diff --git a/spinifex/services/vpcd/mock_ovn.go b/spinifex/services/vpcd/mock_ovn.go index 7740c65db..8a427ab78 100644 --- a/spinifex/services/vpcd/mock_ovn.go +++ b/spinifex/services/vpcd/mock_ovn.go @@ -475,6 +475,17 @@ func (m *MockOVNClient) DeleteAllNATsByExternalIP(_ context.Context, natType, ex return len(foundUUIDs), nil } +func (m *MockOVNClient) FindNATByExternalIP(_ context.Context, natType, externalIP string) (*nbdb.NAT, error) { + m.mu.Lock() + defer m.mu.Unlock() + for _, n := range m.nats { + if n.Type == natType && n.ExternalIP == externalIP { + return n, nil + } + } + return nil, nil +} + // Static Routes func (m *MockOVNClient) AddStaticRoute(_ context.Context, routerName string, route *nbdb.LogicalRouterStaticRoute) error { diff --git a/spinifex/services/vpcd/ovn.go b/spinifex/services/vpcd/ovn.go index 536bc689d..ca529351b 100644 --- a/spinifex/services/vpcd/ovn.go +++ b/spinifex/services/vpcd/ovn.go @@ -91,6 +91,7 @@ type OVNClient interface { DeleteNAT(ctx context.Context, routerName string, natType, logicalIP string) error DeleteNATByExternalIP(ctx context.Context, routerName string, natType, externalIP string) error DeleteAllNATsByExternalIP(ctx context.Context, natType, externalIP string) (int, error) + FindNATByExternalIP(ctx context.Context, natType, externalIP string) (*nbdb.NAT, error) // Static routes AddStaticRoute(ctx context.Context, routerName string, route *nbdb.LogicalRouterStaticRoute) error @@ -871,6 +872,22 @@ func (c *LiveOVNClient) DeleteAllNATsByExternalIP(ctx context.Context, natType, return len(nats), nil } +// FindNATByExternalIP returns the first NAT rule matching the given type and +// external IP, or nil if none exists. Used by the startup reconcile to check +// whether an EIP already has a dnat_and_snat rule before (re-)creating one. +func (c *LiveOVNClient) FindNATByExternalIP(ctx context.Context, natType, externalIP string) (*nbdb.NAT, error) { + var nats []nbdb.NAT + if err := c.client.WhereCache(func(n *nbdb.NAT) bool { + return n.Type == natType && n.ExternalIP == externalIP + }).List(ctx, &nats); err != nil { + return nil, fmt.Errorf("find NAT by external IP: %w", err) + } + if len(nats) == 0 { + return nil, nil + } + return &nats[0], nil +} + func (c *LiveOVNClient) AddStaticRoute(ctx context.Context, routerName string, route *nbdb.LogicalRouterStaticRoute) error { // Set a named UUID so the route can be referenced in the same transaction if route.UUID == "" { diff --git a/spinifex/services/vpcd/reconcile.go b/spinifex/services/vpcd/reconcile.go index 84c872e49..b1e603432 100644 --- a/spinifex/services/vpcd/reconcile.go +++ b/spinifex/services/vpcd/reconcile.go @@ -7,6 +7,7 @@ import ( "log/slog" "time" + handlers_ec2_eip "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eip" 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" @@ -61,6 +62,7 @@ type ReconcileResult struct { SwitchesCreated int IGWsCreated int PortsCreated int + NATsReconciled int } // Reconcile ensures OVN topology matches the expected state from the bootstrap config. @@ -276,7 +278,10 @@ func ReconcileFromKV(ctx context.Context, nc *nats.Conn, topo *TopologyHandler, } } - // 4. Reconcile ENI ports: ensure each ENI has a logical switch port + // 4. Reconcile ENI ports: ensure each ENI has a logical switch port. + // eniMAC is populated here and consumed by Step 5 to build distributed NAT + // rules (direct bridge mode only; centralized mode doesn't need the MAC). + eniMAC := make(map[string]string) // eniID → MAC eniKV, err := js.KeyValue(handlers_ec2_vpc.KVBucketENIs) if err != nil { slog.Debug("vpcd reconcile-kv: ENI KV bucket not available", "err", err) @@ -299,6 +304,8 @@ func ReconcileFromKV(ctx context.Context, nc *nats.Conn, topo *TopologyHandler, continue } + eniMAC[rec.NetworkInterfaceId] = rec.MacAddress + portName := "port-" + rec.NetworkInterfaceId if _, err := topo.ovn.GetLogicalSwitchPort(ctx, portName); err != nil { switchName := "subnet-" + rec.SubnetId @@ -330,11 +337,88 @@ func ReconcileFromKV(ctx context.Context, nc *nats.Conn, topo *TopologyHandler, } } + // 5. Reconcile EIP NAT rules: ensure each associated EIP has a dnat_and_snat + // rule in OVN NB DB. NAT rules are only created by AssociateAddress events + // and are never otherwise reconciled — an OVN NB DB wipe or any state drift + // silently breaks inbound EIP traffic. This step is the safety net. + eipKV, err := js.KeyValue(handlers_ec2_eip.KVBucketEIPs) + if err != nil { + slog.Debug("vpcd reconcile-kv: EIP KV bucket not available", "err", err) + } else { + eipKeys, err := eipKV.Keys() + if err != nil && !errors.Is(err, nats.ErrNoKeysFound) { + slog.Warn("vpcd reconcile-kv: failed to list EIP keys", "err", err) + } + for _, key := range eipKeys { + if key == utils.VersionKey { + continue + } + entry, err := eipKV.Get(key) + if err != nil { + continue + } + var rec handlers_ec2_eip.EIPRecord + if err := json.Unmarshal(entry.Value(), &rec); err != nil { + slog.Warn("vpcd reconcile-kv: failed to unmarshal EIP record", "key", key, "err", err) + continue + } + if rec.State != "associated" || rec.VpcId == "" || rec.PublicIp == "" || rec.PrivateIp == "" { + continue + } + + existing, err := topo.ovn.FindNATByExternalIP(ctx, "dnat_and_snat", rec.PublicIp) + if err != nil { + slog.Warn("vpcd reconcile-kv: failed to query NAT rule", "external_ip", rec.PublicIp, "err", err) + continue + } + if existing != nil && existing.LogicalIP == rec.PrivateIp { + slog.Debug("vpcd reconcile-kv: NAT rule correct, skipping", "external_ip", rec.PublicIp) + continue + } + + routerName := "vpc-" + rec.VpcId + natRule := &nbdb.NAT{ + Type: "dnat_and_snat", + ExternalIP: rec.PublicIp, + LogicalIP: rec.PrivateIp, + ExternalIDs: map[string]string{ + "spinifex:vpc_id": rec.VpcId, + "spinifex:public_ip": rec.PublicIp, + }, + } + if !topo.useCentralizedNAT() && rec.ENIId != "" { + portName := "port-" + rec.ENIId + natRule.LogicalPort = &portName + if mac, ok := eniMAC[rec.ENIId]; ok && mac != "" { + natRule.ExternalMAC = &mac + } + } + + if existing != nil { + if removed, err := topo.ovn.DeleteAllNATsByExternalIP(ctx, "dnat_and_snat", rec.PublicIp); err != nil { + slog.Warn("vpcd reconcile-kv: failed to remove stale NAT rule", "external_ip", rec.PublicIp, "err", err) + continue + } else if removed > 0 { + slog.Info("vpcd reconcile-kv: removed stale NAT rule before re-add", "external_ip", rec.PublicIp, "removed", removed) + } + } + + if err := topo.ovn.AddNAT(ctx, routerName, natRule); err != nil { + slog.Error("vpcd reconcile-kv: failed to add NAT rule", "external_ip", rec.PublicIp, "logical_ip", rec.PrivateIp, "err", err) + continue + } + slog.Info("vpcd reconcile-kv: reconciled NAT rule", + "external_ip", rec.PublicIp, "logical_ip", rec.PrivateIp, "router", routerName) + result.NATsReconciled++ + } + } + slog.Info("vpcd reconcile-kv: complete", "routers_created", result.RoutersCreated, "switches_created", result.SwitchesCreated, "igws_created", result.IGWsCreated, "ports_created", result.PortsCreated, + "nats_reconciled", result.NATsReconciled, ) return result diff --git a/spinifex/services/vpcd/reconcile_test.go b/spinifex/services/vpcd/reconcile_test.go index bed675f83..ca602b7a9 100644 --- a/spinifex/services/vpcd/reconcile_test.go +++ b/spinifex/services/vpcd/reconcile_test.go @@ -6,6 +6,7 @@ import ( "testing" "time" + handlers_ec2_eip "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eip" 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" @@ -656,3 +657,153 @@ func TestReconcileGatewayChassis_NoOpWhenAlreadyCorrect(t *testing.T) { assert.Equal(t, 0, ovn.DeleteGatewayChassisCalls, "no deletes expected") assert.Equal(t, 0, ovn.UpdateGatewayChassisPriorityCalls, "no priority updates expected") } + +// seedEIPKVBucket creates the EIP KV bucket and populates it with test records. +func seedEIPKVBucket(t *testing.T, nc *nats.Conn, eips []handlers_ec2_eip.EIPRecord) { + t.Helper() + js, err := nc.JetStream() + require.NoError(t, err) + eipKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_eip.KVBucketEIPs, History: 1}) + require.NoError(t, err) + for _, e := range eips { + data, _ := json.Marshal(e) + _, err := eipKV.Put(utils.AccountKey("000000000001", e.AllocationId), data) + require.NoError(t, err) + } +} + +func TestReconcileFromKV_ReconcilesEIPNATRules(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"}), + ) + + // Seed VPC + IGW so the router exists for AddNAT. + seedKVBuckets(t, nc, + []handlers_ec2_vpc.VPCRecord{{ + VpcId: "vpc-nat1", CidrBlock: "10.10.0.0/16", State: "available", CreatedAt: time.Now(), + }}, + nil, + []handlers_ec2_igw.IGWRecord{{ + InternetGatewayId: "igw-nat1", VpcId: "vpc-nat1", State: "attached", CreatedAt: time.Now(), + }}, + nil, + ) + + // One associated EIP (missing NAT rule), one unassociated EIP (must be skipped). + seedEIPKVBucket(t, nc, []handlers_ec2_eip.EIPRecord{ + { + AllocationId: "eipalloc-1", PublicIp: "10.0.0.210", PoolName: "wan", + State: "associated", VpcId: "vpc-nat1", PrivateIp: "10.10.1.5", ENIId: "eni-abc123", + CreatedAt: time.Now(), + }, + { + AllocationId: "eipalloc-2", PublicIp: "10.0.0.211", PoolName: "wan", + State: "allocated", + CreatedAt: time.Now(), + }, + }) + + result := ReconcileFromKV(ctx, nc, topo, nil) + assert.Equal(t, 1, result.NATsReconciled, "one missing NAT rule must be created") + + nat, err := ovn.FindNATByExternalIP(ctx, "dnat_and_snat", "10.0.0.210") + require.NoError(t, err) + require.NotNil(t, nat, "dnat_and_snat rule must exist after reconcile") + assert.Equal(t, "10.10.1.5", nat.LogicalIP) + assert.Equal(t, "dnat_and_snat", nat.Type) + + // Unassociated EIP must produce no NAT rule. + nat2, err := ovn.FindNATByExternalIP(ctx, "dnat_and_snat", "10.0.0.211") + require.NoError(t, err) + assert.Nil(t, nat2, "unassociated EIP must not produce a NAT rule") +} + +func TestReconcileFromKV_EIPNATIdempotent(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-nat2", CidrBlock: "10.20.0.0/16", State: "available", CreatedAt: time.Now(), + }}, + nil, + []handlers_ec2_igw.IGWRecord{{ + InternetGatewayId: "igw-nat2", VpcId: "vpc-nat2", State: "attached", CreatedAt: time.Now(), + }}, + nil, + ) + seedEIPKVBucket(t, nc, []handlers_ec2_eip.EIPRecord{{ + AllocationId: "eipalloc-3", PublicIp: "10.0.0.220", PoolName: "wan", + State: "associated", VpcId: "vpc-nat2", PrivateIp: "10.20.1.7", ENIId: "eni-def456", + CreatedAt: time.Now(), + }}) + + // First reconcile: creates the NAT rule. + r1 := ReconcileFromKV(ctx, nc, topo, nil) + assert.Equal(t, 1, r1.NATsReconciled) + + // Second reconcile: rule already correct, must skip. + r2 := ReconcileFromKV(ctx, nc, topo, nil) + assert.Equal(t, 0, r2.NATsReconciled, "correct NAT rule must not be re-created") +} + +func TestReconcileFromKV_EIPNATStaleRuleReplaced(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-nat3", CidrBlock: "10.30.0.0/16", State: "available", CreatedAt: time.Now(), + }}, + nil, + []handlers_ec2_igw.IGWRecord{{ + InternetGatewayId: "igw-nat3", VpcId: "vpc-nat3", State: "attached", CreatedAt: time.Now(), + }}, + nil, + ) + seedEIPKVBucket(t, nc, []handlers_ec2_eip.EIPRecord{{ + AllocationId: "eipalloc-4", PublicIp: "10.0.0.230", PoolName: "wan", + State: "associated", VpcId: "vpc-nat3", PrivateIp: "10.30.1.9", ENIId: "eni-ghi789", + CreatedAt: time.Now(), + }}) + + // Create the router so we can pre-seed a stale NAT rule against it. + require.NoError(t, ovn.CreateLogicalRouter(ctx, &nbdb.LogicalRouter{Name: "vpc-vpc-nat3"})) + require.NoError(t, ovn.AddNAT(ctx, "vpc-vpc-nat3", &nbdb.NAT{ + Type: "dnat_and_snat", ExternalIP: "10.0.0.230", LogicalIP: "10.30.1.99", + })) + + result := ReconcileFromKV(ctx, nc, topo, nil) + assert.Equal(t, 1, result.NATsReconciled, "stale NAT rule must be replaced") + + nat, err := ovn.FindNATByExternalIP(ctx, "dnat_and_snat", "10.0.0.230") + require.NoError(t, err) + require.NotNil(t, nat) + assert.Equal(t, "10.30.1.9", nat.LogicalIP, "logical IP must be updated to match KV") +}