fix: move etcd to cri containerd runner

This fixes a problem when Talos pulls `etcd` image one every reboot, as `etcd` was running in the system containerd which is completely ephemeral (backed by `tmpfs`). Also skip pulling if image is already present and unpacked (same fix for the `kubelet` image). Fixes #3229 Signed-off-by: Andrey Smirnov <smirnov.andrey@gmail.com>
siderolabs · Mar 2, 2021 · 561f8aa · 561f8aa
1 parent 1d8ed9b
commit 561f8aa
Show file tree

Hide file tree

Showing 3 changed files with 46 additions and 10 deletions.
diff --git a/internal/app/machined/pkg/system/services/etcd.go b/internal/app/machined/pkg/system/services/etcd.go
@@ -70,7 +70,7 @@ func (e *Etcd) PreFunc(ctx context.Context, r runtime.Runtime) (err error) {
 		return fmt.Errorf("failed to generate etcd PKI: %w", err)
 	}
 
-	client, err := containerdapi.New(constants.SystemContainerdAddress)
+	client, err := containerdapi.New(constants.ContainerdAddress)
 	if err != nil {
 		return err
 	}
@@ -79,7 +79,9 @@ func (e *Etcd) PreFunc(ctx context.Context, r runtime.Runtime) (err error) {
 
 	// Pull the image and unpack it.
 	containerdctx := namespaces.WithNamespace(ctx, constants.SystemContainerdNamespace)
-	if _, err = image.Pull(containerdctx, r.Config().Machine().Registries(), client, r.Config().Cluster().Etcd().Image()); err != nil {
+
+	_, err = image.Pull(containerdctx, r.Config().Machine().Registries(), client, r.Config().Cluster().Etcd().Image(), image.WithSkipIfAlreadyPulled())
+	if err != nil {
 		return fmt.Errorf("failed to pull image %q: %w", r.Config().Cluster().Etcd().Image(), err)
 	}
 
@@ -114,10 +116,10 @@ func (e *Etcd) Condition(r runtime.Runtime) conditions.Condition {
 // DependsOn implements the Service interface.
 func (e *Etcd) DependsOn(r runtime.Runtime) []string {
 	if r.State().Platform().Mode() == runtime.ModeContainer || r.Config().Machine().Time().Disabled() {
-		return []string{"containerd", "networkd"}
+		return []string{"cri", "networkd"}
 	}
 
-	return []string{"containerd", "networkd", "timed"}
+	return []string{"cri", "networkd", "timed"}
 }
 
 // Runner implements the Service interface.
@@ -147,7 +149,7 @@ func (e *Etcd) Runner(r runtime.Runtime) (runner.Runner, error) {
 		&args,
 		runner.WithLoggingManager(r.Logging()),
 		runner.WithNamespace(constants.SystemContainerdNamespace),
-		runner.WithContainerdAddress(constants.SystemContainerdAddress),
+		runner.WithContainerImage(r.Config().Machine().Kubelet().Image()),
 		runner.WithContainerImage(r.Config().Cluster().Etcd().Image()),
 		runner.WithEnv(env),
 		runner.WithOCISpecOpts(

diff --git a/internal/app/machined/pkg/system/services/kubelet.go b/internal/app/machined/pkg/system/services/kubelet.go
@@ -19,7 +19,6 @@ import (
 	containerdapi "github.com/containerd/containerd"
 	"github.com/containerd/containerd/namespaces"
 	"github.com/containerd/containerd/oci"
-	criconstants "github.com/containerd/cri/pkg/constants"
 	cni "github.com/containerd/go-cni"
 	specs "github.com/opencontainers/runtime-spec/specs-go"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -65,6 +64,8 @@ func (k *Kubelet) ID(r runtime.Runtime) string {
 }
 
 // PreFunc implements the Service interface.
+//
+//nolint: gocyclo
 func (k *Kubelet) PreFunc(ctx context.Context, r runtime.Runtime) error {
 	cfg := struct {
 		Server               string
@@ -110,9 +111,9 @@ func (k *Kubelet) PreFunc(ctx context.Context, r runtime.Runtime) error {
 	defer client.Close()
 
 	// Pull the image and unpack it.
-	containerdctx := namespaces.WithNamespace(ctx, "k8s.io")
+	containerdctx := namespaces.WithNamespace(ctx, constants.SystemContainerdNamespace)
 
-	_, err = image.Pull(containerdctx, r.Config().Machine().Registries(), client, r.Config().Machine().Kubelet().Image())
+	_, err = image.Pull(containerdctx, r.Config().Machine().Registries(), client, r.Config().Machine().Kubelet().Image(), image.WithSkipIfAlreadyPulled())
 	if err != nil {
 		return err
 	}
@@ -182,7 +183,7 @@ func (k *Kubelet) Runner(r runtime.Runtime) (runner.Runner, error) {
 		r.Config().Debug(),
 		&args,
 		runner.WithLoggingManager(r.Logging()),
-		runner.WithNamespace(criconstants.K8sContainerdNamespace),
+		runner.WithNamespace(constants.SystemContainerdNamespace),
 		runner.WithContainerImage(r.Config().Machine().Kubelet().Image()),
 		runner.WithEnv(env),
 		runner.WithOCISpecOpts(

diff --git a/internal/pkg/containers/image/image.go b/internal/pkg/containers/image/image.go
@@ -32,9 +32,42 @@ const (
 	ImportRetryJitter   = time.Second
 )
 
+// PullOption is an option for Pull function.
+type PullOption func(*PullOptions)
+
+// PullOptions configure Pull function.
+type PullOptions struct {
+	SkipIfAlreadyPulled bool
+}
+
+// WithSkipIfAlreadyPulled skips pulling if image is already pulled and unpacked.
+func WithSkipIfAlreadyPulled() PullOption {
+	return func(opts *PullOptions) {
+		opts.SkipIfAlreadyPulled = true
+	}
+}
+
 // Pull is a convenience function that wraps the containerd image pull func with
 // retry functionality.
-func Pull(ctx context.Context, reg config.Registries, client *containerd.Client, ref string) (img containerd.Image, err error) {
+func Pull(ctx context.Context, reg config.Registries, client *containerd.Client, ref string, opt ...PullOption) (img containerd.Image, err error) {
+	var opts PullOptions
+
+	for _, o := range opt {
+		o(&opts)
+	}
+
+	if opts.SkipIfAlreadyPulled {
+		img, err = client.GetImage(ctx, ref)
+		if err == nil {
+			var unpacked bool
+
+			unpacked, err = img.IsUnpacked(ctx, "")
+			if err == nil && unpacked {
+				return img, nil
+			}
+		}
+	}
+
 	resolver := NewResolver(reg)
 
 	err = retry.Exponential(PullTimeout, retry.WithUnits(PullRetryInterval), retry.WithErrorLogging(true)).Retry(func() error {