Add embedded registry implementation

pkg/agent/config/config.go

@@ -16,6 +16,7 @@ import (
 	"os/exec"
 	"path/filepath"
 	"regexp"
+	"strconv"
 	"strings"
 	"time"
 
@@ -26,6 +27,7 @@ import (
 	"github.com/k3s-io/k3s/pkg/clientaccess"
 	"github.com/k3s-io/k3s/pkg/daemons/config"
 	"github.com/k3s-io/k3s/pkg/daemons/control/deps"
+	"github.com/k3s-io/k3s/pkg/spegel"
 	"github.com/k3s-io/k3s/pkg/util"
 	"github.com/k3s-io/k3s/pkg/version"
 	"github.com/k3s-io/k3s/pkg/vpn"
@@ -682,6 +684,29 @@ func get(ctx context.Context, envInfo *cmds.Agent, proxy proxy.Proxy) (*config.N
 	}
 	nodeConfig.AgentConfig.Registry = privRegistries.Registry
 
+	if nodeConfig.EmbeddedRegistry {
+		psk, err := hex.DecodeString(controlConfig.IPSECPSK)
+		if err != nil {
+			return nil, err
+		}
+		if len(psk) < 32 {
+			return nil, errors.New("insufficient PSK bytes")
+		}
+
+		conf := spegel.DefaultRegistry
+		conf.ExternalAddress = nodeConfig.AgentConfig.NodeIP
+		conf.InternalAddress = controlConfig.Loopback(false)
+		conf.RegistryPort = strconv.Itoa(controlConfig.SupervisorPort)
+		conf.ClientCAFile = clientCAFile
+		conf.ClientCertFile = clientK3sControllerCert
+		conf.ClientKeyFile = clientK3sControllerKey
+		conf.ServerCAFile = serverCAFile
+		conf.ServerCertFile = servingKubeletCert
+		conf.ServerKeyFile = servingKubeletKey
+		conf.PSK = psk[:32]
+		conf.InjectMirror(nodeConfig)
+	}
+
 	if err := validateNetworkConfig(nodeConfig); err != nil {
 		return nil, err
 	}

pkg/agent/containerd/config.go

 package containerd
 
 import (
+	"net"
 	"net/url"
 	"os"
 	"path/filepath"
@@ -10,6 +11,7 @@ import (
 	"github.com/k3s-io/k3s/pkg/agent/templates"
 	util2 "github.com/k3s-io/k3s/pkg/agent/util"
 	"github.com/k3s-io/k3s/pkg/daemons/config"
+	"github.com/k3s-io/k3s/pkg/spegel"
 	"github.com/k3s-io/k3s/pkg/version"
 	"github.com/sirupsen/logrus"
 )
@@ -36,6 +38,7 @@ func writeContainerdConfig(cfg *config.Node, containerdConfig templates.Containe
 
 // writeContainerdHosts merges registry mirrors/configs, and renders and saves hosts.toml from the filled template
 func writeContainerdHosts(cfg *config.Node, containerdConfig templates.ContainerdConfig) error {
+	mirrorAddr := net.JoinHostPort(spegel.DefaultRegistry.InternalAddress, spegel.DefaultRegistry.RegistryPort)
 	registry := containerdConfig.PrivateRegistryConfig
 	hosts := map[string]templates.HostConfig{}
 
@@ -57,12 +60,17 @@ func writeContainerdHosts(cfg *config.Node, containerdConfig templates.Container
 		// structure, which is defined in rancher/wharfie.
 		for _, endpoint := range mirror.Endpoints {
 			if endpointURL, err := url.Parse(endpoint); err == nil {
-				config.Endpoints = append(config.Endpoints, templates.RegistryEndpoint{
+				re := templates.RegistryEndpoint{
 					OverridePath: endpointURL.Path != "" && endpointURL.Path != "/" && !strings.HasSuffix(endpointURL.Path, "/v2"),
 					Config:       registry.Configs[endpointURL.Host],
 					Rewrites:     mirror.Rewrites,
 					URI:          endpoint,
-				})
+				}
+				// Do not apply rewrites to the embedded registry endpoint
+				if endpointURL.Host == mirrorAddr {
+					re.Rewrites = nil
+				}
+				config.Endpoints = append(config.Endpoints, re)
 			}
 		}
 		hosts[host] = config

pkg/agent/run.go

@@ -29,6 +29,7 @@ import (
 	"github.com/k3s-io/k3s/pkg/daemons/executor"
 	"github.com/k3s-io/k3s/pkg/nodeconfig"
 	"github.com/k3s-io/k3s/pkg/rootless"
+	"github.com/k3s-io/k3s/pkg/spegel"
 	"github.com/k3s-io/k3s/pkg/util"
 	"github.com/k3s-io/k3s/pkg/version"
 	"github.com/pkg/errors"
@@ -102,6 +103,16 @@ func run(ctx context.Context, cfg cmds.Agent, proxy proxy.Proxy) error {
 	nodeConfig.AgentConfig.EnableIPv4 = enableIPv4
 	nodeConfig.AgentConfig.EnableIPv6 = enableIPv6
 
+	if nodeConfig.EmbeddedRegistry {
+		if nodeConfig.Docker || nodeConfig.ContainerRuntimeEndpoint != "" {
+			return errors.New("embedded registry mirror requires embedded containerd")
+		}
+
+		if err := spegel.DefaultRegistry.Start(ctx, nodeConfig); err != nil {
+			return errors.Wrap(err, "failed to start embedded registry")
+		}
+	}
+
 	if err := setupCriCtlConfig(cfg, nodeConfig); err != nil {
 		return err
 	}

pkg/cli/agent/agent.go

@@ -2,21 +2,27 @@ package agent
 
 import (
 	"crypto/tls"
+	"errors"
 	"fmt"
+	"net/http"
 	"os"
 	"path/filepath"
 	"runtime"
 
 	"github.com/erikdubbelboer/gspt"
+	"github.com/gorilla/mux"
 	"github.com/k3s-io/k3s/pkg/agent"
+	"github.com/k3s-io/k3s/pkg/authenticator"
 	"github.com/k3s-io/k3s/pkg/cli/cmds"
 	"github.com/k3s-io/k3s/pkg/datadir"
+	"github.com/k3s-io/k3s/pkg/spegel"
 	"github.com/k3s-io/k3s/pkg/util"
 	"github.com/k3s-io/k3s/pkg/version"
 	"github.com/k3s-io/k3s/pkg/vpn"
 	"github.com/rancher/wrangler/pkg/signals"
 	"github.com/sirupsen/logrus"
 	"github.com/urfave/cli"
+	apiauth "k8s.io/apiserver/pkg/authentication/authenticator"
 )
 
 func Run(ctx *cli.Context) error {
@@ -99,5 +105,37 @@ func Run(ctx *cli.Context) error {
 		}
 	}
 
+	// Until the agent is run and retrieves config from the server, we won't know
+	// if the embedded registry is enabled. If it is not enabled, these are not
+	// used as the registry is never started.
+	conf := spegel.DefaultRegistry
+	conf.Bootstrapper = spegel.NewAgentBootstrapper(cfg.ServerURL, cfg.Token, cfg.DataDir)
+	conf.HandlerFunc = func(conf *spegel.Config, router *mux.Router) error {
+		// Create and bind a new authenticator using the configured client CA
+		authArgs := []string{"--client-ca-file=" + conf.ClientCAFile}
+		auth, err := authenticator.FromArgs(authArgs)
+		if err != nil {
+			return err
+		}
+		conf.AuthFunc = func() apiauth.Request {
+			return auth
+		}
+
+		// Create a new server and listen on the configured port
+		server := &http.Server{
+			Handler: router,
+			Addr:    ":" + conf.RegistryPort,
+			TLSConfig: &tls.Config{
+				ClientAuth: tls.RequestClientCert,
+			},
+		}
+		go func() {
+			if err := server.ListenAndServeTLS(conf.ServerCertFile, conf.ServerKeyFile); err != nil && !errors.Is(err, http.ErrServerClosed) {
+				logrus.Fatalf("registry server failed: %v", err)
+			}
+		}()
+		return nil
+	}
+
 	return agent.Run(contextCtx, cfg)
 }

pkg/cli/server/server.go

@@ -11,6 +11,7 @@ import (
 
 	systemd "github.com/coreos/go-systemd/daemon"
 	"github.com/erikdubbelboer/gspt"
+	"github.com/gorilla/mux"
 	"github.com/k3s-io/k3s/pkg/agent"
 	"github.com/k3s-io/k3s/pkg/agent/loadbalancer"
 	"github.com/k3s-io/k3s/pkg/cli/cmds"
@@ -20,6 +21,7 @@ import (
 	"github.com/k3s-io/k3s/pkg/etcd"
 	"github.com/k3s-io/k3s/pkg/rootless"
 	"github.com/k3s-io/k3s/pkg/server"
+	"github.com/k3s-io/k3s/pkg/spegel"
 	"github.com/k3s-io/k3s/pkg/util"
 	"github.com/k3s-io/k3s/pkg/version"
 	"github.com/k3s-io/k3s/pkg/vpn"
@@ -28,6 +30,7 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/urfave/cli"
 	utilnet "k8s.io/apimachinery/pkg/util/net"
+	"k8s.io/apiserver/pkg/authentication/authenticator"
 	kubeapiserverflag "k8s.io/component-base/cli/flag"
 	"k8s.io/kubernetes/pkg/controlplane/apiserver/options"
 	utilsnet "k8s.io/utils/net"
@@ -548,6 +551,23 @@ func run(app *cli.Context, cfg *cmds.Server, leaderControllers server.CustomCont
 		return agent.RunStandalone(ctx, agentConfig)
 	}
 
+	if cfg.EmbeddedRegistry {
+		conf := spegel.DefaultRegistry
+		conf.Bootstrapper = spegel.NewChainingBootstrapper(
+			spegel.NewServerBootstrapper(&serverConfig.ControlConfig),
+			spegel.NewAgentBootstrapper(cfg.ServerURL, token, agentConfig.DataDir),
+			spegel.NewSelfBootstrapper(),
+		)
+		conf.HandlerFunc = func(_ *spegel.Config, router *mux.Router) error {
+			router.NotFoundHandler = serverConfig.ControlConfig.Runtime.Handler
+			serverConfig.ControlConfig.Runtime.Handler = router
+			return nil
+		}
+		conf.AuthFunc = func() authenticator.Request {
+			return serverConfig.ControlConfig.Runtime.Authenticator
+		}
+	}
+
 	return agent.Run(ctx, agentConfig)
 }
 

pkg/daemons/config/types.go

@@ -152,6 +152,7 @@ type CriticalControlArgs struct {
 	DisableNPC            bool         `cli:"disable-network-policy"`
 	DisableServiceLB      bool         `cli:"disable-service-lb"`
 	EncryptSecrets        bool         `cli:"secrets-encryption"`
+	EmbeddedRegistry      bool         `cli:"embedded-registry"`
 	FlannelBackend        string       `cli:"flannel-backend"`
 	FlannelIPv6Masq       bool         `cli:"flannel-ipv6-masq"`
 	FlannelExternalIP     bool         `cli:"flannel-external-ip"`
@@ -186,7 +187,6 @@ type Control struct {
 	DisableKubeProxy         bool
 	DisableScheduler         bool
 	DisableServiceLB         bool
-	EmbeddedRegistry         bool
 	Rootless                 bool
 	ServiceLBNamespace       string
 	EnablePProf              bool

pkg/spegel/bootstrap.go

+package spegel
+
+import (
+	"context"
+	"math/rand"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/k3s-io/k3s/pkg/clientaccess"
+	"github.com/k3s-io/k3s/pkg/daemons/config"
+	"github.com/k3s-io/k3s/pkg/version"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/pkg/errors"
+	"github.com/rancher/wrangler/pkg/merr"
+	"github.com/sirupsen/logrus"
+	"github.com/xenitab/spegel/pkg/routing"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/util/wait"
+	nodeutil "k8s.io/kubernetes/pkg/controller/util/node"
+)
+
+// explicit interface checks
+var _ routing.Bootstrapper = &selfBootstrapper{}
+var _ routing.Bootstrapper = &agentBootstrapper{}
+var _ routing.Bootstrapper = &serverBootstrapper{}
+var _ routing.Bootstrapper = &chainingBootstrapper{}
+
+type selfBootstrapper struct {
+	id string
+}
+
+// NewSelfBootstrapper returns a stub p2p bootstrapper that just returns its own ID
+func NewSelfBootstrapper() routing.Bootstrapper {
+	return &selfBootstrapper{}
+}
+
+func (s *selfBootstrapper) Run(_ context.Context, id string) error {
+	s.id = id
+	return nil
+}
+
+func (s *selfBootstrapper) GetAddress() (*peer.AddrInfo, error) {
+	return peer.AddrInfoFromString(s.id)
+}
+
+type agentBootstrapper struct {
+	server     string
+	token      string
+	clientCert string
+	clientKey  string
+}
+
+// NewAgentBootstrapper returns a p2p bootstrapper that retrieves a peer address from its server
+func NewAgentBootstrapper(server, token, dataDir string) routing.Bootstrapper {
+	return &agentBootstrapper{
+		clientCert: filepath.Join(dataDir, "agent", "client-kubelet.crt"),
+		clientKey:  filepath.Join(dataDir, "agent", "client-kubelet.key"),
+		server:     server,
+		token:      token,
+	}
+}
+
+func (c *agentBootstrapper) Run(_ context.Context, _ string) error {
+	return nil
+}
+
+func (c *agentBootstrapper) GetAddress() (*peer.AddrInfo, error) {
+	if c.server == "" || c.token == "" {
+		return nil, errors.New("cannot get addresses without server and token")
+	}
+
+	withCert := clientaccess.WithClientCertificate(c.clientCert, c.clientKey)
+	info, err := clientaccess.ParseAndValidateToken(c.server, c.token, withCert)
+	if err != nil {
+		return nil, err
+	}
+
+	addr, err := info.Get("/v1-" + version.Program + "/p2p")
+	if err != nil {
+		return nil, err
+	}
+
+	addrInfo, err := peer.AddrInfoFromString(string(addr))
+	return addrInfo, err
+}
+
+type serverBootstrapper struct {
+	controlConfig *config.Control
+}
+
+// NewServerBootstrapper returns a p2p bootstrapper that returns an address from a random other cluster member.
+func NewServerBootstrapper(controlConfig *config.Control) routing.Bootstrapper {
+	return &serverBootstrapper{
+		controlConfig: controlConfig,
+	}
+}
+
+func (s *serverBootstrapper) Run(_ context.Context, id string) error {
+	s.controlConfig.Runtime.ClusterControllerStarts["spegel-p2p"] = func(ctx context.Context) {
+		nodes := s.controlConfig.Runtime.Core.Core().V1().Node()
+		wait.PollImmediateUntilWithContext(ctx, 1*time.Second, func(ctx context.Context) (bool, error) {
+			nodeName := os.Getenv("NODE_NAME")
+			if nodeName == "" {
+				return false, nil
+			}
+			node, err := nodes.Get(nodeName, metav1.GetOptions{})
+			if err != nil {
+				return false, nil
+			}
+
+			if node.Annotations == nil {
+				node.Annotations = map[string]string{}
+			}
+			node.Annotations[P2pAddressAnnotation] = id
+			if node.Labels == nil {
+				node.Labels = map[string]string{}
+			}
+			node.Labels[P2pEnabledLabel] = "true"
+
+			if _, err = nodes.Update(node); err != nil {
+				return false, nil
+			}
+			logrus.Infof("Node P2P address annotations and labels added: %s", id)
+			return true, nil
+		})
+	}
+	return nil
+}
+
+func (s *serverBootstrapper) GetAddress() (addrInfo *peer.AddrInfo, err error) {
+	if s.controlConfig.Runtime.Core == nil {
+		return nil, errors.New("runtime core not ready")
+	}
+	nodeName := os.Getenv("NODE_NAME")
+	if nodeName == "" {
+		return nil, errors.New("node name not set")
+	}
+	nodes := s.controlConfig.Runtime.Core.Core().V1().Node()
+	labelSelector := labels.Set{P2pEnabledLabel: "true"}.String()
+	nodeList, err := nodes.List(metav1.ListOptions{LabelSelector: labelSelector})
+	if err != nil {
+		return nil, err
+	}
+	for _, i := range rand.Perm(len(nodeList.Items)) {
+		node := nodeList.Items[i]
+		if node.Name == nodeName {
+			// don't return our own address
+			continue
+		}
+		if find, condition := nodeutil.GetNodeCondition(&node.Status, v1.NodeReady); find == -1 || condition.Status != v1.ConditionTrue {
+			// don't return the address of a not-ready node
+			continue
+		}
+		if val, ok := node.Annotations[P2pAddressAnnotation]; ok {
+			for _, addr := range strings.Split(val, ",") {
+				if info, err := peer.AddrInfoFromString(addr); err == nil {
+					return info, nil
+				}
+			}
+		}
+	}
+	return nil, errors.New("no ready p2p peers found")
+}
+
+type chainingBootstrapper struct {
+	bootstrappers []routing.Bootstrapper
+}
+
+// NewChainingBootstrapper returns a p2p bootstrapper that passes through to a list of bootstrappers.
+func NewChainingBootstrapper(bootstrappers ...routing.Bootstrapper) routing.Bootstrapper {
+	return &chainingBootstrapper{
+		bootstrappers: bootstrappers,
+	}
+}
+
+func (c *chainingBootstrapper) Run(ctx context.Context, id string) error {
+	errs := merr.Errors{}
+	for _, b := range c.bootstrappers {
+		if err := b.Run(ctx, id); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return merr.NewErrors(errs...)
+}
+
+func (c *chainingBootstrapper) GetAddress() (*peer.AddrInfo, error) {
+	errs := merr.Errors{}
+	for _, b := range c.bootstrappers {
+		addr, err := b.GetAddress()
+		if err == nil {
+			return addr, nil
+		}
+		errs = append(errs, err)
+	}
+	return nil, merr.NewErrors(errs...)
+}

pkg/spegel/registry.go

+package spegel
+
+import (
+	"net"
+
+	"github.com/k3s-io/k3s/pkg/daemons/config"
+	"github.com/rancher/wharfie/pkg/registries"
+)
+
+// InjectMirror configures TLS for the registry mirror client, and  adds the mirror address as an endpoint
+// to all configured registries.
+func (c *Config) InjectMirror(nodeConfig *config.Node) error {
+	mirrorAddr := net.JoinHostPort(c.InternalAddress, c.RegistryPort)
+	registry := nodeConfig.AgentConfig.Registry
+
+	if registry.Configs == nil {
+		registry.Configs = map[string]registries.RegistryConfig{}
+	}
+	registry.Configs[mirrorAddr] = registries.RegistryConfig{
+		TLS: &registries.TLSConfig{
+			CAFile:   c.ServerCAFile,
+			CertFile: c.ClientCertFile,
+			KeyFile:  c.ClientKeyFile,
+		},
+	}
+
+	if registry.Mirrors == nil {
+		registry.Mirrors = map[string]registries.Mirror{}
+	}
+	for host, mirror := range registry.Mirrors {
+		if host != "*" {
+			mirror.Endpoints = append([]string{"https://" + mirrorAddr}, mirror.Endpoints...)
+			registry.Mirrors[host] = mirror
+		}
+	}
+
+	return nil
+}

pkg/spegel/spegel.go

+package spegel
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strconv"
+	"time"
+
+	"github.com/k3s-io/k3s/pkg/clientaccess"
+	"github.com/k3s-io/k3s/pkg/daemons/config"
+	"github.com/k3s-io/k3s/pkg/version"
+	"github.com/rancher/dynamiclistener/cert"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apiserver/pkg/authentication/authenticator"
+	"k8s.io/apiserver/pkg/authentication/request/union"
+	"k8s.io/utils/pointer"
+
+	"github.com/go-logr/logr"
+	"github.com/go-logr/stdr"
+	"github.com/gorilla/mux"
+	leveldb "github.com/ipfs/go-ds-leveldb"
+	ipfslog "github.com/ipfs/go-log/v2"
+	"github.com/libp2p/go-libp2p"
+	"github.com/libp2p/go-libp2p/core/crypto"
+	"github.com/libp2p/go-libp2p/p2p/host/peerstore/pstoreds"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	"github.com/xenitab/spegel/pkg/metrics"
+	"github.com/xenitab/spegel/pkg/oci"
+	"github.com/xenitab/spegel/pkg/registry"
+	"github.com/xenitab/spegel/pkg/routing"
+	"github.com/xenitab/spegel/pkg/state"
+	"k8s.io/component-base/metrics/legacyregistry"
+)
+
+// DefaultRegistry is the default instance of a Spegel distributed registry
+var DefaultRegistry = &Config{
+	Bootstrapper: NewSelfBootstrapper(),
+	HandlerFunc: func(_ *Config, _ *mux.Router) error {
+		return errors.New("not implemented")
+	},
+	AuthFunc: func() authenticator.Request {
+		return union.New(nil)
+	},
+}
+
+var (
+	P2pAddressAnnotation = "p2p." + version.Program + ".cattle.io/node-address"
+	P2pEnabledLabel      = "p2p." + version.Program + ".cattle.io/enabled"
+	P2pPortEnv           = version.ProgramUpper + "_P2P_PORT"
+)
+
+type authFunc func() authenticator.Request
+type handlerFunc func(config *Config, router *mux.Router) error
+
+// Config holds fields for a distributed registry
+type Config struct {
+	ClientCAFile   string
+	ClientCertFile string
+	ClientKeyFile  string
+
+	ServerCAFile   string
+	ServerCertFile string
+	ServerKeyFile  string
+
+	// ExternalAddress is the address for other nodes to connect to the registry API.
+	ExternalAddress string
+
+	// InternalAddress is the address for the local containerd instance to connect to the registry API.
+	InternalAddress string
+
+	// RegistryPort is the port for the registry API.
+	RegistryPort string
+
+	// PSK is the preshared key required to join the p2p network.
+	PSK []byte
+
+	// Bootstrapper is the bootstrapper that will be used to discover p2p peers.
+	Bootstrapper routing.Bootstrapper
+
+	// HandlerFunc will be called to add the registry API handler to an existing router.
+	HandlerFunc handlerFunc
+
+	// Authenticator will be called to retrieve an authenticator used to validate the request to the registry API.
+	AuthFunc authFunc
+}
+
+// These values are not currently configurable
+const (
+	resolveLatestTag  = false
+	resolveRetries    = 0
+	resolveTimeout    = time.Second * 5
+	registryScheme    = "https"
+	registryNamespace = "k8s.io"
+	defaultRouterPort = "5001"
+)
+
+func init() {
+	// ensure that spegel exposes metrics through the same registry used by Kubernetes components
+	metrics.DefaultRegisterer = legacyregistry.Registerer()
+	metrics.DefaultGatherer = legacyregistry.DefaultGatherer
+}
+
+// Start starts the embedded p2p router, and binds the registry API to an existing HTTP router.
+func (c *Config) Start(ctx context.Context, nodeConfig *config.Node) error {
+	// distribute images for all configured mirrors. there doesn't need to be a
+	// configured endpoint, just having a key for the registry will do.
+	urls := []url.URL{}
+	registries := []string{}
+	for host := range nodeConfig.AgentConfig.Registry.Mirrors {
+		if u, err := url.Parse("https://" + host); err != nil || host == "*" {
+			logrus.Errorf("Distributed registry mirror skipping unsupported registry: %s", host)
+		} else {
+			urls = append(urls, *u)
+			registries = append(registries, host)
+		}
+	}
+
+	if len(registries) == 0 {
+		logrus.Errorf("Not starting distributed registry mirror: no registries configured for distributed mirroring")
+		return nil
+	}
+
+	logrus.Infof("Starting distributed registry mirror at https://%s:%s/v2 for registries %v",
+		c.ExternalAddress, c.RegistryPort, registries)
+
+	// set up the various logging logging frameworks
+	level := ipfslog.LevelInfo
+	if logrus.IsLevelEnabled(logrus.DebugLevel) {
+		level = ipfslog.LevelDebug
+		stdlog := log.New(logrus.StandardLogger().Writer(), "spegel ", log.LstdFlags)
+		logger := stdr.NewWithOptions(stdlog, stdr.Options{Verbosity: pointer.Int(10)})
+		ctx = logr.NewContext(ctx, logger)
+	}
+	ipfslog.SetAllLoggers(level)
+
+	// Get containerd client
+	ociClient, err := oci.NewContainerd(nodeConfig.Containerd.Address, registryNamespace, nodeConfig.Containerd.Registry, urls)
+	if err != nil {
+		return errors.Wrap(err, "failed to create OCI client")
+	}
+
+	// create or load persistent private key
+	keyFile := filepath.Join(nodeConfig.Containerd.Opt, "peer.key")
+	keyBytes, _, err := cert.LoadOrGenerateKeyFile(keyFile, false)
+	if err != nil {
+		return errors.Wrap(err, "failed to load or generate p2p private key")
+	}
+	privKey, err := cert.ParsePrivateKeyPEM(keyBytes)
+	if err != nil {
+		return errors.Wrap(err, "failed to parse p2p private key")
+	}
+	p2pKey, _, err := crypto.KeyPairFromStdKey(privKey)
+	if err != nil {
+		return errors.Wrap(err, "failed to convert p2p private key")
+	}
+
+	// create a peerstore to allow persisting nodes across restarts
+	peerFile := filepath.Join(nodeConfig.Containerd.Opt, "peerstore.db")
+	ds, err := leveldb.NewDatastore(peerFile, nil)
+	if err != nil {
+		return errors.Wrap(err, "failed to create peerstore datastore")
+	}
+	ps, err := pstoreds.NewPeerstore(ctx, ds, pstoreds.DefaultOpts())
+	if err != nil {
+		return errors.Wrap(err, "failed to create peerstore")
+	}
+
+	// get port and start p2p router
+	routerPort := defaultRouterPort
+	if env := os.Getenv(P2pPortEnv); env != "" {
+		if i, err := strconv.Atoi(env); i == 0 || err != nil {
+			logrus.Warnf("Invalid P2P node port; using default")
+		} else {
+			routerPort = env
+		}
+	}
+	routerAddr := net.JoinHostPort(c.ExternalAddress, routerPort)
+
+	logrus.Infof("Starting distributed registry P2P node at %s", routerAddr)
+	opts := []libp2p.Option{
+		libp2p.Identity(p2pKey),
+		libp2p.Peerstore(ps),
+		libp2p.PrivateNetwork(c.PSK),
+		libp2p.PrometheusRegisterer(metrics.DefaultRegisterer),
+	}
+	router, err := routing.NewP2PRouter(ctx, routerAddr, c.Bootstrapper, c.RegistryPort, registryScheme, opts...)
+	if err != nil {
+		return errors.Wrap(err, "failed to create p2p router")
+	}
+
+	caCert, err := os.ReadFile(c.ServerCAFile)
+	if err != nil {
+		return errors.Wrap(err, "failed to read server CA")
+	}
+	localAddr := net.JoinHostPort(c.InternalAddress, c.RegistryPort)
+	client := clientaccess.GetHTTPClient(caCert, c.ClientCertFile, c.ClientKeyFile)
+	metrics.Register()
+	reg := registry.NewRegistry(ociClient, router, localAddr, resolveRetries, resolveTimeout, resolveLatestTag, client.Transport)
+	regSvr := reg.Server(":"+c.RegistryPort, logr.FromContextOrDiscard(ctx))
+
+	// Close router on shutdown
+	go func() {
+		<-ctx.Done()
+		router.Close()
+	}()
+
+	// Track images available in containerd and publish via p2p router
+	go state.Track(ctx, ociClient, router, resolveLatestTag)
+
+	mRouter := mux.NewRouter().SkipClean(true)
+	mRouter.Use(c.authMiddleware())
+	mRouter.PathPrefix("/v2").Handler(regSvr.Handler)
+	mRouter.PathPrefix("/v1-" + version.Program + "/p2p").Handler(c.peerInfo())
+	if err := c.HandlerFunc(c, mRouter); err != nil {
+		return err
+	}
+
+	// Wait up to 5 seconds for the p2p network to find peers. This will return
+	// immediately if the node is bootstrapping from itself.
+	wait.PollImmediateWithContext(ctx, time.Second, resolveTimeout, func(_ context.Context) (bool, error) {
+		return router.HasMirrors()
+	})
+
+	return nil
+}
+
+// peerInfo sends a peer address retrieved from the bootstrapper via HTTP
+func (c *Config) peerInfo() http.HandlerFunc {
+	return http.HandlerFunc(func(resp http.ResponseWriter, req *http.Request) {
+		client, _, _ := net.SplitHostPort(req.RemoteAddr)
+		info, err := c.Bootstrapper.GetAddress()
+		if err != nil {
+			http.Error(resp, "Internal Error", http.StatusInternalServerError)
+			return
+		}
+		logrus.Debugf("Serving p2p peer addr %s to client at %s", info, client)
+		resp.WriteHeader(http.StatusOK)
+		resp.Header().Set("Content-Type", "text/plain")
+		fmt.Fprintf(resp, "%s/p2p/%s", info.Addrs[0].String(), info.ID.Pretty())
+	})
+}
+
+// authMiddleware calls the configured authenticator to gate access to the registry API
+func (c *Config) authMiddleware() mux.MiddlewareFunc {
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(resp http.ResponseWriter, req *http.Request) {
+			if _, ok, err := c.AuthFunc().AuthenticateRequest(req); !ok || err != nil {
+				http.Error(resp, "Unauthorized", http.StatusUnauthorized)
+				return
+			}
+			next.ServeHTTP(resp, req)
+		})
+	}
+}