sandbox: support vsock connection to task api

Signed-off-by: Abel Feng <[email protected]>

Author: abel-von

core/runtime/v2/shim/util_unix.go

@@ -19,17 +19,24 @@
 package shim
 
 import (
+	"bufio"
 	"context"
 	"crypto/sha256"
 	"fmt"
+	"io"
+	"math"
 	"net"
 	"os"
 	"path/filepath"
 	"runtime"
+	"strconv"
 	"strings"
 	"syscall"
 	"time"
 
+	"github.com/containerd/log"
+	"github.com/mdlayher/vsock"
+
 	"github.com/containerd/containerd/v2/defaults"
 	"github.com/containerd/containerd/v2/pkg/namespaces"
 	"github.com/containerd/containerd/v2/pkg/sys"
@@ -38,6 +45,9 @@ import (
 const (
 	shimBinaryFormat = "containerd-shim-%s-%s"
 	socketPathLimit  = 106
+	protoVsock       = "vsock"
+	protoHybridVsock = "hvsock"
+	protoUnix        = "unix"
 )
 
 func getSysProcAttr() *syscall.SysProcAttr {
@@ -76,7 +86,21 @@ func SocketAddress(ctx context.Context, socketPath, id string) (string, error) {
 
 // AnonDialer returns a dialer for a socket
 func AnonDialer(address string, timeout time.Duration) (net.Conn, error) {
-	return net.DialTimeout("unix", socket(address).path(), timeout)
+	proto, addr, ok := strings.Cut(address, "://")
+	if !ok {
+		return net.DialTimeout("unix", socket(address).path(), timeout)
+	}
+	switch proto {
+	case protoVsock:
+		// vsock dialer can not set timeout
+		return dialVsock(addr)
+	case protoHybridVsock:
+		return dialHybridVsock(addr, timeout)
+	case protoUnix:
+		return net.DialTimeout("unix", socket(address).path(), timeout)
+	default:
+		return nil, fmt.Errorf("unsupported protocol: %s", proto)
+	}
 }
 
 // AnonReconnectDialer returns a dialer for an existing socket on reconnection
@@ -177,3 +201,88 @@ func CanConnect(address string) bool {
 	conn.Close()
 	return true
 }
+
+func hybridVsockDialer(addr string, port uint64, timeout time.Duration) (net.Conn, error) {
+	timeoutCh := time.After(timeout)
+	// Do 10 retries before timeout
+	retryInterval := timeout / 10
+	for {
+		conn, err := net.DialTimeout("unix", addr, timeout)
+		if err != nil {
+			return nil, err
+		}
+		if _, err = conn.Write([]byte(fmt.Sprintf("CONNECT %d\n", port))); err != nil {
+			conn.Close()
+			return nil, err
+		}
+		errChan := make(chan error, 1)
+		go func() {
+			reader := bufio.NewReader(conn)
+			response, err := reader.ReadString('\n')
+			if err != nil {
+				errChan <- err
+				return
+			}
+			if strings.Contains(response, "OK") {
+				errChan <- nil
+			} else {
+				errChan <- fmt.Errorf("hybrid vsock handshake response error: %s", response)
+			}
+		}()
+		select {
+		case err = <-errChan:
+			if err != nil {
+				conn.Close()
+				// When it is EOF, maybe the server side is not ready.
+				if err == io.EOF {
+					log.G(context.Background()).Warnf("Read hybrid vsock got EOF, server may not ready")
+					time.Sleep(retryInterval)
+					continue
+				}
+				return nil, err
+			}
+			return conn, nil
+		case <-timeoutCh:
+			conn.Close()
+			return nil, fmt.Errorf("timeout waiting for hybrid vsocket handshake of %s:%d", addr, port)
+		}
+	}
+
+}
+
+func dialVsock(address string) (net.Conn, error) {
+	contextIDString, portString, ok := strings.Cut(address, ":")
+	if !ok {
+		return nil, fmt.Errorf("invalid vsock address %s", address)
+	}
+	contextID, err := strconv.ParseUint(contextIDString, 10, 0)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse vsock context id %s, %v", contextIDString, err)
+	}
+	if contextID > math.MaxUint32 {
+		return nil, fmt.Errorf("vsock context id %d is invalid", contextID)
+	}
+	port, err := strconv.ParseUint(portString, 10, 0)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse vsock port %s, %v", portString, err)
+	}
+	if port > math.MaxUint32 {
+		return nil, fmt.Errorf("vsock port %d is invalid", port)
+	}
+	return vsock.Dial(uint32(contextID), uint32(port), &vsock.Config{})
+}
+
+func dialHybridVsock(address string, timeout time.Duration) (net.Conn, error) {
+	addr, portString, ok := strings.Cut(address, ":")
+	if !ok {
+		return nil, fmt.Errorf("invalid hybrid vsock address %s", address)
+	}
+	port, err := strconv.ParseUint(portString, 10, 0)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse hybrid vsock port %s, %v", portString, err)
+	}
+	if port > math.MaxUint32 {
+		return nil, fmt.Errorf("hybrid vsock port %d is invalid", port)
+	}
+	return hybridVsockDialer(addr, port, timeout)
+}

go.mod

@@ -36,6 +36,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.0.0
 	github.com/intel/goresctrl v0.6.0
 	github.com/klauspost/compress v1.17.6
+	github.com/mdlayher/vsock v1.2.1
 	github.com/minio/sha256-simd v1.0.1
 	github.com/moby/locker v1.0.1
 	github.com/moby/sys/mountinfo v0.7.1
@@ -103,6 +104,7 @@ require (
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.5 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
+	github.com/mdlayher/socket v0.4.1 // indirect
 	github.com/moby/spdystream v0.2.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect

go.sum

@@ -284,6 +284,10 @@ github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJ
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo=
 github.com/matttproud/golang_protobuf_extensions v1.0.4/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
+github.com/mdlayher/socket v0.4.1 h1:eM9y2/jlbs1M615oshPQOHZzj6R6wMT7bX5NPiQvn2U=
+github.com/mdlayher/socket v0.4.1/go.mod h1:cAqeGjoufqdxWkD7DkpyS+wcefOtmu5OQ8KuoJGIReA=
+github.com/mdlayher/vsock v1.2.1 h1:pC1mTJTvjo1r9n9fbm7S1j04rCgCzhCOS5DY0zqHlnQ=
+github.com/mdlayher/vsock v1.2.1/go.mod h1:NRfCibel++DgeMD8z/hP+PPTjlNJsdPOmxcnENvE+SE=
 github.com/minio/sha256-simd v1.0.1 h1:6kaan5IFmwTNynnKKpDHe6FWHohJOHhCPchzK49dzMM=
 github.com/minio/sha256-simd v1.0.1/go.mod h1:Pz6AKMiUdngCLpeTL/RJY1M9rUuPMYujV5xJjtbRSN8=
 github.com/mndrix/tap-go v0.0.0-20171203230836-629fa407e90b/go.mod h1:pzzDgJWZ34fGzaAZGFW22KVZDfyrYW+QABMrWnJBnSs=