From 46f953698fcc5afa72cecb6d0d5b24c255d27994 Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Thu, 20 Aug 2026 13:34:39 +1000 Subject: [PATCH] fix: requeue an incomplete reconcile pass instead of waiting a drift interval (#840) --- .../dhcp/acquire_budget_internal_test.go | 69 +++++ spinifex/network/external/dhcp/manager.go | 58 ++++- spinifex/network/reconcile/apply.go | 66 +++-- spinifex/network/reconcile/apply_test.go | 124 ++++++++- spinifex/network/reconcile/drift.go | 71 ++++- spinifex/network/reconcile/drift_test.go | 246 ++++++++++++++++++ spinifex/network/reconcile/lock.go | 76 +++++- spinifex/network/reconcile/lock_test.go | 52 ++++ spinifex/network/reconcile/reconcile.go | 96 +++++-- spinifex/vpcd/vpcd.go | 9 +- 10 files changed, 806 insertions(+), 61 deletions(-) create mode 100644 spinifex/network/reconcile/drift_test.go diff --git a/spinifex/network/external/dhcp/acquire_budget_internal_test.go b/spinifex/network/external/dhcp/acquire_budget_internal_test.go index ffcf71733..576512659 100644 --- a/spinifex/network/external/dhcp/acquire_budget_internal_test.go +++ b/spinifex/network/external/dhcp/acquire_budget_internal_test.go @@ -28,3 +28,72 @@ func TestAcquireLadderFitsInsideClientReadTimeout(t *testing.T) { assert.Greater(t, defaultRPCTimeout, defaultAcquireBudget, "the RPC must outlast the acquire, or a slow DORA reports a timeout instead of its own error") } + +// newBackoffTestManager builds a Manager for schedule selection only; it never +// DORAs, so a bare fake client and store suffice. +func newBackoffTestManager(t *testing.T, cfg ManagerConfig) *Manager { + t.Helper() + cfg.Client = &Fake{} + cfg.Store = &Store{} + m, err := NewManager(cfg) + assert.NoError(t, err) + return m +} + +// A failed gw-lrp acquire costs the whole VPC its external gateway until a +// reconcile pass returns, and nothing external times that DORA — the allocator +// calls the manager in-process. A failed ENI or EIP acquire blocks an AWS client +// that must hear back inside its read timeout. The purposes must not share a +// budget. +func TestAcquireBackoffIsPerPurpose(t *testing.T) { + m := newBackoffTestManager(t, ManagerConfig{}) + + gwSchedule, gwBudget := m.acquireBackoffFor(PurposeGatewayLRP) + assert.Equal(t, gwLRPAcquireSchedule, gwSchedule) + assert.Equal(t, gwLRPAcquireBudget, gwBudget) + + for _, purpose := range []string{PurposeEIP, PurposeENIPublic, PurposeNATGWExternal, ""} { + schedule, budget := m.acquireBackoffFor(purpose) + assert.Equal(t, defaultAcquireSchedule, schedule, + "purpose %q must keep the default ladder: its caller is an AWS client that will retry untokened", purpose) + assert.Equal(t, defaultAcquireBudget, budget, + "purpose %q must keep the default budget", purpose) + } +} + +// The gw-lrp window has to be wider than the default, or the split buys nothing: +// the ladder was shortened for the EIP path alone and took gw-lrp down with it. +func TestGatewayLRPAcquireWindowExceedsDefault(t *testing.T) { + ladder := func(schedule []time.Duration) time.Duration { + total := time.Duration(0) + for _, d := range schedule { + total += d + acquireAttemptJitter + } + return total + } + assert.Greater(t, ladder(gwLRPAcquireSchedule), ladder(defaultAcquireSchedule), + "the gw-lrp ladder must outlast the default one") + assert.Greater(t, gwLRPAcquireBudget, defaultAcquireBudget, + "the gw-lrp budget must outlast the default one") + assert.GreaterOrEqual(t, gwLRPAcquireBudget, ladder(gwLRPAcquireSchedule), + "the budget must leave room for the full ladder plus jitter, or later attempts never run") +} + +// Both pairs stay configurable, so an operator on a slow upstream can widen +// either without a rebuild. +func TestAcquireBackoffConfigOverrides(t *testing.T) { + m := newBackoffTestManager(t, ManagerConfig{ + AcquireSchedule: []time.Duration{time.Second}, + AcquireBudget: 2 * time.Second, + GatewayLRPAcquireSchedule: []time.Duration{3 * time.Second}, + GatewayLRPAcquireBudget: 4 * time.Second, + }) + + gwSchedule, gwBudget := m.acquireBackoffFor(PurposeGatewayLRP) + assert.Equal(t, []time.Duration{3 * time.Second}, gwSchedule) + assert.Equal(t, 4*time.Second, gwBudget) + + schedule, budget := m.acquireBackoffFor(PurposeEIP) + assert.Equal(t, []time.Duration{time.Second}, schedule) + assert.Equal(t, 2*time.Second, budget) +} diff --git a/spinifex/network/external/dhcp/manager.go b/spinifex/network/external/dhcp/manager.go index 92d280346..faf8720fd 100644 --- a/spinifex/network/external/dhcp/manager.go +++ b/spinifex/network/external/dhcp/manager.go @@ -38,6 +38,19 @@ var defaultAcquireSchedule = []time.Duration{4 * time.Second, 8 * time.Second, 1 // the daemon room to answer inside a 60s client read timeout, per the schedule. const defaultAcquireBudget = 45 * time.Second +// gwLRPAcquireSchedule is the per-attempt ladder for purpose=gw-lrp. Longer than +// the default because no client is timing this DORA: AttachInternetGateway +// publishes vpc.igw-attach fire-and-forget and the allocator calls the manager +// in-process, so the read-timeout ceiling that bounds the EIP path does not +// apply. The two purposes split on that constraint, not only on cost of failure. +var gwLRPAcquireSchedule = []time.Duration{4 * time.Second, 8 * time.Second, 16 * time.Second, 32 * time.Second} + +// gwLRPAcquireBudget caps the wallclock for a gw-lrp DORA loop, with room over +// the ladder for jitter. A lost gw-lrp lease costs the whole VPC its external +// gateway until a reconcile pass returns, so it is worth waiting out a stall an +// ENI or EIP acquire would rightly abandon. +const gwLRPAcquireBudget = 90 * time.Second + // acquireAttemptJitter is the ± window applied to each per-attempt // timeout so concurrent acquires across vpcds don't synchronise. const acquireAttemptJitter = time.Second @@ -51,6 +64,10 @@ type ManagerConfig struct { Now func() time.Time AcquireSchedule []time.Duration AcquireBudget time.Duration + // GatewayLRPAcquireSchedule/Budget override the outer DORA backoff for + // purpose=gw-lrp only; zero values fall back to the gw-lrp defaults. + GatewayLRPAcquireSchedule []time.Duration + GatewayLRPAcquireBudget time.Duration // IfaceIPs lists the IPs bound to a named interface; tests override. // Used to detect MAC-keyed upstream routers on interface-MAC pools. IfaceIPs func(iface string) ([]net.IP, error) @@ -68,6 +85,8 @@ type Manager struct { now func() time.Time acquireSchedule []time.Duration acquireBudget time.Duration + gwLRPSchedule []time.Duration + gwLRPBudget time.Duration ifaceIPs func(iface string) ([]net.IP, error) nodeName string @@ -121,6 +140,14 @@ func NewManager(cfg ManagerConfig) (*Manager, error) { if budget <= 0 { budget = defaultAcquireBudget } + gwSchedule := cfg.GatewayLRPAcquireSchedule + if len(gwSchedule) == 0 { + gwSchedule = gwLRPAcquireSchedule + } + gwBudget := cfg.GatewayLRPAcquireBudget + if gwBudget <= 0 { + gwBudget = gwLRPAcquireBudget + } ifaceIPs := cfg.IfaceIPs if ifaceIPs == nil { ifaceIPs = defaultIfaceIPs @@ -136,6 +163,8 @@ func NewManager(cfg ManagerConfig) (*Manager, error) { now: now, acquireSchedule: schedule, acquireBudget: budget, + gwLRPSchedule: gwSchedule, + gwLRPBudget: gwBudget, ifaceIPs: ifaceIPs, nodeName: nodeName, loops: map[string]*leaseLoop{}, @@ -408,7 +437,7 @@ func (m *Manager) run(ctx context.Context, self *leaseLoop, e Entry, reaffirm bo } func (m *Manager) doAcquire(ctx context.Context, e *Entry) error { - lease, err := m.acquireWithBackoff(ctx, AcquireRequest{ + lease, err := m.acquireWithBackoff(ctx, e.Purpose, AcquireRequest{ Bridge: e.Lease.Bridge, ClientID: e.Lease.ClientID, Hostname: e.Lease.Hostname, @@ -456,16 +485,27 @@ func (m *Manager) applyLease(ctx context.Context, e *Entry, fresh *Lease) error return nil } +// acquireBackoffFor returns the per-attempt ladder and wallclock budget for +// purpose. gw-lrp gets its own, longer pair; every other purpose shares the +// default one. +func (m *Manager) acquireBackoffFor(purpose string) ([]time.Duration, time.Duration) { + if purpose == PurposeGatewayLRP { + return m.gwLRPSchedule, m.gwLRPBudget + } + return m.acquireSchedule, m.acquireBudget +} + // acquireWithBackoff drives client.Acquire with per-attempt timeouts from -// the schedule, capped by AcquireBudget. Jitter prevents synchronised wakes. +// purpose's schedule, capped by its budget. Jitter prevents synchronised wakes. // ctx.Err() short-circuits; returns last error when all attempts fail. -func (m *Manager) acquireWithBackoff(ctx context.Context, req AcquireRequest) (*Lease, error) { - if len(m.acquireSchedule) == 0 { +func (m *Manager) acquireWithBackoff(ctx context.Context, purpose string, req AcquireRequest) (*Lease, error) { + schedule, budget := m.acquireBackoffFor(purpose) + if len(schedule) == 0 { return m.client.Acquire(ctx, req) } - deadline := m.now().Add(m.acquireBudget) + deadline := m.now().Add(budget) var lastErr error - for i, base := range m.acquireSchedule { + for i, base := range schedule { if ctx.Err() != nil { return nil, ctx.Err() } @@ -493,13 +533,13 @@ func (m *Manager) acquireWithBackoff(ctx context.Context, req AcquireRequest) (* } lastErr = err slog.Warn("dhcp manager: acquire attempt failed", - "client_id", req.ClientID, "attempt", i+1, "of", len(m.acquireSchedule), + "client_id", req.ClientID, "purpose", purpose, "attempt", i+1, "of", len(schedule), "timeout_ms", otelsetup.Millis(attempt), "err", err) } if lastErr == nil { lastErr = errors.New("acquire budget exhausted before first attempt") } - return nil, fmt.Errorf("acquire after %d attempts: %w", len(m.acquireSchedule), lastErr) + return nil, fmt.Errorf("acquire after %d attempts: %w", len(schedule), lastErr) } func (m *Manager) doRenew(ctx context.Context, e *Entry) error { @@ -625,7 +665,7 @@ func (m *Manager) acquireLocked(ctx context.Context, req acquireWireRequest) (*E if vendorClass == "" { vendorClass = m.leaseVendorClass() } - lease, err := m.acquireWithBackoff(ctx, AcquireRequest{ + lease, err := m.acquireWithBackoff(ctx, req.Purpose, AcquireRequest{ Bridge: req.Bridge, ClientID: req.ClientID, Hostname: hostname, diff --git a/spinifex/network/reconcile/apply.go b/spinifex/network/reconcile/apply.go index 5db4c4fe3..76aa9cd2f 100644 --- a/spinifex/network/reconcile/apply.go +++ b/spinifex/network/reconcile/apply.go @@ -102,7 +102,7 @@ func eniIDFromPort(lspName string) string { // applyVPCs ensures every intent VPC has a LogicalRouter. Stray OVN-only // routers are left alone. -func (r *reconciler) applyVPCs(ctx context.Context, intent IntentState, actual ActualState) { +func (r *reconciler) applyVPCs(ctx context.Context, intent IntentState, actual ActualState, res *passResult) { for vpcID, spec := range intent.VPCs { routerName := topology.VPCRouter(vpcID) if _, ok := actual.Routers[routerName]; !ok { @@ -115,6 +115,7 @@ func (r *reconciler) applyVPCs(ctx context.Context, intent IntentState, actual A } if _, _, err := r.ovn.EnsureLogicalRouter(ctx, lr); err != nil { slog.Error("reconcile/apply: ensure VPC router failed", "vpc_id", vpcID, "err", err) + res.fail(classVPC, vpcID, err) continue } actual.Routers[routerName] = struct{}{} @@ -125,7 +126,7 @@ func (r *reconciler) applyVPCs(ctx context.Context, intent IntentState, actual A // applySubnets ensures every intent subnet has a LogicalSwitch, LRP, router LSP, // and DHCPOptions row. Each step is idempotent. -func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actual ActualState) { +func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actual ActualState, res *passResult) { for subnetID, spec := range intent.Subnets { switchName := topology.SubnetSwitch(subnetID) routerName := topology.VPCRouter(spec.VPCID) @@ -135,6 +136,7 @@ func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actua gwIP, prefixBits, err := topology.SubnetGatewayCIDR(spec.CIDR) if err != nil { slog.Error("reconcile/apply: subnet gateway calc failed", "subnet_id", subnetID, "err", err) + res.fail(classSubnet, subnetID, err) continue } gwCIDRString := fmt.Sprintf("%s/%d", gwIP, prefixBits) @@ -150,6 +152,7 @@ func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actua } if _, _, err := r.ovn.EnsureLogicalSwitch(ctx, ls); err != nil { slog.Error("reconcile/apply: ensure subnet switch failed", "subnet_id", subnetID, "err", err) + res.fail(classSubnet, subnetID, err) continue } actual.Switches[switchName] = struct{}{} @@ -167,6 +170,7 @@ func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actua } if err := r.ovn.CreateLogicalRouterPort(ctx, routerName, lrp); err != nil { slog.Error("reconcile/apply: create subnet LRP failed", "subnet_id", subnetID, "err", err) + res.fail(classSubnet, subnetID, err) continue } actual.RouterPorts[routerPortName] = struct{}{} @@ -187,6 +191,7 @@ func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actua } if err := r.ovn.CreateLogicalSwitchPort(ctx, switchName, lsp); err != nil { slog.Error("reconcile/apply: create subnet router-LSP failed", "subnet_id", subnetID, "err", err) + res.fail(classSubnet, subnetID, err) continue } } @@ -205,6 +210,7 @@ func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actua } if _, dErr := r.ovn.CreateDHCPOptions(ctx, opts); dErr != nil { slog.Warn("reconcile/apply: create DHCP options failed (non-fatal)", "subnet_id", subnetID, "err", dErr) + res.fail(classSubnet, subnetID, dErr) } case !maps.Equal(existing.Options, want): // Converge, don't leave create-time values in place. Toggling @@ -212,6 +218,7 @@ func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actua // the wide figure on an encrypted path blackholes large segments. if dErr := r.ovn.UpdateDHCPOptionsOptions(ctx, existing.UUID, want); dErr != nil { slog.Warn("reconcile/apply: update DHCP options failed (non-fatal)", "subnet_id", subnetID, "err", dErr) + res.fail(classSubnet, subnetID, dErr) } else { slog.Info("reconcile/apply: converged DHCP options", "subnet_id", subnetID, "mtu", want["mtu"]) } @@ -225,15 +232,17 @@ func (r *reconciler) applySubnets(ctx context.Context, intent IntentState, actua // applySGs ensures every intent SG has a port group; when pruneOrphans is true, // deletes sg_* PGs with no matching intent SG. Startup passes false to avoid // deleting in-flight resources before peer subscribers have converged. -func (r *reconciler) applySGs(ctx context.Context, intent IntentState, actual ActualState, pruneOrphans bool) { +func (r *reconciler) applySGs(ctx context.Context, intent IntentState, actual ActualState, pruneOrphans bool, res *passResult) { for groupID, spec := range intent.SGs { if err := r.topology.EnsureSGPortGroup(ctx, groupID); err != nil { slog.Error("reconcile/apply: EnsureSGPortGroup failed", "sg", groupID, "err", err) + res.fail(classSG, groupID, err) continue } actual.PortGroups[topology.SecurityGroupPortGroup(groupID)] = struct{}{} if err := r.sg.EnsureSG(ctx, spec); err != nil { slog.Error("reconcile/apply: EnsureSG failed", "sg", groupID, "err", err) + res.fail(classSG, groupID, err) } } @@ -254,6 +263,7 @@ func (r *reconciler) applySGs(ctx context.Context, intent IntentState, actual Ac } if err := r.topology.DeleteSGPortGroupByName(ctx, pgName); err != nil { slog.Warn("reconcile/apply: orphan DeleteSGPortGroupByName failed", "pg", pgName, "err", err) + res.fail(classSG, pgName, err) continue } delete(actual.PortGroups, pgName) @@ -265,7 +275,7 @@ func (r *reconciler) applySGs(ctx context.Context, intent IntentState, actual Ac // SGIDs. Existing ports use diff-based UpdatePortGroupMemberships to avoid gaps. // When pruneOrphans is true, ENI LSPs with no matching intent ENI are torn down; // startup passes false so in-flight ports survive until subscribers converge. -func (r *reconciler) applyPorts(ctx context.Context, intent IntentState, actual ActualState, pruneOrphans bool) { +func (r *reconciler) applyPorts(ctx context.Context, intent IntentState, actual ActualState, pruneOrphans bool, res *passResult) { for portID, spec := range intent.Ports { portName := topology.Port(portID) switchName := topology.SubnetSwitch(spec.SubnetID) @@ -297,6 +307,7 @@ func (r *reconciler) applyPorts(ctx context.Context, intent IntentState, actual } if err := r.ovn.CreateLogicalSwitchPortInGroups(ctx, switchName, lsp, desiredPGs); err != nil { slog.Error("reconcile/apply: create ENI port failed", "port", portName, "err", err) + res.fail(classPort, portName, err) } continue } @@ -304,6 +315,7 @@ func (r *reconciler) applyPorts(ctx context.Context, intent IntentState, actual currentPGs, err := r.ovn.ListPortGroupsForPort(ctx, portName) if err != nil { slog.Warn("reconcile/apply: list port groups for port failed", "port", portName, "err", err) + res.fail(classPort, portName, err) continue } addPGs, removePGs := diffSets(desiredPGs, currentPGs) @@ -312,23 +324,25 @@ func (r *reconciler) applyPorts(ctx context.Context, intent IntentState, actual } if err := r.ovn.UpdatePortGroupMemberships(ctx, portName, addPGs, removePGs); err != nil { slog.Warn("reconcile/apply: update port group memberships failed", "port", portName, "err", err) + res.fail(classPort, portName, err) } } if !pruneOrphans { return } - r.pruneOrphanPorts(ctx, intent) + r.pruneOrphanPorts(ctx, intent, res) } // pruneOrphanPorts deletes guest LSPs whose spinifex:eni_id has no matching // intent ENI, closing the create-only gap that leaks ports across instance // terminate and host reinstall. DeletePort clears PG memberships then removes // the LSP (composed cascade). -func (r *reconciler) pruneOrphanPorts(ctx context.Context, intent IntentState) { +func (r *reconciler) pruneOrphanPorts(ctx context.Context, intent IntentState, res *passResult) { lsps, err := r.ovn.ListLogicalSwitchPorts(ctx) if err != nil { slog.Warn("reconcile/apply: list LSPs for orphan prune failed", "err", err) + res.fail(classPort, "orphan-prune", err) return } for i := range lsps { @@ -342,6 +356,7 @@ func (r *reconciler) pruneOrphanPorts(ctx context.Context, intent IntentState) { spec := topology.PortSpec{PortID: eniID, SubnetID: lsps[i].ExternalIDs["spinifex:subnet_id"]} if err := r.topology.DeletePort(ctx, spec); err != nil { slog.Warn("reconcile/apply: orphan ENI DeletePort failed", "port", lsps[i].Name, "err", err) + res.fail(classPort, lsps[i].Name, err) continue } slog.Info("reconcile/apply: removed orphan ENI port", "port", lsps[i].Name, "eni_id", eniID) @@ -352,14 +367,15 @@ func (r *reconciler) pruneOrphanPorts(ctx context.Context, intent IntentState) { // existing IGWs. AttachIGW is idempotent and must run even when the gateway // switch port already exists: its already-attached path re-ensures host state // (routed-NAT ingress routes) that survives in OVN but not across reboots. -func (r *reconciler) applyIGWs(ctx context.Context, intent IntentState, actual ActualState) { +func (r *reconciler) applyIGWs(ctx context.Context, intent IntentState, actual ActualState, res *passResult) { for vpcID, spec := range intent.IGWs { if err := r.igw.AttachIGW(ctx, spec); err != nil { slog.Error("reconcile/apply: AttachIGW failed", "vpc_id", vpcID, "err", err) + res.fail(classIGW, vpcID, err) continue } actual.ExternalSwch[vpcID] = struct{}{} - r.rebindGatewayChassis(ctx, vpcID, eipProbeIP(intent, vpcID)) + r.rebindGatewayChassis(ctx, vpcID, eipProbeIP(intent, vpcID), res) } } @@ -377,19 +393,23 @@ func eipProbeIP(intent IntentState, vpcID string) string { } // rebindGatewayChassis re-asserts chassis priority tuples on the gateway LRP. -func (r *reconciler) rebindGatewayChassis(ctx context.Context, vpcID, eipIP string) { +func (r *reconciler) rebindGatewayChassis(ctx context.Context, vpcID, eipIP string, res *passResult) { if len(r.chassis) == 0 { return } gwPortName := topology.GatewayRouterPort(vpcID) lrp, err := r.ovn.GetLogicalRouterPort(ctx, gwPortName) if err != nil { + slog.Warn("reconcile/apply: gateway LRP read failed; skipping chassis rebind and datapath gate", + "vpc_id", vpcID, "port", gwPortName, "err", err) + res.fail(classIGW, vpcID, err) return } for i, chassis := range r.chassis { priority := max(20-(i*5), 1) if err := r.ovn.SetGatewayChassis(ctx, gwPortName, chassis, priority); err != nil { slog.Warn("reconcile/apply: SetGatewayChassis failed", "vpc_id", vpcID, "chassis", chassis, "err", err) + res.fail(classIGW, vpcID, err) } } r.ensureGatewayClaimed(ctx, topology.GatewayChassisRedirectPort(vpcID)) @@ -550,10 +570,11 @@ func (r *reconciler) ensureGatewayClaimed(ctx context.Context, crPortName string // proves the gateway-chassis flow exists, not the gatewayLRP->guest hop, so a guest // whose port has not converged (e.g. just after a host reboot) stays dark while // every other signal is green. -func (r *reconciler) applyEIPs(ctx context.Context, intent IntentState, _ ActualState) { +func (r *reconciler) applyEIPs(ctx context.Context, intent IntentState, _ ActualState, res *passResult) { for _, spec := range r.floatingIPSpecs(intent) { if err := r.nat.AddEIP(ctx, spec); err != nil { slog.Error("reconcile/apply: AddEIP failed", "external_ip", spec.ExternalIP, "logical_ip", spec.LogicalIP, "err", err) + res.fail(classEIP, spec.ExternalIP, err) } r.ensureGuestPortDatapath(ctx, spec) } @@ -607,19 +628,21 @@ func (r *reconciler) floatingIPSpecs(intent IntentState) []policy.EIPSpec { // and union its ports into the live set so a mid-pass launch counts as live; skip the // prune entirely if the re-read fails rather than risk a false sweep against a snapshot // known to be stale. -func (r *reconciler) pruneOrphanEIPs(ctx context.Context, intent IntentState) { +func (r *reconciler) pruneOrphanEIPs(ctx context.Context, intent IntentState, res *passResult) { live := make(map[string]struct{}, len(intent.Ports)+len(intent.EIPs)) addLivePorts(live, intent) if r.reloadIntent != nil { fresh, err := r.reloadIntent(ctx) if err != nil { slog.Warn("reconcile/apply: fresh intent re-read failed; skipping orphan EIP prune", "err", err) + res.fail(classEIP, "orphan-prune", err) return } addLivePorts(live, fresh) } if pruned, err := r.nat.PruneOrphanEIPs(ctx, live); err != nil { slog.Warn("reconcile/apply: orphan EIP prune failed", "err", err) + res.fail(classEIP, "orphan-prune", err) } else if pruned > 0 { slog.Info("reconcile/apply: pruned orphan dnat_and_snat rows", "count", pruned) } @@ -644,14 +667,14 @@ func addLivePorts(live map[string]struct{}, intent IntentState) { // drop gate. Public IPs come from two disjoint sources: auto-assigned and ELB // addresses recorded on the ENI (intent.Ports) and user EIPs in the EIP bucket // (intent.EIPs). Both need the same /32 reroute above the gate. -func (r *reconciler) applyPublicInstanceEgress(ctx context.Context, intent IntentState, _ ActualState) { +func (r *reconciler) applyPublicInstanceEgress(ctx context.Context, intent IntentState, _ ActualState, res *passResult) { for _, p := range intent.Ports { if p.PublicIP.IsValid() { - r.ensureEIPEgressExemption(ctx, intent, p.VPCID, p.SubnetID, p.PrivateIP.String()) + r.ensureEIPEgressExemption(ctx, intent, p.VPCID, p.SubnetID, p.PrivateIP.String(), res) } } for _, e := range intent.EIPs { - r.ensureEIPEgressExemption(ctx, intent, e.VPCID, subnetIDForIP(intent.Subnets, e.LogicalIP), e.LogicalIP) + r.ensureEIPEgressExemption(ctx, intent, e.VPCID, subnetIDForIP(intent.Subnets, e.LogicalIP), e.LogicalIP, res) } } @@ -663,7 +686,7 @@ func (r *reconciler) applyPublicInstanceEgress(ctx context.Context, intent Inten // the datapath; the instance's dnat_and_snat supplies SNAT. Scoped to subnets that // actually carry a drop gate: routed subnets egress via their priority-1000 reroute // and need no exemption, so the gate presence bounds the blast radius. -func (r *reconciler) ensureEIPEgressExemption(ctx context.Context, intent IntentState, vpcID, subnetID, instanceIP string) { +func (r *reconciler) ensureEIPEgressExemption(ctx context.Context, intent IntentState, vpcID, subnetID, instanceIP string, res *passResult) { if subnetID == "" { slog.Warn("reconcile/apply: public instance maps to no subnet; skipping egress exemption", "vpc_id", vpcID, "instance_ip", instanceIP) @@ -675,6 +698,7 @@ func (r *reconciler) ensureEIPEgressExemption(ctx context.Context, intent Intent if err := r.igw.EnsureEIPInstanceEgress(ctx, vpcID, subnetID, instanceIP); err != nil { slog.Error("reconcile/apply: EnsureEIPInstanceEgress failed", "vpc_id", vpcID, "subnet_id", subnetID, "instance_ip", instanceIP, "err", err) + res.fail(classPublicEgress, instanceIP, err) } } @@ -762,43 +786,47 @@ func (r *reconciler) ensureGuestPortDatapath(ctx context.Context, spec policy.EI } // applyNATGWs runs every intent NAT gateway through NATManager.AddNATGateway. -func (r *reconciler) applyNATGWs(ctx context.Context, intent IntentState, _ ActualState) { +func (r *reconciler) applyNATGWs(ctx context.Context, intent IntentState, _ ActualState, res *passResult) { for _, spec := range intent.NATGWs { if err := r.nat.AddNATGateway(ctx, spec); err != nil { slog.Warn("reconcile/apply: AddNATGateway failed (likely already exists)", "natgw_id", spec.NATGatewayID, "subnet_cidr", spec.SubnetCIDR, "err", err) + res.fail(classNATGW, spec.NATGatewayID, err) } } } // applyIGWRoutes installs per-subnet egress reroute policies from intent. // Closes the bootstrap race: events fire before subscribers attach; KV retains the route. -func (r *reconciler) applyIGWRoutes(ctx context.Context, intent IntentState, _ ActualState) { +func (r *reconciler) applyIGWRoutes(ctx context.Context, intent IntentState, _ ActualState, res *passResult) { for _, spec := range intent.IGWRoutes { if err := r.igw.EnsureSubnetEgress(ctx, spec.VPCID, spec.SubnetID, spec.DestCIDR); err != nil { slog.Error("reconcile/apply: EnsureSubnetEgress failed", "vpc_id", spec.VPCID, "subnet_id", spec.SubnetID, "cidr", spec.DestCIDR.String(), "err", err) + res.fail(classIGWRoute, spec.SubnetID, err) } } } // applyNATGWRoutes is the NATGW priority sibling of applyIGWRoutes. -func (r *reconciler) applyNATGWRoutes(ctx context.Context, intent IntentState, _ ActualState) { +func (r *reconciler) applyNATGWRoutes(ctx context.Context, intent IntentState, _ ActualState, res *passResult) { for _, spec := range intent.NATGWRoutes { if err := r.igw.EnsureNATGatewaySubnetEgress(ctx, spec.VPCID, spec.SubnetID, spec.DestCIDR); err != nil { slog.Error("reconcile/apply: EnsureNATGatewaySubnetEgress failed", "vpc_id", spec.VPCID, "subnet_id", spec.SubnetID, "cidr", spec.DestCIDR.String(), "err", err) + res.fail(classNATGWRoute, spec.SubnetID, err) } } } // applyDropGates installs DROP policies for subnets with an attached IGW but // no 0.0.0.0/0 route, preventing unintended egress via the VPC default route. -func (r *reconciler) applyDropGates(ctx context.Context, intent IntentState, _ ActualState) { +func (r *reconciler) applyDropGates(ctx context.Context, intent IntentState, _ ActualState, res *passResult) { for _, spec := range intent.DropGates { if err := r.igw.EnsureSubnetEgressDrop(ctx, spec.VPCID, spec.SubnetID, spec.DestCIDR); err != nil { slog.Error("reconcile/apply: EnsureSubnetEgressDrop failed", "vpc_id", spec.VPCID, "subnet_id", spec.SubnetID, "cidr", spec.DestCIDR.String(), "err", err) + res.fail(classDropGate, spec.SubnetID, err) } } } diff --git a/spinifex/network/reconcile/apply_test.go b/spinifex/network/reconcile/apply_test.go index 76f2f8950..9536b8d3f 100644 --- a/spinifex/network/reconcile/apply_test.go +++ b/spinifex/network/reconcile/apply_test.go @@ -596,7 +596,7 @@ func TestReconcile_PruneOrphanEIPs_SweepsAbsentOwners(t *testing.T) { } intent := freshIntent(t) // Ports has eni-a only - r.pruneOrphanEIPs(ctx, intent) + r.pruneOrphanEIPs(ctx, intent, &passResult{}) if findNATByExternal(m, "dnat_and_snat", "192.168.1.10") == nil { t.Errorf("live EIP row must survive the prune") @@ -647,7 +647,7 @@ func TestReconcile_PruneOrphanEIPs_SparesMidPassLaunch(t *testing.T) { return fresh, nil } - r.pruneOrphanEIPs(ctx, snapshot) + r.pruneOrphanEIPs(ctx, snapshot, &passResult{}) if findNATByExternal(m, "dnat_and_snat", "192.168.1.20") == nil { t.Errorf("mid-pass launch row must survive the prune via the fresh re-read") @@ -676,7 +676,7 @@ func TestReconcile_PruneOrphanEIPs_SkipsOnReloadError(t *testing.T) { r.reloadIntent = func(context.Context) (IntentState, error) { return IntentState{}, errors.New("kv unavailable") } - r.pruneOrphanEIPs(ctx, freshIntent(t)) + r.pruneOrphanEIPs(ctx, freshIntent(t), &passResult{}) if findNATByExternal(m, "dnat_and_snat", "192.168.1.11") == nil { t.Errorf("prune must be skipped when the fresh re-read fails") @@ -797,3 +797,121 @@ func TestApplySubnets_ConvergesDriftedDHCPOptions(t *testing.T) { t.Errorf("DHCP options row was replaced (%s -> %s), want an in-place update", before.UUID, after.UUID) } } + +// stubIGW forces AttachIGW to fail while delegating everything else to a real +// manager, so a pass fails exactly one resource and the rest still applies. +type stubIGW struct { + external.IGWManager + + attachErr error +} + +func (s *stubIGW) AttachIGW(ctx context.Context, spec external.IGWSpec) error { + if s.attachErr != nil { + return s.attachErr + } + return s.IGWManager.AttachIGW(ctx, spec) +} + +// igwIntent is freshIntent plus an IGW for vpc-a, the resource whose failed +// attach a pass used to swallow. +func igwIntent(t *testing.T) IntentState { + t.Helper() + intent := freshIntent(t) + intent.IGWs["vpc-a"] = external.IGWSpec{VPCID: "vpc-a", InternetGatewayID: "igw-a"} + return intent +} + +// A failed AttachIGW must surface as ErrPassIncomplete rather than a nil return: +// a swallowed failure is indistinguishable from a converged pass, so nothing +// retries it until the next drift tick. +func TestReconcile_FailedAttachIGWReportsPassIncomplete(t *testing.T) { + rec, m := newTestReconciler(t) + ctx := context.Background() + attachErr := errors.New("allocate gateway LRP IP: dhcp gw-lrp acquire: context deadline exceeded") + rec.igw = &stubIGW{IGWManager: rec.igw, attachErr: attachErr} + // Without a chassis the rebind returns on its first line, which would make + // the assertion below pass whether or not the failure short-circuits it. + rec.chassis = []string{"chassis-1"} + + err := rec.Reconcile(ctx, igwIntent(t)) + if !errors.Is(err, ErrPassIncomplete) { + t.Fatalf("Reconcile = %v, want ErrPassIncomplete — a swallowed AttachIGW failure "+ + "leaves the drift loop unable to tell a converged pass from a broken one", err) + } + // The rest of the pass still applies: only the IGW is unconverged. + if _, ok := m.Ports[topology.Port("eni-a")]; !ok { + t.Errorf("ENI port missing: a failed IGW attach must not abort the pass") + } + if _, ok := m.RouterPorts[topology.GatewayRouterPort("vpc-a")]; ok { + t.Errorf("gateway LRP present after a failed attach") + } + if m.SetGatewayChassisCalls != 0 { + t.Errorf("SetGatewayChassis called %d times after a failed attach, want 0 — "+ + "rebinding chassis onto a gateway LRP that was never created", m.SetGatewayChassisCalls) + } +} + +// The converged path must stay clean: a pass with the same intent and a working +// AttachIGW returns nil, so the drift loop resets to DriftInterval. +func TestReconcile_AttachedIGWReportsConverged(t *testing.T) { + rec, m := newTestReconciler(t) + ctx := context.Background() + rec.chassis = []string{"chassis-1"} + + if err := rec.Reconcile(ctx, igwIntent(t)); err != nil { + t.Fatalf("Reconcile = %v, want nil", err) + } + if _, ok := m.RouterPorts[topology.GatewayRouterPort("vpc-a")]; !ok { + t.Errorf("gateway LRP missing after a successful attach") + } + if m.SetGatewayChassisCalls == 0 { + t.Errorf("SetGatewayChassis never called after a successful attach: the rebind " + + "must run, or a rebooted chassis never reclaims the gateway") + } +} + +// stubNAT forces AddEIP to fail, delegating everything else. +type stubNAT struct { + policy.NATManager + + addErr error +} + +func (s *stubNAT) AddEIP(ctx context.Context, spec policy.EIPSpec) error { + if s.addErr != nil { + return s.addErr + } + return s.NATManager.AddEIP(ctx, spec) +} + +// The incomplete signal is a per-pass contract, not an IGW special case: a stage +// other than applyIGWs failing must report itself the same way. +func TestReconcile_FailedAddEIPReportsPassIncomplete(t *testing.T) { + rec, _ := newTestReconciler(t) + ctx := context.Background() + rec.nat = &stubNAT{NATManager: rec.nat, addErr: errors.New("ovsdb unreachable")} + + intent := freshIntent(t) + intent.EIPs["10.0.1.10"] = policy.EIPSpec{ + VPCID: "vpc-a", ExternalIP: "192.168.1.10", LogicalIP: "10.0.1.10", + PortName: topology.Port("eni-a"), + } + if err := rec.Reconcile(ctx, intent); !errors.Is(err, ErrPassIncomplete) { + t.Fatalf("Reconcile = %v, want ErrPassIncomplete for a failed AddEIP", err) + } +} + +// The convergence summary counts failures per class, so an operator reading one +// line knows which stage did not converge and how widely. +func TestPassResult_SummaryCountsPerClass(t *testing.T) { + res := &passResult{} + res.fail(classIGW, "vpc-a", errors.New("x")) + res.fail(classEIP, "192.168.1.10", errors.New("y")) + res.fail(classEIP, "192.168.1.11", errors.New("z")) + + want := []any{"unconverged", 3, classEIP, 2, classIGW, 1} + if got := res.summaryKV(); !slices.Equal(got, want) { + t.Errorf("summaryKV() = %v, want %v", got, want) + } +} diff --git a/spinifex/network/reconcile/drift.go b/spinifex/network/reconcile/drift.go index de4041633..cda06058f 100644 --- a/spinifex/network/reconcile/drift.go +++ b/spinifex/network/reconcile/drift.go @@ -2,9 +2,11 @@ package reconcile import ( "context" + "errors" "log/slog" "time" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" ) @@ -13,42 +15,93 @@ import ( // integration tests can shrink it. var DriftInterval = 5 * time.Minute +// driftBackoffBase is the requeue gap after the first incomplete pass. Var so +// tests can shrink it. +var driftBackoffBase = 5 * time.Second + +// driftBackoffFactor grows the requeue gap per consecutive incomplete pass, up +// to DriftInterval. Without a requeue the repair latency for any transient apply +// failure is a full DriftInterval, so a 57-second DHCP stall costs five minutes +// of a VPC having no external gateway. +const driftBackoffFactor = 3 + // DriftLoop runs Reconcile every DriftInterval, gated on AcquireLeader so -// only one vpcd scans at a time. Returns when ctx is cancelled. -func DriftLoop(ctx context.Context, rec Reconciler, nc *nats.Conn, localAZ, holder string) { +// only one vpcd scans at a time. An incomplete pass requeues on a short +// backoff instead. Returns when ctx is cancelled. +// +// startup is the outcome of the bootstrap ReconcileApplyOnly pass, or nil when +// this node did not run one. Seeding from it matters because the gateway LRP's +// first DORA happens in that pass: without the seed a resource that failed +// before the loop started waits a full DriftInterval for its first retry. +func DriftLoop(ctx context.Context, rec Reconciler, nc *nats.Conn, localAZ, holder string, startup error) { js, err := jetstream.New(nc) if err != nil { slog.Error("reconcile/drift: JetStream context unavailable, drift loop disabled", "err", err) return } - ticker := time.NewTicker(DriftInterval) - defer ticker.Stop() + backoff := nextDriftBackoff(0, startup) + timer := time.NewTimer(driftWait(backoff)) + defer timer.Stop() for { select { case <-ctx.Done(): return - case <-ticker.C: - runDriftCycle(ctx, rec, nc, js, localAZ, holder) + case <-timer.C: + previous := backoff + backoff = nextDriftBackoff(backoff, runDriftCycle(ctx, rec, nc, js, localAZ, holder)) + wait := driftWait(backoff) + if backoff != previous { + slog.Info("reconcile/drift: requeue interval changed", "wait_ms", otelsetup.Millis(wait)) + } + timer.Reset(wait) } } } +// nextDriftBackoff returns the requeue backoff after a pass: zero once a pass +// converges, otherwise the previous backoff tripled from driftBackoffBase and +// held at DriftInterval. Tracked separately from the wait it produces, so a +// backoff sitting at the cap is not mistaken for a converged loop and reset. +func nextDriftBackoff(current time.Duration, err error) time.Duration { + if !errors.Is(err, ErrPassIncomplete) { + return 0 + } + next := driftBackoffBase + if current > 0 { + next = current * driftBackoffFactor + } + return min(next, DriftInterval) +} + +// driftWait is the gap before the next pass for a backoff; no backoff means a +// full DriftInterval. +func driftWait(backoff time.Duration) time.Duration { + if backoff <= 0 { + return DriftInterval + } + return backoff +} + // runDriftCycle is one tick body, split out so tests can drive it directly. -func runDriftCycle(ctx context.Context, rec Reconciler, nc *nats.Conn, js jetstream.JetStream, localAZ, holder string) { +// Returns the reconcile outcome so the caller can size the next wait; a cycle +// this node did not win the election for returns nil. +func runDriftCycle(ctx context.Context, rec Reconciler, nc *nats.Conn, js jetstream.JetStream, localAZ, holder string) error { release, elected := AcquireLeader(ctx, nc, KVBucketVPCDReconcile, holder) if !elected { - return + return nil } defer release() intent, err := LoadIntentFromKV(ctx, js, localAZ) if err != nil { slog.Warn("reconcile/drift: load intent failed", "err", err) - return + return err } if err := rec.Reconcile(ctx, intent); err != nil { slog.Warn("reconcile/drift: reconcile failed", "err", err) + return err } + return nil } diff --git a/spinifex/network/reconcile/drift_test.go b/spinifex/network/reconcile/drift_test.go new file mode 100644 index 000000000..48a10e9a6 --- /dev/null +++ b/spinifex/network/reconcile/drift_test.go @@ -0,0 +1,246 @@ +//test:in-package — the backoff ladder is unexported state (driftBackoffBase, +//nextDriftBackoff, driftWait) and the tests drive runDriftCycle directly to +//pin what DriftLoop does with its return value. + +package reconcile + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/testutil" +) + +// stubReconciler returns a canned outcome per pass, so the requeue schedule can +// be driven without a live OVN. Safe for concurrent use: DriftLoop calls it from +// its own goroutine while the test polls the count. +type stubReconciler struct { + mu sync.Mutex + outcomes []error + calls int +} + +var _ Reconciler = (*stubReconciler)(nil) + +func (s *stubReconciler) Reconcile(context.Context, IntentState) error { + s.mu.Lock() + defer s.mu.Unlock() + err := s.outcomes[min(s.calls, len(s.outcomes)-1)] + s.calls++ + return err +} + +func (s *stubReconciler) ReconcileApplyOnly(ctx context.Context, intent IntentState) error { + return s.Reconcile(ctx, intent) +} + +func (s *stubReconciler) callCount() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.calls +} + +func incompleteErr() error { + return fmt.Errorf("reconcile: 1 resource(s) unconverged: %w", ErrPassIncomplete) +} + +// shrinkDriftTiming compresses the loop's timings so a test observes several +// requeues in milliseconds. The interval stays long so that a loop which +// wrongly falls back to it registers as silence rather than as a fast requeue. +func shrinkDriftTiming(t *testing.T, interval, base time.Duration) { + t.Helper() + oldInterval, oldBase := DriftInterval, driftBackoffBase + DriftInterval, driftBackoffBase = interval, base + t.Cleanup(func() { DriftInterval, driftBackoffBase = oldInterval, oldBase }) +} + +// waitForCalls polls until rec has been called want times or within elapses. +func waitForCalls(t *testing.T, rec *stubReconciler, want int, within time.Duration) int { + t.Helper() + deadline := time.Now().Add(within) + for time.Now().Before(deadline) { + if got := rec.callCount(); got >= want { + return got + } + time.Sleep(time.Millisecond) + } + return rec.callCount() +} + +// The whole point of the change: an incomplete pass must come back in +// milliseconds-scaled backoff rather than a full DriftInterval. Driving the real +// DriftLoop is what pins timer.Reset to the computed wait — a loop that resets +// to DriftInterval instead still satisfies every pure-function test below. +func TestDriftLoop_IncompletePassRequeuesFast(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + shrinkDriftTiming(t, 10*time.Second, time.Millisecond) + + rec := &stubReconciler{outcomes: []error{incompleteErr()}} + go DriftLoop(t.Context(), rec, nc, "us-east-1a", "node-1", incompleteErr()) + + if got := waitForCalls(t, rec, 4, 5*time.Second); got < 4 { + t.Fatalf("reconcile called %d times, want >= 4 — an incomplete pass is waiting "+ + "a full DriftInterval instead of requeueing on the backoff", got) + } +} + +// The startup ReconcileApplyOnly pass is where the gateway LRP's first DORA +// happens. Its outcome must seed the loop, or the resource it failed to converge +// waits a full DriftInterval for its first retry — the original bug, on boot. +func TestDriftLoop_SeedsBackoffFromStartupOutcome(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + shrinkDriftTiming(t, 10*time.Second, time.Millisecond) + + // Converged from here on: only the seed can bring the first pass forward. + rec := &stubReconciler{outcomes: []error{nil}} + go DriftLoop(t.Context(), rec, nc, "us-east-1a", "node-1", incompleteErr()) + + if got := waitForCalls(t, rec, 1, 5*time.Second); got < 1 { + t.Fatalf("reconcile never ran: a failed startup pass must seed the backoff "+ + "rather than leave the loop armed at DriftInterval (calls=%d)", got) + } +} + +// The converse: a clean startup must not pull the first pass forward, or every +// vpcd reconciles immediately on boot regardless of need. +func TestDriftLoop_ConvergedStartupWaitsFullInterval(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + shrinkDriftTiming(t, 10*time.Second, time.Millisecond) + + rec := &stubReconciler{outcomes: []error{incompleteErr()}} + go DriftLoop(t.Context(), rec, nc, "us-east-1a", "node-1", nil) + + time.Sleep(300 * time.Millisecond) + if got := rec.callCount(); got != 0 { + t.Errorf("reconcile ran %d times within 300ms of a converged startup, want 0 "+ + "— the first pass must wait DriftInterval (10s here)", got) + } +} + +// A converged pass must clear the backoff so the loop drops back to its routine +// cadence, rather than staying on the fast requeue forever. +func TestDriftLoop_ConvergedPassReturnsToDriftInterval(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + shrinkDriftTiming(t, 10*time.Second, time.Millisecond) + + // Pass 1 incomplete (requeues in ~3ms), pass 2 converged (must requeue at 10s). + rec := &stubReconciler{outcomes: []error{incompleteErr(), nil}} + go DriftLoop(t.Context(), rec, nc, "us-east-1a", "node-1", incompleteErr()) + + if got := waitForCalls(t, rec, 2, 5*time.Second); got < 2 { + t.Fatalf("reconcile called %d times, want 2 before the loop settles", got) + } + time.Sleep(300 * time.Millisecond) + if got := rec.callCount(); got != 2 { + t.Errorf("reconcile called %d times, want exactly 2 — a converged pass must "+ + "reset the backoff to DriftInterval, not stay on the fast requeue", got) + } +} + +// runDriftCycle's return is what sizes the next wait, so a pass this node did not +// win must not be reported as converged-or-broken; it reports nil and the caller +// leaves the cadence alone. +func TestRunDriftCycle_ReportsReconcileOutcome(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + + rec := &stubReconciler{outcomes: []error{incompleteErr()}} + err := runDriftCycle(t.Context(), rec, nc, js, "us-east-1a", "node-1") + if !errors.Is(err, ErrPassIncomplete) { + t.Fatalf("runDriftCycle = %v, want ErrPassIncomplete propagated from the pass", err) + } + + converged := &stubReconciler{outcomes: []error{nil}} + if err := runDriftCycle(t.Context(), converged, nc, js, "us-east-1a", "node-1"); err != nil { + t.Errorf("runDriftCycle = %v, want nil for a converged pass", err) + } +} + +// A cycle this node loses the election for must not run a pass at all. +func TestRunDriftCycle_SkipsWhenNotElected(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + + release, ok := AcquireLeader(t.Context(), nc, KVBucketVPCDReconcile, "other-node") + if !ok { + t.Fatal("failed to take the lock the test needs held") + } + defer release() + + rec := &stubReconciler{outcomes: []error{incompleteErr()}} + if err := runDriftCycle(t.Context(), rec, nc, js, "us-east-1a", "node-1"); err != nil { + t.Errorf("runDriftCycle = %v, want nil when not elected", err) + } + if got := rec.callCount(); got != 0 { + t.Errorf("reconcile ran %d times without winning the election, want 0", got) + } +} + +// The ladder itself, as pure functions: exact gaps are asserted here because the +// loop tests above run on compressed timings and can only prove ordering. +func TestDrift_IncompletePassBacksOffThenResets(t *testing.T) { + incomplete := incompleteErr() + outcomes := []error{incomplete, incomplete, incomplete, incomplete, incomplete, incomplete, nil} + + var backoff time.Duration + var waits []time.Duration + for _, outcome := range outcomes { + backoff = nextDriftBackoff(backoff, outcome) + waits = append(waits, driftWait(backoff)) + } + + want := []time.Duration{ + 5 * time.Second, 15 * time.Second, 45 * time.Second, + 135 * time.Second, DriftInterval, DriftInterval, // capped, then held + DriftInterval, // converged: back to the routine interval + } + for i := range want { + if waits[i] != want[i] { + t.Fatalf("wait after pass %d = %v, want %v (full sequence %v)", i+1, waits[i], want[i], waits) + } + } + // The cap and the reset produce the same wait, so pin the state behind it: + // a converged pass must clear the backoff, not sit at the cap. + if backoff != 0 { + t.Errorf("backoff after a converged pass = %v, want 0", backoff) + } +} + +// A pass that converged, or failed for a reason other than non-convergence +// (e.g. a scan failure), must not shorten the interval — only an incomplete +// pass has a known-transient resource worth retrying early. +func TestDrift_NonIncompleteOutcomesKeepDriftInterval(t *testing.T) { + tests := []struct { + name string + err error + }{ + {"converged", nil}, + {"scan failure", errors.New("scan actual OVN state: ovsdb unreachable")}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // Start from a backoff already in force, so a reset is observable. + if got := nextDriftBackoff(45*time.Second, tt.err); got != 0 { + t.Errorf("nextDriftBackoff(45s, %v) = %v, want 0", tt.err, got) + } + if got := driftWait(0); got != DriftInterval { + t.Errorf("driftWait(0) = %v, want %v", got, DriftInterval) + } + }) + } +} + +// The backoff must never exceed DriftInterval: a permanently broken resource +// degrades to today's behaviour instead of hammering the reconcile loop. +func TestDrift_BackoffCapsAtDriftInterval(t *testing.T) { + incomplete := incompleteErr() + backoff := DriftInterval + for range 3 { + backoff = nextDriftBackoff(backoff, incomplete) + if backoff > DriftInterval { + t.Fatalf("backoff = %v, want <= DriftInterval (%v)", backoff, DriftInterval) + } + } +} diff --git a/spinifex/network/reconcile/lock.go b/spinifex/network/reconcile/lock.go index 9a35299e5..628f89632 100644 --- a/spinifex/network/reconcile/lock.go +++ b/spinifex/network/reconcile/lock.go @@ -5,6 +5,7 @@ import ( "errors" "github.com/mulgadc/spinifex/spinifex/otelsetup" "log/slog" + "sync" "time" "github.com/nats-io/nats.go" @@ -15,13 +16,22 @@ import ( const ( KVBucketVPCDReconcile = "spinifex-vpcd-reconcile" reconcileLeaderKey = "leader" - reconcileLeaderTTL = 60 * time.Second // reconcileReleaseTimeout bounds the lock delete on the way out, which runs // on a detached context and so cannot inherit the caller's deadline. reconcileReleaseTimeout = 5 * time.Second ) +// Leader-key lifetime. Vars (not consts) so tests can shrink them. +// +// reconcileLeaderRenew is the gap between refreshes: a pass can outlive the TTL +// on its own — one stalled gw-lrp DORA runs ~64s — so the key is renewed while +// the pass runs rather than left to expire underneath it. +var ( + reconcileLeaderTTL = 60 * time.Second + reconcileLeaderRenew = 20 * time.Second +) + // Bounded wait for JetStream quorum on cold multi-node start. Vars (not // consts) so tests can shrink them. var ( @@ -72,20 +82,80 @@ func AcquireLeader(ctx context.Context, nc *nats.Conn, bucket, holder string) (f } } - if _, err := kv.Create(ctx, reconcileLeaderKey, []byte(holder)); err != nil { + rev, err := kv.Create(ctx, reconcileLeaderKey, []byte(holder)) + if err != nil { slog.Info("reconcile/lock: another holder is leader, skipping reconcile", "holder", holder, "bucket", bucket, "err", err) return nil, false } slog.Info("reconcile/lock: elected", "holder", holder, "bucket", bucket) + lease := &leaderLease{kv: kv, rev: rev} + renewCtx, stopRenew := context.WithCancel(ctx) + go lease.renew(renewCtx, holder, bucket) + return func() { + stopRenew() + rev, held := lease.revision() + if !held { + // Renewal lost the key, so another node may already hold it. An + // unguarded delete here would drop that node's lock, not ours. + slog.Warn("reconcile/lock: lock already lost before release; not deleting", "holder", holder, "bucket", bucket) + return + } // Release outlives ctx: shutdown is the common reason to release, and // skipping the delete would park the lock for the full TTL and stall // every other node's reconcile. releaseCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), reconcileReleaseTimeout) defer cancel() - if err := kv.Delete(releaseCtx, reconcileLeaderKey); err != nil { + if err := kv.Delete(releaseCtx, reconcileLeaderKey, jetstream.LastRevision(rev)); err != nil { slog.Warn("reconcile/lock: failed to release lock (TTL will reap)", "holder", holder, "bucket", bucket, "err", err) } }, true } + +// leaderLease tracks the revision of a held leader key. Renewal and the final +// delete are both CAS-guarded on it, so a pass that outlives the TTL can never +// delete the key of the node that took over from it. +type leaderLease struct { + mu sync.Mutex + kv jetstream.KeyValue + rev uint64 + lost bool +} + +// revision returns the current revision and whether the lease is still held. +func (l *leaderLease) revision() (uint64, bool) { + l.mu.Lock() + defer l.mu.Unlock() + return l.rev, !l.lost +} + +// renew refreshes the leader key until ctx is cancelled or a refresh loses the +// CAS, which means the key expired and another node claimed it. +func (l *leaderLease) renew(ctx context.Context, holder, bucket string) { + ticker := time.NewTicker(reconcileLeaderRenew) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + rev, held := l.revision() + if !held { + return + } + next, err := l.kv.Update(ctx, reconcileLeaderKey, []byte(holder), rev) + if err != nil { + l.mu.Lock() + l.lost = true + l.mu.Unlock() + slog.Warn("reconcile/lock: leader key renewal failed; another node may take over mid-pass", + "holder", holder, "bucket", bucket, "err", err) + return + } + l.mu.Lock() + l.rev = next + l.mu.Unlock() + } + } +} diff --git a/spinifex/network/reconcile/lock_test.go b/spinifex/network/reconcile/lock_test.go index 121385b53..f79dca928 100644 --- a/spinifex/network/reconcile/lock_test.go +++ b/spinifex/network/reconcile/lock_test.go @@ -3,6 +3,7 @@ package reconcile import ( "context" "testing" + "time" "github.com/mulgadc/spinifex/spinifex/testutil" "github.com/stretchr/testify/assert" @@ -51,3 +52,54 @@ func TestAcquireLeader_ReleaseSurvivesCancelledContext(t *testing.T) { require.True(t, ok, "lock must be free immediately after release, not held until the TTL reaps it") release2() } + +// shrinkLeaderTiming compresses the leader key's lifetime so a test can outlast +// the TTL in a second rather than a minute. +func shrinkLeaderTiming(t *testing.T, ttl, renew time.Duration) { + t.Helper() + oldTTL, oldRenew := reconcileLeaderTTL, reconcileLeaderRenew + reconcileLeaderTTL, reconcileLeaderRenew = ttl, renew + t.Cleanup(func() { reconcileLeaderTTL, reconcileLeaderRenew = oldTTL, oldRenew }) +} + +// A pass can outlive the TTL on its own — one stalled gw-lrp DORA runs ~64s +// against a 60s TTL — so the key must be renewed while the pass runs. Without +// this a peer elects itself mid-pass and two reconcilers race the same OVN NB. +func TestAcquireLeader_RenewsKeyBeyondTTL(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + shrinkLeaderTiming(t, time.Second, 100*time.Millisecond) + + release, ok := AcquireLeader(t.Context(), nc, KVBucketVPCDReconcile, "node-1") + require.True(t, ok) + defer release() + + // Well past the TTL: an unrenewed key would have expired by now. + time.Sleep(2500 * time.Millisecond) + + loserRelease, ok := AcquireLeader(t.Context(), nc, KVBucketVPCDReconcile, "node-2") + assert.False(t, ok, "leader key expired mid-pass: a peer can now reconcile concurrently") + assert.Nil(t, loserRelease) +} + +// If the key was lost anyway, the stale holder's release must not delete the +// successor's key — an unguarded delete hands the lock to a third node while +// the successor believes it still holds it. +func TestAcquireLeader_ReleaseDoesNotDeleteSuccessorKey(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + + release, ok := AcquireLeader(t.Context(), nc, KVBucketVPCDReconcile, "node-1") + require.True(t, ok) + + // Stand in for a successor that claimed the key after it expired: the write + // bumps the revision, so node-1's release is now CAS-stale. + kv, err := js.KeyValue(t.Context(), KVBucketVPCDReconcile) + require.NoError(t, err) + _, err = kv.Put(t.Context(), reconcileLeaderKey, []byte("node-2")) + require.NoError(t, err) + + release() + + entry, err := kv.Get(t.Context(), reconcileLeaderKey) + require.NoError(t, err, "stale release deleted the successor's leader key") + assert.Equal(t, "node-2", string(entry.Value()), "successor's key must survive a stale release") +} diff --git a/spinifex/network/reconcile/reconcile.go b/spinifex/network/reconcile/reconcile.go index 805e8f84b..18a9fb1d3 100644 --- a/spinifex/network/reconcile/reconcile.go +++ b/spinifex/network/reconcile/reconcile.go @@ -8,6 +8,8 @@ import ( "errors" "fmt" "log/slog" + "maps" + "slices" "sync" "time" @@ -17,9 +19,18 @@ import ( "github.com/mulgadc/spinifex/spinifex/network/topology" ) +// ErrPassIncomplete marks a pass that applied what it could but left at least +// one resource unconverged. A pass keeps applying past a per-resource failure, +// so without this the caller cannot tell a clean pass from one that dropped ten +// resources on the floor. DriftLoop tests for it with errors.Is and requeues on +// a short backoff instead of waiting a full DriftInterval. +var ErrPassIncomplete = errors.New("reconcile: pass incomplete") + // Reconciler converges OVN NB DB to a declared IntentState. Implementations // are idempotent: a second call with the same IntentState is a no-op. type Reconciler interface { + // Reconcile returns a scan failure as-is, and wraps ErrPassIncomplete when + // the scan succeeded but some resources did not converge. Reconcile(ctx context.Context, intent IntentState) error // ReconcileApplyOnly skips orphan-pruning. Startup uses this to avoid // racing peer subscribers that haven't processed in-flight create events @@ -186,8 +197,58 @@ func New(cfg Config) (Reconciler, error) { }, nil } +// passFailure is one resource an apply stage could not converge. +type passFailure struct { + class string + id string + err error +} + +// passResult accumulates per-resource apply failures so a pass that logged and +// continued past them still reports itself as incomplete to the drift loop. +type passResult struct { + failures []passFailure +} + +// fail records that id in class did not converge. +func (p *passResult) fail(class, id string, err error) { + p.failures = append(p.failures, passFailure{class: class, id: id, err: err}) +} + +// summaryKV renders per-class failure counts as slog key/values, class-sorted so +// the summary line is stable across passes. +func (p *passResult) summaryKV() []any { + counts := make(map[string]int, len(p.failures)) + for _, f := range p.failures { + counts[f.class]++ + } + kv := make([]any, 0, 2+2*len(counts)) + kv = append(kv, "unconverged", len(p.failures)) + for _, class := range slices.Sorted(maps.Keys(counts)) { + kv = append(kv, class, counts[class]) + } + return kv +} + +// Failure classes for the per-pass convergence summary, one per apply stage so +// the summary names the stage that did not converge. +const ( + classVPC = "vpc" + classSubnet = "subnet" + classSG = "sg" + classPort = "port" + classIGW = "igw" + classEIP = "eip" + classNATGW = "natgw" + classIGWRoute = "igw_route" + classNATGWRoute = "natgw_route" + classDropGate = "drop_gate" + classPublicEgress = "public_egress" +) + // Reconcile diffs intent vs. actual OVN state and applies in topological order. -// Per-stage errors are logged; only a scan failure is returned. +// Per-stage errors are logged and the pass continues; a scan failure is returned +// as-is and an unconverged resource surfaces as ErrPassIncomplete. func (r *reconciler) Reconcile(ctx context.Context, intent IntentState) error { return r.reconcile(ctx, intent, true) } @@ -217,21 +278,26 @@ func (r *reconciler) reconcile(ctx context.Context, intent IntentState, pruneOrp "intent_natgw_routes", len(intent.NATGWRoutes), ) - r.applyVPCs(ctx, intent, actual) - r.applySubnets(ctx, intent, actual) - r.applySGs(ctx, intent, actual, pruneOrphans) - r.applyPorts(ctx, intent, actual, pruneOrphans) - r.applyIGWs(ctx, intent, actual) - r.applyEIPs(ctx, intent, actual) + res := &passResult{} + r.applyVPCs(ctx, intent, actual, res) + r.applySubnets(ctx, intent, actual, res) + r.applySGs(ctx, intent, actual, pruneOrphans, res) + r.applyPorts(ctx, intent, actual, pruneOrphans, res) + r.applyIGWs(ctx, intent, actual, res) + r.applyEIPs(ctx, intent, actual, res) if pruneOrphans { - r.pruneOrphanEIPs(ctx, intent) + r.pruneOrphanEIPs(ctx, intent, res) } - r.applyNATGWs(ctx, intent, actual) - r.applyIGWRoutes(ctx, intent, actual) - r.applyNATGWRoutes(ctx, intent, actual) - r.applyDropGates(ctx, intent, actual) - r.applyPublicInstanceEgress(ctx, intent, actual) + r.applyNATGWs(ctx, intent, actual, res) + r.applyIGWRoutes(ctx, intent, actual, res) + r.applyNATGWRoutes(ctx, intent, actual, res) + r.applyDropGates(ctx, intent, actual, res) + r.applyPublicInstanceEgress(ctx, intent, actual, res) - slog.Info("reconcile: complete") - return nil + if len(res.failures) == 0 { + slog.Info("reconcile: complete", "converged", true) + return nil + } + slog.Warn("reconcile: pass incomplete", res.summaryKV()...) + return fmt.Errorf("reconcile: %d resource(s) unconverged: %w", len(res.failures), ErrPassIncomplete) } diff --git a/spinifex/vpcd/vpcd.go b/spinifex/vpcd/vpcd.go index 31fd31069..f1b2697bd 100644 --- a/spinifex/vpcd/vpcd.go +++ b/spinifex/vpcd/vpcd.go @@ -700,12 +700,15 @@ func launchService(cfg *Config) error { // Startup reconcile (leader-gated, apply-only). Orphan pruning is skipped because intent may be stale: // a peer's vpc.create-sg could be mid-flight and a prune would sweep those port groups as orphans. // Drift loop uses full Reconcile. + // startupErr seeds the drift loop's backoff so a resource the bootstrap pass + // could not converge is retried on the short requeue, not a full interval. + var startupErr error if isLeader { intent, intentErr := reconcile.LoadIntentFromKV(ctx, js, cfg.AZ) if intentErr != nil { slog.Warn("vpcd: startup intent load failed", "err", intentErr) - } else if err := rec.ReconcileApplyOnly(ctx, intent); err != nil { - slog.Warn("vpcd: startup reconcile failed", "err", err) + } else if startupErr = rec.ReconcileApplyOnly(ctx, intent); startupErr != nil { + slog.Warn("vpcd: startup reconcile failed", "err", startupErr) } releaseLeader() } @@ -715,7 +718,7 @@ func launchService(cfg *Config) error { defer loopCancel() loopDone := make(chan struct{}) go func() { - reconcile.DriftLoop(loopCtx, rec, nc, cfg.AZ, holder) + reconcile.DriftLoop(loopCtx, rec, nc, cfg.AZ, holder, startupErr) close(loopDone) }()