fix(vpcd): reconcile EIP dnat_and_snat rules on startup

NAT rules are purely event-driven — only AssociateAddress creates them via vpc.add-nat. No startup reconcile path means any OVN NB DB state drift silently breaks EIP inbound traffic after a node reboot. Add Step 5 to ReconcileFromKV that reads associated EIPs from the spinifex-vpc-elastic-ips KV bucket and ensures each has a dnat_and_snat rule in OVN NB DB. Correct rules are skipped (idempotent). Stale rules (wrong LogicalIP) are replaced. Missing rules are created. Also adds FindNATByExternalIP to OVNClient interface + LiveOVNClient/Mock implementations, and three new tests covering: missing rule creation, idempotent skip on correct rule, and stale rule replacement.
2026-10-19 10:52:04 +00:00 · 2026-05-07 23:41:41 +10:00
parent 5ea7df6e7c
commit f44db4d84e
4 changed files with 264 additions and 1 deletions
@@ -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 {
@@ -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 == "" {
@@ -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
@@ -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")
+}