gsadmin/spinifex
1
0
Fork 0
mirror of https://github.com/mulgadc/spinifex.git synced 2026-10-19 10:52:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(vpcd): delete dead retrofit/reconcileIGW helpers + bridge-mode bleed

Browse Source 
Phase 4 / mulga-siv-125.6 Slice B.

Production path no longer touches any of:

- RetrofitAllGatewayPortNetworks / RetrofitAllExternalLocalnetOptions —
  the 5-pass startup leftovers that network/reconcile (Phase 2.2) made
  redundant. Nothing called them outside tests.
- reconcileIGW + its allocateGatewayLRPIP / gwLrpRange helpers + the
  ipv4ToUint32 / uint32ToIPv4 utilities — superseded by
  external.IGWManager (Phase 2.4) which carries its own pool logic.
- ensureLocalnetOptions / ensureGatewayPortNetworks /
  expectedGatewayPortNetwork — read-before-write retrofit shims for the
  same code paths.
- generateMAC — only callers were reconcileIGW + its tests.

useCentralizedNAT (the bridge-mode bleed flagged in plan §2.2) is
replaced by a policy.NATMode value injected via WithNATMode(...). The
mode is resolved once in vpcd.go from host.UplinkMode and threaded into
both the topology-handler lazy managers and the reconciler stack, so
"is this veth?" is no longer a question topology.go can ask. The
TopologyHandler defaults to NATModeDistributed when WithNATMode is
omitted, matching the previous default.

Drops services/vpcd/topology.go from 1220 LOC to 693, deletes ~700 LOC
of obsolete tests (TestGenerateMAC, TestEnsureLocalnetOptions_*,
TestRetrofitAllExternalLocalnetOptions_FixesClearedOptions,
TestReconcileIGW_*, TestGwLrpRange), and frees the nbdb + utils +
net imports in topology.go.

