| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "path/filepath" |
| 8 | "strconv" |
| 9 | "strings" |
| 10 | "sync" |
| 11 | "sync/atomic" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/metrics" |
| 16 | "github.com/carlosframework/platform/internal/release" |
| 17 | ) |
| 18 | |
| 19 | // newMetricsFixture builds the minimal Server LoadMetrics's tests need: a |
| 20 | // console Store (NewServer's SigningKey dance requires one) and a bare |
| 21 | // release.DirStore, with none of consoleFixture's registry/keymail/HTTP |
| 22 | // scaffolding — mirrors server_test.go's minimal Config shape, trimmed to |
| 23 | // what a bucket-only read path touches. |
| 24 | func newMetricsFixture(t *testing.T) (*Server, release.Store) { |
| 25 | t.Helper() |
| 26 | st, err := Open(filepath.Join(t.TempDir(), "console.db")) |
| 27 | if err != nil { |
| 28 | t.Fatal(err) |
| 29 | } |
| 30 | t.Cleanup(func() { st.Close() }) |
| 31 | rel := release.NewDirStore(filepath.Join(t.TempDir(), "bucket")) |
| 32 | srv, err := NewServer(Config{Releases: rel}, st) |
| 33 | if err != nil { |
| 34 | t.Fatal(err) |
| 35 | } |
| 36 | return srv, rel |
| 37 | } |
| 38 | |
| 39 | func p(f float64) *float64 { return &f } |
| 40 | func pi64(i int64) *int64 { return &i } |
| 41 | |
| 42 | // writeAlertState publishes an AlertState the way the box agent would, |
| 43 | // create-only since the fixture never re-publishes the same box twice. |
| 44 | func writeAlertState(t *testing.T, rel release.Store, st metrics.AlertState) { |
| 45 | t.Helper() |
| 46 | b, err := json.Marshal(st) |
| 47 | if err != nil { |
| 48 | t.Fatal(err) |
| 49 | } |
| 50 | if _, err := rel.Put(context.Background(), metrics.AlertKey(st.Box), b, ""); err != nil { |
| 51 | t.Fatal(err) |
| 52 | } |
| 53 | } |
| 54 | |
| 55 | // TestLoadMetricsSweepsAndCaches pins the whole read path in one pass: |
| 56 | // discovery, the 48h raw sweep landing on the newest sample's values, |
| 57 | // non-empty sparklines, CPU-desc unit ordering with nil-CPU last, firing-only |
| 58 | // alerts — and then that a second call inside the 60s TTL returns the SAME |
| 59 | // result without touching the store again (proven by deleting every object |
| 60 | // LoadMetrics would need and asserting the second call doesn't notice). |
| 61 | func TestLoadMetricsSweepsAndCaches(t *testing.T) { |
| 62 | srv, rel := newMetricsFixture(t) |
| 63 | ctx := context.Background() |
| 64 | |
| 65 | olderAt := time.Now().UTC().Add(-90 * time.Minute) |
| 66 | latestAt := time.Now().UTC().Add(-5 * time.Minute) |
| 67 | |
| 68 | older := metrics.RawChunk{ |
| 69 | V: 1, |
| 70 | Box: "box-a", |
| 71 | Hour: metrics.HourOf(olderAt), |
| 72 | Samples: []metrics.Sample{ |
| 73 | { |
| 74 | At: olderAt, |
| 75 | Host: metrics.HostSample{CPUFrac: p(0.2), MemFrac: p(0.3), DiskFrac: p(0.4)}, |
| 76 | Units: map[string]metrics.UnitSample{ |
| 77 | "carlos-app@a.service": {CPUFrac: p(0.05), MemBytes: pi64(1000)}, |
| 78 | }, |
| 79 | }, |
| 80 | }, |
| 81 | } |
| 82 | latest := metrics.RawChunk{ |
| 83 | V: 1, |
| 84 | Box: "box-a", |
| 85 | Hour: metrics.HourOf(latestAt), |
| 86 | Samples: []metrics.Sample{ |
| 87 | { |
| 88 | At: latestAt, |
| 89 | Host: metrics.HostSample{CPUFrac: p(0.5), MemFrac: p(0.6), DiskFrac: p(0.7)}, |
| 90 | Units: map[string]metrics.UnitSample{ |
| 91 | "carlos-app@a.service": {CPUFrac: p(0.9), MemBytes: pi64(2000)}, |
| 92 | "carlos-app@b.service": {CPUFrac: p(0.1), MemBytes: pi64(500)}, |
| 93 | "carlos-app@nilcpu.service": {MemBytes: pi64(100)}, |
| 94 | }, |
| 95 | }, |
| 96 | }, |
| 97 | } |
| 98 | for _, c := range []metrics.RawChunk{older, latest} { |
| 99 | if err := metrics.WriteRaw(ctx, rel, c); err != nil { |
| 100 | t.Fatal(err) |
| 101 | } |
| 102 | } |
| 103 | |
| 104 | writeAlertState(t, rel, metrics.AlertState{ |
| 105 | V: 1, |
| 106 | Box: "box-a", |
| 107 | Rules: map[string]metrics.RuleState{ |
| 108 | "disk": {Firing: true, Value: 0.95, FiredAt: time.Date(2026, 8, 2, 10, 0, 0, 0, time.UTC)}, |
| 109 | "cpu": {Firing: false, Value: 0.5}, |
| 110 | }, |
| 111 | }) |
| 112 | |
| 113 | boxes, alerts, err := srv.LoadMetrics(ctx) |
| 114 | if err != nil { |
| 115 | t.Fatal(err) |
| 116 | } |
| 117 | if len(boxes) != 1 { |
| 118 | t.Fatalf("boxes = %d, want 1: %+v", len(boxes), boxes) |
| 119 | } |
| 120 | bv := boxes[0] |
| 121 | if bv.Box != "box-a" { |
| 122 | t.Fatalf("box = %q, want box-a", bv.Box) |
| 123 | } |
| 124 | if bv.CPU == nil || *bv.CPU != 0.5 { |
| 125 | t.Fatalf("CPU = %v, want 0.5 (the newest sample's value)", bv.CPU) |
| 126 | } |
| 127 | if !bv.At.Equal(latestAt) { |
| 128 | t.Fatalf("At = %v, want %v (the newest sample's time)", bv.At, latestAt) |
| 129 | } |
| 130 | if bv.SparkCPU == "" { |
| 131 | t.Fatal("SparkCPU is empty despite two samples") |
| 132 | } |
| 133 | if len(bv.Units) != 3 { |
| 134 | t.Fatalf("units = %d, want 3: %+v", len(bv.Units), bv.Units) |
| 135 | } |
| 136 | if bv.Units[0].Unit != "carlos-app@a.service" || bv.Units[0].CPUFrac == nil || *bv.Units[0].CPUFrac != 0.9 { |
| 137 | t.Fatalf("units[0] = %+v, want carlos-app@a.service at 0.9 (CPU desc)", bv.Units[0]) |
| 138 | } |
| 139 | if bv.Units[1].Unit != "carlos-app@b.service" { |
| 140 | t.Fatalf("units[1] = %+v, want carlos-app@b.service", bv.Units[1]) |
| 141 | } |
| 142 | if bv.Units[2].Unit != "carlos-app@nilcpu.service" || bv.Units[2].CPUFrac != nil { |
| 143 | t.Fatalf("units[2] = %+v, want the nil-CPU unit sorted last", bv.Units[2]) |
| 144 | } |
| 145 | if len(alerts) != 1 { |
| 146 | t.Fatalf("alerts = %d, want 1 (firing rules only, cpu excluded): %+v", len(alerts), alerts) |
| 147 | } |
| 148 | if alerts[0].Box != "box-a" || alerts[0].Rule != "disk" || alerts[0].Value != 0.95 { |
| 149 | t.Fatalf("alert = %+v", alerts[0]) |
| 150 | } |
| 151 | if !alerts[0].FiredAt.Equal(time.Date(2026, 8, 2, 10, 0, 0, 0, time.UTC)) { |
| 152 | t.Fatalf("alert.FiredAt = %v, want the state's FiredAt as-is", alerts[0].FiredAt) |
| 153 | } |
| 154 | |
| 155 | // Poison the store: delete everything a fresh sweep would need to read. |
| 156 | if err := rel.Delete(ctx, metrics.RawKey("box-a", metrics.HourOf(latestAt))); err != nil { |
| 157 | t.Fatal(err) |
| 158 | } |
| 159 | if err := rel.Delete(ctx, metrics.RawKey("box-a", metrics.HourOf(olderAt))); err != nil { |
| 160 | t.Fatal(err) |
| 161 | } |
| 162 | if err := rel.Delete(ctx, metrics.AlertKey("box-a")); err != nil { |
| 163 | t.Fatal(err) |
| 164 | } |
| 165 | |
| 166 | boxes2, alerts2, err := srv.LoadMetrics(ctx) |
| 167 | if err != nil { |
| 168 | t.Fatal(err) |
| 169 | } |
| 170 | if len(boxes2) != 1 || boxes2[0].Box != "box-a" || boxes2[0].CPU == nil || *boxes2[0].CPU != 0.5 { |
| 171 | t.Fatalf("second call within TTL re-read a poisoned store: %+v", boxes2) |
| 172 | } |
| 173 | if len(alerts2) != 1 { |
| 174 | t.Fatalf("second call within TTL re-read alerts from a poisoned store: %+v", alerts2) |
| 175 | } |
| 176 | } |
| 177 | |
| 178 | // TestLoadMetricsEmptyBucket: no box has ever published anything — the |
| 179 | // normal state of a freshly deployed console — is a clean empty result, not |
| 180 | // an error. |
| 181 | func TestLoadMetricsEmptyBucket(t *testing.T) { |
| 182 | srv, _ := newMetricsFixture(t) |
| 183 | boxes, alerts, err := srv.LoadMetrics(context.Background()) |
| 184 | if err != nil { |
| 185 | t.Fatal(err) |
| 186 | } |
| 187 | if len(boxes) != 0 || len(alerts) != 0 { |
| 188 | t.Fatalf("boxes=%+v alerts=%+v, want both empty", boxes, alerts) |
| 189 | } |
| 190 | } |
| 191 | |
| 192 | // ctxStore is a release.Store that honours its context: every read whose |
| 193 | // context is already done fails with that context's error. DirStore ignores |
| 194 | // its ctx entirely (see internal/release/store.go), so a cancelled-caller |
| 195 | // test built on a bare DirStore would pass vacuously no matter what |
| 196 | // LoadMetrics does with the context it is handed. |
| 197 | type ctxStore struct{ release.Store } |
| 198 | |
| 199 | func (c ctxStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 200 | if err := ctx.Err(); err != nil { |
| 201 | return nil, "", err |
| 202 | } |
| 203 | return c.Store.Get(ctx, key) |
| 204 | } |
| 205 | |
| 206 | func (c ctxStore) List(ctx context.Context, prefix string) ([]string, error) { |
| 207 | if err := ctx.Err(); err != nil { |
| 208 | return nil, err |
| 209 | } |
| 210 | return c.Store.List(ctx, prefix) |
| 211 | } |
| 212 | |
| 213 | // newCtxMetricsFixture is newMetricsFixture with the server reading through a |
| 214 | // context-honouring store. The returned release.Store is the raw one, for |
| 215 | // seeding. |
| 216 | func newCtxMetricsFixture(t *testing.T) (*Server, release.Store) { |
| 217 | t.Helper() |
| 218 | st, err := Open(filepath.Join(t.TempDir(), "console.db")) |
| 219 | if err != nil { |
| 220 | t.Fatal(err) |
| 221 | } |
| 222 | t.Cleanup(func() { st.Close() }) |
| 223 | rel := release.NewDirStore(filepath.Join(t.TempDir(), "bucket")) |
| 224 | srv, err := NewServer(Config{Releases: ctxStore{rel}}, st) |
| 225 | if err != nil { |
| 226 | t.Fatal(err) |
| 227 | } |
| 228 | return srv, rel |
| 229 | } |
| 230 | |
| 231 | // seedOneBox writes a single box's chunk with one sample, the minimum a sweep |
| 232 | // needs to produce a non-empty result. |
| 233 | func seedOneBox(t *testing.T, rel release.Store, box string) { |
| 234 | t.Helper() |
| 235 | at := time.Now().UTC().Add(-time.Minute) |
| 236 | if err := metrics.WriteRaw(context.Background(), rel, metrics.RawChunk{ |
| 237 | V: 1, Box: box, Hour: metrics.HourOf(at), |
| 238 | Samples: []metrics.Sample{{At: at, Host: metrics.HostSample{CPUFrac: p(0.5)}}}, |
| 239 | }); err != nil { |
| 240 | t.Fatal(err) |
| 241 | } |
| 242 | } |
| 243 | |
| 244 | // TestLoadMetricsSweepSurvivesCancelledCaller pins the review fix: the shared |
| 245 | // sweep runs on its own context (WithoutCancel + its own timeout), never the |
| 246 | // request's. Otherwise the first caller to navigate away mid-sweep aborts the |
| 247 | // sweep every other caller is waiting on, and — since a failed sweep is never |
| 248 | // cached — the next request starts over, so a page reloaded impatiently can |
| 249 | // starve the cache indefinitely. A caller whose context is ALREADY cancelled |
| 250 | // is the extreme of that case and must still get real data. |
| 251 | func TestLoadMetricsSweepSurvivesCancelledCaller(t *testing.T) { |
| 252 | srv, rel := newCtxMetricsFixture(t) |
| 253 | seedOneBox(t, rel, "box-a") |
| 254 | |
| 255 | ctx, cancel := context.WithCancel(context.Background()) |
| 256 | cancel() |
| 257 | |
| 258 | boxes, _, err := srv.LoadMetrics(ctx) |
| 259 | if err != nil { |
| 260 | t.Fatalf("LoadMetrics with a cancelled caller context: %v — the sweep must not inherit the request's cancellation", err) |
| 261 | } |
| 262 | if len(boxes) != 1 || boxes[0].Box != "box-a" || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 { |
| 263 | t.Fatalf("boxes = %+v, want box-a at 0.5", boxes) |
| 264 | } |
| 265 | } |
| 266 | |
| 267 | // TestLoadMetricsConcurrentColdCallers: several callers arriving on a cold |
| 268 | // cache at once — every tenant's app page plus the dashboard on a console |
| 269 | // that just restarted — all get the same correct data, whether they coalesce |
| 270 | // onto one sweep or duplicate it. Run under -race, this is also the guard on |
| 271 | // the cache fields being written while the lock is released across the IO. |
| 272 | func TestLoadMetricsConcurrentColdCallers(t *testing.T) { |
| 273 | srv, rel := newMetricsFixture(t) |
| 274 | seedOneBox(t, rel, "box-a") |
| 275 | |
| 276 | const callers = 4 |
| 277 | type result struct { |
| 278 | boxes []BoxMetricsView |
| 279 | err error |
| 280 | } |
| 281 | results := make([]result, callers) |
| 282 | var start sync.WaitGroup |
| 283 | var done sync.WaitGroup |
| 284 | start.Add(1) |
| 285 | for i := 0; i < callers; i++ { |
| 286 | done.Add(1) |
| 287 | go func(i int) { |
| 288 | defer done.Done() |
| 289 | start.Wait() |
| 290 | boxes, _, err := srv.LoadMetrics(context.Background()) |
| 291 | results[i] = result{boxes: boxes, err: err} |
| 292 | }(i) |
| 293 | } |
| 294 | start.Done() |
| 295 | done.Wait() |
| 296 | |
| 297 | for i, r := range results { |
| 298 | if r.err != nil { |
| 299 | t.Fatalf("caller %d: %v", i, r.err) |
| 300 | } |
| 301 | if len(r.boxes) != 1 || r.boxes[0].Box != "box-a" || r.boxes[0].CPU == nil || *r.boxes[0].CPU != 0.5 { |
| 302 | t.Fatalf("caller %d got %+v, want box-a at 0.5", i, r.boxes) |
| 303 | } |
| 304 | } |
| 305 | } |
| 306 | |
| 307 | // writeBoxSample (re)publishes box's current-hour raw chunk with a single |
| 308 | // host CPU reading — metrics.WriteRaw's own "replace whatever is there" |
| 309 | // semantics (see chunk.go) let a later call simulate the agent's next flush |
| 310 | // landing with a different reading, at the same key. |
| 311 | func writeBoxSample(t *testing.T, rel release.Store, box string, cpu float64) { |
| 312 | t.Helper() |
| 313 | at := time.Now().UTC().Add(-time.Minute) |
| 314 | if err := metrics.WriteRaw(context.Background(), rel, metrics.RawChunk{ |
| 315 | V: 1, Box: box, Hour: metrics.HourOf(at), |
| 316 | Samples: []metrics.Sample{{At: at, Host: metrics.HostSample{CPUFrac: p(cpu)}}}, |
| 317 | }); err != nil { |
| 318 | t.Fatal(err) |
| 319 | } |
| 320 | } |
| 321 | |
| 322 | // slowCountingStore wraps a release.Store, counting List/Get calls and |
| 323 | // optionally sleeping getDelay before each Get. It is the fixture the |
| 324 | // stale-while-revalidate tests use to (a) prove a background sweep never |
| 325 | // makes its trigger's caller wait for it and (b) count how many sweeps |
| 326 | // actually ran. |
| 327 | type slowCountingStore struct { |
| 328 | release.Store |
| 329 | getDelay time.Duration |
| 330 | lists atomic.Int64 |
| 331 | gets atomic.Int64 |
| 332 | // listGate, when non-nil, is received from (blocking) before every List |
| 333 | // call proceeds to the underlying store. discoverBoxes's List is the |
| 334 | // very FIRST store operation any sweep makes (before any Get), so |
| 335 | // gating it deterministically freezes a background sweep before it |
| 336 | // touches the store at all — letting a test assert "the synchronous |
| 337 | // caller performed zero store reads" without racing against how fast |
| 338 | // the background goroutine happens to get scheduled. Set only from the |
| 339 | // test goroutine strictly before the call that triggers the sweep, so |
| 340 | // there is a happens-before edge (via the goroutine-creation rule) to |
| 341 | // whichever goroutine reads it — never written concurrently with a read. |
| 342 | // A CLOSED gate never blocks again (a receive on a closed channel |
| 343 | // returns immediately), so once released it stays released for any |
| 344 | // later List call in the same test, gated or not. |
| 345 | listGate chan struct{} |
| 346 | // listErr, when set, is what every List call returns (after clearing |
| 347 | // listGate, if any) instead of touching the underlying store — the |
| 348 | // fixture for simulating an outage that starts exactly when a gated |
| 349 | // sweep is released, so both that sweep AND whatever sweep a coalesced |
| 350 | // waiter runs on its own afterward see the same failure. |
| 351 | listErr error |
| 352 | } |
| 353 | |
| 354 | func (s *slowCountingStore) List(ctx context.Context, prefix string) ([]string, error) { |
| 355 | s.lists.Add(1) |
| 356 | if s.listGate != nil { |
| 357 | <-s.listGate |
| 358 | } |
| 359 | if s.listErr != nil { |
| 360 | return nil, s.listErr |
| 361 | } |
| 362 | return s.Store.List(ctx, prefix) |
| 363 | } |
| 364 | |
| 365 | func (s *slowCountingStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 366 | s.gets.Add(1) |
| 367 | if s.getDelay > 0 { |
| 368 | time.Sleep(s.getDelay) |
| 369 | } |
| 370 | return s.Store.Get(ctx, key) |
| 371 | } |
| 372 | |
| 373 | // newSlowMetricsFixture is newMetricsFixture with the server reading through |
| 374 | // a slowCountingStore, for tests that need to observe how many sweeps ran or |
| 375 | // need a sweep slow enough to prove a caller didn't wait on it. |
| 376 | func newSlowMetricsFixture(t *testing.T, getDelay time.Duration) (*Server, *slowCountingStore) { |
| 377 | t.Helper() |
| 378 | st, err := Open(filepath.Join(t.TempDir(), "console.db")) |
| 379 | if err != nil { |
| 380 | t.Fatal(err) |
| 381 | } |
| 382 | t.Cleanup(func() { st.Close() }) |
| 383 | dir := release.NewDirStore(filepath.Join(t.TempDir(), "bucket")) |
| 384 | slow := &slowCountingStore{Store: dir, getDelay: getDelay} |
| 385 | srv, err := NewServer(Config{Releases: slow}, st) |
| 386 | if err != nil { |
| 387 | t.Fatal(err) |
| 388 | } |
| 389 | return srv, slow |
| 390 | } |
| 391 | |
| 392 | // waitUntilNotSweeping polls the cache's sweeping flag directly (same |
| 393 | // package, no exported seam needed) until a background sweep finishes or |
| 394 | // the deadline passes. |
| 395 | func waitUntilNotSweeping(t *testing.T, srv *Server, timeout time.Duration) { |
| 396 | t.Helper() |
| 397 | deadline := time.Now().Add(timeout) |
| 398 | for { |
| 399 | srv.metricsCache.mu.Lock() |
| 400 | sweeping := srv.metricsCache.sweeping |
| 401 | srv.metricsCache.mu.Unlock() |
| 402 | if !sweeping { |
| 403 | return |
| 404 | } |
| 405 | if time.Now().After(deadline) { |
| 406 | t.Fatalf("background sweep did not finish within %v", timeout) |
| 407 | } |
| 408 | time.Sleep(10 * time.Millisecond) |
| 409 | } |
| 410 | } |
| 411 | |
| 412 | // metricsStaleTestBackdate is how far the tests below wind c.at back to |
| 413 | // simulate an expired-but-servable cache: comfortably past metricsCacheTTL |
| 414 | // (60s) so the cache is definitely STALE, and comfortably short of |
| 415 | // metricsStaleServeCeiling (2min) so it's just as definitely still within |
| 416 | // serving range — 30s of margin on each side, generous enough that neither |
| 417 | // boundary is at risk from scheduling jitter. |
| 418 | const metricsStaleTestBackdate = metricsCacheTTL + 30*time.Second |
| 419 | |
| 420 | // TestLoadMetricsStaleWhileRevalidate is fix B's core behaviour change: an |
| 421 | // expired-but-still-within-ceiling cache is no longer a reason to make the |
| 422 | // caller wait. The first call after expiry gets the OLD data back |
| 423 | // immediately — proven by a zero Get-count delta on the store, not a wall- |
| 424 | // clock guess — while a sweep runs in the background; once that sweep |
| 425 | // lands, a LATER call sees the NEW data, still without ever blocking. |
| 426 | func TestLoadMetricsStaleWhileRevalidate(t *testing.T) { |
| 427 | const getDelay = 80 * time.Millisecond |
| 428 | srv, store := newSlowMetricsFixture(t, getDelay) |
| 429 | ctx := context.Background() |
| 430 | |
| 431 | writeBoxSample(t, store, "box-a", 0.5) |
| 432 | boxes, _, err := srv.LoadMetrics(ctx) |
| 433 | if err != nil { |
| 434 | t.Fatal(err) |
| 435 | } |
| 436 | if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 { |
| 437 | t.Fatalf("initial cold sweep = %+v, want box-a at 0.5", boxes) |
| 438 | } |
| 439 | |
| 440 | // Expire the cache (within the ceiling) and change what the store |
| 441 | // holds. From here on a FRESH sweep would see 0.9 — the immediate call |
| 442 | // below must still answer with the cache's last-known 0.5, proving it |
| 443 | // didn't wait for a fresh read. |
| 444 | srv.metricsCache.mu.Lock() |
| 445 | srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate) |
| 446 | srv.metricsCache.mu.Unlock() |
| 447 | writeBoxSample(t, store, "box-a", 0.9) |
| 448 | |
| 449 | // Freeze the background sweep this call is about to launch at its very |
| 450 | // first store operation (discoverBoxes's List, before any Get) — this |
| 451 | // makes the zero-Gets assertion below deterministic rather than a race |
| 452 | // against the background goroutine's scheduling. |
| 453 | store.listGate = make(chan struct{}) |
| 454 | getsBefore := store.gets.Load() |
| 455 | boxes, _, err = srv.LoadMetrics(ctx) |
| 456 | getsAfter := store.gets.Load() |
| 457 | if err != nil { |
| 458 | t.Fatal(err) |
| 459 | } |
| 460 | if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 { |
| 461 | t.Fatalf("stale call = %+v, want the OLD 0.5 (a background refresh must not be awaited)", boxes) |
| 462 | } |
| 463 | if getsAfter != getsBefore { |
| 464 | t.Fatalf("stale call performed %d store Gets, want 0 — it must return the cache without touching the store at all", getsAfter-getsBefore) |
| 465 | } |
| 466 | close(store.listGate) // release the frozen background sweep to finish |
| 467 | |
| 468 | waitUntilNotSweeping(t, srv, 5*time.Second) |
| 469 | |
| 470 | boxes, _, err = srv.LoadMetrics(ctx) |
| 471 | if err != nil { |
| 472 | t.Fatal(err) |
| 473 | } |
| 474 | if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.9 { |
| 475 | t.Fatalf("after the background sweep landed = %+v, want the NEW 0.9", boxes) |
| 476 | } |
| 477 | } |
| 478 | |
| 479 | // TestLoadMetricsStaleCacheSingleFlight: several callers arriving on an |
| 480 | // EXPIRED-BUT-WITHIN-CEILING (not cold) cache at once must trigger exactly |
| 481 | // one background sweep between them, never one each and never zero. The |
| 482 | // baseline — however many List calls one full sweepAll makes (discoverBoxes |
| 483 | // plus the alerts pass) — is measured from the initial cold sweep rather |
| 484 | // than hard-coded, so this test doesn't have to know sweepAll's internals to |
| 485 | // prove "exactly one more sweep happened." |
| 486 | func TestLoadMetricsStaleCacheSingleFlight(t *testing.T) { |
| 487 | srv, store := newSlowMetricsFixture(t, 30*time.Millisecond) |
| 488 | ctx := context.Background() |
| 489 | writeBoxSample(t, store, "box-a", 0.5) |
| 490 | |
| 491 | // Measured as a delta across just this call (not the cumulative total), |
| 492 | // so any List traffic from fixture/server construction before this |
| 493 | // point doesn't get folded into the "one sweep" baseline below. |
| 494 | beforeCold := store.lists.Load() |
| 495 | if _, _, err := srv.LoadMetrics(ctx); err != nil { |
| 496 | t.Fatal(err) |
| 497 | } |
| 498 | oneSweepLists := store.lists.Load() - beforeCold // however many List calls one sweepAll makes |
| 499 | |
| 500 | srv.metricsCache.mu.Lock() |
| 501 | srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate) |
| 502 | srv.metricsCache.mu.Unlock() |
| 503 | before := store.lists.Load() |
| 504 | |
| 505 | const callers = 8 |
| 506 | var wg sync.WaitGroup |
| 507 | for i := 0; i < callers; i++ { |
| 508 | wg.Add(1) |
| 509 | go func() { |
| 510 | defer wg.Done() |
| 511 | if _, _, err := srv.LoadMetrics(ctx); err != nil { |
| 512 | t.Error(err) |
| 513 | } |
| 514 | }() |
| 515 | } |
| 516 | wg.Wait() |
| 517 | |
| 518 | waitUntilNotSweeping(t, srv, 5*time.Second) |
| 519 | |
| 520 | if got := store.lists.Load() - before; got != oneSweepLists { |
| 521 | t.Fatalf("List calls triggered by %d concurrent callers on a stale cache = %d, want exactly %d (one sweep, shared)", callers, got, oneSweepLists) |
| 522 | } |
| 523 | } |
| 524 | |
| 525 | // panicListStore panics on every List call — the fixture for proving a |
| 526 | // panicking BACKGROUND sweep is recovered (startBackgroundSweep's explicit |
| 527 | // recover — a detached goroutine has no net/http per-connection recover |
| 528 | // sitting above it the way a request's own goroutine does) rather than |
| 529 | // taking the whole process down, and that the cache is left usable |
| 530 | // afterward rather than wedged. |
| 531 | type panicListStore struct { |
| 532 | release.Store |
| 533 | } |
| 534 | |
| 535 | func (panicListStore) List(context.Context, string) ([]string, error) { |
| 536 | panic("simulated bucket panic") |
| 537 | } |
| 538 | |
| 539 | // TestLoadMetricsBackgroundSweepPanicRecovers: a background sweep |
| 540 | // (triggered by a stale-but-within-ceiling cache) that panics must not crash |
| 541 | // the test process, must leave the stale cache intact for its trigger's |
| 542 | // caller (no error), must clear sweeping so nothing is wedged, and must not |
| 543 | // prevent a later, healthy sweep from succeeding. |
| 544 | func TestLoadMetricsBackgroundSweepPanicRecovers(t *testing.T) { |
| 545 | srv, rel := newMetricsFixture(t) |
| 546 | ctx := context.Background() |
| 547 | writeBoxSample(t, rel, "box-a", 0.5) |
| 548 | |
| 549 | if _, _, err := srv.LoadMetrics(ctx); err != nil { |
| 550 | t.Fatal(err) |
| 551 | } |
| 552 | |
| 553 | srv.metricsCache.mu.Lock() |
| 554 | srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate) |
| 555 | srv.metricsCache.mu.Unlock() |
| 556 | srv.cfg.Releases = panicListStore{Store: rel} |
| 557 | |
| 558 | boxes, _, err := srv.LoadMetrics(ctx) |
| 559 | if err != nil { |
| 560 | t.Fatalf("stale call with a panicking background sweep returned an error: %v", err) |
| 561 | } |
| 562 | if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 { |
| 563 | t.Fatalf("stale call = %+v, want the OLD cached data despite the panic", boxes) |
| 564 | } |
| 565 | |
| 566 | waitUntilNotSweeping(t, srv, 5*time.Second) |
| 567 | |
| 568 | srv.metricsCache.mu.Lock() |
| 569 | sweeping := srv.metricsCache.sweeping |
| 570 | srv.metricsCache.mu.Unlock() |
| 571 | if sweeping { |
| 572 | t.Fatal("cache left wedged (sweeping still true) after a panicking background sweep") |
| 573 | } |
| 574 | |
| 575 | // Nothing is permanently wedged: restore a healthy store and force the |
| 576 | // COLD path (c.at zeroed) for a clean, unambiguous assertion that a |
| 577 | // later sweep still works. |
| 578 | srv.cfg.Releases = rel |
| 579 | srv.metricsCache.mu.Lock() |
| 580 | srv.metricsCache.at = time.Time{} |
| 581 | srv.metricsCache.mu.Unlock() |
| 582 | |
| 583 | boxes, _, err = srv.LoadMetrics(ctx) |
| 584 | if err != nil { |
| 585 | t.Fatalf("sweep after a prior background panic failed: %v", err) |
| 586 | } |
| 587 | if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 { |
| 588 | t.Fatalf("recovered sweep = %+v, want box-a at 0.5", boxes) |
| 589 | } |
| 590 | } |
| 591 | |
| 592 | // errListStore fails every List call with a fixed error — the fixture for |
| 593 | // proving a background sweep's failure is only logged while its trigger's |
| 594 | // caller sees no error and the stale cache survives (within the ceiling), |
| 595 | // and that the SAME failure surfaces to a blocking caller once the cache |
| 596 | // ages past metricsStaleServeCeiling — restoring outage visibility rather |
| 597 | // than serving indefinitely-stale data with no error forever. |
| 598 | type errListStore struct { |
| 599 | release.Store |
| 600 | err error |
| 601 | } |
| 602 | |
| 603 | func (e errListStore) List(context.Context, string) ([]string, error) { |
| 604 | return nil, e.err |
| 605 | } |
| 606 | |
| 607 | // TestLoadMetricsBackgroundSweepFailureThenCeilingSurfacesError covers both |
| 608 | // halves of that contract against the one failing store. |
| 609 | func TestLoadMetricsBackgroundSweepFailureThenCeilingSurfacesError(t *testing.T) { |
| 610 | srv, rel := newMetricsFixture(t) |
| 611 | ctx := context.Background() |
| 612 | writeBoxSample(t, rel, "box-a", 0.5) |
| 613 | |
| 614 | if _, _, err := srv.LoadMetrics(ctx); err != nil { |
| 615 | t.Fatal(err) |
| 616 | } |
| 617 | |
| 618 | wantErr := errors.New("bucket had a bad day") |
| 619 | srv.cfg.Releases = errListStore{Store: rel, err: wantErr} |
| 620 | |
| 621 | // Within the ceiling: the stale cache survives, the failing background |
| 622 | // sweep is only logged, and this caller sees no error at all. |
| 623 | srv.metricsCache.mu.Lock() |
| 624 | srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate) |
| 625 | srv.metricsCache.mu.Unlock() |
| 626 | |
| 627 | boxes, _, err := srv.LoadMetrics(ctx) |
| 628 | if err != nil { |
| 629 | t.Fatalf("stale call with a failing background sweep returned an error: %v", err) |
| 630 | } |
| 631 | if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 { |
| 632 | t.Fatalf("stale call = %+v, want the OLD cached data", boxes) |
| 633 | } |
| 634 | |
| 635 | waitUntilNotSweeping(t, srv, 5*time.Second) |
| 636 | srv.metricsCache.mu.Lock() |
| 637 | sweeping := srv.metricsCache.sweeping |
| 638 | srv.metricsCache.mu.Unlock() |
| 639 | if sweeping { |
| 640 | t.Fatal("cache left wedged after a failing background sweep") |
| 641 | } |
| 642 | |
| 643 | // Past the ceiling: the SAME failing store must now surface its error |
| 644 | // to a blocking caller instead of handing back stale data forever. |
| 645 | srv.metricsCache.mu.Lock() |
| 646 | srv.metricsCache.at = time.Now().Add(-(metricsStaleServeCeiling + time.Second)) |
| 647 | srv.metricsCache.mu.Unlock() |
| 648 | |
| 649 | if _, _, err := srv.LoadMetrics(ctx); err == nil { |
| 650 | t.Fatal("call past the ceiling with a failing store returned no error") |
| 651 | } else if !errors.Is(err, wantErr) { |
| 652 | t.Fatalf("error = %v, want it to wrap %v", err, wantErr) |
| 653 | } |
| 654 | } |
| 655 | |
| 656 | // TestLoadMetricsPastCeilingBlocksForFreshData: a cache older than |
| 657 | // metricsStaleServeCeiling is no longer eligible for stale-while-revalidate |
| 658 | // even with a perfectly healthy store — the caller blocks on a synchronous |
| 659 | // sweep and gets genuinely fresh data, proven by a non-zero Get-count delta |
| 660 | // (the sweep ran INSIDE this call, not off to the side) rather than a |
| 661 | // timing guess. |
| 662 | func TestLoadMetricsPastCeilingBlocksForFreshData(t *testing.T) { |
| 663 | srv, store := newSlowMetricsFixture(t, 0) |
| 664 | ctx := context.Background() |
| 665 | writeBoxSample(t, store, "box-a", 0.5) |
| 666 | |
| 667 | if _, _, err := srv.LoadMetrics(ctx); err != nil { |
| 668 | t.Fatal(err) |
| 669 | } |
| 670 | |
| 671 | writeBoxSample(t, store, "box-a", 0.9) |
| 672 | srv.metricsCache.mu.Lock() |
| 673 | srv.metricsCache.at = time.Now().Add(-(metricsStaleServeCeiling + time.Second)) |
| 674 | srv.metricsCache.mu.Unlock() |
| 675 | |
| 676 | getsBefore := store.gets.Load() |
| 677 | boxes, _, err := srv.LoadMetrics(ctx) |
| 678 | getsAfter := store.gets.Load() |
| 679 | if err != nil { |
| 680 | t.Fatal(err) |
| 681 | } |
| 682 | if getsAfter == getsBefore { |
| 683 | t.Fatal("call past the ceiling performed no store Gets — it must sweep synchronously, not serve stale data") |
| 684 | } |
| 685 | if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.9 { |
| 686 | t.Fatalf("boxes = %+v, want the FRESH 0.9 — this call must have swept, not served the stale 0.5", boxes) |
| 687 | } |
| 688 | } |
| 689 | |
| 690 | // TestLoadMetricsPastCeilingWaiterCoalescesOntoBackgroundSweep pins the one |
| 691 | // state-machine transition the other tests only exercise as a side effect: |
| 692 | // a caller landing in the blocking COLD-or-STALE-PAST-CEILING branch while a |
| 693 | // BACKGROUND sweep (started earlier by a within-ceiling stale caller) is |
| 694 | // already in flight. LoadMetrics's blocking branch waits on c.sweeping/ |
| 695 | // c.done regardless of which kind of sweep set them — this proves that |
| 696 | // holds for a background sweep specifically, not just another blocking |
| 697 | // caller's, by gating the background sweep at its very first store call so |
| 698 | // the two callers' timing is deterministic rather than a race. |
| 699 | func TestLoadMetricsPastCeilingWaiterCoalescesOntoBackgroundSweep(t *testing.T) { |
| 700 | srv, store := newSlowMetricsFixture(t, 0) |
| 701 | ctx := context.Background() |
| 702 | writeBoxSample(t, store, "box-a", 0.5) |
| 703 | |
| 704 | // Baseline: however many List calls one full sweepAll makes, measured |
| 705 | // the same way the single-flight test does. |
| 706 | beforeCold := store.lists.Load() |
| 707 | if _, _, err := srv.LoadMetrics(ctx); err != nil { |
| 708 | t.Fatal(err) |
| 709 | } |
| 710 | oneSweepLists := store.lists.Load() - beforeCold |
| 711 | |
| 712 | listsBeforeTrigger := store.lists.Load() |
| 713 | |
| 714 | // Gate the NEXT sweep at its first store call, then trigger it via a |
| 715 | // within-ceiling stale call — this is startBackgroundSweep's path, not |
| 716 | // the blocking one. |
| 717 | store.listGate = make(chan struct{}) |
| 718 | srv.metricsCache.mu.Lock() |
| 719 | srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate) |
| 720 | srv.metricsCache.mu.Unlock() |
| 721 | writeBoxSample(t, store, "box-a", 0.9) // what the gated sweep will see once released |
| 722 | |
| 723 | if boxes, _, err := srv.LoadMetrics(ctx); err != nil { |
| 724 | t.Fatal(err) |
| 725 | } else if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 { |
| 726 | t.Fatalf("triggering call = %+v, want the OLD 0.5 (background sweep just launched, now gated)", boxes) |
| 727 | } |
| 728 | |
| 729 | // Age the SAME cache past the ceiling while the background sweep is |
| 730 | // still stuck at the gate — exactly the state the waiter below lands in. |
| 731 | srv.metricsCache.mu.Lock() |
| 732 | srv.metricsCache.at = time.Now().Add(-(metricsStaleServeCeiling + time.Second)) |
| 733 | srv.metricsCache.mu.Unlock() |
| 734 | |
| 735 | waiterDone := make(chan struct{}) |
| 736 | var waiterBoxes []BoxMetricsView |
| 737 | var waiterErr error |
| 738 | go func() { |
| 739 | defer close(waiterDone) |
| 740 | waiterBoxes, _, waiterErr = srv.LoadMetrics(ctx) |
| 741 | }() |
| 742 | |
| 743 | select { |
| 744 | case <-waiterDone: |
| 745 | t.Fatal("past-ceiling waiter returned before the gated background sweep was released") |
| 746 | case <-time.After(200 * time.Millisecond): |
| 747 | // still blocked, as it must be — coalesced onto the background |
| 748 | // sweep's c.done rather than racing off to start its own. |
| 749 | } |
| 750 | |
| 751 | close(store.listGate) |
| 752 | |
| 753 | select { |
| 754 | case <-waiterDone: |
| 755 | case <-time.After(5 * time.Second): |
| 756 | t.Fatal("past-ceiling waiter never returned after the gate was released") |
| 757 | } |
| 758 | if waiterErr != nil { |
| 759 | t.Fatalf("waiter error = %v, want nil", waiterErr) |
| 760 | } |
| 761 | if len(waiterBoxes) != 1 || waiterBoxes[0].CPU == nil || *waiterBoxes[0].CPU != 0.9 { |
| 762 | t.Fatalf("waiter boxes = %+v, want the FRESH 0.9 the background sweep it coalesced onto produced", waiterBoxes) |
| 763 | } |
| 764 | if got := store.lists.Load() - listsBeforeTrigger; got != oneSweepLists { |
| 765 | 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) |
| 766 | } |
| 767 | srv.metricsCache.mu.Lock() |
| 768 | sweeping := srv.metricsCache.sweeping |
| 769 | srv.metricsCache.mu.Unlock() |
| 770 | if sweeping { |
| 771 | t.Fatal("cache left wedged (sweeping still true) after the waiter returned") |
| 772 | } |
| 773 | } |
| 774 | |
| 775 | // TestLoadMetricsPastCeilingWaiterTakesOverFailedBackgroundSweep is the |
| 776 | // failure half of the same coalescing transition: the background sweep the |
| 777 | // waiter coalesced onto fails (rather than lands), so the cache stays past |
| 778 | // the ceiling — the waiter must then take the sweep over ITSELF (the |
| 779 | // c.sweeping check it started blocked on now reads false) rather than |
| 780 | // staying blocked or silently giving up, and surface that second attempt's |
| 781 | // error like any ordinary blocking caller would. Nothing may wedge: the |
| 782 | // flag clears either way, and a later call against a healthy store must |
| 783 | // still succeed. |
| 784 | func TestLoadMetricsPastCeilingWaiterTakesOverFailedBackgroundSweep(t *testing.T) { |
| 785 | srv, store := newSlowMetricsFixture(t, 0) |
| 786 | ctx := context.Background() |
| 787 | writeBoxSample(t, store, "box-a", 0.5) |
| 788 | |
| 789 | if _, _, err := srv.LoadMetrics(ctx); err != nil { |
| 790 | t.Fatal(err) |
| 791 | } |
| 792 | |
| 793 | store.listGate = make(chan struct{}) |
| 794 | srv.metricsCache.mu.Lock() |
| 795 | srv.metricsCache.at = time.Now().Add(-metricsStaleTestBackdate) |
| 796 | srv.metricsCache.mu.Unlock() |
| 797 | |
| 798 | if _, _, err := srv.LoadMetrics(ctx); err != nil { |
| 799 | t.Fatal(err) |
| 800 | } // stale 0.5 returned, background sweep launched and now gated |
| 801 | |
| 802 | srv.metricsCache.mu.Lock() |
| 803 | srv.metricsCache.at = time.Now().Add(-(metricsStaleServeCeiling + time.Second)) |
| 804 | srv.metricsCache.mu.Unlock() |
| 805 | |
| 806 | waiterDone := make(chan struct{}) |
| 807 | var waiterErr error |
| 808 | go func() { |
| 809 | defer close(waiterDone) |
| 810 | _, _, waiterErr = srv.LoadMetrics(ctx) |
| 811 | }() |
| 812 | |
| 813 | select { |
| 814 | case <-waiterDone: |
| 815 | t.Fatal("past-ceiling waiter returned before the gated background sweep was released") |
| 816 | case <-time.After(200 * time.Millisecond): |
| 817 | } |
| 818 | |
| 819 | // The outage starts exactly when the gate opens: the background |
| 820 | // sweep's blocked List call, and any later List call (including the |
| 821 | // waiter's own follow-up sweep), all see the same error from here on. |
| 822 | wantErr := errors.New("bucket had a bad day") |
| 823 | store.listErr = wantErr |
| 824 | close(store.listGate) |
| 825 | |
| 826 | select { |
| 827 | case <-waiterDone: |
| 828 | case <-time.After(5 * time.Second): |
| 829 | t.Fatal("past-ceiling waiter never returned after the gate was released") |
| 830 | } |
| 831 | if waiterErr == nil { |
| 832 | t.Fatal("waiter that took over a failed background sweep returned no error") |
| 833 | } |
| 834 | if !errors.Is(waiterErr, wantErr) { |
| 835 | t.Fatalf("waiter error = %v, want it to wrap %v", waiterErr, wantErr) |
| 836 | } |
| 837 | |
| 838 | waitUntilNotSweeping(t, srv, 5*time.Second) |
| 839 | |
| 840 | // Nothing is permanently wedged: clear the outage and confirm a later |
| 841 | // call against the same (now healthy) store still succeeds. |
| 842 | store.listErr = nil |
| 843 | boxes, _, err := srv.LoadMetrics(ctx) |
| 844 | if err != nil { |
| 845 | t.Fatalf("sweep after the prior failures still failed: %v", err) |
| 846 | } |
| 847 | if len(boxes) != 1 || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 { |
| 848 | t.Fatalf("recovered sweep = %+v, want box-a at 0.5", boxes) |
| 849 | } |
| 850 | } |
| 851 | |
| 852 | // errOnKeyStore fails one specific Get with a non-NotExist error and passes |
| 853 | // everything else through — the "bucket is answering, but this object will |
| 854 | // not read" case, which must fail the whole sweep rather than being skipped |
| 855 | // like a missing hour. |
| 856 | type errOnKeyStore struct { |
| 857 | release.Store |
| 858 | key string |
| 859 | err error |
| 860 | } |
| 861 | |
| 862 | func (e errOnKeyStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 863 | if key == e.key { |
| 864 | return nil, "", e.err |
| 865 | } |
| 866 | return e.Store.Get(ctx, key) |
| 867 | } |
| 868 | |
| 869 | // newMetricsFixtureStore is newMetricsFixture with the server reading through |
| 870 | // a wrapped store. The returned release.Store is the raw one, for seeding. |
| 871 | func newMetricsFixtureStore(t *testing.T, wrap func(release.Store) release.Store) (*Server, release.Store) { |
| 872 | t.Helper() |
| 873 | st, err := Open(filepath.Join(t.TempDir(), "console.db")) |
| 874 | if err != nil { |
| 875 | t.Fatal(err) |
| 876 | } |
| 877 | t.Cleanup(func() { st.Close() }) |
| 878 | rel := release.NewDirStore(filepath.Join(t.TempDir(), "bucket")) |
| 879 | srv, err := NewServer(Config{Releases: wrap(rel)}, st) |
| 880 | if err != nil { |
| 881 | t.Fatal(err) |
| 882 | } |
| 883 | return srv, rel |
| 884 | } |
| 885 | |
| 886 | // writeChunk publishes one hour's raw chunk for box, the way the box agent |
| 887 | // would. |
| 888 | func writeChunk(t *testing.T, rel release.Store, box, hour string, samples []metrics.Sample) { |
| 889 | t.Helper() |
| 890 | if err := metrics.WriteRaw(context.Background(), rel, metrics.RawChunk{ |
| 891 | V: 1, Box: box, Hour: hour, Samples: samples, |
| 892 | }); err != nil { |
| 893 | t.Fatal(err) |
| 894 | } |
| 895 | } |
| 896 | |
| 897 | // seedTiedHours writes three hours for box-a whose samples deliberately SHARE |
| 898 | // a timestamp across different hours, plus a second box. Ties are the whole |
| 899 | // point: sort.Slice is not stable, so which tied sample ends up last — and so |
| 900 | // which reading the box's row prints as current — is decided by the order the |
| 901 | // hours were concatenated in, not by the sort. Returns the shared timestamp. |
| 902 | func seedTiedHours(t *testing.T, rel release.Store) time.Time { |
| 903 | t.Helper() |
| 904 | now := time.Now().UTC() |
| 905 | tie := now.Truncate(time.Second) |
| 906 | unit := func(cpu float64, mem int64) map[string]metrics.UnitSample { |
| 907 | return map[string]metrics.UnitSample{"carlos-app@a.service": {CPUFrac: p(cpu), MemBytes: pi64(mem)}} |
| 908 | } |
| 909 | writeChunk(t, rel, "box-a", metrics.HourOf(now.Add(-2*time.Hour)), []metrics.Sample{ |
| 910 | {At: tie.Add(-2 * time.Hour), Host: metrics.HostSample{CPUFrac: p(0.10)}, Units: unit(0.11, 110)}, |
| 911 | {At: tie, Host: metrics.HostSample{CPUFrac: p(0.20)}, Units: unit(0.22, 220)}, |
| 912 | }) |
| 913 | writeChunk(t, rel, "box-a", metrics.HourOf(now.Add(-time.Hour)), []metrics.Sample{ |
| 914 | {At: tie.Add(-time.Hour), Host: metrics.HostSample{CPUFrac: p(0.30)}, Units: unit(0.33, 330)}, |
| 915 | {At: tie, Host: metrics.HostSample{CPUFrac: p(0.40)}, Units: unit(0.44, 440)}, |
| 916 | }) |
| 917 | writeChunk(t, rel, "box-a", metrics.HourOf(now), []metrics.Sample{ |
| 918 | {At: tie, Host: metrics.HostSample{CPUFrac: p(0.50)}, Units: unit(0.55, 550)}, |
| 919 | }) |
| 920 | writeChunk(t, rel, "box-b", metrics.HourOf(now), []metrics.Sample{ |
| 921 | {At: tie, Host: metrics.HostSample{CPUFrac: p(0.60)}}, |
| 922 | }) |
| 923 | return tie |
| 924 | } |
| 925 | |
| 926 | // TestSweepAllDeterministicOrder pins the sweep's output shape against the |
| 927 | // timing of whatever reads it in: boxes come back in sorted name order, and a |
| 928 | // box's samples in hour order (newest hour first, as the loop counts back), |
| 929 | // so tied timestamps resolve the same way on every render. The pinned values |
| 930 | // are the serial sweep's own answers — a sweep that assembled hours or boxes |
| 931 | // in completion order would land on a different reading and the row would |
| 932 | // flap between page loads. |
| 933 | func TestSweepAllDeterministicOrder(t *testing.T) { |
| 934 | srv, rel := newMetricsFixture(t) |
| 935 | tie := seedTiedHours(t, rel) |
| 936 | |
| 937 | views, _, err := srv.sweepAll(context.Background()) |
| 938 | if err != nil { |
| 939 | t.Fatal(err) |
| 940 | } |
| 941 | if len(views) != 2 || views[0].Box != "box-a" || views[1].Box != "box-b" { |
| 942 | t.Fatalf("boxes = %+v, want box-a then box-b (sorted)", views) |
| 943 | } |
| 944 | a := views[0] |
| 945 | if !a.At.Equal(tie) { |
| 946 | t.Fatalf("At = %v, want the tied timestamp %v", a.At, tie) |
| 947 | } |
| 948 | // Hour order decides the tie: the LAST tied sample appended is the oldest |
| 949 | // hour's (0.20), not the current hour's (0.50). |
| 950 | if a.CPU == nil || *a.CPU != 0.20 { |
| 951 | t.Fatalf("CPU = %v, want 0.20 (the tie broken in hour order)", a.CPU) |
| 952 | } |
| 953 | if len(a.Units) != 1 || a.Units[0].CPUFrac == nil || *a.Units[0].CPUFrac != 0.22 { |
| 954 | t.Fatalf("units = %+v, want the same sample's unit reading 0.22", a.Units) |
| 955 | } |
| 956 | // The sparkline coordinate string encodes the whole ordered series, so it |
| 957 | // is the sharpest oracle for "the samples arrived in the same order". |
| 958 | const wantSpark = "0.00,36.00 60.00,28.00 120.00,20.00 180.00,24.00 240.00,32.00" |
| 959 | if a.SparkCPU != wantSpark { |
| 960 | t.Fatalf("SparkCPU = %q, want %q (samples in hour order)", a.SparkCPU, wantSpark) |
| 961 | } |
| 962 | |
| 963 | // Same seed, repeated sweeps: identical every time, not merely plausible |
| 964 | // once. |
| 965 | for i := 0; i < 20; i++ { |
| 966 | got, _, err := srv.sweepAll(context.Background()) |
| 967 | if err != nil { |
| 968 | t.Fatal(err) |
| 969 | } |
| 970 | if len(got) != 2 || got[0].Box != "box-a" || got[1].Box != "box-b" { |
| 971 | t.Fatalf("sweep %d box order = %+v", i, got) |
| 972 | } |
| 973 | if got[0].CPU == nil || *got[0].CPU != *a.CPU || got[0].SparkCPU != a.SparkCPU { |
| 974 | t.Fatalf("sweep %d differs: CPU=%v spark=%q, want %v / %q", i, got[0].CPU, got[0].SparkCPU, *a.CPU, a.SparkCPU) |
| 975 | } |
| 976 | if got[0].Units[0].SparkCPU != a.Units[0].SparkCPU || got[0].Units[0].SparkMem != a.Units[0].SparkMem { |
| 977 | t.Fatalf("sweep %d unit sparklines differ: %+v", i, got[0].Units[0]) |
| 978 | } |
| 979 | } |
| 980 | } |
| 981 | |
| 982 | // TestSweepAllFailsOnUnreadableChunk: one hour of one box answering with |
| 983 | // something other than os.ErrNotExist fails the WHOLE sweep — no partial view |
| 984 | // is returned and nothing is cached, so the next caller retries rather than |
| 985 | // inheriting a box's truncated history as fact. |
| 986 | func TestSweepAllFailsOnUnreadableChunk(t *testing.T) { |
| 987 | boom := errors.New("bucket said no") |
| 988 | badKey := metrics.RawKey("box-a", metrics.HourOf(time.Now().UTC().Add(-time.Hour))) |
| 989 | srv, rel := newMetricsFixtureStore(t, func(s release.Store) release.Store { |
| 990 | return errOnKeyStore{Store: s, key: badKey, err: boom} |
| 991 | }) |
| 992 | seedTiedHours(t, rel) |
| 993 | |
| 994 | if _, _, err := srv.LoadMetrics(context.Background()); err == nil { |
| 995 | t.Fatal("LoadMetrics succeeded despite an unreadable chunk") |
| 996 | } else { |
| 997 | if !errors.Is(err, boom) { |
| 998 | t.Fatalf("error = %v, want it to wrap the store's error", err) |
| 999 | } |
| 1000 | if !strings.Contains(err.Error(), "read "+badKey) { |
| 1001 | t.Fatalf("error = %v, want it to name the key it failed to read", err) |
| 1002 | } |
| 1003 | } |
| 1004 | // A failed sweep is never cached. |
| 1005 | srv.metricsCache.mu.Lock() |
| 1006 | cachedAt := srv.metricsCache.at |
| 1007 | srv.metricsCache.mu.Unlock() |
| 1008 | if !cachedAt.IsZero() { |
| 1009 | t.Fatalf("a failed sweep was cached (at = %v)", cachedAt) |
| 1010 | } |
| 1011 | } |
| 1012 | |
| 1013 | // TestOwnerAppMetricsEmptyHosts: an app with no routes asks for nothing and |
| 1014 | // gets nothing — empty, no error, and (proven by the poisoned-free fixture |
| 1015 | // having no boxes at all) never an error path the app page has to handle. |
| 1016 | func TestOwnerAppMetricsEmptyHosts(t *testing.T) { |
| 1017 | srv, _ := newMetricsFixture(t) |
| 1018 | rows, err := srv.OwnerAppMetrics(context.Background(), nil, nil) |
| 1019 | if err != nil { |
| 1020 | t.Fatalf("empty hosts errored: %v", err) |
| 1021 | } |
| 1022 | if len(rows) != 0 { |
| 1023 | t.Fatalf("empty hosts = %+v, want no rows", rows) |
| 1024 | } |
| 1025 | rows, err = srv.OwnerAppMetrics(context.Background(), []string{}, nil) |
| 1026 | if err != nil || len(rows) != 0 { |
| 1027 | t.Fatalf("empty slice = %+v, %v; want no rows, no error", rows, err) |
| 1028 | } |
| 1029 | } |
| 1030 | |
| 1031 | // TestOwnerAppMetricsFiltersToOwnUnits is the boundary at the function level: |
| 1032 | // one box carries this owner's unit, another tenant's unit, and host-level |
| 1033 | // readings; only the owner's unit row comes back, and a host in the allowlist |
| 1034 | // with no unit data in the sweep comes back as an unknown (nil) row, never |
| 1035 | // zeros and never dropped. |
| 1036 | func TestOwnerAppMetricsFiltersToOwnUnits(t *testing.T) { |
| 1037 | srv, rel := newMetricsFixture(t) |
| 1038 | ctx := context.Background() |
| 1039 | // Inside metricsFreshWindow: these are the box's CURRENT readings. |
| 1040 | at := time.Now().UTC().Add(-time.Minute) |
| 1041 | if err := metrics.WriteRaw(ctx, rel, metrics.RawChunk{ |
| 1042 | V: 1, Box: "box-secret-a", Hour: metrics.HourOf(at), |
| 1043 | Samples: []metrics.Sample{{ |
| 1044 | At: at, |
| 1045 | Host: metrics.HostSample{CPUFrac: p(0.777), MemFrac: p(0.888), DiskFrac: p(0.999)}, |
| 1046 | Units: map[string]metrics.UnitSample{ |
| 1047 | "carlos-app@owned.example.com.service": {CPUFrac: p(0.11), MemBytes: pi64(1024)}, |
| 1048 | "carlos-app@other-tenant.example.com.service": {CPUFrac: p(0.55), MemBytes: pi64(4096)}, |
| 1049 | "carlos-litestream.service": {CPUFrac: p(0.66), MemBytes: pi64(8192)}, |
| 1050 | }, |
| 1051 | }}, |
| 1052 | }); err != nil { |
| 1053 | t.Fatal(err) |
| 1054 | } |
| 1055 | |
| 1056 | rows, err := srv.OwnerAppMetrics(ctx, []string{"owned.example.com", "asleep.example.com"}, nil) |
| 1057 | if err != nil { |
| 1058 | t.Fatal(err) |
| 1059 | } |
| 1060 | if len(rows) != 2 { |
| 1061 | t.Fatalf("rows = %+v, want exactly 2 (the two allowlisted hosts)", rows) |
| 1062 | } |
| 1063 | byHost := map[string]InstanceMetrics{} |
| 1064 | for _, r := range rows { |
| 1065 | byHost[r.Host] = r |
| 1066 | } |
| 1067 | owned, ok := byHost["owned.example.com"] |
| 1068 | if !ok { |
| 1069 | t.Fatalf("owned.example.com missing from %+v", rows) |
| 1070 | } |
| 1071 | if owned.CPUFrac == nil || *owned.CPUFrac != 0.11 { |
| 1072 | t.Fatalf("owned CPU = %v, want 0.11", owned.CPUFrac) |
| 1073 | } |
| 1074 | if owned.MemBytes == nil || *owned.MemBytes != 1024 { |
| 1075 | t.Fatalf("owned mem = %v, want 1024", owned.MemBytes) |
| 1076 | } |
| 1077 | if owned.SparkCPU == "" || owned.SparkMem == "" { |
| 1078 | t.Fatalf("owned sparklines empty: %+v", owned) |
| 1079 | } |
| 1080 | if !owned.Reporting() { |
| 1081 | t.Fatalf("owned should be reporting: %+v", owned) |
| 1082 | } |
| 1083 | asleep, ok := byHost["asleep.example.com"] |
| 1084 | if !ok { |
| 1085 | t.Fatalf("asleep.example.com dropped instead of rendering as unknown: %+v", rows) |
| 1086 | } |
| 1087 | if asleep.CPUFrac != nil || asleep.MemBytes != nil { |
| 1088 | t.Fatalf("a host with no unit data must be nil (unknown), not zero: %+v", asleep) |
| 1089 | } |
| 1090 | if asleep.Reporting() { |
| 1091 | t.Fatalf("asleep must not report: %+v", asleep) |
| 1092 | } |
| 1093 | // Nothing else can enter the view: no other tenant's unit, no non-app |
| 1094 | // unit, no host row. |
| 1095 | for _, r := range rows { |
| 1096 | if r.Host != "owned.example.com" && r.Host != "asleep.example.com" { |
| 1097 | t.Fatalf("row outside the allowlist: %+v", r) |
| 1098 | } |
| 1099 | if r.CPUFrac != nil && (*r.CPUFrac == 0.55 || *r.CPUFrac == 0.66 || *r.CPUFrac == 0.777) { |
| 1100 | t.Fatalf("row carries another unit's or the host's reading: %+v", r) |
| 1101 | } |
| 1102 | } |
| 1103 | } |
| 1104 | |
| 1105 | // TestOwnerAppMetricsSurvivesHealthyFlushCycle pins metricsFreshWindow's |
| 1106 | // derivation (flush cadence floor 5min + stale-serve ceiling 2min + 1min |
| 1107 | // slack = 8 minutes): a healthy agent's newest VISIBLE sample can be as old |
| 1108 | // as flush-cadence-plus-ceiling and still be current, but a sample old |
| 1109 | // enough that NOTHING short of a dead agent explains it must still flap to |
| 1110 | // no-recent-readings. Three ages, three different hosts/boxes (so the |
| 1111 | // "freshest box wins" rule in OwnerAppMetrics can't let one case's fresh |
| 1112 | // reading paper over another's staleness): |
| 1113 | // |
| 1114 | // - 5m30s: the original healthy-worst-case under the OLD (5min-flush-only) |
| 1115 | // math — comfortably inside both the old and new window. |
| 1116 | // - 7m0s: the NEW healthy worst case this fix widens the window to cover |
| 1117 | // — flush cadence's floor (5min) plus the full stale-serve ceiling |
| 1118 | // (2min) a caller's cache may be behind by. |
| 1119 | // - 8m30s: comfortably past the 8-minute window regardless of derivation |
| 1120 | // — must NOT read as current. |
| 1121 | func TestOwnerAppMetricsSurvivesHealthyFlushCycle(t *testing.T) { |
| 1122 | srv, rel := newMetricsFixture(t) |
| 1123 | ctx := context.Background() |
| 1124 | |
| 1125 | write := func(box, host string, age time.Duration) { |
| 1126 | at := time.Now().UTC().Add(-age) |
| 1127 | if err := metrics.WriteRaw(ctx, rel, metrics.RawChunk{ |
| 1128 | V: 1, Box: box, Hour: metrics.HourOf(at), |
| 1129 | Samples: []metrics.Sample{{ |
| 1130 | At: at, |
| 1131 | Units: map[string]metrics.UnitSample{ |
| 1132 | "carlos-app@" + host + ".service": {CPUFrac: p(0.42), MemBytes: pi64(2048)}, |
| 1133 | }, |
| 1134 | }}, |
| 1135 | }); err != nil { |
| 1136 | t.Fatal(err) |
| 1137 | } |
| 1138 | } |
| 1139 | write("box-healthy-1", "flush1.example.com", 5*time.Minute+30*time.Second) |
| 1140 | write("box-healthy-2", "flush2.example.com", 7*time.Minute) |
| 1141 | write("box-dead", "dead.example.com", 8*time.Minute+30*time.Second) |
| 1142 | |
| 1143 | rows, err := srv.OwnerAppMetrics(ctx, []string{"flush1.example.com", "flush2.example.com", "dead.example.com"}, nil) |
| 1144 | if err != nil { |
| 1145 | t.Fatal(err) |
| 1146 | } |
| 1147 | byHost := map[string]InstanceMetrics{} |
| 1148 | for _, r := range rows { |
| 1149 | byHost[r.Host] = r |
| 1150 | } |
| 1151 | if r := byHost["flush1.example.com"]; !r.Reporting() { |
| 1152 | t.Fatalf("a 5m30s-old sample flapped to no-recent-readings: %+v", r) |
| 1153 | } |
| 1154 | if r := byHost["flush2.example.com"]; !r.Reporting() { |
| 1155 | t.Fatalf("a 7m0s-old sample (flush floor + stale-serve ceiling) flapped to no-recent-readings: %+v", r) |
| 1156 | } |
| 1157 | if r := byHost["dead.example.com"]; r.Reporting() { |
| 1158 | t.Fatalf("an 8m30s-old sample — past the fresh window — still read as current: %+v", r) |
| 1159 | } |
| 1160 | } |
| 1161 | |
| 1162 | // TestSparkPointsInvertsYAndSkipsNil pins the coordinate math the brief |
| 1163 | // calls out explicitly: a nil sample contributes no pair at all (not a |
| 1164 | // zero), and a higher fraction plots at a SMALLER y (SVG's y axis points |
| 1165 | // down, so "up" on screen means a smaller number). |
| 1166 | func TestSparkPointsInvertsYAndSkipsNil(t *testing.T) { |
| 1167 | pts := sparkPoints([]*float64{p(0.1), nil, p(0.9)}, 240, 40) |
| 1168 | fields := strings.Fields(pts) |
| 1169 | if len(fields) != 2 { |
| 1170 | t.Fatalf("got %d coordinate pairs, want 2 (the nil sample skipped): %q", len(fields), pts) |
| 1171 | } |
| 1172 | xy := func(pair string) (x, y float64) { |
| 1173 | parts := strings.SplitN(pair, ",", 2) |
| 1174 | if len(parts) != 2 { |
| 1175 | t.Fatalf("malformed pair %q", pair) |
| 1176 | } |
| 1177 | xf, err := strconv.ParseFloat(parts[0], 64) |
| 1178 | if err != nil { |
| 1179 | t.Fatalf("bad x in %q: %v", pair, err) |
| 1180 | } |
| 1181 | yf, err := strconv.ParseFloat(parts[1], 64) |
| 1182 | if err != nil { |
| 1183 | t.Fatalf("bad y in %q: %v", pair, err) |
| 1184 | } |
| 1185 | return xf, yf |
| 1186 | } |
| 1187 | x0, y0 := xy(fields[0]) |
| 1188 | x1, y1 := xy(fields[1]) |
| 1189 | if x1 <= x0 { |
| 1190 | t.Fatalf("x did not advance across the skipped nil slot: x0=%v x1=%v", x0, x1) |
| 1191 | } |
| 1192 | if y1 >= y0 { |
| 1193 | t.Fatalf("0.9 (fields[1]) should plot ABOVE 0.1 (fields[0]) — smaller y: y0=%v y1=%v", y0, y1) |
| 1194 | } |
| 1195 | } |
| 1196 | |
| 1197 | // TestSparkPointsEmptyOrAllNil: no readable data renders no line, per the |
| 1198 | // brief ("" when no data) — never a fabricated flat line at zero. |
| 1199 | func TestSparkPointsEmptyOrAllNil(t *testing.T) { |
| 1200 | if got := sparkPoints(nil, 240, 40); got != "" { |
| 1201 | t.Fatalf("nil input = %q, want empty", got) |
| 1202 | } |
| 1203 | if got := sparkPoints([]*float64{}, 240, 40); got != "" { |
| 1204 | t.Fatalf("empty slice = %q, want empty", got) |
| 1205 | } |
| 1206 | if got := sparkPoints([]*float64{nil, nil, nil}, 240, 40); got != "" { |
| 1207 | t.Fatalf("all-nil input = %q, want empty", got) |
| 1208 | } |
| 1209 | } |
| 1210 | |
| 1211 | // TestSparkPointsDownsamples: 48h of minute samples is ~2880 readings into a |
| 1212 | // 240px-wide viewBox — 12 points per pixel of invisible detail, held in the |
| 1213 | // shared cache for every unit on every box. sparkPoints buckets down to at |
| 1214 | // most the viewBox width, and a bucket's value is the MEAN of its non-nil |
| 1215 | // readings, so a spike still lifts its bucket rather than vanishing. |
| 1216 | func TestSparkPointsDownsamples(t *testing.T) { |
| 1217 | vals := make([]*float64, 2880) |
| 1218 | for i := range vals { |
| 1219 | vals[i] = p(0.1) |
| 1220 | } |
| 1221 | vals[1000] = p(1.0) // the spike |
| 1222 | pts := strings.Fields(sparkPoints(vals, 240, 40)) |
| 1223 | if len(pts) == 0 || len(pts) > 240 { |
| 1224 | t.Fatalf("got %d coordinate pairs, want 1..240", len(pts)) |
| 1225 | } |
| 1226 | ys := make([]float64, len(pts)) |
| 1227 | for i, pair := range pts { |
| 1228 | parts := strings.SplitN(pair, ",", 2) |
| 1229 | if len(parts) != 2 { |
| 1230 | t.Fatalf("malformed pair %q", pair) |
| 1231 | } |
| 1232 | y, err := strconv.ParseFloat(parts[1], 64) |
| 1233 | if err != nil { |
| 1234 | t.Fatalf("bad y in %q: %v", pair, err) |
| 1235 | } |
| 1236 | ys[i] = y |
| 1237 | } |
| 1238 | // Exactly one bucket contains the spike, and it plots higher (smaller y) |
| 1239 | // than the flat 0.1 baseline every other bucket sits at. |
| 1240 | baseline := ys[0] |
| 1241 | raised := 0 |
| 1242 | for _, y := range ys { |
| 1243 | if y < baseline-0.001 { |
| 1244 | raised++ |
| 1245 | } |
| 1246 | } |
| 1247 | if raised != 1 { |
| 1248 | t.Fatalf("%d buckets rose above the baseline, want exactly 1 (the spike's): %v", raised, ys) |
| 1249 | } |
| 1250 | } |
| 1251 | |
| 1252 | // TestSparkPointsDownsampleKeepsNilUnknown: a bucket with no readable value |
| 1253 | // at all contributes NO point — the gap stays a gap, never a zero, even |
| 1254 | // after bucketing. |
| 1255 | func TestSparkPointsDownsampleKeepsNilUnknown(t *testing.T) { |
| 1256 | vals := make([]*float64, 1000) |
| 1257 | for i := range vals { |
| 1258 | if i < 500 { |
| 1259 | vals[i] = p(0.5) |
| 1260 | } |
| 1261 | } |
| 1262 | pts := strings.Fields(sparkPoints(vals, 240, 40)) |
| 1263 | if len(pts) == 0 || len(pts) > 120 { |
| 1264 | t.Fatalf("got %d pairs, want at most the ~120 buckets covering the readable half: %v", len(pts), pts) |
| 1265 | } |
| 1266 | for _, pair := range pts { |
| 1267 | if strings.HasSuffix(pair, ",40.00") { |
| 1268 | t.Fatalf("an all-nil bucket was plotted as zero: %q", pair) |
| 1269 | } |
| 1270 | } |
| 1271 | } |
| 1272 | |
| 1273 | // TestOwnerAppMetricsSeesExecBackedInstances pins the second allowlist key: |
| 1274 | // an exec-backed host reports under carlos-exec@<host> (the delegated |
| 1275 | // cgroup's verbatim name, no .service), and its readings reach the owner |
| 1276 | // row exactly like a unit-backed tenant's. Another tenant's exec cgroup |
| 1277 | // stays invisible — the allowlist boundary holds for the new key form too. |
| 1278 | func TestOwnerAppMetricsSeesExecBackedInstances(t *testing.T) { |
| 1279 | srv, rel := newMetricsFixture(t) |
| 1280 | ctx := context.Background() |
| 1281 | at := time.Now().UTC().Add(-time.Minute) |
| 1282 | if err := metrics.WriteRaw(ctx, rel, metrics.RawChunk{ |
| 1283 | V: 1, Box: "box-a", Hour: metrics.HourOf(at), |
| 1284 | Samples: []metrics.Sample{{ |
| 1285 | At: at, |
| 1286 | Units: map[string]metrics.UnitSample{ |
| 1287 | "carlos-exec@keymail.example.com": {CPUFrac: p(0.22), MemBytes: pi64(2048)}, |
| 1288 | "carlos-exec@other.example.com": {CPUFrac: p(0.44), MemBytes: pi64(4096)}, |
| 1289 | }, |
| 1290 | }}, |
| 1291 | }); err != nil { |
| 1292 | t.Fatal(err) |
| 1293 | } |
| 1294 | |
| 1295 | rows, err := srv.OwnerAppMetrics(ctx, []string{"keymail.example.com"}, nil) |
| 1296 | if err != nil { |
| 1297 | t.Fatal(err) |
| 1298 | } |
| 1299 | if len(rows) != 1 { |
| 1300 | t.Fatalf("rows = %+v, want exactly the one allowlisted host", rows) |
| 1301 | } |
| 1302 | got := rows[0] |
| 1303 | if got.Host != "keymail.example.com" || !got.Reporting() { |
| 1304 | t.Fatalf("exec-backed host not reporting: %+v", got) |
| 1305 | } |
| 1306 | if got.CPUFrac == nil || *got.CPUFrac != 0.22 { |
| 1307 | t.Fatalf("CPU = %v, want 0.22", got.CPUFrac) |
| 1308 | } |
| 1309 | if got.MemBytes == nil || *got.MemBytes != 2048 { |
| 1310 | t.Fatalf("mem = %v, want 2048", got.MemBytes) |
| 1311 | } |
| 1312 | if got.SparkCPU == "" || got.SparkMem == "" { |
| 1313 | t.Fatalf("sparklines empty: %+v", got) |
| 1314 | } |
| 1315 | } |
| 1316 | |
| 1317 | // TestOwnerAppMetricsExecAndUnitKeysNeverDoubleReport pins the spec's |
| 1318 | // no-double-report property: a host can't legally report under both key |
| 1319 | // forms at once (one route, one backing), but if a sweep ever carries both |
| 1320 | // — a backing flip mid-window, two boxes — the existing freshest-box rule |
| 1321 | // picks ONE deterministically. No special case, still exactly one row. |
| 1322 | func TestOwnerAppMetricsExecAndUnitKeysNeverDoubleReport(t *testing.T) { |
| 1323 | srv, rel := newMetricsFixture(t) |
| 1324 | ctx := context.Background() |
| 1325 | fresher := time.Now().UTC().Add(-time.Minute) |
| 1326 | older := fresher.Add(-2 * time.Minute) |
| 1327 | // The older box reports the host as a systemd unit, the fresher box as |
| 1328 | // an exec cgroup — the flip-mid-window shape. |
| 1329 | if err := metrics.WriteRaw(ctx, rel, metrics.RawChunk{ |
| 1330 | V: 1, Box: "box-old", Hour: metrics.HourOf(older), |
| 1331 | Samples: []metrics.Sample{{ |
| 1332 | At: older, |
| 1333 | Units: map[string]metrics.UnitSample{ |
| 1334 | "carlos-app@flip.example.com.service": {CPUFrac: p(0.99), MemBytes: pi64(9999)}, |
| 1335 | }, |
| 1336 | }}, |
| 1337 | }); err != nil { |
| 1338 | t.Fatal(err) |
| 1339 | } |
| 1340 | if err := metrics.WriteRaw(ctx, rel, metrics.RawChunk{ |
| 1341 | V: 1, Box: "box-new", Hour: metrics.HourOf(fresher), |
| 1342 | Samples: []metrics.Sample{{ |
| 1343 | At: fresher, |
| 1344 | Units: map[string]metrics.UnitSample{ |
| 1345 | "carlos-exec@flip.example.com": {CPUFrac: p(0.11), MemBytes: pi64(1111)}, |
| 1346 | }, |
| 1347 | }}, |
| 1348 | }); err != nil { |
| 1349 | t.Fatal(err) |
| 1350 | } |
| 1351 | |
| 1352 | rows, err := srv.OwnerAppMetrics(ctx, []string{"flip.example.com"}, nil) |
| 1353 | if err != nil { |
| 1354 | t.Fatal(err) |
| 1355 | } |
| 1356 | if len(rows) != 1 { |
| 1357 | t.Fatalf("rows = %+v, want exactly one row for the host, never two", rows) |
| 1358 | } |
| 1359 | got := rows[0] |
| 1360 | if got.Box != "box-new" { |
| 1361 | t.Fatalf("Box = %q, want the freshest box-new: %+v", got.Box, got) |
| 1362 | } |
| 1363 | if got.CPUFrac == nil || *got.CPUFrac != 0.11 { |
| 1364 | t.Fatalf("CPU = %v, want the freshest box's 0.11", got.CPUFrac) |
| 1365 | } |
| 1366 | } |
| 1367 | |
| 1368 | // TestOwnerAppMetricsMatchesStampedUnit pins the third allowlist key: an |
| 1369 | // operator-stamped bespoke unit (registry Route.Unit, #110) reports under |
| 1370 | // its real systemd unit name, not carlos-app@<host>.service — and the |
| 1371 | // sweep can only match it when the caller opts a host into that name via |
| 1372 | // unitByHost. Without the entry the same host stays unreached, proving the |
| 1373 | // third key is opt-in per host, not a wildcard that reopens the boundary. |
| 1374 | func TestOwnerAppMetricsMatchesStampedUnit(t *testing.T) { |
| 1375 | srv, rel := newMetricsFixture(t) |
| 1376 | ctx := context.Background() |
| 1377 | at := time.Now().UTC().Add(-time.Minute) |
| 1378 | if err := metrics.WriteRaw(ctx, rel, metrics.RawChunk{ |
| 1379 | V: 1, Box: "b1", Hour: metrics.HourOf(at), |
| 1380 | Samples: []metrics.Sample{{ |
| 1381 | At: at, |
| 1382 | Units: map[string]metrics.UnitSample{ |
| 1383 | "titogo@vicky.service": {CPUFrac: p(0.33), MemBytes: pi64(3333)}, |
| 1384 | }, |
| 1385 | }}, |
| 1386 | }); err != nil { |
| 1387 | t.Fatal(err) |
| 1388 | } |
| 1389 | |
| 1390 | rows, err := srv.OwnerAppMetrics(ctx, []string{"vicky.go.tito.io"}, |
| 1391 | map[string]string{"vicky.go.tito.io": "titogo@vicky.service"}) |
| 1392 | if err != nil { |
| 1393 | t.Fatal(err) |
| 1394 | } |
| 1395 | if len(rows) != 1 { |
| 1396 | t.Fatalf("rows = %+v, want exactly the one allowlisted host", rows) |
| 1397 | } |
| 1398 | got := rows[0] |
| 1399 | if got.Host != "vicky.go.tito.io" || !got.Reporting() { |
| 1400 | t.Fatalf("stamped-unit host not reporting: %+v", got) |
| 1401 | } |
| 1402 | if got.Box != "b1" { |
| 1403 | t.Fatalf("Box = %q, want b1", got.Box) |
| 1404 | } |
| 1405 | if got.CPUFrac == nil || *got.CPUFrac != 0.33 { |
| 1406 | t.Fatalf("CPU = %v, want 0.33", got.CPUFrac) |
| 1407 | } |
| 1408 | |
| 1409 | // Without the unitByHost entry, the same host must NOT match — the |
| 1410 | // third key is opt-in per host, not a wildcard. |
| 1411 | rows, err = srv.OwnerAppMetrics(ctx, []string{"vicky.go.tito.io"}, nil) |
| 1412 | if err != nil { |
| 1413 | t.Fatal(err) |
| 1414 | } |
| 1415 | if len(rows) != 1 { |
| 1416 | t.Fatalf("rows = %+v, want exactly the one allowlisted host", rows) |
| 1417 | } |
| 1418 | if rows[0].Reporting() { |
| 1419 | t.Fatalf("host reported without an opt-in unitByHost entry: %+v", rows[0]) |
| 1420 | } |
| 1421 | } |
| 1422 | |
amadan