Implement executor on Windows

This change splits the containerdexecutor.Run() function into smaller
pieces and enables it to run on Windows.

Signed-off-by: Gabriel Adrian Samfira <gsamfira@cloudbasesolutions.com>

Author: gabriel-samfira

executor/containerdexecutor/executor.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
+	"strings"
 	"sync"
 	"syscall"
 	"time"
@@ -17,19 +18,13 @@ import (
 	"github.com/containerd/containerd"
 	"github.com/containerd/containerd/cio"
 	"github.com/containerd/containerd/mount"
-	containerdoci "github.com/containerd/containerd/oci"
-	"github.com/containerd/continuity/fs"
-	"github.com/docker/docker/pkg/idtools"
 	"github.com/moby/buildkit/executor"
 	"github.com/moby/buildkit/executor/oci"
 	resourcestypes "github.com/moby/buildkit/executor/resources/types"
 	gatewayapi "github.com/moby/buildkit/frontend/gateway/pb"
 	"github.com/moby/buildkit/identity"
-	"github.com/moby/buildkit/snapshot"
 	"github.com/moby/buildkit/solver/pb"
 	"github.com/moby/buildkit/util/network"
-	rootlessspecconv "github.com/moby/buildkit/util/rootless/specconv"
-	"github.com/opencontainers/runtime-spec/specs-go"
 	"github.com/pkg/errors"
 )
 
