amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "bytes"
5 "context"
6 "encoding/json"
7 "errors"
8 "log"
9 "math"
10 "os"
11 "path/filepath"
12 "strings"
13 "testing"
14 "time"
15
16 "github.com/carlosframework/platform/internal/metrics"
17 "github.com/carlosframework/platform/internal/release"
18 "github.com/carlosframework/platform/internal/status"
19)
20
21// TestSamplerTickAndFlush is the brief's Step-1 test, verbatim: two ticks
22// across a delta (first has no CPU reading, second derives a fraction from
23// both /proc/stat and a unit's cgroup usage_usec counters), then a flush
24// within the hour followed by a flush across an hour boundary, checking the
25// raw chunk and the day rollup it produces.
26func TestSamplerTickAndFlush(t *testing.T) {
27 ctx := context.Background()
28 store := release.NewDirStore(t.TempDir())
29 at := time.Date(2026, 8, 2, 10, 59, 30, 0, time.UTC)
30 c0, c1 := &metrics.CPUCounters{Busy: 100, Total: 1000}, &metrics.CPUCounters{Busy: 160, Total: 1100}
31 mem, disk := 0.5, 0.3
32 usec0, usec1 := int64(0), int64(3_000_000)
33 sm := &sampler{}
34 probes := func(c *metrics.CPUCounters, usec int64, now time.Time) sampleProbes {
35 return sampleProbes{
36 procStat: func() *metrics.CPUCounters { return c },
37 loadavg: func() (*float64, *float64, *float64) { return nil, nil, nil },
38 memFrac: func() *float64 { return &mem },
39 diskFrac: func() *float64 { return &disk },
40 units: func() map[string]unitRaw {
41 return map[string]unitRaw{"carlos-app@a.example.com.service": {cpuUsec: &usec}}
42 },
43 now: func() time.Time { return now },
44 }
45 }
46 sm.tick(probes(c0, usec0, at))
47 sm.tick(probes(c1, usec1, at.Add(time.Minute)))
48 if len(sm.buf) != 2 {
49 t.Fatalf("buf=%d, want 2", len(sm.buf))
50 }
51 if sm.buf[0].Host.CPUFrac != nil {
52 t.Fatal("first tick has no delta — CPU must be nil, not 0")
53 }
54 if f := sm.buf[1].Host.CPUFrac; f == nil || math.Abs(*f-0.6) > 1e-9 {
55 t.Fatalf("second tick CPU=%v, want 0.6", f)
56 }
57 if f := sm.buf[1].Units["carlos-app@a.example.com.service"].CPUFrac; f == nil || math.Abs(*f-0.05) > 1e-9 {
58 t.Fatalf("unit CPU=%v, want 0.05 (3s of CPU over 60s)", f)
59 }
60
61 // This flush call's buf already spans the 10->11 hour boundary (one
62 // sample in each hour): hour 10 is the "old" hour relative to `now` and
63 // is written to its own chunk AND rolled up into the day rollup right
64 // here, on this first flush — not deferred to the next one. Only hour
65 // 11's sample stays buffered.
66 if err := metricsFlush(ctx, store, "box-a", sm, at.Add(time.Minute)); err != nil {
67 t.Fatal(err)
68 }
69 if _, _, err := store.Get(ctx, "metrics/boxes/box-a/raw/2026-08-02-10.json"); err != nil {
70 t.Fatalf("raw chunk missing: %v", err)
71 }
72 // Both samples now buffered are in hour 11: this second flush has
73 // nothing to roll over, and the day rollup checked below is simply what
74 // the first flush already wrote.
75 sm.tick(probes(c1, usec1, at.Add(2*time.Minute))) // 11:01:30
76 if err := metricsFlush(ctx, store, "box-a", sm, at.Add(2*time.Minute)); err != nil {
77 t.Fatal(err)
78 }
79 body, _, err := store.Get(ctx, "metrics/boxes/box-a/days/2026-08-02.json")
80 if err != nil {
81 t.Fatalf("day rollup missing: %v", err)
82 }
83 var day metrics.DayRollup
84 if err := json.Unmarshal(body, &day); err != nil {
85 t.Fatal(err)
86 }
87 if _, ok := day.Hours["10"]; !ok {
88 t.Fatalf("hour 10 not rolled up: %v", day.Hours)
89 }
90}
91
92// ---- Task 5 (customer-fleets plan) fix round: metrics loop honors the fleet prefix ----
93
94// TestMetricsLoopStoreHonorsFleetPrefix pins the review fix: metricsLoop's
95// store must be wrapped in release.NewPrefixStore(CARLOS_STATUS_PREFIX)
96// exactly like statusLoop's, so a fleet box's raw samples, day rollup, and
97// alert state land under its own fleets/<name>/ namespace — not the shared
98// metrics/boxes/ + alerts/boxes/ paths a platform box uses. Without this, a
99// customer's Pi would surface on the platform operator's monitoring page
100// (console metricsview lists metrics/boxes/) and a threshold breach would
101// page the platform's own alert webhook for customer hardware. Exercises
102// metricsFlush and alertTick through a release.NewPrefixStore-wrapped
103// store — the same composition metricsLoop's store-construction line now
104// produces — rather than reimplementing the wrapping.
105func TestMetricsLoopStoreHonorsFleetPrefix(t *testing.T) {
106 ctx := context.Background()
107 under := release.NewDirStore(t.TempDir())
108 s := release.NewPrefixStore(under, "fleets/pi/")
109 box := "pi-box"
110
111 // --- metricsFlush: raw chunk + day rollup land under the prefix -------
112 sm := &sampler{}
113 at := time.Date(2026, 8, 2, 10, 59, 30, 0, time.UTC)
114 mem, disk := 0.5, 0.3
115 probes := func(now time.Time) sampleProbes {
116 return sampleProbes{
117 procStat: func() *metrics.CPUCounters { return nil },
118 loadavg: func() (*float64, *float64, *float64) { return nil, nil, nil },
119 memFrac: func() *float64 { return &mem },
120 diskFrac: func() *float64 { return &disk },
121 units: func() map[string]unitRaw { return nil },
122 now: func() time.Time { return now },
123 }
124 }
125 sm.tick(probes(at))
126 sm.tick(probes(at.Add(2 * time.Minute))) // crosses the 10 -> 11 hour boundary
127 if err := metricsFlush(ctx, s, box, sm, at.Add(2*time.Minute)); err != nil {
128 t.Fatal(err)
129 }
130 if _, _, err := under.Get(ctx, "fleets/pi/metrics/boxes/pi-box/raw/2026-08-02-10.json"); err != nil {
131 t.Errorf("raw chunk not under the fleet prefix: %v", err)
132 }
133 if _, _, err := under.Get(ctx, "metrics/boxes/pi-box/raw/2026-08-02-10.json"); err == nil {
134 t.Error("raw chunk leaked into the shared (unprefixed) metrics namespace")
135 }
136 if _, _, err := under.Get(ctx, "fleets/pi/metrics/boxes/pi-box/days/2026-08-02.json"); err != nil {
137 t.Errorf("day rollup not under the fleet prefix: %v", err)
138 }
139
140 // --- alertTick: published alert state lands under the prefix too ------
141 rules := metrics.DefaultRules()
142 st := &metrics.AlertState{V: 1, Box: box, Rules: map[string]metrics.RuleState{}}
143 firing := 0.95
144 latest := metrics.Sample{At: at, Host: metrics.HostSample{DiskFrac: &firing}}
145 statePath := filepath.Join(t.TempDir(), "monitor", "alerts.json")
146 if err := alertTick(ctx, s, box, rules, st, latest, statePath, nil); err != nil {
147 t.Fatal(err)
148 }
149 if _, _, err := under.Get(ctx, "fleets/pi/"+metrics.AlertKey(box)); err != nil {
150 t.Errorf("alert state not under the fleet prefix: %v", err)
151 }
152 if _, _, err := under.Get(ctx, metrics.AlertKey(box)); err == nil {
153 t.Error("alert state leaked into the shared (unprefixed) alerts namespace — would page the platform webhook for customer hardware")
154 }
155
156 // --- empty prefix (every platform box today) is a pure passthrough ----
157 if got := release.NewPrefixStore(under, ""); got != release.Store(under) {
158 t.Errorf("NewPrefixStore with empty prefix = %v, want the underlying store unchanged", got)
159 }
160}
161
162// TestMetricsFlushPrunesOldChunks pins the 48h prune: a raw chunk seeded
163// well past the retention window must be gone after a flush that crosses an
164// hour boundary (the only time pruning runs), while a fresh chunk survives.
165func TestMetricsFlushPrunesOldChunks(t *testing.T) {
166 ctx := context.Background()
167 store := release.NewDirStore(t.TempDir())
168 box := "box-a"
169
170 oldHour := time.Date(2026, 7, 30, 9, 0, 0, 0, time.UTC) // 3 days before "now" below
171 if err := metrics.WriteRaw(ctx, store, metrics.RawChunk{
172 V: 1, Box: box, Hour: metrics.HourOf(oldHour), Samples: []metrics.Sample{{At: oldHour}},
173 }); err != nil {
174 t.Fatal(err)
175 }
176 freshHour := time.Date(2026, 8, 2, 9, 0, 0, 0, time.UTC)
177 if err := metrics.WriteRaw(ctx, store, metrics.RawChunk{
178 V: 1, Box: box, Hour: metrics.HourOf(freshHour), Samples: []metrics.Sample{{At: freshHour}},
179 }); err != nil {
180 t.Fatal(err)
181 }
182
183 // A sampler buffer holding one sample from the hour just before "now" —
184 // this is what triggers the hour-rollover branch (rollup + prune).
185 sm := &sampler{buf: []metrics.Sample{
186 {At: time.Date(2026, 8, 2, 9, 30, 0, 0, time.UTC)},
187 }}
188 now := time.Date(2026, 8, 2, 10, 0, 5, 0, time.UTC)
189 if err := metricsFlush(ctx, store, box, sm, now); err != nil {
190 t.Fatal(err)
191 }
192
193 if _, _, err := store.Get(ctx, metrics.RawKey(box, metrics.HourOf(oldHour))); !errors.Is(err, os.ErrNotExist) {
194 t.Errorf("old raw chunk should be pruned, got err=%v", err)
195 }
196 if _, _, err := store.Get(ctx, metrics.RawKey(box, metrics.HourOf(freshHour))); err != nil {
197 t.Errorf("fresh raw chunk should survive pruning: %v", err)
198 }
199}
200
201// TestAlertTickPersistsPublishesAndPosts pins the fire path: a disk reading
202// over threshold persists local state, publishes the new AlertState to the
203// store, and posts exactly one webhook event for the transition — a second,
204// identical tick (still firing, no transition) must not post again.
205func TestAlertTickPersistsPublishesAndPosts(t *testing.T) {
206 ctx := context.Background()
207 store := release.NewDirStore(t.TempDir())
208 statePath := filepath.Join(t.TempDir(), "monitor", "alerts.json")
209 box := "box-a"
210 rules := metrics.DefaultRules()
211 st := &metrics.AlertState{V: 1, Box: box, Rules: map[string]metrics.RuleState{}}
212
213 disk := 0.95
214 latest := metrics.Sample{
215 At: time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC),
216 Host: metrics.HostSample{DiskFrac: &disk},
217 }
218
219 var posts [][]byte
220 post := func(url string, body []byte) error {
221 posts = append(posts, body)
222 return nil
223 }
224
225 if err := alertTick(ctx, store, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err != nil {
226 t.Fatal(err)
227 }
228
229 if _, err := os.Stat(statePath); err != nil {
230 t.Fatalf("state file not written: %v", err)
231 }
232 body, _, err := store.Get(ctx, metrics.AlertKey(box))
233 if err != nil {
234 t.Fatalf("published alert state missing: %v", err)
235 }
236 var published metrics.AlertState
237 if err := json.Unmarshal(body, &published); err != nil {
238 t.Fatal(err)
239 }
240 if !published.Rules["disk"].Firing {
241 t.Errorf("disk rule not firing in published state: %+v", published.Rules["disk"])
242 }
243 if len(posts) != 1 {
244 t.Fatalf("posts = %d, want 1", len(posts))
245 }
246 if !strings.Contains(string(posts[0]), `"rule":"disk"`) {
247 t.Errorf("post body = %s, want it to contain the disk rule", posts[0])
248 }
249
250 // Second, identical tick: already firing, no new transition -> no post.
251 if err := alertTick(ctx, store, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err != nil {
252 t.Fatal(err)
253 }
254 if len(posts) != 1 {
255 t.Errorf("posts after second tick = %d, want still 1 (no new transition)", len(posts))
256 }
257}
258
259// TestAlertTickWebhookFailureDoesNotBlock pins the fire-and-forget webhook
260// contract: a post failure is logged, never returned — state must still be
261// persisted locally and published to the store.
262func TestAlertTickWebhookFailureDoesNotBlock(t *testing.T) {
263 ctx := context.Background()
264 store := release.NewDirStore(t.TempDir())
265 statePath := filepath.Join(t.TempDir(), "monitor", "alerts.json")
266 box := "box-a"
267 rules := metrics.DefaultRules()
268 st := &metrics.AlertState{V: 1, Box: box, Rules: map[string]metrics.RuleState{}}
269
270 disk := 0.95
271 latest := metrics.Sample{
272 At: time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC),
273 Host: metrics.HostSample{DiskFrac: &disk},
274 }
275 post := func(url string, body []byte) error { return errors.New("webhook unreachable") }
276
277 if err := alertTick(ctx, store, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err != nil {
278 t.Fatalf("alertTick must not fail on a webhook error: %v", err)
279 }
280 if _, err := os.Stat(statePath); err != nil {
281 t.Errorf("state file must still be written: %v", err)
282 }
283 if _, _, err := store.Get(ctx, metrics.AlertKey(box)); err != nil {
284 t.Errorf("state must still be published: %v", err)
285 }
286}
287
288// TestAlertTickPersistFailureDoesNotSuppressPublish pins the review fix: a
289// failed local state write (statePath's parent is a file, not a directory,
290// so os.MkdirAll fails deterministically) must not suppress the alert
291// itself — the store publish and the webhook post are the reliable
292// channels, and the local write is most likely to fail exactly when the
293// disk rule is the one firing (disk full).
294func TestAlertTickPersistFailureDoesNotSuppressPublish(t *testing.T) {
295 ctx := context.Background()
296 store := release.NewDirStore(t.TempDir())
297 box := "box-a"
298 rules := metrics.DefaultRules()
299 st := &metrics.AlertState{V: 1, Box: box, Rules: map[string]metrics.RuleState{}}
300
301 // statePath's parent directory is actually a file: MkdirAll must fail.
302 blocker := filepath.Join(t.TempDir(), "not-a-dir")
303 if err := os.WriteFile(blocker, []byte("x"), 0o644); err != nil {
304 t.Fatal(err)
305 }
306 statePath := filepath.Join(blocker, "monitor", "alerts.json")
307
308 disk := 0.95
309 latest := metrics.Sample{
310 At: time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC),
311 Host: metrics.HostSample{DiskFrac: &disk},
312 }
313 var posted bool
314 post := func(url string, body []byte) error {
315 posted = true
316 return nil
317 }
318
319 if err := alertTick(ctx, store, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err != nil {
320 t.Fatalf("alertTick must not fail just because the local state write failed: %v", err)
321 }
322 if !posted {
323 t.Error("webhook must still be posted when only the local persist failed")
324 }
325 body, _, err := store.Get(ctx, metrics.AlertKey(box))
326 if err != nil {
327 t.Fatalf("alert state must still be published to the store: %v", err)
328 }
329 var published metrics.AlertState
330 if err := json.Unmarshal(body, &published); err != nil {
331 t.Fatal(err)
332 }
333 if !published.Rules["disk"].Firing {
334 t.Errorf("disk rule not firing in published state: %+v", published.Rules["disk"])
335 }
336}
337
338// failingPutStore wraps a release.Store and makes every Put fail — the
339// fixture for TestSamplerBufCappedUnderSustainedFlushFailure's "IAM allows
340// PutObject... no wait, denies it" scenario: reads and Deletes still work
341// (delegated via embedding), only Put is broken.
342type failingPutStore struct {
343 release.Store
344}
345
346func (f *failingPutStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
347 return "", errors.New("put always fails (test)")
348}
349
350// TestSamplerBufCappedUnderSustainedFlushFailure pins the review fix: a
351// store whose Put always fails means metricsFlush never gets far enough to
352// trim sm.buf (it returns early on the first failing WriteRaw), so without
353// a cap, buf would grow one sample per tick forever. tick itself enforces
354// samplerMaxBuf regardless of flush's fate; this ticks well past the cap,
355// confirms it held, confirms a failing flush leaves buf untouched, and then
356// confirms a subsequent flush against a real, working store still produces
357// a coherent (single-hour) chunk from whatever survived in buf.
358func TestSamplerBufCappedUnderSustainedFlushFailure(t *testing.T) {
359 ctx := context.Background()
360 box := "box-a"
361 realStore := release.NewDirStore(t.TempDir())
362 failStore := &failingPutStore{Store: realStore}
363
364 sm := &sampler{}
365 start := time.Date(2026, 8, 2, 0, 0, 0, 0, time.UTC)
366 mem, disk := 0.5, 0.3
367 for i := 0; i < samplerMaxBuf+30; i++ {
368 now := start.Add(time.Duration(i) * time.Minute)
369 sm.tick(sampleProbes{
370 procStat: func() *metrics.CPUCounters { return nil },
371 loadavg: func() (*float64, *float64, *float64) { return nil, nil, nil },
372 memFrac: func() *float64 { return &mem },
373 diskFrac: func() *float64 { return &disk },
374 units: func() map[string]unitRaw { return nil },
375 now: func() time.Time { return now },
376 })
377 }
378 if len(sm.buf) != samplerMaxBuf {
379 t.Fatalf("buf=%d after %d ticks, want capped at %d", len(sm.buf), samplerMaxBuf+30, samplerMaxBuf)
380 }
381
382 lastAt := sm.buf[len(sm.buf)-1].At
383 if err := metricsFlush(ctx, failStore, box, sm, lastAt); err == nil {
384 t.Fatal("expected metricsFlush to fail against a Put-always-fails store")
385 }
386 if len(sm.buf) != samplerMaxBuf {
387 t.Fatalf("buf=%d after a failed flush, want still capped at %d (flush must not further grow or corrupt it)", len(sm.buf), samplerMaxBuf)
388 }
389
390 // Recovery: a flush against a real, working store must still succeed
391 // and write a coherent (single-hour) chunk from what's left in buf.
392 if err := metricsFlush(ctx, realStore, box, sm, lastAt); err != nil {
393 t.Fatalf("flush against a working store: %v", err)
394 }
395 body, _, err := realStore.Get(ctx, metrics.RawKey(box, metrics.HourOf(lastAt)))
396 if err != nil {
397 t.Fatalf("current-hour chunk missing after recovery flush: %v", err)
398 }
399 var chunk metrics.RawChunk
400 if err := json.Unmarshal(body, &chunk); err != nil {
401 t.Fatal(err)
402 }
403 if len(chunk.Samples) == 0 {
404 t.Fatal("recovered chunk has no samples")
405 }
406 for _, samp := range chunk.Samples {
407 if metrics.HourOf(samp.At) != metrics.HourOf(lastAt) {
408 t.Errorf("chunk mixes in a sample from a different hour: %v", samp.At)
409 }
410 }
411}
412
413// TestMetricsFlushMergesExistingChunkAfterRestart pins the review fix: a
414// restart resets sm.buf to empty, but the still-open current hour's raw
415// chunk may already hold samples this same box wrote before the crash.
416// WriteRaw is a wholesale replace, so the first flush after a restart must
417// merge the pre-existing chunk's samples in rather than clobber them — that
418// loss would otherwise become permanent the moment the hour rolls up.
419func TestMetricsFlushMergesExistingChunkAfterRestart(t *testing.T) {
420 ctx := context.Background()
421 store := release.NewDirStore(t.TempDir())
422 box := "box-a"
423
424 hourStart := time.Date(2026, 8, 2, 14, 0, 0, 0, time.UTC)
425 preRestart := []metrics.Sample{
426 {At: hourStart.Add(1 * time.Minute)},
427 {At: hourStart.Add(2 * time.Minute)},
428 }
429 if err := metrics.WriteRaw(ctx, store, metrics.RawChunk{
430 V: 1, Box: box, Hour: metrics.HourOf(hourStart), Samples: preRestart,
431 }); err != nil {
432 t.Fatal(err)
433 }
434
435 // A fresh sampler, as if the process just restarted, with one sample
436 // collected after the restart — still within the same hour.
437 sm := &sampler{buf: []metrics.Sample{
438 {At: hourStart.Add(3 * time.Minute)},
439 }}
440 now := hourStart.Add(3 * time.Minute)
441 if err := metricsFlush(ctx, store, box, sm, now); err != nil {
442 t.Fatal(err)
443 }
444
445 body, _, err := store.Get(ctx, metrics.RawKey(box, metrics.HourOf(hourStart)))
446 if err != nil {
447 t.Fatalf("chunk missing after merge flush: %v", err)
448 }
449 var chunk metrics.RawChunk
450 if err := json.Unmarshal(body, &chunk); err != nil {
451 t.Fatal(err)
452 }
453 if len(chunk.Samples) != 3 {
454 t.Fatalf("chunk has %d samples, want 3 (2 pre-restart + 1 post-restart)", len(chunk.Samples))
455 }
456}
457
458// TestMetricsFlushMergesOldHourAfterRestartAcrossBoundary pins the
459// round-2 review fix: mergeExistingChunk must apply to EVERY hour this
460// process writes for the first time, not just whichever hour happens to be
461// "current" at the moment of the first flush. Reproduces the exact
462// scenario the review found: 58 pre-crash samples for hour 09, a restart
463// whose very first flush already spans the 09->10 boundary (so hour 09
464// lands in metricsFlush's "old" bucket on this process's first-ever write
465// to it) — the hour-09 chunk must still come out merged (59 samples, not
466// 2), and the day rollup's "09" entry must be computed from that merged
467// set, not just the post-restart tick.
468func TestMetricsFlushMergesOldHourAfterRestartAcrossBoundary(t *testing.T) {
469 ctx := context.Background()
470 store := release.NewDirStore(t.TempDir())
471 box := "box-a"
472
473 hour09Start := time.Date(2026, 8, 2, 9, 0, 0, 0, time.UTC)
474 preCrashCPU := 0.1
475 preCrash := make([]metrics.Sample, 0, 58)
476 for i := 0; i < 58; i++ {
477 v := preCrashCPU
478 preCrash = append(preCrash, metrics.Sample{
479 At: hour09Start.Add(time.Duration(i) * time.Minute),
480 Host: metrics.HostSample{CPUFrac: &v},
481 })
482 }
483 if err := metrics.WriteRaw(ctx, store, metrics.RawChunk{
484 V: 1, Box: box, Hour: metrics.HourOf(hour09Start), Samples: preCrash,
485 }); err != nil {
486 t.Fatal(err)
487 }
488
489 // Restart at 09:58: a fresh sampler whose buf, by the time of its
490 // first-ever flush, already holds one hour-09 tick and one hour-10
491 // tick — the flush call itself spans the boundary.
492 postRestartCPU := 0.2
493 sm := &sampler{buf: []metrics.Sample{
494 {At: time.Date(2026, 8, 2, 9, 58, 0, 0, time.UTC), Host: metrics.HostSample{CPUFrac: &postRestartCPU}},
495 {At: time.Date(2026, 8, 2, 10, 3, 0, 0, time.UTC), Host: metrics.HostSample{CPUFrac: &postRestartCPU}},
496 }}
497 now := time.Date(2026, 8, 2, 10, 3, 0, 0, time.UTC)
498 if err := metricsFlush(ctx, store, box, sm, now); err != nil {
499 t.Fatal(err)
500 }
501
502 body, _, err := store.Get(ctx, metrics.RawKey(box, metrics.HourOf(hour09Start)))
503 if err != nil {
504 t.Fatalf("hour-09 chunk missing: %v", err)
505 }
506 var chunk metrics.RawChunk
507 if err := json.Unmarshal(body, &chunk); err != nil {
508 t.Fatal(err)
509 }
510 if len(chunk.Samples) != 59 {
511 t.Fatalf("hour-09 chunk has %d samples, want 59 (58 pre-crash + 1 post-restart) — the pre-crash samples were clobbered", len(chunk.Samples))
512 }
513
514 dayBody, _, err := store.Get(ctx, metrics.DayKey(box, "2026-08-02"))
515 if err != nil {
516 t.Fatalf("day rollup missing: %v", err)
517 }
518 var day metrics.DayRollup
519 if err := json.Unmarshal(dayBody, &day); err != nil {
520 t.Fatal(err)
521 }
522 if n := day.Hours["09"].Host["cpu"].N; n != 59 {
523 t.Fatalf("day rollup hour 09 cpu N=%d, want 59 — the rollup must be computed from the merged set, not just the post-restart tick", n)
524 }
525}
526
527// TestAlertTickRetriesTransitionAfterPublishFailure pins the round-2 review
528// fix: a publish failure must NOT advance *st, so the next identical
529// observation reproduces the same transition and retries both the store
530// publish and the webhook post — committing *st unconditionally would
531// otherwise make Evaluate believe the rule is already firing, emit no
532// further events, and permanently swallow the alert.
533func TestAlertTickRetriesTransitionAfterPublishFailure(t *testing.T) {
534 ctx := context.Background()
535 box := "box-a"
536 rules := metrics.DefaultRules()
537 st := &metrics.AlertState{V: 1, Box: box, Rules: map[string]metrics.RuleState{}}
538 statePath := filepath.Join(t.TempDir(), "monitor", "alerts.json")
539
540 disk := 0.95
541 latest := metrics.Sample{
542 At: time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC),
543 Host: metrics.HostSample{DiskFrac: &disk},
544 }
545
546 posts := 0
547 post := func(url string, body []byte) error {
548 posts++
549 return nil
550 }
551
552 failStore := &failingPutStore{Store: release.NewDirStore(t.TempDir())}
553 if err := alertTick(ctx, failStore, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err == nil {
554 t.Fatal("expected alertTick to return the publish error against a Put-always-fails store")
555 }
556 if st.Rules["disk"].Firing {
557 t.Fatal("st must NOT be mutated to firing when publish failed — the transition must remain pending for a retry")
558 }
559 if posts != 1 {
560 t.Errorf("posts = %d, want 1 (webhook still attempted despite the publish failure)", posts)
561 }
562
563 // Retry with the SAME observation against a working store: the pending
564 // transition must reproduce, publish, and post again.
565 workingStore := release.NewDirStore(t.TempDir())
566 if err := alertTick(ctx, workingStore, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err != nil {
567 t.Fatalf("retry against a working store: %v", err)
568 }
569 if !st.Rules["disk"].Firing {
570 t.Fatal("st must be firing after the retried publish succeeds")
571 }
572 if posts != 2 {
573 t.Errorf("posts = %d, want 2 (event re-emitted and re-posted on retry)", posts)
574 }
575 body, _, err := workingStore.Get(ctx, metrics.AlertKey(box))
576 if err != nil {
577 t.Fatalf("AlertKey must be published on the retry: %v", err)
578 }
579 var published metrics.AlertState
580 if err := json.Unmarshal(body, &published); err != nil {
581 t.Fatal(err)
582 }
583 if !published.Rules["disk"].Firing {
584 t.Errorf("published state disk rule not firing: %+v", published.Rules["disk"])
585 }
586}
587
588// putFailsOnceForKey wraps a release.Store and makes Put fail exactly once
589// for one specific key — the fixture for
590// TestMetricsFlushPartialFailureDoesNotRegressEarlierHour's "a later hour's
591// write fails, an earlier hour's succeeded" scenario. Every other key, and
592// every call to failKey after the first, passes through unchanged.
593type putFailsOnceForKey struct {
594 release.Store
595 failKey string
596 failed bool
597}
598
599func (f *putFailsOnceForKey) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
600 if key == f.failKey && !f.failed {
601 f.failed = true
602 return "", errors.New("put fails once for this key (test)")
603 }
604 return f.Store.Put(ctx, key, body, ifMatch)
605}
606
607// TestMetricsFlushPartialFailureDoesNotRegressEarlierHour pins the
608// round-3 review's Critical fix: writtenHours must only be marked (and a
609// hour retired from buf) once ALL of that hour's durable writes actually
610// succeeded. Two old hours (A older, B less old) each already hold a
611// pre-crash sample in the store; a freshly "restarted" sampler holds one
612// post-restart sample for each in buf. The first flush call's Put fails
613// only for hour B's raw key (hour A, processed first, succeeds in full).
614// Before the round-3 fix, hour A would have been marked written and
615// dropped from buf only via the final `sm.buf = curSamples` at the very
616// end of metricsFlush, which this failure never reaches — so the retry
617// would reprocess hour A too, see it (wrongly) already marked, skip the
618// merge, and overwrite its correct 2-sample chunk with just the bare
619// 1-sample in-memory set. This test asserts hour A's chunk survives the
620// retry intact, and hour B ends up correctly merged too.
621func TestMetricsFlushPartialFailureDoesNotRegressEarlierHour(t *testing.T) {
622 ctx := context.Background()
623 realStore := release.NewDirStore(t.TempDir())
624 box := "box-a"
625
626 hourAStart := time.Date(2026, 8, 2, 8, 0, 0, 0, time.UTC)
627 hourBStart := time.Date(2026, 8, 2, 9, 0, 0, 0, time.UTC)
628
629 cpuA := 0.1
630 preA := []metrics.Sample{{At: hourAStart.Add(10 * time.Minute), Host: metrics.HostSample{CPUFrac: &cpuA}}}
631 if err := metrics.WriteRaw(ctx, realStore, metrics.RawChunk{
632 V: 1, Box: box, Hour: metrics.HourOf(hourAStart), Samples: preA,
633 }); err != nil {
634 t.Fatal(err)
635 }
636 cpuB := 0.2
637 preB := []metrics.Sample{{At: hourBStart.Add(10 * time.Minute), Host: metrics.HostSample{CPUFrac: &cpuB}}}
638 if err := metrics.WriteRaw(ctx, realStore, metrics.RawChunk{
639 V: 1, Box: box, Hour: metrics.HourOf(hourBStart), Samples: preB,
640 }); err != nil {
641 t.Fatal(err)
642 }
643
644 // A fresh (post-restart) sampler with one post-restart tick in each old
645 // hour — no current-hour sample, so the whole flush is "old hours".
646 postA, postB := 0.3, 0.4
647 sm := &sampler{buf: []metrics.Sample{
648 {At: hourAStart.Add(20 * time.Minute), Host: metrics.HostSample{CPUFrac: &postA}},
649 {At: hourBStart.Add(20 * time.Minute), Host: metrics.HostSample{CPUFrac: &postB}},
650 }}
651 now := time.Date(2026, 8, 2, 10, 5, 0, 0, time.UTC) // current hour 10 — both A and B are old
652
653 hourBKey := metrics.RawKey(box, metrics.HourOf(hourBStart))
654 failingStore := &putFailsOnceForKey{Store: realStore, failKey: hourBKey}
655
656 if err := metricsFlush(ctx, failingStore, box, sm, now); err == nil {
657 t.Fatal("expected the first flush to fail (hour B's Put fails once)")
658 }
659
660 // Hour A must already be fully durable: merged (2-sample) chunk.
661 bodyA, _, err := realStore.Get(ctx, metrics.RawKey(box, metrics.HourOf(hourAStart)))
662 if err != nil {
663 t.Fatalf("hour A chunk missing after the partial failure: %v", err)
664 }
665 var chunkA metrics.RawChunk
666 if err := json.Unmarshal(bodyA, &chunkA); err != nil {
667 t.Fatal(err)
668 }
669 if len(chunkA.Samples) != 2 {
670 t.Fatalf("hour A chunk has %d samples after the partial failure, want 2 (merged)", len(chunkA.Samples))
671 }
672
673 // Retry: hour B's Put now succeeds.
674 if err := metricsFlush(ctx, failingStore, box, sm, now); err != nil {
675 t.Fatalf("retry flush: %v", err)
676 }
677
678 // Hour A must STILL be the merged 2-sample chunk — the retry must not
679 // have reprocessed and regressed it.
680 bodyA2, _, err := realStore.Get(ctx, metrics.RawKey(box, metrics.HourOf(hourAStart)))
681 if err != nil {
682 t.Fatal(err)
683 }
684 var chunkA2 metrics.RawChunk
685 if err := json.Unmarshal(bodyA2, &chunkA2); err != nil {
686 t.Fatal(err)
687 }
688 if len(chunkA2.Samples) != 2 {
689 t.Fatalf("hour A chunk regressed to %d samples after the retry, want still 2", len(chunkA2.Samples))
690 }
691
692 // Hour B must now be correctly merged.
693 bodyB, _, err := realStore.Get(ctx, metrics.RawKey(box, metrics.HourOf(hourBStart)))
694 if err != nil {
695 t.Fatalf("hour B chunk missing after retry: %v", err)
696 }
697 var chunkB metrics.RawChunk
698 if err := json.Unmarshal(bodyB, &chunkB); err != nil {
699 t.Fatal(err)
700 }
701 if len(chunkB.Samples) != 2 {
702 t.Fatalf("hour B chunk has %d samples, want 2 (merged)", len(chunkB.Samples))
703 }
704
705 dayBody, _, err := realStore.Get(ctx, metrics.DayKey(box, "2026-08-02"))
706 if err != nil {
707 t.Fatalf("day rollup missing: %v", err)
708 }
709 var day metrics.DayRollup
710 if err := json.Unmarshal(dayBody, &day); err != nil {
711 t.Fatal(err)
712 }
713 if n := day.Hours["08"].Host["cpu"].N; n != 2 {
714 t.Errorf("day rollup hour 08 cpu N=%d, want 2 (merged count)", n)
715 }
716 if n := day.Hours["09"].Host["cpu"].N; n != 2 {
717 t.Errorf("day rollup hour 09 cpu N=%d, want 2 (merged count)", n)
718 }
719}
720
721// TestMetricsFlushBackfillsPrevHourRollupAfterRestart pins the round-3 review
722// fix: rollups were derived only from in-memory buf, so an agent that stopped
723// at 09:58 and came back at 10:07 left hour 09's raw chunk — durably written
724// before the stop — with no day-rollup entry, ever. Nothing else rolls that
725// hour up afterwards, and the 48h prune deletes the chunk two days later, so
726// the hour was permanently lost from the trend history. The first flush of a
727// process now looks back one hour and rolls up the stored chunk if the day
728// object lacks it.
729func TestMetricsFlushBackfillsPrevHourRollupAfterRestart(t *testing.T) {
730 ctx := context.Background()
731 store := release.NewDirStore(t.TempDir())
732 box := "box-a"
733
734 hour09Start := time.Date(2026, 8, 2, 9, 0, 0, 0, time.UTC)
735 cpu := 0.4
736 preStop := make([]metrics.Sample, 0, 10)
737 for i := 0; i < 10; i++ {
738 v := cpu
739 preStop = append(preStop, metrics.Sample{
740 At: hour09Start.Add(time.Duration(i) * time.Minute),
741 Host: metrics.HostSample{CPUFrac: &v},
742 })
743 }
744 if err := metrics.WriteRaw(ctx, store, metrics.RawChunk{
745 V: 1, Box: box, Hour: metrics.HourOf(hour09Start), Samples: preStop,
746 }); err != nil {
747 t.Fatal(err)
748 }
749 // No day rollup at all: the agent stopped before hour 09 ever closed.
750 if _, _, err := store.Get(ctx, metrics.DayKey(box, "2026-08-02")); err == nil {
751 t.Fatal("fixture invalid: a day rollup already exists")
752 }
753
754 // Restart at 10:07: a fresh sampler with one hour-10 tick, so the flush
755 // itself never touches hour 09.
756 postCPU := 0.9
757 sm := &sampler{buf: []metrics.Sample{
758 {At: time.Date(2026, 8, 2, 10, 7, 0, 0, time.UTC), Host: metrics.HostSample{CPUFrac: &postCPU}},
759 }}
760 now := time.Date(2026, 8, 2, 10, 7, 0, 0, time.UTC)
761 if err := metricsFlush(ctx, store, box, sm, now); err != nil {
762 t.Fatal(err)
763 }
764
765 body, _, err := store.Get(ctx, metrics.DayKey(box, "2026-08-02"))
766 if err != nil {
767 t.Fatalf("day rollup missing — hour 09 was never rolled up: %v", err)
768 }
769 var day metrics.DayRollup
770 if err := json.Unmarshal(body, &day); err != nil {
771 t.Fatal(err)
772 }
773 agg, ok := day.Hours["09"]
774 if !ok {
775 t.Fatalf("day rollup has no hour 09: %+v", day.Hours)
776 }
777 if n := agg.Host["cpu"].N; n != 10 {
778 t.Fatalf("hour 09 cpu N=%d, want 10 (rolled up from the stored raw chunk)", n)
779 }
780 if avg := agg.Host["cpu"].Avg; avg < 0.399 || avg > 0.401 {
781 t.Fatalf("hour 09 cpu avg=%v, want ~0.4 (the stored chunk's readings)", avg)
782 }
783}
784
785// TestMetricsFlushBackfillLeavesExistingRollupAlone: the backfill is repair
786// work for a gap, not a rewrite. An hour whose rollup is already present —
787// the normal case, written when the hour closed under the previous process —
788// must be left exactly as it is, and the backfill must not run again on
789// later flushes of the same process.
790func TestMetricsFlushBackfillLeavesExistingRollupAlone(t *testing.T) {
791 ctx := context.Background()
792 store := release.NewDirStore(t.TempDir())
793 box := "box-a"
794
795 hour09Start := time.Date(2026, 8, 2, 9, 0, 0, 0, time.UTC)
796 cpu := 0.4
797 if err := metrics.WriteRaw(ctx, store, metrics.RawChunk{
798 V: 1, Box: box, Hour: metrics.HourOf(hour09Start),
799 Samples: []metrics.Sample{
800 {At: hour09Start, Host: metrics.HostSample{CPUFrac: &cpu}},
801 {At: hour09Start.Add(time.Minute), Host: metrics.HostSample{CPUFrac: &cpu}},
802 },
803 }); err != nil {
804 t.Fatal(err)
805 }
806 // A rollup that plainly did NOT come from the two samples above, so a
807 // clobber is unmistakable.
808 sentinel := metrics.HourAgg{Host: map[string]metrics.Agg{"cpu": {Min: 0.1, Avg: 0.2, Max: 0.3, N: 57}}}
809 if err := metrics.UpsertRollup(ctx, store, box, hour09Start, sentinel); err != nil {
810 t.Fatal(err)
811 }
812
813 postCPU := 0.9
814 sm := &sampler{buf: []metrics.Sample{
815 {At: time.Date(2026, 8, 2, 10, 7, 0, 0, time.UTC), Host: metrics.HostSample{CPUFrac: &postCPU}},
816 }}
817 if err := metricsFlush(ctx, store, box, sm, time.Date(2026, 8, 2, 10, 7, 0, 0, time.UTC)); err != nil {
818 t.Fatal(err)
819 }
820
821 readHour09 := func() metrics.HourAgg {
822 t.Helper()
823 body, _, err := store.Get(ctx, metrics.DayKey(box, "2026-08-02"))
824 if err != nil {
825 t.Fatal(err)
826 }
827 var day metrics.DayRollup
828 if err := json.Unmarshal(body, &day); err != nil {
829 t.Fatal(err)
830 }
831 return day.Hours["09"]
832 }
833 if got := readHour09().Host["cpu"]; got != sentinel.Host["cpu"] {
834 t.Fatalf("hour 09 cpu = %+v, want the pre-existing %+v untouched", got, sentinel.Host["cpu"])
835 }
836
837 // A second flush must not reconsider it either.
838 sm.buf = []metrics.Sample{{At: time.Date(2026, 8, 2, 10, 12, 0, 0, time.UTC), Host: metrics.HostSample{CPUFrac: &postCPU}}}
839 if err := metricsFlush(ctx, store, box, sm, time.Date(2026, 8, 2, 10, 12, 0, 0, time.UTC)); err != nil {
840 t.Fatal(err)
841 }
842 if got := readHour09().Host["cpu"]; got != sentinel.Host["cpu"] {
843 t.Fatalf("second flush changed hour 09 to %+v", got)
844 }
845}
846
847// TestLogRejectedAlertOverrides: a threshold override that RulesFromEnv
848// rejects (it wants a percent integer, and "0.95" — the fraction the README
849// used to describe — is not one) must not vanish silently. An accepted value
850// and an unset one say nothing.
851func TestLogRejectedAlertOverrides(t *testing.T) {
852 env := map[string]string{
853 "CARLOS_ALERT_DISK_PCT": "0.95", // a fraction: rejected
854 "CARLOS_ALERT_MEM_PCT": "85", // accepted
855 // CARLOS_ALERT_CPU_PCT unset
856 }
857 var buf bytes.Buffer
858 prev := log.Writer()
859 log.SetOutput(&buf)
860 defer log.SetOutput(prev)
861
862 logRejectedAlertOverrides(func(k string) string { return env[k] })
863
864 out := buf.String()
865 if !strings.Contains(out, "CARLOS_ALERT_DISK_PCT") {
866 t.Errorf("a rejected override logged nothing: %q", out)
867 }
868 if strings.Contains(out, "CARLOS_ALERT_MEM_PCT") || strings.Contains(out, "CARLOS_ALERT_CPU_PCT") {
869 t.Errorf("an accepted or unset override was logged: %q", out)
870 }
871}
872
873// TestMetricsBoxLabelUsesFleetIdentity is I5 (final customer-fleets
874// review): metricsLoop's published label must be the SAME resolved
875// fleet-box identity statusLoop uses (fleetidentity.go's
876// FleetIdentity.Label), never status.Label(machineID)'s hashed pseudonym,
877// whenever CARLOS_FLEET is set — otherwise a fleet box's metrics/alerts
878// would land under a different key than its own heartbeat.
879func TestMetricsBoxLabelUsesFleetIdentity(t *testing.T) {
880 const machineID = "test-machine-id"
881
882 t.Run("no fleet env falls back to the hashed pseudonym", func(t *testing.T) {
883 got := metricsBoxLabel(envFunc(map[string]string{}), machineID)
884 if want := status.Label(machineID); got != want {
885 t.Fatalf("box label = %q, want the hashed pseudonym %q", got, want)
886 }
887 })
888
889 t.Run("fleet env resolves to the fleet-box label", func(t *testing.T) {
890 got := metricsBoxLabel(envFunc(map[string]string{
891 "CARLOS_FLEET": "pi",
892 "CARLOS_FLEET_BOX": "pi-box",
893 }), machineID)
894 if got != "pi-box" {
895 t.Fatalf("box label = %q, want %q (the fleet-box label, not a hash of %q)", got, "pi-box", machineID)
896 }
897 if got == status.Label(machineID) {
898 t.Fatalf("box label collided with the hashed pseudonym — fleet identity was not used")
899 }
900 })
901
902 t.Run("invalid fleet identity degrades to the hashed pseudonym and warns", func(t *testing.T) {
903 var buf bytes.Buffer
904 prev := log.Writer()
905 log.SetOutput(&buf)
906 defer log.SetOutput(prev)
907
908 got := metricsBoxLabel(envFunc(map[string]string{
909 "CARLOS_FLEET": "pi",
910 // no CARLOS_FLEET_BOX, no CARLOS_TUNNEL_NAME to derive one from.
911 }), machineID)
912 if want := status.Label(machineID); got != want {
913 t.Fatalf("box label = %q, want the hashed pseudonym %q on invalid identity", got, want)
914 }
915 if !strings.Contains(buf.String(), "hashed pseudonym") {
916 t.Errorf("invalid fleet identity logged nothing: %q", buf.String())
917 }
918 })
919}
920
921// webhookSinks is the sink set the pre-SNS tests meant: a webhook and
922// nothing else.
923func webhookSinks(url string, post func(url string, body []byte) error) *alertSinks {
924 return newAlertSinks(url, post, "", nil, &alertCounters{})
925}
926