gsadmin/spinifex
1
0
Fork 0
mirror of https://github.com/mulgadc/spinifex.git synced 2026-10-15 17:11:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(vpcd): public-IP boot race + DHCP DORA hardening (siv-33/34/36/37/38/39) (#177)

Browse Source 
* fix(installer): emit kernel cmdline to ttyS0 alongside tty0 (mulga-siv-33)

GRUB_CMDLINE_LINUX in /etc/default/grub only listed console=tty0, so the
installed kernel printed nothing on serial after the GRUB menu. Headless
installs (CI, racks with serial-only IPMI, remote-dev SSH-to-QEMU) saw a
black screen forever and had no path to debug boot failures.

Append console=ttyS0,115200 — the last `console=` on the cmdline becomes
the system console, so ttyS0 wins on serial-only hardware while tty0
still receives output on local displays.

Plan: docs/development/bugs/iso-public-ip-boot-race.md

* fix(ipam): widen vpc.dhcp.acquire retry budget to 60 s (mulga-siv-34)

ObtainDHCPLease retried only 5 attempts × 1 s before returning a hard
error. On a cold ISO boot, NATS comes up well before vpcd's reconcile
leader-election + dhcp.Manager subscribe (typical 5–30 s, pathological
~60 s during full reconcile), so daemon's NewExternalIPAM races vpcd's
dhcp.acquire subscription and fails permanently — d.externalIPAM stays
nil for the daemon's lifetime, daemon_handlers_instance.go silently
skips all auto-public-IP allocation, and instances launch with no
public IP, no dnat_and_snat, and no NAT event ever published.

Bump maxAttempts to 60 with 1 s spacing; expose both as package vars
so tests can shrink the budget for fast timeout-path coverage. Pure
widening of the existing retry loop — no behaviour change on the
success path, only the worst-case wallclock when vpcd never comes up.

The dominant cause of slow vpcd starts (mulga-siv-33: systemd-networkd
not enabled in the ISO → veth pair vanishes on reboot → vpcd
crashloops) is fixed alongside this. A 60 s budget fits comfortably
above all observed healthy-startup times.

Plan: docs/development/bugs/iso-public-ip-boot-race.md

* fix(setup-ovn): declare Linux WAN bridge as NetDev so networkd-managed veth resolves Bridge= on reboot (mulga-siv-33)

The veth-mode reboot persistence drop-ins added in mulga-998.b were
incomplete: setup-ovn.sh wrote `15-spinifex-veth-wan.network` with
`Bridge=br-wan`, but never declared br-wan itself to networkd.

br-wan is created by `networking.service` (ifupdown via
`/etc/network/interfaces`, generated by `install.go writeNetworkConfig`),
not networkd. networkd's `Bridge=` directive requires the target to be a
known NetDev — when it isn't, networkd silently logs
`br-wan NetDev could not be found, ignoring assignment` and the veth
ends up with no master. vpcd's bridge-mode sanity check then fails on
every reboot.

Fix: also write `14-spinifex-br-wan.netdev` declaring br-wan as
`Kind=bridge`. networkd matches the existing kernel bridge by name+kind
(the `Failed to create netdev: File exists` log is harmless), and
`Bridge=br-wan` resolves on next boot. ifupdown still owns DHCP and the
IP — networkd just needs the NetDev declaration to enslave the veth.

The write is gated to avoid clobbering operator-managed networkd config:
if any other `.netdev` under `/etc/systemd/network`,
`/run/systemd/network`, or `/usr/lib/systemd/network` already declares
the WAN bridge by name, the write is skipped with a log line pointing
at the operator file. Cleanup hook also removes the new file when
switching away from veth mode.

Verified end-to-end via QEMU repro: clean install + first boot + reboot,
veth-wan-br master persists across reboots, vpcd starts cleanly.

* fix(vpcd/dhcp): enable AF_PACKET promisc on bridge for DORA (mulga-siv-33)

Real-world DHCP servers commonly ignore the BOOTP BROADCAST flag and
reply with a unicast OFFER addressed to the synthetic per-tenant chaddr
(e.g. 02:f3:...). A Linux bridge floods unknown unicast to slave ports
but does NOT pass it up to the bridge netdev unless IFF_PROMISC is set,
so vpcd's AF_PACKET socket on br-wan misses the OFFER and DORA times
out at 15s ("unable to receive an offer: context deadline exceeded").

