feat: registry proxy

Implement container registry proxy.

Signed-off-by: Dmitriy Matrenichev <[email protected]>

.golangci.yml

@@ -124,7 +124,6 @@ linters:
     - gochecknoinits
     - gocognit
     - godox
-    - gomnd
     - gosec
     - ireturn # we return interfaces
     - maintidx

Dockerfile

@@ -664,6 +664,17 @@ EOF
 FROM scratch AS modules-arm64
 COPY --from=depmod-arm64 /build/lib/modules /lib/modules
 
+FROM build AS talos-image-cache
+ARG TARGETOS
+ARG TARGETARCH
+COPY --from=talosctl-targetarch /talosctl-${TARGETOS}-${TARGETARCH} /bin/talosctl
+RUN env HOME=/home/user TAG=latest /bin/talosctl images cache-create --images=ghcr.io/siderolabs/kubelet:v1.32.0-beta.0 --image-cache-path=kubelet1320.tar
+RUN env HOME=/home/user TAG=latest /bin/talosctl images cache-create --images=gcr.io/etcd-development/etcd:v3.5.17 --image-cache-path=etcd.tar
+RUN mkdir -pv /imagecache && ls *.tar | xargs -n1 tar -xvf
+RUN ls *.tar.gz | xargs -n1 tar -xvzf
+RUN mv manifests /imagecache/manifests
+RUN mv blob /imagecache/blob
+
 # The rootfs target provides the Talos rootfs.
 FROM build AS rootfs-base-amd64
 COPY --link --from=pkg-fhs / /rootfs
@@ -697,6 +708,7 @@ COPY --link --from=pkg-kmod-amd64 /usr/lib/libkmod.* /rootfs/lib/
 COPY --link --from=pkg-kmod-amd64 /usr/bin/kmod /rootfs/sbin/modprobe
 COPY --link --from=modules-amd64 /lib/modules /rootfs/lib/modules
 COPY --link --from=machined-build-amd64 /machined /rootfs/sbin/init
+COPY --link --from=talos-image-cache /imagecache /rootfs/imagecache
 
 # this is a no-op as it copies from a scratch image when WITH_DEBUG_SHELL is not set
 COPY --link --from=pkg-debug-tools-amd64 * /rootfs/
@@ -716,7 +728,7 @@ END
 COPY ./hack/cleanup.sh /toolchain/bin/cleanup.sh
 RUN <<END
     cleanup.sh /rootfs
-    mkdir -pv /rootfs/{boot/EFI,etc/cri/conf.d/hosts,lib/firmware,usr/etc,usr/local/share,usr/share/zoneinfo/Etc,mnt,system,opt,.extra}
+    mkdir -pv /rootfs/{boot/EFI,etc/cri/conf.d/hosts,lib/firmware,usr/etc,usr/local/share,usr/share/zoneinfo/Etc,mnt,system,opt,.extra,imagecache}
     mkdir -pv /rootfs/{etc/kubernetes/manifests,etc/cni/net.d,etc/ssl/certs,usr/libexec/kubernetes,/usr/local/lib/kubelet/credentialproviders}
     mkdir -pv /rootfs/opt/{containerd/bin,containerd/lib}
 END

cmd/talosctl/cmd/mgmt/cluster/create.go

