| index d54d4213c..27370f67b 100644 |
| --- a/internal/console/metricsview_test.go |
| +++ b/internal/console/metricsview_test.go |
| @@ -3,10 +3,12 @@ package console |
| import ( |
| "context" |
| "encoding/json" |
| + "errors" |
| "path/filepath" |
| "strconv" |
| "strings" |
| "sync" |
| + "sync/atomic" |
| "testing" |
| "time" |
| |
| @@ -302,6 +304,694 @@ func TestLoadMetricsConcurrentColdCallers(t *testing.T) { |
| } |
| } |
| |
| +// writeBoxSample (re)publishes box's current-hour raw chunk with a single |
| +// host CPU reading — metrics.WriteRaw's own "replace whatever is there" |
| +// semantics (see chunk.go) let a later call simulate the agent's next flush |
| +// landing with a different reading, at the same key. |
| +func writeBoxSample(t *testing.T, rel release.Store, box string, cpu float64) { |
| + t.Helper() |
| + at := time.Now().UTC().Add(-time.Minute) |
| + if err := metrics.WriteRaw(context.Background(), rel, metrics.RawChunk{ |
| + V: 1, Box: box, Hour: metrics.HourOf(at), |
| + Samples: []metrics.Sample{{At: at, Host: metrics.HostSample{CPUFrac: p(cpu)}}}, |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// slowCountingStore wraps a release.Store, counting List/Get calls and |
| +// optionally sleeping getDelay before each Get. It is the fixture the |
| +// stale-while-revalidate tests use to (a) prove a background sweep never |
| +// makes its trigger's caller wait for it and (b) count how many sweeps |
| +// actually ran. |
| +type slowCountingStore struct { |
| + release.Store |
| + getDelay time.Duration |
| + lists atomic.Int64 |
| + gets atomic.Int64 |
| + // listGate, when non-nil, is received from (blocking) before every List |
| + // call proceeds to the underlying store. discoverBoxes's List is the |
| + // very FIRST store operation any sweep makes (before any Get), so |
| + // gating it deterministically freezes a background sweep before it |
| + // touches the store at all — letting a test assert "the synchronous |
| + // caller performed zero store reads" without racing against how fast |
| + // the background goroutine happens to get scheduled. Set only from the |
| + // test goroutine strictly before the call that triggers the sweep, so |
| + // there is a happens-before edge (via the goroutine-creation rule) to |
| + // whichever goroutine reads it — never written concurrently with a read. |
| + // A CLOSED gate never blocks again (a receive on a closed channel |
| + // returns immediately), so once released it stays released for any |
| + // later List call in the same test, gated or not. |
| + listGate chan struct{} |
| + // listErr, when set, is what every List call returns (after clearing |
| + // listGate, if any) instead of touching the underlying store — the |
| + // fixture for simulating an outage that starts exactly when a gated |
| + // sweep is released, so both that sweep AND whatever sweep a coalesced |
| + // waiter runs on its own afterward see the same failure. |
| + listErr error |
| +} |
| + |
| +func (s *slowCountingStore) List(ctx context.Context, prefix string) ([]string, error) { |
| + s.lists.Add(1) |
| + if s.listGate != nil { |
| + <-s.listGate |
| + } |
| + if s.listErr != nil { |
| + return nil, s.listErr |
| + } |
| + return s.Store.List(ctx, prefix) |
| +} |
| + |
| +func (s *slowCountingStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + s.gets.Add(1) |
| + if s.getDelay > 0 { |
| + time.Sleep(s.getDelay) |
| + } |
| + return s.Store.Get(ctx, key) |
| +} |
| + |
| +// newSlowMetricsFixture is newMetricsFixture with the server reading through |
| +// a slowCountingStore, for tests that need to observe how many sweeps ran or |
| +// need a sweep slow enough to prove a caller didn't wait on it. |
| +func newSlowMetricsFixture(t *testing.T, getDelay time.Duration) (*Server, *slowCountingStore) { |
| + t.Helper() |
| + st, err := Open(filepath.Join(t.TempDir(), "console.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { st.Close() }) |
| + dir := release.NewDirStore(filepath.Join(t.TempDir(), "bucket")) |
| + slow := &slowCountingStore{Store: dir, getDelay: getDelay} |
| + srv, err := NewServer(Config{Releases: slow}, st) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return srv, slow |
| +} |
| + |
| +// waitUntilNotSweeping polls the cache's sweeping flag directly (same |
| +// package, no exported seam needed) until a background sweep finishes or |
| +// the deadline passes. |
| +func waitUntilNotSweeping(t *testing.T, srv *Server, timeout time.Duration) { |
| + t.Helper() |
| + deadline := time.Now().Add(timeout) |
| + for { |
| + srv.metricsCache.mu.Lock() |
| + sweeping := srv.metricsCache.sweeping |
| + srv.metricsCache.mu.Unlock() |
| + if !sweeping { |
| + return |
| + } |
| + if time.Now().After(deadline) { |
| + t.Fatalf("background sweep did not finish within %v", timeout) |
| + } |
| + time.Sleep(10 * time.Millisecond) |
| + } |
| +} |
| + |
| +// metricsStaleTestBackdate is how far the tests below wind c.at back to |
| +// simulate an expired-but-servable cache: comfortably past metricsCacheTTL |
| +// (60s) so the cache is definitely STALE, and comfortably short of |
| +// metricsStaleServeCeiling (2min) so it's just as definitely still within |
| +// serving range — 30s of margin on each side, generous enough that neither |
| +// boundary is at risk from scheduling jitter. |
| +const metricsStaleTestBackdate = metricsCacheTTL + 30*time.Second |
| + |
| +// TestLoadMetricsStaleWhileRevalidate is fix B's core behaviour change: an |
| +// expired-but-still-within-ceiling cache is no longer a reason to make the |
| +// caller wait. The first call after expiry gets the OLD data back |
| +// immediately — proven by a zero Get-count delta on the store, not a wall- |
| +// clock guess — while a sweep runs in the background; once that sweep |
| +// lands, a LATER call sees the NEW data, still without ever blocking. |
| +func TestLoadMetricsStaleWhileRevalidate(t *testing.T) { |
| + const getDelay = 80 * time.Millisecond |
| + srv, store := newSlowMetricsFixture(t, getDelay) |
| + ctx := context.Background() |
| + |
| + writeBoxSample(t, store, "box-a", 0.5) |
| + boxes, _, err := srv.LoadMetrics(ctx) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 { |
| + t.Fatalf("initial cold sweep = %+v, want box-a at 0.5", boxes) |
| + } |
| + |
| + // Expire the cache (within the ceiling) and change what the store |
| + // holds. From here on a FRESH sweep would see 0.9 — the immediate call |
| + // below must still answer with the cache's last-known 0.5, proving it |
| + // didn't wait for a fresh read. |
| + srv.metricsCache.mu.Lock() |
| + srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate) |
| + srv.metricsCache.mu.Unlock() |
| + writeBoxSample(t, store, "box-a", 0.9) |
| + |
| + // Freeze the background sweep this call is about to launch at its very |
| + // first store operation (discoverBoxes's List, before any Get) — this |
| + // makes the zero-Gets assertion below deterministic rather than a race |
| + // against the background goroutine's scheduling. |
| + store.listGate = make(chan struct{}) |
| + getsBefore := store.gets.Load() |
| + boxes, _, err = srv.LoadMetrics(ctx) |
| + getsAfter := store.gets.Load() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 { |
| + t.Fatalf("stale call = %+v, want the OLD 0.5 (a background refresh must not be awaited)", boxes) |
| + } |
| + if getsAfter != getsBefore { |
| + t.Fatalf("stale call performed %d store Gets, want 0 — it must return the cache without touching the store at all", getsAfter-getsBefore) |
| + } |
| + close(store.listGate) // release the frozen background sweep to finish |
| + |
| + waitUntilNotSweeping(t, srv, 5*time.Second) |
| + |
| + boxes, _, err = srv.LoadMetrics(ctx) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.9 { |
| + t.Fatalf("after the background sweep landed = %+v, want the NEW 0.9", boxes) |
| + } |
| +} |
| + |
| +// TestLoadMetricsStaleCacheSingleFlight: several callers arriving on an |
| +// EXPIRED-BUT-WITHIN-CEILING (not cold) cache at once must trigger exactly |
| +// one background sweep between them, never one each and never zero. The |
| +// baseline — however many List calls one full sweepAll makes (discoverBoxes |
| +// plus the alerts pass) — is measured from the initial cold sweep rather |
| +// than hard-coded, so this test doesn't have to know sweepAll's internals to |
| +// prove "exactly one more sweep happened." |
| +func TestLoadMetricsStaleCacheSingleFlight(t *testing.T) { |
| + srv, store := newSlowMetricsFixture(t, 30*time.Millisecond) |
| + ctx := context.Background() |
| + writeBoxSample(t, store, "box-a", 0.5) |
| + |
| + // Measured as a delta across just this call (not the cumulative total), |
| + // so any List traffic from fixture/server construction before this |
| + // point doesn't get folded into the "one sweep" baseline below. |
| + beforeCold := store.lists.Load() |
| + if _, _, err := srv.LoadMetrics(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + oneSweepLists := store.lists.Load() - beforeCold // however many List calls one sweepAll makes |
| + |
| + srv.metricsCache.mu.Lock() |
| + srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate) |
| + srv.metricsCache.mu.Unlock() |
| + before := store.lists.Load() |
| + |
| + const callers = 8 |
| + var wg sync.WaitGroup |
| + for i := 0; i < callers; i++ { |
| + wg.Add(1) |
| + go func() { |
| + defer wg.Done() |
| + if _, _, err := srv.LoadMetrics(ctx); err != nil { |
| + t.Error(err) |
| + } |
| + }() |
| + } |
| + wg.Wait() |
| + |
| + waitUntilNotSweeping(t, srv, 5*time.Second) |
| + |
| + if got := store.lists.Load() - before; got != oneSweepLists { |
| + t.Fatalf("List calls triggered by %d concurrent callers on a stale cache = %d, want exactly %d (one sweep, shared)", callers, got, oneSweepLists) |
| + } |
| +} |
| + |
| +// panicListStore panics on every List call — the fixture for proving a |
| +// panicking BACKGROUND sweep is recovered (startBackgroundSweep's explicit |
| +// recover — a detached goroutine has no net/http per-connection recover |
| +// sitting above it the way a request's own goroutine does) rather than |
| +// taking the whole process down, and that the cache is left usable |
| +// afterward rather than wedged. |
| +type panicListStore struct { |
| + release.Store |
| +} |
| + |
| +func (panicListStore) List(context.Context, string) ([]string, error) { |
| + panic("simulated bucket panic") |
| +} |
| + |
| +// TestLoadMetricsBackgroundSweepPanicRecovers: a background sweep |
| +// (triggered by a stale-but-within-ceiling cache) that panics must not crash |
| +// the test process, must leave the stale cache intact for its trigger's |
| +// caller (no error), must clear sweeping so nothing is wedged, and must not |
| +// prevent a later, healthy sweep from succeeding. |
| +func TestLoadMetricsBackgroundSweepPanicRecovers(t *testing.T) { |
| + srv, rel := newMetricsFixture(t) |
| + ctx := context.Background() |
| + writeBoxSample(t, rel, "box-a", 0.5) |
| + |
| + if _, _, err := srv.LoadMetrics(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + srv.metricsCache.mu.Lock() |
| + srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate) |
| + srv.metricsCache.mu.Unlock() |
| + srv.cfg.Releases = panicListStore{Store: rel} |
| + |
| + boxes, _, err := srv.LoadMetrics(ctx) |
| + if err != nil { |
| + t.Fatalf("stale call with a panicking background sweep returned an error: %v", err) |
| + } |
| + if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 { |
| + t.Fatalf("stale call = %+v, want the OLD cached data despite the panic", boxes) |
| + } |
| + |
| + waitUntilNotSweeping(t, srv, 5*time.Second) |
| + |
| + srv.metricsCache.mu.Lock() |
| + sweeping := srv.metricsCache.sweeping |
| + srv.metricsCache.mu.Unlock() |
| + if sweeping { |
| + t.Fatal("cache left wedged (sweeping still true) after a panicking background sweep") |
| + } |
| + |
| + // Nothing is permanently wedged: restore a healthy store and force the |
| + // COLD path (c.at zeroed) for a clean, unambiguous assertion that a |
| + // later sweep still works. |
| + srv.cfg.Releases = rel |
| + srv.metricsCache.mu.Lock() |
| + srv.metricsCache.at = time.Time{} |
| + srv.metricsCache.mu.Unlock() |
| + |
| + boxes, _, err = srv.LoadMetrics(ctx) |
| + if err != nil { |
| + t.Fatalf("sweep after a prior background panic failed: %v", err) |
| + } |
| + if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 { |
| + t.Fatalf("recovered sweep = %+v, want box-a at 0.5", boxes) |
| + } |
| +} |
| + |
| +// errListStore fails every List call with a fixed error — the fixture for |
| +// proving a background sweep's failure is only logged while its trigger's |
| +// caller sees no error and the stale cache survives (within the ceiling), |
| +// and that the SAME failure surfaces to a blocking caller once the cache |
| +// ages past metricsStaleServeCeiling — restoring outage visibility rather |
| +// than serving indefinitely-stale data with no error forever. |
| +type errListStore struct { |
| + release.Store |
| + err error |
| +} |
| + |
| +func (e errListStore) List(context.Context, string) ([]string, error) { |
| + return nil, e.err |
| +} |
| + |
| +// TestLoadMetricsBackgroundSweepFailureThenCeilingSurfacesError covers both |
| +// halves of that contract against the one failing store. |
| +func TestLoadMetricsBackgroundSweepFailureThenCeilingSurfacesError(t *testing.T) { |
| + srv, rel := newMetricsFixture(t) |
| + ctx := context.Background() |
| + writeBoxSample(t, rel, "box-a", 0.5) |
| + |
| + if _, _, err := srv.LoadMetrics(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + wantErr := errors.New("bucket had a bad day") |
| + srv.cfg.Releases = errListStore{Store: rel, err: wantErr} |
| + |
| + // Within the ceiling: the stale cache survives, the failing background |
| + // sweep is only logged, and this caller sees no error at all. |
| + srv.metricsCache.mu.Lock() |
| + srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate) |
| + srv.metricsCache.mu.Unlock() |
| + |
| + boxes, _, err := srv.LoadMetrics(ctx) |
| + if err != nil { |
| + t.Fatalf("stale call with a failing background sweep returned an error: %v", err) |
| + } |
| + if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 { |
| + t.Fatalf("stale call = %+v, want the OLD cached data", boxes) |
| + } |
| + |
| + waitUntilNotSweeping(t, srv, 5*time.Second) |
| + srv.metricsCache.mu.Lock() |
| + sweeping := srv.metricsCache.sweeping |
| + srv.metricsCache.mu.Unlock() |
| + if sweeping { |
| + t.Fatal("cache left wedged after a failing background sweep") |
| + } |
| + |
| + // Past the ceiling: the SAME failing store must now surface its error |
| + // to a blocking caller instead of handing back stale data forever. |
| + srv.metricsCache.mu.Lock() |
| + srv.metricsCache.at = time.Now().Add(-(metricsStaleServeCeiling + time.Second)) |
| + srv.metricsCache.mu.Unlock() |
| + |
| + if _, _, err := srv.LoadMetrics(ctx); err == nil { |
| + t.Fatal("call past the ceiling with a failing store returned no error") |
| + } else if !errors.Is(err, wantErr) { |
| + t.Fatalf("error = %v, want it to wrap %v", err, wantErr) |
| + } |
| +} |
| + |
| +// TestLoadMetricsPastCeilingBlocksForFreshData: a cache older than |
| +// metricsStaleServeCeiling is no longer eligible for stale-while-revalidate |
| +// even with a perfectly healthy store — the caller blocks on a synchronous |
| +// sweep and gets genuinely fresh data, proven by a non-zero Get-count delta |
| +// (the sweep ran INSIDE this call, not off to the side) rather than a |
| +// timing guess. |
| +func TestLoadMetricsPastCeilingBlocksForFreshData(t *testing.T) { |
| + srv, store := newSlowMetricsFixture(t, 0) |
| + ctx := context.Background() |
| + writeBoxSample(t, store, "box-a", 0.5) |
| + |
| + if _, _, err := srv.LoadMetrics(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + writeBoxSample(t, store, "box-a", 0.9) |
| + srv.metricsCache.mu.Lock() |
| + srv.metricsCache.at = time.Now().Add(-(metricsStaleServeCeiling + time.Second)) |
| + srv.metricsCache.mu.Unlock() |
| + |
| + getsBefore := store.gets.Load() |
| + boxes, _, err := srv.LoadMetrics(ctx) |
| + getsAfter := store.gets.Load() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if getsAfter == getsBefore { |
| + t.Fatal("call past the ceiling performed no store Gets — it must sweep synchronously, not serve stale data") |
| + } |
| + if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.9 { |
| + t.Fatalf("boxes = %+v, want the FRESH 0.9 — this call must have swept, not served the stale 0.5", boxes) |
| + } |
| +} |
| + |
| +// TestLoadMetricsPastCeilingWaiterCoalescesOntoBackgroundSweep pins the one |
| +// state-machine transition the other tests only exercise as a side effect: |
| +// a caller landing in the blocking COLD-or-STALE-PAST-CEILING branch while a |
| +// BACKGROUND sweep (started earlier by a within-ceiling stale caller) is |
| +// already in flight. LoadMetrics's blocking branch waits on c.sweeping/ |
| +// c.done regardless of which kind of sweep set them — this proves that |
| +// holds for a background sweep specifically, not just another blocking |
| +// caller's, by gating the background sweep at its very first store call so |
| +// the two callers' timing is deterministic rather than a race. |
| +func TestLoadMetricsPastCeilingWaiterCoalescesOntoBackgroundSweep(t *testing.T) { |
| + srv, store := newSlowMetricsFixture(t, 0) |
| + ctx := context.Background() |
| + writeBoxSample(t, store, "box-a", 0.5) |
| + |
| + // Baseline: however many List calls one full sweepAll makes, measured |
| + // the same way the single-flight test does. |
| + beforeCold := store.lists.Load() |
| + if _, _, err := srv.LoadMetrics(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + oneSweepLists := store.lists.Load() - beforeCold |
| + |
| + listsBeforeTrigger := store.lists.Load() |
| + |
| + // Gate the NEXT sweep at its first store call, then trigger it via a |
| + // within-ceiling stale call — this is startBackgroundSweep's path, not |
| + // the blocking one. |
| + store.listGate = make(chan struct{}) |
| + srv.metricsCache.mu.Lock() |
| + srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate) |
| + srv.metricsCache.mu.Unlock() |
| + writeBoxSample(t, store, "box-a", 0.9) // what the gated sweep will see once released |
| + |
| + if boxes, _, err := srv.LoadMetrics(ctx); err != nil { |
| + t.Fatal(err) |
| + } else if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 { |
| + t.Fatalf("triggering call = %+v, want the OLD 0.5 (background sweep just launched, now gated)", boxes) |
| + } |
| + |
| + // Age the SAME cache past the ceiling while the background sweep is |
| + // still stuck at the gate — exactly the state the waiter below lands in. |
| + srv.metricsCache.mu.Lock() |
| + srv.metricsCache.at = time.Now().Add(-(metricsStaleServeCeiling + time.Second)) |
| + srv.metricsCache.mu.Unlock() |
| + |
| + waiterDone := make(chan struct{}) |
| + var waiterBoxes []BoxMetricsView |
| + var waiterErr error |
| + go func() { |
| + defer close(waiterDone) |
| + waiterBoxes, _, waiterErr = srv.LoadMetrics(ctx) |
| + }() |
| + |
| + select { |
| + case <-waiterDone: |
| + t.Fatal("past-ceiling waiter returned before the gated background sweep was released") |
| + case <-time.After(200 * time.Millisecond): |
| + // still blocked, as it must be — coalesced onto the background |
| + // sweep's c.done rather than racing off to start its own. |
| + } |
| + |
| + close(store.listGate) |
| + |
| + select { |
| + case <-waiterDone: |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("past-ceiling waiter never returned after the gate was released") |
| + } |
| + if waiterErr != nil { |
| + t.Fatalf("waiter error = %v, want nil", waiterErr) |
| + } |
| + if len(waiterBoxes) != 1 || waiterBoxes[0].CPU == nil || *waiterBoxes[0].CPU != 0.9 { |
| + t.Fatalf("waiter boxes = %+v, want the FRESH 0.9 the background sweep it coalesced onto produced", waiterBoxes) |
| + } |
| + if got := store.lists.Load() - listsBeforeTrigger; got != oneSweepLists { |
| + t.Fatalf("List calls across the whole gated-sweep-plus-waiter span = %d, want exactly %d (one sweep, not duplicated by the waiter)", got, oneSweepLists) |
| + } |
| + srv.metricsCache.mu.Lock() |
| + sweeping := srv.metricsCache.sweeping |
| + srv.metricsCache.mu.Unlock() |
| + if sweeping { |
| + t.Fatal("cache left wedged (sweeping still true) after the waiter returned") |
| + } |
| +} |
| + |
| +// TestLoadMetricsPastCeilingWaiterTakesOverFailedBackgroundSweep is the |
| +// failure half of the same coalescing transition: the background sweep the |
| +// waiter coalesced onto fails (rather than lands), so the cache stays past |
| +// the ceiling — the waiter must then take the sweep over ITSELF (the |
| +// c.sweeping check it started blocked on now reads false) rather than |
| +// staying blocked or silently giving up, and surface that second attempt's |
| +// error like any ordinary blocking caller would. Nothing may wedge: the |
| +// flag clears either way, and a later call against a healthy store must |
| +// still succeed. |
| +func TestLoadMetricsPastCeilingWaiterTakesOverFailedBackgroundSweep(t *testing.T) { |
| + srv, store := newSlowMetricsFixture(t, 0) |
| + ctx := context.Background() |
| + writeBoxSample(t, store, "box-a", 0.5) |
| + |
| + if _, _, err := srv.LoadMetrics(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + store.listGate = make(chan struct{}) |
| + srv.metricsCache.mu.Lock() |
| + srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate) |
| + srv.metricsCache.mu.Unlock() |
| + |
| + if _, _, err := srv.LoadMetrics(ctx); err != nil { |
| + t.Fatal(err) |
| + } // stale 0.5 returned, background sweep launched and now gated |
| + |
| + srv.metricsCache.mu.Lock() |
| + srv.metricsCache.at = time.Now().Add(-(metricsStaleServeCeiling + time.Second)) |
| + srv.metricsCache.mu.Unlock() |
| + |
| + waiterDone := make(chan struct{}) |
| + var waiterErr error |
| + go func() { |
| + defer close(waiterDone) |
| + _, _, waiterErr = srv.LoadMetrics(ctx) |
| + }() |
| + |
| + select { |
| + case <-waiterDone: |
| + t.Fatal("past-ceiling waiter returned before the gated background sweep was released") |
| + case <-time.After(200 * time.Millisecond): |
| + } |
| + |
| + // The outage starts exactly when the gate opens: the background |
| + // sweep's blocked List call, and any later List call (including the |
| + // waiter's own follow-up sweep), all see the same error from here on. |
| + wantErr := errors.New("bucket had a bad day") |
| + store.listErr = wantErr |
| + close(store.listGate) |
| + |
| + select { |
| + case <-waiterDone: |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("past-ceiling waiter never returned after the gate was released") |
| + } |
| + if waiterErr == nil { |
| + t.Fatal("waiter that took over a failed background sweep returned no error") |
| + } |
| + if !errors.Is(waiterErr, wantErr) { |
| + t.Fatalf("waiter error = %v, want it to wrap %v", waiterErr, wantErr) |
| + } |
| + |
| + waitUntilNotSweeping(t, srv, 5*time.Second) |
| + |
| + // Nothing is permanently wedged: clear the outage and confirm a later |
| + // call against the same (now healthy) store still succeeds. |
| + store.listErr = nil |
| + boxes, _, err := srv.LoadMetrics(ctx) |
| + if err != nil { |
| + t.Fatalf("sweep after the prior failures still failed: %v", err) |
| + } |
| + if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 { |
| + t.Fatalf("recovered sweep = %+v, want box-a at 0.5", boxes) |
| + } |
| +} |
| + |
| +// concurrencyProbeStore counts how many Get calls are in flight at once — |
| +// the fixture proving sweepBox's parallel GETs are actually bounded at |
| +// metricsSweepBoxConcurrency: neither serial (the cost fix B removes) nor |
| +// unbounded (the thing errgroup.SetLimit exists to prevent). |
| +type concurrencyProbeStore struct { |
| + release.Store |
| + mu sync.Mutex |
| + cur, max int |
| +} |
| + |
| +func (c *concurrencyProbeStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + c.mu.Lock() |
| + c.cur++ |
| + if c.cur > c.max { |
| + c.max = c.cur |
| + } |
| + c.mu.Unlock() |
| + time.Sleep(5 * time.Millisecond) // hold the "in flight" window open long enough to overlap |
| + body, etag, err := c.Store.Get(ctx, key) |
| + c.mu.Lock() |
| + c.cur-- |
| + c.mu.Unlock() |
| + return body, etag, err |
| +} |
| + |
| +// TestSweepBoxBoundsConcurrency: a box with all 48 hours present sweeps with |
| +// more than one GET in flight at once (the point of fix B) but never more |
| +// than metricsSweepBoxConcurrency (the point of SetLimit). |
| +func TestSweepBoxBoundsConcurrency(t *testing.T) { |
| + srv, rel := newMetricsFixture(t) |
| + ctx := context.Background() |
| + now := time.Now().UTC() |
| + for i := 0; i < metricsSweepHours; i++ { |
| + at := now.Add(-time.Duration(i) * time.Hour) |
| + if err := metrics.WriteRaw(ctx, rel, metrics.RawChunk{ |
| + V: 1, Box: "box-conc", Hour: metrics.HourOf(at), |
| + Samples: []metrics.Sample{{At: at, Host: metrics.HostSample{CPUFrac: p(0.1)}}}, |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + |
| + probe := &concurrencyProbeStore{Store: rel} |
| + srv.cfg.Releases = probe |
| + |
| + if _, err := srv.sweepBox(ctx, "box-conc"); err != nil { |
| + t.Fatal(err) |
| + } |
| + probe.mu.Lock() |
| + max := probe.max |
| + probe.mu.Unlock() |
| + if max <= 1 { |
| + t.Fatalf("max concurrent Gets = %d, want > 1 (the parallelization fix B adds)", max) |
| + } |
| + if max > metricsSweepBoxConcurrency { |
| + t.Fatalf("max concurrent Gets = %d, want <= metricsSweepBoxConcurrency (%d)", max, metricsSweepBoxConcurrency) |
| + } |
| +} |
| + |
| +// TestSweepBoxParallelGetsPreserveSemantics: mixing present, missing |
| +// (os.ErrNotExist), and corrupt-JSON hours through the now-parallel GETs |
| +// must yield exactly the same sample set the old sequential version did — |
| +// fix B changes the sweep's speed, not its behaviour. |
| +func TestSweepBoxParallelGetsPreserveSemantics(t *testing.T) { |
| + srv, rel := newMetricsFixture(t) |
| + ctx := context.Background() |
| + now := time.Now().UTC() |
| + |
| + presentAt := now.Add(-2 * time.Hour) |
| + if err := metrics.WriteRaw(ctx, rel, metrics.RawChunk{ |
| + V: 1, Box: "box-mix", Hour: metrics.HourOf(presentAt), |
| + Samples: []metrics.Sample{{At: presentAt, Host: metrics.HostSample{CPUFrac: p(0.31)}}}, |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + // A corrupt hour: an object exists at the key, but it isn't a RawChunk. |
| + corruptAt := now.Add(-10 * time.Hour) |
| + if _, err := rel.Put(ctx, metrics.RawKey("box-mix", metrics.HourOf(corruptAt)), []byte("not json"), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + // Every other one of the 48 hours is left missing (os.ErrNotExist). |
| + |
| + view, err := srv.sweepBox(ctx, "box-mix") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if view.CPU == nil || *view.CPU != 0.31 { |
| + t.Fatalf("CPU = %v, want 0.31 (the only readable sample; missing/corrupt hours skipped)", view.CPU) |
| + } |
| + if !view.At.Equal(presentAt) { |
| + t.Fatalf("At = %v, want %v", view.At, presentAt) |
| + } |
| +} |
| + |
| +// errKeyStore fails Get for exactly one key with a non-NotExist error and |
| +// passes everything else through — proves sweepBox's parallel GETs still |
| +// fail the WHOLE sweep on a hard error, exactly like the sequential version |
| +// did, rather than silently skipping it like a missing or corrupt hour. |
| +type errKeyStore struct { |
| + release.Store |
| + failKey string |
| + err error |
| +} |
| + |
| +func (e errKeyStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if key == e.failKey { |
| + return nil, "", e.err |
| + } |
| + return e.Store.Get(ctx, key) |
| +} |
| + |
| +// TestSweepBoxParallelGetsFailOnHardError pins the "any other error fails |
| +// the whole sweep" contract under parallel GETs: one hour returning a real |
| +// error (not os.ErrNotExist) must fail sweepBox and surface that error, |
| +// even though 47 other GETs are in flight or already succeeded. |
| +func TestSweepBoxParallelGetsFailOnHardError(t *testing.T) { |
| + srv, rel := newMetricsFixture(t) |
| + ctx := context.Background() |
| + now := time.Now().UTC() |
| + |
| + goodAt := now.Add(-time.Hour) |
| + if err := metrics.WriteRaw(ctx, rel, metrics.RawChunk{ |
| + V: 1, Box: "box-fail", Hour: metrics.HourOf(goodAt), |
| + Samples: []metrics.Sample{{At: goodAt, Host: metrics.HostSample{CPUFrac: p(0.5)}}}, |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + failAt := now.Add(-20 * time.Hour) |
| + wantErr := errors.New("bucket had a bad day") |
| + srv.cfg.Releases = errKeyStore{ |
| + Store: rel, |
| + failKey: metrics.RawKey("box-fail", metrics.HourOf(failAt)), |
| + err: wantErr, |
| + } |
| + |
| + if _, err := srv.sweepBox(ctx, "box-fail"); err == nil { |
| + t.Fatal("sweepBox with a hard Get error on one hour returned no error") |
| + } else if !errors.Is(err, wantErr) { |
| + t.Fatalf("sweepBox error = %v, want it to wrap %v", err, wantErr) |
| + } |
| +} |
| + |
| // TestOwnerAppMetricsEmptyHosts: an app with no routes asks for nothing and |
| // gets nothing — empty, no error, and (proven by the poisoned-free fixture |
| // having no boxes at all) never an error path the app page has to handle. |
| @@ -394,6 +1084,63 @@ func TestOwnerAppMetricsFiltersToOwnUnits(t *testing.T) { |
| } |
| } |
| |
| +// TestOwnerAppMetricsSurvivesHealthyFlushCycle pins metricsFreshWindow's |
| +// derivation (flush cadence floor 5min + stale-serve ceiling 2min + 1min |
| +// slack = 8 minutes): a healthy agent's newest VISIBLE sample can be as old |
| +// as flush-cadence-plus-ceiling and still be current, but a sample old |
| +// enough that NOTHING short of a dead agent explains it must still flap to |
| +// no-recent-readings. Three ages, three different hosts/boxes (so the |
| +// "freshest box wins" rule in OwnerAppMetrics can't let one case's fresh |
| +// reading paper over another's staleness): |
| +// |
| +// - 5m30s: the original healthy-worst-case under the OLD (5min-flush-only) |
| +// math — comfortably inside both the old and new window. |
| +// - 7m0s: the NEW healthy worst case this fix widens the window to cover |
| +// — flush cadence's floor (5min) plus the full stale-serve ceiling |
| +// (2min) a caller's cache may be behind by. |
| +// - 8m30s: comfortably past the 8-minute window regardless of derivation |
| +// — must NOT read as current. |
| +func TestOwnerAppMetricsSurvivesHealthyFlushCycle(t *testing.T) { |
| + srv, rel := newMetricsFixture(t) |
| + ctx := context.Background() |
| + |
| + write := func(box, host string, age time.Duration) { |
| + at := time.Now().UTC().Add(-age) |
| + if err := metrics.WriteRaw(ctx, rel, metrics.RawChunk{ |
| + V: 1, Box: box, Hour: metrics.HourOf(at), |
| + Samples: []metrics.Sample{{ |
| + At: at, |
| + Units: map[string]metrics.UnitSample{ |
| + "carlos-app@" + host + ".service": {CPUFrac: p(0.42), MemBytes: pi64(2048)}, |
| + }, |
| + }}, |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + write("box-healthy-1", "flush1.example.com", 5*time.Minute+30*time.Second) |
| + write("box-healthy-2", "flush2.example.com", 7*time.Minute) |
| + write("box-dead", "dead.example.com", 8*time.Minute+30*time.Second) |
| + |
| + rows, err := srv.OwnerAppMetrics(ctx, []string{"flush1.example.com", "flush2.example.com", "dead.example.com"}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + byHost := map[string]InstanceMetrics{} |
| + for _, r := range rows { |
| + byHost[r.Host] = r |
| + } |
| + if r := byHost["flush1.example.com"]; !r.Reporting() { |
| + t.Fatalf("a 5m30s-old sample flapped to no-recent-readings: %+v", r) |
| + } |
| + if r := byHost["flush2.example.com"]; !r.Reporting() { |
| + t.Fatalf("a 7m0s-old sample (flush floor + stale-serve ceiling) flapped to no-recent-readings: %+v", r) |
| + } |
| + if r := byHost["dead.example.com"]; r.Reporting() { |
| + t.Fatalf("an 8m30s-old sample — past the fresh window — still read as current: %+v", r) |
| + } |
| +} |
| + |
| // TestSparkPointsInvertsYAndSkipsNil pins the coordinate math the brief |
| // calls out explicitly: a nil sample contributes no pair at all (not a |
| // zero), and a higher fraction plots at a SMALLER y (SVG's y axis points |
amadan