fix/pr: fixes from pr review

NOTE: all these follow-up commits will be squashed after the review
is done.

What changed in this commit:

- refactor the code to make the dial retry logic be the same across
  platforms.
- a number of code duplication fixes and refactors, raised by @jedevc

Signed-off-by: Anthony Nandaa <profnandaa@gmail.com>

util/testutil/integration/util.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"os"
 	"os/exec"
+	"strings"
 	"sync"
 	"syscall"
 	"testing"
@@ -102,7 +103,25 @@ func StartCmd(cmd *exec.Cmd, logs map[string]*bytes.Buffer) (func() error, error
 // 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)
+	address = strings.TrimPrefix(address, socketScheme)
+	step := 50 * time.Millisecond
+	i := 0
+	for {
+		if cmd != nil && cmd.ProcessState != nil {
+			return errors.Errorf("process exited: %s", cmd.String())
+		}
+
+		if conn, err := dialPipe(address); err == nil {
+			conn.Close()
+			break
+		}
+		i++
+		if time.Duration(i)*step > d {
+			return errors.Errorf("failed dialing: %s", address)
+		}
+		time.Sleep(step)
+	}
+	return nil
 }
 
 func LookupBinary(name string) error {

util/testutil/integration/util_unix.go

@@ -5,36 +5,19 @@ 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://")
+var socketScheme = "unix://"
+
+// abstracted function to handle pipe dialing on unix.
+// some simplification has been made to discard
+// laddr for unix -- left as nil.
+func dialPipe(address string) (net.Conn, error) {
 	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, errors.Wrapf(err, "failed resolving unix addr: %s", address)
 	}
-	return nil
+	return net.DialUnix("unix", nil, addr)
 }

util/testutil/integration/util_windows.go

@@ -1,33 +1,15 @@
 package integration
 
 import (
-	"os/exec"
-	"strings"
-	"time"
+	"net"
 
 	"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
+var socketScheme = "npipe://"
 
-	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
+// abstracted function to handle pipe dialing on windows.
+// some simplification has been made to discard timeout param.
+func dialPipe(address string) (net.Conn, error) {
+	return winio.DialPipe(address, nil)
 }

util/testutil/workers/containerd.go

@@ -7,7 +7,6 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
-	"runtime"
 	"strconv"
 	"strings"
 	"time"
@@ -108,7 +107,7 @@ func (c *Containerd) New(ctx context.Context, cfg *integration.BackendConfig) (b
 	}()
 
 	rootless := false
-	if runtime.GOOS != "windows" && c.UID != 0 {
+	if c.UID != 0 {
 		if c.GID == 0 {
 			return nil, nil, errors.Errorf("unsupported id pair: uid=%d, gid=%d", c.UID, c.GID)
 		}
@@ -197,10 +196,22 @@ disabled_plugins = ["cri"]
 	}
 	deferF.Append(ctdStop)
 
-	buildkitdArgs := append(getBuildkitdArgs(address), snBuildkitdArgs...)
+	// handles only windows case, no effect on unix
+	address = normalizeAddress(address)
+	buildkitdArgs := []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
+	}
+	if ociWorkerFlag != "" {
+		buildkitdArgs = append(buildkitdArgs, ociWorkerFlag)
+	}
+	buildkitdArgs = append(buildkitdArgs, snBuildkitdArgs...)
 
-	if runtime.GOOS != "windows" && c.Snapshotter != "native" {
-		c.ExtraEnv = append(c.ExtraEnv, "BUILDKIT_DEBUG_FORCE_OVERLAY_DIFF=true")
+	if envForceOverlayDiff != "" && c.Snapshotter != "native" {
+		c.ExtraEnv = append(c.ExtraEnv, envForceOverlayDiff)
 	}
 	if rootless {
 		pidStr, err := os.ReadFile(filepath.Join(rootlessKitState, "child_pid"))

util/testutil/workers/util.go

@@ -1,7 +1,13 @@
 package workers
 
 import (
+	"bytes"
+	"context"
 	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"time"
 
 	"github.com/moby/buildkit/util/testutil/integration"
 )
@@ -19,3 +25,77 @@ func (osp otelSocketPath) UpdateConfigFile(in string) string {
   socketPath = %q
 `, in, osp)
 }
+
+func runBuildkitd(
+	ctx context.Context,
+	conf *integration.BackendConfig,
+	args []string,
+	logs map[string]*bytes.Buffer,
+	uid, gid int,
+	extraEnv []string,
+) (address string, cl func() error, err error) {
+	deferF := &integration.MultiCloser{}
+	cl = deferF.F()
+
+	defer func() {
+		if err != nil {
+			deferF.F()()
+			cl = nil
+		}
+	}()
+
+	tmpdir, err := os.MkdirTemp("", "bktest_buildkitd")
+	if err != nil {
+		return "", nil, err
+	}
+
+	if err := chown(tmpdir, uid, gid); err != nil {
+		return "", nil, err
+	}
+
+	if err := os.MkdirAll(filepath.Join(tmpdir, "tmp"), 0711); err != nil {
+		return "", nil, err
+	}
+
+	if err := chown(filepath.Join(tmpdir, "tmp"), uid, gid); err != nil {
+		return "", nil, err
+	}
+	deferF.Append(func() error { return os.RemoveAll(tmpdir) })
+
+	cfgfile, err := integration.WriteConfig(
+		append(conf.DaemonConfig, withOTELSocketPath(getTraceSocketPath(tmpdir))))
+	if err != nil {
+		return "", nil, err
+	}
+	deferF.Append(func() error {
+		return os.RemoveAll(filepath.Dir(cfgfile))
+	})
+
+	args = append(args, "--config="+cfgfile)
+	address = getBuildkitdAddr(tmpdir)
+
+	args = append(args, "--root", tmpdir, "--addr", address, "--debug")
+	cmd := exec.Command(args[0], args[1:]...) //nolint:gosec // test utility
+	cmd.Env = append(
+		os.Environ(),
+		"BUILDKIT_DEBUG_EXEC_OUTPUT=1",
+		"BUILDKIT_DEBUG_PANIC_ON_ERROR=1",
+		"TMPDIR="+filepath.Join(tmpdir, "tmp"))
+	cmd.Env = append(cmd.Env, extraEnv...)
+	cmd.SysProcAttr = getSysProcAttr()
+
+	stop, err := integration.StartCmd(cmd, logs)
+	if err != nil {
+		return "", nil, err
+	}
+	deferF.Append(stop)
+
+	if err := integration.WaitSocket(address, 15*time.Second, cmd); err != nil {
+		return "", nil, err
+	}
+
+	// separated out since it's not required in windows
+	mountInfo(deferF, tmpdir)
+
+	return address, cl, err
+}