Dev/CI didn't hit this:
- CI uses spx_wan_mode=static — no DORA via vpcd.
- QEMU dev DHCP servers (SLIRP/dnsmasq) honor the BROADCAST flag, so
  their replies are L2 broadcast which the bridge always passes up.

Replace the three nclient4.New(iface, ...) call sites in Acquire/Renew/
Release with nclient4.NewWithConn(...) wrapping a custom AF_PACKET
DGRAM conn that calls SetPromiscuous(true). Promisc is per-socket
(per-DORA, ~15s) and refcounted by the kernel — auto-clears on close.

* revert(vpcd/dhcp): drop AF_PACKET promisc wrapper, restore nclient4.New

ecfbd4f swapped nclient4.New for openPromiscBridgeConn + NewWithConn to
catch unicast OFFER replies that bypass the bridge netdev. On dev hosts
where upstream honors the BOOTP BROADCAST flag, the wrapper regressed
DORA — public-IP allocation failed where main works. Reverting until
the promisc path is validated against both broadcast- and unicast-OFFER
servers.

* fix(setup-ovn): mark br-int/br-ext Unmanaged for systemd-networkd (mulga-siv-37)

Fresh ISO install on Trixie+systemd-networkd left the OVS internal netdevs
admin-down: networkd has no matching .network so it tags them as unmanaged
but never brings them up. OVS dataplane works but ovn-controller flow
programming and any tooling that probes link state misbehaves.

Drop /etc/systemd/network/05-spinifex-ovs-internal.network with
Unmanaged=yes so OVS itself owns the link state — `ovs-vsctl add-br`
already runs the netdev up at create time.

* fix(vpcd): allocate gateway LRP IP from pool in centralized NAT (mulga-siv-36)

In veth/macvlan modes the gateway LRP carries all egress, so it must hold
a WAN-subnet IP — otherwise the upstream router silently drops ARP for
the WAN nexthop (RFC 826) and the default route never resolves. The
hardcoded 169.254.0.1/30 was correct for distributed NAT but regressed
centralized NAT (introduced under mulga-siv-26).

Add gw_lrp_range_start/end to ExternalPool. In centralized mode at IGW
attach the gateway LRP gets a fresh IP from that range, persisted as
external_ids:spinifex:gateway_ip on the LRP for idempotent reconcile and
retrofit. Direct mode keeps link-local. Per-VM EIP allocator skips the
gw range; ValidatePoolConfig rejects overlap with the EIP range.

Adds TestTopologyHandler_IGWAttach_CentralizedNATAllocatesGwLrpIP and
extends RetrofitAllGatewayPortNetworks to migrate stale link-local LRPs
to allocated pool IPs at startup.

* fix(vpcd): drop link-local fallback, auto-derive gw_lrp_range (mulga-siv-36)

In centralized NAT (veth/macvlan) the gateway LRP must hold a WAN-subnet
IP — link-local cannot ARP-resolve the upstream nexthop, so the upstream
router silently drops egress. Hardware deployment confirmed the previous
fallback to 169.254.0.1/30 was always reached when gw_lrp_range was not
explicitly configured.

Auto-derive a 16-IP range at the top of the WAN subnet (broadcast - 16
.. broadcast - 1) when gw_lrp_range is unset; shift below RangeStart on
overlap with the per-VM EIP range; clamp inside the subnet; guard the
upstream gateway IP. IGW attach now errors out instead of falling back
to link-local when allocation fails — surfaces misconfiguration loudly.

Adds focused unit tests for gwLrpRange, ValidatePoolConfig and
nextAvailableExternalIP gw-lrp skip; updates the multi-VPC and
PoolWithGatewayIP regression tests.

* feat(vpcd): DHCP-acquire gateway LRP IPs (mulga-siv-38)

Per-VPC OVN gateway LRPs now obtain their WAN-subnet IP via vpc.dhcp.acquire
when the pool is source="dhcp", instead of carving from a static gw_lrp_range.
The upstream DHCP server becomes the single arbiter of LAN address space; a
cluster grows by N nodes without operator-side range carving in spinifex.toml.

- Consolidate ObtainDHCPLease/ReleaseDHCPLease into the dhcp package as
  RequestAcquire/RequestRelease so vpcd and the daemon-side ExternalIPAM
  share one helper. handlers/ec2/vpc keeps its existing ENI/EIP DHCP path.

- TopologyHandler dispatches by pool source: DHCP pools call DHCPManager
  via NATS (client-id "gw-lrp-<vpcId>", idempotent on retry); static pools
  keep the siv-36 gw_lrp_range / auto-derive allocator. Direct mode and
  no-pool deployments still use link-local.

- IGW detach issues vpc.dhcp.release so the upstream server reclaims the
  IP. Best-effort — server-side expiry is the safety net.

- Tests: TestTopologyHandler_IGWAttach_DHCPGwLrpIP (3 VPCs share one DHCP
  pool, distinct IPs, idempotent attach, release on detach),
  IGWAttach_DHCPAcquireFailureAborts (no link-local fallback when DHCP
  errors), TestPrefixFromMask. Existing static / auto-derive tests
  preserved.

* fix(vpcd): subscribe DHCPManager before reconcile to fix boot race (mulga-siv-38)

DHCPManager.Subscribe ran AFTER ReconcileFromKV / RetrofitAllGatewayPortNetworks,
so retrofit's vpc.dhcp.acquire requests for existing IGWs hit "no responders" and
burned the 60-attempt × 1 s cold-boot retry budget per VPC before failing. On a
dev box with one bootstrap VPC the symptom is a 60 s vpcd startup hang followed
by "allocate gw LRP IP failed", and the bootstrap IGW topology missing from OVN
until the next reconcile pass.

Move dhcpManager.Subscribe ahead of the two reconcile passes so the subscription
is live before any topology code can fan out an acquire.

* fix(vpcd): also subscribe DHCPManager before bootstrap reconcile pass

Previous fix moved DHCPManager.Subscribe ahead of ReconcileFromKV but
missed Pass 1 (the bootstrap-config Reconcile that runs even earlier).
On a dev box with a source="dhcp" pool and a bootstrap IGW, that earlier
pass still hit "no responders" and burned the 60-retry cold-boot budget
before the Manager came up. Move both reconcile passes after Subscribe.

* fix(vpcd): RFC 2131 §4.1 DORA exponential backoff (mulga-siv-39)

DHCPManager.Acquire previously did one DISCOVER and gave up after 15s
because the outer ctx.WithTimeout matched the per-attempt timeout —
nclient4's WithRetry(3) could never fire. Real DHCP servers (consumer
routers, ISC dhcpd, dnsmasq) silently drop OFFERs under load; the spec
mandates retransmission with exponential backoff, and a single lost
DISCOVER was killing legitimate boots.

Move retransmission into DHCPManager.acquireWithBackoff with a
4s/8s/16s/32s schedule (±1s jitter, ~60s worst case). NClient4Client
becomes single-attempt; manager owns retry policy. Default
acquireTimeout lifts from 15s to 90s (sum of schedule + slack).

* fix(vpcd): retry NewDHCPManager on JetStream-not-ready (mulga-siv-40)

NewDHCPManager creates the spinifex-dhcp-leases KV bucket via
utils.GetOrCreateKVBucket → js.CreateKeyValue with no retry. After the
mulga-siv-38 launchService reorder, this runs before any other JetStream
user, so on multi-node cold-boot vpcd hits "JetStream system temporarily
unavailable" and exits before cluster quorum forms. Pseudo-multinode CI
calls start_node_services sequentially, so node2's vpcd failure aborts
the script before node3 ever launches and quorum then never forms.

Wrap NewDHCPManager in newDHCPManagerWithRetry (500ms→10s, 5 min cap,
ctx-cancellable) mirroring daemon.initJetStream so vpcd waits as
patiently as the daemon for the cluster to come up.

* fix(daemon): idempotent DetachVolume QMP path (mulga-siv-41)

