Merge pull request #5602 from twz123/win-containerd-endpoint

Use correct containerd endpoints on Windows

Author: twz123

cmd/ctr/ctr.go

@@ -18,9 +18,9 @@ package ctr
 
 import (
 	"os"
-	"path"
 
 	"github.com/k0sproject/k0s/internal/pkg/dir"
+	"github.com/k0sproject/k0s/pkg/component/worker"
 	"github.com/k0sproject/k0s/pkg/config"
 
 	"github.com/containerd/containerd/cmd/ctr/app"
@@ -58,7 +58,7 @@ func setDefaultValues(runDir string, flags []cli.Flag) {
 	for i, flag := range flags {
 		if f, ok := flag.(cli.StringFlag); ok {
 			if f.Name == "address, a" {
-				f.Value = path.Join(runDir, "containerd.sock")
+				f.Value = worker.GetContainerdAddress(runDir)
 				flags[i] = f
 			} else if f.Name == "namespace, n" {
 				f.Value = "k8s.io"

pkg/component/worker/cri_runtime.go

@@ -30,15 +30,27 @@ type RuntimeEndpoint = url.URL
 // Parses the CRI runtime flag and returns the parsed values.
 // If the flag is empty, provide k0s's defaults.
 func GetContainerRuntimeEndpoint(criSocketFlag, k0sRunDir string) (*RuntimeEndpoint, error) {
-	switch {
-	case criSocketFlag != "":
+	if criSocketFlag != "" {
 		return parseCRISocketFlag(criSocketFlag)
-	case runtime.GOOS == "windows":
-		return &url.URL{Scheme: "npipe", Path: "//./pipe/containerd-containerd"}, nil
-	default:
-		socketPath := filepath.Join(k0sRunDir, "containerd.sock")
-		return &url.URL{Scheme: "unix", Path: filepath.ToSlash(socketPath)}, nil
 	}
+
+	return getContainerRuntimeEndpoint(k0sRunDir), nil
+}
+
+func getContainerRuntimeEndpoint(k0sRunDir string) *RuntimeEndpoint {
+	if runtime.GOOS == "windows" {
+		return &url.URL{Scheme: "npipe", Path: "//./pipe/containerd-containerd"}
+	}
+
+	return &url.URL{Scheme: "unix", Path: filepath.ToSlash(GetContainerdAddress(k0sRunDir))}
+}
+
+func GetContainerdAddress(k0sRunDir string) string {
+	if runtime.GOOS == "windows" {
+		return getContainerRuntimeEndpoint(k0sRunDir).String()
+	}
+
+	return filepath.Join(k0sRunDir, "containerd.sock")
 }
 
 func parseCRISocketFlag(criSocketFlag string) (*RuntimeEndpoint, error) {

pkg/component/worker/ocibundle.go

@@ -49,12 +49,13 @@ const (
 
 // OCIBundleReconciler tries to import OCI bundle into the running containerd instance
 type OCIBundleReconciler struct {
-	k0sVars         *config.CfgVars
-	log             *logrus.Entry
-	alreadyImported map[string]time.Time
-	mtx             sync.Mutex
-	cancel          context.CancelFunc
-	end             chan struct{}
+	ociBundleDir      string
+	containerdAddress string
+	log               *logrus.Entry
+	alreadyImported   map[string]time.Time
+	mtx               sync.Mutex
+	cancel            context.CancelFunc
+	end               chan struct{}
 	*prober.EventEmitter
 }
 
@@ -63,25 +64,25 @@ var _ manager.Component = (*OCIBundleReconciler)(nil)
 // NewOCIBundleReconciler builds new reconciler
 func NewOCIBundleReconciler(vars *config.CfgVars) *OCIBundleReconciler {
 	return &OCIBundleReconciler{
-		k0sVars:         vars,
-		log:             logrus.WithField("component", "OCIBundleReconciler"),
-		EventEmitter:    prober.NewEventEmitter(),
-		alreadyImported: map[string]time.Time{},
-		end:             make(chan struct{}),
+		ociBundleDir:      vars.OCIBundleDir,
+		containerdAddress: GetContainerdAddress(vars.RunDir),
+		log:               logrus.WithField("component", "OCIBundleReconciler"),
+		EventEmitter:      prober.NewEventEmitter(),
+		alreadyImported:   map[string]time.Time{},
+		end:               make(chan struct{}),
 	}
 }
 
 func (a *OCIBundleReconciler) Init(_ context.Context) error {
-	return dir.Init(a.k0sVars.OCIBundleDir, constant.ManifestsDirMode)
+	return dir.Init(a.ociBundleDir, constant.ManifestsDirMode)
 }
 
 // containerdClient returns a connected containerd client.
 func (a *OCIBundleReconciler) containerdClient(ctx context.Context) (*containerd.Client, error) {
 	var client *containerd.Client
-	sock := filepath.Join(a.k0sVars.RunDir, "containerd.sock")
 	if err := retry.Do(func() (err error) {
 		client, err = containerd.New(
-			sock,
+			a.containerdAddress,
 			containerd.WithDefaultNamespace("k8s.io"),
 			containerd.WithDefaultPlatform(
 				platforms.Only(platforms.DefaultSpec()),
@@ -126,14 +127,14 @@ func (a *OCIBundleReconciler) loadAll(ctx context.Context) {
 	defer a.mtx.Unlock()
 
 	a.log.Info("Loading OCI bundles directory")
-	files, err := os.ReadDir(a.k0sVars.OCIBundleDir)
+	files, err := os.ReadDir(a.ociBundleDir)
 	if err != nil {
 		a.log.WithError(err).Errorf("Failed to read bundles directory")
 		return
 	}
 	a.EmitWithPayload("importing OCI bundles", files)
 	for _, file := range files {
-		fpath := filepath.Join(a.k0sVars.OCIBundleDir, file.Name())
+		fpath := filepath.Join(a.ociBundleDir, file.Name())
 		finfo, err := os.Stat(fpath)
 		if err != nil {
 			a.log.WithError(err).Errorf("failed to stat %s", fpath)
@@ -237,7 +238,7 @@ func (a *OCIBundleReconciler) installWatcher(ctx context.Context) error {
 		return fmt.Errorf("failed to create watcher: %w", err)
 	}
 
-	if err := watcher.Add(a.k0sVars.OCIBundleDir); err != nil {
+	if err := watcher.Add(a.ociBundleDir); err != nil {
 		return fmt.Errorf("failed to add watcher: %w", err)
 	}
 
@@ -261,7 +262,7 @@ func (a *OCIBundleReconciler) installWatcher(ctx context.Context) error {
 
 	go func() {
 		defer close(a.end)
-		a.log.Infof("Started to watch events on %s", a.k0sVars.OCIBundleDir)
+		a.log.Infof("Started to watch events on %s", a.ociBundleDir)
 		_ = debouncer.Run(ctx)
 		if err := watcher.Close(); err != nil {
 			a.log.Errorf("Failed to close watcher: %s", err)