@@ -39,7 +34,7 @@ type containerdExecutor struct {
 	networkProviders map[pb.NetMode]network.Provider
 	cgroupParent     string
 	dnsConfig        *oci.DNSConfig
-	running          map[string]chan error
+	running          map[string]*jobDetails
 	mu               sync.Mutex
 	apparmorProfile  string
 	selinux          bool
@@ -72,23 +67,36 @@ func New(client *containerd.Client, root, cgroup string, networkProviders map[pb
 		networkProviders: networkProviders,
 		cgroupParent:     cgroup,
 		dnsConfig:        dnsConfig,
-		running:          make(map[string]chan error),
+		running:          make(map[string]*jobDetails),
 		apparmorProfile:  apparmorProfile,
 		selinux:          selinux,
 		traceSocket:      traceSocket,
 		rootless:         rootless,
 	}
 }
 
+type jobDetails struct {
+	done chan error
+	// On linux the rootfsPath is used to ensure the CWD exists, to fetch user information
+	// and as a bind mount for the root FS of the container.
+	rootfsPath string
+	// On Windows we need to use the root mounts to achieve the same thing that Linux does
+	// with rootfsPath. So we save both in details.
+	rootMounts []mount.Mount
+}
+
 func (w *containerdExecutor) Run(ctx context.Context, id string, root executor.Mount, mounts []executor.Mount, process executor.ProcessInfo, started chan<- struct{}) (rec resourcestypes.Recorder, err error) {
 	if id == "" {
 		id = identity.NewID()
 	}
 
 	startedOnce := sync.Once{}
 	done := make(chan error, 1)
+	details := &jobDetails{
+		done: done,
+	}
 	w.mu.Lock()
-	w.running[id] = done
+	w.running[id] = details
 	w.mu.Unlock()
 	defer func() {
 		w.mu.Lock()
@@ -104,60 +112,16 @@ func (w *containerdExecutor) Run(ctx context.Context, id string, root executor.M
 	}()
 
 	meta := process.Meta
-
-	resolvConf, err := oci.GetResolvConf(ctx, w.root, nil, w.dnsConfig)
-	if err != nil {
-		return nil, err
-	}
-
-	hostsFile, clean, err := oci.GetHostsFile(ctx, w.root, meta.ExtraHosts, nil, meta.Hostname)
-	if err != nil {
-		return nil, err
-	}
-	if clean != nil {
-		defer clean()
-	}
-
-	mountable, err := root.Src.Mount(ctx, false)
+	releasers, resolvConf, hostsFile, err := w.prepareExecutionEnv(ctx, root, mounts, meta, details)
 	if err != nil {
+		releasers()
 		return nil, err
 	}
+	defer releasers()
 
-	rootMounts, release, err := mountable.Mount()
-	if err != nil {
+	if err := w.ensureCWD(ctx, details, meta); err != nil {
 		return nil, err
 	}
-	if release != nil {
-		defer release()
-	}
-
-	lm := snapshot.LocalMounterWithMounts(rootMounts)
-	rootfsPath, err := lm.Mount()
-	if err != nil {
-		return nil, err
-	}
-	defer lm.Unmount()
-	defer executor.MountStubsCleaner(ctx, rootfsPath, mounts, meta.RemoveMountStubsRecursive)()
-
-	uid, gid, sgids, err := oci.GetUser(rootfsPath, meta.User)
-	if err != nil {
-		return nil, err
-	}
-
-	identity := idtools.Identity{
-		UID: int(uid),
-		GID: int(gid),
-	}
-
-	newp, err := fs.RootPath(rootfsPath, meta.Cwd)
-	if err != nil {
-		return nil, errors.Wrapf(err, "working dir %s points to invalid target", newp)
-	}
-	if _, err := os.Stat(newp); err != nil {
-		if err := idtools.MkdirAllAndChown(newp, 0755, identity); err != nil {
-			return nil, errors.Wrapf(err, "failed to create working directory %s", newp)
-		}
-	}
 
 	provider, ok := w.networkProviders[meta.NetMode]
 	if !ok {
@@ -173,23 +137,12 @@ func (w *containerdExecutor) Run(ctx context.Context, id string, root executor.M
 		bklog.G(ctx).Info("enabling HostNetworking")
 	}
 
-	opts := []containerdoci.SpecOpts{oci.WithUIDGID(uid, gid, sgids)}
-	if meta.ReadonlyRootFS {
-		opts = append(opts, containerdoci.WithRootFSReadonly())
-	}
-
-	processMode := oci.ProcessSandbox // FIXME(AkihiroSuda)
-	spec, cleanup, err := oci.GenerateSpec(ctx, meta, mounts, id, resolvConf, hostsFile, namespace, w.cgroupParent, processMode, nil, w.apparmorProfile, w.selinux, w.traceSocket, opts...)
+	spec, specReleasers, err := w.getOCISpec(ctx, id, resolvConf, hostsFile, namespace, mounts, meta, details)
 	if err != nil {
+		specReleasers()
 		return nil, err
 	}
-	defer cleanup()
-	spec.Process.Terminal = meta.Tty
-	if w.rootless {
-		if err := rootlessspecconv.ToRootless(spec); err != nil {
-			return nil, err
-		}
-	}
+	defer specReleasers()
 
 	container, err := w.client.NewContainer(ctx, id,
 		containerd.WithSpec(spec),
@@ -210,20 +163,12 @@ func (w *containerdExecutor) Run(ctx context.Context, id string, root executor.M
 		cioOpts = append(cioOpts, cio.WithTerminal)
 	}
 
-	rootfs := containerd.WithRootFS([]mount.Mount{{
-		Source:  rootfsPath,
-		Type:    "bind",
-		Options: []string{"rbind"},
-	}})
-	if runtime.GOOS == "freebsd" {
-		rootfs = containerd.WithRootFS([]mount.Mount{{
-			Source:  rootfsPath,
-			Type:    "nullfs",
-			Options: []string{},
-		}})
+	taskOpts, err := w.getTaskOpts(ctx, details)
+	if err != nil {
+		return nil, err
 	}
 
-	task, err := container.NewTask(ctx, cio.NewCreator(cioOpts...), rootfs)
+	task, err := container.NewTask(ctx, cio.NewCreator(cioOpts...), taskOpts)
 	if err != nil {
 		return nil, err
 	}
@@ -259,17 +204,16 @@ func (w *containerdExecutor) Exec(ctx context.Context, id string, process execut
 	// is in the process of being created and check again every 100ms or until
 	// context is canceled.
 
+	w.mu.Lock()
+	details, ok := w.running[id]
+	w.mu.Unlock()
+
+	if !ok {
+		return errors.Errorf("container %s not found", id)
+	}
 	var container containerd.Container
 	var task containerd.Task
 	for {
-		w.mu.Lock()
-		done, ok := w.running[id]
-		w.mu.Unlock()
-
-		if !ok {
-			return errors.Errorf("container %s not found", id)
-		}
-
 		if container == nil {
 			container, _ = w.client.LoadContainer(ctx, id)
 		}
@@ -285,7 +229,7 @@ func (w *containerdExecutor) Exec(ctx context.Context, id string, process execut
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
-		case err, ok := <-done:
+		case err, ok := <-details.done:
 			if !ok || err == nil {
 				return errors.Errorf("container %s has stopped", id)
 			}
@@ -301,23 +245,24 @@ func (w *containerdExecutor) Exec(ctx context.Context, id string, process execut
 	}
 
 	proc := spec.Process
-
-	// TODO how do we get rootfsPath for oci.GetUser in case user passed in username rather than uid:gid?
-	// For now only support uid:gid
 	if meta.User != "" {
-		uid, gid, err := oci.ParseUIDGID(meta.User)
+		userSpec, err := getUserSpec(meta.User, details.rootfsPath)
 		if err != nil {
 			return errors.WithStack(err)
 		}
-		proc.User = specs.User{
-			UID:            uid,
-			GID:            gid,
-			AdditionalGids: []uint32{},
-		}
+		proc.User = userSpec
 	}
 
 	proc.Terminal = meta.Tty
-	proc.Args = meta.Args
+
+	if runtime.GOOS == "windows" {
+		// On Windows passing in Args will lead to double escaping by hcsshim, which leads to errors.
+		// The recommendation is to use CommandLine.
+		proc.CommandLine = strings.Join(meta.Args, " ")
+	} else {
+		proc.Args = meta.Args
+	}
+
 	if meta.Cwd != "" {
 		spec.Process.Cwd = meta.Cwd
 	}