feat(cni): Spread egress across all EgressShards in hashed mode

config/galactic-cni/daemonset.yaml

@@ -97,6 +97,16 @@ spec:
           # placeholder, not a published tag.
           image: ghcr.io/datum-cloud/galactic-cni:latest
           command: ["/galactic-cni", "run"]
+          env:
+            # How tenant VRFs reach the cluster's egress shards. "ordered"
+            # sends each VRF to the first reachable shard in
+            # GALACTIC_CNI_EGRESS_SHARD_SIDS. "hashed" spreads tenants across
+            # every EgressShard in this namespace, and also reads
+            # GALACTIC_CNI_EGRESS_HASH, GALACTIC_CNI_EGRESS_PIN_IDLE and
+            # GALACTIC_CNI_EGRESS_POOL_MIN_ACTIVE. See
+            # docs/nat/configuration.md#active-active-egress.
+            - name: GALACTIC_CNI_EGRESS_MODE
+              value: ordered
           # Documentation only: hostNetwork: true means these are the
           # process's actual listen ports on the node, not something
           # Kubernetes publishes -- containerPort here doesn't change
@@ -128,7 +138,7 @@ spec:
             readOnlyRootFilesystem: true
           # The kernel charges eBPF map memory to the cgroup of the process
           # that creates the map, and this container creates the uSID
-          # datapath's maps (about 59Mi) on a node that has none pinned.
+          # datapath's maps (about 67Mi) on a node that has none pinned.
           # TestUsid_MapMemoryFitsGalacticCNILimit fails when the maps plus
           # the process outgrow this limit -- see config/galactic-cni/README.md.
           resources:

config/galactic-cni/rbac.yaml

@@ -12,6 +12,13 @@ rules:
       - bgpadvertisements
       - bgpvrfinstances
     verbs: ["get", "list", "create", "update", "patch", "delete"]
+  # egressshards (read only): in hashed egress mode, galactic-cni run spreads
+  # tenant egress across every EgressShard in the cluster and watches them for
+  # membership and drain changes. See internal/installer/egressgroup.go.
+  - apiGroups: ["network.datumapis.com"]
+    resources:
+      - egressshards
+    verbs: ["get", "list", "watch"]
   - apiGroups: ["k8s.cni.cncf.io"]
     resources:
       - network-attachment-definitions

go.mod

