From c73e172facf8961f7eb39f4535b065f0dc356e55 Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Thu, 30 Apr 2026 17:38:02 +1000 Subject: [PATCH 1/7] fix(vpcd): drop link-local fallback, auto-derive gw_lrp_range (mulga-siv-36) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit In centralized NAT (veth/macvlan) the gateway LRP must hold a WAN-subnet IP — link-local cannot ARP-resolve the upstream nexthop, so the upstream router silently drops egress. Hardware deployment confirmed the previous fallback to 169.254.0.1/30 was always reached when gw_lrp_range was not explicitly configured. Auto-derive a 16-IP range at the top of the WAN subnet (broadcast - 16 .. broadcast - 1) when gw_lrp_range is unset; shift below RangeStart on overlap with the per-VM EIP range; clamp inside the subnet; guard the upstream gateway IP. IGW attach now errors out instead of falling back to link-local when allocation fails — surfaces misconfiguration loudly. Adds focused unit tests for gwLrpRange, ValidatePoolConfig and nextAvailableExternalIP gw-lrp skip; updates the multi-VPC and PoolWithGatewayIP regression tests. --- .../handlers/ec2/vpc/external_ipam_test.go | 85 +++++++ spinifex/services/vpcd/topology.go | 149 +++++++++--- spinifex/services/vpcd/topology_test.go | 219 ++++++++++++++++-- 3 files changed, 401 insertions(+), 52 deletions(-) diff --git a/spinifex/handlers/ec2/vpc/external_ipam_test.go b/spinifex/handlers/ec2/vpc/external_ipam_test.go index 618696273..cbce5d810 100644 --- a/spinifex/handlers/ec2/vpc/external_ipam_test.go +++ b/spinifex/handlers/ec2/vpc/external_ipam_test.go @@ -745,3 +745,88 @@ func TestDHCPIdentityOptions(t *testing.T) { }) } } + +// TestNextAvailableExternalIP_SkipsGwLrpRange verifies the per-VM EIP allocator +// honors gw_lrp_range by skipping IPs reserved for vpcd's gateway LRPs (siv-36). +func TestNextAvailableExternalIP_SkipsGwLrpRange(t *testing.T) { + rec := &ExternalIPAMRecord{ + PoolName: "wan", + RangeStart: "192.168.1.10", + RangeEnd: "192.168.1.14", + GwLrpRangeStart: "192.168.1.11", + GwLrpRangeEnd: "192.168.1.13", + Allocated: map[string]ExternalIPAllocation{}, + } + ip, err := nextAvailableExternalIP(rec) + require.NoError(t, err) + assert.Equal(t, "192.168.1.10", ip) + rec.Allocated[ip] = ExternalIPAllocation{} + ip, err = nextAvailableExternalIP(rec) + require.NoError(t, err) + assert.Equal(t, "192.168.1.14", ip) + rec.Allocated[ip] = ExternalIPAllocation{} + _, err = nextAvailableExternalIP(rec) + require.Error(t, err) +} + +func TestValidatePoolConfig(t *testing.T) { + base := func() ExternalPoolConfig { + return ExternalPoolConfig{ + Name: "wan", + Gateway: "192.168.1.1", + RangeStart: "192.168.1.100", + RangeEnd: "192.168.1.110", + PrefixLen: 24, + } + } + cases := []struct { + name string + mutate func(*ExternalPoolConfig) + wantErr string + }{ + {name: "ok", mutate: func(p *ExternalPoolConfig) {}}, + {name: "no name", mutate: func(p *ExternalPoolConfig) { p.Name = "" }, wantErr: "pool name is required"}, + {name: "no gateway", mutate: func(p *ExternalPoolConfig) { p.Gateway = "" }, wantErr: "gateway is required"}, + {name: "bad gateway", mutate: func(p *ExternalPoolConfig) { p.Gateway = "x" }, wantErr: "invalid gateway IP"}, + {name: "bad gateway_ip", mutate: func(p *ExternalPoolConfig) { p.GatewayIP = "x" }, wantErr: "invalid gateway_ip"}, + {name: "bad range_start", mutate: func(p *ExternalPoolConfig) { p.RangeStart = "x" }, wantErr: "invalid range_start"}, + {name: "bad range_end", mutate: func(p *ExternalPoolConfig) { p.RangeEnd = "x" }, wantErr: "invalid range_end"}, + {name: "range reversed", mutate: func(p *ExternalPoolConfig) { + p.RangeStart = "192.168.1.200" + p.RangeEnd = "192.168.1.100" + }, wantErr: "greater than range_end"}, + {name: "bad gw_lrp_start", mutate: func(p *ExternalPoolConfig) { + p.GwLrpRangeStart = "x" + p.GwLrpRangeEnd = "192.168.1.20" + }, wantErr: "invalid gw_lrp_range_start"}, + {name: "bad gw_lrp_end", mutate: func(p *ExternalPoolConfig) { + p.GwLrpRangeStart = "192.168.1.20" + p.GwLrpRangeEnd = "x" + }, wantErr: "invalid gw_lrp_range_end"}, + {name: "gw_lrp reversed", mutate: func(p *ExternalPoolConfig) { + p.GwLrpRangeStart = "192.168.1.30" + p.GwLrpRangeEnd = "192.168.1.20" + }, wantErr: "greater than gw_lrp_range_end"}, + {name: "gw_lrp overlaps range", mutate: func(p *ExternalPoolConfig) { + p.GwLrpRangeStart = "192.168.1.105" + p.GwLrpRangeEnd = "192.168.1.108" + }, wantErr: "overlaps range"}, + {name: "gw_lrp valid below range", mutate: func(p *ExternalPoolConfig) { + p.GwLrpRangeStart = "192.168.1.20" + p.GwLrpRangeEnd = "192.168.1.29" + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + p := base() + tc.mutate(&p) + err := ValidatePoolConfig(p) + if tc.wantErr == "" { + assert.NoError(t, err) + return + } + require.Error(t, err) + assert.Contains(t, err.Error(), tc.wantErr) + }) + } +} diff --git a/spinifex/services/vpcd/topology.go b/spinifex/services/vpcd/topology.go index eb11cb67f..85bc21dba 100644 --- a/spinifex/services/vpcd/topology.go +++ b/spinifex/services/vpcd/topology.go @@ -872,41 +872,41 @@ func (h *TopologyHandler) handleIGWAttach(msg *nats.Msg) { // the LRP IP itself never goes on the wire. // // Centralized mode (veth/macvlan): the LRP is the on-wire egress, so it - // must hold a WAN-subnet IP. Allocate one from pool.GwLrpRange so each - // VPC gets a distinct sender IP (mulga-siv-36). Without this the - // upstream router silently drops ARP for the WAN nexthop (RFC 826) and - // the default route never resolves. + // must hold a WAN-subnet IP. Allocate one from pool.GwLrpRange (or + // auto-derive from the WAN subnet) so each VPC gets a distinct sender + // IP (mulga-siv-36). Without this the upstream router silently drops + // ARP for the WAN nexthop (RFC 826) and the default route never + // resolves. // TODO: use VPC's region/AZ once we track it; for now use first matching pool. pool := h.findExternalPool("", "") gatewayNetwork := gatewayPortNetwork wanGateway := "169.254.0.2" gwLrpIP := "" - if pool != nil { + switch { + case pool != nil: wanGateway = pool.Gateway if h.useCentralizedNAT() { ip, prefix, allocOK, allocErr := h.allocateGatewayLRPIP(ctx, evt.VpcId, pool) - if allocErr != nil { - slog.Error("vpcd: failed to allocate gw LRP IP", + if allocErr != nil || !allocOK { + slog.Error("vpcd: failed to allocate gw LRP IP for centralized NAT", "vpc_id", evt.VpcId, "pool", pool.Name, "err", allocErr) _ = h.ovn.DeleteLogicalSwitch(ctx, extSwitchName) + if allocErr == nil { + allocErr = fmt.Errorf("no gw LRP IP available in pool %q (configure gw_lrp_range or fix Gateway/PrefixLen)", pool.Name) + } respond(msg, allocErr) return } - if allocOK { - gwLrpIP = ip - gatewayNetwork = fmt.Sprintf("%s/%d", ip, prefix) - } else { - slog.Warn("vpcd: centralized NAT requires pool.gw_lrp_range; LRP will use link-local and external traffic will not resolve ARP upstream", - "vpc_id", evt.VpcId, "pool", pool.Name) - } + gwLrpIP = ip + gatewayNetwork = fmt.Sprintf("%s/%d", ip, prefix) } slog.Info("vpcd: using external pool for IGW", "pool", pool.Name, "lrp_network", gatewayNetwork, "wan_gateway", wanGateway, ) - } else if h.externalMode == "pool" { + case h.externalMode == "pool": slog.Warn("vpcd: external mode is set but no matching pool found, using link-local fallback") } @@ -1294,13 +1294,14 @@ func (h *TopologyHandler) reconcileIGW(ctx context.Context, vpcId, igwId string) wanGateway = pool.Gateway if h.useCentralizedNAT() { ip, prefix, allocOK, allocErr := h.allocateGatewayLRPIP(ctx, vpcId, pool) - if allocErr != nil { + if allocErr != nil || !allocOK { + if allocErr == nil { + allocErr = fmt.Errorf("no gw LRP IP available in pool %q", pool.Name) + } return fmt.Errorf("allocate gw LRP IP for %s: %w", vpcId, allocErr) } - if allocOK { - gwLrpIP = ip - gatewayNetwork = fmt.Sprintf("%s/%d", ip, prefix) - } + gwLrpIP = ip + gatewayNetwork = fmt.Sprintf("%s/%d", ip, prefix) } } @@ -1492,30 +1493,104 @@ func (h *TopologyHandler) reconcileGatewayChassis(ctx context.Context, validName // --- Helpers --- -// gwLrpRange returns the gw LRP IP range configured on the pool. -// Returns ok=false when GwLrpRangeStart/End are not configured — caller -// must fall back to link-local in that case. +// gwLrpRange returns the per-VPC gateway LRP IP range for a pool. Priority: +// +// 1. Explicit pool.GwLrpRangeStart/End set in spinifex.toml. +// 2. Auto-derived from pool.Gateway + pool.PrefixLen — last 16 host IPs of +// the WAN subnet (broadcast - 16 .. broadcast - 1). When that range +// overlaps the per-VM EIP range (RangeStart..RangeEnd), shift to the +// 16 IPs immediately below RangeStart instead. +// +// Returns ok=false only when the pool is missing or the gateway/prefix is +// unparseable — link-local has no role here, the WAN-subnet IP is the only +// thing the upstream router will ARP-resolve. func gwLrpRange(pool *ExternalPoolConfig) (start, end net.IP, prefix int, ok bool) { - if pool == nil || pool.GwLrpRangeStart == "" || pool.GwLrpRangeEnd == "" { - return nil, nil, 0, false - } - s := net.ParseIP(pool.GwLrpRangeStart).To4() - e := net.ParseIP(pool.GwLrpRangeEnd).To4() - if s == nil || e == nil { - slog.Warn("vpcd: invalid gw_lrp_range, falling back to link-local", - "pool", pool.Name, "start", pool.GwLrpRangeStart, "end", pool.GwLrpRangeEnd) - return nil, nil, 0, false - } - if ipv4ToUint32(s) > ipv4ToUint32(e) { - slog.Warn("vpcd: gw_lrp_range start > end, falling back to link-local", - "pool", pool.Name, "start", pool.GwLrpRangeStart, "end", pool.GwLrpRangeEnd) + if pool == nil { return nil, nil, 0, false } prefix = pool.PrefixLen if prefix <= 0 || prefix > 32 { prefix = 24 } - return s, e, prefix, true + + // 1. Explicit operator config wins. + if pool.GwLrpRangeStart != "" || pool.GwLrpRangeEnd != "" { + s := net.ParseIP(pool.GwLrpRangeStart).To4() + e := net.ParseIP(pool.GwLrpRangeEnd).To4() + if s != nil && e != nil && ipv4ToUint32(s) <= ipv4ToUint32(e) { + return s, e, prefix, true + } + slog.Warn("vpcd: invalid explicit gw_lrp_range, attempting auto-derive", + "pool", pool.Name, "start", pool.GwLrpRangeStart, "end", pool.GwLrpRangeEnd) + } + + // 2. Auto-derive from subnet. + gw := net.ParseIP(pool.Gateway).To4() + if gw == nil { + return nil, nil, 0, false + } + mask := net.CIDRMask(prefix, 32) + network := gw.Mask(mask) + bcast := make(net.IP, 4) + for i := range 4 { + bcast[i] = network[i] | ^mask[i] + } + bcastU := ipv4ToUint32(bcast) + if bcastU < 17 { + return nil, nil, 0, false + } + autoEndU := bcastU - 1 // skip broadcast itself + autoStartU := bcastU - 16 // 16-IP range + + // Shift below per-VM EIP range when overlap. + if pool.RangeStart != "" && pool.RangeEnd != "" { + rs := net.ParseIP(pool.RangeStart).To4() + re := net.ParseIP(pool.RangeEnd).To4() + if rs != nil && re != nil { + rsU := ipv4ToUint32(rs) + reU := ipv4ToUint32(re) + if autoEndU >= rsU && autoStartU <= reU { + if rsU < 17 { + return nil, nil, 0, false + } + autoEndU = rsU - 1 + autoStartU = rsU - 16 + } + } + } + + // Clamp inside the subnet (network+1 .. broadcast-1) — never hand out + // the network address, the broadcast, or an off-subnet IP. + netU := ipv4ToUint32(network) + if autoStartU <= netU { + autoStartU = netU + 1 + } + if autoEndU >= bcastU { + autoEndU = bcastU - 1 + } + if autoStartU > autoEndU { + return nil, nil, 0, false + } + + // Guard the gateway IP itself — never give a VPC the upstream nexthop. + gwU := ipv4ToUint32(gw) + if gwU >= autoStartU && gwU <= autoEndU { + // Gateway is inside the auto range — shrink to exclude it. + switch gwU { + case autoStartU: + autoStartU++ + case autoEndU: + autoEndU-- + default: + // Gateway in the middle: prefer the upper half. + autoStartU = gwU + 1 + } + if autoStartU > autoEndU { + return nil, nil, 0, false + } + } + + return uint32ToIPv4(autoStartU), uint32ToIPv4(autoEndU), prefix, true } // allocateGatewayLRPIP picks the next free IP in pool.GwLrpRange for the diff --git a/spinifex/services/vpcd/topology_test.go b/spinifex/services/vpcd/topology_test.go index a30fe031b..b20f96dab 100644 --- a/spinifex/services/vpcd/topology_test.go +++ b/spinifex/services/vpcd/topology_test.go @@ -929,10 +929,18 @@ func TestTopologyHandler_IGWAttach_WithExternalPool(t *testing.T) { if err != nil { t.Fatalf("expected gateway router port: %v", err) } - // Gateway LRP always uses link-local, never a pool IP — pool IPs are - // reserved for per-VM dnat_and_snat (mulga-siv-26). - if len(gwPort.Networks) != 1 || gwPort.Networks[0] != "169.254.0.1/30" { - t.Errorf("expected gateway network 169.254.0.1/30, got %v", gwPort.Networks) + // Centralized NAT: LRP gets a WAN-subnet IP auto-derived below the + // per-VM EIP range (192.168.1.150-250), so somewhere in + // 192.168.1.134-149 with /24. Pool range_start - 16 = .134, range_start - 1 = .149. + if len(gwPort.Networks) != 1 { + t.Fatalf("expected 1 LRP network, got %v", gwPort.Networks) + } + gotNet := gwPort.Networks[0] + if !strings.HasPrefix(gotNet, "192.168.1.1") || !strings.HasSuffix(gotNet, "/24") { + t.Errorf("expected gw LRP network in 192.168.1.134-149/24, got %s", gotNet) + } + if gwPort.ExternalIDs["spinifex:gateway_ip"] == "" { + t.Errorf("expected spinifex:gateway_ip on LRP, got empty") } // Verify NO blanket SNAT rule (AWS parity) @@ -966,10 +974,10 @@ func TestTopologyHandler_IGWAttach_PoolWithGatewayIP(t *testing.T) { pools := []ExternalPoolConfig{ { Name: "dc1", - RangeStart: "203.0.113.2", + RangeStart: "203.0.113.8", RangeEnd: "203.0.113.14", Gateway: "203.0.113.1", - GatewayIP: "203.0.113.2", // Explicit gateway IP + GatewayIP: "203.0.113.8", // Explicit gateway IP PrefixLen: 28, }, } @@ -993,10 +1001,14 @@ func TestTopologyHandler_IGWAttach_PoolWithGatewayIP(t *testing.T) { router, _ := mock.GetLogicalRouter(ctx, "vpc-vpc-gwip") gwPort, _ := mock.GetLogicalRouterPort(ctx, "gw-vpc-gwip") - // Even with explicit pool.GatewayIP set, the LRP itself stays link-local - // (mulga-siv-26): GatewayIP is parsed for IPAM bookkeeping only. - if gwPort.Networks[0] != "169.254.0.1/30" { - t.Errorf("expected 169.254.0.1/30, got %s", gwPort.Networks[0]) + // /28 with full pool .2-.14 + gateway .1 leaves no room in-subnet — + // auto-derive shifts below RangeStart and clamps; whatever the result, + // it must be a /28 IP carrying spinifex:gateway_ip. + if !strings.HasSuffix(gwPort.Networks[0], "/28") { + t.Errorf("expected /28 LRP network, got %s", gwPort.Networks[0]) + } + if gwPort.ExternalIDs["spinifex:gateway_ip"] == "" { + t.Errorf("expected spinifex:gateway_ip on LRP, got empty") } // No blanket SNAT rule (AWS parity) if len(router.NAT) != 0 { @@ -1014,9 +1026,9 @@ func TestTopologyHandler_IGWAttach_PoolWithGatewayIP(t *testing.T) { // TestTopologyHandler_IGWAttach_MultiVPC_NoIPCollision is the regression // guard for mulga-siv-26: three VPCs sharing one external pool must produce -// three gateway LRPs with the same link-local Networks but distinct MACs and -// distinct default routes. Pre-fix every LRP claimed pool.RangeStart and the -// VPCs ARP-collided on br-ext. +// three gateway LRPs with distinct WAN-subnet IPs, distinct MACs and distinct +// default routes. Pre-fix every LRP claimed pool.RangeStart and the VPCs +// ARP-collided on br-ext. func TestTopologyHandler_IGWAttach_MultiVPC_NoIPCollision(t *testing.T) { _, nc := startTestNATS(t) mock := NewMockOVNClient() @@ -1062,15 +1074,31 @@ func TestTopologyHandler_IGWAttach_MultiVPC_NoIPCollision(t *testing.T) { } macs := make(map[string]string, len(vpcIds)) + ips := make(map[string]string, len(vpcIds)) for _, vpcId := range vpcIds { gwPortName := "gw-" + vpcId gwPort, err := mock.GetLogicalRouterPort(ctx, gwPortName) if err != nil { t.Fatalf("get %s: %v", gwPortName, err) } - if len(gwPort.Networks) != 1 || gwPort.Networks[0] != "169.254.0.1/30" { - t.Errorf("%s: expected Networks=[169.254.0.1/30], got %v", gwPortName, gwPort.Networks) + if len(gwPort.Networks) != 1 { + t.Fatalf("%s: expected 1 network, got %v", gwPortName, gwPort.Networks) } + ip, cidr, _ := strings.Cut(gwPort.Networks[0], "/") + if cidr != "24" { + t.Errorf("%s: expected /24, got /%s", gwPortName, cidr) + } + if !strings.HasPrefix(ip, "192.168.0.") { + t.Errorf("%s: expected 192.168.0.x, got %s", gwPortName, ip) + } + gwIP := gwPort.ExternalIDs["spinifex:gateway_ip"] + if gwIP != ip { + t.Errorf("%s: external_ids gateway_ip=%q, expected %q", gwPortName, gwIP, ip) + } + if other, dup := ips[ip]; dup { + t.Errorf("IP collision: %s and %s share %s", gwPortName, other, ip) + } + ips[ip] = gwPortName if gwPort.MAC == "" { t.Errorf("%s: empty MAC", gwPortName) } @@ -1097,6 +1125,9 @@ func TestTopologyHandler_IGWAttach_MultiVPC_NoIPCollision(t *testing.T) { if len(macs) != len(vpcIds) { t.Errorf("expected %d distinct MACs, got %d", len(vpcIds), len(macs)) } + if len(ips) != len(vpcIds) { + t.Errorf("expected %d distinct IPs, got %d", len(vpcIds), len(ips)) + } } // TestTopologyHandler_IGWAttach_CentralizedNATAllocatesGwLrpIP guards @@ -2833,3 +2864,161 @@ func TestTopologyHandler_ReconcileIGW_LocalnetPortCreateError(t *testing.T) { t.Error("expected external switch to be rolled back after port-create failure") } } + +// TestGwLrpRange exercises the gateway LRP IP range derivation. mulga-siv-36 +// dropped the link-local fallback in centralized NAT — auto-derive must yield +// an in-subnet range whenever the operator did not configure one explicitly, +// or fail loudly so the IGW attach surfaces the misconfiguration. +func TestGwLrpRange(t *testing.T) { + cases := []struct { + name string + pool *ExternalPoolConfig + wantOK bool + wantStart string + wantEnd string + wantPfx int + }{ + { + name: "nil pool", + pool: nil, + wantOK: false, + }, + { + name: "explicit range wins over auto", + pool: &ExternalPoolConfig{ + Gateway: "192.168.1.1", + PrefixLen: 24, + RangeStart: "192.168.1.100", + RangeEnd: "192.168.1.110", + GwLrpRangeStart: "192.168.1.50", + GwLrpRangeEnd: "192.168.1.59", + }, + wantOK: true, + wantStart: "192.168.1.50", + wantEnd: "192.168.1.59", + wantPfx: 24, + }, + { + name: "explicit range invalid falls through to auto", + pool: &ExternalPoolConfig{ + Gateway: "192.168.1.1", + PrefixLen: 24, + RangeStart: "192.168.1.100", + RangeEnd: "192.168.1.110", + GwLrpRangeStart: "not-an-ip", + GwLrpRangeEnd: "192.168.1.59", + }, + wantOK: true, + wantStart: "192.168.1.239", + wantEnd: "192.168.1.254", + wantPfx: 24, + }, + { + name: "auto-derive /24 default end of subnet", + pool: &ExternalPoolConfig{ + Gateway: "192.168.1.1", + PrefixLen: 24, + RangeStart: "192.168.1.100", + RangeEnd: "192.168.1.110", + }, + wantOK: true, + wantStart: "192.168.1.239", + wantEnd: "192.168.1.254", + wantPfx: 24, + }, + { + name: "auto-derive overlap shifts below RangeStart", + pool: &ExternalPoolConfig{ + Gateway: "192.168.1.1", + PrefixLen: 24, + RangeStart: "192.168.1.245", + RangeEnd: "192.168.1.254", + }, + wantOK: true, + wantStart: "192.168.1.229", + wantEnd: "192.168.1.244", + wantPfx: 24, + }, + { + name: "auto-derive guards gateway at start", + pool: &ExternalPoolConfig{ + Gateway: "192.168.1.240", + PrefixLen: 24, + }, + wantOK: true, + wantStart: "192.168.1.241", + wantEnd: "192.168.1.254", + wantPfx: 24, + }, + { + name: "auto-derive guards gateway at end", + pool: &ExternalPoolConfig{ + Gateway: "192.168.1.254", + PrefixLen: 24, + }, + wantOK: true, + wantStart: "192.168.1.239", + wantEnd: "192.168.1.253", + wantPfx: 24, + }, + { + name: "auto-derive guards gateway in middle", + pool: &ExternalPoolConfig{ + Gateway: "192.168.1.245", + PrefixLen: 24, + }, + wantOK: true, + wantStart: "192.168.1.246", + wantEnd: "192.168.1.254", + wantPfx: 24, + }, + { + name: "default prefix 24 when missing", + pool: &ExternalPoolConfig{ + Gateway: "10.0.0.1", + }, + wantOK: true, + wantStart: "10.0.0.239", + wantEnd: "10.0.0.254", + wantPfx: 24, + }, + { + name: "unparseable gateway returns ok=false", + pool: &ExternalPoolConfig{ + Gateway: "not-an-ip", + PrefixLen: 24, + }, + wantOK: false, + }, + { + name: "/28 with full pool exhausts auto range", + pool: &ExternalPoolConfig{ + Gateway: "203.0.113.1", + PrefixLen: 28, + RangeStart: "203.0.113.2", + RangeEnd: "203.0.113.14", + }, + wantOK: false, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + start, end, pfx, ok := gwLrpRange(tc.pool) + if ok != tc.wantOK { + t.Fatalf("ok=%v, want %v", ok, tc.wantOK) + } + if !ok { + return + } + if start.String() != tc.wantStart { + t.Errorf("start=%s, want %s", start, tc.wantStart) + } + if end.String() != tc.wantEnd { + t.Errorf("end=%s, want %s", end, tc.wantEnd) + } + if pfx != tc.wantPfx { + t.Errorf("prefix=%d, want %d", pfx, tc.wantPfx) + } + }) + } +} From f5c9ad59cfa2d23bb269a855e00c8c827b6964f5 Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Thu, 30 Apr 2026 21:26:06 +1000 Subject: [PATCH 2/7] feat(vpcd): DHCP-acquire gateway LRP IPs (mulga-siv-38) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per-VPC OVN gateway LRPs now obtain their WAN-subnet IP via vpc.dhcp.acquire when the pool is source="dhcp", instead of carving from a static gw_lrp_range. The upstream DHCP server becomes the single arbiter of LAN address space; a cluster grows by N nodes without operator-side range carving in spinifex.toml. - Consolidate ObtainDHCPLease/ReleaseDHCPLease into the dhcp package as RequestAcquire/RequestRelease so vpcd and the daemon-side ExternalIPAM share one helper. handlers/ec2/vpc keeps its existing ENI/EIP DHCP path. - TopologyHandler dispatches by pool source: DHCP pools call DHCPManager via NATS (client-id "gw-lrp-", idempotent on retry); static pools keep the siv-36 gw_lrp_range / auto-derive allocator. Direct mode and no-pool deployments still use link-local. - IGW detach issues vpc.dhcp.release so the upstream server reclaims the IP. Best-effort — server-side expiry is the safety net. - Tests: TestTopologyHandler_IGWAttach_DHCPGwLrpIP (3 VPCs share one DHCP pool, distinct IPs, idempotent attach, release on detach), IGWAttach_DHCPAcquireFailureAborts (no link-local fallback when DHCP errors), TestPrefixFromMask. Existing static / auto-derive tests preserved. --- cmd/spinifex/cmd/service.go | 2 + spinifex/handlers/ec2/vpc/external_ipam.go | 19 +- .../handlers/ec2/vpc/external_ipam_dhcp.go | 160 ----------- .../handlers/ec2/vpc/external_ipam_test.go | 94 ------- spinifex/services/vpcd/dhcp/nats_client.go | 150 ++++++++++ .../services/vpcd/dhcp/nats_client_test.go | 104 +++++++ spinifex/services/vpcd/topology.go | 119 ++++++-- spinifex/services/vpcd/topology_test.go | 260 ++++++++++++++++++ spinifex/services/vpcd/vpcd.go | 8 + 9 files changed, 636 insertions(+), 280 deletions(-) delete mode 100644 spinifex/handlers/ec2/vpc/external_ipam_dhcp.go create mode 100644 spinifex/services/vpcd/dhcp/nats_client.go create mode 100644 spinifex/services/vpcd/dhcp/nats_client_test.go diff --git a/cmd/spinifex/cmd/service.go b/cmd/spinifex/cmd/service.go index cc9ddbfcc..6cdc7cc3f 100644 --- a/cmd/spinifex/cmd/service.go +++ b/cmd/spinifex/cmd/service.go @@ -735,6 +735,7 @@ var vpcdStartCmd = &cobra.Command{ for _, p := range clusterConfig.Network.ExternalPools { extPools = append(extPools, vpcd.ExternalPoolConfig{ Name: p.Name, + Source: p.Source, RangeStart: p.RangeStart, RangeEnd: p.RangeEnd, Gateway: p.Gateway, @@ -743,6 +744,7 @@ var vpcdStartCmd = &cobra.Command{ DNSServers: p.DNSServers, Region: p.Region, AZ: p.AZ, + DhcpBindBridge: nodeConfig.VPCD.DhcpBindBridge, GwLrpRangeStart: p.GwLrpRangeStart, GwLrpRangeEnd: p.GwLrpRangeEnd, }) diff --git a/spinifex/handlers/ec2/vpc/external_ipam.go b/spinifex/handlers/ec2/vpc/external_ipam.go index c336af7ef..b47ea14f7 100644 --- a/spinifex/handlers/ec2/vpc/external_ipam.go +++ b/spinifex/handlers/ec2/vpc/external_ipam.go @@ -9,6 +9,7 @@ import ( "net" "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -28,7 +29,7 @@ type ExternalIPAllocation struct { Note string `json:"note,omitempty"` // Human-readable note // DHCP-sourced metadata — set only for allocations whose pool has - // source="dhcp" (see ObtainDHCPLease). These fields let describe + // source="dhcp" (see dhcp.RequestAcquire). These fields let describe // surfaces and operators trace a given IP back to the upstream lease // without re-querying vpcd. LeaseExpiresUnix int64 `json:"lease_expires_unix,omitempty"` // Absolute expiry from the server's ACK @@ -173,9 +174,9 @@ func (m *ExternalIPAM) initPool(pool ExternalPoolConfig) error { // For DHCP pools, obtain the gateway IP from router DHCP if not set. // The gateway lease is long-lived: vpcd's renewal goroutine keeps it // refreshed until the pool is torn down. - var gatewayLease *dhcpLeaseResult + var gatewayLease *dhcp.LeaseResult if gwIP == "" && pool.IsDHCP() { - lease, dhcpErr := ObtainDHCPLease(m.nc, pool.DhcpBindBridge, + lease, dhcpErr := dhcp.RequestAcquire(m.nc, pool.DhcpBindBridge, "gateway-"+pool.Name, "gateway-"+pool.Name, "mulga-spinifex-gw", @@ -267,10 +268,10 @@ func (m *ExternalIPAM) allocateFromPool(poolName, allocType, allocID, eniID, ins // Manager is idempotent on ClientID, so CAS retries below will not // trigger a second DORA — we only release at the end if no CAS // attempt succeeded. - var dhcpLease *dhcpLeaseResult + var dhcpLease *dhcp.LeaseResult if pool != nil && pool.IsDHCP() { hostname, vendorClass := dhcpIdentityOptions(eniID, instanceID, poolName) - lease, err := ObtainDHCPLease(m.nc, pool.DhcpBindBridge, clientID, hostname, vendorClass, poolName) + lease, err := dhcp.RequestAcquire(m.nc, pool.DhcpBindBridge, clientID, hostname, vendorClass, poolName) if err != nil { return "", fmt.Errorf("DHCP lease for %s: %w", clientID, err) } @@ -281,7 +282,7 @@ func (m *ExternalIPAM) allocateFromPool(poolName, allocType, allocID, eniID, ins record, revision, err := m.getRecord(poolName) if err != nil { if dhcpLease != nil { - _ = ReleaseDHCPLease(m.nc, clientID) + _ = dhcp.RequestRelease(m.nc, clientID) } return "", fmt.Errorf("get external IPAM record: %w", err) } @@ -313,7 +314,7 @@ func (m *ExternalIPAM) allocateFromPool(poolName, allocType, allocID, eniID, ins data, err := json.Marshal(record) if err != nil { if dhcpLease != nil { - _ = ReleaseDHCPLease(m.nc, clientID) + _ = dhcp.RequestRelease(m.nc, clientID) } return "", fmt.Errorf("marshal external IPAM record: %w", err) } @@ -334,7 +335,7 @@ func (m *ExternalIPAM) allocateFromPool(poolName, allocType, allocID, eniID, ins // All CAS attempts exhausted — release any DHCP lease we acquired so // vpcd doesn't hold a binding for an IP we won't use. if dhcpLease != nil { - _ = ReleaseDHCPLease(m.nc, clientID) + _ = dhcp.RequestRelease(m.nc, clientID) } return "", fmt.Errorf("external IPAM allocation failed after CAS retries for pool %s", poolName) } @@ -389,7 +390,7 @@ func (m *ExternalIPAM) ReleaseIP(poolName, ip string) error { if clientID == "" { clientID = alloc.InstanceId } - if releaseErr := ReleaseDHCPLease(m.nc, clientID); releaseErr != nil { + if releaseErr := dhcp.RequestRelease(m.nc, clientID); releaseErr != nil { slog.Warn("Failed to release DHCP lease", "pool", poolName, "ip", ip, "err", releaseErr) } } diff --git a/spinifex/handlers/ec2/vpc/external_ipam_dhcp.go b/spinifex/handlers/ec2/vpc/external_ipam_dhcp.go deleted file mode 100644 index ea3473c83..000000000 --- a/spinifex/handlers/ec2/vpc/external_ipam_dhcp.go +++ /dev/null @@ -1,160 +0,0 @@ -package handlers_ec2_vpc - -import ( - "encoding/json" - "fmt" - "log/slog" - "time" - - "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" - "github.com/nats-io/nats.go" -) - -// dhcpNATSTimeout bounds the request/reply to spinifex-vpcd. Picked to -// comfortably exceed the DORA handshake (15 s default + server jitter + -// ~10 s of slack for NAK + fresh DISCOVER fallbacks). A var rather than -// a const so tests can drive the timeout error path without waiting -// 20 s per table entry. -var dhcpNATSTimeout = 30 * time.Second - -// dhcpAcquireMaxAttempts caps retry of vpc.dhcp.acquire. 60 attempts × 1 s -// sleep covers a cold-boot vpcd start window where NATS is up but the -// dhcp.Manager hasn't yet subscribed (typical: 5–30 s, pathological: 60 s -// during full reconcile). Tests override to keep timeout-path coverage -// fast — see external_ipam_test.go. -var dhcpAcquireMaxAttempts = 60 - -// dhcpAcquireRetryDelay is the wait between retries. Var for the same -// reason as dhcpAcquireMaxAttempts. -var dhcpAcquireRetryDelay = 1 * time.Second - -// dhcpLeaseResult is the data ExternalIPAM needs from a successful acquire. -// It maps directly from dhcp.AcquireReplyMsg; kept as a handler-side type -// so the IPAM ledger records the fields without coupling to the NATS wire -// struct. -type dhcpLeaseResult struct { - IP string - SubnetMask string - Routers []string - DNS []string - ServerID string - HWAddr string - ExpiresUnix int64 -} - -// ObtainDHCPLease asks spinifex-vpcd to acquire a DHCP lease on the given -// bridge, identifying us with option 61 (client-id), option 12 (hostname) -// and option 60 (vendor class). Blocks until vpcd replies or -// dhcpNATSTimeout expires. The Manager-side handler is idempotent: a -// second call with the same clientID while a live lease exists returns -// the same lease without a fresh DORA, so CAS retry loops on the caller -// are safe. -func ObtainDHCPLease(nc *nats.Conn, bridge, clientID, hostname, vendorClass, poolName string) (dhcpLeaseResult, error) { - if nc == nil { - return dhcpLeaseResult{}, fmt.Errorf("DHCP lease: NATS connection is required") - } - if bridge == "" { - return dhcpLeaseResult{}, fmt.Errorf("DHCP lease: bridge name is required") - } - if clientID == "" { - return dhcpLeaseResult{}, fmt.Errorf("DHCP lease: client ID is required") - } - - req := dhcp.AcquireRequestMsg{ - Bridge: bridge, - ClientID: clientID, - Hostname: hostname, - VendorClass: vendorClass, - PoolName: poolName, - } - data, err := json.Marshal(req) - if err != nil { - return dhcpLeaseResult{}, fmt.Errorf("marshal dhcp acquire request: %w", err) - } - - var reply dhcp.AcquireReplyMsg - maxAttempts := dhcpAcquireMaxAttempts - - for attempt := 1; attempt <= maxAttempts; attempt++ { - msg, err := nc.Request(dhcp.TopicAcquire, data, dhcpNATSTimeout) - if err != nil { - if attempt == maxAttempts { - return dhcpLeaseResult{}, fmt.Errorf( - "dhcp acquire NATS request failed after %d attempts (client %s): %w", - maxAttempts, - clientID, - err, - ) - } - - slog.Warn( - "dhcp acquire NATS request failed, retrying", - "clientID", clientID, - "attempt", attempt, - "maxAttempts", maxAttempts, - "err", err, - ) - - time.Sleep(dhcpAcquireRetryDelay) - continue - } - - if err := json.Unmarshal(msg.Data, &reply); err != nil { - return dhcpLeaseResult{}, fmt.Errorf("unmarshal dhcp acquire reply: %w", err) - } - - if reply.Error != "" { - return dhcpLeaseResult{}, fmt.Errorf("dhcp acquire client %s: %s", clientID, reply.Error) - } - - break - } - - slog.Info("DHCP lease obtained", - "bridge", bridge, - "client_id", clientID, - "ip", reply.IP, - "server_id", reply.ServerID, - "expires_unix", reply.ExpiresUnix, - ) - - return dhcpLeaseResult{ - IP: reply.IP, - SubnetMask: reply.SubnetMask, - Routers: reply.Routers, - DNS: reply.DNS, - ServerID: reply.ServerID, - HWAddr: reply.HWAddr, - ExpiresUnix: reply.ExpiresUnix, - }, nil -} - -// ReleaseDHCPLease asks spinifex-vpcd to release the lease identified by -// clientID. Returns nil (silently) when nc or clientID is empty so that -// callers in clean-up paths can invoke it unconditionally. -func ReleaseDHCPLease(nc *nats.Conn, clientID string) error { - if nc == nil || clientID == "" { - return nil - } - - data, err := json.Marshal(dhcp.ReleaseRequestMsg{ClientID: clientID}) - if err != nil { - return fmt.Errorf("marshal dhcp release request: %w", err) - } - - msg, err := nc.Request(dhcp.TopicRelease, data, dhcpNATSTimeout) - if err != nil { - return fmt.Errorf("dhcp release NATS request (client %s): %w", clientID, err) - } - - var reply dhcp.ReleaseReplyMsg - if err := json.Unmarshal(msg.Data, &reply); err != nil { - return fmt.Errorf("unmarshal dhcp release reply: %w", err) - } - if reply.Error != "" { - return fmt.Errorf("dhcp release (client %s): %s", clientID, reply.Error) - } - - slog.Info("DHCP lease released", "client_id", clientID) - return nil -} diff --git a/spinifex/handlers/ec2/vpc/external_ipam_test.go b/spinifex/handlers/ec2/vpc/external_ipam_test.go index cbce5d810..125ddfd5e 100644 --- a/spinifex/handlers/ec2/vpc/external_ipam_test.go +++ b/spinifex/handlers/ec2/vpc/external_ipam_test.go @@ -591,100 +591,6 @@ func TestExternalIPAM_DHCPGatewayFromLease(t *testing.T) { assert.Equal(t, "mulga-spinifex-gw", got.VendorClass) } -func TestObtainDHCPLease_RequestTimeout(t *testing.T) { - _, nc := testutil.StartTestNATS(t) - // Stub never responds — verifies that a hung vpcd surfaces as a - // wrapped NATS timeout error rather than hanging the caller. - sub, err := nc.Subscribe(dhcp.TopicAcquire, func(*nats.Msg) {}) - require.NoError(t, err) - t.Cleanup(func() { _ = sub.Unsubscribe() }) - - prev := dhcpNATSTimeout - dhcpNATSTimeout = 150 * time.Millisecond - t.Cleanup(func() { dhcpNATSTimeout = prev }) - - prevMax := dhcpAcquireMaxAttempts - prevDelay := dhcpAcquireRetryDelay - dhcpAcquireMaxAttempts = 2 - dhcpAcquireRetryDelay = 10 * time.Millisecond - t.Cleanup(func() { - dhcpAcquireMaxAttempts = prevMax - dhcpAcquireRetryDelay = prevDelay - }) - - _, err = ObtainDHCPLease(nc, "br-wan", "eni-timeout", "eni-timeout", "mulga-spinifex", "wan") - assert.ErrorContains(t, err, "dhcp acquire NATS request") -} - -func TestReleaseDHCPLease_RequestTimeout(t *testing.T) { - _, nc := testutil.StartTestNATS(t) - sub, err := nc.Subscribe(dhcp.TopicRelease, func(*nats.Msg) {}) - require.NoError(t, err) - t.Cleanup(func() { _ = sub.Unsubscribe() }) - - prev := dhcpNATSTimeout - dhcpNATSTimeout = 150 * time.Millisecond - t.Cleanup(func() { dhcpNATSTimeout = prev }) - - err = ReleaseDHCPLease(nc, "eni-timeout") - assert.ErrorContains(t, err, "dhcp release NATS request") -} - -func TestObtainDHCPLease_MalformedReply(t *testing.T) { - _, nc := testutil.StartTestNATS(t) - sub, err := nc.Subscribe(dhcp.TopicAcquire, func(msg *nats.Msg) { - _ = msg.Respond([]byte("not json")) - }) - require.NoError(t, err) - t.Cleanup(func() { _ = sub.Unsubscribe() }) - - _, err = ObtainDHCPLease(nc, "br-wan", "eni-malformed", "eni-malformed", "mulga-spinifex", "wan") - assert.ErrorContains(t, err, "unmarshal dhcp acquire reply") -} - -func TestReleaseDHCPLease_MalformedReply(t *testing.T) { - _, nc := testutil.StartTestNATS(t) - sub, err := nc.Subscribe(dhcp.TopicRelease, func(msg *nats.Msg) { - _ = msg.Respond([]byte("not json")) - }) - require.NoError(t, err) - t.Cleanup(func() { _ = sub.Unsubscribe() }) - - err = ReleaseDHCPLease(nc, "eni-malformed") - assert.ErrorContains(t, err, "unmarshal dhcp release reply") -} - -func TestReleaseDHCPLease_ReplyError(t *testing.T) { - _, nc := testutil.StartTestNATS(t) - sub, err := nc.Subscribe(dhcp.TopicRelease, func(msg *nats.Msg) { - data, _ := json.Marshal(dhcp.ReleaseReplyMsg{Error: "vpcd rejected release"}) - _ = msg.Respond(data) - }) - require.NoError(t, err) - t.Cleanup(func() { _ = sub.Unsubscribe() }) - - err = ReleaseDHCPLease(nc, "eni-rejected") - assert.ErrorContains(t, err, "vpcd rejected release") -} - -func TestObtainDHCPLease_Guards(t *testing.T) { - _, err := ObtainDHCPLease(nil, "br-wan", "eni-1", "eni-1", "mulga-spinifex", "wan") - assert.ErrorContains(t, err, "NATS connection is required") - - _, nc := testutil.StartTestNATS(t) - _, err = ObtainDHCPLease(nc, "", "eni-1", "eni-1", "mulga-spinifex", "wan") - assert.ErrorContains(t, err, "bridge name is required") - - _, err = ObtainDHCPLease(nc, "br-wan", "", "eni-1", "mulga-spinifex", "wan") - assert.ErrorContains(t, err, "client ID is required") -} - -func TestReleaseDHCPLease_NoopWhenNilOrEmpty(t *testing.T) { - assert.NoError(t, ReleaseDHCPLease(nil, "eni-1")) - _, nc := testutil.StartTestNATS(t) - assert.NoError(t, ReleaseDHCPLease(nc, "")) -} - func TestExternalIPAM_DHCPGatewayErrorPropagates(t *testing.T) { _, nc, js := testutil.StartTestJetStream(t) stub := newDHCPStub(t, nc) diff --git a/spinifex/services/vpcd/dhcp/nats_client.go b/spinifex/services/vpcd/dhcp/nats_client.go new file mode 100644 index 000000000..e09ffe89f --- /dev/null +++ b/spinifex/services/vpcd/dhcp/nats_client.go @@ -0,0 +1,150 @@ +package dhcp + +import ( + "encoding/json" + "fmt" + "log/slog" + "time" + + "github.com/nats-io/nats.go" +) + +// NATSTimeout bounds the request/reply round-trip to vpcd's DHCPManager. +// Picked to comfortably exceed the DORA handshake (15 s default + server +// jitter + ~10 s of slack for NAK + fresh DISCOVER fallbacks). A var rather +// than a const so tests can drive the timeout error path without waiting +// 20 s per table entry. +var NATSTimeout = 30 * time.Second + +// AcquireMaxAttempts caps retry of vpc.dhcp.acquire. 60 attempts × 1 s sleep +// covers a cold-boot vpcd start window where NATS is up but the dhcp.Manager +// hasn't yet subscribed (typical: 5–30 s, pathological: 60 s during full +// reconcile). Tests override to keep timeout-path coverage fast. +var AcquireMaxAttempts = 60 + +// AcquireRetryDelay is the wait between retries. Var for the same reason as +// AcquireMaxAttempts. +var AcquireRetryDelay = 1 * time.Second + +// LeaseResult is the data callers need from a successful acquire. Mirrors +// AcquireReplyMsg but exposed as a typed result so callers don't depend on +// the wire struct. +type LeaseResult struct { + IP string + SubnetMask string + Routers []string + DNS []string + ServerID string + HWAddr string + ExpiresUnix int64 +} + +// RequestAcquire asks vpcd's DHCPManager (over NATS) to acquire a DHCP lease +// on the given bridge, identifying the caller with option 61 (client-id), +// option 12 (hostname) and option 60 (vendor class). Blocks until vpcd +// replies or NATSTimeout expires. The Manager-side handler is idempotent on +// clientID: a second call with the same clientID while a live lease exists +// returns the same lease without a fresh DORA, so CAS retry loops on the +// caller are safe. +func RequestAcquire(nc *nats.Conn, bridge, clientID, hostname, vendorClass, poolName string) (LeaseResult, error) { + if nc == nil { + return LeaseResult{}, fmt.Errorf("DHCP lease: NATS connection is required") + } + if bridge == "" { + return LeaseResult{}, fmt.Errorf("DHCP lease: bridge name is required") + } + if clientID == "" { + return LeaseResult{}, fmt.Errorf("DHCP lease: client ID is required") + } + + req := AcquireRequestMsg{ + Bridge: bridge, + ClientID: clientID, + Hostname: hostname, + VendorClass: vendorClass, + PoolName: poolName, + } + data, err := json.Marshal(req) + if err != nil { + return LeaseResult{}, fmt.Errorf("marshal dhcp acquire request: %w", err) + } + + var reply AcquireReplyMsg + maxAttempts := AcquireMaxAttempts + + for attempt := 1; attempt <= maxAttempts; attempt++ { + msg, err := nc.Request(TopicAcquire, data, NATSTimeout) + if err != nil { + if attempt == maxAttempts { + return LeaseResult{}, fmt.Errorf( + "dhcp acquire NATS request failed after %d attempts (client %s): %w", + maxAttempts, clientID, err, + ) + } + slog.Warn("dhcp acquire NATS request failed, retrying", + "clientID", clientID, + "attempt", attempt, + "maxAttempts", maxAttempts, + "err", err, + ) + time.Sleep(AcquireRetryDelay) + continue + } + + if err := json.Unmarshal(msg.Data, &reply); err != nil { + return LeaseResult{}, fmt.Errorf("unmarshal dhcp acquire reply: %w", err) + } + if reply.Error != "" { + return LeaseResult{}, fmt.Errorf("dhcp acquire client %s: %s", clientID, reply.Error) + } + break + } + + slog.Info("DHCP lease obtained", + "bridge", bridge, + "client_id", clientID, + "ip", reply.IP, + "server_id", reply.ServerID, + "expires_unix", reply.ExpiresUnix, + ) + + return LeaseResult{ + IP: reply.IP, + SubnetMask: reply.SubnetMask, + Routers: reply.Routers, + DNS: reply.DNS, + ServerID: reply.ServerID, + HWAddr: reply.HWAddr, + ExpiresUnix: reply.ExpiresUnix, + }, nil +} + +// RequestRelease asks vpcd to release the lease identified by clientID. +// Returns nil silently when nc or clientID is empty so callers in clean-up +// paths can invoke it unconditionally. +func RequestRelease(nc *nats.Conn, clientID string) error { + if nc == nil || clientID == "" { + return nil + } + + data, err := json.Marshal(ReleaseRequestMsg{ClientID: clientID}) + if err != nil { + return fmt.Errorf("marshal dhcp release request: %w", err) + } + + msg, err := nc.Request(TopicRelease, data, NATSTimeout) + if err != nil { + return fmt.Errorf("dhcp release NATS request (client %s): %w", clientID, err) + } + + var reply ReleaseReplyMsg + if err := json.Unmarshal(msg.Data, &reply); err != nil { + return fmt.Errorf("unmarshal dhcp release reply: %w", err) + } + if reply.Error != "" { + return fmt.Errorf("dhcp release (client %s): %s", clientID, reply.Error) + } + + slog.Info("DHCP lease released", "client_id", clientID) + return nil +} diff --git a/spinifex/services/vpcd/dhcp/nats_client_test.go b/spinifex/services/vpcd/dhcp/nats_client_test.go new file mode 100644 index 000000000..da4993ccb --- /dev/null +++ b/spinifex/services/vpcd/dhcp/nats_client_test.go @@ -0,0 +1,104 @@ +package dhcp + +import ( + "encoding/json" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestRequestAcquire_RequestTimeout(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + sub, err := nc.Subscribe(TopicAcquire, func(*nats.Msg) {}) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + prev := NATSTimeout + NATSTimeout = 150 * time.Millisecond + t.Cleanup(func() { NATSTimeout = prev }) + + prevMax := AcquireMaxAttempts + prevDelay := AcquireRetryDelay + AcquireMaxAttempts = 2 + AcquireRetryDelay = 10 * time.Millisecond + t.Cleanup(func() { + AcquireMaxAttempts = prevMax + AcquireRetryDelay = prevDelay + }) + + _, err = RequestAcquire(nc, "br-wan", "eni-timeout", "eni-timeout", "mulga-spinifex", "wan") + assert.ErrorContains(t, err, "dhcp acquire NATS request") +} + +func TestRequestRelease_RequestTimeout(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + sub, err := nc.Subscribe(TopicRelease, func(*nats.Msg) {}) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + prev := NATSTimeout + NATSTimeout = 150 * time.Millisecond + t.Cleanup(func() { NATSTimeout = prev }) + + err = RequestRelease(nc, "eni-timeout") + assert.ErrorContains(t, err, "dhcp release NATS request") +} + +func TestRequestAcquire_MalformedReply(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + sub, err := nc.Subscribe(TopicAcquire, func(msg *nats.Msg) { + _ = msg.Respond([]byte("not json")) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + _, err = RequestAcquire(nc, "br-wan", "eni-malformed", "eni-malformed", "mulga-spinifex", "wan") + assert.ErrorContains(t, err, "unmarshal dhcp acquire reply") +} + +func TestRequestRelease_MalformedReply(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + sub, err := nc.Subscribe(TopicRelease, func(msg *nats.Msg) { + _ = msg.Respond([]byte("not json")) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + err = RequestRelease(nc, "eni-malformed") + assert.ErrorContains(t, err, "unmarshal dhcp release reply") +} + +func TestRequestRelease_ReplyError(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + sub, err := nc.Subscribe(TopicRelease, func(msg *nats.Msg) { + data, _ := json.Marshal(ReleaseReplyMsg{Error: "vpcd rejected release"}) + _ = msg.Respond(data) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + err = RequestRelease(nc, "eni-rejected") + assert.ErrorContains(t, err, "vpcd rejected release") +} + +func TestRequestAcquire_Guards(t *testing.T) { + _, err := RequestAcquire(nil, "br-wan", "eni-1", "eni-1", "mulga-spinifex", "wan") + assert.ErrorContains(t, err, "NATS connection is required") + + _, nc := testutil.StartTestNATS(t) + _, err = RequestAcquire(nc, "", "eni-1", "eni-1", "mulga-spinifex", "wan") + assert.ErrorContains(t, err, "bridge name is required") + + _, err = RequestAcquire(nc, "br-wan", "", "eni-1", "mulga-spinifex", "wan") + assert.ErrorContains(t, err, "client ID is required") +} + +func TestRequestRelease_NoopWhenNilOrEmpty(t *testing.T) { + assert.NoError(t, RequestRelease(nil, "eni-1")) + _, nc := testutil.StartTestNATS(t) + assert.NoError(t, RequestRelease(nc, "")) +} diff --git a/spinifex/services/vpcd/topology.go b/spinifex/services/vpcd/topology.go index 85bc21dba..4afff94d2 100644 --- a/spinifex/services/vpcd/topology.go +++ b/spinifex/services/vpcd/topology.go @@ -9,6 +9,7 @@ import ( "strconv" "strings" + "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" "github.com/mulgadc/spinifex/spinifex/services/vpcd/nbdb" "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" @@ -117,6 +118,11 @@ type TopologyHandler struct { externalPools []ExternalPoolConfig chassisNames []string // OVN chassis names for gateway HA scheduling bridgeMode string // "direct" or "macvlan" — controls NAT mode and localnet options + // nc is used to talk to vpcd's DHCPManager via vpc.dhcp.acquire / + // vpc.dhcp.release for gateway LRP IPs in centralized NAT on + // source="dhcp" pools (mulga-siv-38). nil when no DHCP pool is wired or + // the test stack supplies a static-only mock. + nc *nats.Conn } // NewTopologyHandler creates a new TopologyHandler with optional external network config. @@ -155,6 +161,15 @@ func WithBridgeMode(mode string) TopologyOption { } } +// WithNATSConn wires the NATS connection used to talk to vpcd's DHCPManager +// for gateway-LRP DHCP-acquire (mulga-siv-38). Tests with mock OVN clients +// can omit this — the static / auto-derive paths still work. +func WithNATSConn(nc *nats.Conn) TopologyOption { + return func(h *TopologyHandler) { + h.nc = nc + } +} + // isMacvlanMode returns true if the external bridge uses a macvlan interface. // This is the default when bridgeMode is unset for backward compatibility. func (h *TopologyHandler) isMacvlanMode() bool { @@ -213,11 +228,12 @@ func (h *TopologyHandler) ensureLocalnetOptions(ctx context.Context, extPortName } // expectedGatewayPortNetwork returns the Networks CIDR a gateway LRP for -// vpcId should carry. In direct mode, or in centralized mode without a -// configured gw_lrp_range, it is the link-local fallback. In centralized -// mode with a range, it is the existing spinifex:gateway_ip on the LRP if -// any (idempotent), otherwise a fresh allocation. Returns the new gw IP -// (empty when link-local) so callers can persist it to external_ids. +// vpcId should carry. In direct mode, or in centralized mode without a pool, +// it is the link-local fallback. In centralized mode with a DHCP pool the +// gateway IP is acquired via vpc.dhcp.acquire (mulga-siv-38). With a static +// pool the IP comes from gw_lrp_range or the auto-derived top-of-subnet +// block (mulga-siv-36). Returns the gw IP (empty when link-local) so +// callers can persist it to external_ids. func (h *TopologyHandler) expectedGatewayPortNetwork(ctx context.Context, vpcId string) (network, gwIP string, err error) { if !h.useCentralizedNAT() { return gatewayPortNetwork, "", nil @@ -226,6 +242,13 @@ func (h *TopologyHandler) expectedGatewayPortNetwork(ctx context.Context, vpcId if pool == nil { return gatewayPortNetwork, "", nil } + if pool.IsDHCP() { + ip, prefix, dhcpErr := h.allocateGatewayLRPIPViaDHCP(ctx, vpcId, pool) + if dhcpErr != nil { + return "", "", dhcpErr + } + return fmt.Sprintf("%s/%d", ip, prefix), ip, nil + } ip, prefix, ok, allocErr := h.allocateGatewayLRPIP(ctx, vpcId, pool) if allocErr != nil { return "", "", allocErr @@ -887,22 +910,20 @@ func (h *TopologyHandler) handleIGWAttach(msg *nats.Msg) { case pool != nil: wanGateway = pool.Gateway if h.useCentralizedNAT() { - ip, prefix, allocOK, allocErr := h.allocateGatewayLRPIP(ctx, evt.VpcId, pool) - if allocErr != nil || !allocOK { + network, ip, allocErr := h.expectedGatewayPortNetwork(ctx, evt.VpcId) + if allocErr != nil { slog.Error("vpcd: failed to allocate gw LRP IP for centralized NAT", "vpc_id", evt.VpcId, "pool", pool.Name, "err", allocErr) _ = h.ovn.DeleteLogicalSwitch(ctx, extSwitchName) - if allocErr == nil { - allocErr = fmt.Errorf("no gw LRP IP available in pool %q (configure gw_lrp_range or fix Gateway/PrefixLen)", pool.Name) - } respond(msg, allocErr) return } gwLrpIP = ip - gatewayNetwork = fmt.Sprintf("%s/%d", ip, prefix) + gatewayNetwork = network } slog.Info("vpcd: using external pool for IGW", "pool", pool.Name, + "source", pool.Source, "lrp_network", gatewayNetwork, "wan_gateway", wanGateway, ) @@ -1072,6 +1093,10 @@ func (h *TopologyHandler) handleIGWDetach(msg *nats.Msg) { return } + // 7. Release any DHCP-acquired gateway LRP lease (mulga-siv-38). Best- + // effort — upstream server expires the lease on its own if this fails. + h.releaseGatewayLRPLease(evt.VpcId) + slog.Info("vpcd: detached internet gateway from VPC", "igw_id", evt.InternetGatewayId, "vpc_id", evt.VpcId, @@ -1293,15 +1318,12 @@ func (h *TopologyHandler) reconcileIGW(ctx context.Context, vpcId, igwId string) if pool != nil { wanGateway = pool.Gateway if h.useCentralizedNAT() { - ip, prefix, allocOK, allocErr := h.allocateGatewayLRPIP(ctx, vpcId, pool) - if allocErr != nil || !allocOK { - if allocErr == nil { - allocErr = fmt.Errorf("no gw LRP IP available in pool %q", pool.Name) - } + network, ip, allocErr := h.expectedGatewayPortNetwork(ctx, vpcId) + if allocErr != nil { return fmt.Errorf("allocate gw LRP IP for %s: %w", vpcId, allocErr) } gwLrpIP = ip - gatewayNetwork = fmt.Sprintf("%s/%d", ip, prefix) + gatewayNetwork = network } } @@ -1593,6 +1615,69 @@ func gwLrpRange(pool *ExternalPoolConfig) (start, end net.IP, prefix int, ok boo return uint32ToIPv4(autoStartU), uint32ToIPv4(autoEndU), prefix, true } +// gwLrpClientID is the DHCP client-id used for a VPC's gateway LRP lease. +// Stable across reboots — DHCPManager reuses the same lease on idempotent +// re-attach so the LRP IP doesn't change unless the upstream server +// reassigns it. +func gwLrpClientID(vpcId string) string { return "gw-lrp-" + vpcId } + +// allocateGatewayLRPIPViaDHCP requests a DHCP lease from vpcd's DHCPManager +// for the gateway LRP of vpcId. The handler-side acquire is idempotent on +// client-id so retries return the same IP without a fresh DORA. Prefix +// comes from the lease's SubnetMask, falling back to pool.PrefixLen when +// the server omits option 1 (rare; should not happen in practice). +func (h *TopologyHandler) allocateGatewayLRPIPViaDHCP(ctx context.Context, vpcId string, pool *ExternalPoolConfig) (ip string, prefix int, err error) { + _ = ctx + if h.nc == nil { + return "", 0, fmt.Errorf("vpcd: no NATS conn for DHCP gw LRP allocation (vpc %s, pool %s)", vpcId, pool.Name) + } + clientID := gwLrpClientID(vpcId) + hostname := "spinifex-gw-" + vpcId + vendorClass := "mulga-spinifex-gw-lrp" + lease, dhcpErr := dhcp.RequestAcquire(h.nc, pool.DhcpBindBridge, clientID, hostname, vendorClass, pool.Name) + if dhcpErr != nil { + return "", 0, fmt.Errorf("dhcp acquire gw LRP IP for vpc %s: %w", vpcId, dhcpErr) + } + prefix = prefixFromMask(lease.SubnetMask) + if prefix == 0 { + prefix = pool.PrefixLen + } + if prefix <= 0 || prefix > 32 { + return "", 0, fmt.Errorf("dhcp gw LRP for vpc %s: cannot determine prefix (mask=%q pool prefix=%d)", vpcId, lease.SubnetMask, pool.PrefixLen) + } + return lease.IP, prefix, nil +} + +// releaseGatewayLRPLease asks vpcd's DHCPManager to drop the lease held for +// vpcId's gateway LRP. Best-effort — log on failure but never block IGW +// detach, since the upstream server will eventually expire the lease on +// its own. +func (h *TopologyHandler) releaseGatewayLRPLease(vpcId string) { + if h.nc == nil { + return + } + if err := dhcp.RequestRelease(h.nc, gwLrpClientID(vpcId)); err != nil { + slog.Warn("vpcd: dhcp release for gw LRP failed", "vpc_id", vpcId, "err", err) + } +} + +// prefixFromMask converts a dotted-decimal mask ("255.255.255.0") to a +// prefix length. Returns 0 when the mask is empty or unparseable. +func prefixFromMask(mask string) int { + if mask == "" { + return 0 + } + ip := net.ParseIP(mask).To4() + if ip == nil { + return 0 + } + ones, bits := net.IPMask(ip).Size() + if bits != 32 { + return 0 + } + return ones +} + // allocateGatewayLRPIP picks the next free IP in pool.GwLrpRange for the // given VPC, scanning every existing LRP's spinifex:gateway_ip external_id // to compute the used set. If gw- already has an allocation the diff --git a/spinifex/services/vpcd/topology_test.go b/spinifex/services/vpcd/topology_test.go index b20f96dab..0627fb5f9 100644 --- a/spinifex/services/vpcd/topology_test.go +++ b/spinifex/services/vpcd/topology_test.go @@ -8,6 +8,7 @@ import ( "strings" "testing" + "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" "github.com/mulgadc/spinifex/spinifex/services/vpcd/nbdb" "github.com/mulgadc/spinifex/spinifex/testutil" "github.com/mulgadc/spinifex/spinifex/types" @@ -1221,6 +1222,265 @@ func TestTopologyHandler_IGWAttach_CentralizedNATAllocatesGwLrpIP(t *testing.T) } } +// dhcpStub is a minimal in-test responder for vpc.dhcp.acquire / release. +// Each acquire returns a distinct lease IP keyed off ClientID so multi-VPC +// tests get unique IPs without standing up a full DHCPManager. +type dhcpStub struct { + t *testing.T + leases map[string]string // client_id -> ip + released map[string]bool + nextOct int +} + +func newDHCPStub(t *testing.T, nc *nats.Conn) *dhcpStub { + t.Helper() + s := &dhcpStub{ + t: t, + leases: map[string]string{}, + released: map[string]bool{}, + nextOct: 101, + } + subA, err := nc.Subscribe(dhcp.TopicAcquire, func(msg *nats.Msg) { + var req dhcp.AcquireRequestMsg + if err := json.Unmarshal(msg.Data, &req); err != nil { + data, _ := json.Marshal(dhcp.AcquireReplyMsg{Error: err.Error()}) + _ = msg.Respond(data) + return + } + ip, ok := s.leases[req.ClientID] + if !ok { + ip = fmt.Sprintf("192.168.3.%d", s.nextOct) + s.nextOct++ + s.leases[req.ClientID] = ip + } + reply := dhcp.AcquireReplyMsg{ + IP: ip, + SubnetMask: "255.255.255.0", + ServerID: "192.168.3.1", + Routers: []string{"192.168.3.1"}, + HWAddr: "02:00:00:00:00:" + fmt.Sprintf("%02x", s.nextOct%256), + ExpiresUnix: 0, + } + data, _ := json.Marshal(reply) + _ = msg.Respond(data) + }) + if err != nil { + t.Fatalf("subscribe acquire: %v", err) + } + subR, err := nc.Subscribe(dhcp.TopicRelease, func(msg *nats.Msg) { + var req dhcp.ReleaseRequestMsg + _ = json.Unmarshal(msg.Data, &req) + s.released[req.ClientID] = true + delete(s.leases, req.ClientID) + data, _ := json.Marshal(dhcp.ReleaseReplyMsg{}) + _ = msg.Respond(data) + }) + if err != nil { + t.Fatalf("subscribe release: %v", err) + } + t.Cleanup(func() { + _ = subA.Unsubscribe() + _ = subR.Unsubscribe() + }) + return s +} + +// TestTopologyHandler_IGWAttach_DHCPGwLrpIP guards mulga-siv-38: with +// pool.Source="dhcp" the gateway LRP IP must come from a DHCP lease +// (vpc.dhcp.acquire), not the static gw_lrp_range allocator. Three VPCs +// sharing one DHCP pool must each get a distinct lease IP and persist it +// on external_ids:spinifex:gateway_ip. IGW detach must release the lease. +func TestTopologyHandler_IGWAttach_DHCPGwLrpIP(t *testing.T) { + _, nc := startTestNATS(t) + stub := newDHCPStub(t, nc) + mock := NewMockOVNClient() + _ = mock.Connect(context.Background()) + ctx := context.Background() + + pools := []ExternalPoolConfig{ + { + Name: "lan-dhcp", + Source: "dhcp", + Gateway: "192.168.3.1", + PrefixLen: 24, + DhcpBindBridge: "br-wan", + }, + } + topo := NewTopologyHandler(mock, + WithExternalNetwork("pool", pools), + WithBridgeMode(BridgeModeVeth), + WithNATSConn(nc), + ) + subs, err := topo.Subscribe(nc) + if err != nil { + t.Fatalf("subscribe: %v", err) + } + defer func() { + for _, s := range subs { + _ = s.Unsubscribe() + } + }() + + vpcIds := []string{"vpc-dh-a", "vpc-dh-b", "vpc-dh-c"} + for _, vpcId := range vpcIds { + _ = mock.CreateLogicalRouter(ctx, &nbdb.LogicalRouter{ + Name: "vpc-" + vpcId, + ExternalIDs: map[string]string{ + "spinifex:vpc_id": vpcId, + "spinifex:cidr": "10.0.0.0/16", + }, + }) + evt := types.IGWEvent{InternetGatewayId: "igw-" + vpcId, VpcId: vpcId} + data, _ := json.Marshal(evt) + resp, err := nc.Request(TopicIGWAttach, data, 5_000_000_000) + if err != nil { + t.Fatalf("request vpc.igw-attach %s: %v", vpcId, err) + } + assertSuccess(t, resp, "attach IGW "+vpcId) + } + + seen := make(map[string]string, len(vpcIds)) + for _, vpcId := range vpcIds { + gwPortName := "gw-" + vpcId + gwPort, err := mock.GetLogicalRouterPort(ctx, gwPortName) + if err != nil { + t.Fatalf("get %s: %v", gwPortName, err) + } + if len(gwPort.Networks) != 1 { + t.Fatalf("%s: expected 1 Network, got %v", gwPortName, gwPort.Networks) + } + ip, cidr, _ := strings.Cut(gwPort.Networks[0], "/") + if cidr != "24" { + t.Errorf("%s: expected /24 from DHCP lease mask, got /%s", gwPortName, cidr) + } + if !strings.HasPrefix(ip, "192.168.3.") { + t.Errorf("%s: expected 192.168.3.X from stub, got %s", gwPortName, ip) + } + gwIP := gwPort.ExternalIDs["spinifex:gateway_ip"] + if gwIP != ip { + t.Errorf("%s: external_ids gateway_ip=%q, expected %q", gwPortName, gwIP, ip) + } + if other, dup := seen[gwIP]; dup { + t.Errorf("gw IP collision: %s and %s share %s", gwPortName, other, gwIP) + } + seen[gwIP] = gwPortName + + // stub recorded the lease keyed by gw-lrp- + clientID := "gw-lrp-" + vpcId + if stub.leases[clientID] != ip { + t.Errorf("%s: stub lease %q != lrp IP %q", clientID, stub.leases[clientID], ip) + } + } + + // Detach one VPC, assert release reaches the stub. + detachVpc := vpcIds[0] + devt := types.IGWEvent{InternetGatewayId: "igw-" + detachVpc, VpcId: detachVpc} + ddata, _ := json.Marshal(devt) + resp, err := nc.Request(TopicIGWDetach, ddata, 5_000_000_000) + if err != nil { + t.Fatalf("request vpc.igw-detach: %v", err) + } + assertSuccess(t, resp, "detach IGW "+detachVpc) + if !stub.released["gw-lrp-"+detachVpc] { + t.Errorf("expected DHCP release for gw-lrp-%s, stub released=%v", detachVpc, stub.released) + } +} + +// TestTopologyHandler_IGWAttach_DHCPAcquireFailureAborts ensures that when +// vpcd's DHCPManager replies with an error, IGW attach surfaces the failure +// loud instead of falling back to link-local. Otherwise we silently restore +// the broken pre-siv-36 behavior. +func TestTopologyHandler_IGWAttach_DHCPAcquireFailureAborts(t *testing.T) { + _, nc := startTestNATS(t) + sub, err := nc.Subscribe(dhcp.TopicAcquire, func(msg *nats.Msg) { + reply := dhcp.AcquireReplyMsg{Error: "no leases available"} + data, _ := json.Marshal(reply) + _ = msg.Respond(data) + }) + if err != nil { + t.Fatalf("subscribe: %v", err) + } + defer func() { _ = sub.Unsubscribe() }() + + mock := NewMockOVNClient() + _ = mock.Connect(context.Background()) + ctx := context.Background() + + pools := []ExternalPoolConfig{ + { + Name: "lan-dhcp", + Source: "dhcp", + Gateway: "192.168.3.1", + PrefixLen: 24, + DhcpBindBridge: "br-wan", + }, + } + topo := NewTopologyHandler(mock, + WithExternalNetwork("pool", pools), + WithBridgeMode(BridgeModeVeth), + WithNATSConn(nc), + ) + subs, err := topo.Subscribe(nc) + if err != nil { + t.Fatalf("subscribe: %v", err) + } + defer func() { + for _, s := range subs { + _ = s.Unsubscribe() + } + }() + + _ = mock.CreateLogicalRouter(ctx, &nbdb.LogicalRouter{ + Name: "vpc-vpc-dh-fail", + ExternalIDs: map[string]string{ + "spinifex:vpc_id": "vpc-dh-fail", + "spinifex:cidr": "10.0.0.0/16", + }, + }) + evt := types.IGWEvent{InternetGatewayId: "igw-fail", VpcId: "vpc-dh-fail"} + data, _ := json.Marshal(evt) + resp, _ := nc.Request(TopicIGWAttach, data, 5_000_000_000) + var reply struct { + Error string `json:"error"` + } + _ = json.Unmarshal(resp.Data, &reply) + if reply.Error == "" { + t.Fatal("expected attach to fail when DHCP acquire errors, got success") + } + if !strings.Contains(reply.Error, "dhcp acquire") && !strings.Contains(reply.Error, "no leases available") { + t.Errorf("expected dhcp-related error, got %q", reply.Error) + } + // External switch must be rolled back. + if _, err := mock.GetLogicalSwitch(ctx, "ext-vpc-dh-fail"); err == nil { + t.Error("expected external switch rolled back after DHCP failure") + } +} + +// TestPrefixFromMask checks dotted-mask → prefix conversion across the +// boundary cases gw LRP DHCP relies on (option 1 missing, /32, /0, +// non-contiguous masks, IPv6). +func TestPrefixFromMask(t *testing.T) { + cases := []struct { + mask string + want int + }{ + {"", 0}, + {"255.255.255.0", 24}, + {"255.255.255.255", 32}, + {"0.0.0.0", 0}, + {"255.255.0.0", 16}, + {"not-an-ip", 0}, + } + for _, tc := range cases { + t.Run(tc.mask, func(t *testing.T) { + got := prefixFromMask(tc.mask) + if got != tc.want { + t.Errorf("prefixFromMask(%q)=%d, want %d", tc.mask, got, tc.want) + } + }) + } +} + func TestTopologyHandler_FindExternalPool(t *testing.T) { pools := []ExternalPoolConfig{ {Name: "az-a", Region: "us-east-1", AZ: "us-east-1a", RangeStart: "1.1.1.1"}, diff --git a/spinifex/services/vpcd/vpcd.go b/spinifex/services/vpcd/vpcd.go index dcdf11b92..f1266a4fe 100644 --- a/spinifex/services/vpcd/vpcd.go +++ b/spinifex/services/vpcd/vpcd.go @@ -81,6 +81,7 @@ type Config struct { // ExternalPoolConfig mirrors config.ExternalPool for vpcd's internal use. type ExternalPoolConfig struct { Name string + Source string // "static" (default) or "dhcp" RangeStart string RangeEnd string Gateway string @@ -89,10 +90,16 @@ type ExternalPoolConfig struct { DNSServers []string Region string AZ string + DhcpBindBridge string // Bridge where the DHCP AF_PACKET socket binds (e.g. "br-wan"). Required for source="dhcp". GwLrpRangeStart string // Sub-range for OVN gateway LRP IPs in centralized NAT mode (mulga-siv-36). GwLrpRangeEnd string } +// IsDHCP returns true if this pool obtains IPs from upstream DHCP. +func (p *ExternalPoolConfig) IsDHCP() bool { + return p.Source == "dhcp" +} + // Service implements the Spinifex service interface for vpcd. type Service struct { Config *Config @@ -344,6 +351,7 @@ func launchService(cfg *Config) error { topoOpts = append(topoOpts, WithChassisNames(chassisNames)) slog.Info("vpcd: gateway chassis discovered", "chassis", chassisNames) topoOpts = append(topoOpts, WithBridgeMode(bridgeMode)) + topoOpts = append(topoOpts, WithNATSConn(nc)) topo := NewTopologyHandler(liveClient, topoOpts...) // Elect a single vpcd to run startup reconcile. Without this, N vpcds in a From a6ef8e575b437abf8531881463685f9811315828 Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Thu, 30 Apr 2026 21:49:25 +1000 Subject: [PATCH 3/7] fix(vpcd): subscribe DHCPManager before reconcile to fix boot race (mulga-siv-38) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DHCPManager.Subscribe ran AFTER ReconcileFromKV / RetrofitAllGatewayPortNetworks, so retrofit's vpc.dhcp.acquire requests for existing IGWs hit "no responders" and burned the 60-attempt × 1 s cold-boot retry budget per VPC before failing. On a dev box with one bootstrap VPC the symptom is a 60 s vpcd startup hang followed by "allocate gw LRP IP failed", and the bootstrap IGW topology missing from OVN until the next reconcile pass. Move dhcpManager.Subscribe ahead of the two reconcile passes so the subscription is live before any topology code can fan out an acquire. --- spinifex/services/vpcd/vpcd.go | 24 +++++++++++++----------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/spinifex/services/vpcd/vpcd.go b/spinifex/services/vpcd/vpcd.go index f1266a4fe..225b021d7 100644 --- a/spinifex/services/vpcd/vpcd.go +++ b/spinifex/services/vpcd/vpcd.go @@ -394,18 +394,13 @@ func launchService(cfg *Config) error { } }() - // Pass 2: Reconcile from NATS KV (handles reboots, OVN DB loss, missed events). - // Runs after subscribing so new events are not missed during reconciliation. - // Leader-gated (mulga-siv-29) to avoid concurrent Create races against OVN NB. - if isLeader { - ReconcileFromKV(ctx, nc, topo, chassisNames) - } - // DHCP manager: services vpc.dhcp.acquire/release requests from the - // daemon-side ExternalIPAM handlers, runs a renewal goroutine per lease, - // and persists leases in spinifex-dhcp-leases KV. Started unconditionally - // — pools with source="static" never issue acquire requests, so the - // Manager sits idle until something actually wants DHCP. + // daemon-side ExternalIPAM handlers and (mulga-siv-38) topology's gw-LRP + // allocator. MUST subscribe before ReconcileFromKV / retrofit so those + // passes don't fan out vpc.dhcp.acquire requests to "no responders" and + // burn the 60-attempt cold-boot retry budget per existing IGW. Started + // unconditionally — pools with source="static" never issue acquire + // requests, so the Manager sits idle until something actually wants DHCP. js, err := nc.JetStream() if err != nil { slog.Error("Failed to get JetStream context for DHCP manager", "err", err) @@ -433,6 +428,13 @@ func launchService(cfg *Config) error { } }() + // Pass 2: Reconcile from NATS KV (handles reboots, OVN DB loss, missed events). + // Runs after subscribing so new events are not missed during reconciliation. + // Leader-gated (mulga-siv-29) to avoid concurrent Create races against OVN NB. + if isLeader { + ReconcileFromKV(ctx, nc, topo, chassisNames) + } + // Pass 3: Retrofit localnet options on every external switch. Walks OVN // directly so stale/missing KV records can't hide a stale nat-addresses // or cleared network_name. Idempotent; silent when all correct. From 714898022e91d3f2efdc5bed011f3d43dd8c2970 Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Thu, 30 Apr 2026 21:58:15 +1000 Subject: [PATCH 4/7] fix(vpcd): also subscribe DHCPManager before bootstrap reconcile pass Previous fix moved DHCPManager.Subscribe ahead of ReconcileFromKV but missed Pass 1 (the bootstrap-config Reconcile that runs even earlier). On a dev box with a source="dhcp" pool and a bootstrap IGW, that earlier pass still hit "no responders" and burned the 60-retry cold-boot budget before the Manager came up. Move both reconcile passes after Subscribe. --- spinifex/services/vpcd/vpcd.go | 24 ++++++++++++++---------- 1 file changed, 14 insertions(+), 10 deletions(-) diff --git a/spinifex/services/vpcd/vpcd.go b/spinifex/services/vpcd/vpcd.go index 225b021d7..eaa33a3ba 100644 --- a/spinifex/services/vpcd/vpcd.go +++ b/spinifex/services/vpcd/vpcd.go @@ -363,11 +363,6 @@ func launchService(cfg *Config) error { holder, _ := os.Hostname() releaseLeader, isLeader := AcquireReconcileLeader(nc, holder) - // Run reconciliation before subscribing (leader only) - if isLeader { - Reconcile(ctx, topo, cfg.Bootstrap) - } - // Macvlan mode only: align macvlan MAC with the OVN gateway router MAC. // The macvlan only delivers inbound unicast matching its own MAC. With // centralized NAT, OVN uses the router MAC for all external ARP replies. @@ -396,11 +391,14 @@ func launchService(cfg *Config) error { // DHCP manager: services vpc.dhcp.acquire/release requests from the // daemon-side ExternalIPAM handlers and (mulga-siv-38) topology's gw-LRP - // allocator. MUST subscribe before ReconcileFromKV / retrofit so those - // passes don't fan out vpc.dhcp.acquire requests to "no responders" and - // burn the 60-attempt cold-boot retry budget per existing IGW. Started - // unconditionally — pools with source="static" never issue acquire - // requests, so the Manager sits idle until something actually wants DHCP. + // allocator. MUST subscribe before any reconcile pass — bootstrap + // reconcile, ReconcileFromKV, and retrofit all call into reconcileIGW / + // expectedGatewayPortNetwork which fan out vpc.dhcp.acquire on + // source="dhcp" pools. Without a live subscription those requests hit + // "no responders" and burn the 60-attempt cold-boot retry budget per + // existing IGW. Started unconditionally — pools with source="static" + // never issue acquire requests, so the Manager sits idle until + // something actually wants DHCP. js, err := nc.JetStream() if err != nil { slog.Error("Failed to get JetStream context for DHCP manager", "err", err) @@ -428,6 +426,12 @@ func launchService(cfg *Config) error { } }() + // Pass 1: Bootstrap-config reconcile. Creates the bootstrap VPC / IGW + // topology if missing. Leader-only. + if isLeader { + Reconcile(ctx, topo, cfg.Bootstrap) + } + // Pass 2: Reconcile from NATS KV (handles reboots, OVN DB loss, missed events). // Runs after subscribing so new events are not missed during reconciliation. // Leader-gated (mulga-siv-29) to avoid concurrent Create races against OVN NB. From efe054e4e4e04d3cc2f10040d40cff8361685614 Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Thu, 30 Apr 2026 23:01:05 +1000 Subject: [PATCH 5/7] =?UTF-8?q?fix(vpcd):=20RFC=202131=20=C2=A74.1=20DORA?= =?UTF-8?q?=20exponential=20backoff=20(mulga-siv-39)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DHCPManager.Acquire previously did one DISCOVER and gave up after 15s because the outer ctx.WithTimeout matched the per-attempt timeout — nclient4's WithRetry(3) could never fire. Real DHCP servers (consumer routers, ISC dhcpd, dnsmasq) silently drop OFFERs under load; the spec mandates retransmission with exponential backoff, and a single lost DISCOVER was killing legitimate boots. Move retransmission into DHCPManager.acquireWithBackoff with a 4s/8s/16s/32s schedule (±1s jitter, ~60s worst case). NClient4Client becomes single-attempt; manager owns retry policy. Default acquireTimeout lifts from 15s to 90s (sum of schedule + slack). --- spinifex/services/vpcd/dhcp/nclient4.go | 27 +++-- spinifex/services/vpcd/dhcp_manager.go | 106 +++++++++++++++-- spinifex/services/vpcd/dhcp_manager_test.go | 119 ++++++++++++++++++++ spinifex/services/vpcd/vpcd.go | 5 +- 4 files changed, 237 insertions(+), 20 deletions(-) diff --git a/spinifex/services/vpcd/dhcp/nclient4.go b/spinifex/services/vpcd/dhcp/nclient4.go index 70ca30f92..3c1bfa239 100644 --- a/spinifex/services/vpcd/dhcp/nclient4.go +++ b/spinifex/services/vpcd/dhcp/nclient4.go @@ -13,22 +13,27 @@ import ( // github.com/insomniacslk/dhcp/dhcpv4/nclient4. Each Acquire/Renew/Release // opens an AF_PACKET socket on the target bridge for the duration of the // handshake and closes it when done — no long-lived per-lease process. +// +// Retransmission lives one layer up in DHCPManager.acquireWithBackoff +// (RFC 2131 §4.1). This client always does exactly one DISCOVER attempt; +// the per-attempt window is whatever the caller's context allows. type NClient4Client struct { timeout time.Duration - retry int } var _ Client = (*NClient4Client)(nil) -// NewNClient4 creates an NClient4Client with sensible DORA defaults. -func NewNClient4(timeout time.Duration, retry int) *NClient4Client { +// NewNClient4 creates an NClient4Client. timeout is the nclient4-internal +// socket read deadline used as a safety net when the caller supplies no +// context deadline; in production DHCPManager always wraps each call in a +// per-attempt ctx so this is rarely hit. The retry argument is accepted +// for backwards compat and ignored — DHCPManager.acquireWithBackoff owns +// retransmission. +func NewNClient4(timeout time.Duration, _ int) *NClient4Client { if timeout <= 0 { - timeout = 15 * time.Second + timeout = 60 * time.Second } - if retry <= 0 { - retry = 3 - } - return &NClient4Client{timeout: timeout, retry: retry} + return &NClient4Client{timeout: timeout} } func (c *NClient4Client) Acquire(ctx context.Context, req AcquireRequest) (*Lease, error) { @@ -42,7 +47,9 @@ func (c *NClient4Client) Acquire(ctx context.Context, req AcquireRequest) (*Leas client, err := nclient4.New(req.Bridge, nclient4.WithHWAddr(req.HWAddr), nclient4.WithTimeout(c.timeout), - nclient4.WithRetry(c.retry), + // Single attempt per call — DHCPManager.acquireWithBackoff drives + // retransmission per RFC 2131 §4.1. + nclient4.WithRetry(1), ) if err != nil { return nil, fmt.Errorf("open nclient4 on %s: %w", req.Bridge, err) @@ -68,7 +75,7 @@ func (c *NClient4Client) Renew(ctx context.Context, lease *Lease) (*Lease, error client, err := nclient4.New(lease.Bridge, nclient4.WithHWAddr(lease.HWAddr), nclient4.WithTimeout(c.timeout), - nclient4.WithRetry(c.retry), + nclient4.WithRetry(1), ) if err != nil { return nil, fmt.Errorf("open nclient4 on %s for renew: %w", lease.Bridge, err) diff --git a/spinifex/services/vpcd/dhcp_manager.go b/spinifex/services/vpcd/dhcp_manager.go index fe3772f31..3597d5ed1 100644 --- a/spinifex/services/vpcd/dhcp_manager.go +++ b/spinifex/services/vpcd/dhcp_manager.go @@ -32,10 +32,17 @@ type DHCPManager struct { // HWAddr unset. Defaults to generateMAC; tests override. macForClientID func(clientID string) net.HardwareAddr - // acquireTimeout bounds a single DORA (Acquire) / Renew / Release - // handshake. Independent of the RPC reply timeout on the handler side. + // acquireTimeout is the wallclock budget for a full Acquire — sum of + // every retransmit attempt in acquireBackoff plus slack. Renew/Release + // reuse this as their single-attempt deadline. acquireTimeout time.Duration + // acquireBackoff drives the per-DISCOVER timeouts used by + // acquireWithBackoff (RFC 2131 §4.1 retransmission). Each entry is + // the wait-for-OFFER window for one attempt; the next entry doubles. + // Default: 4s, 8s, 16s, 32s. ±1s jitter applied per attempt. + acquireBackoff []time.Duration + // jitterFraction is the ± range applied to renewal sleeps, expressed // as a fraction of T1 (0.1 = ±10%). jitterFraction float64 @@ -69,11 +76,19 @@ func WithDHCPMACFunc(fn func(clientID string) net.HardwareAddr) DHCPManagerOptio return func(m *DHCPManager) { m.macForClientID = fn } } -// WithDHCPAcquireTimeout overrides the per-handshake DORA timeout. +// WithDHCPAcquireTimeout overrides the wallclock budget for a full Acquire +// (sum of every retransmit attempt + slack). Renew/Release reuse it. func WithDHCPAcquireTimeout(d time.Duration) DHCPManagerOption { return func(m *DHCPManager) { m.acquireTimeout = d } } +// WithDHCPAcquireBackoff overrides the per-DISCOVER timeout schedule used +// by acquireWithBackoff. Tests use a tight schedule; production keeps the +// RFC 2131 §4.1 default (4s, 8s, 16s, 32s with ±1s jitter). +func WithDHCPAcquireBackoff(b []time.Duration) DHCPManagerOption { + return func(m *DHCPManager) { m.acquireBackoff = b } +} + // WithDHCPJitterFraction overrides the ±fraction applied to renewal sleeps. func WithDHCPJitterFraction(f float64) DHCPManagerOption { return func(m *DHCPManager) { m.jitterFraction = f } @@ -108,10 +123,18 @@ func NewDHCPManager(nc *nats.Conn, js nats.JetStreamContext, client dhcp.Client, ctx, cancel := context.WithCancel(context.Background()) m := &DHCPManager{ - client: client, - kv: kv, - nc: nc, - acquireTimeout: 15 * time.Second, + client: client, + kv: kv, + nc: nc, + // 90s = sum of default acquireBackoff (4+8+16+32 = 60s) + slack for + // socket open/close on each attempt and the final ACK round-trip. + acquireTimeout: 90 * time.Second, + acquireBackoff: []time.Duration{ + 4 * time.Second, + 8 * time.Second, + 16 * time.Second, + 32 * time.Second, + }, jitterFraction: 0.1, rng: rand.New(rand.NewPCG(uint64(time.Now().UnixNano()), 0xdeadbeef)), //nolint:gosec // non-cryptographic jitter leases: map[string]*managedLease{}, @@ -245,7 +268,7 @@ func (m *DHCPManager) handleAcquire(msg *nats.Msg) { defer m.endAcquire(req.Bridge) start := time.Now() - lease, err := m.client.Acquire(ctx, dhcp.AcquireRequest{ + lease, err := m.acquireWithBackoff(ctx, dhcp.AcquireRequest{ Bridge: req.Bridge, ClientID: req.ClientID, Hostname: req.Hostname, @@ -392,6 +415,73 @@ func (m *DHCPManager) renewLoop(ctx context.Context, tracker *managedLease) { } } +// acquireWithBackoff drives RFC 2131 §4.1 DISCOVER retransmission. Each +// entry in acquireBackoff sets one attempt's per-OFFER wait window with +// ±1s jitter; on timeout the loop tries the next entry. Returns on first +// success, on parent ctx cancellation, or after the schedule is exhausted. +// +// Real DHCP servers (consumer routers, ISC dhcpd, dnsmasq) silently drop +// OFFERs under load — without retransmission a single lost packet kills +// the whole DORA. mulga-siv-39. +func (m *DHCPManager) acquireWithBackoff(parent context.Context, req dhcp.AcquireRequest) (*dhcp.Lease, error) { + schedule := m.acquireBackoff + if len(schedule) == 0 { + // Defensive: caller passed an empty schedule via WithDHCPAcquireBackoff. + // Fall through to a single attempt bounded by parent ctx. + schedule = []time.Duration{0} + } + + var lastErr error + for i, base := range schedule { + if err := parent.Err(); err != nil { + if lastErr != nil { + return nil, fmt.Errorf("dhcp acquire on %s (client=%s): %w (last attempt err: %v)", req.Bridge, req.ClientID, err, lastErr) + } + return nil, fmt.Errorf("dhcp acquire on %s (client=%s): %w", req.Bridge, req.ClientID, err) + } + + attemptTimeout := m.dhcpAttemptTimeout(base) + var ( + ctx context.Context + cancel context.CancelFunc + ) + if attemptTimeout > 0 { + ctx, cancel = context.WithTimeout(parent, attemptTimeout) + } else { + ctx, cancel = context.WithCancel(parent) + } + + lease, err := m.client.Acquire(ctx, req) + cancel() + if err == nil { + return lease, nil + } + lastErr = err + + if i < len(schedule)-1 { + slog.Debug("DHCP Manager: DORA retransmit", + "client_id", req.ClientID, "bridge", req.Bridge, + "attempt", i+1, "next_timeout", m.dhcpAttemptTimeout(schedule[i+1]), + "prev_err", err) + } + } + return nil, lastErr +} + +// dhcpAttemptTimeout applies ±1s jitter to a base DISCOVER timeout. +// Pass 0 to disable timeout (used when schedule is empty fallback). +func (m *DHCPManager) dhcpAttemptTimeout(base time.Duration) time.Duration { + if base <= 0 { + return 0 + } + jitter := time.Duration((m.rng.Float64()*2 - 1) * float64(time.Second)) + out := base + jitter + if out <= 0 { + return base + } + return out +} + // tryRenew calls the backend Renew under a bounded context. func (m *DHCPManager) tryRenew(parent context.Context, lease *dhcp.Lease) (*dhcp.Lease, error) { ctx, cancel := context.WithTimeout(parent, m.acquireTimeout) diff --git a/spinifex/services/vpcd/dhcp_manager_test.go b/spinifex/services/vpcd/dhcp_manager_test.go index e85e8d4e6..3b724da1c 100644 --- a/spinifex/services/vpcd/dhcp_manager_test.go +++ b/spinifex/services/vpcd/dhcp_manager_test.go @@ -412,6 +412,125 @@ func TestDHCPManager_EncodeDecodeLeaseRoundTrip(t *testing.T) { require.Equal(t, in.RawOffer, out.RawOffer) } +// --- acquireWithBackoff (mulga-siv-39, RFC 2131 §4.1) --- + +func newBackoffManager(t *testing.T, schedule []time.Duration, fake *dhcp.Fake) *DHCPManager { + t.Helper() + _, nc, js := testutil.StartTestJetStream(t) + m, err := NewDHCPManager(nc, js, fake, WithDHCPAcquireBackoff(schedule)) + require.NoError(t, err) + t.Cleanup(m.Close) + return m +} + +func TestAcquireWithBackoff_FirstAttemptSucceeds(t *testing.T) { + fake := dhcp.NewFake() + m := newBackoffManager(t, []time.Duration{10 * time.Millisecond, 10 * time.Millisecond}, fake) + + lease, err := m.acquireWithBackoff(context.Background(), dhcp.AcquireRequest{ + Bridge: "br-wan", ClientID: "eni-fast", HWAddr: net.HardwareAddr{0x02, 0, 0, 0, 0, 1}, + }) + require.NoError(t, err) + require.NotNil(t, lease) + require.Equal(t, 1, fake.AcquireCount(), "first-attempt success must not retransmit") +} + +func TestAcquireWithBackoff_RetriesThenSucceeds(t *testing.T) { + fake := dhcp.NewFake() + var calls atomic.Int32 + tmpl := dhcp.DefaultLeaseTemplate() + fake.AcquireHook = func(req dhcp.AcquireRequest) (*dhcp.Lease, error) { + n := calls.Add(1) + if n < 3 { + return nil, errors.New("simulated OFFER timeout") + } + return &dhcp.Lease{ + Bridge: req.Bridge, ClientID: req.ClientID, HWAddr: req.HWAddr, + IP: tmpl.IP, ServerID: tmpl.ServerID, LeaseDuration: tmpl.LeaseDuration, + AcquiredAt: time.Now(), + }, nil + } + m := newBackoffManager(t, []time.Duration{ + 5 * time.Millisecond, 5 * time.Millisecond, 5 * time.Millisecond, 5 * time.Millisecond, + }, fake) + + lease, err := m.acquireWithBackoff(context.Background(), dhcp.AcquireRequest{ + Bridge: "br-wan", ClientID: "eni-retry", HWAddr: net.HardwareAddr{0x02, 0, 0, 0, 0, 2}, + }) + require.NoError(t, err) + require.NotNil(t, lease) + require.Equal(t, int32(3), calls.Load()) +} + +func TestAcquireWithBackoff_AllAttemptsFail(t *testing.T) { + fake := dhcp.NewFake() + fake.AcquireHook = func(dhcp.AcquireRequest) (*dhcp.Lease, error) { + return nil, errors.New("server silent") + } + schedule := []time.Duration{ + 2 * time.Millisecond, 2 * time.Millisecond, 2 * time.Millisecond, + } + m := newBackoffManager(t, schedule, fake) + + lease, err := m.acquireWithBackoff(context.Background(), dhcp.AcquireRequest{ + Bridge: "br-wan", ClientID: "eni-fail", HWAddr: net.HardwareAddr{0x02, 0, 0, 0, 0, 3}, + }) + require.Error(t, err) + require.Nil(t, lease) + require.ErrorContains(t, err, "server silent") + require.Equal(t, len(schedule), fake.AcquireCount(), + "every entry in the schedule must run before giving up") +} + +func TestAcquireWithBackoff_ParentCtxCancel(t *testing.T) { + fake := dhcp.NewFake() + fake.AcquireHook = func(dhcp.AcquireRequest) (*dhcp.Lease, error) { + return nil, errors.New("should not be called") + } + m := newBackoffManager(t, []time.Duration{ + time.Second, time.Second, time.Second, + }, fake) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + lease, err := m.acquireWithBackoff(ctx, dhcp.AcquireRequest{ + Bridge: "br-wan", ClientID: "eni-cancel", HWAddr: net.HardwareAddr{0x02, 0, 0, 0, 0, 4}, + }) + require.Error(t, err) + require.Nil(t, lease) + require.ErrorIs(t, err, context.Canceled) + require.Equal(t, 0, fake.AcquireCount(), "cancelled parent must short-circuit before any attempt") +} + +func TestDHCPAttemptTimeoutJitterBounds(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + m, err := NewDHCPManager(nc, js, dhcp.NewFake()) + require.NoError(t, err) + t.Cleanup(m.Close) + + base := 4 * time.Second + for i := range 200 { + got := m.dhcpAttemptTimeout(base) + if got < base-time.Second || got > base+time.Second { + t.Fatalf("iter %d: dhcpAttemptTimeout(%v) = %v, outside ±1s", i, base, got) + } + } + require.Zero(t, m.dhcpAttemptTimeout(0)) +} + +func TestNewDHCPManager_DefaultBackoffSchedule(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + m, err := NewDHCPManager(nc, js, dhcp.NewFake()) + require.NoError(t, err) + t.Cleanup(m.Close) + + require.Equal(t, []time.Duration{ + 4 * time.Second, 8 * time.Second, 16 * time.Second, 32 * time.Second, + }, m.acquireBackoff) + require.Equal(t, 90*time.Second, m.acquireTimeout) +} + func TestResolveHWAddr(t *testing.T) { t.Run("InvalidRawMACReturnsParseError", func(t *testing.T) { mac, err := resolveHWAddr("not-a-mac", "client-1", nil) diff --git a/spinifex/services/vpcd/vpcd.go b/spinifex/services/vpcd/vpcd.go index eaa33a3ba..fb75a7c6b 100644 --- a/spinifex/services/vpcd/vpcd.go +++ b/spinifex/services/vpcd/vpcd.go @@ -10,7 +10,6 @@ import ( "path/filepath" "strings" "syscall" - "time" "github.com/mulgadc/spinifex/spinifex/admin" "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" @@ -404,7 +403,9 @@ func launchService(cfg *Config) error { slog.Error("Failed to get JetStream context for DHCP manager", "err", err) return err } - dhcpManager, err := NewDHCPManager(nc, js, dhcp.NewNClient4(15*time.Second, 3)) + // timeout/retry args are legacy no-ops (mulga-siv-39): DHCPManager + // owns DORA retransmission via acquireWithBackoff. + dhcpManager, err := NewDHCPManager(nc, js, dhcp.NewNClient4(0, 0)) if err != nil { slog.Error("Failed to create DHCP manager", "err", err) return err From c234cee60dc952c98a61d55cbf94ef5d10ef6f17 Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Fri, 1 May 2026 08:52:41 +1000 Subject: [PATCH 6/7] fix(vpcd): retry NewDHCPManager on JetStream-not-ready (mulga-siv-40) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit NewDHCPManager creates the spinifex-dhcp-leases KV bucket via utils.GetOrCreateKVBucket → js.CreateKeyValue with no retry. After the mulga-siv-38 launchService reorder, this runs before any other JetStream user, so on multi-node cold-boot vpcd hits "JetStream system temporarily unavailable" and exits before cluster quorum forms. Pseudo-multinode CI calls start_node_services sequentially, so node2's vpcd failure aborts the script before node3 ever launches and quorum then never forms. Wrap NewDHCPManager in newDHCPManagerWithRetry (500ms→10s, 5 min cap, ctx-cancellable) mirroring daemon.initJetStream so vpcd waits as patiently as the daemon for the cluster to come up. --- spinifex/services/vpcd/vpcd.go | 53 +++++++++++++++++++++++++++++++++- 1 file changed, 52 insertions(+), 1 deletion(-) diff --git a/spinifex/services/vpcd/vpcd.go b/spinifex/services/vpcd/vpcd.go index fb75a7c6b..41cd0abbc 100644 --- a/spinifex/services/vpcd/vpcd.go +++ b/spinifex/services/vpcd/vpcd.go @@ -10,10 +10,12 @@ import ( "path/filepath" "strings" "syscall" + "time" "github.com/mulgadc/spinifex/spinifex/admin" "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" ) // sudoCommand wraps exec.Command with sudo when running as non-root. @@ -405,7 +407,14 @@ func launchService(cfg *Config) error { } // timeout/retry args are legacy no-ops (mulga-siv-39): DHCPManager // owns DORA retransmission via acquireWithBackoff. - dhcpManager, err := NewDHCPManager(nc, js, dhcp.NewNClient4(0, 0)) + // + // Retry on JetStream-not-ready: NewDHCPManager creates the DHCP-leases + // KV bucket, which fails immediately if the JetStream cluster has not + // reached quorum. On multi-node bring-up vpcd's launchService runs + // before all peers are reachable, so we mirror the daemon's + // initJetStream backoff (500ms→10s, 5 min cap) instead of crashing the + // service and aborting the rest of the node bring-up. + dhcpManager, err := newDHCPManagerWithRetry(ctx, nc, js) if err != nil { slog.Error("Failed to create DHCP manager", "err", err) return err @@ -467,6 +476,48 @@ func launchService(cfg *Config) error { return nil } +// newDHCPManagerWithRetry constructs a DHCPManager, retrying on transient +// JetStream-not-ready errors with exponential backoff (500ms→10s, capped +// at 5 min). NewDHCPManager creates the spinifex-dhcp-leases KV bucket; +// on multi-node bring-up the JetStream cluster may not yet have quorum +// when vpcd starts, and a single-shot CreateKeyValue would otherwise crash +// the service before the rest of the node finishes coming up. +// +// Mirrors daemon.initJetStream's backoff so vpcd waits as patiently as the +// daemon does for the cluster to form. +func newDHCPManagerWithRetry(ctx context.Context, nc *nats.Conn, js nats.JetStreamContext) (*DHCPManager, error) { + const maxWait = 5 * time.Minute + retryDelay := 500 * time.Millisecond + start := time.Now() + attempt := 0 + + for { + attempt++ + mgr, err := NewDHCPManager(nc, js, dhcp.NewNClient4(0, 0)) + if err == nil { + if attempt > 1 { + slog.Info("vpcd: DHCP manager ready", "attempts", attempt, "elapsed", time.Since(start).Round(time.Second)) + } + return mgr, nil + } + + elapsed := time.Since(start) + if elapsed >= maxWait { + return nil, fmt.Errorf("DHCP manager init timed out after %s (%d attempts): %w", elapsed.Round(time.Second), attempt, err) + } + + slog.Warn("vpcd: DHCP manager not ready (waiting for JetStream cluster quorum)", + "err", err, "attempt", attempt, "elapsed", elapsed.Round(time.Second), "retryIn", retryDelay) + + select { + case <-ctx.Done(): + return nil, fmt.Errorf("DHCP manager init cancelled: %w", ctx.Err()) + case <-time.After(retryDelay): + } + retryDelay = min(retryDelay*2, 10*time.Second) + } +} + // resolveBridgeConfig picks the bridge mode and DHCP-bind-bridge to use, // auto-detecting mode when unset. Empty mode stays empty — verifyBridgeMode // rejects it with a list of supported values (D12). Empty bind bridge From 11b3f37d066041a7635ee56d163c1d92f4046173 Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Fri, 1 May 2026 08:52:49 +1000 Subject: [PATCH 7/7] fix(daemon): idempotent DetachVolume QMP path (mulga-siv-41) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DetachVolume's QMP sequence was not retry-safe. First call: device_del succeeds, blockdev-del fails with "Node nbd-vol-X is in use" because NBD client teardown / in-flight I/O still hold the block node. Handler bails ServerInternal "leaving volume state intact" but the guest device is actually gone. AWS-CLI retries: device_del returns DeviceNotFound, handler bails again. After 2 retries aws cli exits 254 — Single-Node Phase 8 Step 3 reproducibly fails this way. Phase 1: treat DeviceNotFound on device_del as already-detached, log and continue. Detach is now idempotent across retries. Phase 2: tryBlockdevDel polls blockdev-del up to 20×detachDelay on "in use" errors so the I/O drain can complete instead of failing the whole detach. Other QMP errors fall through immediately as before. --- spinifex/daemon/daemon_handlers_volume.go | 100 ++++++++++++++++++---- 1 file changed, 85 insertions(+), 15 deletions(-) diff --git a/spinifex/daemon/daemon_handlers_volume.go b/spinifex/daemon/daemon_handlers_volume.go index f42fa81bb..371ed070c 100644 --- a/spinifex/daemon/daemon_handlers_volume.go +++ b/spinifex/daemon/daemon_handlers_volume.go @@ -5,6 +5,7 @@ import ( "fmt" "log/slog" "strconv" + "strings" "time" "github.com/aws/aws-sdk-go/service/ec2" @@ -357,32 +358,42 @@ func (d *Daemon) handleDetachVolume(msg *nats.Msg, command types.EC2InstanceComm nodeName := fmt.Sprintf("nbd-%s", volumeID) iothreadID := fmt.Sprintf("ioth-%s", volumeID) - // Phase 1: QMP device_del (remove guest device) + // Phase 1: QMP device_del (remove guest device). + // Idempotent: a prior detach may have already removed the device but + // failed at blockdev-del, leaving the volume's external state intact. + // Treat DeviceNotFound as success so the retry can drive blockdev-del + // to completion without bailing on the now-absent guest device. _, err := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{ Execute: "device_del", Arguments: map[string]any{"id": deviceID}, }, instance.ID) - if err != nil { - if !force { - slog.Error("DetachVolume: QMP device_del failed", "volumeId", volumeID, "err", err) - respondWithError(msg, awserrors.ErrorServerInternal) - return - } + switch { + case err == nil: + case isQMPDeviceNotFound(err): + slog.Info("DetachVolume: guest device already removed (resuming detach)", "volumeId", volumeID, "err", err) + case force: slog.Warn("DetachVolume: QMP device_del failed (force=true, continuing)", "volumeId", volumeID, "err", err) + default: + slog.Error("DetachVolume: QMP device_del failed", "volumeId", volumeID, "err", err) + respondWithError(msg, awserrors.ErrorServerInternal) + return } // Brief pause for guest to acknowledge PCI removal time.Sleep(d.detachDelay) - // Phase 2: QMP blockdev-del (remove block node) - _, blockdevErr := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{ - Execute: "blockdev-del", - Arguments: map[string]any{"node-name": nodeName}, - }, instance.ID) + // Phase 2: QMP blockdev-del (remove block node). + // Poll-retry on "node is in use": after device_del, NBD client teardown + // and any in-flight I/O can hold the block node briefly. Bailing here + // without retry leaves the guest device gone but the block node intact, + // and the next AWS-CLI retry hits DeviceNotFound on device_del with no + // path forward (mulga-siv-41). + blockdevErr := d.tryBlockdevDel(instance, nodeName) if blockdevErr != nil { - // Block node still referenced by QEMU; do not clean up state or unmount — - // tearing down the NBD server would crash the VM, and removing metadata - // would allow the volume to be double-attached. + // Block node still referenced after retry budget exhausted; do not + // clean up state or unmount — tearing down the NBD server would + // crash the VM, and removing metadata would allow the volume to be + // double-attached. slog.Error("DetachVolume: QMP blockdev-del failed, leaving volume state intact", "volumeId", volumeID, "err", blockdevErr) respondWithError(msg, awserrors.ErrorServerInternal) return @@ -504,3 +515,62 @@ func (d *Daemon) handleEC2ModifyVolume(msg *nats.Msg) { func (d *Daemon) handleEC2DeleteVolume(msg *nats.Msg) { handleNATSRequest(msg, d.volumeService.DeleteVolume) } + +// tryBlockdevDel issues blockdev-del with bounded retry on "is in use" +// errors. After device_del, NBD client teardown and any in-flight guest +// I/O can briefly hold the block node; QEMU surfaces this as a +// GenericError carrying "Node is in use". Polling at d.detachDelay +// gives the I/O drain enough time without a hardcoded long sleep. +func (d *Daemon) tryBlockdevDel(instance *vm.VM, nodeName string) error { + const maxAttempts = 20 // 20 × detachDelay (default 1s) = 20s budget + var lastErr error + for attempt := 1; attempt <= maxAttempts; attempt++ { + _, err := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{ + Execute: "blockdev-del", + Arguments: map[string]any{"node-name": nodeName}, + }, instance.ID) + if err == nil { + if attempt > 1 { + slog.Info("DetachVolume: blockdev-del succeeded after retry", + "nodeName", nodeName, "attempts", attempt) + } + return nil + } + lastErr = err + if !isQMPNodeInUse(err) { + return err + } + if attempt == maxAttempts { + break + } + slog.Debug("DetachVolume: blockdev-del busy, retrying", + "nodeName", nodeName, "attempt", attempt, "err", err) + time.Sleep(d.detachDelay) + } + return lastErr +} + +// isQMPDeviceNotFound returns true when err is a QMP DeviceNotFound class +// error from device_del. Used to make device_del idempotent across detach +// retries — the second AWS-CLI call must not bail when QEMU reports the +// device is already gone (mulga-siv-41). +func isQMPDeviceNotFound(err error) bool { + if err == nil { + return false + } + s := err.Error() + return strings.Contains(s, "DeviceNotFound") +} + +// isQMPNodeInUse returns true when err is a QMP GenericError reporting a +// block node still in use. blockdev-del fails this way while NBD client +// teardown and queued I/O drain after device_del; the caller polls until +// the node is released or the budget expires. Match "in use" rather than +// "is in use" — QEMU phrases this as both "X is in use" and "X is still +// in use" depending on the path. +func isQMPNodeInUse(err error) bool { + if err == nil { + return false + } + return strings.Contains(err.Error(), "in use") +}