client/windows integration tests: plumbing work for running on windows

cmd/buildkitd/main_containerd_worker.go

@@ -354,7 +354,7 @@ func validContainerdSocket(cfg config.ContainerdConfig) bool {
 		// FIXME(AkihiroSuda): prohibit tcp?
 		return true
 	}
-	socketPath := strings.TrimPrefix(socket, "unix://")
+	socketPath := strings.TrimPrefix(socket, socketScheme)
 	if _, err := os.Stat(socketPath); errors.Is(err, os.ErrNotExist) {
 		// FIXME(AkihiroSuda): add more conditions
 		bklog.L.Warnf("skipping containerd worker, as %q does not exist", socketPath)

cmd/buildkitd/main_unix.go

@@ -14,6 +14,8 @@ import (
 	"github.com/pkg/errors"
 )
 
+const socketScheme = "unix://"
+
 func init() {
 	syscall.Umask(0)
 }

cmd/buildkitd/main_windows.go

@@ -13,6 +13,8 @@ import (
 	"github.com/pkg/errors"
 )
 
+const socketScheme = "npipe://"
+
 func listenFD(addr string, tlsConfig *tls.Config) (net.Listener, error) {
 	return nil, errors.New("listening server on fd not supported on windows")
 }

util/testutil/integration/sandbox.go

@@ -99,7 +99,7 @@ func newSandbox(ctx context.Context, w Worker, mirror string, mv matrixValue) (s
 
 	b, closer, err := w.New(ctx, cfg)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, errors.Wrap(err, "creating worker")
 	}
 	deferF.Append(closer)
 

util/testutil/integration/util.go

@@ -5,10 +5,8 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"net"
 	"os"
 	"os/exec"
-	"strings"
 	"sync"
 	"syscall"
 	"testing"
@@ -100,31 +98,11 @@ func StartCmd(cmd *exec.Cmd, logs map[string]*bytes.Buffer) (func() error, error
 	}, nil
 }
 