@@ -514,12 +514,12 @@ func create(ctx context.Context) error {
 		}
 
 		for _, registryMirror := range registryMirrors {
-			components := strings.SplitN(registryMirror, "=", 2)
-			if len(components) != 2 {
+			left, right, ok := strings.Cut(registryMirror, "=")
+			if !ok {
 				return fmt.Errorf("invalid registry mirror spec: %q", registryMirror)
 			}
 
-			genOptions = append(genOptions, generate.WithRegistryMirror(components[0], components[1]))
+			genOptions = append(genOptions, generate.WithRegistryMirror(left, right))
 		}
 
 		for _, registryHost := range registryInsecure {

cmd/talosctl/cmd/mgmt/gen/config.go

@@ -187,12 +187,12 @@ func writeConfig(args []string) error {
 	var genOptions []generate.Option //nolint:prealloc
 
 	for _, registryMirror := range genConfigCmdFlags.registryMirrors {
-		components := strings.SplitN(registryMirror, "=", 2)
-		if len(components) != 2 {
+		left, right, ok := strings.Cut(registryMirror, "=")
+		if !ok {
 			return fmt.Errorf("invalid registry mirror spec: %q", registryMirror)
 		}
 
-		genOptions = append(genOptions, generate.WithRegistryMirror(components[0], components[1]))
+		genOptions = append(genOptions, generate.WithRegistryMirror(left, right))
 	}
 
 	if genConfigCmdFlags.talosVersion != "" {

cmd/talosctl/cmd/talos/image.go

@@ -5,7 +5,6 @@
 package talos
 
 import (
-	"bytes"
 	"context"
 	"fmt"
 	"io"
@@ -174,7 +173,7 @@ talosctl images default | talosctl images cache-create --image-cache-path=/tmp/c
 		}
 
 		if imageCacheCreateCmdFlags.images[0] == "-" {
-			var imagesListData bytes.Buffer
+			var imagesListData strings.Builder
 
 			if _, err := io.Copy(&imagesListData, os.Stdin); err != nil {
 				return fmt.Errorf("error reading from stdin: %w", err)

internal/app/machined/pkg/runtime/v1alpha1/v1alpha1_sequencer_tasks.go

@@ -425,6 +425,7 @@ func StartAllServices(runtime.Sequence, any) (runtime.TaskExecutionFunc, string)
 
 		serviceList := []system.Service{
 			&services.CRI{},
+			&services.Registryd{},
 		}
 
 		switch t := r.Config().Machine().Type(); t {

internal/app/machined/pkg/system/services/apid.go

@@ -53,7 +53,7 @@ func (o *APID) ID(r runtime.Runtime) string {
 }
 
 // apidResourceFilter filters access to COSI state for apid.
-func apidResourceFilter(ctx context.Context, access state.Access) error {
+func apidResourceFilter(_ context.Context, access state.Access) error {
 	if !access.Verb.Readonly() {
 		return errors.New("write access denied")
 	}
@@ -73,7 +73,7 @@ func apidResourceFilter(ctx context.Context, access state.Access) error {
 }
 
 // PreFunc implements the Service interface.
-func (o *APID) PreFunc(ctx context.Context, r runtime.Runtime) error {
+func (o *APID) PreFunc(_ context.Context, r runtime.Runtime) error {
 	// filter apid access to make sure apid can only access its certificates
 	resources := state.Filter(r.State().V1Alpha2().Resources(), apidResourceFilter)
 
@@ -117,7 +117,7 @@ func (o *APID) PreFunc(ctx context.Context, r runtime.Runtime) error {
 }
 
 // PostFunc implements the Service interface.
-func (o *APID) PostFunc(r runtime.Runtime, state events.ServiceState) (err error) {
+func (o *APID) PostFunc(runtime.Runtime, events.ServiceState) (err error) {
 	o.runtimeServer.Stop()
 
 	return os.RemoveAll(constants.APIRuntimeSocketPath)
@@ -129,7 +129,7 @@ func (o *APID) Condition(r runtime.Runtime) conditions.Condition {
 }
 
 // DependsOn implements the Service interface.
-func (o *APID) DependsOn(r runtime.Runtime) []string {
+func (o *APID) DependsOn(runtime.Runtime) []string {
 	return []string{"containerd"}
 }
 

internal/app/machined/pkg/system/services/auditd.go

@@ -26,27 +26,27 @@ var _ system.HealthcheckedService = (*Auditd)(nil)
 type Auditd struct{}
 
 // ID implements the Service interface.
-func (s *Auditd) ID(r runtime.Runtime) string {
+func (s *Auditd) ID(runtime.Runtime) string {
 	return auditdServiceID
 }
 
 // PreFunc implements the Service interface.
-func (s *Auditd) PreFunc(ctx context.Context, r runtime.Runtime) error {
+func (s *Auditd) PreFunc(context.Context, runtime.Runtime) error {
 	return nil
 }
 
 // PostFunc implements the Service interface.
-func (s *Auditd) PostFunc(r runtime.Runtime, state events.ServiceState) error {
+func (s *Auditd) PostFunc(runtime.Runtime, events.ServiceState) error {
 	return nil
 }
 
 // Condition implements the Service interface.
-func (s *Auditd) Condition(r runtime.Runtime) conditions.Condition {
+func (s *Auditd) Condition(runtime.Runtime) conditions.Condition {
 	return nil
 }
 
 // DependsOn implements the Service interface.
-func (s *Auditd) DependsOn(r runtime.Runtime) []string {
+func (s *Auditd) DependsOn(runtime.Runtime) []string {
 	return nil
 }
 
@@ -56,13 +56,13 @@ func (s *Auditd) Runner(r runtime.Runtime) (runner.Runner, error) {
 }
 
 // HealthFunc implements the HealthcheckedService interface.
-func (s *Auditd) HealthFunc(r runtime.Runtime) health.Check {
+func (s *Auditd) HealthFunc(runtime.Runtime) health.Check {
 	return func(ctx context.Context) error {
 		return nil
 	}
 }
 
 // HealthSettings implements the HealthcheckedService interface.
-func (s *Auditd) HealthSettings(r runtime.Runtime) *health.Settings {
+func (s *Auditd) HealthSettings(runtime.Runtime) *health.Settings {
 	return &health.DefaultSettings
 }

internal/app/machined/pkg/system/services/containerd.go

@@ -50,17 +50,17 @@ func (c *Containerd) Client() (*containerd.Client, error) {
 }
 
 // ID implements the Service interface.
-func (c *Containerd) ID(r runtime.Runtime) string {
+func (c *Containerd) ID(runtime.Runtime) string {
 	return "containerd"
 }
 
 // PreFunc implements the Service interface.
-func (c *Containerd) PreFunc(ctx context.Context, r runtime.Runtime) error {
+func (c *Containerd) PreFunc(context.Context, runtime.Runtime) error {
 	return nil
 }
 
 // PostFunc implements the Service interface.
-func (c *Containerd) PostFunc(r runtime.Runtime, state events.ServiceState) (err error) {
+func (c *Containerd) PostFunc(runtime.Runtime, events.ServiceState) (err error) {
 	if c.client != nil {
 		return c.client.Close()
 	}
@@ -69,12 +69,12 @@ func (c *Containerd) PostFunc(r runtime.Runtime, state events.ServiceState) (err
 }
 
 // Condition implements the Service interface.
-func (c *Containerd) Condition(r runtime.Runtime) conditions.Condition {
+func (c *Containerd) Condition(runtime.Runtime) conditions.Condition {
 	return nil
 }
 
 // DependsOn implements the Service interface.
-func (c *Containerd) DependsOn(r runtime.Runtime) []string {
+func (c *Containerd) DependsOn(runtime.Runtime) []string {
 	return nil
 }
 

internal/app/machined/pkg/system/services/cri.go

@@ -52,17 +52,17 @@ func (c *CRI) Client() (*containerd.Client, error) {
 }
 
 // ID implements the Service interface.
-func (c *CRI) ID(r runtime.Runtime) string {
+func (c *CRI) ID(runtime.Runtime) string {
 	return "cri"
 }
 
 // PreFunc implements the Service interface.
-func (c *CRI) PreFunc(ctx context.Context, r runtime.Runtime) error {
+func (c *CRI) PreFunc(context.Context, runtime.Runtime) error {
 	return os.MkdirAll(defaults.DefaultRootDir, os.ModeDir)
 }
 
 // PostFunc implements the Service interface.
-func (c *CRI) PostFunc(r runtime.Runtime, state events.ServiceState) (err error) {
+func (c *CRI) PostFunc(runtime.Runtime, events.ServiceState) (err error) {
 	if c.client != nil {
 		return c.client.Close()
 	}
@@ -76,7 +76,7 @@ func (c *CRI) Condition(r runtime.Runtime) conditions.Condition {
 }
 
 // DependsOn implements the Service interface.
-func (c *CRI) DependsOn(r runtime.Runtime) []string {
+func (c *CRI) DependsOn(runtime.Runtime) []string {
 	return nil
 }
 

internal/app/machined/pkg/system/services/etcd.go

@@ -23,11 +23,11 @@ import (
 	"github.com/cosi-project/runtime/pkg/resource"
 	"github.com/cosi-project/runtime/pkg/safe"
 	"github.com/cosi-project/runtime/pkg/state"
-	specs "github.com/opencontainers/runtime-spec/specs-go"
+	"github.com/opencontainers/runtime-spec/specs-go"
 	"github.com/siderolabs/gen/xslices"
 	"github.com/siderolabs/go-retry/retry"
 	clientv3 "go.etcd.io/etcd/client/v3"
-	snapshot "go.etcd.io/etcd/etcdutl/v3/snapshot"
+	"go.etcd.io/etcd/etcdutl/v3/snapshot"
 
 	"github.com/siderolabs/talos/internal/app/machined/pkg/runtime"
 	"github.com/siderolabs/talos/internal/app/machined/pkg/system"

internal/app/machined/pkg/system/services/kubelet.go

@@ -17,7 +17,7 @@ import (
 	"github.com/cosi-project/runtime/pkg/resource"
 	"github.com/cosi-project/runtime/pkg/safe"
 	"github.com/cosi-project/runtime/pkg/state"
-	specs "github.com/opencontainers/runtime-spec/specs-go"
+	"github.com/opencontainers/runtime-spec/specs-go"
 
 	"github.com/siderolabs/talos/internal/app/machined/pkg/runtime"
 	"github.com/siderolabs/talos/internal/app/machined/pkg/system"

internal/app/machined/pkg/system/services/machined.go

@@ -110,7 +110,7 @@ type machinedService struct {
 }
 
 // Main is an entrypoint to the API service.
-func (s *machinedService) Main(ctx context.Context, r runtime.Runtime, logWriter io.Writer) error {
+func (s *machinedService) Main(ctx context.Context, _ runtime.Runtime, logWriter io.Writer) error {
 	injector := &authz.Injector{
 		Mode: authz.MetadataOnly,
 	}
@@ -171,19 +171,26 @@ func (s *machinedService) Main(ctx context.Context, r runtime.Runtime, logWriter
 		return err
 	}
 
-	go func() {
-		//nolint:errcheck
-		server.Serve(listener)
-	}()
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
 
-	<-ctx.Done()
+	closed := make(chan struct{})
 
-	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer shutdownCancel()
+	context.AfterFunc(ctx, func() {
+		defer close(closed)
 
-	factory.ServerGracefulStop(server, shutdownCtx) //nolint:contextcheck
+		shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer shutdownCancel()
 
-	return nil
+		factory.ServerGracefulStop(server, shutdownCtx) //nolint:contextcheck
+	})
+
+	err = server.Serve(listener)
+
+	cancel()
+	<-closed
+
+	return err
 }
 
 var _ system.HealthcheckedService = (*Machined)(nil)
@@ -195,27 +202,27 @@ type Machined struct {
 }
 
 // ID implements the Service interface.
-func (m *Machined) ID(r runtime.Runtime) string {
+func (m *Machined) ID(runtime.Runtime) string {
 	return machinedServiceID
 }
 
 // PreFunc implements the Service interface.
-func (m *Machined) PreFunc(ctx context.Context, r runtime.Runtime) error {
+func (m *Machined) PreFunc(context.Context, runtime.Runtime) error {
 	return nil
 }
 
 // PostFunc implements the Service interface.
-func (m *Machined) PostFunc(r runtime.Runtime, state events.ServiceState) (err error) {
+func (m *Machined) PostFunc(runtime.Runtime, events.ServiceState) (err error) {
 	return nil
 }
 
 // Condition implements the Service interface.
-func (m *Machined) Condition(r runtime.Runtime) conditions.Condition {
+func (m *Machined) Condition(runtime.Runtime) conditions.Condition {
 	return nil
 }
 
 // DependsOn implements the Service interface.
-func (m *Machined) DependsOn(r runtime.Runtime) []string {
+func (m *Machined) DependsOn(runtime.Runtime) []string {
 	return nil
 }
 

internal/app/machined/pkg/system/services/registry/app/main.go

@@ -0,0 +1,41 @@
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"path/filepath"
+
+	"go.uber.org/zap"
+
+	"github.com/siderolabs/talos/internal/app/machined/pkg/system/services/registry"
+)
+
+func main() {
+	if err := app(); err != nil {
+		fmt.Fprintf(os.Stderr, "error: %v\n", err)
+		os.Exit(1)
+	}
+}
+
+func app() error {
+	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
+	defer cancel()
+
+	development, err := zap.NewDevelopment()
+	if err != nil {
+		return fmt.Errorf("failed to create development logger: %w", err)
+	}
+
+	homeDir, err := os.UserHomeDir()
+	if err != nil {
+		return fmt.Errorf("failed to get user home directory: %w", err)
+	}
+
+	return registry.NewService(filepath.Join(homeDir, "registry-cache"), development).Run(ctx)
+}