DetachVolume's QMP sequence was not retry-safe. First call: device_del
succeeds, blockdev-del fails with "Node nbd-vol-X is in use" because NBD
client teardown / in-flight I/O still hold the block node. Handler bails
ServerInternal "leaving volume state intact" but the guest device is
actually gone. AWS-CLI retries: device_del returns DeviceNotFound,
handler bails again. After 2 retries aws cli exits 254 — Single-Node
Phase 8 Step 3 reproducibly fails this way.

Phase 1: treat DeviceNotFound on device_del as already-detached, log and
continue. Detach is now idempotent across retries.

Phase 2: tryBlockdevDel polls blockdev-del up to 20×detachDelay on "in
use" errors so the I/O drain can complete instead of failing the whole
detach. Other QMP errors fall through immediately as before.
This commit is contained in:
Josh Siviour
2026-05-01 09:48:52 +10:00
committed by GitHub
parent 6ec2f4bd08
commit 69404b38ce
18 changed files with 1866 additions and 403 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/installer/install/install.go
+6 -5
View File
@@ -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 {
cmd/spinifex/cmd/service.go
+13 -9
View File
@@ -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,
})
}
go.mod
+2 -2
View File
@@ -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
scripts/setup-ovn.sh
+50 -1
View File
@@ -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 <<NETDEV
[NetDev]
Name=$LINUX_BRIDGE
Kind=bridge
NETDEV
fi
sudo tee "$VETH_NETDEV" >/dev/null <<NETDEV
[NetDev]
Name=veth-wan-br
spinifex/config/config.go
+8
View File
@@ -31,6 +31,14 @@ type ExternalPool struct {
DNSServers []string `mapstructure:"dns_servers"` // DNS servers for VM DHCP (auto-detected from host; fallback: 8.8.8.8, 1.1.1.1)
Region string `mapstructure:"region"` // Scope to region (optional — empty means any region)
AZ string `mapstructure:"az"` // Scope to AZ (optional — empty means any AZ in region)
// GwLrpRangeStart/End reserve a sub-range of the LAN for OVN gateway LRP IPs
// in centralized NAT mode (veth/macvlan). Each VPC consumes one IP from this
// range so its gateway router port can ARP with a sender IP on the LAN
// subnet — link-local 169.254.0.1/30 makes upstream routers reject ARP per
// RFC 826 (mulga-siv-36). Must NOT overlap [RangeStart, RangeEnd] or IPAM
// and vpcd will fight over the same IP.
GwLrpRangeStart string `mapstructure:"gw_lrp_range_start"`
GwLrpRangeEnd string `mapstructure:"gw_lrp_range_end"`
}
// NetworkConfig holds cluster-wide external network settings.
spinifex/daemon/daemon.go
+12 -10
View File
@@ -692,16 +692,18 @@ func (d *Daemon) Start() error {
}
for _, p := range d.clusterConfig.Network.ExternalPools {
pools = append(pools, handlers_ec2_vpc.ExternalPoolConfig{
Name: p.Name,
Source: p.Source,
RangeStart: p.RangeStart,
RangeEnd: p.RangeEnd,
Gateway: p.Gateway,
GatewayIP: p.GatewayIP,
PrefixLen: p.PrefixLen,
Region: p.Region,
AZ: p.AZ,
DhcpBindBridge: dhcpBindBridge,
Name: p.Name,
Source: p.Source,
RangeStart: p.RangeStart,
RangeEnd: p.RangeEnd,
Gateway: p.Gateway,
GatewayIP: p.GatewayIP,
PrefixLen: p.PrefixLen,
Region: p.Region,
AZ: p.AZ,
DhcpBindBridge: dhcpBindBridge,
GwLrpRangeStart: p.GwLrpRangeStart,
GwLrpRangeEnd: p.GwLrpRangeEnd,
})
}
d.externalIPAM, err = handlers_ec2_vpc.NewExternalIPAM(d.natsConn, js, pools)
spinifex/daemon/daemon_handlers_volume.go
+85 -15
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"log/slog"
"strconv"
"strings"
"time"
"github.com/aws/aws-sdk-go/service/ec2"
@@ -357,32 +358,42 @@ func (d *Daemon) handleDetachVolume(msg *nats.Msg, command types.EC2InstanceComm
nodeName := fmt.Sprintf("nbd-%s", volumeID)
iothreadID := fmt.Sprintf("ioth-%s", volumeID)
// Phase 1: QMP device_del (remove guest device)
// Phase 1: QMP device_del (remove guest device).
// Idempotent: a prior detach may have already removed the device but
// failed at blockdev-del, leaving the volume's external state intact.
// Treat DeviceNotFound as success so the retry can drive blockdev-del
// to completion without bailing on the now-absent guest device.
_, err := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{
Execute: "device_del",
Arguments: map[string]any{"id": deviceID},
}, instance.ID)
if err != nil {
if !force {
slog.Error("DetachVolume: QMP device_del failed", "volumeId", volumeID, "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
switch {
case err == nil:
case isQMPDeviceNotFound(err):
slog.Info("DetachVolume: guest device already removed (resuming detach)", "volumeId", volumeID, "err", err)
case force:
slog.Warn("DetachVolume: QMP device_del failed (force=true, continuing)", "volumeId", volumeID, "err", err)
default:
slog.Error("DetachVolume: QMP device_del failed", "volumeId", volumeID, "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
// Brief pause for guest to acknowledge PCI removal
time.Sleep(d.detachDelay)
// Phase 2: QMP blockdev-del (remove block node)
_, blockdevErr := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{
Execute: "blockdev-del",
Arguments: map[string]any{"node-name": nodeName},
}, instance.ID)
// Phase 2: QMP blockdev-del (remove block node).
// Poll-retry on "node is in use": after device_del, NBD client teardown
// and any in-flight I/O can hold the block node briefly. Bailing here
// without retry leaves the guest device gone but the block node intact,
// and the next AWS-CLI retry hits DeviceNotFound on device_del with no
// path forward (mulga-siv-41).
blockdevErr := d.tryBlockdevDel(instance, nodeName)
if blockdevErr != nil {
// Block node still referenced by QEMU; do not clean up state or unmount —
// tearing down the NBD server would crash the VM, and removing metadata
// would allow the volume to be double-attached.
// Block node still referenced after retry budget exhausted; do not
// clean up state or unmount — tearing down the NBD server would
// crash the VM, and removing metadata would allow the volume to be
// double-attached.
slog.Error("DetachVolume: QMP blockdev-del failed, leaving volume state intact", "volumeId", volumeID, "err", blockdevErr)
respondWithError(msg, awserrors.ErrorServerInternal)
return
@@ -504,3 +515,62 @@ func (d *Daemon) handleEC2ModifyVolume(msg *nats.Msg) {
func (d *Daemon) handleEC2DeleteVolume(msg *nats.Msg) {
handleNATSRequest(msg, d.volumeService.DeleteVolume)
}
// tryBlockdevDel issues blockdev-del with bounded retry on "is in use"
// errors. After device_del, NBD client teardown and any in-flight guest
// I/O can briefly hold the block node; QEMU surfaces this as a
// GenericError carrying "Node <name> 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")
}
spinifex/handlers/ec2/vpc/external_ipam.go
+89 -30
View File
@@ -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
}
spinifex/handlers/ec2/vpc/external_ipam_dhcp.go
-150
View File
@@ -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
}
spinifex/handlers/ec2/vpc/external_ipam_test.go
+85 -85
View File
@@ -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)
})
}
}
spinifex/services/vpcd/dhcp/nats_client.go
+150
View File
@@ -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
}
spinifex/services/vpcd/dhcp/nats_client_test.go
+104
View File
@@ -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, ""))
}
spinifex/services/vpcd/dhcp/nclient4.go
+17 -10
View File
@@ -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)
spinifex/services/vpcd/dhcp_manager.go
+98 -8
View File
@@ -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)
spinifex/services/vpcd/dhcp_manager_test.go
+119
View File
@@ -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)
spinifex/services/vpcd/topology.go
+379 -37
View File
@@ -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-<vpcId> 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) {
spinifex/services/vpcd/topology_test.go
+555 -15
View File
@@ -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-<vpcId>
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)
}
})
}
}
spinifex/services/vpcd/vpcd.go
+94 -26
View File
@@ -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