diff --git a/spinifex/services/vpcd/topology.go b/spinifex/services/vpcd/topology.go index 431356294..b83370bd4 100644 --- a/spinifex/services/vpcd/topology.go +++ b/spinifex/services/vpcd/topology.go @@ -8,6 +8,7 @@ import ( "net" "strconv" "strings" + "time" "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" "github.com/mulgadc/spinifex/spinifex/services/vpcd/nbdb" @@ -16,6 +17,39 @@ import ( "github.com/nats-io/nats.go" ) +// waitForFlowsHV shells out to `ovn-nbctl --wait=hv sync`, which bumps +// NB_Global.nb_cfg and blocks until every connected chassis has acknowledged +// the new sequence number — i.e. ovn-northd has compiled NB -> SB and +// ovn-controller has installed the resulting flows. Used after IGW attach +// so newly-launched VMs aren't unreachable while their gateway chassis is +// still catching up (mulga-siv-105). +// +// Bounded by ovn-nbctl --timeout=30 (seconds). On overrun we log a WARN +// and return nil — the caller continues. In practice flows converge within +// seconds; a 30s overrun means something OVN-side is wedged enough that +// failing VPC create wouldn't improve things. +// +// Declared as a var so tests can stub it. +var waitForFlowsHV = func() error { + start := time.Now() + cmd := sudoCommand("ovn-nbctl", + "--no-leader-only", + "--timeout=30", + "--wait=hv", + "sync", + ) + if out, err := cmd.CombinedOutput(); err != nil { + slog.Warn("vpcd: OVN flows-ready barrier overran; continuing without confirmation", + "elapsed", time.Since(start), + "err", err, + "output", strings.TrimSpace(string(out)), + ) + return nil + } + slog.Debug("vpcd: OVN flows-ready barrier complete", "elapsed", time.Since(start)) + return nil +} + // NATS topics for VPC lifecycle events published by the daemon. const ( TopicVPCCreate = "vpc.create" @@ -1174,6 +1208,14 @@ func (h *TopologyHandler) handleIGWAttach(msg *nats.Msg) { "wan_gateway", wanGateway, "chassis_count", len(h.chassisNames), ) + + // Block until ovn-northd compiles + the gateway chassis has installed + // flows for this VPC. Without this, a subsequent RunInstances -> + // vpc.add-nat can complete while the datapath is still dark, and the + // new VM is unreachable on its public IP for the duration of the + // flow-install latency (typically seconds; observed up to minutes when + // the chassis was loaded during CI). mulga-siv-105. + _ = waitForFlowsHV() respond(msg, nil) } diff --git a/spinifex/services/vpcd/topology_test.go b/spinifex/services/vpcd/topology_test.go index 587f7838a..81c26c749 100644 --- a/spinifex/services/vpcd/topology_test.go +++ b/spinifex/services/vpcd/topology_test.go @@ -877,6 +877,49 @@ func TestTopologyHandler_IGWAttach(t *testing.T) { } } +// TestTopologyHandler_IGWAttach_InvokesFlowBarrier asserts handleIGWAttach +// calls waitForFlowsHV after the OVN writes complete (mulga-siv-105). The +// barrier itself shells out to ovn-nbctl --wait=hv sync; stub it here so +// the test doesn't depend on a live OVN. +func TestTopologyHandler_IGWAttach_InvokesFlowBarrier(t *testing.T) { + orig := waitForFlowsHV + var called int + waitForFlowsHV = func() error { called++; return nil } + defer func() { waitForFlowsHV = orig }() + + _, nc := startTestNATS(t) + mock := NewMockOVNClient() + _ = mock.Connect(context.Background()) + ctx := context.Background() + + topo := NewTopologyHandler(mock) + subs, err := topo.Subscribe(nc) + require.NoError(t, err) + defer func() { + for _, s := range subs { + _ = s.Unsubscribe() + } + }() + + require.NoError(t, mock.CreateLogicalRouter(ctx, &nbdb.LogicalRouter{ + Name: "vpc-vpc-barrier1", + ExternalIDs: map[string]string{ + "spinifex:vpc_id": "vpc-barrier1", + "spinifex:cidr": "10.0.0.0/16", + }, + })) + + evt := types.IGWEvent{InternetGatewayId: "igw-barrier1", VpcId: "vpc-barrier1"} + data, _ := json.Marshal(evt) + resp, err := nc.Request(TopicIGWAttach, data, 5_000_000_000) + require.NoError(t, err) + assertSuccess(t, resp, "attach IGW") + + if called != 1 { + t.Errorf("expected waitForFlowsHV to run exactly once after IGW attach, got %d", called) + } +} + func TestTopologyHandler_IGWAttach_WithExternalPool(t *testing.T) { _, nc := startTestNATS(t) mock := NewMockOVNClient()