-func WaitUnix(address string, d time.Duration, cmd *exec.Cmd) error {
-	address = strings.TrimPrefix(address, "unix://")
-	addr, err := net.ResolveUnixAddr("unix", address)
-	if err != nil {
-		return errors.Wrapf(err, "failed resolving unix addr: %s", address)
-	}
-
-	step := 50 * time.Millisecond
-	i := 0
-	for {
-		if cmd != nil && cmd.ProcessState != nil {
-			return errors.Errorf("process exited: %s", cmd.String())
-		}
-
-		if conn, err := net.DialUnix("unix", nil, addr); err == nil {
-			conn.Close()
-			break
-		}
-		i++
-		if time.Duration(i)*step > d {
-			return errors.Errorf("failed dialing: %s", address)
-		}
-		time.Sleep(step)
-	}
-	return nil
+// WaitSocket will dial a socket opened by a command passed in as cmd.
+// On Linux this socket is typically a Unix socket,
+// while on Windows this will be a named pipe.
+func WaitSocket(address string, d time.Duration, cmd *exec.Cmd) error {
+	return waitSocket(address, d, cmd)
 }
 
 func LookupBinary(name string) error {

util/testutil/integration/util_unix.go

@@ -0,0 +1,40 @@
+//go:build !windows
+// +build !windows
+
+package integration
+
+import (
+	"net"
+	"os/exec"
+	"strings"
+	"time"
+
+	"github.com/pkg/errors"
+)
+
+func waitSocket(address string, d time.Duration, cmd *exec.Cmd) error {
+	address = strings.TrimPrefix(address, "unix://")
+	addr, err := net.ResolveUnixAddr("unix", address)
+	if err != nil {
+		return errors.Wrapf(err, "failed resolving unix addr: %s", address)
+	}
+
+	step := 50 * time.Millisecond
+	i := 0
+	for {
+		if cmd != nil && cmd.ProcessState != nil {
+			return errors.Errorf("process exited: %s", cmd.String())
+		}
+
+		if conn, err := net.DialUnix("unix", nil, addr); err == nil {
+			conn.Close()
+			break
+		}
+		i++
+		if time.Duration(i)*step > d {
+			return errors.Errorf("failed dialing: %s", address)
+		}
+		time.Sleep(step)
+	}
+	return nil
+}

util/testutil/integration/util_windows.go

@@ -0,0 +1,33 @@
+package integration
+
+import (
+	"os/exec"
+	"strings"
+	"time"
+
+	"github.com/Microsoft/go-winio"
+	"github.com/pkg/errors"
+)
+
+func waitSocket(address string, d time.Duration, cmd *exec.Cmd) error {
+	address = strings.TrimPrefix(address, "npipe://")
+	step := 50 * time.Millisecond
+	i := 0
+
+	for {
+		if cmd != nil && cmd.ProcessState != nil {
+			return errors.Errorf("process exited: %s", cmd.String())
+		}
+
+		if conn, err := winio.DialPipe(address, nil); err == nil {
+			conn.Close()
+			break
+		}
+		i++
+		if time.Duration(i)*step > d {
+			return errors.Errorf("failed dialing: %s", address)
+		}
+		time.Sleep(step)
+	}
+	return nil
+}

util/testutil/workers/containerd.go

@@ -88,9 +88,11 @@ func (c *Containerd) New(ctx context.Context, cfg *integration.BackendConfig) (b
 	if err := integration.LookupBinary(c.Containerd); err != nil {
 		return nil, nil, err
 	}
+
 	if err := integration.LookupBinary("buildkitd"); err != nil {
 		return nil, nil, err
 	}
+
 	if err := requireRoot(); err != nil {
 		return nil, nil, err
 	}
@@ -106,7 +108,7 @@ func (c *Containerd) New(ctx context.Context, cfg *integration.BackendConfig) (b
 	}()
 
 	rootless := false
-	if c.UID != 0 {
+	if runtime.GOOS != "windows" && c.UID != 0 {
 		if c.GID == 0 {
 			return nil, nil, errors.Errorf("unsupported id pair: uid=%d, gid=%d", c.UID, c.GID)
 		}
@@ -117,6 +119,7 @@ func (c *Containerd) New(ctx context.Context, cfg *integration.BackendConfig) (b
 	if err != nil {
 		return nil, nil, err
 	}
+
 	if rootless {
 		if err := os.Chown(tmpdir, c.UID, c.GID); err != nil {
 			return nil, nil, err
@@ -125,7 +128,7 @@ func (c *Containerd) New(ctx context.Context, cfg *integration.BackendConfig) (b
 
 	deferF.Append(func() error { return os.RemoveAll(tmpdir) })
 
-	address := filepath.Join(tmpdir, "containerd.sock")
+	address := getContainerdSock(tmpdir)
 	config := fmt.Sprintf(`root = %q
 state = %q
 # CRI plugins listens on 10010/tcp for stream server.
@@ -137,8 +140,11 @@ disabled_plugins = ["cri"]
 
 [debug]
   level = "debug"
-  address = %q
-`, filepath.Join(tmpdir, "root"), filepath.Join(tmpdir, "state"), address, filepath.Join(tmpdir, "debug.sock"))
+  address = %q`,
+		filepath.Join(tmpdir, "root"),
+		filepath.Join(tmpdir, "state"),
+		address, getContainerdDebugSock(tmpdir),
+	)
 
 	var snBuildkitdArgs []string
 	if c.Snapshotter != "" {
@@ -185,19 +191,13 @@ disabled_plugins = ["cri"]
 	if err != nil {
 		return nil, nil, err
 	}
-	if err := integration.WaitUnix(address, 10*time.Second, cmd); err != nil {
+	if err := integration.WaitSocket(address, 10*time.Second, cmd); err != nil {
 		ctdStop()
 		return nil, nil, errors.Wrapf(err, "containerd did not start up: %s", integration.FormatLogs(cfg.Logs))
 	}
 	deferF.Append(ctdStop)
 
-	buildkitdArgs := append([]string{"buildkitd",
-		"--oci-worker=false",
-		"--containerd-worker-gc=false",
-		"--containerd-worker=true",
-		"--containerd-worker-addr", address,
-		"--containerd-worker-labels=org.mobyproject.buildkit.worker.sandbox=true", // Include use of --containerd-worker-labels to trigger https://github.com/moby/buildkit/pull/603
-	}, snBuildkitdArgs...)
+	buildkitdArgs := append(getBuildkitdArgs(address), snBuildkitdArgs...)
 
 	if runtime.GOOS != "windows" && c.Snapshotter != "native" {
 		c.ExtraEnv = append(c.ExtraEnv, "BUILDKIT_DEBUG_FORCE_OVERLAY_DIFF=true")
@@ -266,9 +266,13 @@ func runStargzSnapshotter(cfg *integration.BackendConfig) (address string, cl fu
 	if err != nil {
 		return "", nil, err
 	}
-	if err = integration.WaitUnix(address, 10*time.Second, cmd); err != nil {
+	if err = integration.WaitSocket(address, 10*time.Second, cmd); err != nil {
 		snStop()
-		return "", nil, errors.Wrapf(err, "containerd-stargz-grpc did not start up: %s", integration.FormatLogs(cfg.Logs))
+		errMsg := fmt.Sprintf(
+			"containerd-stargz-grpc did not start up: %s",
+			integration.FormatLogs(cfg.Logs),
+		)
+		return "", nil, errors.Wrapf(err, errMsg)
 	}
 	deferF.Append(snStop)
 

util/testutil/workers/dockerd.go

@@ -159,7 +159,7 @@ func (c Moby) New(ctx context.Context, cfg *integration.BackendConfig) (b integr
 	}
 	deferF.Append(d.StopWithError)
 
-	if err := integration.WaitUnix(d.Sock(), 5*time.Second, nil); err != nil {
+	if err := integration.WaitSocket(d.Sock(), 5*time.Second, nil); err != nil {
 		return nil, nil, errors.Errorf("dockerd did not start up: %q, %s", err, integration.FormatLogs(cfg.Logs))
 	}
 

util/testutil/workers/sysprocattr_unix.go

@@ -21,3 +21,21 @@ func getBuildkitdAddr(tmpdir string) string {
 func getTraceSocketPath(tmpdir string) string {
 	return filepath.Join(tmpdir, "otel-grpc.sock")
 }
+
+func getContainerdSock(tmpdir string) string {
+	return filepath.Join(tmpdir, "containerd.sock")
+}
+
+func getContainerdDebugSock(tmpdir string) string {
+	return filepath.Join(tmpdir, "debug.sock")
+}
+
+func getBuildkitdArgs(address string) []string {
+	return []string{"buildkitd",
+		"--oci-worker=false",
+		"--containerd-worker-gc=false",
+		"--containerd-worker=true",
+		"--containerd-worker-addr", address,
+		"--containerd-worker-labels=org.mobyproject.buildkit.worker.sandbox=true", // Include use of --containerd-worker-labels to trigger https://github.com/moby/buildkit/pull/603
+	}
+}

util/testutil/workers/sysprocattr_windows.go

@@ -5,6 +5,7 @@ package workers
 
 import (
 	"path/filepath"
+	"strings"
 	"syscall"
 )
 
@@ -13,9 +14,30 @@ func getSysProcAttr() *syscall.SysProcAttr {
 }
 
 func getBuildkitdAddr(tmpdir string) string {
-	return "//./pipe/buildkitd-" + filepath.Base(tmpdir)
+	return "npipe:////./pipe/buildkitd-" + filepath.Base(tmpdir)
 }
 
 func getTraceSocketPath(tmpdir string) string {
 	return `\\.\pipe\buildkit-otel-grpc-` + filepath.Base(tmpdir)
 }
+
+func getContainerdSock(tmpdir string) string {
+	return `\\.\pipe\containerd-` + filepath.Base(tmpdir)
+}
+
+func getContainerdDebugSock(tmpdir string) string {
+	return `\\.\pipe\containerd-` + filepath.Base(tmpdir) + `debug`
+}
+
+func getBuildkitdArgs(address string) []string {
+	address = filepath.ToSlash(address)
+	if !strings.HasPrefix(address, "npipe://") {
+		address = "npipe://" + address
+	}
+	return []string{"buildkitd",
+		"--containerd-worker-gc=false",
+		"--containerd-worker=true",
+		"--containerd-worker-addr", address,
+		"--containerd-worker-labels=org.mobyproject.buildkit.worker.sandbox=true", // Include use of --containerd-worker-labels to trigger https://github.com/moby/buildkit/pull/603
+	}
+}