mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-20 03:11:46 +00:00
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.
This commit is contained in:
@@ -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,
|
||||
})
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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, ""))
|
||||
}
|
||||
@@ -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-<vpcId> already has an allocation the
|
||||
|
||||
@@ -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-<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) {
|
||||
pools := []ExternalPoolConfig{
|
||||
{Name: "az-a", Region: "us-east-1", AZ: "us-east-1a", RangeStart: "1.1.1.1"},
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user