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 <[email protected]>

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

@@ -108,7 +108,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,7 +197,18 @@ disabled_plugins = ["cri"]
  }
  deferF.Append(ctdStop)
 
- buildkitdArgs := append(getBuildkitdArgs(address), snBuildkitdArgs...)
+ // handles only windows case, no effect on unix
+ address = normalizeAddress(address)
+ buildkitdArgs := append([]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:/moby/buildkit/pull/603
+ }, snBuildkitdArgs...)
+
+ if ociWorkerFlag != "" {
+ buildkitdArgs = append(buildkitdArgs, ociWorkerFlag)
+ }
 
  if runtime.GOOS != "windows" && c.Snapshotter != "native" {
  c.ExtraEnv = append(c.ExtraEnv, "BUILDKIT_DEBUG_FORCE_OVERLAY_DIFF=true")

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
+}