@@ -22,7 +22,7 @@ require (
 	github.com/spf13/viper v1.21.0
 	github.com/vishvananda/netlink v1.3.2-0.20261002201547-1d1c62507e3c
 	go.datum.net/cloud v0.1.2
-	go.datum.net/network v0.4.0
+	go.datum.net/network v0.4.1-0.20261009192728-36ee528904ff
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.72.0
 	go.opentelemetry.io/otel v1.47.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.47.0
@@ -63,7 +63,6 @@ require (
 	github.com/go-openapi/swag/cmdutils v0.28.0 // indirect
 	github.com/go-openapi/swag/conv v0.28.0 // indirect
 	github.com/go-openapi/swag/fileutils v0.28.0 // indirect
-	github.com/go-openapi/swag/jsonname v0.26.0 // indirect
 	github.com/go-openapi/swag/jsonutils v0.28.0 // indirect
 	github.com/go-openapi/swag/loading v0.28.0 // indirect
 	github.com/go-openapi/swag/mangling v0.28.0 // indirect

internal/config/cni.go

@@ -5,11 +5,14 @@
 package config
 
 import (
+	"errors"
 	"fmt"
 	"net"
 	"net/netip"
 	"os"
+	"strconv"
 	"strings"
+	"time"
 
 	"go.datum.net/galactic/internal/plumbing/dan"
 )
@@ -95,8 +98,114 @@ const (
 	// Unset or empty means this fabric has no NAT64, and a VRF gets no such
 	// route -- not an error.
 	EnvCNINAT64Prefix = "GALACTIC_CNI_NAT64_PREFIX"
+
+	// EnvCNIEgressMode selects how a tenant VRF's egress reaches the
+	// cluster's egress shards. EgressModeOrdered, the default, sends each VRF
+	// to the first shard in EnvCNIEgressShardSIDs that resolves.
+	// EgressModeHashed spreads tenants across every healthy EgressShard in the
+	// cluster, picking a shard per tenant address in the datapath, and does not
+	// read EnvCNIEgressShardSIDs. Read only by galactic-cni run, which owns
+	// the shard group CNI ADD then follows.
+	EnvCNIEgressMode = "GALACTIC_CNI_EGRESS_MODE"
+
+	// EnvCNIEgressHash selects what hashed mode hashes: "source" (the
+	// default) keeps every connection from one tenant address on one shard
+	// and one public address, as RFC 4787 REQ-2 asks; "flow" hashes the
+	// 5-tuple, spreading one heavy address over every shard at the cost of
+	// that pairing.
+	EnvCNIEgressHash = "GALACTIC_CNI_EGRESS_HASH"
+
+	// EnvCNIEgressPinIdle is how long, as a Go duration, hashed mode keeps a
+	// tenant address on the shard it was first sent to after its last packet,
+	// so adding or draining a shard does not move established sessions. "0"
+	// turns pinning off. The default, DefaultEgressPinIdle, is the egress
+	// shard's longest idle timeout.
+	EnvCNIEgressPinIdle = "GALACTIC_CNI_EGRESS_PIN_IDLE"
+
+	// EnvCNIEgressPoolMinActive is the fewest active egress shards the cluster
+	// is expected to have. It is exported as a metric for the pool alert to
+	// compare against, and does not change forwarding. Default 1.
+	EnvCNIEgressPoolMinActive = "GALACTIC_CNI_EGRESS_POOL_MIN_ACTIVE"
+)
+
+// EgressMode is how tenant VRFs reach the cluster's egress shards; see
+// EnvCNIEgressMode.
+type EgressMode string
+
+// The EgressMode values.
+const (
+	EgressModeOrdered EgressMode = "ordered"
+	EgressModeHashed  EgressMode = "hashed"
 )
 
+// The EnvCNIEgressHash values.
+const (
+	EgressHashSource = "source"
+	EgressHashFlow   = "flow"
+)
+
+// DefaultEgressPinIdle is EnvCNIEgressPinIdle's default: 2 h 4 min, the
+// egress shard's established-TCP idle timeout (RFC 5382 REQ-5).
+const DefaultEgressPinIdle = 2*time.Hour + 4*time.Minute
+
+// EgressGroupConfig is galactic-cni run's hashed-mode configuration.
+type EgressGroupConfig struct {
+	Mode    EgressMode
+	Hash    string
+	PinIdle time.Duration
+	// MinActive is EnvCNIEgressPoolMinActive.
+	MinActive int
+}
+
+// ParseEgressGroupConfig reads the hashed-mode settings from getenv. Unset
+// values take their defaults; a set but invalid one is an error, so a typo
+// fails the installer at startup instead of silently running ordered.
+func ParseEgressGroupConfig(getenv func(string) string) (EgressGroupConfig, error) {
+	cfg := EgressGroupConfig{
+		Mode:      EgressModeOrdered,
+		Hash:      EgressHashSource,
+		PinIdle:   DefaultEgressPinIdle,
+		MinActive: 1,
+	}
+	var errs []error
+	switch mode := EgressMode(strings.TrimSpace(getenv(EnvCNIEgressMode))); mode {
+	case "":
+	case EgressModeOrdered, EgressModeHashed:
+		cfg.Mode = mode
+	default:
+		errs = append(errs, fmt.Errorf("%s=%q: want %q or %q", EnvCNIEgressMode, mode, EgressModeOrdered, EgressModeHashed))
+	}
+	switch hash := strings.TrimSpace(getenv(EnvCNIEgressHash)); hash {
+	case "":
+	case EgressHashSource, EgressHashFlow:
+		cfg.Hash = hash
+	default:
+		errs = append(errs, fmt.Errorf("%s=%q: want %q or %q", EnvCNIEgressHash, hash, EgressHashSource, EgressHashFlow))
+	}
+	if raw := strings.TrimSpace(getenv(EnvCNIEgressPinIdle)); raw != "" {
+		idle, err := time.ParseDuration(raw)
+		switch {
+		case raw == "0":
+			cfg.PinIdle = 0
+		case err != nil:
+			errs = append(errs, fmt.Errorf("%s=%q: %w", EnvCNIEgressPinIdle, raw, err))
+		case idle < time.Second:
+			errs = append(errs, fmt.Errorf("%s=%q: want 0 or at least 1s", EnvCNIEgressPinIdle, raw))
+		default:
+			cfg.PinIdle = idle
+		}
+	}
+	if raw := strings.TrimSpace(getenv(EnvCNIEgressPoolMinActive)); raw != "" {
+		n, err := strconv.Atoi(raw)
+		if err != nil || n < 0 {
+			errs = append(errs, fmt.Errorf("%s=%q: want a non-negative integer", EnvCNIEgressPoolMinActive, raw))
+		} else {
+			cfg.MinActive = n
+		}
+	}
+	return cfg, errors.Join(errs...)
+}
+
 // --- CNIConfig -------------------------------------------------------------
 
 // CNIConfig resolves CNI configuration with three-tier precedence: environment

internal/config/cni_test.go

@@ -6,6 +6,7 @@ package config
 
 import (
 	"testing"
+	"time"
 
 	"go.datum.net/galactic/internal/plumbing/dan"
 )
@@ -288,3 +289,48 @@ func TestParseNAT64Prefixes(t *testing.T) {
 		})
 	}
 }
