diff --git a/cmd/spinifex/cmd/service.go b/cmd/spinifex/cmd/service.go index cc9ddbfcc..6cdc7cc3f 100644 --- a/cmd/spinifex/cmd/service.go +++ b/cmd/spinifex/cmd/service.go @@ -735,6 +735,7 @@ var vpcdStartCmd = &cobra.Command{ for _, p := range clusterConfig.Network.ExternalPools { extPools = append(extPools, vpcd.ExternalPoolConfig{ Name: p.Name, + Source: p.Source, RangeStart: p.RangeStart, RangeEnd: p.RangeEnd, Gateway: p.Gateway, @@ -743,6 +744,7 @@ var vpcdStartCmd = &cobra.Command{ DNSServers: p.DNSServers, Region: p.Region, AZ: p.AZ, + DhcpBindBridge: nodeConfig.VPCD.DhcpBindBridge, GwLrpRangeStart: p.GwLrpRangeStart, GwLrpRangeEnd: p.GwLrpRangeEnd, }) diff --git a/spinifex/handlers/ec2/vpc/external_ipam.go b/spinifex/handlers/ec2/vpc/external_ipam.go index c336af7ef..b47ea14f7 100644 --- a/spinifex/handlers/ec2/vpc/external_ipam.go +++ b/spinifex/handlers/ec2/vpc/external_ipam.go @@ -9,6 +9,7 @@ import ( "net" "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -28,7 +29,7 @@ type ExternalIPAllocation struct { Note string `json:"note,omitempty"` // Human-readable note // DHCP-sourced metadata — set only for allocations whose pool has - // source="dhcp" (see ObtainDHCPLease). These fields let describe + // source="dhcp" (see dhcp.RequestAcquire). These fields let describe // surfaces and operators trace a given IP back to the upstream lease // without re-querying vpcd. LeaseExpiresUnix int64 `json:"lease_expires_unix,omitempty"` // Absolute expiry from the server's ACK @@ -173,9 +174,9 @@ func (m *ExternalIPAM) initPool(pool ExternalPoolConfig) error { // For DHCP pools, obtain the gateway IP from router DHCP if not set. // The gateway lease is long-lived: vpcd's renewal goroutine keeps it // refreshed until the pool is torn down. - var gatewayLease *dhcpLeaseResult + var gatewayLease *dhcp.LeaseResult if gwIP == "" && pool.IsDHCP() { - lease, dhcpErr := ObtainDHCPLease(m.nc, pool.DhcpBindBridge, + lease, dhcpErr := dhcp.RequestAcquire(m.nc, pool.DhcpBindBridge, "gateway-"+pool.Name, "gateway-"+pool.Name, "mulga-spinifex-gw", @@ -267,10 +268,10 @@ func (m *ExternalIPAM) allocateFromPool(poolName, allocType, allocID, eniID, ins // Manager is idempotent on ClientID, so CAS retries below will not // trigger a second DORA — we only release at the end if no CAS // attempt succeeded. - var dhcpLease *dhcpLeaseResult + var dhcpLease *dhcp.LeaseResult if pool != nil && pool.IsDHCP() { hostname, vendorClass := dhcpIdentityOptions(eniID, instanceID, poolName) - lease, err := ObtainDHCPLease(m.nc, pool.DhcpBindBridge, clientID, hostname, vendorClass, poolName) + lease, err := dhcp.RequestAcquire(m.nc, pool.DhcpBindBridge, clientID, hostname, vendorClass, poolName) if err != nil { return "", fmt.Errorf("DHCP lease for %s: %w", clientID, err) } @@ -281,7 +282,7 @@ func (m *ExternalIPAM) allocateFromPool(poolName, allocType, allocID, eniID, ins record, revision, err := m.getRecord(poolName) if err != nil { if dhcpLease != nil { - _ = ReleaseDHCPLease(m.nc, clientID) + _ = dhcp.RequestRelease(m.nc, clientID) } return "", fmt.Errorf("get external IPAM record: %w", err) } @@ -313,7 +314,7 @@ func (m *ExternalIPAM) allocateFromPool(poolName, allocType, allocID, eniID, ins data, err := json.Marshal(record) if err != nil { if dhcpLease != nil { - _ = ReleaseDHCPLease(m.nc, clientID) + _ = dhcp.RequestRelease(m.nc, clientID) } return "", fmt.Errorf("marshal external IPAM record: %w", err) } @@ -334,7 +335,7 @@ func (m *ExternalIPAM) allocateFromPool(poolName, allocType, allocID, eniID, ins // All CAS attempts exhausted — release any DHCP lease we acquired so // vpcd doesn't hold a binding for an IP we won't use. if dhcpLease != nil { - _ = ReleaseDHCPLease(m.nc, clientID) + _ = dhcp.RequestRelease(m.nc, clientID) } return "", fmt.Errorf("external IPAM allocation failed after CAS retries for pool %s", poolName) } @@ -389,7 +390,7 @@ func (m *ExternalIPAM) ReleaseIP(poolName, ip string) error { if clientID == "" { clientID = alloc.InstanceId } - if releaseErr := ReleaseDHCPLease(m.nc, clientID); releaseErr != nil { + if releaseErr := dhcp.RequestRelease(m.nc, clientID); releaseErr != nil { slog.Warn("Failed to release DHCP lease", "pool", poolName, "ip", ip, "err", releaseErr) } } diff --git a/spinifex/handlers/ec2/vpc/external_ipam_dhcp.go b/spinifex/handlers/ec2/vpc/external_ipam_dhcp.go deleted file mode 100644 index ea3473c83..000000000 --- a/spinifex/handlers/ec2/vpc/external_ipam_dhcp.go +++ /dev/null @@ -1,160 +0,0 @@ -package handlers_ec2_vpc - -import ( - "encoding/json" - "fmt" - "log/slog" - "time" - - "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" - "github.com/nats-io/nats.go" -) - -// dhcpNATSTimeout bounds the request/reply to spinifex-vpcd. Picked to -// comfortably exceed the DORA handshake (15 s default + server jitter + -// ~10 s of slack for NAK + fresh DISCOVER fallbacks). A var rather than -// a const so tests can drive the timeout error path without waiting -// 20 s per table entry. -var dhcpNATSTimeout = 30 * time.Second - -// dhcpAcquireMaxAttempts caps retry of vpc.dhcp.acquire. 60 attempts × 1 s -// sleep covers a cold-boot vpcd start window where NATS is up but the -// dhcp.Manager hasn't yet subscribed (typical: 5–30 s, pathological: 60 s -// during full reconcile). Tests override to keep timeout-path coverage -// fast — see external_ipam_test.go. -var dhcpAcquireMaxAttempts = 60 - -// dhcpAcquireRetryDelay is the wait between retries. Var for the same -// reason as dhcpAcquireMaxAttempts. -var dhcpAcquireRetryDelay = 1 * time.Second - -// dhcpLeaseResult is the data ExternalIPAM needs from a successful acquire. -// It maps directly from dhcp.AcquireReplyMsg; kept as a handler-side type -// so the IPAM ledger records the fields without coupling to the NATS wire -// struct. -type dhcpLeaseResult struct { - IP string - SubnetMask string - Routers []string - DNS []string - ServerID string - HWAddr string - ExpiresUnix int64 -} - -// ObtainDHCPLease asks spinifex-vpcd to acquire a DHCP lease on the given -// bridge, identifying us with option 61 (client-id), option 12 (hostname) -// and option 60 (vendor class). Blocks until vpcd replies or -// dhcpNATSTimeout expires. The Manager-side handler is idempotent: a -// second call with the same clientID while a live lease exists returns -// the same lease without a fresh DORA, so CAS retry loops on the caller -// are safe. -func ObtainDHCPLease(nc *nats.Conn, bridge, clientID, hostname, vendorClass, poolName string) (dhcpLeaseResult, error) { - if nc == nil { - return dhcpLeaseResult{}, fmt.Errorf("DHCP lease: NATS connection is required") - } - if bridge == "" { - return dhcpLeaseResult{}, fmt.Errorf("DHCP lease: bridge name is required") - } - if clientID == "" { - return dhcpLeaseResult{}, fmt.Errorf("DHCP lease: client ID is required") - } - - req := dhcp.AcquireRequestMsg{ - Bridge: bridge, - ClientID: clientID, - Hostname: hostname, - VendorClass: vendorClass, - PoolName: poolName, - } - data, err := json.Marshal(req) - if err != nil { - return dhcpLeaseResult{}, fmt.Errorf("marshal dhcp acquire request: %w", err) - } - - var reply dhcp.AcquireReplyMsg - maxAttempts := dhcpAcquireMaxAttempts - - for attempt := 1; attempt <= maxAttempts; attempt++ { - msg, err := nc.Request(dhcp.TopicAcquire, data, dhcpNATSTimeout) - if err != nil { - if attempt == maxAttempts { - return dhcpLeaseResult{}, fmt.Errorf( - "dhcp acquire NATS request failed after %d attempts (client %s): %w", - maxAttempts, - clientID, - err, - ) - } - - slog.Warn( - "dhcp acquire NATS request failed, retrying", - "clientID", clientID, - "attempt", attempt, - "maxAttempts", maxAttempts, - "err", err, - ) - - time.Sleep(dhcpAcquireRetryDelay) - continue - } - - if err := json.Unmarshal(msg.Data, &reply); err != nil { - return dhcpLeaseResult{}, fmt.Errorf("unmarshal dhcp acquire reply: %w", err) - } - - if reply.Error != "" { - return dhcpLeaseResult{}, fmt.Errorf("dhcp acquire client %s: %s", clientID, reply.Error) - } - - break - } - - slog.Info("DHCP lease obtained", - "bridge", bridge, - "client_id", clientID, - "ip", reply.IP, - "server_id", reply.ServerID, - "expires_unix", reply.ExpiresUnix, - ) - - return dhcpLeaseResult{ - IP: reply.IP, - SubnetMask: reply.SubnetMask, - Routers: reply.Routers, - DNS: reply.DNS, - ServerID: reply.ServerID, - HWAddr: reply.HWAddr, - ExpiresUnix: reply.ExpiresUnix, - }, nil -} - -// ReleaseDHCPLease asks spinifex-vpcd to release the lease identified by -// clientID. Returns nil (silently) when nc or clientID is empty so that -// callers in clean-up paths can invoke it unconditionally. -func ReleaseDHCPLease(nc *nats.Conn, clientID string) error { - if nc == nil || clientID == "" { - return nil - } - - data, err := json.Marshal(dhcp.ReleaseRequestMsg{ClientID: clientID}) - if err != nil { - return fmt.Errorf("marshal dhcp release request: %w", err) - } - - msg, err := nc.Request(dhcp.TopicRelease, data, dhcpNATSTimeout) - if err != nil { - return fmt.Errorf("dhcp release NATS request (client %s): %w", clientID, err) - } - - var reply dhcp.ReleaseReplyMsg - if err := json.Unmarshal(msg.Data, &reply); err != nil { - return fmt.Errorf("unmarshal dhcp release reply: %w", err) - } - if reply.Error != "" { - return fmt.Errorf("dhcp release (client %s): %s", clientID, reply.Error) - } - - slog.Info("DHCP lease released", "client_id", clientID) - return nil -} diff --git a/spinifex/handlers/ec2/vpc/external_ipam_test.go b/spinifex/handlers/ec2/vpc/external_ipam_test.go index cbce5d810..125ddfd5e 100644 --- a/spinifex/handlers/ec2/vpc/external_ipam_test.go +++ b/spinifex/handlers/ec2/vpc/external_ipam_test.go @@ -591,100 +591,6 @@ func TestExternalIPAM_DHCPGatewayFromLease(t *testing.T) { assert.Equal(t, "mulga-spinifex-gw", got.VendorClass) } -func TestObtainDHCPLease_RequestTimeout(t *testing.T) { - _, nc := testutil.StartTestNATS(t) - // Stub never responds — verifies that a hung vpcd surfaces as a - // wrapped NATS timeout error rather than hanging the caller. - sub, err := nc.Subscribe(dhcp.TopicAcquire, func(*nats.Msg) {}) - require.NoError(t, err) - t.Cleanup(func() { _ = sub.Unsubscribe() }) - - prev := dhcpNATSTimeout - dhcpNATSTimeout = 150 * time.Millisecond - t.Cleanup(func() { dhcpNATSTimeout = prev }) - - prevMax := dhcpAcquireMaxAttempts - prevDelay := dhcpAcquireRetryDelay - dhcpAcquireMaxAttempts = 2 - dhcpAcquireRetryDelay = 10 * time.Millisecond - t.Cleanup(func() { - dhcpAcquireMaxAttempts = prevMax - dhcpAcquireRetryDelay = prevDelay - }) - - _, err = ObtainDHCPLease(nc, "br-wan", "eni-timeout", "eni-timeout", "mulga-spinifex", "wan") - assert.ErrorContains(t, err, "dhcp acquire NATS request") -} - -func TestReleaseDHCPLease_RequestTimeout(t *testing.T) { - _, nc := testutil.StartTestNATS(t) - sub, err := nc.Subscribe(dhcp.TopicRelease, func(*nats.Msg) {}) - require.NoError(t, err) - t.Cleanup(func() { _ = sub.Unsubscribe() }) - - prev := dhcpNATSTimeout - dhcpNATSTimeout = 150 * time.Millisecond - t.Cleanup(func() { dhcpNATSTimeout = prev }) - - err = ReleaseDHCPLease(nc, "eni-timeout") - assert.ErrorContains(t, err, "dhcp release NATS request") -} - -func TestObtainDHCPLease_MalformedReply(t *testing.T) { - _, nc := testutil.StartTestNATS(t) - sub, err := nc.Subscribe(dhcp.TopicAcquire, func(msg *nats.Msg) { - _ = msg.Respond([]byte("not json")) - }) - require.NoError(t, err) - t.Cleanup(func() { _ = sub.Unsubscribe() }) - - _, err = ObtainDHCPLease(nc, "br-wan", "eni-malformed", "eni-malformed", "mulga-spinifex", "wan") - assert.ErrorContains(t, err, "unmarshal dhcp acquire reply") -} - -func TestReleaseDHCPLease_MalformedReply(t *testing.T) { - _, nc := testutil.StartTestNATS(t) - sub, err := nc.Subscribe(dhcp.TopicRelease, func(msg *nats.Msg) { - _ = msg.Respond([]byte("not json")) - }) - require.NoError(t, err) - t.Cleanup(func() { _ = sub.Unsubscribe() }) - - err = ReleaseDHCPLease(nc, "eni-malformed") - assert.ErrorContains(t, err, "unmarshal dhcp release reply") -} - -func TestReleaseDHCPLease_ReplyError(t *testing.T) { - _, nc := testutil.StartTestNATS(t) - sub, err := nc.Subscribe(dhcp.TopicRelease, func(msg *nats.Msg) { - data, _ := json.Marshal(dhcp.ReleaseReplyMsg{Error: "vpcd rejected release"}) - _ = msg.Respond(data) - }) - require.NoError(t, err) - t.Cleanup(func() { _ = sub.Unsubscribe() }) - - err = ReleaseDHCPLease(nc, "eni-rejected") - assert.ErrorContains(t, err, "vpcd rejected release") -} - -func TestObtainDHCPLease_Guards(t *testing.T) { - _, err := ObtainDHCPLease(nil, "br-wan", "eni-1", "eni-1", "mulga-spinifex", "wan") - assert.ErrorContains(t, err, "NATS connection is required") - - _, nc := testutil.StartTestNATS(t) - _, err = ObtainDHCPLease(nc, "", "eni-1", "eni-1", "mulga-spinifex", "wan") - assert.ErrorContains(t, err, "bridge name is required") - - _, err = ObtainDHCPLease(nc, "br-wan", "", "eni-1", "mulga-spinifex", "wan") - assert.ErrorContains(t, err, "client ID is required") -} - -func TestReleaseDHCPLease_NoopWhenNilOrEmpty(t *testing.T) { - assert.NoError(t, ReleaseDHCPLease(nil, "eni-1")) - _, nc := testutil.StartTestNATS(t) - assert.NoError(t, ReleaseDHCPLease(nc, "")) -} - func TestExternalIPAM_DHCPGatewayErrorPropagates(t *testing.T) { _, nc, js := testutil.StartTestJetStream(t) stub := newDHCPStub(t, nc) diff --git a/spinifex/services/vpcd/dhcp/nats_client.go b/spinifex/services/vpcd/dhcp/nats_client.go new file mode 100644 index 000000000..e09ffe89f --- /dev/null +++ b/spinifex/services/vpcd/dhcp/nats_client.go @@ -0,0 +1,150 @@ +package dhcp + +import ( + "encoding/json" + "fmt" + "log/slog" + "time" + + "github.com/nats-io/nats.go" +) + +// NATSTimeout bounds the request/reply round-trip to vpcd's DHCPManager. +// Picked to comfortably exceed the DORA handshake (15 s default + server +// jitter + ~10 s of slack for NAK + fresh DISCOVER fallbacks). A var rather +// than a const so tests can drive the timeout error path without waiting +// 20 s per table entry. +var NATSTimeout = 30 * time.Second + +// AcquireMaxAttempts caps retry of vpc.dhcp.acquire. 60 attempts × 1 s sleep +// covers a cold-boot vpcd start window where NATS is up but the dhcp.Manager +// hasn't yet subscribed (typical: 5–30 s, pathological: 60 s during full +// reconcile). Tests override to keep timeout-path coverage fast. +var AcquireMaxAttempts = 60 + +// AcquireRetryDelay is the wait between retries. Var for the same reason as +// AcquireMaxAttempts. +var AcquireRetryDelay = 1 * time.Second + +// LeaseResult is the data callers need from a successful acquire. Mirrors +// AcquireReplyMsg but exposed as a typed result so callers don't depend on +// the wire struct. +type LeaseResult struct { + IP string + SubnetMask string + Routers []string + DNS []string + ServerID string + HWAddr string + ExpiresUnix int64 +} + +// RequestAcquire asks vpcd's DHCPManager (over NATS) to acquire a DHCP lease +// on the given bridge, identifying the caller with option 61 (client-id), +// option 12 (hostname) and option 60 (vendor class). Blocks until vpcd +// replies or NATSTimeout expires. The Manager-side handler is idempotent on +// clientID: a second call with the same clientID while a live lease exists +// returns the same lease without a fresh DORA, so CAS retry loops on the +// caller are safe. +func RequestAcquire(nc *nats.Conn, bridge, clientID, hostname, vendorClass, poolName string) (LeaseResult, error) { + if nc == nil { + return LeaseResult{}, fmt.Errorf("DHCP lease: NATS connection is required") + } + if bridge == "" { + return LeaseResult{}, fmt.Errorf("DHCP lease: bridge name is required") + } + if clientID == "" { + return LeaseResult{}, fmt.Errorf("DHCP lease: client ID is required") + } + + req := AcquireRequestMsg{ + Bridge: bridge, + ClientID: clientID, + Hostname: hostname, + VendorClass: vendorClass, + PoolName: poolName, + } + data, err := json.Marshal(req) + if err != nil { + return LeaseResult{}, fmt.Errorf("marshal dhcp acquire request: %w", err) + } + + var reply AcquireReplyMsg + maxAttempts := AcquireMaxAttempts + + for attempt := 1; attempt <= maxAttempts; attempt++ { + msg, err := nc.Request(TopicAcquire, data, NATSTimeout) + if err != nil { + if attempt == maxAttempts { + return LeaseResult{}, fmt.Errorf( + "dhcp acquire NATS request failed after %d attempts (client %s): %w", + maxAttempts, clientID, err, + ) + } + slog.Warn("dhcp acquire NATS request failed, retrying", + "clientID", clientID, + "attempt", attempt, + "maxAttempts", maxAttempts, + "err", err, + ) + time.Sleep(AcquireRetryDelay) + continue + } + + if err := json.Unmarshal(msg.Data, &reply); err != nil { + return LeaseResult{}, fmt.Errorf("unmarshal dhcp acquire reply: %w", err) + } + if reply.Error != "" { + return LeaseResult{}, fmt.Errorf("dhcp acquire client %s: %s", clientID, reply.Error) + } + break + } + + slog.Info("DHCP lease obtained", + "bridge", bridge, + "client_id", clientID, + "ip", reply.IP, + "server_id", reply.ServerID, + "expires_unix", reply.ExpiresUnix, + ) + + return LeaseResult{ + IP: reply.IP, + SubnetMask: reply.SubnetMask, + Routers: reply.Routers, + DNS: reply.DNS, + ServerID: reply.ServerID, + HWAddr: reply.HWAddr, + ExpiresUnix: reply.ExpiresUnix, + }, nil +} + +// RequestRelease asks vpcd to release the lease identified by clientID. +// Returns nil silently when nc or clientID is empty so callers in clean-up +// paths can invoke it unconditionally. +func RequestRelease(nc *nats.Conn, clientID string) error { + if nc == nil || clientID == "" { + return nil + } + + data, err := json.Marshal(ReleaseRequestMsg{ClientID: clientID}) + if err != nil { + return fmt.Errorf("marshal dhcp release request: %w", err) + } + + msg, err := nc.Request(TopicRelease, data, NATSTimeout) + if err != nil { + return fmt.Errorf("dhcp release NATS request (client %s): %w", clientID, err) + } + + var reply ReleaseReplyMsg + if err := json.Unmarshal(msg.Data, &reply); err != nil { + return fmt.Errorf("unmarshal dhcp release reply: %w", err) + } + if reply.Error != "" { + return fmt.Errorf("dhcp release (client %s): %s", clientID, reply.Error) + } + + slog.Info("DHCP lease released", "client_id", clientID) + return nil +} diff --git a/spinifex/services/vpcd/dhcp/nats_client_test.go b/spinifex/services/vpcd/dhcp/nats_client_test.go new file mode 100644 index 000000000..da4993ccb --- /dev/null +++ b/spinifex/services/vpcd/dhcp/nats_client_test.go @@ -0,0 +1,104 @@ +package dhcp + +import ( + "encoding/json" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestRequestAcquire_RequestTimeout(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + sub, err := nc.Subscribe(TopicAcquire, func(*nats.Msg) {}) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + prev := NATSTimeout + NATSTimeout = 150 * time.Millisecond + t.Cleanup(func() { NATSTimeout = prev }) + + prevMax := AcquireMaxAttempts + prevDelay := AcquireRetryDelay + AcquireMaxAttempts = 2 + AcquireRetryDelay = 10 * time.Millisecond + t.Cleanup(func() { + AcquireMaxAttempts = prevMax + AcquireRetryDelay = prevDelay + }) + + _, err = RequestAcquire(nc, "br-wan", "eni-timeout", "eni-timeout", "mulga-spinifex", "wan") + assert.ErrorContains(t, err, "dhcp acquire NATS request") +} + +func TestRequestRelease_RequestTimeout(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + sub, err := nc.Subscribe(TopicRelease, func(*nats.Msg) {}) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + prev := NATSTimeout + NATSTimeout = 150 * time.Millisecond + t.Cleanup(func() { NATSTimeout = prev }) + + err = RequestRelease(nc, "eni-timeout") + assert.ErrorContains(t, err, "dhcp release NATS request") +} + +func TestRequestAcquire_MalformedReply(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + sub, err := nc.Subscribe(TopicAcquire, func(msg *nats.Msg) { + _ = msg.Respond([]byte("not json")) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + _, err = RequestAcquire(nc, "br-wan", "eni-malformed", "eni-malformed", "mulga-spinifex", "wan") + assert.ErrorContains(t, err, "unmarshal dhcp acquire reply") +} + +func TestRequestRelease_MalformedReply(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + sub, err := nc.Subscribe(TopicRelease, func(msg *nats.Msg) { + _ = msg.Respond([]byte("not json")) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + err = RequestRelease(nc, "eni-malformed") + assert.ErrorContains(t, err, "unmarshal dhcp release reply") +} + +func TestRequestRelease_ReplyError(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + sub, err := nc.Subscribe(TopicRelease, func(msg *nats.Msg) { + data, _ := json.Marshal(ReleaseReplyMsg{Error: "vpcd rejected release"}) + _ = msg.Respond(data) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + err = RequestRelease(nc, "eni-rejected") + assert.ErrorContains(t, err, "vpcd rejected release") +} + +func TestRequestAcquire_Guards(t *testing.T) { + _, err := RequestAcquire(nil, "br-wan", "eni-1", "eni-1", "mulga-spinifex", "wan") + assert.ErrorContains(t, err, "NATS connection is required") + + _, nc := testutil.StartTestNATS(t) + _, err = RequestAcquire(nc, "", "eni-1", "eni-1", "mulga-spinifex", "wan") + assert.ErrorContains(t, err, "bridge name is required") + + _, err = RequestAcquire(nc, "br-wan", "", "eni-1", "mulga-spinifex", "wan") + assert.ErrorContains(t, err, "client ID is required") +} + +func TestRequestRelease_NoopWhenNilOrEmpty(t *testing.T) { + assert.NoError(t, RequestRelease(nil, "eni-1")) + _, nc := testutil.StartTestNATS(t) + assert.NoError(t, RequestRelease(nc, "")) +} diff --git a/spinifex/services/vpcd/topology.go b/spinifex/services/vpcd/topology.go index 85bc21dba..4afff94d2 100644 --- a/spinifex/services/vpcd/topology.go +++ b/spinifex/services/vpcd/topology.go @@ -9,6 +9,7 @@ import ( "strconv" "strings" + "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" "github.com/mulgadc/spinifex/spinifex/services/vpcd/nbdb" "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" @@ -117,6 +118,11 @@ type TopologyHandler struct { externalPools []ExternalPoolConfig chassisNames []string // OVN chassis names for gateway HA scheduling bridgeMode string // "direct" or "macvlan" — controls NAT mode and localnet options + // nc is used to talk to vpcd's DHCPManager via vpc.dhcp.acquire / + // vpc.dhcp.release for gateway LRP IPs in centralized NAT on + // source="dhcp" pools (mulga-siv-38). nil when no DHCP pool is wired or + // the test stack supplies a static-only mock. + nc *nats.Conn } // NewTopologyHandler creates a new TopologyHandler with optional external network config. @@ -155,6 +161,15 @@ func WithBridgeMode(mode string) TopologyOption { } } +// WithNATSConn wires the NATS connection used to talk to vpcd's DHCPManager +// for gateway-LRP DHCP-acquire (mulga-siv-38). Tests with mock OVN clients +// can omit this — the static / auto-derive paths still work. +func WithNATSConn(nc *nats.Conn) TopologyOption { + return func(h *TopologyHandler) { + h.nc = nc + } +} + // isMacvlanMode returns true if the external bridge uses a macvlan interface. // This is the default when bridgeMode is unset for backward compatibility. func (h *TopologyHandler) isMacvlanMode() bool { @@ -213,11 +228,12 @@ func (h *TopologyHandler) ensureLocalnetOptions(ctx context.Context, extPortName } // expectedGatewayPortNetwork returns the Networks CIDR a gateway LRP for -// vpcId should carry. In direct mode, or in centralized mode without a -// configured gw_lrp_range, it is the link-local fallback. In centralized -// mode with a range, it is the existing spinifex:gateway_ip on the LRP if -// any (idempotent), otherwise a fresh allocation. Returns the new gw IP -// (empty when link-local) so callers can persist it to external_ids. +// vpcId should carry. In direct mode, or in centralized mode without a pool, +// it is the link-local fallback. In centralized mode with a DHCP pool the +// gateway IP is acquired via vpc.dhcp.acquire (mulga-siv-38). With a static +// pool the IP comes from gw_lrp_range or the auto-derived top-of-subnet +// block (mulga-siv-36). Returns the gw IP (empty when link-local) so +// callers can persist it to external_ids. func (h *TopologyHandler) expectedGatewayPortNetwork(ctx context.Context, vpcId string) (network, gwIP string, err error) { if !h.useCentralizedNAT() { return gatewayPortNetwork, "", nil @@ -226,6 +242,13 @@ func (h *TopologyHandler) expectedGatewayPortNetwork(ctx context.Context, vpcId if pool == nil { return gatewayPortNetwork, "", nil } + if pool.IsDHCP() { + ip, prefix, dhcpErr := h.allocateGatewayLRPIPViaDHCP(ctx, vpcId, pool) + if dhcpErr != nil { + return "", "", dhcpErr + } + return fmt.Sprintf("%s/%d", ip, prefix), ip, nil + } ip, prefix, ok, allocErr := h.allocateGatewayLRPIP(ctx, vpcId, pool) if allocErr != nil { return "", "", allocErr @@ -887,22 +910,20 @@ func (h *TopologyHandler) handleIGWAttach(msg *nats.Msg) { case pool != nil: wanGateway = pool.Gateway if h.useCentralizedNAT() { - ip, prefix, allocOK, allocErr := h.allocateGatewayLRPIP(ctx, evt.VpcId, pool) - if allocErr != nil || !allocOK { + network, ip, allocErr := h.expectedGatewayPortNetwork(ctx, evt.VpcId) + if allocErr != nil { slog.Error("vpcd: failed to allocate gw LRP IP for centralized NAT", "vpc_id", evt.VpcId, "pool", pool.Name, "err", allocErr) _ = h.ovn.DeleteLogicalSwitch(ctx, extSwitchName) - if allocErr == nil { - allocErr = fmt.Errorf("no gw LRP IP available in pool %q (configure gw_lrp_range or fix Gateway/PrefixLen)", pool.Name) - } respond(msg, allocErr) return } gwLrpIP = ip - gatewayNetwork = fmt.Sprintf("%s/%d", ip, prefix) + gatewayNetwork = network } slog.Info("vpcd: using external pool for IGW", "pool", pool.Name, + "source", pool.Source, "lrp_network", gatewayNetwork, "wan_gateway", wanGateway, ) @@ -1072,6 +1093,10 @@ func (h *TopologyHandler) handleIGWDetach(msg *nats.Msg) { return } + // 7. Release any DHCP-acquired gateway LRP lease (mulga-siv-38). Best- + // effort — upstream server expires the lease on its own if this fails. + h.releaseGatewayLRPLease(evt.VpcId) + slog.Info("vpcd: detached internet gateway from VPC", "igw_id", evt.InternetGatewayId, "vpc_id", evt.VpcId, @@ -1293,15 +1318,12 @@ func (h *TopologyHandler) reconcileIGW(ctx context.Context, vpcId, igwId string) if pool != nil { wanGateway = pool.Gateway if h.useCentralizedNAT() { - ip, prefix, allocOK, allocErr := h.allocateGatewayLRPIP(ctx, vpcId, pool) - if allocErr != nil || !allocOK { - if allocErr == nil { - allocErr = fmt.Errorf("no gw LRP IP available in pool %q", pool.Name) - } + network, ip, allocErr := h.expectedGatewayPortNetwork(ctx, vpcId) + if allocErr != nil { return fmt.Errorf("allocate gw LRP IP for %s: %w", vpcId, allocErr) } gwLrpIP = ip - gatewayNetwork = fmt.Sprintf("%s/%d", ip, prefix) + gatewayNetwork = network } } @@ -1593,6 +1615,69 @@ func gwLrpRange(pool *ExternalPoolConfig) (start, end net.IP, prefix int, ok boo return uint32ToIPv4(autoStartU), uint32ToIPv4(autoEndU), prefix, true } +// gwLrpClientID is the DHCP client-id used for a VPC's gateway LRP lease. +// Stable across reboots — DHCPManager reuses the same lease on idempotent +// re-attach so the LRP IP doesn't change unless the upstream server +// reassigns it. +func gwLrpClientID(vpcId string) string { return "gw-lrp-" + vpcId } + +// allocateGatewayLRPIPViaDHCP requests a DHCP lease from vpcd's DHCPManager +// for the gateway LRP of vpcId. The handler-side acquire is idempotent on +// client-id so retries return the same IP without a fresh DORA. Prefix +// comes from the lease's SubnetMask, falling back to pool.PrefixLen when +// the server omits option 1 (rare; should not happen in practice). +func (h *TopologyHandler) allocateGatewayLRPIPViaDHCP(ctx context.Context, vpcId string, pool *ExternalPoolConfig) (ip string, prefix int, err error) { + _ = ctx + if h.nc == nil { + return "", 0, fmt.Errorf("vpcd: no NATS conn for DHCP gw LRP allocation (vpc %s, pool %s)", vpcId, pool.Name) + } + clientID := gwLrpClientID(vpcId) + hostname := "spinifex-gw-" + vpcId + vendorClass := "mulga-spinifex-gw-lrp" + lease, dhcpErr := dhcp.RequestAcquire(h.nc, pool.DhcpBindBridge, clientID, hostname, vendorClass, pool.Name) + if dhcpErr != nil { + return "", 0, fmt.Errorf("dhcp acquire gw LRP IP for vpc %s: %w", vpcId, dhcpErr) + } + prefix = prefixFromMask(lease.SubnetMask) + if prefix == 0 { + prefix = pool.PrefixLen + } + if prefix <= 0 || prefix > 32 { + return "", 0, fmt.Errorf("dhcp gw LRP for vpc %s: cannot determine prefix (mask=%q pool prefix=%d)", vpcId, lease.SubnetMask, pool.PrefixLen) + } + return lease.IP, prefix, nil +} + +// releaseGatewayLRPLease asks vpcd's DHCPManager to drop the lease held for +// vpcId's gateway LRP. Best-effort — log on failure but never block IGW +// detach, since the upstream server will eventually expire the lease on +// its own. +func (h *TopologyHandler) releaseGatewayLRPLease(vpcId string) { + if h.nc == nil { + return + } + if err := dhcp.RequestRelease(h.nc, gwLrpClientID(vpcId)); err != nil { + slog.Warn("vpcd: dhcp release for gw LRP failed", "vpc_id", vpcId, "err", err) + } +} + +// prefixFromMask converts a dotted-decimal mask ("255.255.255.0") to a +// prefix length. Returns 0 when the mask is empty or unparseable. +func prefixFromMask(mask string) int { + if mask == "" { + return 0 + } + ip := net.ParseIP(mask).To4() + if ip == nil { + return 0 + } + ones, bits := net.IPMask(ip).Size() + if bits != 32 { + return 0 + } + return ones +} + // allocateGatewayLRPIP picks the next free IP in pool.GwLrpRange for the // given VPC, scanning every existing LRP's spinifex:gateway_ip external_id // to compute the used set. If gw- already has an allocation the diff --git a/spinifex/services/vpcd/topology_test.go b/spinifex/services/vpcd/topology_test.go index b20f96dab..0627fb5f9 100644 --- a/spinifex/services/vpcd/topology_test.go +++ b/spinifex/services/vpcd/topology_test.go @@ -8,6 +8,7 @@ import ( "strings" "testing" + "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" "github.com/mulgadc/spinifex/spinifex/services/vpcd/nbdb" "github.com/mulgadc/spinifex/spinifex/testutil" "github.com/mulgadc/spinifex/spinifex/types" @@ -1221,6 +1222,265 @@ func TestTopologyHandler_IGWAttach_CentralizedNATAllocatesGwLrpIP(t *testing.T) } } +// dhcpStub is a minimal in-test responder for vpc.dhcp.acquire / release. +// Each acquire returns a distinct lease IP keyed off ClientID so multi-VPC +// tests get unique IPs without standing up a full DHCPManager. +type dhcpStub struct { + t *testing.T + leases map[string]string // client_id -> ip + released map[string]bool + nextOct int +} + +func newDHCPStub(t *testing.T, nc *nats.Conn) *dhcpStub { + t.Helper() + s := &dhcpStub{ + t: t, + leases: map[string]string{}, + released: map[string]bool{}, + nextOct: 101, + } + subA, err := nc.Subscribe(dhcp.TopicAcquire, func(msg *nats.Msg) { + var req dhcp.AcquireRequestMsg + if err := json.Unmarshal(msg.Data, &req); err != nil { + data, _ := json.Marshal(dhcp.AcquireReplyMsg{Error: err.Error()}) + _ = msg.Respond(data) + return + } + ip, ok := s.leases[req.ClientID] + if !ok { + ip = fmt.Sprintf("192.168.3.%d", s.nextOct) + s.nextOct++ + s.leases[req.ClientID] = ip + } + reply := dhcp.AcquireReplyMsg{ + IP: ip, + SubnetMask: "255.255.255.0", + ServerID: "192.168.3.1", + Routers: []string{"192.168.3.1"}, + HWAddr: "02:00:00:00:00:" + fmt.Sprintf("%02x", s.nextOct%256), + ExpiresUnix: 0, + } + data, _ := json.Marshal(reply) + _ = msg.Respond(data) + }) + if err != nil { + t.Fatalf("subscribe acquire: %v", err) + } + subR, err := nc.Subscribe(dhcp.TopicRelease, func(msg *nats.Msg) { + var req dhcp.ReleaseRequestMsg + _ = json.Unmarshal(msg.Data, &req) + s.released[req.ClientID] = true + delete(s.leases, req.ClientID) + data, _ := json.Marshal(dhcp.ReleaseReplyMsg{}) + _ = msg.Respond(data) + }) + if err != nil { + t.Fatalf("subscribe release: %v", err) + } + t.Cleanup(func() { + _ = subA.Unsubscribe() + _ = subR.Unsubscribe() + }) + return s +} + +// TestTopologyHandler_IGWAttach_DHCPGwLrpIP guards mulga-siv-38: with +// pool.Source="dhcp" the gateway LRP IP must come from a DHCP lease +// (vpc.dhcp.acquire), not the static gw_lrp_range allocator. Three VPCs +// sharing one DHCP pool must each get a distinct lease IP and persist it +// on external_ids:spinifex:gateway_ip. IGW detach must release the lease. +func TestTopologyHandler_IGWAttach_DHCPGwLrpIP(t *testing.T) { + _, nc := startTestNATS(t) + stub := newDHCPStub(t, nc) + mock := NewMockOVNClient() + _ = mock.Connect(context.Background()) + ctx := context.Background() + + pools := []ExternalPoolConfig{ + { + Name: "lan-dhcp", + Source: "dhcp", + Gateway: "192.168.3.1", + PrefixLen: 24, + DhcpBindBridge: "br-wan", + }, + } + topo := NewTopologyHandler(mock, + WithExternalNetwork("pool", pools), + WithBridgeMode(BridgeModeVeth), + WithNATSConn(nc), + ) + subs, err := topo.Subscribe(nc) + if err != nil { + t.Fatalf("subscribe: %v", err) + } + defer func() { + for _, s := range subs { + _ = s.Unsubscribe() + } + }() + + vpcIds := []string{"vpc-dh-a", "vpc-dh-b", "vpc-dh-c"} + for _, vpcId := range vpcIds { + _ = mock.CreateLogicalRouter(ctx, &nbdb.LogicalRouter{ + Name: "vpc-" + vpcId, + ExternalIDs: map[string]string{ + "spinifex:vpc_id": vpcId, + "spinifex:cidr": "10.0.0.0/16", + }, + }) + evt := types.IGWEvent{InternetGatewayId: "igw-" + vpcId, VpcId: vpcId} + data, _ := json.Marshal(evt) + resp, err := nc.Request(TopicIGWAttach, data, 5_000_000_000) + if err != nil { + t.Fatalf("request vpc.igw-attach %s: %v", vpcId, err) + } + assertSuccess(t, resp, "attach IGW "+vpcId) + } + + seen := make(map[string]string, len(vpcIds)) + for _, vpcId := range vpcIds { + gwPortName := "gw-" + vpcId + gwPort, err := mock.GetLogicalRouterPort(ctx, gwPortName) + if err != nil { + t.Fatalf("get %s: %v", gwPortName, err) + } + if len(gwPort.Networks) != 1 { + t.Fatalf("%s: expected 1 Network, got %v", gwPortName, gwPort.Networks) + } + ip, cidr, _ := strings.Cut(gwPort.Networks[0], "/") + if cidr != "24" { + t.Errorf("%s: expected /24 from DHCP lease mask, got /%s", gwPortName, cidr) + } + if !strings.HasPrefix(ip, "192.168.3.") { + t.Errorf("%s: expected 192.168.3.X from stub, got %s", gwPortName, ip) + } + gwIP := gwPort.ExternalIDs["spinifex:gateway_ip"] + if gwIP != ip { + t.Errorf("%s: external_ids gateway_ip=%q, expected %q", gwPortName, gwIP, ip) + } + if other, dup := seen[gwIP]; dup { + t.Errorf("gw IP collision: %s and %s share %s", gwPortName, other, gwIP) + } + seen[gwIP] = gwPortName + + // stub recorded the lease keyed by gw-lrp- + clientID := "gw-lrp-" + vpcId + if stub.leases[clientID] != ip { + t.Errorf("%s: stub lease %q != lrp IP %q", clientID, stub.leases[clientID], ip) + } + } + + // Detach one VPC, assert release reaches the stub. + detachVpc := vpcIds[0] + devt := types.IGWEvent{InternetGatewayId: "igw-" + detachVpc, VpcId: detachVpc} + ddata, _ := json.Marshal(devt) + resp, err := nc.Request(TopicIGWDetach, ddata, 5_000_000_000) + if err != nil { + t.Fatalf("request vpc.igw-detach: %v", err) + } + assertSuccess(t, resp, "detach IGW "+detachVpc) + if !stub.released["gw-lrp-"+detachVpc] { + t.Errorf("expected DHCP release for gw-lrp-%s, stub released=%v", detachVpc, stub.released) + } +} + +// TestTopologyHandler_IGWAttach_DHCPAcquireFailureAborts ensures that when +// vpcd's DHCPManager replies with an error, IGW attach surfaces the failure +// loud instead of falling back to link-local. Otherwise we silently restore +// the broken pre-siv-36 behavior. +func TestTopologyHandler_IGWAttach_DHCPAcquireFailureAborts(t *testing.T) { + _, nc := startTestNATS(t) + sub, err := nc.Subscribe(dhcp.TopicAcquire, func(msg *nats.Msg) { + reply := dhcp.AcquireReplyMsg{Error: "no leases available"} + data, _ := json.Marshal(reply) + _ = msg.Respond(data) + }) + if err != nil { + t.Fatalf("subscribe: %v", err) + } + defer func() { _ = sub.Unsubscribe() }() + + mock := NewMockOVNClient() + _ = mock.Connect(context.Background()) + ctx := context.Background() + + pools := []ExternalPoolConfig{ + { + Name: "lan-dhcp", + Source: "dhcp", + Gateway: "192.168.3.1", + PrefixLen: 24, + DhcpBindBridge: "br-wan", + }, + } + topo := NewTopologyHandler(mock, + WithExternalNetwork("pool", pools), + WithBridgeMode(BridgeModeVeth), + WithNATSConn(nc), + ) + subs, err := topo.Subscribe(nc) + if err != nil { + t.Fatalf("subscribe: %v", err) + } + defer func() { + for _, s := range subs { + _ = s.Unsubscribe() + } + }() + + _ = mock.CreateLogicalRouter(ctx, &nbdb.LogicalRouter{ + Name: "vpc-vpc-dh-fail", + ExternalIDs: map[string]string{ + "spinifex:vpc_id": "vpc-dh-fail", + "spinifex:cidr": "10.0.0.0/16", + }, + }) + evt := types.IGWEvent{InternetGatewayId: "igw-fail", VpcId: "vpc-dh-fail"} + data, _ := json.Marshal(evt) + resp, _ := nc.Request(TopicIGWAttach, data, 5_000_000_000) + var reply struct { + Error string `json:"error"` + } + _ = json.Unmarshal(resp.Data, &reply) + if reply.Error == "" { + t.Fatal("expected attach to fail when DHCP acquire errors, got success") + } + if !strings.Contains(reply.Error, "dhcp acquire") && !strings.Contains(reply.Error, "no leases available") { + t.Errorf("expected dhcp-related error, got %q", reply.Error) + } + // External switch must be rolled back. + if _, err := mock.GetLogicalSwitch(ctx, "ext-vpc-dh-fail"); err == nil { + t.Error("expected external switch rolled back after DHCP failure") + } +} + +// TestPrefixFromMask checks dotted-mask → prefix conversion across the +// boundary cases gw LRP DHCP relies on (option 1 missing, /32, /0, +// non-contiguous masks, IPv6). +func TestPrefixFromMask(t *testing.T) { + cases := []struct { + mask string + want int + }{ + {"", 0}, + {"255.255.255.0", 24}, + {"255.255.255.255", 32}, + {"0.0.0.0", 0}, + {"255.255.0.0", 16}, + {"not-an-ip", 0}, + } + for _, tc := range cases { + t.Run(tc.mask, func(t *testing.T) { + got := prefixFromMask(tc.mask) + if got != tc.want { + t.Errorf("prefixFromMask(%q)=%d, want %d", tc.mask, got, tc.want) + } + }) + } +} + func TestTopologyHandler_FindExternalPool(t *testing.T) { pools := []ExternalPoolConfig{ {Name: "az-a", Region: "us-east-1", AZ: "us-east-1a", RangeStart: "1.1.1.1"}, diff --git a/spinifex/services/vpcd/vpcd.go b/spinifex/services/vpcd/vpcd.go index dcdf11b92..f1266a4fe 100644 --- a/spinifex/services/vpcd/vpcd.go +++ b/spinifex/services/vpcd/vpcd.go @@ -81,6 +81,7 @@ type Config struct { // ExternalPoolConfig mirrors config.ExternalPool for vpcd's internal use. type ExternalPoolConfig struct { Name string + Source string // "static" (default) or "dhcp" RangeStart string RangeEnd string Gateway string @@ -89,10 +90,16 @@ type ExternalPoolConfig struct { DNSServers []string Region string AZ string + DhcpBindBridge string // Bridge where the DHCP AF_PACKET socket binds (e.g. "br-wan"). Required for source="dhcp". GwLrpRangeStart string // Sub-range for OVN gateway LRP IPs in centralized NAT mode (mulga-siv-36). GwLrpRangeEnd string } +// IsDHCP returns true if this pool obtains IPs from upstream DHCP. +func (p *ExternalPoolConfig) IsDHCP() bool { + return p.Source == "dhcp" +} + // Service implements the Spinifex service interface for vpcd. type Service struct { Config *Config @@ -344,6 +351,7 @@ func launchService(cfg *Config) error { topoOpts = append(topoOpts, WithChassisNames(chassisNames)) slog.Info("vpcd: gateway chassis discovered", "chassis", chassisNames) topoOpts = append(topoOpts, WithBridgeMode(bridgeMode)) + topoOpts = append(topoOpts, WithNATSConn(nc)) topo := NewTopologyHandler(liveClient, topoOpts...) // Elect a single vpcd to run startup reconcile. Without this, N vpcds in a