From e9404f1d7057da0015334ca1454184a7eb9030bb Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Mon, 31 Aug 2026 14:50:32 +1000 Subject: [PATCH] feat: wake the EKS cluster reconciler on the signals it already receives (#952) The loop polled every 30s for two things it was already being told about: the bootstrap artifacts are keys in the account bucket a watch can see land, and the control plane's health arrives over core NATS into a field the reconciler read only on the next tick. It now runs on reconciler.Run over a watch of the account bucket, plus a new Config.Trigger fed by the state-report subscription, and returns a revisit deadline for what elapsed time alone decides: the create timeout, and the instant the newest report goes stale. A degraded cluster keeps the old interval rather than restating the three recovery ladders' grace and backoff arithmetic in a second place. The trigger only fires on a report that changes health. The control plane publishes on its own timer regardless, so forwarding every report would have been the same 30s tick under another name. --- spinifex/handlers/eks/cluster_reconciler.go | 243 ++++++++++++++---- .../eks/cluster_reconciler_recovery_test.go | 2 +- .../eks/cluster_reconciler_replace_test.go | 2 +- .../eks/cluster_reconciler_state_test.go | 14 +- .../handlers/eks/cluster_reconciler_test.go | 13 +- spinifex/handlers/eks/revisit_test.go | 183 +++++++++++++ spinifex/reconciler/reconciler.go | 31 +++ spinifex/reconciler/trigger_test.go | 152 +++++++++++ 8 files changed, 578 insertions(+), 62 deletions(-) create mode 100644 spinifex/handlers/eks/revisit_test.go create mode 100644 spinifex/reconciler/trigger_test.go diff --git a/spinifex/handlers/eks/cluster_reconciler.go b/spinifex/handlers/eks/cluster_reconciler.go index 65852cbf2..51bb0695c 100644 --- a/spinifex/handlers/eks/cluster_reconciler.go +++ b/spinifex/handlers/eks/cluster_reconciler.go @@ -6,13 +6,17 @@ import ( "errors" "fmt" "log/slog" + "maps" "net/http" "strings" + "sync" "sync/atomic" "time" "github.com/mulgadc/spinifex/spinifex/kvlease" + "github.com/mulgadc/spinifex/spinifex/kvstore" "github.com/mulgadc/spinifex/spinifex/otelsetup" + "github.com/mulgadc/spinifex/spinifex/reconciler" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -21,7 +25,11 @@ import ( const ( defaultReconcileLeaseRefresh = 20 * time.Second defaultReconcileInterval = 30 * time.Second - defaultHealthzTimeout = 5 * time.Second + // reconcileResync is the drift backstop for a loop whose signals now arrive as + // watch updates and state reports. It also bounds every deadline below it, so + // the 15-minute create timeout is served by the resync rather than by a timer. + reconcileResync = 5 * time.Minute + defaultHealthzTimeout = 5 * time.Second // defaultCreateTimeout bounds how long a cluster may sit in CREATING before // being marked FAILED. Measured from meta.CreatedAt so a VM that never boots // reaches a terminal state even across daemon restarts. @@ -148,6 +156,17 @@ type ClusterReconciler struct { stateSubject string stateStaleAfter time.Duration latest atomic.Pointer[ServerStateReport] + // wake carries a report that changed the cluster's health into the reconcile + // loop, which watches KV and would otherwise never see it: the report arrives + // over core NATS and writes nothing. + wake chan struct{} + + // terminal records why the loop is ending — a terminal cluster status or a + // lost lease — and stopRun cancels it. reconciler.Run has no exit condition + // of its own, so ending it is cancelling it, and this is what Run reports. + terminalMu sync.Mutex + terminal error + stopRun context.CancelFunc // Add-on delivery status source. When addonStatusSub is non-nil the // reconciler subscribes to the per-cluster add-on status subject and CASes @@ -334,17 +353,20 @@ func NewClusterReconciler(leaderKV, acctKV jetstream.KeyValue, accountID, cluste return nil, errors.New("eks: NewClusterReconciler empty holderID") } r := &ClusterReconciler{ - leaderKV: leaderKV, - acctKV: acctKV, - accountID: accountID, - clusterName: clusterName, - holderID: holderID, - healthURL: healthURL, - leaseRefresh: defaultReconcileLeaseRefresh, - interval: defaultReconcileInterval, - healthTimeout: defaultHealthzTimeout, - createTimeout: defaultCreateTimeout, - stateStaleAfter: defaultStateStaleAfter, + leaderKV: leaderKV, + acctKV: acctKV, + accountID: accountID, + clusterName: clusterName, + holderID: holderID, + healthURL: healthURL, + leaseRefresh: defaultReconcileLeaseRefresh, + interval: defaultReconcileInterval, + healthTimeout: defaultHealthzTimeout, + createTimeout: defaultCreateTimeout, + stateStaleAfter: defaultStateStaleAfter, + // One slot: it means "a report changed something", and a second one + // arriving before the pass runs says nothing the first did not. + wake: make(chan struct{}, 1), restartGrace: defaultCPRestartGrace, restartBackoff: defaultCPRestartBackoff, maxRestartAttempts: defaultMaxCPRestartAttempts, @@ -402,6 +424,10 @@ func (r *ClusterReconciler) Run(ctx context.Context) error { if r.lease == nil { return errors.New("ClusterReconciler: Run called without AcquireLease") } + runCtx, stop := context.WithCancel(ctx) + defer stop() + r.stopRun = stop + if r.stateSub != nil && r.stateSubject != "" { sub, err := r.stateSub.Subscribe(r.stateSubject, func(m *nats.Msg) { report, perr := unmarshalServerStateReport(m.Data) @@ -410,7 +436,7 @@ func (r *ClusterReconciler) Run(ctx context.Context) error { "cluster", r.clusterName, "subject", m.Subject, "err", perr) return } - r.latest.Store(&report) + r.storeReport(&report) }) if err != nil { return fmt.Errorf("subscribe state report %s: %w", r.stateSubject, err) @@ -434,34 +460,95 @@ func (r *ClusterReconciler) Run(ctx context.Context) error { defer func() { _ = sub.Unsubscribe() }() } - reconcileT := time.NewTicker(r.interval) - defer reconcileT.Stop() - - lost := r.lease.Lost() - if err := r.reconcileOnce(ctx); err != nil { - if terminalReconcileErr(err) { - return err - } - slog.Warn("ClusterReconciler: initial reconcile failed", - "cluster", r.clusterName, "err", err) - } - - for { + // The lease turning over is not a KV write this loop watches, so losing it has + // to end the loop from outside it. + go func() { select { - case <-ctx.Done(): - return ctx.Err() - case <-lost: - return ErrReconcilerLeaseLost - case <-reconcileT.C: - if err := r.reconcileOnce(ctx); err != nil { - if terminalReconcileErr(err) { - return err - } - slog.Warn("ClusterReconciler: reconcile failed", - "cluster", r.clusterName, "err", err) - } + case <-runCtx.Done(): + case <-r.lease.Lost(): + r.endRun(ErrReconcilerLeaseLost) } + }() + + // The whole account bucket, and that is forced rather than lazy: a cluster key + // is "clusters//meta", "/" is not the NATS subject separator, so the key + // is one token and no prefix wildcard selects a single cluster. The cost is + // waking on a sibling cluster's writes, which real activity bounds. + bucket := kvstore.NewOpenBucket(nil, r.acctKV, kvstore.Config{Name: r.acctKV.Bucket()}) + reconciler.Run(runCtx, reconciler.Config{ + Name: "eks/" + r.clusterName, + Sources: []reconciler.Source{reconciler.Fixed(bucket, ">")}, + Reconcile: r.reconcilePass, + Trigger: r.wake, + Resync: reconcileResync, + }) + + if err := r.terminalErr(); err != nil { + return err } + return ctx.Err() +} + +// reconcilePass runs one reconcile and says when the loop should next run with +// nothing arriving. A terminal outcome ends the loop rather than being returned: +// the loop has no notion of finishing, so finishing it is cancelling it. +func (r *ClusterReconciler) reconcilePass(ctx context.Context) (time.Duration, error) { + revisit, err := r.reconcileOnce(ctx) + switch { + case err == nil: + return revisit, nil + case terminalReconcileErr(err): + r.endRun(err) + return 0, nil + default: + // Retried at the old cadence rather than left to the resync: a pass that + // failed has learnt nothing, so waiting five minutes to try again would be + // slower than the ticker this replaced. + return r.interval, err + } +} + +// endRun records the first reason the loop ended and cancels it. First wins: a +// lost lease and a terminal status can land together, and which one is reported +// should not depend on the race. +func (r *ClusterReconciler) endRun(reason error) { + r.terminalMu.Lock() + if r.terminal == nil { + r.terminal = reason + } + r.terminalMu.Unlock() + if r.stopRun != nil { + r.stopRun() + } +} + +func (r *ClusterReconciler) terminalErr() error { + r.terminalMu.Lock() + defer r.terminalMu.Unlock() + return r.terminal +} + +// storeReport keeps the newest control-plane report and wakes the loop only when +// it says something about health the last one did not. The control plane +// publishes on its own timer whether or not anything changed, so waking on every +// report would be the old tick again under another name. +func (r *ClusterReconciler) storeReport(report *ServerStateReport) { + prev := r.latest.Swap(report) + if prev != nil && sameHealth(prev, report) { + return + } + select { + case r.wake <- struct{}{}: + default: + } +} + +// sameHealth reports whether two reports would produce the same observe result. +// Freshness is deliberately excluded: a report that only repeats itself pushes +// the staleness deadline out, which the next pass reads for itself. +func sameHealth(a, b *ServerStateReport) bool { + return a.Healthz == b.Healthz && a.Reason == b.Reason && + a.NodeCount == b.NodeCount && maps.Equal(a.NodegroupReady, b.NodegroupReady) } func terminalReconcileErr(err error) bool { @@ -470,40 +557,45 @@ func terminalReconcileErr(err error) bool { errors.Is(err, ErrClusterNotFound) } -func (r *ClusterReconciler) reconcileOnce(ctx context.Context) error { +// reconcileOnce runs one pass and reports how long the loop may wait with +// nothing arriving. The signals that do arrive — a bootstrap artifact landing, a +// status change, a control plane changing its mind about its own health — are +// watched or triggered, so the deadline covers only what elapsed time decides. +func (r *ClusterReconciler) reconcileOnce(ctx context.Context) (time.Duration, error) { meta, err := GetClusterMeta(ctx, r.acctKV, r.clusterName) if err != nil { - return err + return 0, err } switch meta.Status { case ClusterStatusCreating: ready, reason := r.bootstrapReady(ctx, meta) if !ready { - // Info, not Debug: this is the per-tick reason a CREATING cluster has + // Info, not Debug: this is the per-pass reason a CREATING cluster has // not yet flipped ACTIVE. At Debug it is invisible at the default // level, so a cluster that stalls to the create-timeout FAILED gives // no diagnosable cause. Logged only during the CREATING window. slog.Info("ClusterReconciler: bootstrap not ready", "cluster", r.clusterName, "reason", reason) - return r.failIfCreateTimedOut(ctx, meta, "bootstrap not ready: "+reason) + return r.createRevisit(meta), r.failIfCreateTimedOut(ctx, meta, "bootstrap not ready: "+reason) } issue, nodeCount, nodegroupReady := r.observe(ctx) if issue != "" { slog.Info("ClusterReconciler: health not ready", "cluster", r.clusterName, "issue", issue) - return r.failIfCreateTimedOut(ctx, meta, "healthz not ready: "+issue) + return r.createRevisit(meta), r.failIfCreateTimedOut(ctx, meta, "healthz not ready: "+issue) } if err := SetClusterStatus(ctx, r.acctKV, r.clusterName, ClusterStatusActive); err != nil { - return err + return 0, err } if err := SetClusterHealthState(ctx, r.acctKV, r.clusterName, "", nodeCount, nodegroupReady); err != nil { if errors.Is(err, ErrClusterNotFound) { - return ErrClusterNotFound + return 0, ErrClusterNotFound } - return fmt.Errorf("record cluster health: %w", err) + return 0, fmt.Errorf("record cluster health: %w", err) } slog.Info("ClusterReconciler: cluster transitioned to ACTIVE", "cluster", r.clusterName, "nodes", nodeCount) + return r.activeRevisit(""), nil case ClusterStatusActive: issue, nodeCount, nodegroupReady := r.observe(ctx) if issue != "" { @@ -512,19 +604,68 @@ func (r *ClusterReconciler) reconcileOnce(ctx context.Context) error { } if err := SetClusterHealthState(ctx, r.acctKV, r.clusterName, issue, nodeCount, nodegroupReady); err != nil { if errors.Is(err, ErrClusterNotFound) { - return ErrClusterNotFound + return 0, ErrClusterNotFound } - return fmt.Errorf("record cluster health: %w", err) + return 0, fmt.Errorf("record cluster health: %w", err) } r.maybeRecoverControlPlane(ctx, meta, issue) r.maybeReformEtcdQuorum(ctx, meta, issue) r.maybeReplaceControlPlaneMember(ctx, meta, issue) + return r.activeRevisit(issue), nil case ClusterStatusDeleting: - return ErrReconcilerClusterDeleting + return 0, ErrReconcilerClusterDeleting case ClusterStatusFailed: - return ErrReconcilerClusterFailed + return 0, ErrReconcilerClusterFailed } - return nil + return 0, nil +} + +// createRevisit is what remains of the create timeout. Everything else a +// CREATING cluster waits for is an event now: the bootstrap artifacts are keys +// in the watched bucket, and the first healthy report is a trigger. Only the +// timeout expiring announces itself to nobody. +func (r *ClusterReconciler) createRevisit(meta *ClusterMeta) time.Duration { + if r.createTimeout <= 0 || meta.CreatedAt.IsZero() { + return 0 + } + if remaining := time.Until(meta.CreatedAt.Add(r.createTimeout)); remaining > 0 { + return remaining + } + // Already due and still not failed, so the pass that fails it could not. Retry + // at the old cadence rather than immediately, which would spin. + return r.interval +} + +// activeRevisit is the deadline for a cluster that is already up. +func (r *ClusterReconciler) activeRevisit(issue string) time.Duration { + if issue != "" { + // A degraded cluster is on the recovery ladders' clocks — three graces, + // three backoffs and three attempt caps that interact — and queries its + // members' VM state every pass anyway. Keeping the old interval leaves + // that timing exactly as it is rather than restating it here. + return r.interval + } + return r.reportRevisit() +} + +// reportRevisit is the instant the newest control-plane report goes stale, which +// is what a healthy cluster is waiting for: silence is published by nobody, so +// nothing else will wake the loop to notice it. +func (r *ClusterReconciler) reportRevisit() time.Duration { + if r.stateSub == nil || r.stateSubject == "" { + // Health comes from an HTTP probe, which answers only when asked. + return r.interval + } + report := r.latest.Load() + if report == nil { + return r.interval + } + if fresh := time.Until(time.Unix(report.TS, 0).Add(r.stateStaleAfter)); fresh > 0 { + return fresh + } + // Already stale, so there is no expiry left to wait for and the next pass is + // what notices. Returning nothing here would leave that pass unscheduled. + return r.interval } // maybeRecoverControlPlane attempts a bounded in-place restart of a wedged diff --git a/spinifex/handlers/eks/cluster_reconciler_recovery_test.go b/spinifex/handlers/eks/cluster_reconciler_recovery_test.go index 58d070940..3959c9993 100644 --- a/spinifex/handlers/eks/cluster_reconciler_recovery_test.go +++ b/spinifex/handlers/eks/cluster_reconciler_recovery_test.go @@ -189,7 +189,7 @@ func TestActiveBranchRestartsWedgedCP(t *testing.T) { r.degradedSince = time.Now().Add(-10 * time.Minute) r.latest.Store(&ServerStateReport{Healthz: "ok", NodeCount: 2, TS: time.Now().Add(-5 * time.Minute).Unix()}) - require.NoError(t, r.reconcileOnce(context.Background())) + require.NoError(t, onePass(t, r)) meta, err := GetClusterMeta(t.Context(), acctKV, "alpha") require.NoError(t, err) diff --git a/spinifex/handlers/eks/cluster_reconciler_replace_test.go b/spinifex/handlers/eks/cluster_reconciler_replace_test.go index 99e1c5b34..ec81eeb4a 100644 --- a/spinifex/handlers/eks/cluster_reconciler_replace_test.go +++ b/spinifex/handlers/eks/cluster_reconciler_replace_test.go @@ -326,7 +326,7 @@ func TestActiveBranchReplacesLostCPMember(t *testing.T) { r.replacingSince = time.Now().Add(-10 * time.Minute) r.latest.Store(freshReport("ok", 3)) - require.NoError(t, r.reconcileOnce(context.Background())) + require.NoError(t, onePass(t, r)) require.Equal(t, 1, prov.calls, "wedged member replaced through the ACTIVE arm") got, err := GetClusterMeta(t.Context(), acctKV, "alpha") diff --git a/spinifex/handlers/eks/cluster_reconciler_state_test.go b/spinifex/handlers/eks/cluster_reconciler_state_test.go index 94f613367..badbe5ba0 100644 --- a/spinifex/handlers/eks/cluster_reconciler_state_test.go +++ b/spinifex/handlers/eks/cluster_reconciler_state_test.go @@ -43,7 +43,7 @@ func TestClusterReconciler_CreatingTransitionsToActiveOnStateReport(t *testing.T seedBootstrapState(t, acctKV) r.latest.Store(freshReport("ok", 3)) - require.NoError(t, r.reconcileOnce(context.Background())) + require.NoError(t, onePass(t, r)) meta, err := GetClusterMeta(t.Context(), acctKV, "alpha") require.NoError(t, err) @@ -58,7 +58,7 @@ func TestClusterReconciler_CreatingStaysWithoutStateReport(t *testing.T) { seedBootstrapState(t, acctKV) // No report stored. - require.NoError(t, r.reconcileOnce(context.Background())) + require.NoError(t, onePass(t, r)) meta, err := GetClusterMeta(t.Context(), acctKV, "alpha") require.NoError(t, err) @@ -71,7 +71,7 @@ func TestClusterReconciler_CreatingStaysOnUnhealthyReport(t *testing.T) { seedBootstrapState(t, acctKV) r.latest.Store(freshReport("fail", 0)) - require.NoError(t, r.reconcileOnce(context.Background())) + require.NoError(t, onePass(t, r)) meta, err := GetClusterMeta(t.Context(), acctKV, "alpha") require.NoError(t, err) @@ -84,7 +84,7 @@ func TestClusterReconciler_ActiveRecordsNodeCountAndClearsIssue(t *testing.T) { require.NoError(t, SetClusterHealthState(t.Context(), acctKV, "alpha", "stale", 0, nil)) r.latest.Store(freshReport("ok", 4)) - require.NoError(t, r.reconcileOnce(context.Background())) + require.NoError(t, onePass(t, r)) meta, err := GetClusterMeta(t.Context(), acctKV, "alpha") require.NoError(t, err) @@ -98,7 +98,7 @@ func TestClusterReconciler_ActiveFlagsStaleReport(t *testing.T) { require.NoError(t, SetClusterStatus(t.Context(), acctKV, "alpha", ClusterStatusActive)) r.latest.Store(&ServerStateReport{Healthz: "ok", NodeCount: 2, TS: time.Now().Add(-5 * time.Minute).Unix()}) - require.NoError(t, r.reconcileOnce(context.Background())) + require.NoError(t, onePass(t, r)) meta, err := GetClusterMeta(t.Context(), acctKV, "alpha") require.NoError(t, err) @@ -114,7 +114,7 @@ func TestClusterReconciler_ActiveUnhealthyReportSurfacesReason(t *testing.T) { Reason: "readyz:[etcd]; etcd:unreachable; disk:ok", }) - require.NoError(t, r.reconcileOnce(context.Background())) + require.NoError(t, onePass(t, r)) meta, err := GetClusterMeta(t.Context(), acctKV, "alpha") require.NoError(t, err) @@ -128,7 +128,7 @@ func TestClusterReconciler_ActiveUnhealthyReportWithoutReasonStaysTerse(t *testi require.NoError(t, SetClusterStatus(t.Context(), acctKV, "alpha", ClusterStatusActive)) r.latest.Store(freshReport("fail", 0)) - require.NoError(t, r.reconcileOnce(context.Background())) + require.NoError(t, onePass(t, r)) meta, err := GetClusterMeta(t.Context(), acctKV, "alpha") require.NoError(t, err) diff --git a/spinifex/handlers/eks/cluster_reconciler_test.go b/spinifex/handlers/eks/cluster_reconciler_test.go index 33170eb89..3da6987b6 100644 --- a/spinifex/handlers/eks/cluster_reconciler_test.go +++ b/spinifex/handlers/eks/cluster_reconciler_test.go @@ -45,6 +45,15 @@ func (s *stubHTTPDoer) setResponse(status int, err error) { s.err = err } +// onePass runs one reconcile and keeps only its error, for the tests that assert +// on what converged rather than on when the loop should next run. The tests that +// do care about the deadline call reconcileOnce directly. +func onePass(t *testing.T, r *ClusterReconciler) error { + t.Helper() + _, err := r.reconcileOnce(t.Context()) + return err +} + func newReconcilerHarness(t *testing.T, healthURL string, opts ...ReconcilerOption) (*ClusterReconciler, jetstream.KeyValue, jetstream.KeyValue) { t.Helper() _, nc, _ := testutil.StartTestJetStream(t) @@ -418,14 +427,14 @@ func TestClusterReconciler_ActiveHealthRecoversClearsIssue(t *testing.T) { // First probe fails and records the issue. require.Error(t, r.probeHealthz(context.Background())) - require.NoError(t, r.reconcileOnce(context.Background())) + require.NoError(t, onePass(t, r)) meta, err := GetClusterMeta(t.Context(), acctKV, "alpha") require.NoError(t, err) require.NotEmpty(t, meta.HealthIssue) // Probe recovers: the issue must be cleared. stub.setResponse(http.StatusOK, nil) - require.NoError(t, r.reconcileOnce(context.Background())) + require.NoError(t, onePass(t, r)) meta, err = GetClusterMeta(t.Context(), acctKV, "alpha") require.NoError(t, err) assert.Empty(t, meta.HealthIssue, "recovered /healthz must clear the recorded issue") diff --git a/spinifex/handlers/eks/revisit_test.go b/spinifex/handlers/eks/revisit_test.go new file mode 100644 index 000000000..49a7cc8c8 --- /dev/null +++ b/spinifex/handlers/eks/revisit_test.go @@ -0,0 +1,183 @@ +package handlers_eks + +//test:in-package — asserts on the deadline reconcileOnce returns and on when a +// state report wakes the loop, both of which are unexported and are the whole +// subject of these tests. + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// woke reports whether storeReport signalled, draining the channel so a later +// call in the same test starts from quiet. +func woke(r *ClusterReconciler) bool { + select { + case <-r.wake: + return true + default: + return false + } +} + +// A CREATING cluster's artifacts are keys in the watched bucket and its first +// healthy report is a trigger, so the timeout expiring is the only thing left +// that announces itself to nobody. +func TestReconcileOnce_ACreatingClusterAsksForWhatIsLeftOfItsCreateTimeout(t *testing.T) { + r, _, acctKV := newStateReconcilerHarness(t, WithCreateTimeout(15*time.Minute)) + freshenClusterCreatedAt(t, acctKV) + // No bootstrap state and no report, so the cluster stays CREATING. + + revisit, err := r.reconcileOnce(t.Context()) + require.NoError(t, err) + + assert.Greater(t, revisit, 14*time.Minute, "a cluster created seconds ago has nearly all of its window left") + assert.LessOrEqual(t, revisit, 15*time.Minute) +} + +// A create timeout already past belongs to the pass that fails the cluster. +// Returning nothing would leave that pass unscheduled. +func TestReconcileOnce_ACreatingClusterPastItsTimeoutAsksForTheInterval(t *testing.T) { + r, _, acctKV := newStateReconcilerHarness(t, + WithCreateTimeout(15*time.Minute), + WithReconcileInterval(30*time.Second), + ) + meta, err := GetClusterMeta(t.Context(), acctKV, "alpha") + require.NoError(t, err) + meta.CreatedAt = time.Now().UTC().Add(-time.Hour) + require.NoError(t, PutClusterMeta(t.Context(), acctKV, meta)) + + // The pass marks it FAILED, which is terminal and reported as such. + revisit, err := r.reconcileOnce(t.Context()) + require.ErrorIs(t, err, ErrReconcilerClusterFailed) + assert.Equal(t, 30*time.Second, revisit) +} + +// A healthy cluster is waiting for its control plane to go silent, and silence +// is published by nobody. The deadline is the instant the newest report expires, +// which lands the pass on it rather than up to a tick after it. +func TestReconcileOnce_AHealthyClusterAsksForItsReportsStalenessInstant(t *testing.T) { + r, _, acctKV := newStateReconcilerHarness(t, WithStateStaleAfter(90*time.Second)) + require.NoError(t, SetClusterStatus(t.Context(), acctKV, "alpha", ClusterStatusActive)) + r.latest.Store(freshReport("ok", 3)) + + revisit, err := r.reconcileOnce(t.Context()) + require.NoError(t, err) + + assert.Greater(t, revisit, 60*time.Second, "a report seconds old must not read as nearly stale") + assert.LessOrEqual(t, revisit, 90*time.Second) +} + +// A report already past its window is the case the next pass has to notice, so +// the deadline falls back to the interval rather than to nothing. +func TestReconcileOnce_AClusterWithAStaleReportAsksForTheInterval(t *testing.T) { + r, _, acctKV := newStateReconcilerHarness(t, + WithStateStaleAfter(90*time.Second), + WithReconcileInterval(30*time.Second), + ) + require.NoError(t, SetClusterStatus(t.Context(), acctKV, "alpha", ClusterStatusActive)) + r.latest.Store(&ServerStateReport{Healthz: "ok", NodeCount: 2, TS: time.Now().Add(-5 * time.Minute).Unix()}) + + revisit, err := r.reconcileOnce(t.Context()) + require.NoError(t, err) + + assert.Equal(t, 30*time.Second, revisit) +} + +// A degraded cluster keeps the old cadence: the recovery ladders run on elapsed +// time that this deliberately does not restate, and they poll their members' +// VM state on every pass regardless. +func TestReconcileOnce_ADegradedClusterAsksForTheInterval(t *testing.T) { + r, _, acctKV := newStateReconcilerHarness(t, WithReconcileInterval(30*time.Second)) + require.NoError(t, SetClusterStatus(t.Context(), acctKV, "alpha", ClusterStatusActive)) + r.latest.Store(freshReport("fail", 0)) + + revisit, err := r.reconcileOnce(t.Context()) + require.NoError(t, err) + + assert.Equal(t, 30*time.Second, revisit) +} + +// Without a state source health comes from an HTTP probe, which answers only +// when asked, so there is nothing to wait for and the poll has to stay. +func TestReconcileOnce_AProbedClusterAsksForTheInterval(t *testing.T) { + stub := &stubHTTPDoer{status: 200} + r, _, acctKV := newReconcilerHarness(t, "https://nlb.example/healthz", + WithHTTPClient(stub), + WithReconcileInterval(30*time.Second), + ) + require.NoError(t, SetClusterStatus(t.Context(), acctKV, "alpha", ClusterStatusActive)) + + revisit, err := r.reconcileOnce(t.Context()) + require.NoError(t, err) + + assert.Equal(t, 30*time.Second, revisit) +} + +// A terminal status ends the loop, so there is no next run to schedule. +func TestReconcileOnce_ADeletingClusterAsksForNoDeadline(t *testing.T) { + r, _, acctKV := newStateReconcilerHarness(t) + require.NoError(t, SetClusterStatus(t.Context(), acctKV, "alpha", ClusterStatusDeleting)) + + revisit, err := r.reconcileOnce(t.Context()) + require.ErrorIs(t, err, ErrReconcilerClusterDeleting) + assert.Zero(t, revisit) +} + +// The first report is always news: there is nothing to compare it against, and +// it is what flips a CREATING cluster. +func TestStoreReport_TheFirstReportWakesTheLoop(t *testing.T) { + r, _, _ := newStateReconcilerHarness(t) + + r.storeReport(freshReport("ok", 3)) + + assert.True(t, woke(r), "the first report must wake the loop") +} + +// The control plane publishes on its own timer whether or not anything changed. +// Waking on every report would be the old 30s tick under another name. +func TestStoreReport_ARepeatedReportDoesNotWakeTheLoop(t *testing.T) { + r, _, _ := newStateReconcilerHarness(t) + r.storeReport(&ServerStateReport{Healthz: "ok", NodeCount: 3, TS: time.Now().Unix()}) + require.True(t, woke(r)) + + r.storeReport(&ServerStateReport{Healthz: "ok", NodeCount: 3, TS: time.Now().Add(30 * time.Second).Unix()}) + + assert.False(t, woke(r), "a report saying the same thing later must not wake the loop") +} + +// The newest report still has to be stored even when it does not wake anything, +// because the staleness deadline is measured from it. +func TestStoreReport_ARepeatedReportStillRefreshesWhatIsStored(t *testing.T) { + r, _, _ := newStateReconcilerHarness(t) + r.storeReport(&ServerStateReport{Healthz: "ok", NodeCount: 3, TS: 1000}) + + r.storeReport(&ServerStateReport{Healthz: "ok", NodeCount: 3, TS: 2000}) + + assert.Equal(t, int64(2000), r.latest.Load().TS) +} + +func TestStoreReport_AChangedReportWakesTheLoop(t *testing.T) { + for _, tc := range []struct { + name string + next *ServerStateReport + }{ + {"healthz", &ServerStateReport{Healthz: "fail", NodeCount: 3}}, + {"reason", &ServerStateReport{Healthz: "ok", NodeCount: 3, Reason: "etcd:unreachable"}}, + {"node count", &ServerStateReport{Healthz: "ok", NodeCount: 2}}, + {"nodegroup counts", &ServerStateReport{Healthz: "ok", NodeCount: 3, NodegroupReady: map[string]int{"ng": 2}}}, + } { + t.Run(tc.name, func(t *testing.T) { + r, _, _ := newStateReconcilerHarness(t) + r.storeReport(&ServerStateReport{Healthz: "ok", NodeCount: 3}) + require.True(t, woke(r)) + + r.storeReport(tc.next) + + assert.True(t, woke(r), "a report that changes health must wake the loop") + }) + } +} diff --git a/spinifex/reconciler/reconciler.go b/spinifex/reconciler/reconciler.go index e11e59b5e..1d310c32c 100644 --- a/spinifex/reconciler/reconciler.go +++ b/spinifex/reconciler/reconciler.go @@ -12,6 +12,10 @@ // waiting on something no KV write announces — a VM reaching running, an agent // falling silent, a timeout expiring — which a watch cannot see at all, and for // which the resync alone is too blunt to be a failure detector. +// +// A caller whose inputs are not all KV can supply a Trigger channel as well, for +// the events that arrive by some other route and would otherwise be seen only +// when the next pass happened to look. package reconciler import ( @@ -89,6 +93,11 @@ type Config struct { // tombstone carries no value, so a caller whose work key lives in the value // cannot name it, and a whole-set pass needs no attribution. KeyFor func(entry jetstream.KeyValueEntry) (keys []string, ok bool) + // Trigger wakes the loop from outside the buckets, for the inputs a KV watch + // cannot see because they never reach KV. A receive is treated exactly like a + // watch update: debounced with them, then served by a whole-set pass, since an + // external signal names no key. Nil means the buckets are the only source. + Trigger <-chan struct{} // Resync bounds how stale the world can get when no update arrives, and is // also when Sources are re-enumerated. Zero means DefaultResync. Resync time.Duration @@ -174,6 +183,10 @@ func Run(ctx context.Context, cfg Config) { // pass that would have seen it ran before the watch existed. w.resync(ctx) + if cfg.Trigger != nil { + go forward(ctx, cfg.Trigger, w) + } + resync := time.NewTicker(cfg.Resync) defer resync.Stop() @@ -213,6 +226,24 @@ func Run(ctx context.Context, cfg Config) { } } +// forward turns an external wake into the same signal a watch update produces, +// so a trigger is debounced and coalesced with them rather than taking a second +// path through the loop. It queues the whole set because a trigger names no key. +func forward(ctx context.Context, trigger <-chan struct{}, w *watchSet) { + for { + select { + case <-ctx.Done(): + return + case _, open := <-trigger: + if !open { + return + } + w.queue.addWhole() + w.signal() + } + } +} + // clampRevisit bounds a deadline to the resync. A caller may ask to be revisited // sooner but never later: the resync is the outer bound on staleness, and is not // something a pass gets to extend. Non-positive means no deadline. diff --git a/spinifex/reconciler/trigger_test.go b/spinifex/reconciler/trigger_test.go new file mode 100644 index 000000000..fb8de5322 --- /dev/null +++ b/spinifex/reconciler/trigger_test.go @@ -0,0 +1,152 @@ +package reconciler_test + +import ( + "context" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/reconciler" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// The reason the trigger exists: a loop whose signal arrives by some other route +// than a KV write would otherwise see it only when the next pass happened to +// look, which is the poll the watch was supposed to replace. +func TestRun_ATriggerWakesTheLoopWithNoWatchTraffic(t *testing.T) { + store, _ := newBucket(t, "trigger-wakes") + p := newPasses() + trigger := make(chan struct{}, 1) + + run(t, reconciler.Config{ + Name: "trigger", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, + Trigger: trigger, + Resync: time.Hour, + }) + + p.waitFor(t, 1, "the startup pass") + trigger <- struct{}{} + + p.waitFor(t, 2, "a pass from the trigger, against a bucket nobody is writing to") +} + +// A quiet trigger must not be a source of passes on its own: the loop still +// falls back to the resync and nothing else. +func TestRun_AQuietTriggerAddsNoPasses(t *testing.T) { + store, _ := newBucket(t, "trigger-quiet") + p := newPasses() + trigger := make(chan struct{}, 1) + + run(t, reconciler.Config{ + Name: "quiet", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, + Trigger: trigger, + Resync: time.Hour, + }) + + p.waitFor(t, 1, "the startup pass") + p.stillAt(t, 1, "a trigger nobody signalled must not produce passes") +} + +// A burst of triggers is one pass, the same as a burst of watch updates. The +// caller signalling several times in a row is describing one thing to converge, +// not several. +func TestRun_ABurstOfTriggersCoalescesIntoOnePass(t *testing.T) { + store, _ := newBucket(t, "trigger-burst") + p := newPasses() + trigger := make(chan struct{}, 8) + + run(t, reconciler.Config{ + Name: "burst", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, + Trigger: trigger, + Resync: time.Hour, + Debounce: 100 * time.Millisecond, + }) + + p.waitFor(t, 1, "the startup pass") + for range 8 { + trigger <- struct{}{} + } + + p.waitFor(t, 2, "one pass for the burst") + p.stillAt(t, 2, "eight triggers inside one debounce window are one pass") +} + +// A trigger names no key, so a per-key loop has to be served by the whole-set +// pass. Anything else would leave the caller's signal describing nothing. +func TestRun_ATriggerRunsTheWholeSetOnAPerKeyLoop(t *testing.T) { + store, _ := newBucket(t, "trigger-whole-set") + whole := newPasses() + keyed := newPasses() + trigger := make(chan struct{}, 1) + + run(t, reconciler.Config{ + Name: "whole-set", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) (time.Duration, error) { whole.record(); return 0, nil }, + ReconcileKey: func(context.Context, string) (time.Duration, error) { keyed.record(); return 0, nil }, + Trigger: trigger, + Resync: time.Hour, + }) + + whole.waitFor(t, 1, "the startup pass") + trigger <- struct{}{} + + whole.waitFor(t, 2, "the trigger served by a whole-set pass") + assert.Zero(t, keyed.now(), "a trigger carries no key, so no per-key pass may run") +} + +// The deadline a triggered pass returns has to be armed like any other, or a +// loop woken by a trigger would silently lose the schedule it just asked for. +func TestRun_ATriggeredPassStillArmsItsDeadline(t *testing.T) { + store, _ := newBucket(t, "trigger-deadline") + p := newPasses() + trigger := make(chan struct{}, 1) + + run(t, reconciler.Config{ + Name: "trigger-deadline", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 80 * time.Millisecond, nil }, + Resync: time.Hour, + Trigger: trigger, + }) + + p.waitFor(t, 1, "the startup pass") + trigger <- struct{}{} + + p.waitFor(t, 5, "deadline passes continuing after the trigger woke the loop") +} + +// Closing the trigger is how a caller says its source is finished. The loop must +// survive it rather than spin on a closed channel. +func TestRun_AClosedTriggerDoesNotSpin(t *testing.T) { + store, _ := newBucket(t, "trigger-closed") + p := newPasses() + trigger := make(chan struct{}) + + run(t, reconciler.Config{ + Name: "closed", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) (time.Duration, error) { p.record(); return 0, nil }, + Trigger: trigger, + Resync: time.Hour, + }) + + p.waitFor(t, 1, "the startup pass") + close(trigger) + + // One pass is allowed for the close itself; what must not happen is the + // forwarder reading a closed channel forever and passing on every read. + time.Sleep(300 * time.Millisecond) + require.LessOrEqual(t, p.now(), 2, "a closed trigger must not drive a pass loop") + + // Still alive: a write after the close is still reconciled. + before := p.now() + require.NoError(t, store.Set(t.Context(), "after-close", &record{Name: "after-close"})) + p.waitFor(t, before+1, "a watch update after the trigger closed") +}