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