MIR-1478: Keep the OCI registry on the internal WireGuard network

api/runner/rpc.yml

@@ -163,16 +163,19 @@ interfaces:
         index: 6
         doc: |
           Report whether the coordinator has a workload identity issuer
-          configured, and its issuer URL. Distributed runners call this once at
-          startup to decide whether to mint workload identity tokens via the
-          coordinator.
+          configured, its issuer URL, and its internal WireGuard address.
+          Distributed runners call this once at startup to configure identity
+          tokens and registry pulls.
         results:
           - name: enabled
             type: bool
             doc: Whether a workload identity issuer is configured on the coordinator
           - name: issuer_url
             type: string
             doc: The issuer URL (iss claim anchor) when enabled
+          - name: coordinator_internal_ip
+            type: string
+            doc: Current coordinator bridge gateway reachable over WireGuard
 
       - name: IssueWorkloadToken
         index: 7

api/runner/runner_v1alpha/rpc.gen.go

@@ -967,8 +967,9 @@ func (v *RunnerRegistrationWorkloadIssuerInfoArgs) UnmarshalJSON(data []byte) er
 }
 
 type runnerRegistrationWorkloadIssuerInfoResultsData struct {
-	Enabled   *bool   `cbor:"0,keyasint,omitempty" json:"enabled,omitempty"`
-	IssuerUrl *string `cbor:"1,keyasint,omitempty" json:"issuer_url,omitempty"`
+	Enabled               *bool   `cbor:"0,keyasint,omitempty" json:"enabled,omitempty"`
+	IssuerUrl             *string `cbor:"1,keyasint,omitempty" json:"issuer_url,omitempty"`
+	CoordinatorInternalIp *string `cbor:"2,keyasint,omitempty" json:"coordinator_internal_ip,omitempty"`
 }
 
 type RunnerRegistrationWorkloadIssuerInfoResults struct {
@@ -984,6 +985,10 @@ func (v *RunnerRegistrationWorkloadIssuerInfoResults) SetIssuerUrl(issuer_url st
 	v.data.IssuerUrl = &issuer_url
 }
 
+func (v *RunnerRegistrationWorkloadIssuerInfoResults) SetCoordinatorInternalIp(coordinator_internal_ip string) {
+	v.data.CoordinatorInternalIp = &coordinator_internal_ip
+}
+
 func (v *RunnerRegistrationWorkloadIssuerInfoResults) MarshalCBOR() ([]byte, error) {
 	return cbor.Marshal(v.data)
 }
@@ -2485,6 +2490,17 @@ func (v *RunnerRegistrationClientWorkloadIssuerInfoResults) IssuerUrl() string {
 	return *v.data.IssuerUrl
 }
 
+func (v *RunnerRegistrationClientWorkloadIssuerInfoResults) HasCoordinatorInternalIp() bool {
+	return v.data.CoordinatorInternalIp != nil
+}
+
+func (v *RunnerRegistrationClientWorkloadIssuerInfoResults) CoordinatorInternalIp() string {
+	if v.data.CoordinatorInternalIp == nil {
+		return ""
+	}
+	return *v.data.CoordinatorInternalIp
+}
+
 func (v RunnerRegistrationClient) WorkloadIssuerInfo(ctx context.Context) (*RunnerRegistrationClientWorkloadIssuerInfoResults, error) {
 	args := RunnerRegistrationWorkloadIssuerInfoArgs{}
 

components/coordinate/coordinate.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"log/slog"
 	"net"
+	"net/netip"
 	"os"
 	"path/filepath"
 	"time"
@@ -27,14 +28,15 @@ type EtcdTLSConfig struct {
 }
 
 type CoordinatorConfig struct {
-	Address         string              `json:"address" yaml:"address"`
-	EtcdEndpoints   []string            `json:"etcd_endpoints" yaml:"etcd_endpoints"`
-	Prefix          string              `json:"prefix" yaml:"prefix"`
-	Resolver        netresolve.Resolver `json:"resolver" yaml:"resolver"`
-	TempDir         string              `json:"temp_dir" yaml:"temp_dir"`
-	DataPath        string              `json:"data_path" yaml:"data_path"`
-	AdditionalNames []string            `json:"additional_names" yaml:"additional_names"`
-	IPs             *IPSet              `json:"ips" yaml:"ips"`
+	Address               string              `json:"address" yaml:"address"`
+	CoordinatorInternalIP netip.Addr          `json:"-" yaml:"-"`
+	EtcdEndpoints         []string            `json:"etcd_endpoints" yaml:"etcd_endpoints"`
+	Prefix                string              `json:"prefix" yaml:"prefix"`
+	Resolver              netresolve.Resolver `json:"resolver" yaml:"resolver"`
+	TempDir               string              `json:"temp_dir" yaml:"temp_dir"`
+	DataPath              string              `json:"data_path" yaml:"data_path"`
+	AdditionalNames       []string            `json:"additional_names" yaml:"additional_names"`
+	IPs                   *IPSet              `json:"ips" yaml:"ips"`
 
 	// ACME certificate configuration
 	AcmeEmail       string `json:"acme_email" yaml:"acme_email"`

components/coordinate/runner_endpoints.go

@@ -52,6 +52,7 @@ func (c *RunnerEndpoints) Start(context.Context) error {
 		Authority:              c.authority,
 		EAC:                    c.eac,
 		CoordinatorAddr:        c.Address,
+		CoordinatorInternalIP:  c.CoordinatorInternalIP,
 		EtcdEndpoints:          c.EtcdEndpoints,
 		EtcdPrefix:             c.Prefix,
 		VictoriametricsAddress: c.VictoriametricsAddress,

components/distributedrunner/boot_sandbox_host.go

@@ -92,7 +92,7 @@ func (b *sandboxHostBoot) start(
 		EtcdEndpoints: append([]string(nil), b.inputs.etcdEndpoints...),
 		EtcdPrefix:    b.inputs.etcdPrefix,
 	}
-	if err := b.prepareNetworkDeps(&dependencies); err != nil {
+	if err := b.prepareNetworkDeps(&dependencies, access.access.CoordinatorInternalIP()); err != nil {
 		return nil, err
 	}
 
@@ -107,9 +107,15 @@ func (b *sandboxHostBoot) start(
 	return b.value, nil
 }
 
-func (b *sandboxHostBoot) prepareNetworkDeps(deps *runner.RunnerDeps) error {
+func (b *sandboxHostBoot) prepareNetworkDeps(deps *runner.RunnerDeps, coordinatorInternalIP netip.Addr) error {
 	resolver, hostMapper := netresolve.NewLocalResolver()
 	deps.Resolver = resolver
+	if coordinatorInternalIP.Is4() {
+		if err := hostMapper.SetHost("cluster.local", coordinatorInternalIP); err != nil {
+			return fmt.Errorf("mapping cluster registry: %w", err)
+		}
+		b.inputs.log.Info("mapped cluster.local to coordinator WireGuard gateway", "addr", coordinatorInternalIP)
+	}
 	coordinatorHost, coordinatorPort, splitErr := net.SplitHostPort(b.inputs.coordinator)
 	if splitErr != nil {
 		b.inputs.log.Warn("in-cluster API access disabled: coordinator address has no usable host and port",
@@ -120,11 +126,16 @@ func (b *sandboxHostBoot) prepareNetworkDeps(deps *runner.RunnerDeps) error {
 		// Sandboxes reach the API on the coordinator rather than the local bridge
 		// router. This must be an IP because sandbox DNS resolves app.miren names
 		// and nothing else, so the coordinator hostname would not resolve there.
-		hostMapper.SetHost("cluster.local", coordinatorAddr)
 		deps.ApiAddress = net.JoinHostPort(coordinatorAddr.String(), coordinatorPort)
 		deps.CACert = []byte(b.inputs.caCert)
-		b.inputs.log.Info("mapped cluster.local to coordinator", "hostname", coordinatorHost, "addr", coordinatorAddr)
 		b.inputs.log.Info("sandboxes will reach the cluster API at", "address", deps.ApiAddress)
+		if !coordinatorInternalIP.IsValid() {
+			// Older coordinators serve the registry on the same address as the API.
+			if err := hostMapper.SetHost("cluster.local", coordinatorAddr); err != nil {
+				return fmt.Errorf("mapping legacy cluster registry: %w", err)
+			}
+			b.inputs.log.Warn("coordinator did not advertise an internal address; using its API address for registry pulls", "addr", coordinatorAddr)
+		}
 	}
 
 	if b.inputs.clientCert == "" || b.inputs.clientKey == "" || b.inputs.caCert == "" {

components/distributedrunner/boot_sandbox_host_test.go

@@ -0,0 +1,35 @@
+//go:build linux
+
+package distributedrunner
+
+import (
+	"net/netip"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"miren.dev/runtime/components/runner"
+)
+
+func TestRegistryResolvesOverWireGuardNotPublicCoordinator(t *testing.T) {
+	boot := &sandboxHostBoot{inputs: sandboxHostBootInputs{
+		log: testLogger(), coordinator: "198.51.100.9:8443",
+	}}
+	var deps runner.RunnerDeps
+	require.NoError(t, boot.prepareNetworkDeps(&deps, netip.MustParseAddr("10.8.42.1")))
+	addr, err := deps.Resolver.LookupHost("cluster.local")
+	require.NoError(t, err)
+	require.Equal(t, netip.MustParseAddr("10.8.42.1"), addr)
+	require.Equal(t, "198.51.100.9:8443", deps.ApiAddress)
+}
+
+func TestLegacyCoordinatorRegistryUsesAPIAddress(t *testing.T) {
+	boot := &sandboxHostBoot{inputs: sandboxHostBootInputs{
+		log: testLogger(), coordinator: "198.51.100.9:8443",
+	}}
+	var deps runner.RunnerDeps
+	require.NoError(t, boot.prepareNetworkDeps(&deps, netip.Addr{}))
+	addr, err := deps.Resolver.LookupHost("cluster.local")
+	require.NoError(t, err)
+	require.Equal(t, netip.MustParseAddr("198.51.100.9"), addr)
+	require.Equal(t, "198.51.100.9:8443", deps.ApiAddress)
+}

components/runner/runner.go

@@ -237,8 +237,9 @@ func NewRunner(log *slog.Logger, deps RunnerDeps, cfg RunnerConfig) (*Runner, er
 // runner. It has no container or host-network responsibilities.
 type ClusterAccess struct {
 	RunnerConfig
-	Log  *slog.Logger
-	deps RunnerDeps
+	Log                   *slog.Logger
+	deps                  RunnerDeps
+	coordinatorInternalIP netip.Addr
 
 	state       *rpc.State
 	eac         *es.EntityAccessClient
@@ -474,8 +475,8 @@ func (r *ClusterAccess) Start(ctx context.Context) (retErr error) {
 	r.state = rs
 	r.eac = es.NewEntityAccessClient(client)
 	r.entityBase = entityserver.NewClient(r.Log, r.eac)
-	if err := r.setupRemoteWorkloadIssuer(ctx, rs); err != nil {
-		r.Log.Warn("failed to set up workload identity issuer", "error", err)
+	if err := r.setupRemoteCoordinatorInfo(ctx, rs); err != nil {
+		return fmt.Errorf("setting up coordinator registry and workload identity: %w", err)
 	}
 	if err := r.setupRemoteSecrets(rs); err != nil {
 		return fmt.Errorf("setting up secret resolution: %w", err)
@@ -601,6 +602,9 @@ func (r *ClusterAccess) WorkloadIssuer() workloadidentity.TokenIssuer {
 	return r.deps.WorkloadIssuer
 }
 
+// CoordinatorInternalIP is the coordinator's current WireGuard-routed bridge gateway.
+func (r *ClusterAccess) CoordinatorInternalIP() netip.Addr { return r.coordinatorInternalIP }
+
 // setupSqliteDisks connects to the coordinator's SQLite backup service so
 // sqlite-provider disks are replicated as they are written.
 //
@@ -645,14 +649,10 @@ func (c sqliteDiskCloser) Close() error {
 	return c.m.Close(ctx)
 }
 
-// setupRemoteWorkloadIssuer wires a remote workload identity issuer for
-// distributed runners. Runners do not hold the cluster signing key, so they
-// mint tokens by calling the coordinator's RunnerRegistration service. When the
-// coordinator reports no issuer is configured, token issuance stays disabled
-// (deps.WorkloadIssuer remains nil). The coordinator's embedded runner
-// (r.Config == nil) keeps the concrete issuer it was constructed with.
-func (r *ClusterAccess) setupRemoteWorkloadIssuer(ctx context.Context, rs *rpc.State) error {
-	if r.Config == nil || r.deps.WorkloadIssuer != nil {
+// setupRemoteCoordinatorInfo obtains the internal registry address and optional
+// workload issuer from the coordinator. The embedded runner already has both.
+func (r *ClusterAccess) setupRemoteCoordinatorInfo(ctx context.Context, rs *rpc.State) error {
+	if r.Config == nil {
 		return nil
 	}
 
@@ -664,8 +664,8 @@ func (r *ClusterAccess) setupRemoteWorkloadIssuer(ctx context.Context, rs *rpc.S
 	regClient := runner_v1alpha.NewRunnerRegistrationClient(client)
 
 	// Retry transient failures: the entities connection was just established, so
-	// a failure here is usually a brief blip. Giving up immediately would leave
-	// the runner with no token issuance until it is restarted.
+	// a failure here is usually a brief blip. This result is required to set up
+	// registry routing, even when workload identity is disabled.
 	var info *runner_v1alpha.RunnerRegistrationClientWorkloadIssuerInfoResults
 	for attempt := 1; ; attempt++ {
 		info, err = queryWorkloadIssuerInfo(ctx, regClient)
@@ -683,7 +683,16 @@ func (r *ClusterAccess) setupRemoteWorkloadIssuer(ctx context.Context, rs *rpc.S
 		case <-time.After(issuerInfoRetryDelay):
 		}
 	}
+	if info.HasCoordinatorInternalIp() {
+		r.coordinatorInternalIP, err = netip.ParseAddr(info.CoordinatorInternalIp())
+		if err != nil || !r.coordinatorInternalIP.Is4() {
+			return fmt.Errorf("invalid coordinator internal IP %q", info.CoordinatorInternalIp())
+		}
+	}
 
+	if r.deps.WorkloadIssuer != nil {
+		return nil
+	}
 	if !info.Enabled() {
 		r.Log.Info("coordinator has no workload identity issuer; sandbox tokens disabled")
 		return nil

components/server/boot_foundation.go

@@ -25,17 +25,18 @@ type foundationBoot struct {
 	output    boot.Output[foundationBootOutput]
 }
 
-func newFoundationBoot(config coordinate.CoordinatorConfig, ipDiscovery boot.Output[ipDiscoveryBootOutput], registration boot.Output[registrationBootOutput], identity boot.Output[workloadIdentityBootOutput], etcd boot.Output[etcdBootOutput], buildkit boot.Output[buildkitBootOutput], observability boot.Output[observabilityBootOutput]) *foundationBoot {
+func newFoundationBoot(config coordinate.CoordinatorConfig, ipDiscovery boot.Output[ipDiscoveryBootOutput], registration boot.Output[registrationBootOutput], identity boot.Output[workloadIdentityBootOutput], etcd boot.Output[etcdBootOutput], buildkit boot.Output[buildkitBootOutput], registryHostMapping boot.Output[registryHostMappingBootOutput], observability boot.Output[observabilityBootOutput]) *foundationBoot {
 	b := &foundationBoot{config: config}
-	b.component, b.output = boot.Provide6(
-		"cluster-foundation", ipDiscovery, registration, identity, etcd, buildkit, observability,
+	b.component, b.output = boot.Provide7(
+		"cluster-foundation", ipDiscovery, registration, identity, etcd, buildkit, registryHostMapping, observability,
 		b.start, boot.WithStop(b.stop, componentStopTimeout),
 	)
 	return b
 }
 
-func (b *foundationBoot) start(ctx context.Context, ipDiscovery ipDiscoveryBootOutput, registration registrationBootOutput, identity workloadIdentityBootOutput, etcd etcdBootOutput, buildkit buildkitBootOutput, observability observabilityBootOutput) (foundationBootOutput, error) {
+func (b *foundationBoot) start(ctx context.Context, ipDiscovery ipDiscoveryBootOutput, registration registrationBootOutput, identity workloadIdentityBootOutput, etcd etcdBootOutput, buildkit buildkitBootOutput, hostMapping registryHostMappingBootOutput, observability observabilityBootOutput) (foundationBootOutput, error) {
 	config := b.config
+	config.CoordinatorInternalIP = hostMapping.registryIP
 	config.IPs = ipDiscovery.ipSet
 	config.CloudAuth = registration.cloudAuth
 	config.WorkloadIssuer = identity.issuer

components/server/boot_oci_registry.go

@@ -4,13 +4,13 @@ package server
 
 import (
 	"context"
+	"net"
 
 	"miren.dev/runtime/components/ocireg"
+	"miren.dev/runtime/network"
 	"miren.dev/runtime/pkg/boot"
 )
 
-const ociRegistryListenAddress = ":5000"
-
 type ociRegistryBootInputs struct {
 	dataPath string
 }
@@ -26,26 +26,30 @@ func ociRegistryInputs(options StartOptions) ociRegistryBootInputs {
 	return ociRegistryBootInputs{dataPath: options.Config.Server.GetDataPath()}
 }
 
-func newOCIRegistryBoot(inputs ociRegistryBootInputs, identity boot.Output[workloadIdentityBootOutput], entityAccess boot.Output[entityAccessBootOutput], hostMapping *boot.Component, observability boot.Output[observabilityBootOutput]) *ociRegistryBoot {
+func newOCIRegistryBoot(inputs ociRegistryBootInputs, identity boot.Output[workloadIdentityBootOutput], entityAccess boot.Output[entityAccessBootOutput], hostMapping boot.Output[registryHostMappingBootOutput], sandboxHost *boot.Component, observability boot.Output[observabilityBootOutput]) *ociRegistryBoot {
 	b := &ociRegistryBoot{inputs: inputs}
-	b.component, b.output = boot.Provide3("oci-registry", identity, entityAccess, observability, b.start,
-		boot.DependsOn(hostMapping),
+	b.component, b.output = boot.Provide4("oci-registry", identity, entityAccess, hostMapping, observability, b.start,
+		boot.DependsOn(sandboxHost),
 		boot.WithStop(b.stop, componentStopTimeout),
 	)
 	return b
 }
 
-func (b *ociRegistryBoot) start(ctx context.Context, identity workloadIdentityBootOutput, entityAccess entityAccessBootOutput, observability observabilityBootOutput) (struct{}, error) {
+func (b *ociRegistryBoot) start(ctx context.Context, identity workloadIdentityBootOutput, entityAccess entityAccessBootOutput, hostMapping registryHostMappingBootOutput, observability observabilityBootOutput) (struct{}, error) {
+	if err := network.AllowRegistryFromWireGuard(); err != nil {
+		return struct{}{}, err
+	}
+	listenAddress := net.JoinHostPort(hostMapping.registryIP.String(), "5000")
 	b.registry = ocireg.NewRegistry(
 		b.inputs.dataPath,
 		observability.log,
 		entityAccess.client,
 		identity.issuer,
 	)
-	if err := b.registry.Start(ctx, ociRegistryListenAddress); err != nil {
+	if err := b.registry.Start(ctx, listenAddress); err != nil {
 		return struct{}{}, err
 	}
-	observability.log.Info("OCI registry listening", "listen-address", ociRegistryListenAddress, "service-address", ocireg.Host)
+	observability.log.Info("OCI registry listening", "listen-address", listenAddress, "service-address", ocireg.Host)
 	return struct{}{}, nil
 }
 

components/server/startup.go

@@ -115,6 +115,7 @@ func newStartup(runtime *Runtime, options StartOptions) *startup {
 		workloadIdentity.output,
 		etcd.output,
 		buildkit.output,
+		registryHostMapping.output,
 		observability.output,
 	)
 	appData := newAppDataBoot(foundation.output)
@@ -160,7 +161,7 @@ func newStartup(runtime *Runtime, options StartOptions) *startup {
 	serverInfo := newServerInfoBoot(instance, foundation.output)
 	serverLifecycle := newServerLifecycleBoot(serverLifecycleInputsFrom(options), instance, foundation.output)
 	cloudUplink := newCloudUplinkBoot(cloudControl.output, deploymentAttempts.output, ingress.output, serverLifecycle.output)
-	ociRegistry := newOCIRegistryBoot(ociRegistryInputs(options), workloadIdentity.output, entityAccess.output, registryHostMapping.component, observability.output)
+	ociRegistry := newOCIRegistryBoot(ociRegistryInputs(options), workloadIdentity.output, entityAccess.output, registryHostMapping.output, sandboxHost.component, observability.output)
 	workAdmission := newWorkAdmissionBoot(applicationManagement.output, workloadControl.component, nodePresence.Component, buildkit.component, ociRegistry.component, registryHostMapping.component)
 	buildSagaRecovery := newBuildSagaRecoveryBoot(
 		buildSagaRecoveryInputs(options),

docs/docs/distributed-runners.md

@@ -199,6 +199,10 @@ Day-to-day fleet management happens through the `runner` subcommands. A quick to
 
 A typical maintenance window looks like: drain the node, do your work, then uncordon it (or remove it if it's not coming back).
 
+:::warning[Upgrading to the internal-only registry]
+When upgrading from a release that serves the registry on the coordinator's public address to one that serves it only over WireGuard, image pulls can briefly fail. Miren Cloud-managed upgrades update the coordinator first, then restart runners one at a time; each runner resumes pulling images after its upgrade. For manual upgrades, upgrade runner binaries first while the old coordinator still serves the registry, then upgrade the coordinator and restart the runners again so they learn its internal address.
+:::
+
 ## Things to know
 
 A few properties of distributed clusters are worth keeping in mind as you plan: