allow executors to define containerd and docker behavior

pkg/agent/containerd/config_linux.go

@@ -4,7 +4,6 @@
 package containerd
 
 import (
-	"context"
 	"os"
 
 	"github.com/containerd/containerd"
@@ -39,9 +38,9 @@ func getContainerdArgs(cfg *config.Node) []string {
 	return args
 }
 
-// setupContainerdConfig generates the containerd.toml, using a template combined with various
+// SetupContainerdConfig generates the containerd.toml, using a template combined with various
 // runtime configurations and registry mirror settings provided by the administrator.
-func setupContainerdConfig(ctx context.Context, cfg *config.Node) error {
+func SetupContainerdConfig(cfg *config.Node) error {
 	isRunningInUserNS := userns.RunningInUserNS()
 	_, _, controllers := cgroups.CheckCgroups()
 	// "/sys/fs/cgroup" is namespaced

pkg/agent/containerd/config_windows.go

@@ -4,8 +4,6 @@
 package containerd
 
 import (
-	"context"
-
 	"github.com/containerd/containerd"
 	"github.com/k3s-io/k3s/pkg/agent/templates"
 	"github.com/k3s-io/k3s/pkg/daemons/config"
@@ -23,9 +21,9 @@ func getContainerdArgs(cfg *config.Node) []string {
 	return args
 }
 
-// setupContainerdConfig generates the containerd.toml, using a template combined with various
+// SetupContainerdConfig generates the containerd.toml, using a template combined with various
 // runtime configurations and registry mirror settings provided by the administrator.
-func setupContainerdConfig(ctx context.Context, cfg *config.Node) error {
+func SetupContainerdConfig(cfg *config.Node) error {
 	if cfg.SELinux {
 		logrus.Warn("SELinux isn't supported on windows")
 	}

pkg/agent/containerd/containerd.go

@@ -41,10 +41,6 @@ var (
 // Run configures and starts containerd as a child process. Once it is up, images are preloaded
 // or pulled from files found in the agent images directory.
 func Run(ctx context.Context, cfg *config.Node) error {
-	if err := setupContainerdConfig(ctx, cfg); err != nil {
-		return err
-	}
-
 	args := getContainerdArgs(cfg)
 	stdOut := io.Writer(os.Stdout)
 	stdErr := io.Writer(os.Stderr)
@@ -111,14 +107,14 @@ func Run(ctx context.Context, cfg *config.Node) error {
 		return err
 	}
 
-	return preloadImages(ctx, cfg)
+	return PreloadImages(ctx, cfg)
 }
 
-// preloadImages reads the contents of the agent images directory, and attempts to
+// PreloadImages reads the contents of the agent images directory, and attempts to
 // import into containerd any files found there. Supported compressed types are decompressed, and
 // any .txt files are processed as a list of images that should be pre-pulled from remote registries.
 // If configured, imported images are retagged as being pulled from additional registries.
-func preloadImages(ctx context.Context, cfg *config.Node) error {
+func PreloadImages(ctx context.Context, cfg *config.Node) error {
 	fileInfo, err := os.Stat(cfg.Images)
 	if os.IsNotExist(err) {
 		return nil

pkg/agent/run.go

@@ -13,8 +13,6 @@ import (
 
 	systemd "github.com/coreos/go-systemd/daemon"
 	"github.com/k3s-io/k3s/pkg/agent/config"
-	"github.com/k3s-io/k3s/pkg/agent/containerd"
-	"github.com/k3s-io/k3s/pkg/agent/cridockerd"
 	"github.com/k3s-io/k3s/pkg/agent/flannel"
 	"github.com/k3s-io/k3s/pkg/agent/netpol"
 	"github.com/k3s-io/k3s/pkg/agent/proxy"
@@ -132,14 +130,11 @@ func run(ctx context.Context, cfg cmds.Agent, proxy proxy.Proxy) error {
 		}
 	}
 
-	if nodeConfig.Docker {
-		if err := cridockerd.Run(ctx, nodeConfig); err != nil {
-			return err
-		}
-	} else if nodeConfig.ContainerRuntimeEndpoint == "" {
-		if err := containerd.Run(ctx, nodeConfig); err != nil {
-			return err
-		}
+	notifySocket := os.Getenv("NOTIFY_SOCKET")
+	os.Unsetenv("NOTIFY_SOCKET")
+
+	if err := setupTunnelAndRunAgent(ctx, nodeConfig, cfg, proxy); err != nil {
+		return err
 	}
 
 	// the agent runtime is ready to host workloads when containerd is up and the airgap
@@ -150,13 +145,6 @@ func run(ctx context.Context, cfg cmds.Agent, proxy proxy.Proxy) error {
 		close(cfg.AgentReady)
 	}
 
-	notifySocket := os.Getenv("NOTIFY_SOCKET")
-	os.Unsetenv("NOTIFY_SOCKET")
-
-	if err := setupTunnelAndRunAgent(ctx, nodeConfig, cfg, proxy); err != nil {
-		return err
-	}
-
 	if err := util.WaitForAPIServerReady(ctx, nodeConfig.AgentConfig.KubeConfigKubelet, util.DefaultAPIServerReadyTimeout); err != nil {
 		return errors.Wrap(err, "failed to wait for apiserver ready")
 	}

pkg/daemons/agent/agent.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/k3s-io/k3s/pkg/agent/config"
+	"github.com/k3s-io/k3s/pkg/agent/containerd"
 	"github.com/k3s-io/k3s/pkg/agent/proxy"
 	daemonconfig "github.com/k3s-io/k3s/pkg/daemons/config"
 	"github.com/k3s-io/k3s/pkg/daemons/executor"
@@ -22,6 +23,17 @@ func Agent(ctx context.Context, nodeConfig *daemonconfig.Node, proxy proxy.Proxy
 	logsapi.ReapplyHandling = logsapi.ReapplyHandlingIgnoreUnchanged
 	logs.InitLogs()
 	defer logs.FlushLogs()
+
+	if nodeConfig.Docker {
+		if err := startDocker(ctx, nodeConfig); err != nil {
+			return err
+		}
+	} else if nodeConfig.ContainerRuntimeEndpoint == "" {
+		if err := startContainerd(ctx, nodeConfig); err != nil {
+			return err
+		}
+	}
+
 	if err := startKubelet(ctx, &nodeConfig.AgentConfig); err != nil {
 		return err
 	}
@@ -53,6 +65,17 @@ func startKubelet(ctx context.Context, cfg *daemonconfig.Agent) error {
 	return executor.Kubelet(ctx, args)
 }
 
+func startContainerd(ctx context.Context, cfg *daemonconfig.Node) error {
+	if err := containerd.SetupContainerdConfig(cfg); err != nil {
+		return err
+	}
+	return executor.Containerd(ctx, cfg)
+}
+
+func startDocker(ctx context.Context, cfg *daemonconfig.Node) error {
+	return executor.Docker(ctx, cfg)
+}
+
 // ImageCredProvAvailable checks to see if the kubelet image credential provider bin dir and config
 // files exist and are of the correct types. This is exported so that it may be used by downstream projects.
 func ImageCredProvAvailable(cfg *daemonconfig.Agent) bool {

pkg/daemons/executor/embed.go

@@ -13,6 +13,8 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/k3s-io/k3s/pkg/agent/containerd"
+	"github.com/k3s-io/k3s/pkg/agent/cridockerd"
 	"github.com/k3s-io/k3s/pkg/cli/cmds"
 	daemonconfig "github.com/k3s-io/k3s/pkg/daemons/config"
 	"github.com/k3s-io/k3s/pkg/util"
@@ -256,6 +258,14 @@ func (e *Embedded) CurrentETCDOptions() (InitialOptions, error) {
 	return InitialOptions{}, nil
 }
 
+func (e *Embedded) Containerd(ctx context.Context, cfg *daemonconfig.Node) error {
+	return containerd.Run(ctx, cfg)
+}
+
+func (e *Embedded) Docker(ctx context.Context, cfg *daemonconfig.Node) error {
+	return cridockerd.Run(ctx, cfg)
+}
+
 // waitForUntaintedNode watches nodes, waiting to find one not tainted as
 // uninitialized by the external cloud provider.
 func waitForUntaintedNode(ctx context.Context, kubeConfig string) error {

pkg/daemons/executor/executor.go

@@ -31,6 +31,8 @@ type Executor interface {
 	CurrentETCDOptions() (InitialOptions, error)
 	ETCD(ctx context.Context, args ETCDConfig, extraArgs []string) error
 	CloudControllerManager(ctx context.Context, ccmRBACReady <-chan struct{}, args []string) error
+	Containerd(ctx context.Context, node *daemonconfig.Node) error
+	Docker(ctx context.Context, node *daemonconfig.Node) error
 }
 
 type ETCDConfig struct {
@@ -169,3 +171,11 @@ func ETCD(ctx context.Context, args ETCDConfig, extraArgs []string) error {
 func CloudControllerManager(ctx context.Context, ccmRBACReady <-chan struct{}, args []string) error {
 	return executor.CloudControllerManager(ctx, ccmRBACReady, args)
 }
+
+func Containerd(ctx context.Context, config *daemonconfig.Node) error {
+	return executor.Containerd(ctx, config)
+}
+
+func Docker(ctx context.Context, config *daemonconfig.Node) error {
+	return executor.Docker(ctx, config)
+}