+
+func TestParseEgressGroupConfig(t *testing.T) {
+	tests := []struct {
+		name    string
+		env     map[string]string
+		want    EgressGroupConfig
+		wantErr bool
+	}{
+		{
+			name: "DefaultsToOrderedSourceWithPins",
+			want: EgressGroupConfig{
+				Mode: EgressModeOrdered, Hash: EgressHashSource, PinIdle: DefaultEgressPinIdle, MinActive: 1,
+			},
+		},
+		{
+			name: "HashedFlowWithoutPins",
+			env: map[string]string{
+				EnvCNIEgressMode: "hashed", EnvCNIEgressHash: "flow", EnvCNIEgressPinIdle: "0",
+				EnvCNIEgressPoolMinActive: "3",
+			},
+			want: EgressGroupConfig{Mode: EgressModeHashed, Hash: EgressHashFlow, PinIdle: 0, MinActive: 3},
+		},
+		{
+			name: "CustomPinIdle",
+			env:  map[string]string{EnvCNIEgressPinIdle: "45m"},
+			want: EgressGroupConfig{Mode: EgressModeOrdered, Hash: EgressHashSource, PinIdle: 45 * time.Minute, MinActive: 1},
+		},
+		{name: "UnknownModeFails", env: map[string]string{EnvCNIEgressMode: "anycast"}, wantErr: true},
+		{name: "UnknownHashFails", env: map[string]string{EnvCNIEgressHash: "5tuple"}, wantErr: true},
+		{name: "BadPinIdleFails", env: map[string]string{EnvCNIEgressPinIdle: "forever"}, wantErr: true},
+		{name: "SubSecondPinIdleFails", env: map[string]string{EnvCNIEgressPinIdle: "10ms"}, wantErr: true},
+		{name: "NegativeMinActiveFails", env: map[string]string{EnvCNIEgressPoolMinActive: "-1"}, wantErr: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := ParseEgressGroupConfig(func(k string) string { return tt.env[k] })
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("ParseEgressGroupConfig() error = %v, wantErr %v", err, tt.wantErr)
+			}
+			if !tt.wantErr && got != tt.want {
+				t.Errorf("ParseEgressGroupConfig() = %+v, want %+v", got, tt.want)
+			}
+		})
+	}
+}