From 318e43c66f6c55acb801f2f72d35cbcccf84cabc Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Wed, 22 Jul 2026 22:27:07 +1000 Subject: [PATCH] fix: allocate mgmt IPs cluster-wide via NATS KV CAS (#576) MgmtIPAllocator allocated from a process-local map, but br-mgmt is trunked onto a shared physical NIC making the /24 one flat L2 segment across the whole cluster. Every node independently allocated from .10, causing collisions and ARP-race losers being blackholed. Rewrite the allocator on the spinifex-cluster-state NATS KV bucket using compare-and-swap, keyed by mgmt-ipam., modeled on IPAM. Bind the KV handle in startCluster (not startLocal) to respect the DDIL boundary. Serve already-allocated instance IDs from the local cache when KV is unhealthy, but refuse new allocations until it recovers. Rebuild now reconciles this node's running VMs into the cluster-wide record instead of just refreshing the local cache. --- spinifex/daemon/daemon.go | 9 +- spinifex/daemon/daemon_mgmt_ip.go | 279 ++++++++++++++-- spinifex/daemon/daemon_mgmt_ip_test.go | 420 ++++++++++++++++++++++++- spinifex/daemon/jetstream.go | 16 + spinifex/daemon/jetstream_test.go | 87 +++++ 5 files changed, 762 insertions(+), 49 deletions(-) diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 06955c8c2..2c06b367d 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1615,8 +1615,13 @@ func (d *Daemon) startCluster() error { return fmt.Errorf("restore instances: %w", err) } - // Rebuild mgmt IP allocator so already-allocated IPs aren't reused. - if d.mgmtIPAllocator != nil { + // Bind the mgmt IP allocator to cluster KV now that JetStream is up (DDIL + // §1e-audit: startLocal built it KV-less). Rebuild then reconciles this + // node's already-running VMs into the cluster-wide record instead of + // just refreshing the local cache, so already-allocated IPs aren't + // reused by another node either. + if d.mgmtIPAllocator != nil && d.jsManager != nil { + d.mgmtIPAllocator.BindKV(d.jsManager, d.node) d.mgmtIPAllocator.Rebuild(d.vmMgr.SnapshotMap()) slog.Info("Rebuilt mgmt IP allocator from restored instances", "allocated", d.mgmtIPAllocator.AllocatedCount()) } diff --git a/spinifex/daemon/daemon_mgmt_ip.go b/spinifex/daemon/daemon_mgmt_ip.go index 7e01059bd..d9f7b63fb 100644 --- a/spinifex/daemon/daemon_mgmt_ip.go +++ b/spinifex/daemon/daemon_mgmt_ip.go @@ -1,27 +1,107 @@ package daemon import ( + "errors" "fmt" "log/slog" "net" "sync" "github.com/mulgadc/spinifex/spinifex/vm" + "github.com/nats-io/nats.go" ) -// MgmtIPAllocator manages static IP allocation for management NICs on system instances. -// IPs are derived from the host's management bridge subnet (.10–.249 range). -// Thread-safe via its own mutex — LaunchSystemInstance does not hold a global lock. +// mgmtIPAMKeyPrefix namespaces mgmt-IP allocation records within the shared +// spinifex-cluster-state KV bucket (see JetStreamManager.UpdateMgmtIPAM). One +// record per management subnet, so clusters running distinct br-mgmt CIDRs +// never contend on the same key. +const mgmtIPAMKeyPrefix = "mgmt-ipam." + +// mgmtCASOuterRetries bounds the outer retry loop Allocate/Release/Rebuild +// wrap around UpdateMgmtIPAM. casUpdate already retries a revision conflict +// internally (maxCASRetries), but that window only absorbs one overlapping +// writer per call; a burst of many nodes or VMs allocating on the same +// subnet at once (e.g. several system instances launched together at +// cluster bootstrap) can exhaust it. Retrying the whole read-mutate-write +// cycle from a fresh Get here converges on success under that contention +// instead of surfacing a transient conflict as a hard allocation failure. +const mgmtCASOuterRetries = 20 + +// updateMgmtIPAMWithRetry wraps jsManager.UpdateMgmtIPAM with the outer +// retry described above. Only a revision-conflict exhaustion is retried — +// any other error (KV not initialized, context errors, etc.) is returned +// immediately. +func updateMgmtIPAMWithRetry(jsManager *JetStreamManager, subnet string, mutate func(*MgmtIPRecord), createIfAbsent bool) (*MgmtIPRecord, error) { + var lastErr error + for range mgmtCASOuterRetries { + record, err := jsManager.UpdateMgmtIPAM(subnet, mutate, createIfAbsent) + if err == nil { + return record, nil + } + if !errors.Is(err, nats.ErrKeyExists) { + return nil, err + } + lastErr = err + } + return nil, fmt.Errorf("mgmt IPAM update for subnet %s exhausted %d outer retries under contention: %w", + subnet, mgmtCASOuterRetries, lastErr) +} + +// MgmtIPEntry records one management IP allocation: the address, the +// instance it belongs to, and the node that made the allocation. Node is +// carried for audit and for a future reclaim sweep against dead nodes' +// heartbeats — it plays no part in allocation itself, which must consider +// every entry regardless of which node owns it. +type MgmtIPEntry struct { + IP string `json:"ip"` + InstanceID string `json:"instance_id"` + Node string `json:"node"` +} + +// MgmtIPRecord is the KV-CAS record for one management subnet. br-mgmt is +// trunked onto a shared physical NIC on the tofu/ansible/baremetal cluster +// paths, so the /24 it lives on is one flat L2 segment across every host — +// every node allocating on that subnet reads and writes this same record. +type MgmtIPRecord struct { + Subnet string `json:"subnet"` // e.g. "10.28.8.0/24" + Allocated []MgmtIPEntry `json:"allocated"` +} + +// MgmtIPAllocator manages static IP allocation for management NICs on +// system instances. IPs are derived from the host's management bridge +// subnet (.10-.249 range). +// +// Allocation is backed by a cluster-wide NATS KV compare-and-swap record +// once BindKV has been called, so nodes sharing the same br-mgmt L2 segment +// never hand out the same address (mulga-f3j2x): every node independently +// scanning from .10 was the bug, since br-mgmt is one flat subnet across the +// whole cluster, not a per-host range. +// +// allocated is a write-through local cache mirroring this node's view of the +// KV record, so AllocatedCount and lookups for an already-known instance ID +// keep working even when KV is unreachable. Allocate refuses to mint a NEW +// address in that case — this node's view of the cluster-wide record may be +// stale — but a cached instance ID and Release both still work locally. type MgmtIPAllocator struct { mu sync.Mutex - allocated map[string]string // instanceID → IP + allocated map[string]string // instanceID -> IP (write-through cache) baseIP net.IP // network base (e.g. 10.15.8.0) + subnet string // "10.15.8.0/24", the CAS record's key suffix rangeMin byte // first host byte (10) rangeMax byte // last host byte (249) + + // jsManager / node are bound by BindKV once the daemon reaches + // startCluster. Both stay nil/empty through startLocal — the DDIL + // boundary there (daemon.go's assertNoClusterServicesInitialised) means + // startLocal must never touch JetStream KV, so the allocator constructed + // there is local-cache-only until BindKV is called. + jsManager *JetStreamManager + node string } // NewMgmtIPAllocator creates an allocator for the /24 subnet of bridgeIP. -// Allocates from .10 to .249 (240 addresses). +// Allocates from .10 to .249 (240 addresses). The allocator is KV-less until +// BindKV is called; see the DDIL boundary note above. func NewMgmtIPAllocator(bridgeIP string) (*MgmtIPAllocator, error) { ip := net.ParseIP(bridgeIP) if ip == nil { @@ -40,70 +120,205 @@ func NewMgmtIPAllocator(bridgeIP string) (*MgmtIPAllocator, error) { return &MgmtIPAllocator{ allocated: make(map[string]string), baseIP: base, + subnet: fmt.Sprintf("%d.%d.%d.0/24", base[0], base[1], base[2]), rangeMin: 10, rangeMax: 249, }, nil } -// Allocate assigns a management IP to the given instance. -// Returns the allocated IP string (e.g. "10.15.8.10"). -func (a *MgmtIPAllocator) Allocate(instanceID string) (string, error) { +// BindKV attaches the cluster-wide KV backing store. Called once from +// daemon.startCluster after JetStream is up — never from startLocal (DDIL +// §1e-audit). node identifies this node's entries in the shared record for +// audit; it plays no part in allocation, which considers every entry in the +// record regardless of which node owns it. +func (a *MgmtIPAllocator) BindKV(jsManager *JetStreamManager, node string) { a.mu.Lock() defer a.mu.Unlock() + a.jsManager = jsManager + a.node = node +} - // Check if already allocated +// Allocate assigns a management IP to the given instance. +// Returns the allocated IP string (e.g. "10.15.8.10"). +// +// A known instance ID is served from the local cache without touching KV, so +// a node that already owns an address can always answer for it. A brand new +// allocation requires a healthy cluster KV — this node's cache cannot prove +// an address is free cluster-wide on its own — and is refused otherwise. +func (a *MgmtIPAllocator) Allocate(instanceID string) (string, error) { + a.mu.Lock() if ip, ok := a.allocated[instanceID]; ok { + a.mu.Unlock() return ip, nil } + jsManager, node, subnet := a.jsManager, a.node, a.subnet + a.mu.Unlock() - // Build set of used IPs for fast lookup - used := make(map[string]struct{}, len(a.allocated)) - for _, ip := range a.allocated { - used[ip] = struct{}{} + if jsManager == nil || !jsManager.KVHealthy() { + return "", fmt.Errorf("mgmt IP allocation unavailable: cluster state KV unreachable for subnet %s", subnet) } - // Find first free IP in range - for i := a.rangeMin; i <= a.rangeMax; i++ { - candidate := fmt.Sprintf("%d.%d.%d.%d", a.baseIP[0], a.baseIP[1], a.baseIP[2], i) + // mutate has no error return (casUpdate's contract), so exhaustion and + // idempotent hits are threaded back out through these closed-over + // locals. On either path mutate leaves the record unchanged, so the + // CAS write that follows is a harmless no-op against the same revision. + var allocatedIP string + var mutateErr error + _, err := updateMgmtIPAMWithRetry(jsManager, subnet, func(r *MgmtIPRecord) { + if r.Subnet == "" { + r.Subnet = subnet + } + + // Idempotent: an instance that already holds an address (allocated + // by this node earlier, or by another node before a restart lost + // the local cache) keeps it rather than taking a second one. + for _, e := range r.Allocated { + if e.InstanceID == instanceID { + allocatedIP = e.IP + return + } + } + + ip, findErr := nextFreeIP(a.baseIP, a.rangeMin, a.rangeMax, r.Allocated) + if findErr != nil { + mutateErr = findErr + return + } + r.Allocated = append(r.Allocated, MgmtIPEntry{IP: ip, InstanceID: instanceID, Node: node}) + allocatedIP = ip + }, true) + if err != nil { + return "", fmt.Errorf("mgmt IP CAS update for subnet %s: %w", subnet, err) + } + if mutateErr != nil { + return "", mutateErr + } + + a.mu.Lock() + a.allocated[instanceID] = allocatedIP + a.mu.Unlock() + + slog.Debug("Management IP allocated", "instance", instanceID, "ip", allocatedIP, "node", node) + return allocatedIP, nil +} + +// nextFreeIP scans rangeMin-rangeMax on the /24 rooted at baseIP, skipping +// every IP already present in allocated regardless of which entry's Node it +// belongs to — that is the whole fix for mulga-f3j2x: the range is shared +// across the cluster, not partitioned per host. +func nextFreeIP(baseIP net.IP, rangeMin, rangeMax byte, allocated []MgmtIPEntry) (string, error) { + used := make(map[string]struct{}, len(allocated)) + for _, e := range allocated { + used[e.IP] = struct{}{} + } + + for i := rangeMin; i <= rangeMax; i++ { + candidate := fmt.Sprintf("%d.%d.%d.%d", baseIP[0], baseIP[1], baseIP[2], i) if _, taken := used[candidate]; !taken { - a.allocated[instanceID] = candidate - slog.Debug("Management IP allocated", "instance", instanceID, "ip", candidate) return candidate, nil } } - return "", fmt.Errorf("management IP range exhausted (%d.%d.%d.%d–%d.%d.%d.%d)", - a.baseIP[0], a.baseIP[1], a.baseIP[2], a.rangeMin, - a.baseIP[0], a.baseIP[1], a.baseIP[2], a.rangeMax) + return "", fmt.Errorf("management IP range exhausted across the cluster (%d.%d.%d.%d-%d.%d.%d.%d, %d addresses)", + baseIP[0], baseIP[1], baseIP[2], rangeMin, + baseIP[0], baseIP[1], baseIP[2], rangeMax, + int(rangeMax-rangeMin)+1) } -// Release frees the management IP for the given instance. +// Release frees the management IP for the given instance, cluster-wide. +// Best-effort against KV: the local cache is always updated so this node +// never reoffers the address, but a KV write failure is logged rather than +// returned — callers (LaunchSystemInstance rollback, CleanupMgmtNetwork) +// cannot act on a Release error, matching the pre-existing signature. func (a *MgmtIPAllocator) Release(instanceID string) { a.mu.Lock() - defer a.mu.Unlock() + ip, hadLocal := a.allocated[instanceID] + delete(a.allocated, instanceID) + jsManager, subnet := a.jsManager, a.subnet + a.mu.Unlock() - if ip, ok := a.allocated[instanceID]; ok { - delete(a.allocated, instanceID) - slog.Debug("Management IP released", "instance", instanceID, "ip", ip) + if hadLocal { + slog.Debug("Management IP released locally", "instance", instanceID, "ip", ip) + } + + if jsManager == nil { + return + } + + // createIfAbsent=false: releasing an instance this node never allocated + // (or that no node has a record for) must not conjure an empty record + // into existence — nats.ErrKeyNotFound just means there was nothing to + // release. + _, err := updateMgmtIPAMWithRetry(jsManager, subnet, func(r *MgmtIPRecord) { + for i, e := range r.Allocated { + if e.InstanceID == instanceID { + r.Allocated = append(r.Allocated[:i], r.Allocated[i+1:]...) + return + } + } + }, false) + if err != nil && !errors.Is(err, nats.ErrKeyNotFound) { + slog.Warn("Failed to release mgmt IP in cluster KV; address remains reserved until reconciled", + "instance", instanceID, "err", err) } } -// Rebuild reconstructs the allocated set from existing VMs. -// Called on daemon startup after loading VMs from KV store. +// Rebuild reconciles the allocator against vms (this node's own VM +// snapshot). It always refreshes the local write-through cache first, then +// — once KV is bound (see BindKV) and healthy — CAS-inserts an entry for +// every local VM carrying a MgmtIP, idempotent by instance ID. This never +// removes another node's entries: reclaiming addresses of a dead node is a +// separate concern (tracked apart from mulga-f3j2x). +// +// Called on daemon startup (startLocal, cache-only since KV isn't bound yet) +// and again after cluster bootstrap (startCluster, once BindKV has run) so +// already-allocated IPs are never handed out a second time. func (a *MgmtIPAllocator) Rebuild(vms map[string]*vm.VM) { a.mu.Lock() - defer a.mu.Unlock() - for id, v := range vms { if v.MgmtIP != "" { a.allocated[id] = v.MgmtIP } } + count := len(a.allocated) + jsManager, node, subnet := a.jsManager, a.node, a.subnet + a.mu.Unlock() - slog.Info("Management IP allocator rebuilt", "count", len(a.allocated)) + slog.Info("Management IP allocator rebuilt", "count", count) + + if jsManager == nil { + // startLocal: no cluster KV yet, cache-only rebuild. + return + } + if !jsManager.KVHealthy() { + slog.Warn("Skipping mgmt IP KV reconcile: cluster state KV unreachable", "subnet", subnet) + return + } + + for id, v := range vms { + if v.MgmtIP == "" { + continue + } + mgmtIP := v.MgmtIP + _, err := updateMgmtIPAMWithRetry(jsManager, subnet, func(r *MgmtIPRecord) { + if r.Subnet == "" { + r.Subnet = subnet + } + for _, e := range r.Allocated { + if e.InstanceID == id { + return // already reconciled + } + } + r.Allocated = append(r.Allocated, MgmtIPEntry{IP: mgmtIP, InstanceID: id, Node: node}) + }, true) + if err != nil { + slog.Warn("Failed to reconcile mgmt IP into cluster KV", "instance", id, "ip", mgmtIP, "err", err) + } + } } -// AllocatedCount returns the number of currently allocated IPs. +// AllocatedCount returns the number of currently allocated IPs known to this +// node's local cache. func (a *MgmtIPAllocator) AllocatedCount() int { a.mu.Lock() defer a.mu.Unlock() diff --git a/spinifex/daemon/daemon_mgmt_ip_test.go b/spinifex/daemon/daemon_mgmt_ip_test.go index cec24e439..d58e7b2ef 100644 --- a/spinifex/daemon/daemon_mgmt_ip_test.go +++ b/spinifex/daemon/daemon_mgmt_ip_test.go @@ -2,12 +2,46 @@ package daemon import ( "fmt" + "strings" "sync" "testing" "github.com/mulgadc/spinifex/spinifex/vm" + "github.com/nats-io/nats.go" ) +// newTestMgmtJSM returns a JetStreamManager backed by the shared per-package +// JetStream test server, with the cluster-state bucket initialised — the +// same bucket UpdateMgmtIPAM writes into (namespaced under "mgmt-ipam."). +func newTestMgmtJSM(t *testing.T) *JetStreamManager { + t.Helper() + nc, err := nats.Connect(sharedJSNATSURL) + if err != nil { + t.Fatalf("connect NATS: %v", err) + } + t.Cleanup(nc.Close) + + jsm, err := NewJetStreamManager(nc, 1) + if err != nil { + t.Fatalf("new JetStreamManager: %v", err) + } + if err := jsm.InitClusterStateBucket(); err != nil { + t.Fatalf("init cluster state bucket: %v", err) + } + return jsm +} + +// cleanupMgmtIPAM deletes the mgmt-ipam record for a's subnet at test end. +// The cluster-state bucket is shared by the whole package's test binary, so +// without this a record left behind by one test (or one -count repetition) +// would leak into the next test/run that reuses the same bridge subnet. +func cleanupMgmtIPAM(t *testing.T, jsm *JetStreamManager, a *MgmtIPAllocator) { + t.Helper() + t.Cleanup(func() { + _ = jsm.clusterKV.Delete(mgmtIPAMKeyPrefix + a.subnet) + }) +} + func TestNewMgmtIPAllocator(t *testing.T) { tests := []struct { name string @@ -40,6 +74,10 @@ func TestNewMgmtIPAllocator(t *testing.T) { } } +// TestMgmtIPAllocator_Allocate exercises the same scan-from-.10 contract as +// before the KV rewrite, but now through a bound cluster KV — Allocate +// refuses new addresses without one (see fail-closed tests below), so every +// allocator under test here binds one first. func TestMgmtIPAllocator_Allocate(t *testing.T) { tests := []struct { name string @@ -52,10 +90,13 @@ func TestMgmtIPAllocator_Allocate(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { + jsm := newTestMgmtJSM(t) a, err := NewMgmtIPAllocator(tt.bridgeIP) if err != nil { t.Fatal(err) } + a.BindKV(jsm, "node-a") + cleanupMgmtIPAM(t, jsm, a) ip, err := a.Allocate("i-first") if err != nil { @@ -90,10 +131,13 @@ func TestMgmtIPAllocator_Allocate(t *testing.T) { } func TestMgmtIPAllocator_Release(t *testing.T) { - a, err := NewMgmtIPAllocator("10.15.8.1") + jsm := newTestMgmtJSM(t) + a, err := NewMgmtIPAllocator("10.16.8.1") if err != nil { t.Fatal(err) } + a.BindKV(jsm, "node-a") + cleanupMgmtIPAM(t, jsm, a) a.Allocate("i-one") a.Allocate("i-two") @@ -108,11 +152,14 @@ func TestMgmtIPAllocator_Release(t *testing.T) { if err != nil { t.Fatal(err) } - if ip != "10.15.8.10" { - t.Errorf("reused IP = %q, want 10.15.8.10", ip) + if ip != "10.16.8.10" { + t.Errorf("reused IP = %q, want 10.16.8.10", ip) } } +// TestMgmtIPAllocator_ReleaseNonexistent asserts Release never panics or +// blocks on an instance ID with no allocation — including with no KV bound +// at all, the shape startLocal constructs. func TestMgmtIPAllocator_ReleaseNonexistent(t *testing.T) { a, err := NewMgmtIPAllocator("10.15.8.1") if err != nil { @@ -123,12 +170,15 @@ func TestMgmtIPAllocator_ReleaseNonexistent(t *testing.T) { } func TestMgmtIPAllocator_Exhaustion(t *testing.T) { - a, err := NewMgmtIPAllocator("10.15.8.1") + jsm := newTestMgmtJSM(t) + a, err := NewMgmtIPAllocator("10.17.8.1") if err != nil { t.Fatal(err) } + a.BindKV(jsm, "node-a") + cleanupMgmtIPAM(t, jsm, a) - // Fill all 240 slots (.10–.249) + // Fill all 240 slots (.10-.249) for i := range 240 { _, err := a.Allocate(fmt.Sprintf("i-%d", i)) if err != nil { @@ -136,11 +186,14 @@ func TestMgmtIPAllocator_Exhaustion(t *testing.T) { } } - // Next should fail + // Next should fail, naming the cluster rather than a host-local limit. _, err = a.Allocate("i-overflow") if err == nil { t.Fatal("expected exhaustion error") } + if !strings.Contains(err.Error(), "exhausted across the cluster") { + t.Errorf("exhaustion error = %q, want it to name the cluster", err.Error()) + } // Release one, then it should work a.Release("i-0") @@ -148,22 +201,28 @@ func TestMgmtIPAllocator_Exhaustion(t *testing.T) { if err != nil { t.Fatalf("after release: %v", err) } - if ip != "10.15.8.10" { - t.Errorf("reused IP = %q, want 10.15.8.10", ip) + if ip != "10.17.8.10" { + t.Errorf("reused IP = %q, want 10.17.8.10", ip) } } +// TestMgmtIPAllocator_Rebuild binds KV before Rebuild so it exercises the +// reconcile path (CAS-inserting local VMs into the shared record), not just +// the local-cache refresh that runs unbound in startLocal. func TestMgmtIPAllocator_Rebuild(t *testing.T) { - a, err := NewMgmtIPAllocator("10.15.8.1") + jsm := newTestMgmtJSM(t) + a, err := NewMgmtIPAllocator("10.18.8.1") if err != nil { t.Fatal(err) } + a.BindKV(jsm, "node-a") + cleanupMgmtIPAM(t, jsm, a) vms := map[string]*vm.VM{ - "i-a": {MgmtIP: "10.15.8.10"}, - "i-b": {MgmtIP: "10.15.8.15"}, + "i-a": {MgmtIP: "10.18.8.10"}, + "i-b": {MgmtIP: "10.18.8.15"}, "i-c": {MgmtIP: ""}, // no mgmt NIC - "i-d": {MgmtIP: "10.15.8.20"}, + "i-d": {MgmtIP: "10.18.8.20"}, } a.Rebuild(vms) @@ -177,16 +236,46 @@ func TestMgmtIPAllocator_Rebuild(t *testing.T) { if err != nil { t.Fatal(err) } - if ip != "10.15.8.11" { - t.Errorf("next IP after rebuild = %q, want 10.15.8.11", ip) + if ip != "10.18.8.11" { + t.Errorf("next IP after rebuild = %q, want 10.18.8.11", ip) + } +} + +// TestMgmtIPAllocator_Rebuild_UnboundIsCacheOnly is the startLocal shape: +// Rebuild with no KV bound must only refresh the local cache and must never +// panic or block waiting on a cluster that doesn't exist yet. +func TestMgmtIPAllocator_Rebuild_UnboundIsCacheOnly(t *testing.T) { + a, err := NewMgmtIPAllocator("10.19.8.1") + if err != nil { + t.Fatal(err) + } + + vms := map[string]*vm.VM{ + "i-a": {MgmtIP: "10.19.8.10"}, + } + a.Rebuild(vms) + + if a.AllocatedCount() != 1 { + t.Errorf("count after unbound rebuild = %d, want 1", a.AllocatedCount()) + } + // A cached ID still resolves without KV. + ip, err := a.Allocate("i-a") + if err != nil { + t.Fatalf("cached instance should resolve without KV: %v", err) + } + if ip != "10.19.8.10" { + t.Errorf("cached IP = %q, want 10.19.8.10", ip) } } func TestMgmtIPAllocator_Concurrent(t *testing.T) { - a, err := NewMgmtIPAllocator("10.15.8.1") + jsm := newTestMgmtJSM(t) + a, err := NewMgmtIPAllocator("10.20.8.1") if err != nil { t.Fatal(err) } + a.BindKV(jsm, "node-a") + cleanupMgmtIPAM(t, jsm, a) var wg sync.WaitGroup results := make(map[string]string) @@ -226,3 +315,304 @@ func TestMgmtIPAllocator_Concurrent(t *testing.T) { seen[ip] = true } } + +// --- Cluster-safety coverage (mulga-f3j2x) --- + +// TestMgmtIPAllocator_SharedKV_NoDuplicates is the core regression test: +// two allocators standing in for two nodes, sharing one KV, must never both +// hand out the same address even though each scans from .10 independently. +func TestMgmtIPAllocator_SharedKV_NoDuplicates(t *testing.T) { + jsm := newTestMgmtJSM(t) + + node1, err := NewMgmtIPAllocator("10.21.8.1") + if err != nil { + t.Fatal(err) + } + node1.BindKV(jsm, "node-1") + cleanupMgmtIPAM(t, jsm, node1) + + node2, err := NewMgmtIPAllocator("10.21.8.2") + if err != nil { + t.Fatal(err) + } + node2.BindKV(jsm, "node-2") + + seen := make(map[string]string) // ip -> owning instance + allocate := func(a *MgmtIPAllocator, id string) { + ip, err := a.Allocate(id) + if err != nil { + t.Fatalf("allocate %s: %v", id, err) + } + if owner, dup := seen[ip]; dup { + t.Fatalf("duplicate address %s: held by %s and %s", ip, owner, id) + } + seen[ip] = id + } + + // Alternate allocation across the two "nodes" — each locally believes it + // is scanning a fresh range from .10, but the shared KV record must + // still serialise them onto distinct addresses. + for i := range 10 { + allocate(node1, fmt.Sprintf("node1-i-%d", i)) + allocate(node2, fmt.Sprintf("node2-i-%d", i)) + } + + if len(seen) != 20 { + t.Errorf("got %d unique addresses, want 20", len(seen)) + } +} + +// TestMgmtIPAllocator_ConcurrentAcrossNodes fires concurrent allocations +// from three simulated nodes against the same KV record, forcing genuine +// CAS conflicts (not just the happy path) as their Get/mutate/Update +// windows overlap. Every address handed out across all three must be +// unique — this is the exact shape of CI run 29912697556, where independent +// per-node allocation raced onto the same br-mgmt address. +func TestMgmtIPAllocator_ConcurrentAcrossNodes(t *testing.T) { + jsm := newTestMgmtJSM(t) + + const nodeCount = 3 + const perNode = 20 + nodes := make([]*MgmtIPAllocator, nodeCount) + for n := range nodeCount { + a, err := NewMgmtIPAllocator("10.22.8.1") + if err != nil { + t.Fatal(err) + } + a.BindKV(jsm, fmt.Sprintf("node-%d", n)) + if n == 0 { + cleanupMgmtIPAM(t, jsm, a) + } + nodes[n] = a + } + + type result struct { + id string + ip string + err error + } + results := make(chan result, nodeCount*perNode) + var wg sync.WaitGroup + for n := range nodeCount { + for i := range perNode { + wg.Add(1) + go func(n, i int) { + defer wg.Done() + id := fmt.Sprintf("node%d-i-%d", n, i) + ip, err := nodes[n].Allocate(id) + results <- result{id: id, ip: ip, err: err} + }(n, i) + } + } + wg.Wait() + close(results) + + seen := make(map[string]string) + for r := range results { + if r.err != nil { + t.Fatalf("allocate %s: %v", r.id, r.err) + } + if owner, dup := seen[r.ip]; dup { + t.Fatalf("duplicate address %s: held by %s and %s", r.ip, owner, r.id) + } + seen[r.ip] = r.id + } + if len(seen) != nodeCount*perNode { + t.Errorf("got %d unique addresses, want %d", len(seen), nodeCount*perNode) + } +} + +// TestMgmtIPAllocator_Allocate_IdempotentAcrossNodes proves idempotency is a +// property of the shared KV record, not of one allocator's local cache: a +// second allocator (standing in for the instance's owning node restarting +// inside a partition and losing its cache) re-requesting a known instance ID +// gets back the same address rather than a new one. +func TestMgmtIPAllocator_Allocate_IdempotentAcrossNodes(t *testing.T) { + jsm := newTestMgmtJSM(t) + + node1, err := NewMgmtIPAllocator("10.23.8.1") + if err != nil { + t.Fatal(err) + } + node1.BindKV(jsm, "node-1") + cleanupMgmtIPAM(t, jsm, node1) + + ip, err := node1.Allocate("i-reattach") + if err != nil { + t.Fatal(err) + } + + // A second allocator, simulating a fresh process on the same node after + // a restart with an empty local cache. + node1Restarted, err := NewMgmtIPAllocator("10.23.8.1") + if err != nil { + t.Fatal(err) + } + node1Restarted.BindKV(jsm, "node-1") + + ip2, err := node1Restarted.Allocate("i-reattach") + if err != nil { + t.Fatal(err) + } + if ip2 != ip { + t.Errorf("re-allocate from fresh allocator = %q, want %q (idempotent by instance ID)", ip2, ip) + } +} + +// TestMgmtIPAllocator_Release_FreesAddressClusterWide proves Release's +// effect is visible cluster-wide: a second allocator, not the one that +// released the address, can immediately take it. +func TestMgmtIPAllocator_Release_FreesAddressClusterWide(t *testing.T) { + jsm := newTestMgmtJSM(t) + + node1, err := NewMgmtIPAllocator("10.24.8.1") + if err != nil { + t.Fatal(err) + } + node1.BindKV(jsm, "node-1") + cleanupMgmtIPAM(t, jsm, node1) + + node2, err := NewMgmtIPAllocator("10.24.8.2") + if err != nil { + t.Fatal(err) + } + node2.BindKV(jsm, "node-2") + + ip, err := node1.Allocate("i-node1-only") + if err != nil { + t.Fatal(err) + } + node1.Release("i-node1-only") + + // node2 must now be able to take the address node1 just released. + ip2, err := node2.Allocate("i-node2-new") + if err != nil { + t.Fatal(err) + } + if ip2 != ip { + t.Errorf("node2 got %q, want the released address %q", ip2, ip) + } +} + +// TestMgmtIPAllocator_Exhaustion_ClusterWide proves the exhaustion check is +// against the cluster-wide record, not this allocator's own local cache: +// two allocators together hold all 240 addresses (120 each, so neither +// locally believes the range is full) and a third allocation from either +// must still fail. +func TestMgmtIPAllocator_Exhaustion_ClusterWide(t *testing.T) { + jsm := newTestMgmtJSM(t) + + node1, err := NewMgmtIPAllocator("10.25.8.1") + if err != nil { + t.Fatal(err) + } + node1.BindKV(jsm, "node-1") + cleanupMgmtIPAM(t, jsm, node1) + + node2, err := NewMgmtIPAllocator("10.25.8.2") + if err != nil { + t.Fatal(err) + } + node2.BindKV(jsm, "node-2") + + for i := range 120 { + if _, err := node1.Allocate(fmt.Sprintf("node1-i-%d", i)); err != nil { + t.Fatalf("node1 allocation %d failed: %v", i, err) + } + } + for i := range 120 { + if _, err := node2.Allocate(fmt.Sprintf("node2-i-%d", i)); err != nil { + t.Fatalf("node2 allocation %d failed: %v", i, err) + } + } + + // Each allocator's local cache only knows about its own 120 — but the + // cluster-wide record holds all 240, so the next allocation from either + // must fail. + if node1.AllocatedCount() != 120 { + t.Fatalf("node1 local cache = %d, want 120 (proves the check isn't local)", node1.AllocatedCount()) + } + if _, err := node1.Allocate("i-overflow-from-node1"); err == nil { + t.Fatal("expected exhaustion error from node1") + } + if _, err := node2.Allocate("i-overflow-from-node2"); err == nil { + t.Fatal("expected exhaustion error from node2") + } +} + +// TestMgmtIPAllocator_Allocate_RefusedWhenKVUnhealthy is the fail-closed +// contract: once KV is unreachable, a brand new allocation is refused, but +// an instance ID already resolved (and cached) before the outage keeps +// resolving from the local cache without touching KV at all. +func TestMgmtIPAllocator_Allocate_RefusedWhenKVUnhealthy(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + if err != nil { + t.Fatal(err) + } + jsm, err := NewJetStreamManager(nc, 1) + if err != nil { + t.Fatal(err) + } + if err := jsm.InitClusterStateBucket(); err != nil { + t.Fatal(err) + } + + a, err := NewMgmtIPAllocator("10.26.8.1") + if err != nil { + t.Fatal(err) + } + a.BindKV(jsm, "node-1") + + ip, err := a.Allocate("i-known") + if err != nil { + t.Fatal(err) + } + + // Sever the connection backing this JetStreamManager: KVHealthy's + // AccountInfo round-trip will now fail, simulating a partition. + nc.Close() + + if jsm.KVHealthy() { + t.Fatal("expected KVHealthy to report false after closing the connection") + } + + // A known instance ID still resolves — served entirely from the local + // cache, no KV round-trip required. + cachedIP, err := a.Allocate("i-known") + if err != nil { + t.Fatalf("cached instance should resolve without KV: %v", err) + } + if cachedIP != ip { + t.Errorf("cached IP = %q, want %q", cachedIP, ip) + } + + // A brand new allocation must be refused rather than risk a duplicate + // against a cluster-wide record this node can no longer trust. + if _, err := a.Allocate("i-new-during-partition"); err == nil { + t.Fatal("expected allocation to be refused while KV is unhealthy") + } +} + +// TestMgmtIPAllocator_Allocate_RefusedWhenKVNeverBound is the startLocal +// shape: an allocator with no KV bound at all (as constructed before +// startCluster runs) must refuse new allocations exactly like an unhealthy +// bound KV, while still serving whatever Rebuild already cached. +func TestMgmtIPAllocator_Allocate_RefusedWhenKVNeverBound(t *testing.T) { + a, err := NewMgmtIPAllocator("10.27.8.1") + if err != nil { + t.Fatal(err) + } + a.Rebuild(map[string]*vm.VM{"i-known": {MgmtIP: "10.27.8.10"}}) + + ip, err := a.Allocate("i-known") + if err != nil { + t.Fatalf("cached instance should resolve without KV: %v", err) + } + if ip != "10.27.8.10" { + t.Errorf("cached IP = %q, want 10.27.8.10", ip) + } + + if _, err := a.Allocate("i-brand-new"); err == nil { + t.Fatal("expected allocation to be refused with no KV bound") + } +} diff --git a/spinifex/daemon/jetstream.go b/spinifex/daemon/jetstream.go index 8dcad7e36..4dace1be9 100644 --- a/spinifex/daemon/jetstream.go +++ b/spinifex/daemon/jetstream.go @@ -478,6 +478,22 @@ func (m *JetStreamManager) UpdateClusterShutdown(mutate func(*ClusterShutdownSta return casUpdate(m.clusterKV, "cluster.shutdown", mutate, false) } +// UpdateMgmtIPAM atomically applies mutate to the mgmt-ipam record for +// subnet and writes it back with optimistic concurrency, retrying on a +// concurrent writer's revision conflict. createIfAbsent lets the first +// allocation for a subnet create its record; releases pass false so a +// release of an instance nobody holds an address for does not conjure an +// empty record into existence. Every node sharing the management bridge's +// L2 segment reads and writes the same record, keyed by subnet, which is +// what makes mgmt IP allocation safe across the whole cluster rather than +// just this process (mulga-f3j2x). +func (m *JetStreamManager) UpdateMgmtIPAM(subnet string, mutate func(*MgmtIPRecord), createIfAbsent bool) (*MgmtIPRecord, error) { + if m.clusterKV == nil { + return nil, errors.New("cluster state KV not initialized") + } + return casUpdate(m.clusterKV, mgmtIPAMKeyPrefix+subnet, mutate, createIfAbsent) +} + // ReadClusterShutdown reads the cluster shutdown state from KV. func (m *JetStreamManager) ReadClusterShutdown() (*ClusterShutdownState, error) { if m.clusterKV == nil { diff --git a/spinifex/daemon/jetstream_test.go b/spinifex/daemon/jetstream_test.go index 8398f7c46..c59f8283a 100644 --- a/spinifex/daemon/jetstream_test.go +++ b/spinifex/daemon/jetstream_test.go @@ -1634,3 +1634,90 @@ func TestJetStreamManager_BestEffort_NilObserver_NoPanic(t *testing.T) { // No observer set — must not panic on success or failure paths. jsm.WriteStateBytesBestEffort("obs-nil", []byte(`{"vms":{}}`), 5*time.Second) } + +// --- UpdateMgmtIPAM CAS tests --- + +// TestJetStreamManager_UpdateMgmtIPAM_ConflictRetry locks the CAS contract +// for mgmt-ipam records: a concurrent writer landing between Get and Update +// must be detected via a revision conflict and retried, and both the +// injected concurrent write and the retried write must land — mirroring +// UpdateStoppedInstance/UpdateTerminatedInstance's conflict-retry tests, +// applied to the allocator's backing store (mulga-f3j2x). +func TestJetStreamManager_UpdateMgmtIPAM_ConflictRetry(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + require.NoError(t, jsm.InitClusterStateBucket()) + + const subnet = "10.99.8.0/24" + defer func() { _ = jsm.clusterKV.Delete(mgmtIPAMKeyPrefix + subnet) }() + + _, err = jsm.UpdateMgmtIPAM(subnet, func(r *MgmtIPRecord) { + r.Subnet = subnet + r.Allocated = append(r.Allocated, MgmtIPEntry{IP: "10.99.8.10", InstanceID: "i-seed", Node: "node-seed"}) + }, true) + require.NoError(t, err) + + var attempts int + var injectOnce sync.Once + updated, err := jsm.UpdateMgmtIPAM(subnet, func(r *MgmtIPRecord) { + attempts++ + r.Allocated = append(r.Allocated, MgmtIPEntry{IP: "10.99.8.11", InstanceID: "i-retried", Node: "node-a"}) + + // Simulate a concurrent writer landing between our Get and Update on + // the first attempt only, forcing exactly one revision conflict. + injectOnce.Do(func() { + _, cerr := jsm.UpdateMgmtIPAM(subnet, func(cr *MgmtIPRecord) { + cr.Allocated = append(cr.Allocated, MgmtIPEntry{IP: "10.99.8.12", InstanceID: "i-concurrent", Node: "node-b"}) + }, true) + require.NoError(t, cerr) + }) + }, true) + require.NoError(t, err, "the retried Update must eventually succeed against the fresh revision") + require.NotNil(t, updated) + assert.Equal(t, 2, attempts, "the injected concurrent write must force exactly one retry") + + ids := make(map[string]bool, len(updated.Allocated)) + for _, e := range updated.Allocated { + ids[e.InstanceID] = true + } + assert.True(t, ids["i-seed"], "the original entry must survive") + assert.True(t, ids["i-concurrent"], "the concurrent writer's entry must survive, not be clobbered") + assert.True(t, ids["i-retried"], "the retried update must also land") +} + +// TestJetStreamManager_UpdateMgmtIPAM_NotFound verifies createIfAbsent=false +// surfaces nats.ErrKeyNotFound instead of silently creating a record — +// MgmtIPAllocator.Release relies on this to no-op cleanly when there is +// nothing to release. +func TestJetStreamManager_UpdateMgmtIPAM_NotFound(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + require.NoError(t, jsm.InitClusterStateBucket()) + + _, err = jsm.UpdateMgmtIPAM("10.98.8.0/24-nonexistent", func(r *MgmtIPRecord) {}, false) + assert.ErrorIs(t, err, nats.ErrKeyNotFound) +} + +// TestJetStreamManager_UpdateMgmtIPAM_ClusterKVNotInitialized tests error +// handling when the cluster-state KV bucket has not been initialized. +func TestJetStreamManager_UpdateMgmtIPAM_ClusterKVNotInitialized(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + // Don't call InitClusterStateBucket + + _, err = jsm.UpdateMgmtIPAM("10.97.8.0/24", func(r *MgmtIPRecord) {}, true) + assert.Error(t, err) + assert.Contains(t, err.Error(), "cluster state KV not initialized") +}