From 216ca9e710b533e68f1ef38943c202add85e3e4d Mon Sep 17 00:00:00 2001 From: Ben Duncan <5178243+benduncan@users.noreply.github.com> Date: Tue, 24 Mar 2026 15:03:04 +1000 Subject: [PATCH] =?UTF-8?q?feat:=20Phase=2012b=20+=20Phase=2016=20?= =?UTF-8?q?=E2=80=94=20KV=20reconcile=20on=20startup=20+=20NAT=20mode=20su?= =?UTF-8?q?pport?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 12b: vpcd now reconciles OVN topology from NATS KV on startup (Pass 2). After subscribing to topics, it reads all VPCs, subnets, IGW attachments, and ENIs from KV and ensures matching OVN objects exist. Handles reboots, OVN DB loss, and missed events. All operations are idempotent. Phase 16 prep: Add --gateway-ip flag to admin init for NAT mode (specifies the OVN gateway router's external IP without DHCP). Update reset-dev-env.sh to support EXTERNAL_MODE=nat env var for NAT/DHCP mode testing alongside the default pool mode. --- cmd/spinifex/cmd/admin.go | 18 +- scripts/reset-dev-env.sh | 28 +++- spinifex/services/vpcd/reconcile.go | 205 ++++++++++++++++++++++- spinifex/services/vpcd/reconcile_test.go | 189 +++++++++++++++++++++ spinifex/services/vpcd/vpcd.go | 4 + 5 files changed, 431 insertions(+), 13 deletions(-) diff --git a/cmd/spinifex/cmd/admin.go b/cmd/spinifex/cmd/admin.go index e342fcc8a..7128d78d0 100644 --- a/cmd/spinifex/cmd/admin.go +++ b/cmd/spinifex/cmd/admin.go @@ -221,6 +221,7 @@ func init() { adminInitCmd.Flags().String("external-iface", "", "WAN NIC for br-external (auto-detected from default route)") adminInitCmd.Flags().String("external-pool", "", "External IP pool range as start-end (e.g., 192.168.1.150-192.168.1.250)") adminInitCmd.Flags().String("external-gateway", "", "WAN gateway IP (auto-detected from default route)") + adminInitCmd.Flags().String("gateway-ip", "", "OVN gateway router's external IP for SNAT (default: pool range_start for pool mode, required for nat mode without DHCP)") adminInitCmd.Flags().Int("external-prefix-len", 24, "External pool subnet prefix length (auto-detected)") adminInitCmd.Flags().Bool("no-external", false, "Disable external networking (overlay-only, no internet for VMs)") @@ -543,6 +544,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { externalPool, _ := cmd.Flags().GetString("external-pool") externalGateway, _ := cmd.Flags().GetString("external-gateway") externalPrefixLen, _ := cmd.Flags().GetInt("external-prefix-len") + gatewayIP, _ := cmd.Flags().GetString("gateway-ip") noExternal, _ := cmd.Flags().GetBool("no-external") // Auto-detect network topology @@ -625,6 +627,10 @@ func runAdminInit(cmd *cobra.Command, args []string) { fmt.Fprintf(os.Stderr, "āŒ Error: --external-gateway is not a valid IP: %s\n", externalGateway) os.Exit(1) } + if gatewayIP != "" && net.ParseIP(gatewayIP) == nil { + fmt.Fprintf(os.Stderr, "āŒ Error: --gateway-ip is not a valid IP: %s\n", gatewayIP) + os.Exit(1) + } // Detect DNS servers from the host for VM DHCP var dnsServers []string @@ -635,9 +641,9 @@ func runAdminInit(cmd *cobra.Command, args []string) { } } - // For nat mode, we need the gateway IP. The DHCP flag tells setup-ovn.sh - // to obtain one via DHCP on the macvlan/bridge interface. - useExternalDHCP := externalMode == "nat" && poolStart == "" + // For nat mode, we need the gateway IP. Either provided via --gateway-ip + // or obtained via DHCP on the macvlan/bridge interface. + useExternalDHCP := externalMode == "nat" && gatewayIP == "" && poolStart == "" // Validate IP address format if net.ParseIP(bindIP) == nil { @@ -844,6 +850,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { PoolStart: poolStart, PoolEnd: poolEnd, PoolGateway: externalGateway, + PoolGatewayIP: gatewayIP, PoolPrefixLen: externalPrefixLen, PoolDNSServers: dnsServers, @@ -858,10 +865,13 @@ func runAdminInit(cmd *cobra.Command, args []string) { // Print external networking summary if externalMode != "" { fmt.Printf("\nšŸ“” External networking: %s\n", externalMode) - fmt.Printf(" WAN interface: %s (macvlan)\n", externalIface) + fmt.Printf(" WAN interface: %s\n", externalIface) if externalMode == "pool" { fmt.Printf(" IP pool: %s - %s\n", poolStart, poolEnd) fmt.Printf(" āš ļø Ensure %s-%s is excluded from your router's DHCP range.\n", poolStart, poolEnd) + } else if gatewayIP != "" { + fmt.Printf(" Gateway IP: %s (static)\n", gatewayIP) + fmt.Println(" VMs get: outbound internet via SNAT (no per-VM public IPs)") } else if useExternalDHCP { fmt.Println(" Gateway IP: DHCP (obtained during OVN setup)") fmt.Println(" VMs get: outbound internet via SNAT") diff --git a/scripts/reset-dev-env.sh b/scripts/reset-dev-env.sh index 8443734bf..7dfff0bfc 100755 --- a/scripts/reset-dev-env.sh +++ b/scripts/reset-dev-env.sh @@ -159,19 +159,31 @@ fi echo "Initializing platform" ADMIN_INIT_ARGS="--region $REGION --az ${REGION}a --node node1 --nodes 1" + +# External networking mode: set EXTERNAL_MODE=nat for NAT/DHCP mode (outbound-only), +# or leave unset / EXTERNAL_MODE=pool for pool mode (per-VM public IPs). +EXTERNAL_MODE="${EXTERNAL_MODE:-pool}" + if [ -n "$WAN_IFACE" ] && [ -n "$WAN_GW" ]; then - # Suggest pool range at high end of WAN subnet to avoid DHCP conflicts WAN_IP=$(ip -4 -o addr show "$WAN_IFACE" 2>/dev/null | awk '{print $4}' | cut -d/ -f1 | head -1) WAN_PREFIX=$(ip -4 -o addr show "$WAN_IFACE" 2>/dev/null | awk '{print $4}' | cut -d/ -f2 | head -1) if [ -z "$WAN_PREFIX" ]; then WAN_PREFIX=24; fi - # Use high range of subnet: .200-.250 for /24, avoid common DHCP ranges - IFS='.' read -r o1 o2 o3 o4 <<< "$WAN_GW" - POOL_START="${o1}.${o2}.${o3}.200" - POOL_END="${o1}.${o2}.${o3}.250" - - echo " External pool: $POOL_START - $POOL_END (gateway: $WAN_GW)" - ADMIN_INIT_ARGS="$ADMIN_INIT_ARGS --external-mode=pool --external-pool=${POOL_START}-${POOL_END} --external-gateway=${WAN_GW} --external-prefix-len=${WAN_PREFIX}" + if [ "$EXTERNAL_MODE" = "nat" ]; then + # NAT mode: outbound-only via shared SNAT. Use a single IP from the + # high end of the WAN subnet as the gateway IP. No per-VM public IPs. + IFS='.' read -r o1 o2 o3 o4 <<< "$WAN_GW" + GATEWAY_IP="${GATEWAY_IP:-${o1}.${o2}.${o3}.200}" + echo " External mode: nat (gateway IP: $GATEWAY_IP, WAN gateway: $WAN_GW)" + ADMIN_INIT_ARGS="$ADMIN_INIT_ARGS --external-mode=nat --gateway-ip=${GATEWAY_IP} --external-gateway=${WAN_GW} --external-prefix-len=${WAN_PREFIX}" + else + # Pool mode: per-VM public IPs from a static range + IFS='.' read -r o1 o2 o3 o4 <<< "$WAN_GW" + POOL_START="${o1}.${o2}.${o3}.200" + POOL_END="${o1}.${o2}.${o3}.250" + echo " External pool: $POOL_START - $POOL_END (gateway: $WAN_GW)" + ADMIN_INIT_ARGS="$ADMIN_INIT_ARGS --external-mode=pool --external-pool=${POOL_START}-${POOL_END} --external-gateway=${WAN_GW} --external-prefix-len=${WAN_PREFIX}" + fi fi ./bin/spx admin init $ADMIN_INIT_ARGS diff --git a/spinifex/services/vpcd/reconcile.go b/spinifex/services/vpcd/reconcile.go index 3cc20252f..2342b03e0 100644 --- a/spinifex/services/vpcd/reconcile.go +++ b/spinifex/services/vpcd/reconcile.go @@ -2,14 +2,23 @@ package vpcd import ( "context" + "encoding/json" + "errors" "log/slog" + + handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw" + handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" + "github.com/mulgadc/spinifex/spinifex/services/vpcd/nbdb" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" ) // ReconcileResult tracks what was created during reconciliation. type ReconcileResult struct { - RoutersCreated int + RoutersCreated int SwitchesCreated int IGWsCreated int + PortsCreated int } // Reconcile ensures OVN topology matches the expected state from the bootstrap config. @@ -81,3 +90,197 @@ func Reconcile(ctx context.Context, topo *TopologyHandler, bootstrap *BootstrapV return result } + +// ReconcileFromKV reads all VPCs, subnets, IGW attachments, and ENIs from NATS KV +// and ensures OVN topology matches. This is Pass 2 from the Phase 12 plan ā€” it +// handles reboots, OVN DB loss, and any missed events. All operations are idempotent. +func ReconcileFromKV(ctx context.Context, nc *nats.Conn, topo *TopologyHandler) ReconcileResult { + var result ReconcileResult + + js, err := nc.JetStream() + if err != nil { + slog.Warn("vpcd reconcile-kv: failed to get JetStream context", "err", err) + return result + } + + // 1. Reconcile VPCs: ensure each VPC has a logical router + vpcKV, err := js.KeyValue(handlers_ec2_vpc.KVBucketVPCs) + if err != nil { + slog.Debug("vpcd reconcile-kv: VPC KV bucket not available (first boot before daemon?)", "err", err) + return result + } + + // Build a map of VPC ID ā†’ CIDR for subnet/IGW reconciliation + type vpcInfo struct { + VpcId string + CidrBlock string + } + vpcMap := make(map[string]vpcInfo) + + vpcKeys, err := vpcKV.Keys() + if err != nil && !errors.Is(err, nats.ErrNoKeysFound) { + slog.Warn("vpcd reconcile-kv: failed to list VPC keys", "err", err) + } + for _, key := range vpcKeys { + if key == utils.VersionKey { + continue + } + entry, err := vpcKV.Get(key) + if err != nil { + continue + } + var rec handlers_ec2_vpc.VPCRecord + if err := json.Unmarshal(entry.Value(), &rec); err != nil { + slog.Warn("vpcd reconcile-kv: failed to unmarshal VPC record", "key", key, "err", err) + continue + } + vpcMap[rec.VpcId] = vpcInfo{VpcId: rec.VpcId, CidrBlock: rec.CidrBlock} + + routerName := "vpc-" + rec.VpcId + if _, err := topo.ovn.GetLogicalRouter(ctx, routerName); err != nil { + slog.Info("vpcd reconcile-kv: creating VPC router", "router", routerName) + if err := topo.reconcileVPC(ctx, rec.VpcId, rec.CidrBlock); err != nil { + slog.Error("vpcd reconcile-kv: failed to create VPC router", "err", err) + } else { + result.RoutersCreated++ + } + } + } + + // 2. Reconcile subnets: ensure each subnet has a logical switch + router port + DHCP + subnetKV, err := js.KeyValue(handlers_ec2_vpc.KVBucketSubnets) + if err != nil { + slog.Debug("vpcd reconcile-kv: subnet KV bucket not available", "err", err) + } else { + subnetKeys, err := subnetKV.Keys() + if err != nil && !errors.Is(err, nats.ErrNoKeysFound) { + slog.Warn("vpcd reconcile-kv: failed to list subnet keys", "err", err) + } + for _, key := range subnetKeys { + if key == utils.VersionKey { + continue + } + entry, err := subnetKV.Get(key) + if err != nil { + continue + } + var rec handlers_ec2_vpc.SubnetRecord + if err := json.Unmarshal(entry.Value(), &rec); err != nil { + slog.Warn("vpcd reconcile-kv: failed to unmarshal subnet record", "key", key, "err", err) + continue + } + + switchName := "subnet-" + rec.SubnetId + if _, err := topo.ovn.GetLogicalSwitch(ctx, switchName); err != nil { + slog.Info("vpcd reconcile-kv: creating subnet topology", "switch", switchName) + if err := topo.reconcileSubnet(ctx, rec.SubnetId, rec.VpcId, rec.CidrBlock); err != nil { + slog.Error("vpcd reconcile-kv: failed to create subnet topology", "err", err) + } else { + result.SwitchesCreated++ + } + } + } + } + + // 3. Reconcile IGW attachments: ensure attached IGWs have external switch + SNAT + igwKV, err := js.KeyValue(handlers_ec2_igw.KVBucketIGW) + if err != nil { + slog.Debug("vpcd reconcile-kv: IGW KV bucket not available", "err", err) + } else { + igwKeys, err := igwKV.Keys() + if err != nil && !errors.Is(err, nats.ErrNoKeysFound) { + slog.Warn("vpcd reconcile-kv: failed to list IGW keys", "err", err) + } + for _, key := range igwKeys { + if key == utils.VersionKey { + continue + } + entry, err := igwKV.Get(key) + if err != nil { + continue + } + var rec handlers_ec2_igw.IGWRecord + if err := json.Unmarshal(entry.Value(), &rec); err != nil { + slog.Warn("vpcd reconcile-kv: failed to unmarshal IGW record", "key", key, "err", err) + continue + } + // Only reconcile attached IGWs + if rec.VpcId == "" || rec.State != "attached" { + continue + } + + extSwitchName := "ext-" + rec.VpcId + if _, err := topo.ovn.GetLogicalSwitch(ctx, extSwitchName); err != nil { + slog.Info("vpcd reconcile-kv: creating IGW topology", "switch", extSwitchName, "igw_id", rec.InternetGatewayId) + if err := topo.reconcileIGW(ctx, rec.VpcId, rec.InternetGatewayId); err != nil { + slog.Error("vpcd reconcile-kv: failed to create IGW topology", "err", err) + } else { + result.IGWsCreated++ + } + } + } + } + + // 4. Reconcile ENI ports: ensure each ENI has a logical switch port + eniKV, err := js.KeyValue(handlers_ec2_vpc.KVBucketENIs) + if err != nil { + slog.Debug("vpcd reconcile-kv: ENI KV bucket not available", "err", err) + } else { + eniKeys, err := eniKV.Keys() + if err != nil && !errors.Is(err, nats.ErrNoKeysFound) { + slog.Warn("vpcd reconcile-kv: failed to list ENI keys", "err", err) + } + for _, key := range eniKeys { + if key == utils.VersionKey { + continue + } + entry, err := eniKV.Get(key) + if err != nil { + continue + } + var rec handlers_ec2_vpc.ENIRecord + if err := json.Unmarshal(entry.Value(), &rec); err != nil { + slog.Warn("vpcd reconcile-kv: failed to unmarshal ENI record", "key", key, "err", err) + continue + } + + portName := "port-" + rec.NetworkInterfaceId + if _, err := topo.ovn.GetLogicalSwitchPort(ctx, portName); err != nil { + switchName := "subnet-" + rec.SubnetId + addrStr := rec.MacAddress + " " + rec.PrivateIpAddress + lsp := &nbdb.LogicalSwitchPort{ + Name: portName, + Addresses: []string{addrStr}, + PortSecurity: []string{addrStr}, + ExternalIDs: map[string]string{ + "spinifex:eni_id": rec.NetworkInterfaceId, + "spinifex:subnet_id": rec.SubnetId, + "spinifex:vpc_id": rec.VpcId, + }, + } + + // Attach DHCP options if available + dhcpOpts, dhcpErr := topo.ovn.FindDHCPOptionsByExternalID(ctx, "spinifex:subnet_id", rec.SubnetId) + if dhcpErr == nil { + lsp.DHCPv4Options = &dhcpOpts.UUID + } + + slog.Info("vpcd reconcile-kv: creating ENI port", "port", portName, "switch", switchName) + if err := topo.ovn.CreateLogicalSwitchPort(ctx, switchName, lsp); err != nil { + slog.Error("vpcd reconcile-kv: failed to create ENI port", "port", portName, "err", err) + } else { + result.PortsCreated++ + } + } + } + } + + slog.Info("vpcd reconcile-kv: complete", + "routers_created", result.RoutersCreated, + "switches_created", result.SwitchesCreated, + "igws_created", result.IGWsCreated, + "ports_created", result.PortsCreated, + ) + + return result +} diff --git a/spinifex/services/vpcd/reconcile_test.go b/spinifex/services/vpcd/reconcile_test.go index 1b0d6d6e0..2dd9fa71d 100644 --- a/spinifex/services/vpcd/reconcile_test.go +++ b/spinifex/services/vpcd/reconcile_test.go @@ -2,8 +2,14 @@ package vpcd import ( "context" + "encoding/json" "testing" + "time" + handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw" + handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -147,3 +153,186 @@ func TestReconcile_PartialState(t *testing.T) { assert.Equal(t, 1, result.SwitchesCreated) assert.Equal(t, 1, result.IGWsCreated) } + +// --- ReconcileFromKV tests --- + +// seedKVBuckets creates NATS KV buckets and populates them with test data. +func seedKVBuckets(t *testing.T, nc *nats.Conn, vpcs []handlers_ec2_vpc.VPCRecord, subnets []handlers_ec2_vpc.SubnetRecord, igws []handlers_ec2_igw.IGWRecord, enis []handlers_ec2_vpc.ENIRecord) { + t.Helper() + js, err := nc.JetStream() + require.NoError(t, err) + + vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1}) + require.NoError(t, err) + for _, v := range vpcs { + data, _ := json.Marshal(v) + _, err := vpcKV.Put(utils.AccountKey("000000000001", v.VpcId), data) + require.NoError(t, err) + } + + subnetKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketSubnets, History: 1}) + require.NoError(t, err) + for _, s := range subnets { + data, _ := json.Marshal(s) + _, err := subnetKV.Put(utils.AccountKey("000000000001", s.SubnetId), data) + require.NoError(t, err) + } + + igwKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_igw.KVBucketIGW, History: 1}) + require.NoError(t, err) + for _, i := range igws { + data, _ := json.Marshal(i) + _, err := igwKV.Put(utils.AccountKey("000000000001", i.InternetGatewayId), data) + require.NoError(t, err) + } + + eniKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketENIs, History: 1}) + require.NoError(t, err) + for _, e := range enis { + data, _ := json.Marshal(e) + _, err := eniKV.Put(utils.AccountKey("000000000001", e.NetworkInterfaceId), data) + require.NoError(t, err) + } +} + +func TestReconcileFromKV_CreatesFullTopology(t *testing.T) { + _, nc := startTestJetStreamNATS(t) + ovn := NewMockOVNClient() + _ = ovn.Connect(context.Background()) + ctx := context.Background() + + topo := NewTopologyHandler(ovn, + WithExternalNetwork("pool", []ExternalPoolConfig{{ + Name: "wan", + RangeStart: "10.0.0.200", + RangeEnd: "10.0.0.250", + Gateway: "10.0.0.1", + GatewayIP: "10.0.0.200", + PrefixLen: 24, + }}), + WithChassisNames([]string{"chassis-node1"}), + ) + + seedKVBuckets(t, nc, + []handlers_ec2_vpc.VPCRecord{{ + VpcId: "vpc-kv1", CidrBlock: "10.100.0.0/16", State: "available", + CreatedAt: time.Now(), + }}, + []handlers_ec2_vpc.SubnetRecord{{ + SubnetId: "subnet-kv1", VpcId: "vpc-kv1", CidrBlock: "10.100.1.0/24", + State: "available", CreatedAt: time.Now(), + }}, + []handlers_ec2_igw.IGWRecord{{ + InternetGatewayId: "igw-kv1", VpcId: "vpc-kv1", State: "attached", + CreatedAt: time.Now(), + }}, + []handlers_ec2_vpc.ENIRecord{{ + NetworkInterfaceId: "eni-kv1", SubnetId: "subnet-kv1", VpcId: "vpc-kv1", + PrivateIpAddress: "10.100.1.10", MacAddress: "02:00:00:aa:bb:cc", + Status: "in-use", CreatedAt: time.Now(), + }}, + ) + + result := ReconcileFromKV(ctx, nc, topo) + assert.Equal(t, 1, result.RoutersCreated) + assert.Equal(t, 1, result.SwitchesCreated) + assert.Equal(t, 1, result.IGWsCreated) + assert.Equal(t, 1, result.PortsCreated) + + // Verify OVN objects + _, err := ovn.GetLogicalRouter(ctx, "vpc-vpc-kv1") + require.NoError(t, err) + + _, err = ovn.GetLogicalSwitch(ctx, "subnet-subnet-kv1") + require.NoError(t, err) + + _, err = ovn.GetLogicalSwitch(ctx, "ext-vpc-kv1") + require.NoError(t, err) + + lsp, err := ovn.GetLogicalSwitchPort(ctx, "port-eni-kv1") + require.NoError(t, err) + assert.Contains(t, lsp.Addresses[0], "10.100.1.10") +} + +func TestReconcileFromKV_Idempotent(t *testing.T) { + _, nc := startTestJetStreamNATS(t) + ovn := NewMockOVNClient() + _ = ovn.Connect(context.Background()) + ctx := context.Background() + + topo := NewTopologyHandler(ovn, + WithExternalNetwork("pool", []ExternalPoolConfig{{ + Name: "wan", Gateway: "10.0.0.1", GatewayIP: "10.0.0.200", PrefixLen: 24, + }}), + WithChassisNames([]string{"chassis-node1"}), + ) + + seedKVBuckets(t, nc, + []handlers_ec2_vpc.VPCRecord{{ + VpcId: "vpc-idem2", CidrBlock: "10.200.0.0/16", State: "available", + CreatedAt: time.Now(), + }}, + []handlers_ec2_vpc.SubnetRecord{{ + SubnetId: "subnet-idem2", VpcId: "vpc-idem2", CidrBlock: "10.200.1.0/24", + State: "available", CreatedAt: time.Now(), + }}, + nil, nil, + ) + + r1 := ReconcileFromKV(ctx, nc, topo) + assert.Equal(t, 1, r1.RoutersCreated) + assert.Equal(t, 1, r1.SwitchesCreated) + + // Second run: everything exists + r2 := ReconcileFromKV(ctx, nc, topo) + assert.Equal(t, 0, r2.RoutersCreated) + assert.Equal(t, 0, r2.SwitchesCreated) +} + +func TestReconcileFromKV_SkipsDetachedIGW(t *testing.T) { + _, nc := startTestJetStreamNATS(t) + ovn := NewMockOVNClient() + _ = ovn.Connect(context.Background()) + ctx := context.Background() + + topo := NewTopologyHandler(ovn, + WithExternalNetwork("pool", []ExternalPoolConfig{{ + Name: "wan", Gateway: "10.0.0.1", GatewayIP: "10.0.0.200", PrefixLen: 24, + }}), + ) + + seedKVBuckets(t, nc, + []handlers_ec2_vpc.VPCRecord{{ + VpcId: "vpc-det", CidrBlock: "10.50.0.0/16", State: "available", + CreatedAt: time.Now(), + }}, + nil, + []handlers_ec2_igw.IGWRecord{{ + InternetGatewayId: "igw-det", VpcId: "", State: "available", + CreatedAt: time.Now(), + }}, + nil, + ) + + result := ReconcileFromKV(ctx, nc, topo) + assert.Equal(t, 1, result.RoutersCreated) // VPC router created + assert.Equal(t, 0, result.IGWsCreated) // Detached IGW skipped + + _, err := ovn.GetLogicalSwitch(ctx, "ext-vpc-det") + assert.Error(t, err) // External switch should NOT exist +} + +func TestReconcileFromKV_NoBuckets(t *testing.T) { + _, nc := startTestJetStreamNATS(t) + ovn := NewMockOVNClient() + _ = ovn.Connect(context.Background()) + + topo := NewTopologyHandler(ovn) + + // No KV buckets seeded ā€” should handle gracefully + result := ReconcileFromKV(context.Background(), nc, topo) + assert.Equal(t, 0, result.RoutersCreated) + assert.Equal(t, 0, result.SwitchesCreated) + assert.Equal(t, 0, result.IGWsCreated) + assert.Equal(t, 0, result.PortsCreated) +} diff --git a/spinifex/services/vpcd/vpcd.go b/spinifex/services/vpcd/vpcd.go index e3410ace7..54b3af8ed 100644 --- a/spinifex/services/vpcd/vpcd.go +++ b/spinifex/services/vpcd/vpcd.go @@ -277,6 +277,10 @@ 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. + ReconcileFromKV(ctx, nc, topo) + slog.Info("vpcd service started, waiting for VPC lifecycle events", "subscriptions", len(subs)) // Wait for shutdown signal