From 69404b38ce6a352c2daf2fbf384cae0bc096c44a Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Fri, 1 May 2026 09:48:52 +1000 Subject: [PATCH] fix(vpcd): public-IP boot race + DHCP DORA hardening (siv-33/34/36/37/38/39) (#177) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(installer): emit kernel cmdline to ttyS0 alongside tty0 (mulga-siv-33) GRUB_CMDLINE_LINUX in /etc/default/grub only listed console=tty0, so the installed kernel printed nothing on serial after the GRUB menu. Headless installs (CI, racks with serial-only IPMI, remote-dev SSH-to-QEMU) saw a black screen forever and had no path to debug boot failures. Append console=ttyS0,115200 — the last `console=` on the cmdline becomes the system console, so ttyS0 wins on serial-only hardware while tty0 still receives output on local displays. Plan: docs/development/bugs/iso-public-ip-boot-race.md * fix(ipam): widen vpc.dhcp.acquire retry budget to 60 s (mulga-siv-34) ObtainDHCPLease retried only 5 attempts × 1 s before returning a hard error. On a cold ISO boot, NATS comes up well before vpcd's reconcile leader-election + dhcp.Manager subscribe (typical 5–30 s, pathological ~60 s during full reconcile), so daemon's NewExternalIPAM races vpcd's dhcp.acquire subscription and fails permanently — d.externalIPAM stays nil for the daemon's lifetime, daemon_handlers_instance.go silently skips all auto-public-IP allocation, and instances launch with no public IP, no dnat_and_snat, and no NAT event ever published. Bump maxAttempts to 60 with 1 s spacing; expose both as package vars so tests can shrink the budget for fast timeout-path coverage. Pure widening of the existing retry loop — no behaviour change on the success path, only the worst-case wallclock when vpcd never comes up. The dominant cause of slow vpcd starts (mulga-siv-33: systemd-networkd not enabled in the ISO → veth pair vanishes on reboot → vpcd crashloops) is fixed alongside this. A 60 s budget fits comfortably above all observed healthy-startup times. Plan: docs/development/bugs/iso-public-ip-boot-race.md * fix(setup-ovn): declare Linux WAN bridge as NetDev so networkd-managed veth resolves Bridge= on reboot (mulga-siv-33) The veth-mode reboot persistence drop-ins added in mulga-998.b were incomplete: setup-ovn.sh wrote `15-spinifex-veth-wan.network` with `Bridge=br-wan`, but never declared br-wan itself to networkd. br-wan is created by `networking.service` (ifupdown via `/etc/network/interfaces`, generated by `install.go writeNetworkConfig`), not networkd. networkd's `Bridge=` directive requires the target to be a known NetDev — when it isn't, networkd silently logs `br-wan NetDev could not be found, ignoring assignment` and the veth ends up with no master. vpcd's bridge-mode sanity check then fails on every reboot. Fix: also write `14-spinifex-br-wan.netdev` declaring br-wan as `Kind=bridge`. networkd matches the existing kernel bridge by name+kind (the `Failed to create netdev: File exists` log is harmless), and `Bridge=br-wan` resolves on next boot. ifupdown still owns DHCP and the IP — networkd just needs the NetDev declaration to enslave the veth. The write is gated to avoid clobbering operator-managed networkd config: if any other `.netdev` under `/etc/systemd/network`, `/run/systemd/network`, or `/usr/lib/systemd/network` already declares the WAN bridge by name, the write is skipped with a log line pointing at the operator file. Cleanup hook also removes the new file when switching away from veth mode. Verified end-to-end via QEMU repro: clean install + first boot + reboot, veth-wan-br master persists across reboots, vpcd starts cleanly. * fix(vpcd/dhcp): enable AF_PACKET promisc on bridge for DORA (mulga-siv-33) Real-world DHCP servers commonly ignore the BOOTP BROADCAST flag and reply with a unicast OFFER addressed to the synthetic per-tenant chaddr (e.g. 02:f3:...). A Linux bridge floods unknown unicast to slave ports but does NOT pass it up to the bridge netdev unless IFF_PROMISC is set, so vpcd's AF_PACKET socket on br-wan misses the OFFER and DORA times out at 15s ("unable to receive an offer: context deadline exceeded"). Dev/CI didn't hit this: - CI uses spx_wan_mode=static — no DORA via vpcd. - QEMU dev DHCP servers (SLIRP/dnsmasq) honor the BROADCAST flag, so their replies are L2 broadcast which the bridge always passes up. Replace the three nclient4.New(iface, ...) call sites in Acquire/Renew/ Release with nclient4.NewWithConn(...) wrapping a custom AF_PACKET DGRAM conn that calls SetPromiscuous(true). Promisc is per-socket (per-DORA, ~15s) and refcounted by the kernel — auto-clears on close. * revert(vpcd/dhcp): drop AF_PACKET promisc wrapper, restore nclient4.New ecfbd4f swapped nclient4.New for openPromiscBridgeConn + NewWithConn to catch unicast OFFER replies that bypass the bridge netdev. On dev hosts where upstream honors the BOOTP BROADCAST flag, the wrapper regressed DORA — public-IP allocation failed where main works. Reverting until the promisc path is validated against both broadcast- and unicast-OFFER servers. * fix(setup-ovn): mark br-int/br-ext Unmanaged for systemd-networkd (mulga-siv-37) Fresh ISO install on Trixie+systemd-networkd left the OVS internal netdevs admin-down: networkd has no matching .network so it tags them as unmanaged but never brings them up. OVS dataplane works but ovn-controller flow programming and any tooling that probes link state misbehaves. Drop /etc/systemd/network/05-spinifex-ovs-internal.network with Unmanaged=yes so OVS itself owns the link state — `ovs-vsctl add-br` already runs the netdev up at create time. * fix(vpcd): allocate gateway LRP IP from pool in centralized NAT (mulga-siv-36) In veth/macvlan modes the gateway LRP carries all egress, so it must hold a WAN-subnet IP — otherwise the upstream router silently drops ARP for the WAN nexthop (RFC 826) and the default route never resolves. The hardcoded 169.254.0.1/30 was correct for distributed NAT but regressed centralized NAT (introduced under mulga-siv-26). Add gw_lrp_range_start/end to ExternalPool. In centralized mode at IGW attach the gateway LRP gets a fresh IP from that range, persisted as external_ids:spinifex:gateway_ip on the LRP for idempotent reconcile and retrofit. Direct mode keeps link-local. Per-VM EIP allocator skips the gw range; ValidatePoolConfig rejects overlap with the EIP range. Adds TestTopologyHandler_IGWAttach_CentralizedNATAllocatesGwLrpIP and extends RetrofitAllGatewayPortNetworks to migrate stale link-local LRPs to allocated pool IPs at startup. * fix(vpcd): drop link-local fallback, auto-derive gw_lrp_range (mulga-siv-36) 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. * feat(vpcd): DHCP-acquire gateway LRP IPs (mulga-siv-38) 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. * fix(vpcd): subscribe DHCPManager before reconcile to fix boot race (mulga-siv-38) 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. * 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. * fix(vpcd): RFC 2131 §4.1 DORA exponential backoff (mulga-siv-39) 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). * fix(vpcd): retry NewDHCPManager on JetStream-not-ready (mulga-siv-40) 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. * fix(daemon): idempotent DetachVolume QMP path (mulga-siv-41) 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. --- cmd/installer/install/install.go | 11 +- cmd/spinifex/cmd/service.go | 22 +- go.mod | 4 +- scripts/setup-ovn.sh | 51 +- spinifex/config/config.go | 8 + spinifex/daemon/daemon.go | 22 +- spinifex/daemon/daemon_handlers_volume.go | 100 ++- spinifex/handlers/ec2/vpc/external_ipam.go | 119 +++- .../handlers/ec2/vpc/external_ipam_dhcp.go | 150 ----- .../handlers/ec2/vpc/external_ipam_test.go | 170 +++--- spinifex/services/vpcd/dhcp/nats_client.go | 150 +++++ .../services/vpcd/dhcp/nats_client_test.go | 104 ++++ 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/topology.go | 416 +++++++++++-- spinifex/services/vpcd/topology_test.go | 570 +++++++++++++++++- spinifex/services/vpcd/vpcd.go | 120 +++- 18 files changed, 1866 insertions(+), 403 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/installer/install/install.go b/cmd/installer/install/install.go index 53268f7f5..5220e6cf3 100644 --- a/cmd/installer/install/install.go +++ b/cmd/installer/install/install.go @@ -465,15 +465,16 @@ func installBootloader(disk string) error { copySplashImage(mountRoot) copyGrubFont(mountRoot) - // Kernel cmdline and basic defaults only — graphics/serial handled by 05_spinifex below. - // Confirm requirement - // console=ttyS0,115200n8 - + // Both consoles listed: tty0 stays primary for local install, ttyS0 mirrors + // kernel output to serial so headless installs (CI, racks with serial-only + // IPMI, remote-dev SSH-to-QEMU) see boot messages. The last `console=` on + // the cmdline becomes the system console, so ttyS0 wins on serial-only + // hardware while tty0 still receives output for local display. grubDefault := `GRUB_DEFAULT=0 GRUB_TIMEOUT=5 GRUB_DISTRIBUTOR=Spinifex GRUB_CMDLINE_LINUX_DEFAULT="" -GRUB_CMDLINE_LINUX="console=tty0 systemd.show_status=1" +GRUB_CMDLINE_LINUX="console=tty0 console=ttyS0,115200 systemd.show_status=1" ` if err := os.WriteFile(filepath.Join(mountRoot, "etc/default/grub"), []byte(grubDefault), 0o644); err != nil { diff --git a/cmd/spinifex/cmd/service.go b/cmd/spinifex/cmd/service.go index 462012e6e..6cdc7cc3f 100644 --- a/cmd/spinifex/cmd/service.go +++ b/cmd/spinifex/cmd/service.go @@ -734,15 +734,19 @@ var vpcdStartCmd = &cobra.Command{ var extPools []vpcd.ExternalPoolConfig for _, p := range clusterConfig.Network.ExternalPools { extPools = append(extPools, vpcd.ExternalPoolConfig{ - Name: p.Name, - RangeStart: p.RangeStart, - RangeEnd: p.RangeEnd, - Gateway: p.Gateway, - GatewayIP: p.GatewayIP, - PrefixLen: p.PrefixLen, - DNSServers: p.DNSServers, - Region: p.Region, - AZ: p.AZ, + Name: p.Name, + Source: p.Source, + RangeStart: p.RangeStart, + RangeEnd: p.RangeEnd, + Gateway: p.Gateway, + GatewayIP: p.GatewayIP, + PrefixLen: p.PrefixLen, + DNSServers: p.DNSServers, + Region: p.Region, + AZ: p.AZ, + DhcpBindBridge: nodeConfig.VPCD.DhcpBindBridge, + GwLrpRangeStart: p.GwLrpRangeStart, + GwLrpRangeEnd: p.GwLrpRangeEnd, }) } diff --git a/go.mod b/go.mod index 602326933..abba016b2 100644 --- a/go.mod +++ b/go.mod @@ -12,6 +12,7 @@ require ( github.com/insomniacslk/dhcp v0.0.0-20260407060928-11b94ed970f2 github.com/kdomanski/iso9660 v0.4.0 github.com/klauspost/cpuid/v2 v2.3.0 + github.com/mdlayher/packet v1.1.2 github.com/mulgadc/predastore v1.1.6 github.com/mulgadc/viperblock v1.2.1 github.com/nats-io/nats-server/v2 v2.12.7 @@ -23,6 +24,7 @@ require ( github.com/spf13/viper v1.21.0 github.com/stretchr/testify v1.11.1 golang.org/x/crypto v0.50.0 + golang.org/x/sys v0.43.0 gopkg.in/ini.v1 v1.67.1 ) @@ -86,7 +88,6 @@ require ( github.com/mattn/go-isatty v0.0.20 // indirect github.com/mattn/go-localereader v0.0.1 // indirect github.com/mattn/go-runewidth v0.0.21 // indirect - github.com/mdlayher/packet v1.1.2 // indirect github.com/mdlayher/socket v0.4.1 // indirect github.com/minio/crc64nvme v1.1.1 // indirect github.com/minio/highwayhash v1.0.4 // indirect @@ -123,7 +124,6 @@ require ( golang.org/x/mod v0.34.0 // indirect golang.org/x/net v0.53.0 // indirect golang.org/x/sync v0.20.0 // indirect - golang.org/x/sys v0.43.0 // indirect golang.org/x/telemetry v0.0.0-20260311193753-579e4da9a98c // indirect golang.org/x/term v0.42.0 // indirect golang.org/x/text v0.36.0 // indirect diff --git a/scripts/setup-ovn.sh b/scripts/setup-ovn.sh index 353197131..15d611e66 100755 --- a/scripts/setup-ovn.sh +++ b/scripts/setup-ovn.sh @@ -290,6 +290,26 @@ sudo ovs-vsctl set Bridge br-int other-config:disable-in-band=true sudo ip link set br-int up echo " br-int: created, fail-mode=secure, up" +# Mark OVS internal netdevs Unmanaged for systemd-networkd. Without this, +# Trixie's networkd takes ownership of any unconfigured iface and may bring +# br-int/br-ext admin-down after setup-ovn.sh's `ip link set up` (no Match +# rule => default management => no carrier => link down). OVS dataplane +# still forwards, but ovn-controller flow programming and any tooling that +# probes link state misbehave. Unmanaged=yes keeps OVS in sole control. +# (mulga-siv-37) +OVS_INTERNAL_NET=/etc/systemd/network/05-spinifex-ovs-internal.network +if [ ! -f "$OVS_INTERNAL_NET" ]; then + sudo tee "$OVS_INTERNAL_NET" >/dev/null <<'NETWORK' +[Match] +Name=br-int br-ext + +[Link] +Unmanaged=yes +NETWORK + sudo networkctl reload 2>/dev/null || true + echo " wrote $OVS_INTERNAL_NET (Unmanaged=yes for br-int br-ext)" +fi + # --- Step 3b: Configure WAN bridge for public subnet uplink --- if [ -n "$WAN_BRIDGE" ]; then echo "" @@ -301,7 +321,8 @@ if [ -n "$WAN_BRIDGE" ]; then # veth pair re-materialises on reboot and fights the current mode's # bridge plumbing (Fix 1, mulga-998.b, per D17). if [ "$WAN_BRIDGE_MODE" != "veth" ]; then - sudo rm -f /etc/systemd/network/15-spinifex-veth-wan.netdev \ + sudo rm -f /etc/systemd/network/14-spinifex-br-wan.netdev \ + /etc/systemd/network/15-spinifex-veth-wan.netdev \ /etc/systemd/network/15-spinifex-veth-wan.network \ /etc/systemd/network/16-spinifex-veth-wan-ovs.network sudo networkctl reload 2>/dev/null || true @@ -366,9 +387,37 @@ if [ -n "$WAN_BRIDGE" ]; then # are kernel-only and vanish on reboot; without persistence vpcd # starts with the OVS port pointing at a nonexistent peer and # silently falls back to direct mode (Fix 1, mulga-998.b). + # + # Declare the Linux bridge as a NetDev too (even though it's + # already created by ifupdown) so networkd recognises it and the + # `Bridge=$LINUX_BRIDGE` directive below resolves. Without this, + # networkd logs `br-wan NetDev could not be found, ignoring + # assignment` and veth-wan-br ends up orphaned with no master, + # which crashes vpcd's veth bridge mode sanity check on reboot. + # `Failed to create netdev: File exists` is harmless — networkd + # then matches the existing kernel bridge by name+kind. + # + # Gate: skip the write if any operator-managed .netdev (cloud-init, + # netplan, manual op) already declares this bridge — don't clobber + # custom networkd config. networkd searches /etc, /run, /usr/lib in + # that priority order; check all three. Our own file is excluded so + # idempotent re-runs still rewrite when needed. + BR_WAN_NETDEV="/etc/systemd/network/14-spinifex-br-wan.netdev" VETH_NETDEV="/etc/systemd/network/15-spinifex-veth-wan.netdev" VETH_NETWORK="/etc/systemd/network/15-spinifex-veth-wan.network" VETH_OVS_NETWORK="/etc/systemd/network/16-spinifex-veth-wan-ovs.network" + EXISTING_BR_NETDEV=$(grep -rls --include="*.netdev" -E "^\s*Name=$LINUX_BRIDGE\s*$" \ + /etc/systemd/network /run/systemd/network /usr/lib/systemd/network 2>/dev/null \ + | grep -v "^$BR_WAN_NETDEV$" || true) + if [ -n "$EXISTING_BR_NETDEV" ]; then + echo " skipping $BR_WAN_NETDEV — operator-managed NetDev already declares $LINUX_BRIDGE: $EXISTING_BR_NETDEV" + else + sudo tee "$BR_WAN_NETDEV" >/dev/null </dev/null < 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") +} diff --git a/spinifex/handlers/ec2/vpc/external_ipam.go b/spinifex/handlers/ec2/vpc/external_ipam.go index e96a9d721..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 @@ -38,16 +39,21 @@ type ExternalIPAllocation struct { // ExternalIPAMRecord tracks allocated external IPs for a single pool. type ExternalIPAMRecord struct { - PoolName string `json:"pool_name"` - Source string `json:"source"` // "static" (default) or "dhcp" - RangeStart string `json:"range_start"` - RangeEnd string `json:"range_end"` - Gateway string `json:"gateway"` - GatewayIP string `json:"gateway_ip"` - PrefixLen int `json:"prefix_len"` - Region string `json:"region,omitempty"` - AZ string `json:"az,omitempty"` - Allocated map[string]ExternalIPAllocation `json:"allocated"` + PoolName string `json:"pool_name"` + Source string `json:"source"` // "static" (default) or "dhcp" + RangeStart string `json:"range_start"` + RangeEnd string `json:"range_end"` + Gateway string `json:"gateway"` + GatewayIP string `json:"gateway_ip"` + PrefixLen int `json:"prefix_len"` + Region string `json:"region,omitempty"` + AZ string `json:"az,omitempty"` + // GwLrpRangeStart/End mirrors ExternalPoolConfig — IPAM skips this + // sub-range so it doesn't collide with vpcd's gateway LRP IPs in + // centralized NAT (mulga-siv-36). + GwLrpRangeStart string `json:"gw_lrp_range_start,omitempty"` + GwLrpRangeEnd string `json:"gw_lrp_range_end,omitempty"` + Allocated map[string]ExternalIPAllocation `json:"allocated"` } // ExternalPoolConfig is the admin-defined pool from spinifex.toml. @@ -62,6 +68,11 @@ type ExternalPoolConfig struct { Region string AZ string DhcpBindBridge string // Bridge where the DHCP AF_PACKET socket binds (e.g. "br-wan"). Linux bridge in veth mode; OVS bridge in direct mode. Never "br-ext". + // GwLrpRangeStart/End reserves a sub-range of the LAN for OVN gateway + // LRP IPs in centralized NAT mode (mulga-siv-36). IPAM must skip these + // addresses or the per-VM EIP allocator and vpcd will fight over them. + GwLrpRangeStart string + GwLrpRangeEnd string } // IsDHCP returns true if this pool obtains IPs from router DHCP. @@ -122,15 +133,20 @@ func (m *ExternalIPAM) initPool(pool ExternalPoolConfig) error { } if err == nil { - if chk.RangeStart != pool.RangeStart || chk.RangeEnd != pool.RangeEnd || chk.Source != pool.Source { + if chk.RangeStart != pool.RangeStart || chk.RangeEnd != pool.RangeEnd || chk.Source != pool.Source || + chk.GwLrpRangeStart != pool.GwLrpRangeStart || chk.GwLrpRangeEnd != pool.GwLrpRangeEnd { slog.Info("external IPAM pool config drift, reconciling KV", "pool", pool.Name, "old_range", chk.RangeStart+"-"+chk.RangeEnd, "new_range", pool.RangeStart+"-"+pool.RangeEnd, - "old_source", chk.Source, "new_source", pool.Source) + "old_source", chk.Source, "new_source", pool.Source, + "old_gw_lrp_range", chk.GwLrpRangeStart+"-"+chk.GwLrpRangeEnd, + "new_gw_lrp_range", pool.GwLrpRangeStart+"-"+pool.GwLrpRangeEnd) chk.RangeStart = pool.RangeStart chk.RangeEnd = pool.RangeEnd chk.Source = pool.Source + chk.GwLrpRangeStart = pool.GwLrpRangeStart + chk.GwLrpRangeEnd = pool.GwLrpRangeEnd data, err := json.Marshal(chk) if err != nil { @@ -158,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", @@ -182,14 +198,16 @@ func (m *ExternalIPAM) initPool(pool ExternalPoolConfig) error { } record := &ExternalIPAMRecord{ - PoolName: pool.Name, - RangeStart: pool.RangeStart, - RangeEnd: pool.RangeEnd, - Gateway: pool.Gateway, - GatewayIP: gwIP, - PrefixLen: pool.PrefixLen, - Region: pool.Region, - AZ: pool.AZ, + PoolName: pool.Name, + RangeStart: pool.RangeStart, + RangeEnd: pool.RangeEnd, + Gateway: pool.Gateway, + GatewayIP: gwIP, + PrefixLen: pool.PrefixLen, + Region: pool.Region, + AZ: pool.AZ, + GwLrpRangeStart: pool.GwLrpRangeStart, + GwLrpRangeEnd: pool.GwLrpRangeEnd, Allocated: map[string]ExternalIPAllocation{ gwIP: gatewayAlloc, }, @@ -250,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) } @@ -264,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) } @@ -296,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) } @@ -317,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) } @@ -372,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) } } @@ -453,7 +471,9 @@ func (m *ExternalIPAM) getRecord(poolName string) (*ExternalIPAMRecord, uint64, return &record, entry.Revision(), nil } -// nextAvailableExternalIP finds the next unallocated IP in the pool's range. +// nextAvailableExternalIP finds the next unallocated IP in the pool's +// range. Addresses inside [GwLrpRangeStart, GwLrpRangeEnd] are skipped — +// vpcd reserves them for OVN gateway LRPs (mulga-siv-36). func nextAvailableExternalIP(record *ExternalIPAMRecord) (string, error) { startIP := net.ParseIP(record.RangeStart).To4() endIP := net.ParseIP(record.RangeEnd).To4() @@ -461,10 +481,23 @@ func nextAvailableExternalIP(record *ExternalIPAMRecord) (string, error) { return "", fmt.Errorf("invalid IP range: %s - %s", record.RangeStart, record.RangeEnd) } + var gwLrpStart, gwLrpEnd int64 = -1, -1 + if record.GwLrpRangeStart != "" && record.GwLrpRangeEnd != "" { + s := net.ParseIP(record.GwLrpRangeStart).To4() + e := net.ParseIP(record.GwLrpRangeEnd).To4() + if s != nil && e != nil { + gwLrpStart = ipToInt(s).Int64() + gwLrpEnd = ipToInt(e).Int64() + } + } + startInt := ipToInt(startIP) endInt := ipToInt(endIP) for i := startInt.Int64(); i <= endInt.Int64(); i++ { + if gwLrpStart >= 0 && i >= gwLrpStart && i <= gwLrpEnd { + continue + } candidate := intToIP(intFromInt64(i)).String() if _, taken := record.Allocated[candidate]; !taken { return candidate, nil @@ -501,6 +534,32 @@ func ValidatePoolConfig(pool ExternalPoolConfig) error { if ipToInt(startIP.To4()).Cmp(ipToInt(endIP.To4())) > 0 { return fmt.Errorf("range_start %s is greater than range_end %s", pool.RangeStart, pool.RangeEnd) } + // gw_lrp_range must be valid IPs and must NOT overlap range_start/end + // — otherwise vpcd's gateway LRP allocator and per-VM EIP allocator + // would fight over the same address (mulga-siv-36). + if pool.GwLrpRangeStart != "" || pool.GwLrpRangeEnd != "" { + gwS := net.ParseIP(pool.GwLrpRangeStart) + gwE := net.ParseIP(pool.GwLrpRangeEnd) + if gwS == nil { + return fmt.Errorf("invalid gw_lrp_range_start: %q", pool.GwLrpRangeStart) + } + if gwE == nil { + return fmt.Errorf("invalid gw_lrp_range_end: %q", pool.GwLrpRangeEnd) + } + gwSi := ipToInt(gwS.To4()) + gwEi := ipToInt(gwE.To4()) + if gwSi.Cmp(gwEi) > 0 { + return fmt.Errorf("gw_lrp_range_start %s is greater than gw_lrp_range_end %s", + pool.GwLrpRangeStart, pool.GwLrpRangeEnd) + } + rangeSi := ipToInt(startIP.To4()) + rangeEi := ipToInt(endIP.To4()) + // Overlap test: !(gwE < rangeS || gwS > rangeE) + if gwEi.Cmp(rangeSi) >= 0 && gwSi.Cmp(rangeEi) <= 0 { + return fmt.Errorf("gw_lrp_range %s-%s overlaps range %s-%s", + pool.GwLrpRangeStart, pool.GwLrpRangeEnd, pool.RangeStart, pool.RangeEnd) + } + } } return nil } 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 7ed5f64ef..000000000 --- a/spinifex/handlers/ec2/vpc/external_ipam_dhcp.go +++ /dev/null @@ -1,150 +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 - -// 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) - } - - // Attempt 5 deliveres to NAT (timeout, service not started, temp errors) - var reply dhcp.AcquireReplyMsg - const maxAttempts = 5 - - 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(1 * time.Second) - 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 3051ed347..125ddfd5e 100644 --- a/spinifex/handlers/ec2/vpc/external_ipam_test.go +++ b/spinifex/handlers/ec2/vpc/external_ipam_test.go @@ -591,91 +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 }) - - _, 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) @@ -736,3 +651,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/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/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/topology.go b/spinifex/services/vpcd/topology.go index f0a98fb26..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" @@ -32,13 +33,23 @@ const ( TopicDeleteNATGateway = "vpc.delete-nat-gateway" ) -// gatewayPortNetwork is the link-local CIDR every IGW gateway-LRP carries. -// Pool IPs are reserved for per-VM dnat_and_snat external addresses; reusing -// them on the LRP causes intra-cluster ARP collisions when multiple VPCs -// share a br-ext localnet (mulga-siv-26). The default route's OutputPort is -// set explicitly so the WAN nexthop need not be on this subnet. +// gatewayPortNetwork is the link-local CIDR every IGW gateway-LRP carries +// in distributed-NAT (direct) mode. Per-VM dnat_and_snat handles ARP per +// chassis so the LRP IP itself never goes on the wire. The default route's +// OutputPort is set explicitly so the WAN nexthop need not be on this subnet. +// +// Centralized-NAT (veth/macvlan) bypasses this constant: the gateway LRP is +// the on-wire egress point and must hold a WAN-subnet IP from the pool's +// gw_lrp_range, otherwise the upstream router silently drops ARP requests +// from an off-subnet sender (RFC 826) and the default route never resolves +// (mulga-siv-36). const gatewayPortNetwork = "169.254.0.1/30" +// gatewayIPExtID is the LRP external_ids key holding the gw IP allocated +// from the pool's gw_lrp_range. Persisted so reconcile/retrofit can recover +// the assignment without re-allocating, and so siblings see it as "used". +const gatewayIPExtID = "spinifex:gateway_ip" + // VPCEvent is published on vpc.create after a VPC is persisted. type VPCEvent struct { VpcId string `json:"vpc_id"` @@ -107,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. @@ -145,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 { @@ -202,22 +227,81 @@ func (h *TopologyHandler) ensureLocalnetOptions(ctx context.Context, extPortName return nil } +// expectedGatewayPortNetwork returns the Networks CIDR a gateway LRP for +// 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 + } + pool := h.findExternalPool("", "") + 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 + } + if !ok { + return gatewayPortNetwork, "", nil + } + return fmt.Sprintf("%s/%d", ip, prefix), ip, nil +} + // ensureGatewayPortNetworks rewrites the gateway LRP's Networks column in -// place when it drifts from the link-local CIDR. CreateLogicalRouterPort is -// a no-op when the row exists, so reconcile of an upgraded cluster otherwise -// keeps the old pool-IP networks (mulga-siv-26 D8). Idempotent — no UPDATE -// is issued when already correct. +// place when it drifts from the mode's expected value. CreateLogicalRouterPort +// is a no-op when the row exists, so reconcile of an upgraded cluster +// otherwise keeps the stale pool-IP networks (mulga-siv-26 D8) or, in +// centralized NAT, the off-subnet link-local that breaks ARP upstream +// (mulga-siv-36). Idempotent — no UPDATE is issued when already correct. func (h *TopologyHandler) ensureGatewayPortNetworks(ctx context.Context, gwPortName string) error { lrp, err := h.ovn.GetLogicalRouterPort(ctx, gwPortName) if err != nil { return fmt.Errorf("get gateway router port %s: %w", gwPortName, err) } - if len(lrp.Networks) == 1 && lrp.Networks[0] == gatewayPortNetwork { + vpcId := lrp.ExternalIDs["spinifex:vpc_id"] + if vpcId == "" { + // No VPC tag — this is from the legacy "gw-vpc-..." naming or a + // hand-edited row; fall back to link-local to keep behavior safe. + if len(lrp.Networks) == 1 && lrp.Networks[0] == gatewayPortNetwork { + return nil + } + slog.Info("vpcd: rewriting stale gateway port networks (no vpc_id tag)", + "port", gwPortName, "old", lrp.Networks, "new", gatewayPortNetwork) + lrp.Networks = []string{gatewayPortNetwork} + return h.ovn.UpdateLogicalRouterPort(ctx, lrp) + } + wantNetwork, wantGwIP, err := h.expectedGatewayPortNetwork(ctx, vpcId) + if err != nil { + return fmt.Errorf("compute expected network for %s: %w", gwPortName, err) + } + currentGwIP := lrp.ExternalIDs[gatewayIPExtID] + if len(lrp.Networks) == 1 && lrp.Networks[0] == wantNetwork && currentGwIP == wantGwIP { return nil } slog.Info("vpcd: rewriting stale gateway port networks", - "port", gwPortName, "old", lrp.Networks, "new", gatewayPortNetwork) - lrp.Networks = []string{gatewayPortNetwork} + "port", gwPortName, "old", lrp.Networks, "new", wantNetwork, + "old_gw_ip", currentGwIP, "new_gw_ip", wantGwIP) + lrp.Networks = []string{wantNetwork} + if lrp.ExternalIDs == nil { + lrp.ExternalIDs = map[string]string{} + } + if wantGwIP == "" { + delete(lrp.ExternalIDs, gatewayIPExtID) + } else { + lrp.ExternalIDs[gatewayIPExtID] = wantGwIP + } if err := h.ovn.UpdateLogicalRouterPort(ctx, lrp); err != nil { return fmt.Errorf("update gateway router port %s: %w", gwPortName, err) } @@ -225,10 +309,10 @@ func (h *TopologyHandler) ensureGatewayPortNetworks(ctx context.Context, gwPortN } // RetrofitAllGatewayPortNetworks walks every LRP tagged -// spinifex:role=gateway and ensures Networks is the link-local CIDR. Needed +// spinifex:role=gateway and ensures Networks matches the current mode. Needed // because Reconcile/ReconcileFromKV early-return when the external switch // already exists, so reconcileIGW never runs against pre-existing topologies -// shipped by the buggy pool-IP build (mulga-siv-26 D8). Idempotent — the +// shipped by older builds (mulga-siv-26 D8 / mulga-siv-36). Idempotent — the // underlying ensureGatewayPortNetworks no-ops when already correct. func (h *TopologyHandler) RetrofitAllGatewayPortNetworks(ctx context.Context) { lrps, err := h.ovn.ListLogicalRouterPorts(ctx) @@ -804,37 +888,64 @@ func (h *TopologyHandler) handleIGWAttach(msg *nats.Msg) { return } - // Resolve external pool for this VPC's WAN nexthop. The router port - // itself always takes link-local 169.254.0.1/30 — pool IPs are reserved - // for per-VM dnat_and_snat external addresses (mulga-siv-26). Sharing a - // pool IP across VPCs causes intra-cluster ARP collisions on br-ext. + // Resolve external pool for this VPC's WAN nexthop and gw LRP IP. + // + // Direct mode: link-local LRP is fine — distributed NAT puts a per-VM + // dnat_and_snat row with external_mac/logical_port on every chassis, so + // 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 (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() { + 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) + respond(msg, allocErr) + return + } + gwLrpIP = ip + gatewayNetwork = network + } slog.Info("vpcd: using external pool for IGW", "pool", pool.Name, + "source", pool.Source, "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") } // 3. Create gateway router port on the VPC router connecting to external switch gwMAC := generateMAC("gw-" + evt.VpcId) + lrpExtIDs := map[string]string{ + "spinifex:vpc_id": evt.VpcId, + "spinifex:igw_id": evt.InternetGatewayId, + "spinifex:role": "gateway", + } + if gwLrpIP != "" { + lrpExtIDs[gatewayIPExtID] = gwLrpIP + } lrp := &nbdb.LogicalRouterPort{ - Name: gwPortName, - MAC: gwMAC, - Networks: []string{gatewayNetwork}, - ExternalIDs: map[string]string{ - "spinifex:vpc_id": evt.VpcId, - "spinifex:igw_id": evt.InternetGatewayId, - "spinifex:role": "gateway", - }, + Name: gwPortName, + MAC: gwMAC, + Networks: []string{gatewayNetwork}, + ExternalIDs: lrpExtIDs, } if err := h.ovn.CreateLogicalRouterPort(ctx, routerName, lrp); err != nil { slog.Error("vpcd: failed to create gateway router port", "port", gwPortName, "err", err) @@ -982,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, @@ -1194,13 +1309,22 @@ func (h *TopologyHandler) reconcileIGW(ctx context.Context, vpcId, igwId string) gwPortName := "gw-" + vpcId switchGWPortName := "gw-port-" + vpcId - // Router port always link-local; pool IPs reserved for per-VM - // dnat_and_snat external addresses (mulga-siv-26). + // Direct mode: link-local LRP. Centralized mode: allocate from + // pool.GwLrpRange so the LRP can ARP on the WAN subnet (mulga-siv-36). pool := h.findExternalPool("", "") gatewayNetwork := gatewayPortNetwork wanGateway := "169.254.0.2" + gwLrpIP := "" if pool != nil { wanGateway = pool.Gateway + if h.useCentralizedNAT() { + 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 = network + } } // Build external IDs with optional IGW ID @@ -1252,14 +1376,18 @@ func (h *TopologyHandler) reconcileIGW(ctx context.Context, vpcId, igwId string) // 3. Create gateway router port gwMAC := generateMAC("gw-" + vpcId) + reconcileLrpExtIDs := map[string]string{ + "spinifex:vpc_id": vpcId, + "spinifex:role": "gateway", + } + if gwLrpIP != "" { + reconcileLrpExtIDs[gatewayIPExtID] = gwLrpIP + } lrp := &nbdb.LogicalRouterPort{ - Name: gwPortName, - MAC: gwMAC, - Networks: []string{gatewayNetwork}, - ExternalIDs: map[string]string{ - "spinifex:vpc_id": vpcId, - "spinifex:role": "gateway", - }, + Name: gwPortName, + MAC: gwMAC, + Networks: []string{gatewayNetwork}, + ExternalIDs: reconcileLrpExtIDs, } if err := h.ovn.CreateLogicalRouterPort(ctx, routerName, lrp); err != nil { _ = h.ovn.DeleteLogicalSwitch(ctx, extSwitchName) @@ -1387,6 +1515,220 @@ func (h *TopologyHandler) reconcileGatewayChassis(ctx context.Context, validName // --- Helpers --- +// 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 { + return nil, nil, 0, false + } + prefix = pool.PrefixLen + if prefix <= 0 || prefix > 32 { + prefix = 24 + } + + // 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 +} + +// 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 +// existing IP is returned (idempotent reconcile). Returns ok=false when +// the pool has no gw_lrp_range — caller falls back to link-local. +func (h *TopologyHandler) allocateGatewayLRPIP(ctx context.Context, vpcId string, pool *ExternalPoolConfig) (ip string, prefix int, ok bool, err error) { + start, end, prefix, ok := gwLrpRange(pool) + if !ok { + return "", 0, false, nil + } + + gwPortName := "gw-" + vpcId + lrps, err := h.ovn.ListLogicalRouterPorts(ctx) + if err != nil { + return "", 0, false, fmt.Errorf("list LRPs for gw IP allocation: %w", err) + } + used := make(map[uint32]struct{}, len(lrps)) + for _, lrp := range lrps { + existing := lrp.ExternalIDs[gatewayIPExtID] + if existing == "" { + continue + } + if lrp.Name == gwPortName { + // Idempotent: already allocated to this VPC. + return existing, prefix, true, nil + } + if v := net.ParseIP(existing).To4(); v != nil { + used[ipv4ToUint32(v)] = struct{}{} + } + } + + startU := ipv4ToUint32(start) + endU := ipv4ToUint32(end) + for n := startU; n <= endU; n++ { + if _, taken := used[n]; taken { + continue + } + return uint32ToIPv4(n).String(), prefix, true, nil + } + return "", 0, false, fmt.Errorf("gw_lrp_range exhausted for pool %q (%s-%s)", pool.Name, pool.GwLrpRangeStart, pool.GwLrpRangeEnd) +} + +func ipv4ToUint32(ip net.IP) uint32 { + v := ip.To4() + return uint32(v[0])<<24 | uint32(v[1])<<16 | uint32(v[2])<<8 | uint32(v[3]) +} + +func uint32ToIPv4(n uint32) net.IP { + return net.IPv4(byte(n>>24&0xff), byte(n>>16&0xff), byte(n>>8&0xff), byte(n&0xff)).To4() +} + // subnetGateway computes the gateway IP (.1) from a CIDR string. // Returns the gateway IP string and the prefix length. func subnetGateway(cidr string) (string, int, error) { diff --git a/spinifex/services/vpcd/topology_test.go b/spinifex/services/vpcd/topology_test.go index 927ff8f50..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" @@ -929,10 +930,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 +975,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 +1002,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 +1027,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 +1075,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 +1126,359 @@ 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 +// mulga-siv-36: in centralized NAT (veth/macvlan) the gateway LRP must +// hold a WAN-subnet IP from pool.GwLrpRange so it can ARP the upstream +// nexthop. Three VPCs sharing one pool must each get a distinct IP and +// persist it on external_ids:spinifex:gateway_ip. +func TestTopologyHandler_IGWAttach_CentralizedNATAllocatesGwLrpIP(t *testing.T) { + _, nc := startTestNATS(t) + mock := NewMockOVNClient() + _ = mock.Connect(context.Background()) + ctx := context.Background() + + pools := []ExternalPoolConfig{ + { + Name: "lan", + RangeStart: "192.168.3.100", + RangeEnd: "192.168.3.150", + Gateway: "192.168.3.1", + PrefixLen: 24, + GwLrpRangeStart: "192.168.3.20", + GwLrpRangeEnd: "192.168.3.29", + }, + } + // Default bridge mode is centralized (useCentralizedNAT returns true + // when bridgeMode != "direct"). Explicit veth here to be unambiguous. + topo := NewTopologyHandler(mock, + WithExternalNetwork("pool", pools), + WithBridgeMode(BridgeModeVeth), + ) + subs, err := topo.Subscribe(nc) + if err != nil { + t.Fatalf("subscribe: %v", err) + } + defer func() { + for _, s := range subs { + _ = s.Unsubscribe() + } + }() + + vpcIds := []string{"vpc-cn-a", "vpc-cn-b", "vpc-cn-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) + } + net := gwPort.Networks[0] + if !strings.HasPrefix(net, "192.168.3.") || !strings.HasSuffix(net, "/24") { + t.Errorf("%s: expected 192.168.3.X/24 from gw_lrp_range, got %s", gwPortName, net) + } + gwIP := gwPort.ExternalIDs["spinifex:gateway_ip"] + if gwIP == "" { + t.Errorf("%s: missing spinifex:gateway_ip external_id", gwPortName) + } + if other, dup := seen[gwIP]; dup { + t.Errorf("gw IP collision: %s and %s share %s", gwPortName, other, gwIP) + } + seen[gwIP] = gwPortName + + // Idempotent: re-attach must not change the assignment. + evt := types.IGWEvent{InternetGatewayId: "igw-" + vpcId, VpcId: vpcId} + data, _ := json.Marshal(evt) + _, _ = nc.Request(TopicIGWAttach, data, 5_000_000_000) + gwPort2, _ := mock.GetLogicalRouterPort(ctx, gwPortName) + if gwPort2.ExternalIDs["spinifex:gateway_ip"] != gwIP { + t.Errorf("%s: re-attach changed gateway_ip from %s to %s", + gwPortName, gwIP, gwPort2.ExternalIDs["spinifex:gateway_ip"]) + } + } +} + +// 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) { @@ -2742,3 +3124,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) + } + }) + } +} diff --git a/spinifex/services/vpcd/vpcd.go b/spinifex/services/vpcd/vpcd.go index dd6ad5806..41cd0abbc 100644 --- a/spinifex/services/vpcd/vpcd.go +++ b/spinifex/services/vpcd/vpcd.go @@ -15,6 +15,7 @@ import ( "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. @@ -80,15 +81,24 @@ type Config struct { // ExternalPoolConfig mirrors config.ExternalPool for vpcd's internal use. type ExternalPoolConfig struct { - Name string - RangeStart string - RangeEnd string - Gateway string - GatewayIP string - PrefixLen int - DNSServers []string - Region string - AZ string + Name string + Source string // "static" (default) or "dhcp" + RangeStart string + RangeEnd string + Gateway string + GatewayIP string + PrefixLen int + 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. @@ -342,6 +352,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 @@ -353,11 +364,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. @@ -384,24 +390,31 @@ 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 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) 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. + // + // 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 @@ -423,6 +436,19 @@ 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. + 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. @@ -450,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