diff --git a/spinifex/network/reconcile/drift.go b/spinifex/network/reconcile/drift.go index cda06058f..8673e8ff4 100644 --- a/spinifex/network/reconcile/drift.go +++ b/spinifex/network/reconcile/drift.go @@ -3,20 +3,28 @@ package reconcile import ( "context" "errors" + "fmt" "log/slog" "time" + handlers_ec2_eip "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eip" + handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw" + handlers_ec2_natgw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/natgw" + handlers_ec2_routetable "github.com/mulgadc/spinifex/spinifex/handlers/ec2/routetable" + handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" + "github.com/mulgadc/spinifex/spinifex/kvstore" "github.com/mulgadc/spinifex/spinifex/otelsetup" + loop "github.com/mulgadc/spinifex/spinifex/reconciler" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" ) -// DriftInterval is the gap between drift passes. Var (not const) so -// integration tests can shrink it. +// DriftInterval is the resync backstop between drift passes when nothing +// changes. Var (not const) so 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. +// driftBackoffBase is the requeue gap after the first incomplete pass, and the +// floor on how often a change may drive a pass. Var so tests can shrink it. var driftBackoffBase = 5 * time.Second // driftBackoffFactor grows the requeue gap per consecutive incomplete pass, up @@ -25,14 +33,57 @@ var driftBackoffBase = 5 * time.Second // 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. An incomplete pass requeues on a short -// backoff instead. Returns when ctx is cancelled. +// intentBuckets names every KV bucket LoadIntentFromKV reads. A write to one of +// them is an intent change and nothing else: none of their writers is periodic, +// and a pass writes OVN rather than KV, so watching them cannot self-trigger. +var intentBuckets = []string{ + handlers_ec2_vpc.KVBucketVPCs, + handlers_ec2_vpc.KVBucketSubnets, + handlers_ec2_vpc.KVBucketSecurityGroups, + handlers_ec2_vpc.KVBucketENIs, + handlers_ec2_igw.KVBucketIGW, + handlers_ec2_eip.KVBucketEIPs, + handlers_ec2_routetable.KVBucketRouteTables, + handlers_ec2_natgw.KVBucketNatGateways, +} + +// intentSource watches the intent buckets. Enumerated rather than fixed so a +// bucket that does not exist yet is left alone: the EC2 handlers own these and +// set their history depth, and an opening watch would create one at the wrong +// depth. A bucket that appears later is picked up on the next resync. +func intentSource(js jetstream.JetStream) loop.Source { + return loop.Dynamic(func(ctx context.Context) ([]*kvstore.Bucket, error) { + buckets := make([]*kvstore.Bucket, 0, len(intentBuckets)) + for _, name := range intentBuckets { + kv, err := js.KeyValue(ctx, name) + switch { + case errors.Is(err, jetstream.ErrBucketNotFound): + // Nothing of this kind has ever been created here. + slog.DebugContext(ctx, "reconcile/drift: intent bucket absent, not watched", + "bucket", name) + continue + case err != nil: + // Reported rather than skipped: a blip is not evidence the bucket + // went away, and a partial list would have the caller tear down + // the watchers it left out. + return nil, fmt.Errorf("resolve intent bucket %s: %w", name, err) + } + // js is passed so a watcher whose stream was lost can reconnect; + // RecreateIfMissing stays false so that reconnect never creates. + buckets = append(buckets, kvstore.NewOpenBucket(js, kv, kvstore.Config{Name: name})) + } + return buckets, nil + }, ">") +} + +// DriftLoop reconciles on every intent change and every DriftInterval, gated on +// AcquireLeader so only one vpcd scans at a time. An incomplete pass asks to be +// revisited 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. +// this node did not run one. It seeds the first deadline rather than triggering +// a pass: that bootstrap pass is the startup scan, and running a second one here +// would duplicate it and pull a scan forward on every clean boot. func DriftLoop(ctx context.Context, rec Reconciler, nc *nats.Conn, localAZ, holder string, startup error) { js, err := jetstream.New(nc) if err != nil { @@ -40,30 +91,53 @@ func DriftLoop(ctx context.Context, rec Reconciler, nc *nats.Conn, localAZ, hold return } - backoff := nextDriftBackoff(0, startup) - timer := time.NewTimer(driftWait(backoff)) - defer timer.Stop() + loop.Run(ctx, loop.Config{ + Name: "vpcd/drift", + Sources: []loop.Source{intentSource(js)}, + Reconcile: driftPass(rec, nc, js, localAZ, holder, startup), + Resync: DriftInterval, + }) +} - for { - select { - case <-ctx.Done(): - return - 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) +// driftPass returns the reconcile pass, closing over the backoff ladder and the +// last pass time. Run never calls it concurrently with itself, which is why +// neither needs a lock. +func driftPass(rec Reconciler, nc *nats.Conn, js jetstream.JetStream, localAZ, holder string, startup error) func(context.Context) (time.Duration, error) { + var backoff time.Duration + var lastPass time.Time + seeded := false + + return func(ctx context.Context) (time.Duration, error) { + if !seeded { + seeded = true + backoff = nextDriftBackoff(0, startup) + return backoff, nil } + // A change-driven pass is a repair check, not the apply path: the event + // subscriber has already applied the change, so coalescing a burst behind + // the floor the ladder starts at costs a scan nothing it would have + // found. Without it a launch storm scans continuously. + if wait := time.Until(lastPass.Add(driftBackoffBase)); wait > 0 { + return wait, nil + } + previous := backoff + backoff = nextDriftBackoff(backoff, runDriftCycle(ctx, rec, nc, js, localAZ, holder)) + // Stamped after the scan, not before: the floor is a gap between passes, + // and a pass slower than the floor would otherwise leave none at all. + lastPass = time.Now() + if backoff != previous { + slog.Info("reconcile/drift: requeue interval changed", "backoff_ms", otelsetup.Millis(backoff)) + } + // The outcome is logged by the cycle, and an unconverged pass is a change + // of schedule rather than a failure of the loop. + return backoff, nil } } // 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. +// held at DriftInterval. Zero means no deadline, leaving a converged loop to the +// resync rather than asking for a revisit it does not need. func nextDriftBackoff(current time.Duration, err error) time.Duration { if !errors.Is(err, ErrPassIncomplete) { return 0 @@ -75,16 +149,7 @@ func nextDriftBackoff(current time.Duration, err error) time.Duration { 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. +// runDriftCycle is one pass body, split out so tests can drive it directly. // 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 { diff --git a/spinifex/network/reconcile/drift_test.go b/spinifex/network/reconcile/drift_test.go index 53b9ba1d9..53a394e60 100644 --- a/spinifex/network/reconcile/drift_test.go +++ b/spinifex/network/reconcile/drift_test.go @@ -1,6 +1,6 @@ //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. +//nextDriftBackoff) and the tests drive runDriftCycle directly to pin what +//DriftLoop does with its return value. package reconcile @@ -200,34 +200,31 @@ func TestRunDriftCycle_SkipsWhenNotElected(t *testing.T) { } } -// 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. +// The ladder itself, as a pure function: exact gaps are asserted here because +// the loop tests above run on compressed timings and can only prove ordering. +// Zero is the converged value and means no deadline, which leaves the next pass +// to the resync the loop configures at DriftInterval. 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 + var got []time.Duration for _, outcome := range outcomes { backoff = nextDriftBackoff(backoff, outcome) - waits = append(waits, driftWait(backoff)) + got = append(got, 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 + 0, // converged: no deadline, the resync takes over } 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) + if got[i] != want[i] { + t.Fatalf("backoff after pass %d = %v, want %v (full sequence %v)", i+1, got[i], want[i], got) } } - // 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 @@ -247,9 +244,6 @@ func TestDrift_NonIncompleteOutcomesKeepDriftInterval(t *testing.T) { 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) - } }) } } diff --git a/spinifex/network/reconcile/drift_watch_test.go b/spinifex/network/reconcile/drift_watch_test.go new file mode 100644 index 000000000..4feaa5439 --- /dev/null +++ b/spinifex/network/reconcile/drift_watch_test.go @@ -0,0 +1,234 @@ +//test:in-package — intentSource, intentBuckets and driftBackoffBase are +//unexported, and the point of these tests is what the loop watches rather than +//what it does once woken. + +package reconcile + +import ( + "os" + "regexp" + "strconv" + "testing" + "time" + + handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw" + handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" + "github.com/mulgadc/spinifex/spinifex/kvutil" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go/jetstream" +) + +// createIntentBucket makes one intent bucket the way its owning handler would, +// so a test can write to it and have the source find it. +func createIntentBucket(t *testing.T, js jetstream.JetStream, name string) jetstream.KeyValue { + t.Helper() + kv, err := kvutil.GetOrCreateBucket(t.Context(), js, name, 1) + if err != nil { + t.Fatalf("create bucket %s: %v", name, err) + } + return kv +} + +// bucketNames is what intentSource resolved this cycle. +func bucketNames(t *testing.T, js jetstream.JetStream) []string { + t.Helper() + buckets, err := intentSource(js).Buckets(t.Context()) + if err != nil { + t.Fatalf("enumerate intent buckets: %v", err) + } + names := make([]string, 0, len(buckets)) + for _, b := range buckets { + names = append(names, b.Name()) + } + return names +} + +// waitForWatch writes probe keys until one of them wakes a pass, and returns the +// call count once it has. Watchers attach asynchronously and are UpdatesOnly, so +// a single write racing startup is lost for good; retrying is what makes a test +// that needs a live watcher deterministic rather than dependent on how fast the +// machine got there. +func waitForWatch(t *testing.T, rec *stubReconciler, kv jetstream.KeyValue, prefix string) int { + t.Helper() + before := rec.callCount() + deadline := time.Now().Add(10 * time.Second) + for i := 0; time.Now().Before(deadline); i++ { + if _, err := kv.Put(t.Context(), prefix+strconv.Itoa(i), []byte(`{}`)); err != nil { + t.Fatalf("probe write %d: %v", i, err) + } + if got := waitForCalls(t, rec, before+1, 250*time.Millisecond); got > before { + return got + } + } + t.Fatal("no intent write woke a pass within 10s, so the loop is still waiting " + + "out DriftInterval rather than watching the bucket") + return 0 +} + +// streamExists reports whether the KV bucket's backing stream is present, which +// is how a test tells "watched" from "created by being watched". +func streamExists(t *testing.T, js jetstream.JetStream, name string) bool { + t.Helper() + _, err := js.Stream(t.Context(), "KV_"+name) + return err == nil +} + +// The whole point of the conversion: a write to an intent bucket must reach the +// loop long before DriftInterval, because that write is the repair signal a +// dropped vpc.* event would otherwise leave for the next scan. +func TestDriftLoop_IntentWriteWakesAPass(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + kv := createIntentBucket(t, js, handlers_ec2_vpc.KVBucketVPCs) + // The interval is long so a pass can only have come from the watch, and the + // floor is short so the write is not merely deferred behind it. + shrinkDriftTiming(t, time.Minute, time.Millisecond) + + rec := &stubReconciler{outcomes: []error{nil}} + startDriftLoop(t, rec, nc, nil) + + // After the startup seed the loop is armed at DriftInterval alone, so the + // watch has to be what brings the pass forward. + time.Sleep(200 * time.Millisecond) + if got := rec.callCount(); got != 0 { + t.Fatalf("reconcile ran %d times before any write, want 0", got) + } + + // Well inside DriftInterval, so the pass can only have come from the write. + waitForWatch(t, rec, kv, "vpc-watch-") +} + +// A burst is one pass, not one per key: the debounce coalesces it and the floor +// keeps a launch storm from turning a five-minute OVN scan into a running one. +func TestDriftLoop_BurstOfWritesIsOnePass(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + kv := createIntentBucket(t, js, handlers_ec2_vpc.KVBucketENIs) + shrinkDriftTiming(t, time.Minute, 2*time.Second) + + rec := &stubReconciler{outcomes: []error{nil}} + startDriftLoop(t, rec, nc, nil) + + // The burst has to land on a watcher that is already attached, or the writes + // it is meant to coalesce are simply never seen. + base := waitForWatch(t, rec, kv, "eni-probe-") + + for i := range 20 { + if _, err := kv.Put(t.Context(), "eni-burst-"+strconv.Itoa(i), []byte(`{}`)); err != nil { + t.Fatalf("write %d: %v", i, err) + } + } + if got := waitForCalls(t, rec, base+1, 10*time.Second); got < base+1 { + t.Fatal("a burst of intent writes never woke a pass") + } + // The burst is long spent, so a further pass here could only come from the + // loop tracking writes rather than coalescing them. + time.Sleep(500 * time.Millisecond) + if got := rec.callCount() - base; got != 1 { + t.Errorf("reconcile ran %d times for one burst of 20 writes, want 1", got) + } +} + +// Watching must not create. These buckets belong to the EC2 handlers, which set +// their history depth; a bucket vpcd created first would fix the wrong depth in +// place, and the handler's own get-or-create would then return it rather than +// correct it. +func TestIntentSource_DoesNotCreateMissingBuckets(t *testing.T) { + _, _, js := testutil.StartTestJetStream(t) + + if got := bucketNames(t, js); len(got) != 0 { + t.Fatalf("intentSource resolved %v against an empty JetStream, want none", got) + } + for _, name := range intentBuckets { + if streamExists(t, js, name) { + t.Errorf("bucket %s exists after enumeration, want it left absent", name) + } + } +} + +// A cluster that has never had an IGW has no IGW bucket, so the source has to +// pick one up when it appears rather than only at startup. That is why it is a +// Dynamic source: JetStream has no bucket-created event, so discovery rides the +// resync. +func TestIntentSource_PicksUpABucketThatAppearsLater(t *testing.T) { + _, _, js := testutil.StartTestJetStream(t) + createIntentBucket(t, js, handlers_ec2_vpc.KVBucketVPCs) + + if got := bucketNames(t, js); len(got) != 1 || got[0] != handlers_ec2_vpc.KVBucketVPCs { + t.Fatalf("intentSource resolved %v, want just %s", got, handlers_ec2_vpc.KVBucketVPCs) + } + + createIntentBucket(t, js, handlers_ec2_igw.KVBucketIGW) + got := bucketNames(t, js) + if len(got) != 2 { + t.Fatalf("intentSource resolved %v after a second bucket appeared, want both", got) + } +} + +// bucketConstants is every handler KV bucket constant referenced by a file. +func bucketConstants(t *testing.T, path string) map[string]struct{} { + t.Helper() + src, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %s: %v", path, err) + } + out := map[string]struct{}{} + for _, m := range regexp.MustCompile(`handlers_ec2_\w+\.KVBucket\w+`).FindAllString(string(src), -1) { + out[m] = struct{}{} + } + return out +} + +// Every bucket LoadIntentFromKV reads must be watched, or a change to it waits +// out the resync while every other kind of change is repaired in seconds. The +// two lists are compared at the source rather than by name, so a load added to +// intent.go and not to intentBuckets fails here instead of going quiet. +func TestIntentSource_CoversEveryBucketTheIntentLoadReads(t *testing.T) { + read := bucketConstants(t, "intent.go") + if len(read) == 0 { + t.Fatal("no bucket constants found in intent.go — the scan is not matching what it should") + } + watched := bucketConstants(t, "drift.go") + for name := range read { + if _, ok := watched[name]; !ok { + t.Errorf("%s is read by LoadIntentFromKV but absent from intentBuckets, "+ + "so a change to it is only repaired on the resync", name) + } + } +} + +// The floor is what stops a busy cluster scanning continuously, so it is pinned +// on the pass rather than inferred from the loop's timing. +func TestDriftPass_FloorsTheChangeDrivenRate(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + shrinkDriftTiming(t, time.Minute, time.Hour) + + rec := &stubReconciler{outcomes: []error{nil}} + pass := driftPass(rec, nc, js, "us-east-1a", "node-1", nil) + + // The seed pass does no work: the bootstrap ReconcileApplyOnly was the + // startup scan and this one must not repeat it. + if revisit, err := pass(t.Context()); err != nil || revisit != 0 { + t.Fatalf("seed pass = (%v, %v), want (0, nil) for a converged startup", revisit, err) + } + if got := rec.callCount(); got != 0 { + t.Fatalf("seed pass ran reconcile %d times, want 0", got) + } + + if _, err := pass(t.Context()); err != nil { + t.Fatalf("first real pass: %v", err) + } + if got := rec.callCount(); got != 1 { + t.Fatalf("reconcile ran %d times, want 1", got) + } + + revisit, err := pass(t.Context()) + if err != nil { + t.Fatalf("second pass: %v", err) + } + if got := rec.callCount(); got != 1 { + t.Errorf("reconcile ran %d times inside the floor, want it deferred to 1", got) + } + if revisit <= 0 || revisit > driftBackoffBase { + t.Errorf("deferred pass asked to be revisited in %v, want a positive wait "+ + "no longer than the floor (%v)", revisit, driftBackoffBase) + } +}