Stale "phase 4 migrates those onto the new managers" comment in vpcd.go
fixed in passing — the subscribers were migrated in Phase 2.6
(mulga-siv-129); only the file-relocation parts of Phase 4 remain.
Stale ensureLocalnetOptions / ensureGatewayPortNetworks references in
network/ovn/{live.go,mock/mock.go} doc comments also cleaned up.
This commit is contained in:
joshsiv-mulga
2026-05-25 14:11:14 +10:00
parent bcad478a4b
commit b4d784bc65
6 changed files with 45 additions and 1261 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spinifex/network/ovn/live.go
+1 -3
View File
@@ -462,9 +462,7 @@ func (c *LiveClient) GetLogicalRouterPort(ctx context.Context, name string) (*nb
}
// UpdateLogicalRouterPort rewrites mutable columns on an existing LRP.
// Used by ensureGatewayPortNetworks to retrofit the link-local Networks
// CIDR onto gateway ports created by older code that used pool IPs
// (mulga-siv-26 D8). Mirrors UpdateLogicalSwitchPort.
// Mirrors UpdateLogicalSwitchPort.
func (c *LiveClient) UpdateLogicalRouterPort(ctx context.Context, lrp *nbdb.LogicalRouterPort) error {
if lrp.UUID == "" {
existing, err := c.GetLogicalRouterPort(ctx, lrp.Name)
spinifex/network/ovn/mock/mock.go
+2 -4
View File
@@ -29,13 +29,11 @@ type Client struct {
GatewayChassis map[string]*nbdb.GatewayChassis // keyed by UUID
// UpdateLogicalSwitchPortCalls counts UpdateLogicalSwitchPort invocations
// so tests can assert idempotent read-before-write paths (e.g.
// ensureLocalnetOptions — mulga-998.b Fix 3).
// so tests can assert idempotent read-before-write paths.
UpdateLogicalSwitchPortCalls int
// UpdateLogicalRouterPortCalls counts UpdateLogicalRouterPort invocations
// so tests can assert ensureGatewayPortNetworks only writes on drift
// (mulga-siv-26 D8).
// so tests can assert writers only emit on drift.
UpdateLogicalRouterPortCalls int
// SetGatewayChassisCalls / DeleteGatewayChassisCalls /
spinifex/services/vpcd/topology.go
+13 -514
View File
@@ -12,11 +12,9 @@ import (
"time"
"github.com/mulgadc/spinifex/spinifex/network/external"
"github.com/mulgadc/spinifex/spinifex/network/ovn/nbdb"
"github.com/mulgadc/spinifex/spinifex/network/policy"
"github.com/mulgadc/spinifex/spinifex/network/topology"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -74,23 +72,6 @@ const (
TopicUpdateSG = "vpc.update-sg"
)
// gatewayPortNetwork is the link-local CIDR every IGW gateway-LRP carries
// in distributed-NAT (direct) mode. Per-VM dnat_and_snat handles ARP per
// chassis so the LRP IP itself never goes on the wire. The default route's
// OutputPort is set explicitly so the WAN nexthop need not be on this subnet.
//
// Centralized-NAT (veth) bypasses this constant: the gateway LRP is
// the on-wire egress point and must hold a WAN-subnet IP from the pool's
// gw_lrp_range, otherwise the upstream router silently drops ARP requests
// from an off-subnet sender (RFC 826) and the default route never resolves
// (mulga-siv-36).
const gatewayPortNetwork = "169.254.0.1/30"
// gatewayIPExtID is the LRP external_ids key holding the gw IP allocated
// from the pool's gw_lrp_range. Persisted so reconcile/retrofit can recover
// the assignment without re-allocating, and so siblings see it as "used".
const gatewayIPExtID = "spinifex:gateway_ip"
// VPCEvent is published on vpc.create after a VPC is persisted.
type VPCEvent struct {
VpcId string `json:"vpc_id"`
@@ -170,8 +151,8 @@ type TopologyHandler struct {
ovn OVNClient
externalMode string
externalPools []ExternalPoolConfig
chassisNames []string // OVN chassis names for gateway HA scheduling
bridgeMode string // "direct" or "veth" — controls NAT mode and localnet options
chassisNames []string // OVN chassis names for gateway HA scheduling
natMode policy.NATMode // distributed (direct uplink) or centralized (veth uplink)
// lm + lmOnce cache the topology.Manager produced from h.ovn so each
// adapter call doesn't re-construct (Phase 2.6, mulga-siv-129).
@@ -183,9 +164,9 @@ type TopologyHandler struct {
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.
// natm + natmOnce cache the policy.NATManager backed by h.ovn. NAT mode is
// the value injected via WithNATMode; the FlowsBarrier closure shells out
// to ovn-nbctl --wait=hv sync via waitForFlowsHV.
natmOnce sync.Once
natm policy.NATManager
natmErr error
@@ -198,8 +179,11 @@ type TopologyHandler struct {
}
// NewTopologyHandler creates a new TopologyHandler with optional external network config.
// NAT mode defaults to Distributed (direct uplink) when WithNATMode is omitted;
// matches the pre-Slice-B behaviour where unset bridgeMode also implied
// distributed via useCentralizedNAT().
func NewTopologyHandler(ovn OVNClient, opts ...TopologyOption) *TopologyHandler {
h := &TopologyHandler{ovn: ovn}
h := &TopologyHandler{ovn: ovn, natMode: policy.NATModeDistributed}
for _, opt := range opts {
opt(h)
}
@@ -224,184 +208,11 @@ func WithChassisNames(names []string) TopologyOption {
}
}
// WithBridgeMode sets the external bridge mode ("direct" or "veth").
// Direct bridge enables distributed NAT; veth uses centralized NAT.
func WithBridgeMode(mode string) TopologyOption {
// WithNATMode sets the NAT mode used by lazily-constructed managers. Derived
// upstream from the uplink/bridge mode (direct→distributed, veth→centralized).
func WithNATMode(mode policy.NATMode) TopologyOption {
return func(h *TopologyHandler) {
h.bridgeMode = mode
}
}
// useCentralizedNAT returns true if the bridge mode requires centralized NAT.
// Veth mode needs centralized NAT because the Linux bridge intermediary breaks
// distributed NAT hairpin routing. Only direct bridge mode supports distributed NAT.
func (h *TopologyHandler) useCentralizedNAT() bool {
return h.bridgeMode == BridgeModeVeth
}
// ensureLocalnetOptions aligns options on an existing localnet port with the
// current bridge mode. CreateLogicalSwitchPort only seeds options at first
// creation; once created, stale options persist forever (e.g. vpcd came up in
// the wrong mode once — pre-Fix 1 — or an operator cleared options manually).
// This read-before-write is a no-op when already correct, so it is safe to
// call on every reconcile.
//
// network_name=external is required regardless of bridge mode — it binds the
// port to the ovn-bridge-mappings entry for the external OVS bridge.
// Centralized mode (veth): options:nat-addresses=router must be set.
// Distributed mode (direct): options:nat-addresses must be absent.
func (h *TopologyHandler) ensureLocalnetOptions(ctx context.Context, extPortName string) error {
lsp, err := h.ovn.GetLogicalSwitchPort(ctx, extPortName)
if err != nil {
return fmt.Errorf("get localnet port %s: %w", extPortName, err)
}
wantNat := h.useCentralizedNAT()
currentNat, hasNat := lsp.Options["nat-addresses"]
currentNetName, hasNetName := lsp.Options["network_name"]
natOK := (wantNat && hasNat && currentNat == "router") || (!wantNat && !hasNat)
netNameOK := hasNetName && currentNetName == "external"
if natOK && netNameOK {
return nil
}
if lsp.Options == nil {
lsp.Options = map[string]string{}
}
lsp.Options["network_name"] = "external"
if wantNat {
lsp.Options["nat-addresses"] = "router"
} else {
delete(lsp.Options, "nat-addresses")
}
slog.Info("vpcd: retrofitting localnet options",
"port", extPortName, "bridge_mode", h.bridgeMode,
"nat-addresses", lsp.Options["nat-addresses"],
"network_name", lsp.Options["network_name"])
if err := h.ovn.UpdateLogicalSwitchPort(ctx, lsp); err != nil {
return fmt.Errorf("update localnet port %s options: %w", extPortName, err)
}
return nil
}
// expectedGatewayPortNetwork returns the Networks CIDR a gateway LRP for
// vpcId should carry. In direct mode, or in centralized mode without a pool,
// it is the link-local fallback. In centralized mode 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
}
pool := h.findExternalPool("", "")
if pool == nil {
return gatewayPortNetwork, "", nil
}
ip, prefix, ok, allocErr := h.allocateGatewayLRPIP(ctx, vpcId, pool)
if allocErr != nil {
return "", "", allocErr
}
if !ok {
return gatewayPortNetwork, "", nil
}
return fmt.Sprintf("%s/%d", ip, prefix), ip, nil
}
// ensureGatewayPortNetworks rewrites the gateway LRP's Networks column in
// place when it drifts from the mode's expected value. CreateLogicalRouterPort
// is a no-op when the row exists, so reconcile of an upgraded cluster
// otherwise keeps the stale pool-IP networks (mulga-siv-26 D8) or, in
// centralized NAT, the off-subnet link-local that breaks ARP upstream
// (mulga-siv-36). Idempotent — no UPDATE is issued when already correct.
func (h *TopologyHandler) ensureGatewayPortNetworks(ctx context.Context, gwPortName string) error {
lrp, err := h.ovn.GetLogicalRouterPort(ctx, gwPortName)
if err != nil {
return fmt.Errorf("get gateway router port %s: %w", gwPortName, err)
}
vpcId := lrp.ExternalIDs["spinifex:vpc_id"]
if vpcId == "" {
// No VPC tag — this is from the legacy "gw-vpc-..." naming or a
// hand-edited row; fall back to link-local to keep behavior safe.
if len(lrp.Networks) == 1 && lrp.Networks[0] == gatewayPortNetwork {
return nil
}
slog.Info("vpcd: rewriting stale gateway port networks (no vpc_id tag)",
"port", gwPortName, "old", lrp.Networks, "new", gatewayPortNetwork)
lrp.Networks = []string{gatewayPortNetwork}
return h.ovn.UpdateLogicalRouterPort(ctx, lrp)
}
wantNetwork, wantGwIP, err := h.expectedGatewayPortNetwork(ctx, vpcId)
if err != nil {
return fmt.Errorf("compute expected network for %s: %w", gwPortName, err)
}
currentGwIP := lrp.ExternalIDs[gatewayIPExtID]
if len(lrp.Networks) == 1 && lrp.Networks[0] == wantNetwork && currentGwIP == wantGwIP {
return nil
}
slog.Info("vpcd: rewriting stale gateway port networks",
"port", gwPortName, "old", lrp.Networks, "new", wantNetwork,
"old_gw_ip", currentGwIP, "new_gw_ip", wantGwIP)
lrp.Networks = []string{wantNetwork}
if lrp.ExternalIDs == nil {
lrp.ExternalIDs = map[string]string{}
}
if wantGwIP == "" {
delete(lrp.ExternalIDs, gatewayIPExtID)
} else {
lrp.ExternalIDs[gatewayIPExtID] = wantGwIP
}
if err := h.ovn.UpdateLogicalRouterPort(ctx, lrp); err != nil {
return fmt.Errorf("update gateway router port %s: %w", gwPortName, err)
}
return nil
}
// RetrofitAllGatewayPortNetworks walks every LRP tagged
// spinifex:role=gateway and ensures Networks matches the current mode. Needed
// because Reconcile/ReconcileFromKV early-return when the external switch
// already exists, so reconcileIGW never runs against pre-existing topologies
// shipped by older builds (mulga-siv-26 D8 / mulga-siv-36). Idempotent — the
// underlying ensureGatewayPortNetworks no-ops when already correct.
func (h *TopologyHandler) RetrofitAllGatewayPortNetworks(ctx context.Context) {
lrps, err := h.ovn.ListLogicalRouterPorts(ctx)
if err != nil {
slog.Warn("vpcd: gateway-port retrofit skipped — list LRPs failed", "err", err)
return
}
for _, lrp := range lrps {
if lrp.ExternalIDs["spinifex:role"] != "gateway" {
continue
}
if err := h.ensureGatewayPortNetworks(ctx, lrp.Name); err != nil {
slog.Error("vpcd: gateway-port retrofit failed", "port", lrp.Name, "err", err)
}
}
}
// RetrofitAllExternalLocalnetOptions walks every OVN logical switch tagged
// spinifex:role=external and calls ensureLocalnetOptions on its localnet
// port. OVN is the source of truth — NATS KV records for IGWs may be absent
// or stale (external switch was created via a live event whose KV record
// later expired, or an operator cleared options manually). Walking OVN
// directly catches all of these on every vpcd startup. Idempotent.
func (h *TopologyHandler) RetrofitAllExternalLocalnetOptions(ctx context.Context) {
switches, err := h.ovn.ListLogicalSwitches(ctx)
if err != nil {
slog.Warn("vpcd: retrofit skipped — list logical switches failed", "err", err)
return
}
for i := range switches {
ls := &switches[i]
if ls.ExternalIDs["spinifex:role"] != "external" {
continue
}
vpcID := ls.ExternalIDs["spinifex:vpc_id"]
if vpcID == "" {
continue
}
extPortName := "ext-port-" + vpcID
if err := h.ensureLocalnetOptions(ctx, extPortName); err != nil {
slog.Error("vpcd: retrofit localnet options failed", "port", extPortName, "err", err)
}
h.natMode = mode
}
}
@@ -828,320 +639,8 @@ func (h *TopologyHandler) handleDeleteNAT(msg *nats.Msg) {
respond(msg, nil)
}
// --- Reconciliation (called on startup, not via NATS) ---
// reconcileIGW ensures the OVN external switch, localnet port, gateway router
// port, router-bridging LSP, default route, and gateway chassis bindings exist
// for a VPC's internet gateway. Every step is Get-then-Create idempotent, with
// no cascade rollback on failure: partial topology is safe — the next
// reconcile pass heals whatever is missing.
func (h *TopologyHandler) reconcileIGW(ctx context.Context, vpcId, igwId string) error {
routerName := "vpc-" + vpcId
extSwitchName := "ext-" + vpcId
extPortName := "ext-port-" + vpcId
gwPortName := "gw-" + vpcId
switchGWPortName := "gw-port-" + vpcId
pool := h.findExternalPool("", "")
wanGateway := "169.254.0.2"
if pool != nil {
wanGateway = pool.Gateway
}
// 1. External logical switch
if _, err := h.ovn.GetLogicalSwitch(ctx, extSwitchName); err != nil {
extIDs := map[string]string{
"spinifex:vpc_id": vpcId,
"spinifex:role": "external",
}
if igwId != "" {
extIDs["spinifex:igw_id"] = igwId
}
if err := h.ovn.CreateLogicalSwitch(ctx, &nbdb.LogicalSwitch{
Name: extSwitchName,
ExternalIDs: extIDs,
}); err != nil {
return fmt.Errorf("create external switch %s: %w", extSwitchName, err)
}
}
// 2. Localnet port. ensureLocalnetOptions runs unconditionally — it is a
// read-before-write no-op when the options are already correct, and the
// recovery path (port exists with stale/missing options) needs it.
if _, err := h.ovn.GetLogicalSwitchPort(ctx, extPortName); err != nil {
portExtIDs := map[string]string{"spinifex:vpc_id": vpcId}
if igwId != "" {
portExtIDs["spinifex:igw_id"] = igwId
}
opts := map[string]string{"network_name": "external"}
if h.useCentralizedNAT() {
opts["nat-addresses"] = "router"
}
if err := h.ovn.CreateLogicalSwitchPort(ctx, extSwitchName, &nbdb.LogicalSwitchPort{
Name: extPortName,
Type: "localnet",
Addresses: []string{"unknown"},
Options: opts,
ExternalIDs: portExtIDs,
}); err != nil {
return fmt.Errorf("create localnet port %s: %w", extPortName, err)
}
}
if err := h.ensureLocalnetOptions(ctx, extPortName); err != nil {
return fmt.Errorf("retrofit localnet options %s: %w", extPortName, err)
}
// 3. Gateway router port. expectedGatewayPortNetwork is only consulted
// when we actually create the LRP — RetrofitAllGatewayPortNetworks
// handles drift on pre-existing ports separately.
if _, err := h.ovn.GetLogicalRouterPort(ctx, gwPortName); err != nil {
gatewayNetwork := gatewayPortNetwork
gwLrpIP := ""
if pool != nil && h.useCentralizedNAT() {
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 = network
}
lrpExtIDs := map[string]string{
"spinifex:vpc_id": vpcId,
"spinifex:role": "gateway",
}
if gwLrpIP != "" {
lrpExtIDs[gatewayIPExtID] = gwLrpIP
}
if err := h.ovn.CreateLogicalRouterPort(ctx, routerName, &nbdb.LogicalRouterPort{
Name: gwPortName,
MAC: generateMAC("gw-" + vpcId),
Networks: []string{gatewayNetwork},
ExternalIDs: lrpExtIDs,
}); err != nil {
return fmt.Errorf("create gateway router port %s: %w", gwPortName, err)
}
}
// 4. Switch port connecting external switch to router
if _, err := h.ovn.GetLogicalSwitchPort(ctx, switchGWPortName); err != nil {
if err := h.ovn.CreateLogicalSwitchPort(ctx, extSwitchName, &nbdb.LogicalSwitchPort{
Name: switchGWPortName,
Type: "router",
Addresses: []string{"router"},
Options: map[string]string{"router-port": gwPortName},
ExternalIDs: map[string]string{"spinifex:vpc_id": vpcId},
}); err != nil {
return fmt.Errorf("create switch gateway port %s: %w", switchGWPortName, err)
}
}
// 5. No blanket SNAT — per-VM dnat_and_snat rules handle public instances.
// See handleIGWAttach comment for rationale (AWS parity).
// 6. Default route. AddStaticRoute is non-idempotent (every retry leaves
// a fresh duplicate row), so look up first via FindStaticRoute. An
// existing row with mismatched nexthop/output is treated as an operator
// override and left alone.
existing, err := h.ovn.FindStaticRoute(ctx, routerName, "0.0.0.0/0")
if err != nil {
slog.Warn("vpcd reconcile: failed to query default route", "router", routerName, "err", err)
}
switch {
case existing == nil && err == nil:
if addErr := h.ovn.AddStaticRoute(ctx, routerName, &nbdb.LogicalRouterStaticRoute{
IPPrefix: "0.0.0.0/0",
Nexthop: wanGateway,
OutputPort: &gwPortName,
ExternalIDs: map[string]string{"spinifex:vpc_id": vpcId},
}); addErr != nil {
slog.Warn("vpcd reconcile: failed to add default route", "err", addErr)
}
case existing != nil:
existingPort := ""
if existing.OutputPort != nil {
existingPort = *existing.OutputPort
}
if existing.Nexthop != wanGateway || existingPort != gwPortName {
slog.Warn("vpcd reconcile: default route differs from expected, leaving existing entry in place",
"router", routerName,
"existing_nexthop", existing.Nexthop, "want_nexthop", wanGateway,
"existing_output_port", existingPort, "want_output_port", gwPortName)
}
}
// 7. Schedule gateway chassis (already idempotent via SetGatewayChassis).
for i, chassis := range h.chassisNames {
priority := max(20-(i*5), 1)
if err := h.ovn.SetGatewayChassis(ctx, gwPortName, chassis, priority); err != nil {
slog.Warn("vpcd reconcile: failed to set gateway chassis", "chassis", chassis, "err", err)
}
}
slog.Info("vpcd reconcile: ensured IGW topology",
"ext_switch", extSwitchName, "gw_port", gwPortName, "wan_gateway", wanGateway)
return nil
}
// --- Helpers ---
// gwLrpRange returns the per-VPC gateway LRP IP range for a pool. Priority:
//
// 1. Explicit pool.GwLrpRangeStart/End set in spinifex.toml.
// 2. Auto-derived from pool.Gateway + pool.PrefixLen — last 16 host IPs of
// the WAN subnet (broadcast - 16 .. broadcast - 1). When that range
// overlaps the per-VM EIP range (RangeStart..RangeEnd), shift to the
// 16 IPs immediately below RangeStart instead.
//
// Returns ok=false only when the pool is missing or the gateway/prefix is
// unparseable — link-local has no role here, the WAN-subnet IP is the only
// thing the upstream router will ARP-resolve.
func gwLrpRange(pool *ExternalPoolConfig) (start, end net.IP, prefix int, ok bool) {
if pool == nil {
return nil, nil, 0, false
}
prefix = pool.PrefixLen
if prefix <= 0 || prefix > 32 {
prefix = 24
}
// 1. Explicit operator config wins.
if pool.GwLrpRangeStart != "" || pool.GwLrpRangeEnd != "" {
s := net.ParseIP(pool.GwLrpRangeStart).To4()
e := net.ParseIP(pool.GwLrpRangeEnd).To4()
if s != nil && e != nil && ipv4ToUint32(s) <= ipv4ToUint32(e) {
return s, e, prefix, true
}
slog.Warn("vpcd: invalid explicit gw_lrp_range, attempting auto-derive",
"pool", pool.Name, "start", pool.GwLrpRangeStart, "end", pool.GwLrpRangeEnd)
}
// 2. Auto-derive from subnet.
gw := net.ParseIP(pool.Gateway).To4()
if gw == nil {
return nil, nil, 0, false
}
mask := net.CIDRMask(prefix, 32)
network := gw.Mask(mask)
bcast := make(net.IP, 4)
for i := range 4 {
bcast[i] = network[i] | ^mask[i]
}
bcastU := ipv4ToUint32(bcast)
if bcastU < 17 {
return nil, nil, 0, false
}
autoEndU := bcastU - 1 // skip broadcast itself
autoStartU := bcastU - 16 // 16-IP range
// Shift below per-VM EIP range when overlap.
if pool.RangeStart != "" && pool.RangeEnd != "" {
rs := net.ParseIP(pool.RangeStart).To4()
re := net.ParseIP(pool.RangeEnd).To4()
if rs != nil && re != nil {
rsU := ipv4ToUint32(rs)
reU := ipv4ToUint32(re)
if autoEndU >= rsU && autoStartU <= reU {
if rsU < 17 {
return nil, nil, 0, false
}
autoEndU = rsU - 1
autoStartU = rsU - 16
}
}
}
// Clamp inside the subnet (network+1 .. broadcast-1) — never hand out
// the network address, the broadcast, or an off-subnet IP.
netU := ipv4ToUint32(network)
if autoStartU <= netU {
autoStartU = netU + 1
}
if autoEndU >= bcastU {
autoEndU = bcastU - 1
}
if autoStartU > autoEndU {
return nil, nil, 0, false
}
// Guard the gateway IP itself — never give a VPC the upstream nexthop.
gwU := ipv4ToUint32(gw)
if gwU >= autoStartU && gwU <= autoEndU {
// Gateway is inside the auto range — shrink to exclude it.
switch gwU {
case autoStartU:
autoStartU++
case autoEndU:
autoEndU--
default:
// Gateway in the middle: prefer the upper half.
autoStartU = gwU + 1
}
if autoStartU > autoEndU {
return nil, nil, 0, false
}
}
return uint32ToIPv4(autoStartU), uint32ToIPv4(autoEndU), prefix, true
}
// 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
// existing IP is returned (idempotent reconcile). Returns ok=false when
// the pool has no gw_lrp_range — caller falls back to link-local.
func (h *TopologyHandler) allocateGatewayLRPIP(ctx context.Context, vpcId string, pool *ExternalPoolConfig) (ip string, prefix int, ok bool, err error) {
start, end, prefix, ok := gwLrpRange(pool)
if !ok {
return "", 0, false, nil
}
gwPortName := "gw-" + vpcId
lrps, err := h.ovn.ListLogicalRouterPorts(ctx)
if err != nil {
return "", 0, false, fmt.Errorf("list LRPs for gw IP allocation: %w", err)
}
used := make(map[uint32]struct{}, len(lrps))
for _, lrp := range lrps {
existing := lrp.ExternalIDs[gatewayIPExtID]
if existing == "" {
continue
}
if lrp.Name == gwPortName {
// Idempotent: already allocated to this VPC.
return existing, prefix, true, nil
}
if v := net.ParseIP(existing).To4(); v != nil {
used[ipv4ToUint32(v)] = struct{}{}
}
}
startU := ipv4ToUint32(start)
endU := ipv4ToUint32(end)
for n := startU; n <= endU; n++ {
if _, taken := used[n]; taken {
continue
}
return uint32ToIPv4(n).String(), prefix, true, nil
}
return "", 0, false, fmt.Errorf("gw_lrp_range exhausted for pool %q (%s-%s)", pool.Name, pool.GwLrpRangeStart, pool.GwLrpRangeEnd)
}
func ipv4ToUint32(ip net.IP) uint32 {
v := ip.To4()
return uint32(v[0])<<24 | uint32(v[1])<<16 | uint32(v[2])<<8 | uint32(v[3])
}
func uint32ToIPv4(n uint32) net.IP {
return net.IPv4(byte(n>>24&0xff), byte(n>>16&0xff), byte(n>>8&0xff), byte(n&0xff)).To4()
}
// generateMAC creates a deterministic locally-administered unicast MAC
// from a resource ID via utils.HashMAC. Inputs are vpcd-owned ids
// (subnet-..., gw-vpc-..., eni-...) which are unique on their own.
func generateMAC(resourceID string) string {
return utils.HashMAC(resourceID)
}
// respond sends a simple JSON response to a NATS request.
func respond(msg *nats.Msg, err error) {
if msg.Reply == "" {
spinifex/services/vpcd/topology_manager.go
+5 -13
View File
@@ -106,16 +106,12 @@ func (h *TopologyHandler) securityGroupManager() policy.SecurityGroupManager {
}
// 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.
// the value injected via WithNATMode; 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))
h.natm, h.natmErr = policy.NewNATManager(h.ovn, h.natMode, policy.WithFlowsBarrier(waitForFlowsHV))
})
return h.natm, h.natmErr
}
@@ -129,10 +125,6 @@ func (h *TopologyHandler) igwManager() (external.IGWManager, error) {
h.igwmErr = err
return
}
mode := policy.NATModeDistributed
if h.useCentralizedNAT() {
mode = policy.NATModeCentralized
}
var poolCfg *external.ExternalPoolConfig
if p := h.findExternalPool("", ""); p != nil {
shared := external.ExternalPoolConfig{
@@ -157,7 +149,7 @@ func (h *TopologyHandler) igwManager() (external.IGWManager, error) {
Pool: poolCfg,
Allocator: external.NewStaticRangeAllocator(h.ovn),
Chassis: h.chassisNames,
NATMode: mode,
NATMode: h.natMode,
FlowsBarrier: waitForFlowsHV,
})
})
spinifex/services/vpcd/topology_test.go
+11 -712
View File
@@ -3,14 +3,13 @@ package vpcd
import (
"context"
"encoding/json"
"fmt"
"net"
"slices"
"strings"
"testing"
"time"
"github.com/mulgadc/spinifex/spinifex/network/ovn/nbdb"
"github.com/mulgadc/spinifex/spinifex/network/policy"
"github.com/mulgadc/spinifex/spinifex/network/topology"
"github.com/mulgadc/spinifex/spinifex/testutil"
"github.com/mulgadc/spinifex/spinifex/types"
@@ -380,47 +379,6 @@ func TestTopologyHandler_VPCDeleteCascade(t *testing.T) {
}
}
func TestGenerateMAC(t *testing.T) {
mac1 := generateMAC("subnet-aaa")
mac2 := generateMAC("subnet-bbb")
// Must be a valid locally-administered unicast MAC. First octet is
// hash-derived, not the literal "02:00:00" of the old 24-bit impl.
hw, err := net.ParseMAC(mac1)
if err != nil {
t.Fatalf("invalid MAC %q: %v", mac1, err)
}
if hw[0]&0x03 != 0x02 {
t.Errorf("expected unicast+LAA bits on first octet, got %#x", hw[0])
}
// Different inputs produce different MACs
if mac1 == mac2 {
t.Error("expected different MACs for different inputs")
}
// Same input produces same MAC (deterministic)
mac1b := generateMAC("subnet-aaa")
if mac1 != mac1b {
t.Error("expected deterministic MAC")
}
// Output must use full 46-bit hash space, not the old 24-bit space that
// pinned the top 3 octets to "02:00:00". Sweep many inputs and assert
// the second octet varies — old impl always emitted 0x00 there.
seen := map[byte]struct{}{}
for i := range 1000 {
hw, err := net.ParseMAC(generateMAC(fmt.Sprintf("subnet-%d", i)))
if err != nil {
t.Fatalf("invalid MAC: %v", err)
}
seen[hw[1]] = struct{}{}
}
if len(seen) <= 1 {
t.Errorf("expected second octet to vary across inputs, got %d distinct values", len(seen))
}
}
func TestTopologyHandler_CreatePort(t *testing.T) {
_, nc := startTestNATS(t)
mock := NewMockOVNClient()
@@ -959,7 +917,7 @@ func TestTopologyHandler_AddNAT_Idempotent(t *testing.T) {
_ = mock.Connect(context.Background())
ctx := context.Background()
topo := NewTopologyHandler(mock, WithBridgeMode(BridgeModeVeth))
topo := NewTopologyHandler(mock, WithNATMode(policy.NATModeCentralized))
subs, err := topo.Subscribe(nc)
require.NoError(t, err)
defer func() {
@@ -1040,7 +998,7 @@ func TestTopologyHandler_IGWAttach_WithExternalPool(t *testing.T) {
PrefixLen: 24,
},
}
topo := NewTopologyHandler(mock, WithExternalNetwork("pool", pools), WithBridgeMode(BridgeModeVeth))
topo := NewTopologyHandler(mock, WithExternalNetwork("pool", pools), WithNATMode(policy.NATModeCentralized))
subs, err := topo.Subscribe(nc)
if err != nil {
t.Fatalf("subscribe: %v", err)
@@ -1134,7 +1092,7 @@ func TestTopologyHandler_IGWAttach_PoolWithGatewayIP(t *testing.T) {
PrefixLen: 28,
},
}
topo := NewTopologyHandler(mock, WithExternalNetwork("pool", pools), WithBridgeMode(BridgeModeVeth))
topo := NewTopologyHandler(mock, WithExternalNetwork("pool", pools), WithNATMode(policy.NATModeCentralized))
subs, _ := topo.Subscribe(nc)
defer func() {
for _, s := range subs {
@@ -1197,7 +1155,7 @@ func TestTopologyHandler_IGWAttach_MultiVPC_NoIPCollision(t *testing.T) {
PrefixLen: 24,
},
}
topo := NewTopologyHandler(mock, WithExternalNetwork("pool", pools), WithBridgeMode(BridgeModeVeth))
topo := NewTopologyHandler(mock, WithExternalNetwork("pool", pools), WithNATMode(policy.NATModeCentralized))
subs, err := topo.Subscribe(nc)
if err != nil {
t.Fatalf("subscribe: %v", err)
@@ -1307,7 +1265,7 @@ func TestTopologyHandler_IGWAttach_CentralizedNATAllocatesGwLrpIP(t *testing.T)
}
topo := NewTopologyHandler(mock,
WithExternalNetwork("pool", pools),
WithBridgeMode(BridgeModeVeth),
WithNATMode(policy.NATModeCentralized),
)
subs, err := topo.Subscribe(nc)
if err != nil {
@@ -2283,7 +2241,7 @@ func TestTopologyHandler_IGWAttach_DirectBridge_NoNatAddresses(t *testing.T) {
_ = mock.Connect(context.Background())
ctx := context.Background()
topo := NewTopologyHandler(mock, WithBridgeMode(BridgeModeDirect))
topo := NewTopologyHandler(mock, WithNATMode(policy.NATModeDistributed))
subs, err := topo.Subscribe(nc)
if err != nil {
t.Fatalf("subscribe: %v", err)
@@ -2329,7 +2287,7 @@ func TestTopologyHandler_AddNAT_DirectBridge_DistributedNAT(t *testing.T) {
_ = mock.Connect(context.Background())
ctx := context.Background()
topo := NewTopologyHandler(mock, WithBridgeMode(BridgeModeDirect))
topo := NewTopologyHandler(mock, WithNATMode(policy.NATModeDistributed))
subs, err := topo.Subscribe(nc)
if err != nil {
t.Fatalf("subscribe: %v", err)
@@ -2388,7 +2346,7 @@ func TestTopologyHandler_IGWAttach_VethMode_HasNatAddresses(t *testing.T) {
_ = mock.Connect(context.Background())
ctx := context.Background()
topo := NewTopologyHandler(mock, WithBridgeMode(BridgeModeVeth))
topo := NewTopologyHandler(mock, WithNATMode(policy.NATModeCentralized))
subs, err := topo.Subscribe(nc)
if err != nil {
t.Fatalf("subscribe: %v", err)
@@ -2431,7 +2389,7 @@ func TestTopologyHandler_AddNAT_VethMode_CentralizedNAT(t *testing.T) {
_ = mock.Connect(context.Background())
ctx := context.Background()
topo := NewTopologyHandler(mock, WithBridgeMode(BridgeModeVeth))
topo := NewTopologyHandler(mock, WithNATMode(policy.NATModeCentralized))
subs, err := topo.Subscribe(nc)
if err != nil {
t.Fatalf("subscribe: %v", err)
@@ -2491,7 +2449,7 @@ func TestTopologyHandler_AddNAT_CleansStaleRulesFromOtherVPCs(t *testing.T) {
_ = mock.Connect(context.Background())
ctx := context.Background()
topo := NewTopologyHandler(mock, WithBridgeMode(BridgeModeVeth))
topo := NewTopologyHandler(mock, WithNATMode(policy.NATModeCentralized))
subs, err := topo.Subscribe(nc)
if err != nil {
t.Fatalf("subscribe: %v", err)
@@ -2568,665 +2526,6 @@ func TestTopologyHandler_AddNAT_CleansStaleRulesFromOtherVPCs(t *testing.T) {
}
}
// ensureLocalnetOptions retrofits options:nat-addresses on a localnet port
// whose mode no longer matches (e.g. created during a pre-Fix 1 run where
// detectBridgeMode returned direct because the veth had not been
// recreated after reboot). mulga-998.b Fix 3.
func TestEnsureLocalnetOptions_AddsNatAddressesInCentralizedMode(t *testing.T) {
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
_ = mock.CreateLogicalSwitch(ctx, &nbdb.LogicalSwitch{Name: "ext-vpc-x"})
_ = mock.CreateLogicalSwitchPort(ctx, "ext-vpc-x", &nbdb.LogicalSwitchPort{
Name: "ext-port-vpc-x",
Type: "localnet",
Addresses: []string{"unknown"},
Options: map[string]string{"network_name": "external"},
})
topo := NewTopologyHandler(mock, WithBridgeMode(BridgeModeVeth))
if err := topo.ensureLocalnetOptions(ctx, "ext-port-vpc-x"); err != nil {
t.Fatalf("ensureLocalnetOptions: %v", err)
}
port, _ := mock.GetLogicalSwitchPort(ctx, "ext-port-vpc-x")
if port.Options["nat-addresses"] != "router" {
t.Errorf("expected nat-addresses=router after retrofit, got %q", port.Options["nat-addresses"])
}
if port.Options["network_name"] != "external" {
t.Errorf("network_name clobbered: got %q", port.Options["network_name"])
}
}
func TestEnsureLocalnetOptions_RemovesNatAddressesInDirectMode(t *testing.T) {
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
_ = mock.CreateLogicalSwitch(ctx, &nbdb.LogicalSwitch{Name: "ext-vpc-y"})
_ = mock.CreateLogicalSwitchPort(ctx, "ext-vpc-y", &nbdb.LogicalSwitchPort{
Name: "ext-port-vpc-y",
Type: "localnet",
Addresses: []string{"unknown"},
Options: map[string]string{"network_name": "external", "nat-addresses": "router"},
})
topo := NewTopologyHandler(mock, WithBridgeMode(BridgeModeDirect))
if err := topo.ensureLocalnetOptions(ctx, "ext-port-vpc-y"); err != nil {
t.Fatalf("ensureLocalnetOptions: %v", err)
}
port, _ := mock.GetLogicalSwitchPort(ctx, "ext-port-vpc-y")
if _, ok := port.Options["nat-addresses"]; ok {
t.Errorf("nat-addresses should be removed in direct mode, got %q", port.Options["nat-addresses"])
}
if port.Options["network_name"] != "external" {
t.Errorf("network_name clobbered: got %q", port.Options["network_name"])
}
}
func TestEnsureLocalnetOptions_NoOpWhenAlreadyCorrect(t *testing.T) {
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
_ = mock.CreateLogicalSwitch(ctx, &nbdb.LogicalSwitch{Name: "ext-vpc-z"})
_ = mock.CreateLogicalSwitchPort(ctx, "ext-vpc-z", &nbdb.LogicalSwitchPort{
Name: "ext-port-vpc-z",
Type: "localnet",
Addresses: []string{"unknown"},
Options: map[string]string{"network_name": "external", "nat-addresses": "router"},
})
startUpdates := mock.UpdateLogicalSwitchPortCalls
topo := NewTopologyHandler(mock, WithBridgeMode(BridgeModeVeth))
if err := topo.ensureLocalnetOptions(ctx, "ext-port-vpc-z"); err != nil {
t.Fatalf("ensureLocalnetOptions: %v", err)
}
if mock.UpdateLogicalSwitchPortCalls != startUpdates {
t.Errorf("expected no UpdateLogicalSwitchPort call when already correct (idempotent); got %d new calls",
mock.UpdateLogicalSwitchPortCalls-startUpdates)
}
}
func TestEnsureLocalnetOptions_NoOpInDirectModeWhenAbsent(t *testing.T) {
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
_ = mock.CreateLogicalSwitch(ctx, &nbdb.LogicalSwitch{Name: "ext-vpc-d"})
_ = mock.CreateLogicalSwitchPort(ctx, "ext-vpc-d", &nbdb.LogicalSwitchPort{
Name: "ext-port-vpc-d",
Type: "localnet",
Addresses: []string{"unknown"},
Options: map[string]string{"network_name": "external"},
})
start := mock.UpdateLogicalSwitchPortCalls
topo := NewTopologyHandler(mock, WithBridgeMode(BridgeModeDirect))
if err := topo.ensureLocalnetOptions(ctx, "ext-port-vpc-d"); err != nil {
t.Fatalf("ensureLocalnetOptions: %v", err)
}
if mock.UpdateLogicalSwitchPortCalls != start {
t.Errorf("direct mode + no nat-addresses should be a no-op; got %d new calls",
mock.UpdateLogicalSwitchPortCalls-start)
}
}
func TestEnsureLocalnetOptions_HandlesNilOptionsMap(t *testing.T) {
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
_ = mock.CreateLogicalSwitch(ctx, &nbdb.LogicalSwitch{Name: "ext-vpc-n"})
_ = mock.CreateLogicalSwitchPort(ctx, "ext-vpc-n", &nbdb.LogicalSwitchPort{
Name: "ext-port-vpc-n",
Type: "localnet",
Addresses: []string{"unknown"},
})
topo := NewTopologyHandler(mock, WithBridgeMode(BridgeModeVeth))
if err := topo.ensureLocalnetOptions(ctx, "ext-port-vpc-n"); err != nil {
t.Fatalf("ensureLocalnetOptions: %v", err)
}
port, _ := mock.GetLogicalSwitchPort(ctx, "ext-port-vpc-n")
if port.Options["nat-addresses"] != "router" {
t.Errorf("expected nat-addresses=router after nil-map init, got %q", port.Options["nat-addresses"])
}
}
func TestEnsureLocalnetOptions_RestoresNetworkName(t *testing.T) {
// Operator (or test drill) cleared options entirely — retrofit must
// restore both network_name=external and nat-addresses=router.
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
_ = mock.CreateLogicalSwitch(ctx, &nbdb.LogicalSwitch{Name: "ext-vpc-r"})
_ = mock.CreateLogicalSwitchPort(ctx, "ext-vpc-r", &nbdb.LogicalSwitchPort{
Name: "ext-port-vpc-r",
Type: "localnet",
Addresses: []string{"unknown"},
Options: map[string]string{},
})
topo := NewTopologyHandler(mock, WithBridgeMode(BridgeModeVeth))
if err := topo.ensureLocalnetOptions(ctx, "ext-port-vpc-r"); err != nil {
t.Fatalf("ensureLocalnetOptions: %v", err)
}
port, _ := mock.GetLogicalSwitchPort(ctx, "ext-port-vpc-r")
if port.Options["network_name"] != "external" {
t.Errorf("expected network_name=external after retrofit, got %q", port.Options["network_name"])
}
if port.Options["nat-addresses"] != "router" {
t.Errorf("expected nat-addresses=router after retrofit, got %q", port.Options["nat-addresses"])
}
}
func TestEnsureLocalnetOptions_RestoresNetworkNameDirectMode(t *testing.T) {
// Direct mode: network_name still required, nat-addresses still absent.
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
_ = mock.CreateLogicalSwitch(ctx, &nbdb.LogicalSwitch{Name: "ext-vpc-rd"})
_ = mock.CreateLogicalSwitchPort(ctx, "ext-vpc-rd", &nbdb.LogicalSwitchPort{
Name: "ext-port-vpc-rd",
Type: "localnet",
Addresses: []string{"unknown"},
Options: map[string]string{},
})
topo := NewTopologyHandler(mock, WithBridgeMode(BridgeModeDirect))
if err := topo.ensureLocalnetOptions(ctx, "ext-port-vpc-rd"); err != nil {
t.Fatalf("ensureLocalnetOptions: %v", err)
}
port, _ := mock.GetLogicalSwitchPort(ctx, "ext-port-vpc-rd")
if port.Options["network_name"] != "external" {
t.Errorf("expected network_name=external after retrofit, got %q", port.Options["network_name"])
}
if _, ok := port.Options["nat-addresses"]; ok {
t.Errorf("direct mode should not set nat-addresses, got %q", port.Options["nat-addresses"])
}
}
func TestRetrofitAllExternalLocalnetOptions_FixesClearedOptions(t *testing.T) {
// Walks every ext-* switch in OVN (authoritative) and retrofits options
// on its localnet port. Covers VPCs whose IGW KV record is missing or
// whose options were cleared manually (mulga-998.b Drill 2).
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
// Two external switches with cleared options, one subnet switch (must be
// skipped because role != external), one external switch already correct.
for _, vpcID := range []string{"vpc-a", "vpc-b"} {
_ = mock.CreateLogicalSwitch(ctx, &nbdb.LogicalSwitch{
Name: "ext-" + vpcID,
ExternalIDs: map[string]string{"spinifex:role": "external", "spinifex:vpc_id": vpcID},
})
_ = mock.CreateLogicalSwitchPort(ctx, "ext-"+vpcID, &nbdb.LogicalSwitchPort{
Name: "ext-port-" + vpcID,
Type: "localnet",
Addresses: []string{"unknown"},
Options: map[string]string{},
})
}
// A subnet switch that must NOT be retrofitted.
_ = mock.CreateLogicalSwitch(ctx, &nbdb.LogicalSwitch{
Name: "subnet-foo",
ExternalIDs: map[string]string{"spinifex:subnet_id": "subnet-foo"},
})
// An ext switch already correct — must remain a no-op.
_ = mock.CreateLogicalSwitch(ctx, &nbdb.LogicalSwitch{
Name: "ext-vpc-ok",
ExternalIDs: map[string]string{"spinifex:role": "external", "spinifex:vpc_id": "vpc-ok"},
})
_ = mock.CreateLogicalSwitchPort(ctx, "ext-vpc-ok", &nbdb.LogicalSwitchPort{
Name: "ext-port-vpc-ok",
Type: "localnet",
Addresses: []string{"unknown"},
Options: map[string]string{"network_name": "external", "nat-addresses": "router"},
})
startUpdates := mock.UpdateLogicalSwitchPortCalls
topo := NewTopologyHandler(mock, WithBridgeMode(BridgeModeVeth))
topo.RetrofitAllExternalLocalnetOptions(ctx)
for _, vpcID := range []string{"vpc-a", "vpc-b"} {
port, err := mock.GetLogicalSwitchPort(ctx, "ext-port-"+vpcID)
if err != nil {
t.Fatalf("get port: %v", err)
}
if port.Options["network_name"] != "external" {
t.Errorf("%s: expected network_name=external, got %q", vpcID, port.Options["network_name"])
}
if port.Options["nat-addresses"] != "router" {
t.Errorf("%s: expected nat-addresses=router, got %q", vpcID, port.Options["nat-addresses"])
}
}
// Expect exactly 2 updates (vpc-a + vpc-b); vpc-ok must be idempotent no-op.
if got := mock.UpdateLogicalSwitchPortCalls - startUpdates; got != 2 {
t.Errorf("expected 2 UpdateLogicalSwitchPort calls (one per stale ext port); got %d", got)
}
}
func TestEnsureLocalnetOptions_ErrorOnMissingPort(t *testing.T) {
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
topo := NewTopologyHandler(mock, WithBridgeMode(BridgeModeVeth))
err := topo.ensureLocalnetOptions(ctx, "ext-port-missing")
if err == nil {
t.Fatal("expected error when port missing")
}
if !strings.Contains(err.Error(), "ext-port-missing") {
t.Errorf("error should name the port; got: %v", err)
}
}
// reconcileIGW is now ensure-then-skip per child with no rollback paths, so
// running against an empty NB DB must produce a fully-formed IGW topology in
// one pass.
func TestReconcileIGW_FullCreate(t *testing.T) {
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
topo := NewTopologyHandler(mock,
WithBridgeMode(BridgeModeDirect),
WithChassisNames([]string{"chassis-A"}),
)
if err := mock.CreateLogicalRouter(ctx, &nbdb.LogicalRouter{Name: "vpc-vpc-fc"}); err != nil {
t.Fatalf("seed router: %v", err)
}
if err := topo.reconcileIGW(ctx, "vpc-fc", "igw-fc"); err != nil {
t.Fatalf("reconcileIGW: %v", err)
}
if _, err := mock.GetLogicalSwitch(ctx, "ext-vpc-fc"); err != nil {
t.Errorf("ext switch missing: %v", err)
}
lsp, err := mock.GetLogicalSwitchPort(ctx, "ext-port-vpc-fc")
if err != nil {
t.Fatalf("localnet port missing: %v", err)
}
if lsp.Type != "localnet" || lsp.Options["network_name"] != "external" {
t.Errorf("localnet port misconfigured: type=%q options=%v", lsp.Type, lsp.Options)
}
if _, err := mock.GetLogicalRouterPort(ctx, "gw-vpc-fc"); err != nil {
t.Errorf("gateway LRP missing: %v", err)
}
if _, err := mock.GetLogicalSwitchPort(ctx, "gw-port-vpc-fc"); err != nil {
t.Errorf("router-LSP missing: %v", err)
}
route, err := mock.FindStaticRoute(ctx, "vpc-vpc-fc", "0.0.0.0/0")
if err != nil {
t.Fatalf("FindStaticRoute: %v", err)
}
if route == nil {
t.Errorf("default route missing")
}
gc, err := mock.GetGatewayChassisByName(ctx, "gw-vpc-fc-chassis-A")
if err != nil {
t.Fatalf("GetGatewayChassisByName: %v", err)
}
if gc == nil {
t.Errorf("gateway chassis binding missing")
}
}
// Partial state: switch + LRP + router-LSP exist but localnet LSP is missing
// (the cell-10 nightly's bootstrap topology shape). reconcileIGW must add the
// missing localnet without touching the rest.
func TestReconcileIGW_PartialState_MissingLocalnet(t *testing.T) {
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
topo := NewTopologyHandler(mock,
WithBridgeMode(BridgeModeDirect),
WithChassisNames([]string{"chassis-A"}),
)
if err := mock.CreateLogicalRouter(ctx, &nbdb.LogicalRouter{Name: "vpc-vpc-pml"}); err != nil {
t.Fatalf("seed router: %v", err)
}
if err := mock.CreateLogicalSwitch(ctx, &nbdb.LogicalSwitch{
Name: "ext-vpc-pml",
ExternalIDs: map[string]string{
"spinifex:vpc_id": "vpc-pml",
"spinifex:role": "external",
},
}); err != nil {
t.Fatalf("seed ext switch: %v", err)
}
if err := mock.CreateLogicalRouterPort(ctx, "vpc-vpc-pml", &nbdb.LogicalRouterPort{
Name: "gw-vpc-pml",
MAC: "02:00:00:aa:bb:01",
Networks: []string{"169.254.0.1/30"},
ExternalIDs: map[string]string{"spinifex:vpc_id": "vpc-pml", "spinifex:role": "gateway"},
}); err != nil {
t.Fatalf("seed LRP: %v", err)
}
if err := mock.CreateLogicalSwitchPort(ctx, "ext-vpc-pml", &nbdb.LogicalSwitchPort{
Name: "gw-port-vpc-pml",
Type: "router",
Addresses: []string{"router"},
Options: map[string]string{"router-port": "gw-vpc-pml"},
ExternalIDs: map[string]string{"spinifex:vpc_id": "vpc-pml"},
}); err != nil {
t.Fatalf("seed router-LSP: %v", err)
}
if err := topo.reconcileIGW(ctx, "vpc-pml", "igw-pml"); err != nil {
t.Fatalf("reconcileIGW: %v", err)
}
lsp, err := mock.GetLogicalSwitchPort(ctx, "ext-port-vpc-pml")
if err != nil {
t.Fatalf("localnet port still missing after self-heal: %v", err)
}
if lsp.Type != "localnet" {
t.Errorf("expected type=localnet, got %q", lsp.Type)
}
if lsp.Options["network_name"] != "external" {
t.Errorf("expected network_name=external, got %q", lsp.Options["network_name"])
}
}
// Partial state: switch + localnet exist but the gateway LRP and router-LSP
// are missing. reconcileIGW must finish what a previously-failed pass started.
func TestReconcileIGW_PartialState_MissingLRP(t *testing.T) {
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
topo := NewTopologyHandler(mock,
WithBridgeMode(BridgeModeDirect),
WithChassisNames([]string{"chassis-A"}),
)
if err := mock.CreateLogicalRouter(ctx, &nbdb.LogicalRouter{Name: "vpc-vpc-pmr"}); err != nil {
t.Fatalf("seed router: %v", err)
}
if err := mock.CreateLogicalSwitch(ctx, &nbdb.LogicalSwitch{
Name: "ext-vpc-pmr",
ExternalIDs: map[string]string{"spinifex:vpc_id": "vpc-pmr", "spinifex:role": "external"},
}); err != nil {
t.Fatalf("seed ext switch: %v", err)
}
if err := mock.CreateLogicalSwitchPort(ctx, "ext-vpc-pmr", &nbdb.LogicalSwitchPort{
Name: "ext-port-vpc-pmr",
Type: "localnet",
Addresses: []string{"unknown"},
Options: map[string]string{"network_name": "external"},
ExternalIDs: map[string]string{"spinifex:vpc_id": "vpc-pmr"},
}); err != nil {
t.Fatalf("seed localnet: %v", err)
}
if err := topo.reconcileIGW(ctx, "vpc-pmr", "igw-pmr"); err != nil {
t.Fatalf("reconcileIGW: %v", err)
}
if _, err := mock.GetLogicalRouterPort(ctx, "gw-vpc-pmr"); err != nil {
t.Errorf("gateway LRP not created: %v", err)
}
if _, err := mock.GetLogicalSwitchPort(ctx, "gw-port-vpc-pmr"); err != nil {
t.Errorf("router-LSP not created: %v", err)
}
gc, err := mock.GetGatewayChassisByName(ctx, "gw-vpc-pmr-chassis-A")
if err != nil {
t.Fatalf("GetGatewayChassisByName: %v", err)
}
if gc == nil {
t.Errorf("gateway chassis binding missing after heal")
}
}
// Full pre-existing topology: reconcileIGW must be a no-op (no errors, no new
// entities, no duplicate default route). This is the steady-state self-heal
// path that runs on every vpcd start.
func TestReconcileIGW_FullExisting(t *testing.T) {
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
topo := NewTopologyHandler(mock,
WithBridgeMode(BridgeModeDirect),
WithChassisNames([]string{"chassis-A"}),
)
if err := mock.CreateLogicalRouter(ctx, &nbdb.LogicalRouter{Name: "vpc-vpc-fe"}); err != nil {
t.Fatalf("seed router: %v", err)
}
// First pass creates the topology end-to-end.
if err := topo.reconcileIGW(ctx, "vpc-fe", "igw-fe"); err != nil {
t.Fatalf("first reconcileIGW: %v", err)
}
extBefore, err := mock.GetLogicalSwitch(ctx, "ext-vpc-fe")
if err != nil {
t.Fatalf("ext switch missing after first pass: %v", err)
}
lrBefore, err := mock.GetLogicalRouter(ctx, "vpc-vpc-fe")
if err != nil {
t.Fatalf("router fetch: %v", err)
}
if got := len(lrBefore.StaticRoutes); got != 1 {
t.Fatalf("first pass: expected 1 default route, got %d", got)
}
// Second pass against the now-complete topology.
if err := topo.reconcileIGW(ctx, "vpc-fe", "igw-fe"); err != nil {
t.Fatalf("second reconcileIGW: %v", err)
}
extAfter, err := mock.GetLogicalSwitch(ctx, "ext-vpc-fe")
if err != nil {
t.Fatalf("ext switch fetch: %v", err)
}
if extAfter.UUID != extBefore.UUID {
t.Errorf("ext switch UUID changed (was %q, now %q) — second pass must not recreate", extBefore.UUID, extAfter.UUID)
}
lrAfter, err := mock.GetLogicalRouter(ctx, "vpc-vpc-fe")
if err != nil {
t.Fatalf("router fetch: %v", err)
}
if got := len(lrAfter.StaticRoutes); got != 1 {
t.Errorf("second pass: expected 1 default route, got %d", got)
}
}
// AddStaticRoute is non-idempotent on its own — without FindStaticRoute, every
// retry leaves a fresh duplicate 0.0.0.0/0 row. Pre-seed a complete topology
// (exercising the path where reconcileIGW finds an existing route) and verify
// no duplicate is added.
func TestReconcileIGW_DuplicateRoute(t *testing.T) {
mock := NewMockOVNClient()
_ = mock.Connect(context.Background())
ctx := context.Background()
topo := NewTopologyHandler(mock,
WithBridgeMode(BridgeModeDirect),
WithChassisNames([]string{"chassis-A"}),
)
if err := mock.CreateLogicalRouter(ctx, &nbdb.LogicalRouter{Name: "vpc-vpc-dr"}); err != nil {
t.Fatalf("seed router: %v", err)
}
if err := topo.reconcileIGW(ctx, "vpc-dr", "igw-dr"); err != nil {
t.Fatalf("first reconcileIGW: %v", err)
}
// Multiple subsequent passes must keep the route count at exactly 1.
for i := range 3 {
if err := topo.reconcileIGW(ctx, "vpc-dr", "igw-dr"); err != nil {
t.Fatalf("reconcileIGW pass %d: %v", i+2, err)
}
lr, err := mock.GetLogicalRouter(ctx, "vpc-vpc-dr")
if err != nil {
t.Fatalf("router fetch pass %d: %v", i+2, err)
}
if got := len(lr.StaticRoutes); got != 1 {
t.Fatalf("pass %d: expected 1 default route, got %d", i+2, got)
}
}
}
// TestGwLrpRange exercises the gateway LRP IP range derivation. mulga-siv-36
// dropped the link-local fallback in centralized NAT — auto-derive must yield
// an in-subnet range whenever the operator did not configure one explicitly,
// or fail loudly so the IGW attach surfaces the misconfiguration.
func TestGwLrpRange(t *testing.T) {
cases := []struct {
name string
pool *ExternalPoolConfig
wantOK bool
wantStart string
wantEnd string
wantPfx int
}{
{
name: "nil pool",
pool: nil,
wantOK: false,
},
{
name: "explicit range wins over auto",
pool: &ExternalPoolConfig{
Gateway: "192.168.1.1",
PrefixLen: 24,
RangeStart: "192.168.1.100",
RangeEnd: "192.168.1.110",
GwLrpRangeStart: "192.168.1.50",
GwLrpRangeEnd: "192.168.1.59",
},
wantOK: true,
wantStart: "192.168.1.50",
wantEnd: "192.168.1.59",
wantPfx: 24,
},
{
name: "explicit range invalid falls through to auto",
pool: &ExternalPoolConfig{
Gateway: "192.168.1.1",
PrefixLen: 24,
RangeStart: "192.168.1.100",
RangeEnd: "192.168.1.110",
GwLrpRangeStart: "not-an-ip",
GwLrpRangeEnd: "192.168.1.59",
},
wantOK: true,
wantStart: "192.168.1.239",
wantEnd: "192.168.1.254",
wantPfx: 24,
},
{
name: "auto-derive /24 default end of subnet",
pool: &ExternalPoolConfig{
Gateway: "192.168.1.1",
PrefixLen: 24,
RangeStart: "192.168.1.100",
RangeEnd: "192.168.1.110",
},
wantOK: true,
wantStart: "192.168.1.239",
wantEnd: "192.168.1.254",
wantPfx: 24,
},
{
name: "auto-derive overlap shifts below RangeStart",
pool: &ExternalPoolConfig{
Gateway: "192.168.1.1",
PrefixLen: 24,
RangeStart: "192.168.1.245",
RangeEnd: "192.168.1.254",
},
wantOK: true,
wantStart: "192.168.1.229",
wantEnd: "192.168.1.244",
wantPfx: 24,
},
{
name: "auto-derive guards gateway at start",
pool: &ExternalPoolConfig{
Gateway: "192.168.1.240",
PrefixLen: 24,
},
wantOK: true,
wantStart: "192.168.1.241",
wantEnd: "192.168.1.254",
wantPfx: 24,
},
{
name: "auto-derive guards gateway at end",
pool: &ExternalPoolConfig{
Gateway: "192.168.1.254",
PrefixLen: 24,
},
wantOK: true,
wantStart: "192.168.1.239",
wantEnd: "192.168.1.253",
wantPfx: 24,
},
{
name: "auto-derive guards gateway in middle",
pool: &ExternalPoolConfig{
Gateway: "192.168.1.245",
PrefixLen: 24,
},
wantOK: true,
wantStart: "192.168.1.246",
wantEnd: "192.168.1.254",
wantPfx: 24,
},
{
name: "default prefix 24 when missing",
pool: &ExternalPoolConfig{
Gateway: "10.0.0.1",
},
wantOK: true,
wantStart: "10.0.0.239",
wantEnd: "10.0.0.254",
wantPfx: 24,
},
{
name: "unparseable gateway returns ok=false",
pool: &ExternalPoolConfig{
Gateway: "not-an-ip",
PrefixLen: 24,
},
wantOK: false,
},
{
name: "/28 with full pool exhausts auto range",
pool: &ExternalPoolConfig{
Gateway: "203.0.113.1",
PrefixLen: 28,
RangeStart: "203.0.113.2",
RangeEnd: "203.0.113.14",
},
wantOK: false,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
start, end, pfx, ok := gwLrpRange(tc.pool)
if ok != tc.wantOK {
t.Fatalf("ok=%v, want %v", ok, tc.wantOK)
}
if !ok {
return
}
if start.String() != tc.wantStart {
t.Errorf("start=%s, want %s", start, tc.wantStart)
}
if end.String() != tc.wantEnd {
t.Errorf("end=%s, want %s", end, tc.wantEnd)
}
if pfx != tc.wantPfx {
t.Errorf("prefix=%d, want %d", pfx, tc.wantPfx)
}
})
}
}
// --- Phase 4: SG enforcement port-group membership ---
// setupPortSGFixture stands up a switch + DHCP options + the port groups for
spinifex/services/vpcd/vpcd.go
+13 -15
View File
@@ -438,7 +438,12 @@ func launchService(cfg *Config) error {
}
topoOpts = append(topoOpts, WithChassisNames(chassisNames))
slog.Info("vpcd: gateway chassis discovered", "chassis", chassisNames)
topoOpts = append(topoOpts, WithBridgeMode(bridgeMode))
uplinkMode := host.UplinkModePhysical
if bridgeMode == BridgeModeVeth {
uplinkMode = host.UplinkModeVeth
}
natMode := policy.NATModeFromUplinkMode(uplinkMode)
topoOpts = append(topoOpts, WithNATMode(natMode))
topo := NewTopologyHandler(liveClient, topoOpts...)
// Elect a single vpcd to run startup reconcile. Without this, N vpcds in a
@@ -467,17 +472,10 @@ func launchService(cfg *Config) error {
return err
}
// Construct the network/reconcile manager stack. The reconciler replaces
// the five-pass startup sequence (Reconcile + ReconcileFromKV +
// ReconcileSGsOnce + RetrofitAllExternalLocalnetOptions +
// RetrofitAllGatewayPortNetworks) with a single intent-driven pass that
// also runs on a 5-minute drift ticker (replaces the 30s ReconcileSGsLoop).
uplinkMode := host.UplinkModePhysical
if bridgeMode == BridgeModeVeth {
uplinkMode = host.UplinkModeVeth
}
natMode := policy.NATModeFromUplinkMode(uplinkMode)
// Construct the network/reconcile manager stack. The reconciler is the
// single intent-driven pass that runs once at startup (leader-gated) and
// then on a 5-minute drift ticker; together they replace the five
// separate passes the old vpcd ran on boot (mulga-siv-125, Phase 2).
sgMgr := policy.NewSecurityGroupManager(liveClient)
natMgr, err := policy.NewNATManager(liveClient, natMode, policy.WithFlowsBarrier(waitForFlowsHV))
if err != nil {
@@ -522,9 +520,9 @@ func launchService(cfg *Config) error {
// Startup reconcile: leader-gated read of NATS KV intent state, applied
// once against OVN NB DB. The drift loop below handles the recovery case
// (KV not yet populated when this fires — daemon's EnsureDefaultVPC may
// race with vpcd's startup). Subscribers above keep handling per-event
// NATS messages via the legacy TopologyHandler path; phase 4 migrates
// those onto the new managers.
// race with vpcd's startup). Per-event NATS subscribers (TopologyHandler)
// already route through the same network/* managers, so the startup pass
// and the runtime fast path share one implementation.
if isLeader {
intent, intentErr := reconcile.LoadIntentFromKV(ctx, js, cfg.AZ)
if intentErr != nil {