Expose API to access VMI via Vsock

Signed-off-by: Zhuchen Wang <[email protected]>

api/openapi-spec/swagger.json

@@ -11911,6 +11911,43 @@
      }
     ]
    },
+   "/apis/subresources.kubevirt.io/v1/namespaces/{namespace:[a-z0-9][a-z0-9\\-]*}/virtualmachineinstances/{name:[a-z0-9][a-z0-9\\-]*}/vsock": {
+    "get": {
+     "description": "Open a websocket connection forwarding traffic to the specified VirtualMachineInstance and port via VSOCK.",
+     "operationId": "v1VSOCK",
+     "responses": {
+      "401": {
+       "description": "Unauthorized"
+      }
+     }
+    },
+    "parameters": [
+     {
+      "uniqueItems": true,
+      "type": "string",
+      "description": "Name of the resource",
+      "name": "name",
+      "in": "path",
+      "required": true
+     },
+     {
+      "uniqueItems": true,
+      "type": "string",
+      "description": "Object name and auth scope, such as for teams and projects",
+      "name": "namespace",
+      "in": "path",
+      "required": true
+     },
+     {
+      "uniqueItems": true,
+      "type": "integer",
+      "description": "The port which the VSOCK application listens to.",
+      "name": "port",
+      "in": "query",
+      "required": true
+     }
+    ]
+   },
    "/apis/subresources.kubevirt.io/v1/namespaces/{namespace:[a-z0-9][a-z0-9\\-]*}/virtualmachines/{name:[a-z0-9][a-z0-9\\-]*}/addvolume": {
     "put": {
      "description": "Add a volume and disk to a running Virtual Machine.",
@@ -13300,6 +13337,43 @@
      }
     ]
    },
