amadan
host: fix three review defects in the metrics loop
Cap sm.buf at ~2h of ticks (samplerMaxBuf=120) so a store that's up for reads but persistently failing writes can't grow it forever — the cap now lives in tick, not in the flush path that only trims on full success. Decouple alertTick's local state persist from publish+webhook: a failed write to statePath is logged and no longer suppresses the alert, since it's most likely to fail exactly when the disk rule is the one firing (disk full). A genuine publish failure (the store write) is still returned. Merge the current hour's pre-existing raw chunk into the first flush after a restart instead of letting WriteRaw's wholesale replace silently clobber samples collected before the crash — the loss would otherwise become permanent once that hour rolls up into the day aggregate. Also corrects a stale comment in TestSamplerTickAndFlush: the first flush in that test already crosses the hour boundary and rolls up hour 10 on the spot, not "no rollup yet".
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2 files changed,
+273
−18
internal/host/metrics_loop.go+95 −16internal/host/metrics_loop_test.go+178 −2
diff --git a/internal/host/metrics_loop.go b/internal/host/metrics_loop.go| index 6b99e7164..e136c9c4b 100644 |
| --- a/internal/host/metrics_loop.go |
| +++ b/internal/host/metrics_loop.go |
| @@ -38,16 +38,28 @@ type sampleProbes struct { |
| now func() time.Time |
| } |
| +// samplerMaxBuf hard-caps sm.buf at ~2h of 60s ticks. metricsFlush only |
| +// trims buf on a fully successful flush, so a store that's up for reads but |
| +// persistently failing writes (an IAM policy that allows PutObject but |
| +// denies DeleteObject, a full disk, whatever) would otherwise grow buf by |
| +// one sample every tick forever, re-marshaling the whole (ever-growing) |
| +// buffer on every retry — this bounds that regardless of how flush is |
| +// doing, since the cap is enforced in tick, not in flush. |
| +const samplerMaxBuf = 120 |
| + |
| // sampler carries the delta state between ticks: the previous /proc/stat |
| // reading (for host CPU), the previous per-unit cgroup usage_usec reading |
| // (for unit CPU), the wall-clock time of the previous tick (unit CPU needs |
| -// elapsed wall time, not just the counter delta), and the buffered samples |
| -// awaiting their next flush. |
| +// elapsed wall time, not just the counter delta), the buffered samples |
| +// awaiting their next flush (hard-capped at samplerMaxBuf), and whether |
| +// this sampler has completed at least one flush of the current hour yet — |
| +// see metricsFlush's restart-merge doc comment for why that matters. |
| type sampler struct { |
| - prevCPU *metrics.CPUCounters |
| - prevUnit map[string]*int64 // unit -> cpu usec at previous tick |
| - prevAt time.Time |
| - buf []metrics.Sample |
| + prevCPU *metrics.CPUCounters |
| + prevUnit map[string]*int64 // unit -> cpu usec at previous tick |
| + prevAt time.Time |
| + buf []metrics.Sample |
| + firstFlushDone bool |
| } |
| // tick reads p once and appends one metrics.Sample to sm.buf. The very |
| @@ -103,6 +115,12 @@ func (sm *sampler) tick(p sampleProbes) { |
| Units: unitSamples, |
| }) |
| + if len(sm.buf) > samplerMaxBuf { |
| + drop := len(sm.buf) - samplerMaxBuf |
| + log.Printf("host: sampler buffer exceeded %d samples (flush stuck?) — dropping %d oldest unflushed sample(s)", samplerMaxBuf, drop) |
| + sm.buf = sm.buf[drop:] |
| + } |
| + |
| sm.prevCPU = curCPU |
| sm.prevUnit = newPrevUnit |
| sm.prevAt = now |
| @@ -165,6 +183,14 @@ func metricsFlush(ctx context.Context, s release.Store, box string, sm *sampler, |
| } |
| if len(curSamples) > 0 { |
| + if !sm.firstFlushDone { |
| + merged, err := mergeExistingChunk(ctx, s, box, curHour, curSamples) |
| + if err != nil { |
| + return err |
| + } |
| + curSamples = merged |
| + sm.firstFlushDone = true |
| + } |
| if err := metrics.WriteRaw(ctx, s, metrics.RawChunk{V: 1, Box: box, Hour: curHour, Samples: curSamples}); err != nil { |
| return err |
| } |
| @@ -174,6 +200,51 @@ func metricsFlush(ctx context.Context, s release.Store, box string, sm *sampler, |
| return nil |
| } |
| +// mergeExistingChunk reads whatever raw chunk already exists for hour (the |
| +// box's still-open current-hour chunk, most likely written by this same box |
| +// before a restart) and merges it under inMemory: existing samples first, |
| +// then the in-memory ones, deduping on exact At-timestamp collisions with |
| +// the in-memory sample winning (it's the fresher read of that instant). A |
| +// fresh sampler's buf always starts empty after a restart, and WriteRaw is |
| +// a wholesale replace — without this, the first flush after a restart would |
| +// silently clobber everything the box measured earlier in the still-open |
| +// hour, and that loss would propagate permanently into the hour's day |
| +// rollup once it closes. Only called once per sampler lifetime (gated by |
| +// sm.firstFlushDone), since every flush after the first already carries |
| +// forward everything this process itself wrote to the current hour. A |
| +// missing chunk (no prior write this hour — the common case) is not an |
| +// error; a corrupt one is logged and skipped rather than failing the flush, |
| +// since a garbled pre-restart file must not block ongoing monitoring. |
| +func mergeExistingChunk(ctx context.Context, s release.Store, box, hour string, inMemory []metrics.Sample) ([]metrics.Sample, error) { |
| + key := metrics.RawKey(box, hour) |
| + body, _, err := s.Get(ctx, key) |
| + if err != nil { |
| + if errors.Is(err, os.ErrNotExist) { |
| + return inMemory, nil |
| + } |
| + return nil, err |
| + } |
| + var existing metrics.RawChunk |
| + if err := json.Unmarshal(body, &existing); err != nil { |
| + log.Printf("host: existing raw chunk %s unreadable, flushing without it: %v", key, err) |
| + return inMemory, nil |
| + } |
| + |
| + inMemoryAt := make(map[time.Time]bool, len(inMemory)) |
| + for _, samp := range inMemory { |
| + inMemoryAt[samp.At] = true |
| + } |
| + |
| + merged := make([]metrics.Sample, 0, len(existing.Samples)+len(inMemory)) |
| + for _, samp := range existing.Samples { |
| + if !inMemoryAt[samp.At] { |
| + merged = append(merged, samp) |
| + } |
| + } |
| + merged = append(merged, inMemory...) |
| + return merged, nil |
| +} |
| + |
| // pruneStaleRaw deletes this box's raw chunks older than 48h — List + |
| // metrics.StaleRawKeys (which never returns a key it cannot parse — see its |
| // own doc comment) + Delete. |
| @@ -197,10 +268,18 @@ func pruneStaleRaw(ctx context.Context, s release.Store, box string, now time.Ti |
| // actually changed (Evaluate emits Events exactly on those transitions — |
| // see its own doc comment — so "any events" is exactly the "state changed" |
| // test), and posts each event to webhookURL ("" skips the webhook |
| -// entirely). A webhook post failure is logged and otherwise ignored: |
| -// alertTick's job is to persist and publish reliably, and it must not let an |
| -// unreachable third party stop that — the console watcher polling |
| -// AlertKey(box) is the reliable channel, the webhook is best-effort. |
| +// entirely). |
| +// |
| +// A local persist failure is logged and otherwise ignored — it must NEVER |
| +// suppress publish or the webhook post below. The local write is most |
| +// likely to fail exactly when the disk rule is the one firing (disk full), |
| +// which is precisely the alert that must not go silent because of it; the |
| +// store publish and the webhook are the channels that actually reach |
| +// anyone, and neither depends on the local file. A webhook post failure is |
| +// likewise logged and otherwise ignored: the console watcher polling |
| +// AlertKey(box) is the reliable channel, the webhook is best-effort. A |
| +// genuine publish failure (the store write itself) IS still returned, since |
| +// unlike the local file, that's the reliable channel actually failing. |
| func alertTick(ctx context.Context, s release.Store, box string, rules metrics.Rules, |
| st *metrics.AlertState, latest metrics.Sample, statePath, webhookURL string, |
| post func(url string, body []byte) error) error { |
| @@ -215,16 +294,16 @@ func alertTick(ctx context.Context, s release.Store, box string, rules metrics.R |
| newState, events := metrics.Evaluate(rules, *st, obs) |
| if err := persistAlertState(statePath, newState); err != nil { |
| - return err |
| + log.Printf("host: alert state persist to %s failed, continuing to publish/post: %v", statePath, err) |
| } |
| + var publishErr error |
| if len(events) > 0 { |
| body, err := json.MarshalIndent(newState, "", " ") |
| if err != nil { |
| - return err |
| - } |
| - if err := putAlertState(ctx, s, metrics.AlertKey(box), body); err != nil { |
| - return err |
| + publishErr = err |
| + } else if err := putAlertState(ctx, s, metrics.AlertKey(box), body); err != nil { |
| + publishErr = err |
| } |
| } |
| @@ -243,7 +322,7 @@ func alertTick(ctx context.Context, s release.Store, box string, rules metrics.R |
| } |
| *st = newState |
| - return nil |
| + return publishErr |
| } |
| // persistAlertState writes st to path as indented JSON via the repo's |
diff --git a/internal/host/metrics_loop_test.go b/internal/host/metrics_loop_test.go| index 741c6f146..faa1de618 100644 |
| --- a/internal/host/metrics_loop_test.go |
| +++ b/internal/host/metrics_loop_test.go |
| @@ -55,14 +55,20 @@ func TestSamplerTickAndFlush(t *testing.T) { |
| t.Fatalf("unit CPU=%v, want 0.05 (3s of CPU over 60s)", f) |
| } |
| - // Flush within the hour: one raw chunk, no rollup yet. |
| + // This flush call's buf already spans the 10->11 hour boundary (one |
| + // sample in each hour): hour 10 is the "old" hour relative to `now` and |
| + // is written to its own chunk AND rolled up into the day rollup right |
| + // here, on this first flush — not deferred to the next one. Only hour |
| + // 11's sample stays buffered. |
| if err := metricsFlush(ctx, store, "box-a", sm, at.Add(time.Minute)); err != nil { |
| t.Fatal(err) |
| } |
| if _, _, err := store.Get(ctx, "metrics/boxes/box-a/raw/2026-08-02-10.json"); err != nil { |
| t.Fatalf("raw chunk missing: %v", err) |
| } |
| - // Tick past the hour boundary, flush again: previous hour rolled up. |
| + // Both samples now buffered are in hour 11: this second flush has |
| + // nothing to roll over, and the day rollup checked below is simply what |
| + // the first flush already wrote. |
| sm.tick(probes(c1, usec1, at.Add(2*time.Minute))) // 11:01:30 |
| if err := metricsFlush(ctx, store, "box-a", sm, at.Add(2*time.Minute)); err != nil { |
| t.Fatal(err) |
| @@ -205,3 +211,173 @@ func TestAlertTickWebhookFailureDoesNotBlock(t *testing.T) { |
| t.Errorf("state must still be published: %v", err) |
| } |
| } |
| + |
| +// TestAlertTickPersistFailureDoesNotSuppressPublish pins the review fix: a |
| +// failed local state write (statePath's parent is a file, not a directory, |
| +// so os.MkdirAll fails deterministically) must not suppress the alert |
| +// itself — the store publish and the webhook post are the reliable |
| +// channels, and the local write is most likely to fail exactly when the |
| +// disk rule is the one firing (disk full). |
| +func TestAlertTickPersistFailureDoesNotSuppressPublish(t *testing.T) { |
| + ctx := context.Background() |
| + store := release.NewDirStore(t.TempDir()) |
| + box := "box-a" |
| + rules := metrics.DefaultRules() |
| + st := &metrics.AlertState{V: 1, Box: box, Rules: map[string]metrics.RuleState{}} |
| + |
| + // statePath's parent directory is actually a file: MkdirAll must fail. |
| + blocker := filepath.Join(t.TempDir(), "not-a-dir") |
| + if err := os.WriteFile(blocker, []byte("x"), 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + statePath := filepath.Join(blocker, "monitor", "alerts.json") |
| + |
| + disk := 0.95 |
| + latest := metrics.Sample{ |
| + At: time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC), |
| + Host: metrics.HostSample{DiskFrac: &disk}, |
| + } |
| + var posted bool |
| + post := func(url string, body []byte) error { |
| + posted = true |
| + return nil |
| + } |
| + |
| + if err := alertTick(ctx, store, box, rules, st, latest, statePath, "http://example.test/webhook", post); err != nil { |
| + t.Fatalf("alertTick must not fail just because the local state write failed: %v", err) |
| + } |
| + if !posted { |
| + t.Error("webhook must still be posted when only the local persist failed") |
| + } |
| + body, _, err := store.Get(ctx, metrics.AlertKey(box)) |
| + if err != nil { |
| + t.Fatalf("alert state must still be published to the store: %v", err) |
| + } |
| + var published metrics.AlertState |
| + if err := json.Unmarshal(body, &published); err != nil { |
| + t.Fatal(err) |
| + } |
| + if !published.Rules["disk"].Firing { |
| + t.Errorf("disk rule not firing in published state: %+v", published.Rules["disk"]) |
| + } |
| +} |
| + |
| +// failingPutStore wraps a release.Store and makes every Put fail — the |
| +// fixture for TestSamplerBufCappedUnderSustainedFlushFailure's "IAM allows |
| +// PutObject... no wait, denies it" scenario: reads and Deletes still work |
| +// (delegated via embedding), only Put is broken. |
| +type failingPutStore struct { |
| + release.Store |
| +} |
| + |
| +func (f *failingPutStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + return "", errors.New("put always fails (test)") |
| +} |
| + |
| +// TestSamplerBufCappedUnderSustainedFlushFailure pins the review fix: a |
| +// store whose Put always fails means metricsFlush never gets far enough to |
| +// trim sm.buf (it returns early on the first failing WriteRaw), so without |
| +// a cap, buf would grow one sample per tick forever. tick itself enforces |
| +// samplerMaxBuf regardless of flush's fate; this ticks well past the cap, |
| +// confirms it held, confirms a failing flush leaves buf untouched, and then |
| +// confirms a subsequent flush against a real, working store still produces |
| +// a coherent (single-hour) chunk from whatever survived in buf. |
| +func TestSamplerBufCappedUnderSustainedFlushFailure(t *testing.T) { |
| + ctx := context.Background() |
| + box := "box-a" |
| + realStore := release.NewDirStore(t.TempDir()) |
| + failStore := &failingPutStore{Store: realStore} |
| + |
| + sm := &sampler{} |
| + start := time.Date(2026, 8, 2, 0, 0, 0, 0, time.UTC) |
| + mem, disk := 0.5, 0.3 |
| + for i := 0; i < samplerMaxBuf+30; i++ { |
| + now := start.Add(time.Duration(i) * time.Minute) |
| + sm.tick(sampleProbes{ |
| + procStat: func() *metrics.CPUCounters { return nil }, |
| + loadavg: func() (*float64, *float64, *float64) { return nil, nil, nil }, |
| + memFrac: func() *float64 { return &mem }, |
| + diskFrac: func() *float64 { return &disk }, |
| + units: func() map[string]unitRaw { return nil }, |
| + now: func() time.Time { return now }, |
| + }) |
| + } |
| + if len(sm.buf) != samplerMaxBuf { |
| + t.Fatalf("buf=%d after %d ticks, want capped at %d", len(sm.buf), samplerMaxBuf+30, samplerMaxBuf) |
| + } |
| + |
| + lastAt := sm.buf[len(sm.buf)-1].At |
| + if err := metricsFlush(ctx, failStore, box, sm, lastAt); err == nil { |
| + t.Fatal("expected metricsFlush to fail against a Put-always-fails store") |
| + } |
| + if len(sm.buf) != samplerMaxBuf { |
| + 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) |
| + } |
| + |
| + // Recovery: a flush against a real, working store must still succeed |
| + // and write a coherent (single-hour) chunk from what's left in buf. |
| + if err := metricsFlush(ctx, realStore, box, sm, lastAt); err != nil { |
| + t.Fatalf("flush against a working store: %v", err) |
| + } |
| + body, _, err := realStore.Get(ctx, metrics.RawKey(box, metrics.HourOf(lastAt))) |
| + if err != nil { |
| + t.Fatalf("current-hour chunk missing after recovery flush: %v", err) |
| + } |
| + var chunk metrics.RawChunk |
| + if err := json.Unmarshal(body, &chunk); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(chunk.Samples) == 0 { |
| + t.Fatal("recovered chunk has no samples") |
| + } |
| + for _, samp := range chunk.Samples { |
| + if metrics.HourOf(samp.At) != metrics.HourOf(lastAt) { |
| + t.Errorf("chunk mixes in a sample from a different hour: %v", samp.At) |
| + } |
| + } |
| +} |
| + |
| +// TestMetricsFlushMergesExistingChunkAfterRestart pins the review fix: a |
| +// restart resets sm.buf to empty, but the still-open current hour's raw |
| +// chunk may already hold samples this same box wrote before the crash. |
| +// WriteRaw is a wholesale replace, so the first flush after a restart must |
| +// merge the pre-existing chunk's samples in rather than clobber them — that |
| +// loss would otherwise become permanent the moment the hour rolls up. |
| +func TestMetricsFlushMergesExistingChunkAfterRestart(t *testing.T) { |
| + ctx := context.Background() |
| + store := release.NewDirStore(t.TempDir()) |
| + box := "box-a" |
| + |
| + hourStart := time.Date(2026, 8, 2, 14, 0, 0, 0, time.UTC) |
| + preRestart := []metrics.Sample{ |
| + {At: hourStart.Add(1 * time.Minute)}, |
| + {At: hourStart.Add(2 * time.Minute)}, |
| + } |
| + if err := metrics.WriteRaw(ctx, store, metrics.RawChunk{ |
| + V: 1, Box: box, Hour: metrics.HourOf(hourStart), Samples: preRestart, |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + // A fresh sampler, as if the process just restarted, with one sample |
| + // collected after the restart — still within the same hour. |
| + sm := &sampler{buf: []metrics.Sample{ |
| + {At: hourStart.Add(3 * time.Minute)}, |
| + }} |
| + now := hourStart.Add(3 * time.Minute) |
| + if err := metricsFlush(ctx, store, box, sm, now); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + body, _, err := store.Get(ctx, metrics.RawKey(box, metrics.HourOf(hourStart))) |
| + if err != nil { |
| + t.Fatalf("chunk missing after merge flush: %v", err) |
| + } |
| + var chunk metrics.RawChunk |
| + if err := json.Unmarshal(body, &chunk); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(chunk.Samples) != 3 { |
| + t.Fatalf("chunk has %d samples, want 3 (2 pre-restart + 1 post-restart)", len(chunk.Samples)) |
| + } |
| +} |