From 090faf504667c2e60fbf6b2edb5886dfdb319f33 Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Thu, 21 May 2026 22:27:50 +1000 Subject: [PATCH] refactor(vpcd): migrate NAT GW subscribers to policy.NATManager MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit handleAddNATGateway / handleDeleteNATGateway now decode the NATS event and delegate to policy.NATManager.{AddNATGateway,DeleteNATGateway}. The manager carries identical OVN logic — same snat type, same ExternalIDs (spinifex:vpc_id + spinifex:nat_gateway_id) — so existing TestTopologyHandler_AddNATGateway / DeleteNATGateway suites stay green. Lazy-constructed NATManager added on TopologyHandler (natmOnce). Mode is derived from the bridge mode (veth → centralized, direct → distributed) and the FlowsBarrier is bound to waitForFlowsHV for parity with the legacy EIP path's tail barrier (used by the next slice, vpc.add-nat). Slice 4 of mulga-siv-129, sub-step b. --- spinifex/services/vpcd/topology.go | 74 ++++++++++------------ spinifex/services/vpcd/topology_manager.go | 15 +++++ 2 files changed, 50 insertions(+), 39 deletions(-) diff --git a/spinifex/services/vpcd/topology.go b/spinifex/services/vpcd/topology.go index a04f6d1cd..4a827f63c 100644 --- a/spinifex/services/vpcd/topology.go +++ b/spinifex/services/vpcd/topology.go @@ -188,6 +188,13 @@ type TopologyHandler struct { // SG subscribers reuse one manager instance across events. sgmOnce sync.Once sgm policy.SecurityGroupManager + + // natm + natmOnce cache the policy.NATManager backed by h.ovn. NAT mode + // resolves from h.bridgeMode at first use; the FlowsBarrier closure + // shells out to ovn-nbctl --wait=hv sync via waitForFlowsHV. + natmOnce sync.Once + natm policy.NATManager + natmErr error } // NewTopologyHandler creates a new TopologyHandler with optional external network config. @@ -1531,63 +1538,52 @@ func respond(msg *nats.Msg, err error) { } } -// handleAddNATGateway creates an OVN SNAT rule for a private subnet via a NAT Gateway. -// The SNAT rule rewrites source IPs from the private subnet CIDR to the NAT GW's public IP. +// handleAddNATGateway is a thin wrapper that decodes the NATS event and +// delegates the SNAT installation to policy.NATManager.AddNATGateway. func (h *TopologyHandler) handleAddNATGateway(msg *nats.Msg) { var evt NATGatewayEvent if err := json.Unmarshal(msg.Data, &evt); err != nil { slog.Error("vpcd: failed to unmarshal vpc.add-nat-gateway event", "err", err) return } - - slog.Info("vpcd: adding NAT Gateway SNAT rule", - "vpcId", evt.VpcId, "natGatewayId", evt.NatGatewayId, - "publicIp", evt.PublicIp, "subnetCidr", evt.SubnetCidr) - - ctx := context.Background() - routerName := "vpc-" + evt.VpcId - - snatRule := &nbdb.NAT{ - Type: "snat", - ExternalIP: evt.PublicIp, - LogicalIP: evt.SubnetCidr, - ExternalIDs: map[string]string{ - "spinifex:vpc_id": evt.VpcId, - "spinifex:nat_gateway_id": evt.NatGatewayId, - }, - } - - if err := h.ovn.AddNAT(ctx, routerName, snatRule); err != nil { - slog.Error("vpcd: failed to add NAT Gateway SNAT rule", - "router", routerName, "publicIp", evt.PublicIp, - "subnetCidr", evt.SubnetCidr, "err", err) + nm, err := h.natManager() + if err != nil { + slog.Error("vpcd: NAT manager init failed", "err", err) + return + } + if err := nm.AddNATGateway(context.Background(), policy.NATGWSpec{ + VPCID: evt.VpcId, + NATGatewayID: evt.NatGatewayId, + PublicIP: evt.PublicIp, + SubnetCIDR: evt.SubnetCidr, + }); err != nil { + slog.Error("vpcd: AddNATGateway failed", + "vpc_id", evt.VpcId, "natgw_id", evt.NatGatewayId, + "public_ip", evt.PublicIp, "subnet_cidr", evt.SubnetCidr, "err", err) return } - slog.Info("vpcd: NAT Gateway SNAT rule added", - "router", routerName, "publicIp", evt.PublicIp, "subnetCidr", evt.SubnetCidr) + "vpc_id", evt.VpcId, "natgw_id", evt.NatGatewayId, + "public_ip", evt.PublicIp, "subnet_cidr", evt.SubnetCidr) } -// handleDeleteNATGateway removes the OVN SNAT rule for a private subnet's NAT Gateway. +// handleDeleteNATGateway delegates SNAT teardown to policy.NATManager. func (h *TopologyHandler) handleDeleteNATGateway(msg *nats.Msg) { var evt NATGatewayEvent if err := json.Unmarshal(msg.Data, &evt); err != nil { slog.Error("vpcd: failed to unmarshal vpc.delete-nat-gateway event", "err", err) return } - - slog.Info("vpcd: removing NAT Gateway SNAT rule", - "vpcId", evt.VpcId, "natGatewayId", evt.NatGatewayId, "subnetCidr", evt.SubnetCidr) - - ctx := context.Background() - routerName := "vpc-" + evt.VpcId - - if err := h.ovn.DeleteNAT(ctx, routerName, "snat", evt.SubnetCidr); err != nil { - slog.Warn("vpcd: failed to delete NAT Gateway SNAT rule", - "router", routerName, "subnetCidr", evt.SubnetCidr, "err", err) + nm, err := h.natManager() + if err != nil { + slog.Error("vpcd: NAT manager init failed", "err", err) + return + } + if err := nm.DeleteNATGateway(context.Background(), evt.VpcId, evt.SubnetCidr); err != nil { + slog.Warn("vpcd: DeleteNATGateway failed", + "vpc_id", evt.VpcId, "subnet_cidr", evt.SubnetCidr, "err", err) return } - slog.Info("vpcd: NAT Gateway SNAT rule removed", - "router", routerName, "subnetCidr", evt.SubnetCidr) + "vpc_id", evt.VpcId, "natgw_id", evt.NatGatewayId, "subnet_cidr", evt.SubnetCidr) } diff --git a/spinifex/services/vpcd/topology_manager.go b/spinifex/services/vpcd/topology_manager.go index b2617a16e..43c8a9ff3 100644 --- a/spinifex/services/vpcd/topology_manager.go +++ b/spinifex/services/vpcd/topology_manager.go @@ -103,3 +103,18 @@ func (h *TopologyHandler) securityGroupManager() policy.SecurityGroupManager { }) return h.sgm } + +// natManager returns the lazily-constructed policy.NATManager. NAT mode is +// derived from the configured bridge mode (veth → centralized, direct → +// distributed) and the FlowsBarrier is bound to waitForFlowsHV so the EIP +// subscriber blocks on hypervisor flow-install before responding. +func (h *TopologyHandler) natManager() (policy.NATManager, error) { + h.natmOnce.Do(func() { + mode := policy.NATModeDistributed + if h.useCentralizedNAT() { + mode = policy.NATModeCentralized + } + h.natm, h.natmErr = policy.NewNATManager(h.ovn, mode, policy.WithFlowsBarrier(waitForFlowsHV)) + }) + return h.natm, h.natmErr +}