+   "/apis/subresources.kubevirt.io/v1alpha3/namespaces/{namespace:[a-z0-9][a-z0-9\\-]*}/virtualmachineinstances/{name:[a-z0-9][a-z0-9\\-]*}/vsock": {
+    "get": {
+     "description": "Open a websocket connection forwarding traffic to the specified VirtualMachineInstance and port via VSOCK.",
+     "operationId": "v1alpha3VSOCK",
+     "responses": {
+      "401": {
+       "description": "Unauthorized"
+      }
+     }
+    },
+    "parameters": [
+     {
+      "uniqueItems": true,
+      "type": "string",
+      "description": "Name of the resource",
+      "name": "name",
+      "in": "path",
+      "required": true
+     },
+     {
+      "uniqueItems": true,
+      "type": "string",
+      "description": "Object name and auth scope, such as for teams and projects",
+      "name": "namespace",
+      "in": "path",
+      "required": true
+     },
+     {
+      "uniqueItems": true,
+      "type": "integer",
+      "description": "The port which the VSOCK application listens to.",
+      "name": "port",
+      "in": "query",
+      "required": true
+     }
+    ]
+   },
    "/apis/subresources.kubevirt.io/v1alpha3/namespaces/{namespace:[a-z0-9][a-z0-9\\-]*}/virtualmachines/{name:[a-z0-9][a-z0-9\\-]*}/addvolume": {
     "put": {
      "description": "Add a volume and disk to a running Virtual Machine.",

cmd/virt-handler/virt-handler.go

@@ -516,6 +516,7 @@ func (app *virtHandlerApp) runServer(errCh chan error, consoleHandler *rest.Cons
 	ws.Route(ws.GET("/v1/namespaces/{namespace}/virtualmachineinstances/{name}/guestosinfo").To(lifecycleHandler.GetGuestInfo).Produces(restful.MIME_JSON).Consumes(restful.MIME_JSON).Returns(http.StatusOK, "OK", v1.VirtualMachineInstanceGuestAgentInfo{}))
 	ws.Route(ws.GET("/v1/namespaces/{namespace}/virtualmachineinstances/{name}/userlist").To(lifecycleHandler.GetUsers).Produces(restful.MIME_JSON).Consumes(restful.MIME_JSON).Returns(http.StatusOK, "OK", v1.VirtualMachineInstanceGuestOSUserList{}))
 	ws.Route(ws.GET("/v1/namespaces/{namespace}/virtualmachineinstances/{name}/filesystemlist").To(lifecycleHandler.GetFilesystems).Produces(restful.MIME_JSON).Consumes(restful.MIME_JSON).Returns(http.StatusOK, "OK", v1.VirtualMachineInstanceFileSystemList{}))
+	ws.Route(ws.GET("/v1/namespaces/{namespace}/virtualmachineinstances/{name}/vsock").Param(restful.QueryParameter("port", "Target VSOCK port")).To(consoleHandler.VSOCKHandler))
 	restful.DefaultContainer.Add(ws)
 	server := &http.Server{
 		Addr:    fmt.Sprintf("%s:%d", app.ServiceListen.BindAddress, app.consoleServerPort),

go.mod

@@ -35,6 +35,7 @@ require (
 	github.com/kisielk/errcheck v1.6.2
 	github.com/krolaw/dhcp4 v0.0.0-20180925202202-7cead472c414
 	github.com/kubernetes-csi/external-snapshotter/client/v4 v4.2.0
+	github.com/mdlayher/vsock v1.1.1
 	github.com/mitchellh/go-ps v0.0.0-20190716172923-621e5597135b
 	github.com/mitchellh/go-vnc v0.0.0-20150629162542-723ed9867aed
 	github.com/moby/sys/mountinfo v0.5.0
@@ -119,7 +120,7 @@ require (
 	github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0 // indirect
 	github.com/godbus/dbus/v5 v5.0.6 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
-	github.com/google/go-cmp v0.5.6 // indirect
+	github.com/google/go-cmp v0.5.7 // indirect
 	github.com/google/go-querystring v1.0.0 // indirect
 	github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38 // indirect
 	github.com/google/renameio v0.1.0 // indirect
@@ -132,6 +133,7 @@ require (
 	github.com/mattn/go-isatty v0.0.12 // indirect
 	github.com/mattn/go-runewidth v0.0.12 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
+	github.com/mdlayher/socket v0.2.0 // indirect
 	github.com/mitchellh/mapstructure v1.4.1 // indirect
 	github.com/moby/spdystream v0.2.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
@@ -148,6 +150,7 @@ require (
 	go.mongodb.org/mongo-driver v1.8.4 // indirect
 	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect
 	golang.org/x/oauth2 v0.0.0-20211104180415-d3ed0bb246c8 // indirect
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
 	golang.org/x/text v0.3.7 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/genproto v0.0.0-20210831024726-fe130286e0e2 // indirect

go.sum

@@ -499,8 +499,8 @@ github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/google/go-cmp v0.5.6 h1:BKbKCqvP6I+rmFHt06ZmyQtvB8xAkWdhFyr0ZUNZcxQ=
-github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.7 h1:81/ik6ipDQS2aGcBfIN5dHDB36BwrStyeAQquSYCV4o=
+github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8/DtOE=
 github.com/google/go-github/v32 v32.0.0 h1:q74KVb22spUq0U5HqZ9VCYqQz8YRuOtL/39ZnfwO+NM=
 github.com/google/go-github/v32 v32.0.0/go.mod h1:rIEpZD9CTDQwDK9GDrtMTycQNA4JU3qBsCizh3q2WCI=
 github.com/google/go-querystring v1.0.0 h1:Xkwi/a1rcvNg1PPYe5vI8GbeBY/jrVuDX5ASuANWTrk=
@@ -736,6 +736,10 @@ github.com/mdlayher/netlink v1.1.0/go.mod h1:H4WCitaheIsdF9yOYu8CFmCgQthAPIWZmcK
 github.com/mdlayher/netlink v1.1.1/go.mod h1:WTYpFb/WTvlRJAyKhZL5/uy69TDDpHHu2VZmb2XgV7o=
 github.com/mdlayher/raw v0.0.0-20190606142536-fef19f00fc18/go.mod h1:7EpbotpCmVZcu+KCX4g9WaRNuu11uyhiW7+Le1dKawg=
 github.com/mdlayher/raw v0.0.0-20191009151244-50f2db8cc065/go.mod h1:7EpbotpCmVZcu+KCX4g9WaRNuu11uyhiW7+Le1dKawg=
+github.com/mdlayher/socket v0.2.0 h1:EY4YQd6hTAg2tcXF84p5DTHazShE50u5HeBzBaNgjkA=
+github.com/mdlayher/socket v0.2.0/go.mod h1:QLlNPkFR88mRUNQIzRBMfXxwKal8H7u1h3bL1CV+f0E=
+github.com/mdlayher/vsock v1.1.1 h1:8lFuiXQnmICBrCIIA9PMgVSke6Fg6V4+r0v7r55k88I=
+github.com/mdlayher/vsock v1.1.1/go.mod h1:Y43jzcy7KM3QB+/FK15pfqGxDMCMzUXWegEfIbSM18U=
 github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
 github.com/miekg/dns v1.1.15/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
 github.com/miekg/dns v1.1.22/go.mod h1:bPDLeHnStXmXAq1m/Ch/hvfNHr14JKNPMBo3VZKjuso=
@@ -1256,6 +1260,7 @@ golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

pkg/virt-api/api.go

@@ -358,6 +358,11 @@ func (app *virtAPIApp) composeSubresources() {
 			Param(definitions.PortForwardProtocolParameter(subws)).
 			Operation(version.Version + "vmi-PortForwardWithProtocol").
 			Doc("Open a websocket connection forwarding traffic of the specified protocol (either tcp or udp) to the specified VirtualMachineInstance and port."))
+		subws.Route(subws.GET(definitions.NamespacedResourcePath(subresourcesvmiGVR) + definitions.SubResourcePath("vsock")).
+			To(subresourceApp.VSOCKRequestHandler).
+			Param(definitions.NamespaceParam(subws)).Param(definitions.NameParam(subws)).Param(definitions.VSOCKPortParameter(subws)).
+			Operation(version.Version + "VSOCK").
+			Doc("Open a websocket connection forwarding traffic to the specified VirtualMachineInstance and port via VSOCK."))
 
 		// VM endpoint
 		subws.Route(subws.GET(definitions.NamespacedResourcePath(subresourcesvmGVR) + definitions.SubResourcePath("portforward") + definitions.PortPath).

pkg/virt-api/definitions/definitions.go

@@ -678,3 +678,7 @@ func PortForwardProtocolParameter(ws *restful.WebService) *restful.Parameter {
 }
 
 func noop(_ *restful.Request, _ *restful.Response) {}
+
+func VSOCKPortParameter(ws *restful.WebService) *restful.Parameter {
+	return ws.QueryParameter(PortParamName, "The port which the VSOCK application listens to.").DataType("integer").Required(true)
+}

pkg/virt-api/rest/BUILD.bazel

@@ -14,6 +14,7 @@ go_library(
         "subresource.go",
         "usbredir.go",
         "vnc.go",
+        "vsock.go",
     ],
     importpath = "kubevirt.io/kubevirt/pkg/virt-api/rest",
     visibility = ["//visibility:public"],

pkg/virt-api/rest/vsock.go

@@ -0,0 +1,35 @@
+package rest
+
+import (
+	"fmt"
+
+	restful "github.com/emicklei/go-restful"
+	"k8s.io/apimachinery/pkg/api/errors"
+
+	v1 "kubevirt.io/api/core/v1"
+	"kubevirt.io/client-go/kubecli"
+	"kubevirt.io/client-go/log"
+
+	"kubevirt.io/kubevirt/pkg/util"
+)
+
+func (app *SubresourceAPIApp) VSOCKRequestHandler(request *restful.Request, response *restful.Response) {
+	streamer := NewRawStreamer(
+		app.FetchVirtualMachineInstance,
+		validateVMIForVSOCK,
+		app.virtHandlerDialer(func(vmi *v1.VirtualMachineInstance, conn kubecli.VirtHandlerConn) (string, error) {
+			return conn.VSOCKURI(vmi, request.QueryParameter("port"))
+		}),
+	)
+
+	streamer.Handle(request, response)
+}
+
+func validateVMIForVSOCK(vmi *v1.VirtualMachineInstance) *errors.StatusError {
+	if !util.IsAutoAttachVSOCK(vmi) {
+		err := fmt.Errorf("VSOCK is not attached.")
+		log.Log.Object(vmi).Reason(err).Error("Can't establish Vsock connection.")
+		return errors.NewBadRequest(err.Error())
+	}
+	return nil
+}

pkg/virt-handler/rest/BUILD.bazel

@@ -10,12 +10,14 @@ go_library(
     importpath = "kubevirt.io/kubevirt/pkg/virt-handler/rest",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/util:go_default_library",
         "//pkg/virt-handler/cmd-client:go_default_library",
         "//pkg/virt-handler/isolation:go_default_library",
         "//staging/src/kubevirt.io/api/core/v1:go_default_library",
         "//staging/src/kubevirt.io/client-go/kubecli:go_default_library",
         "//staging/src/kubevirt.io/client-go/log:go_default_library",
         "//vendor/github.com/emicklei/go-restful:go_default_library",
+        "//vendor/github.com/mdlayher/vsock:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/yaml:go_default_library",

pkg/virt-handler/rest/console.go

@@ -31,13 +31,15 @@ import (
 	"sync"
 
 	"github.com/emicklei/go-restful"
+	"github.com/mdlayher/vsock"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
 
 	v1 "kubevirt.io/api/core/v1"
 	"kubevirt.io/client-go/kubecli"
 	"kubevirt.io/client-go/log"
 
+	"kubevirt.io/kubevirt/pkg/util"
 	"kubevirt.io/kubevirt/pkg/virt-handler/isolation"
 )
 
@@ -131,7 +133,7 @@ func (t *ConsoleHandler) USBRedirHandler(request *restful.Request, response *res
 		usbHandler := t.usbredir[uid]
 		delete(usbHandler.stopChans, slotId)
 	}()
-	t.stream(vmi, request, response, unixSocketPath, stopChan)
+	t.stream(vmi, request, response, unixSocketDialer(vmi, unixSocketPath), stopChan)
 }
 
 func (t *ConsoleHandler) VNCHandler(request *restful.Request, response *restful.Response) {
@@ -150,7 +152,7 @@ func (t *ConsoleHandler) VNCHandler(request *restful.Request, response *restful.
 	uid := vmi.GetUID()
 	stopChn := newStopChan(uid, t.vncLock, t.vncStopChans)
 	defer deleteStopChan(uid, stopChn, t.vncLock, t.vncStopChans)
-	t.stream(vmi, request, response, unixSocketPath, stopChn)
+	t.stream(vmi, request, response, unixSocketDialer(vmi, unixSocketPath), stopChn)
 }
 
 func (t *ConsoleHandler) SerialHandler(request *restful.Request, response *restful.Response) {
@@ -169,7 +171,46 @@ func (t *ConsoleHandler) SerialHandler(request *restful.Request, response *restf
 	uid := vmi.GetUID()
 	stopCh := newStopChan(uid, t.serialLock, t.serialStopChans)
 	defer deleteStopChan(uid, stopCh, t.serialLock, t.serialStopChans)
-	t.stream(vmi, request, response, unixSocketPath, stopCh)
+	t.stream(vmi, request, response, unixSocketDialer(vmi, unixSocketPath), stopCh)
+}
+
+func (t *ConsoleHandler) VSOCKHandler(request *restful.Request, response *restful.Response) {
+	vmi, code, err := getVMI(request, t.vmiInformer)
+	if err != nil {
+		log.Log.Object(vmi).Reason(err).Error(failedRetrieveVMI)
+		response.WriteError(code, err)
+		return
+	}
+	log.Log.Object(vmi).Info("In VSOCKHandler")
+	if !util.IsAutoAttachVSOCK(vmi) {
+		response.WriteError(http.StatusBadRequest, errors.New("VM doesn't have VSOCK enabled"))
+		return
+	}
+	if vmi.Status.VSOCKCID == nil {
+		// This should not happen.
+		err := errors.New("VSOCK CID is nil")
+		log.Log.Object(vmi).Error(err.Error())
+		response.WriteError(http.StatusInternalServerError, err)
+		return
+	}
+	portParam := request.QueryParameter("port")
+	port, err := strconv.ParseUint(portParam, 10, 32)
+	if err != nil {
+		log.Log.Object(vmi).Reason(err).Errorf("Failed parsing the path parameter port %s", portParam)
+		response.WriteError(http.StatusBadRequest, err)
+		return
+	}
+	cid := *vmi.Status.VSOCKCID
+	t.stream(vmi, request, response, func() (net.Conn, error) {
+		log.Log.Object(vmi).Infof("Connecting to %d:%d", cid, port)
+		conn, err := vsock.Dial(cid, uint32(port), &vsock.Config{})
+		if err != nil {
+			log.Log.Object(vmi).Reason(err).Errorf("failed to dial vsock %d:%d", cid, port)
+			return nil, err
+		}
+		log.Log.Object(vmi).Infof("Connected to %d:%d", cid, port)
+		return conn, nil
+	}, make(chan struct{})) // It is legitimate and up to the guest-application to accept multiple connections.
 }
 
 func newStopChan(uid types.UID, lock *sync.Mutex, stopChans map[types.UID](chan struct{})) chan struct{} {
@@ -209,7 +250,20 @@ func (t *ConsoleHandler) getUnixSocketPath(vmi *v1.VirtualMachineInstance, socke
 	return socketPath, nil
 }
 
-func (t *ConsoleHandler) stream(vmi *v1.VirtualMachineInstance, request *restful.Request, response *restful.Response, unixSocketPath string, stopCh chan struct{}) {
+func unixSocketDialer(vmi *v1.VirtualMachineInstance, unixSocketPath string) func() (net.Conn, error) {
+	return func() (net.Conn, error) {
+		log.Log.Object(vmi).Infof("Connecting to %s", unixSocketPath)
+		fd, err := net.Dial("unix", unixSocketPath)
+		if err != nil {
+			log.Log.Object(vmi).Reason(err).Errorf("failed to dial unix socket %s", unixSocketPath)
+			return nil, err
+		}
+		log.Log.Object(vmi).Infof("Connected to %s", unixSocketPath)
+		return fd, nil
+	}
+}
+
+func (t *ConsoleHandler) stream(vmi *v1.VirtualMachineInstance, request *restful.Request, response *restful.Response, dial func() (net.Conn, error), stopCh chan struct{}) {
 	var upgrader = kubecli.NewUpgrader()
 	clientSocket, err := upgrader.Upgrade(response.ResponseWriter, request.Request, nil)
 	if err != nil {
@@ -220,27 +274,23 @@ func (t *ConsoleHandler) stream(vmi *v1.VirtualMachineInstance, request *restful
 	defer clientSocket.Close()
 
 	log.Log.Object(vmi).Infof("Websocket connection upgraded")
-	log.Log.Object(vmi).Infof("Connecting to %s", unixSocketPath)
 
-	fd, err := net.Dial("unix", unixSocketPath)
+	conn, err := dial()
 	if err != nil {
-		log.Log.Object(vmi).Reason(err).Errorf("failed to dial unix socket %s", unixSocketPath)
 		response.WriteHeader(http.StatusInternalServerError)
 		return
 	}
-	defer fd.Close()
-
-	log.Log.Object(vmi).Infof("Connected to %s", unixSocketPath)
+	defer conn.Close()
 
 	errCh := make(chan error, 2)
 	go func() {
-		_, err := kubecli.CopyTo(clientSocket, fd)
+		_, err := kubecli.CopyTo(clientSocket, conn)
 		log.Log.Object(vmi).Reason(err).Error("error encountered reading from unix socket")
 		errCh <- err
 	}()
 
 	go func() {
-		_, err := kubecli.CopyFrom(fd, clientSocket)
+		_, err := kubecli.CopyFrom(conn, clientSocket)
 		log.Log.Object(vmi).Reason(err).Error("error encountered reading from client (virt-api) websocket")
 		errCh <- err
 	}()