feat: Take the egress shard's identity from its EgressShard spec

cmd/galactic-nat/nat.go

@@ -7,13 +7,13 @@ package main
 import (
 	"fmt"
 	"log/slog"
-	"net/netip"
-	"sync/atomic"
+	"sync"
 
 	"github.com/cilium/ebpf/link"
 	"github.com/prometheus/client_golang/prometheus"
 
 	"go.datum.net/galactic/internal/config"
+	"go.datum.net/galactic/internal/controller"
 	"go.datum.net/galactic/internal/plumbing/ebpf/natattach"
 	"go.datum.net/galactic/internal/plumbing/ebpf/natmap"
 	"go.datum.net/galactic/internal/plumbing/ebpf/natprog"
@@ -36,111 +36,143 @@ var natDatapathKeepAlive struct {
 	links []link.Link
 }
 
-// natDatapathStatus reports whether setup has completed a successful load,
-// attach, and configure pass. The flag is set true once, after every step
-// succeeds, and never cleared: there is no runtime detach detection, the
-// datapath being expected to survive the process's whole lifetime. That makes
-// an atomic sufficient, it being written once and read concurrently by every
-// later reconcile.
-type natDatapathStatus struct {
-	attached atomic.Bool
+// natDatapath is this node's attached egress translation datapath, as
+// controller.EgressShardReconciler drives it (controller.EgressDatapath).
+//
+// Attachment happens once, at startup, and needs no identity: until Program
+// writes one, shard_config_table holds a row serving no family and the
+// datapath claims no packet. Program and Clear are called from the reconciler
+// only, but Attached and Programmed are read concurrently with them, hence the
+// mutex.
+type natDatapath struct {
+	uplinks     []string
+	shardConfig *natmap.ShardConfigTable
+
+	mu       sync.Mutex
+	attached bool
 }
 
-func (s *natDatapathStatus) Attached() bool { return s.attached.Load() }
+// Attached reports whether startup completed its load and attach pass. It is
+// set once and never cleared: there is no runtime detach detection, the
+// datapath being expected to survive the process's whole lifetime.
+func (d *natDatapath) Attached() bool {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	return d.attached
+}
 
-// shardConfigFromFlags translates validated process configuration into the row
-// the datapath reads. Config validation has already rejected a half-configured
-// family, so an unset address here means that family is genuinely off rather
-// than missing.
-func shardConfigFromFlags(cfg *config.NATConfig) (natmap.ShardConfig, error) {
-	sidAddr, err := netip.ParseAddr(cfg.ShardSID)
-	if err != nil {
-		return natmap.ShardConfig{}, fmt.Errorf("parse shard SID %q: %w", cfg.ShardSID, err)
+// Program writes identity into shard_config_table. A blind overwrite: see
+// natmap.ShardConfigTable.Set.
+func (d *natDatapath) Program(identity controller.EgressShardIdentity) error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	// The IPv4 counterpart of the forwarding sysctls startup sets, needed for
+	// the same reason on the one leg that leaves this datapath as IPv4: a
+	// NAT64 forward packet resolves its next hop with an IPv4 FIB lookup,
+	// which a kernel with IPv4 forwarding off refuses. The datapath counts
+	// that refusal, but a shard whose every IPv4 flow dies at the last
+	// instruction is a shard that does not work. Set only once a shard
+	// actually serves NAT64, so a node that never does keeps IPv4
+	// forwarding as it was.
+	if identity.ShardAddressIPv4.IsValid() {
+		for _, iface := range d.uplinks {
+			if err := sysctl.ConfigureFIBLookupUplinkSysctlsIPv4(iface); err != nil {
+				return fmt.Errorf("configure IPv4 forwarding on uplink interface %q: %w", iface, err)
+			}
+		}
 	}
 
-	shardCfg := natmap.ShardConfig{
-		ShardSID: sidAddr,
+	cfg := natmap.ShardConfig{
+		ShardSID:      identity.ShardSID,
+		ShardPubAddr6: identity.ShardAddressIPv6,
+		ShardPubAddr4: identity.ShardAddressIPv4,
+		NAT64Prefix:   identity.NAT64Prefix,
 	}
-
-	if cfg.ServesNAT66() {
-		pubAddr, err := netip.ParseAddr(cfg.ShardPubAddr6)
-		if err != nil {
-			return natmap.ShardConfig{}, fmt.Errorf("parse shard public address %q: %w", cfg.ShardPubAddr6, err)
-		}
-		shardCfg.ShardPubAddr6 = pubAddr
+	if current, ok, err := d.shardConfig.Get(); err == nil && ok && current == cfg {
+		return nil
+	}
+	if err := d.shardConfig.Set(cfg); err != nil {
+		return err
 	}
+	slog.Info("Egress translation datapath programmed",
+		"shardSID", identity.ShardSID,
+		"nat66", identity.ShardAddressIPv6.IsValid(),
+		"nat64", identity.ShardAddressIPv4.IsValid(),
+	)
+	return nil
+}
 
-	if cfg.ServesNAT64() {
-		pubAddr4, err := netip.ParseAddr(cfg.ShardPubAddr4)
-		if err != nil {
-			return natmap.ShardConfig{}, fmt.Errorf(
-				"parse shard public IPv4 address %q: %w", cfg.ShardPubAddr4, err)
-		}
-		prefix, err := netip.ParsePrefix(cfg.NAT64Prefix)
-		if err != nil {
-			return natmap.ShardConfig{}, fmt.Errorf("parse NAT64 prefix %q: %w", cfg.NAT64Prefix, err)
-		}
-		shardCfg.ShardPubAddr4 = pubAddr4
-		shardCfg.NAT64Prefix = prefix
+// Clear removes the programmed identity, so the datapath claims no packet.
+func (d *natDatapath) Clear() error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	if _, ok, err := d.shardConfig.Get(); err == nil && !ok {
+		return nil
 	}
+	if err := d.shardConfig.Clear(); err != nil {
+		return err
+	}
+	slog.Info("Egress translation datapath cleared; claiming no packets")
+	return nil
+}
 
-	return shardCfg, nil
+// Programmed reads back the identity shard_config_table holds. A read failure
+// reports no identity, which publishes as an empty status rather than as one
+// the datapath might not be translating with.
+func (d *natDatapath) Programmed() (controller.EgressShardIdentity, bool) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	cfg, ok, err := d.shardConfig.Get()
+	if err != nil || !ok {
+		return controller.EgressShardIdentity{}, false
+	}
+	return controller.EgressShardIdentity{
+		ShardSID:         cfg.ShardSID,
+		ShardAddressIPv6: cfg.ShardPubAddr6,
+		ShardAddressIPv4: cfg.ShardPubAddr4,
+		NAT64Prefix:      cfg.NAT64Prefix,
+	}, true
 }
 
 // setupNatDatapath loads and attaches the egress translation datapath to every
-// one of this shard's uplinks, writes its identity into the config map, and
-// registers this shard's metrics. It returns the health reporter the reconciler
-// uses for its Ready condition.
+// one of this node's uplinks and registers this shard's metrics. It returns
+// the datapath the reconciler programs from the node's EgressShard.
 //
-// Every configured uplink is attached, not only the one this node's traffic
-// uses today: a packet arriving on an uplink with no program reaches no
-// translation at all and leaves untranslated and uncounted, so a multi-homed
-// shard node that attached to one uplink would lose the shard role the moment
-// routing moved. Attachment is all-or-nothing -- natattach.Attach unwinds its
-// own partial work -- so a shard that cannot claim every uplink fails to start
-// rather than running with a hole in its coverage.
+// Uplinks come from natattach.ResolveUplinks: the override when one is
+// configured, otherwise the same auto-detection the CNI's SRv6 datapath uses.
+// Every one is attached, not only the one this node's traffic uses today: a
+// packet arriving on an uplink with no program reaches no translation at all
+// and leaves untranslated and uncounted, so a multi-homed shard node that
+// attached to one uplink would lose the shard role the moment routing moved.
+// Attachment is all-or-nothing -- natattach.Attach unwinds its own partial
+// work -- so a shard that cannot claim every uplink fails to start rather than
+// running with a hole in its coverage.
 //
 // The loaded objects and every returned link are stashed in natDatapathKeepAlive
 // rather than closed here: they, and the attachment itself, must survive for the
 // life of this process.
-func setupNatDatapath(cfg *config.NATConfig, metricsReg prometheus.Registerer) (*natDatapathStatus, error) {
-	shardCfg, err := shardConfigFromFlags(cfg)
-	if err != nil {
-		return nil, err
-	}
-
-	// A bonding master is attached through its slaves, so every step below
-	// works on the resolved targets rather than the configured names.
-	targets, err := natattach.ResolveTargets(cfg.UplinkInterfaces)
+func setupNatDatapath(cfg *config.NATConfig, metricsReg prometheus.Registerer) (*natDatapath, error) {
+	uplinks, err := natattach.ResolveUplinks(cfg.UplinkInterfaces)
 	if err != nil {
-		return nil, fmt.Errorf("resolve uplink interfaces %v: %w", cfg.UplinkInterfaces, err)
+		return nil, fmt.Errorf("resolve uplink interfaces: %w", err)
 	}
 
 	// Required for the FIB lookup in both the forward and return paths to
 	// succeed on the interface the program actually runs on. The lookup uses
-	// the ingress interface, meaning whichever resolved target the packet
-	// arrived on, so this must be applied to every one of them rather than to
-	// a primary: once an uplink is a bond, its slaves rather than the master
-	// are what the kernel reports as ingress. Best-effort and non-fatal,
-	// matching how the gateway binary configures the same sysctls.
-	for _, iface := range targets {
+	// the ingress interface, meaning whichever uplink the packet arrived on, so
+	// this must be applied to every one of them rather than to a primary:
+	// once an uplink is a bond, its slaves rather than the master are what the
+	// kernel reports as ingress, and uplinks already holds the slaves.
+	// Best-effort and non-fatal, matching how the gateway binary configures the
+	// same sysctls.
+	for _, iface := range uplinks {
 		if err := sysctl.ConfigureFIBLookupUplinkSysctls(iface); err != nil {
 			return nil, fmt.Errorf("configure IPv6 forwarding on uplink interface %q: %w", iface, err)
 		}
 	}
-	// The IPv4 counterpart, needed for the same reason on the one leg that
-	// leaves this datapath as IPv4: a NAT64 forward packet resolves its next
-	// hop with an IPv4 FIB lookup, which a kernel with IPv4 forwarding off
-	// refuses. The datapath counts that refusal, but a shard whose every IPv4
-	// flow dies at the last instruction is a shard that does not work.
-	if cfg.ServesNAT64() {
-		for _, iface := range targets {
-			if err := sysctl.ConfigureFIBLookupUplinkSysctlsIPv4(iface); err != nil {
-				return nil, fmt.Errorf(
-					"configure IPv4 forwarding on uplink interface %q: %w", iface, err)
-			}
-		}
-	}
 
 	objs, err := natattach.Load(natattach.PinDir)
 	if err != nil {
@@ -154,17 +186,16 @@ func setupNatDatapath(cfg *config.NATConfig, metricsReg prometheus.Registerer) (
 		return nil, fmt.Errorf("populate egress translation tail-call array: %w", err)
 	}
 
-	shardConfigTable := natmap.NewShardConfigTable(natmap.KernelTable{Map: objs.ShardConfigTable})
-	if err := shardConfigTable.Set(shardCfg); err != nil {
-		_ = objs.Close()
-		return nil, fmt.Errorf("write shard_config_table: %w", err)
-	}
+	// shard_config_table is deliberately not written here. Its pinned row is
+	// whatever this node's previous process programmed, so a restart keeps
+	// translating established flows until the first reconcile re-derives it
+	// from the EgressShard spec -- or clears it, if that shard is gone.
+	shardConfig := natmap.NewShardConfigTable(natmap.KernelTable{Map: objs.ShardConfigTable})
 
-	xdpLinks, err := natattach.Attach(objs.NatIngress, targets)
+	xdpLinks, err := natattach.Attach(objs.NatIngress, uplinks)
 	if err != nil {
 		_ = objs.Close()
-		return nil, fmt.Errorf("attach egress translation datapath to uplink interfaces %v: %w",
-			cfg.UplinkInterfaces, err)
+		return nil, fmt.Errorf("attach egress translation datapath to uplink interfaces %v: %w", uplinks, err)
 	}
 
 	collector := newNatCollector(objs)
@@ -178,16 +209,11 @@ func setupNatDatapath(cfg *config.NATConfig, metricsReg prometheus.Registerer) (
 	natDatapathKeepAlive.links = xdpLinks
 
 	slog.Info("Egress translation datapath attached",
-		"interfaces", cfg.UplinkInterfaces,
-		"targets", targets,
-		"shardSID", cfg.ShardSID,
-		"nat66", cfg.ServesNAT66(),
-		"nat64", cfg.ServesNAT64(),
+		"interfaces", uplinks,
+		"autoDetected", len(cfg.UplinkInterfaces) == 0,
 	)
 
-	status := &natDatapathStatus{}
-	status.attached.Store(true)
-	return status, nil
+	return &natDatapath{uplinks: uplinks, shardConfig: shardConfig, attached: true}, nil
 }
 
 // closeAll best-effort closes every link, to unwind a partially set-up datapath

cmd/galactic-nat/root.go

@@ -37,9 +37,10 @@ const (
  Find more information at: https://www.datum.net/docs`
 )
 
-// runCmd is the application startup: it loads and attaches this shard's NAT66
-// egress datapath to its fabric-facing uplink and registers the reconciler that
-// publishes this shard's identity and health.
+// runCmd is the application startup: it loads and attaches this node's egress
+// translation datapath to its fabric-facing uplinks and registers the
+// reconciler that programs it from this node's EgressShard and publishes the
+// result.
 func runCmd(cfg *config.NATConfig) error {
 	nodeName := cfg.NodeName
 	metricsPort := cfg.MetricsPort
@@ -111,28 +112,26 @@ func runCmd(cfg *config.NATConfig) error {
 	// Pre-flight RBAC check.
 	checkWatchPermissions(mgr)
 
-	// Load and attach the egress translation datapath. Always a real datapath:
-	// configuration validation rejects an empty uplink or SID, and a shard
-	// serving neither address family, before this is reached -- this binary
-	// exists only to run a shard.
-	datapathHealth, err := setupNatDatapath(cfg, ctrlmetrics.Registry)
+	// Load and attach the egress translation datapath. It needs no identity to
+	// attach: that comes from this node's EgressShard spec, which the
+	// reconciler below programs on its first reconcile.
+	datapath, err := setupNatDatapath(cfg, ctrlmetrics.Registry)
 	if err != nil {
 		return fmt.Errorf("setup egress translation eBPF datapath: %w", err)
 	}
 	// Only now is the datapath attached. Report serving from here on, not from
-	// process start.
+	// process start. Serving does not wait for an identity to be programmed:
+	// a node whose shard has not been assigned one yet would otherwise never
+	// turn ready and would block the DaemonSet's rollout. The EgressShard's
+	// Programmed condition reports that instead.
 	healthSrv.SetServingStatus("", grpc_health_v1.HealthCheckResponse_SERVING)
 
 	// Register EgressShard controller.
 	if err := (&controller.EgressShardReconciler{
-		Client:           mgr.GetClient(),
-		Scheme:           mgr.GetScheme(),
-		NodeName:         nodeName,
-		ShardSID:         cfg.ShardSID,
-		ShardAddressIPv6: cfg.ShardPubAddr6,
-		ShardAddressIPv4: cfg.ShardPubAddr4,
-		NAT64Prefix:      cfg.NAT64Prefix,
-		Datapath:         datapathHealth,
+		Client:   mgr.GetClient(),
+		Scheme:   mgr.GetScheme(),
+		NodeName: nodeName,
+		Datapath: datapath,
 	}).SetupWithManager(mgr); err != nil {
 		return fmt.Errorf("setup EgressShard controller: %w", err)
 	}
@@ -184,17 +183,8 @@ func newRootCommand() *cobra.Command {
 		config.DefaultNATGRPCHealthPort,
 		"gRPC health check port")
 	cmd.Flags().StringP("nat-uplink-interfaces", "", "",
-		"Comma-separated fabric-facing uplink interfaces this shard's XDP datapath attaches to; "+
-			"name every fabric uplink, not just the primary (required)")
-	cmd.Flags().StringP("nat-shard-sid", "", "",
-		"This shard's own SRv6 uSID, encapsulation target for tenant egress traffic (required)")
-	cmd.Flags().StringP("nat-shard-pub-addr6", "", "",
-		"This shard's own publicly-routable IPv6 masquerade source address, enabling NAT66")
-	cmd.Flags().StringP("nat-shard-pub-addr4", "", "",
-		"This shard's own publicly-routable IPv4 masquerade source address, enabling NAT64 "+
-			"together with --nat64-prefix")
-	cmd.Flags().StringP("nat64-prefix", "", "",
-		"Fabric-wide NAT64 /96 this shard translates for; must match what DNS64 synthesizes into")
+		"Comma-separated fabric-facing uplink interfaces this shard's XDP datapath attaches to, "+
+			"overriding auto-detection; name every fabric uplink, not just the primary")
 	cmd.Flags().Bool("build-info", false, "Print build information and exit")
 	cmd.Flags().BoolP("version", "V", false, "Print version and exit")
 	return cmd