From ac1c2349cf975fc832af115e3c3f9c125817d154 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 17 Aug 2026 17:48:50 +0000 Subject: [PATCH 1/4] Evict stale per-device history series in the collector diskHist, rxHist, and txHist previously grew without bound: fastTick created a ring buffer per mountpoint/interface on first sight but never removed one, so churning veth interfaces (container start/stop) or USB drives/autofs mounts that come and go left dead series behind forever, growing memory usage and the /api/v1/metrics/history payload. Evict a device's series once its newest sample falls outside the retained history window (FastInterval * HistoryCapacity), not merely for being absent from a single tick, since DiskCollector.Collect already skips mountpoints that transiently fail to stat and such a device should not lose its history over one missed tick. Closes #24 --- internal/collector/collector.go | 35 +++++++++++++++++++ internal/collector/collector_test.go | 48 +++++++++++++++++++++++++++ internal/collector/ringbuffer.go | 14 ++++++++ internal/collector/ringbuffer_test.go | 26 +++++++++++++++ 4 files changed, 123 insertions(+) diff --git a/internal/collector/collector.go b/internal/collector/collector.go index e990578..5c56000 100644 --- a/internal/collector/collector.go +++ b/internal/collector/collector.go @@ -330,6 +330,7 @@ func (c *Collector) fastTick(ctx context.Context) { c.memHist.Add(HistoryPoint{Timestamp: now, Value: mem.UsedPercent}) c.swapHist.Add(HistoryPoint{Timestamp: now, Value: swap.UsedPercent}) + diskKeys := make(map[string]struct{}, len(disks)) for _, d := range disks { rb, ok := c.diskHist[d.Mountpoint] if !ok { @@ -337,7 +338,9 @@ func (c *Collector) fastTick(ctx context.Context) { c.diskHist[d.Mountpoint] = rb } rb.Add(HistoryPoint{Timestamp: now, Value: d.UsedPercent}) + diskKeys[d.Mountpoint] = struct{}{} } + netKeys := make(map[string]struct{}, len(netIfaces)) for _, n := range netIfaces { rxRB, ok := c.rxHist[n.Name] if !ok { @@ -352,8 +355,22 @@ func (c *Collector) fastTick(ctx context.Context) { c.txHist[n.Name] = txRB } txRB.Add(HistoryPoint{Timestamp: now, Value: n.TxBytesPerSec}) + netKeys[n.Name] = struct{}{} } + // Drop per-device series for mountpoints/interfaces that have vanished + // (unplugged USB drive, torn-down veth interface, ...), or diskHist/ + // rxHist/txHist would otherwise grow without bound as devices churn. A + // device is only evicted once its *newest* sample falls outside the + // retained history window, not merely for being absent from this single + // tick: DiskCollector.Collect already skips mountpoints that fail to + // stat, so a device missing for one tick must keep its history rather + // than losing it immediately. + historyWindow := c.cfg.FastInterval * time.Duration(c.cfg.HistoryCapacity) + evictStaleSeries(c.diskHist, diskKeys, now, historyWindow) + evictStaleSeries(c.rxHist, netKeys, now, historyWindow) + evictStaleSeries(c.txHist, netKeys, now, historyWindow) + // Evaluate the freshly collected values against the alert thresholds. // The engine has its own lock and never calls back into the collector, // so doing this while c.mu is held cannot deadlock. Metrics whose @@ -394,6 +411,24 @@ func (c *Collector) Alerts() alert.Report { return c.alerts.Report() } +// evictStaleSeries deletes entries from hist whose key is not in current +// and whose newest sample is older than window. window <= 0 disables +// eviction (no history window configured to measure staleness against). +func evictStaleSeries(hist map[string]*RingBuffer[HistoryPoint], current map[string]struct{}, now time.Time, window time.Duration) { + if window <= 0 { + return + } + for key, rb := range hist { + if _, ok := current[key]; ok { + continue + } + newest, ok := rb.Newest() + if !ok || now.Sub(newest.Timestamp) > window { + delete(hist, key) + } + } +} + func (c *Collector) slowTick(ctx context.Context) { updates, err := c.updates.Collect(ctx) if err != nil { diff --git a/internal/collector/collector_test.go b/internal/collector/collector_test.go index f7e42b5..3ce585a 100644 --- a/internal/collector/collector_test.go +++ b/internal/collector/collector_test.go @@ -175,6 +175,54 @@ func TestCollector_Alerts_EvaluatedOnFastTick(t *testing.T) { } } +func TestCollector_History_EvictsStaleDeviceSeries(t *testing.T) { + c := New(Config{ + FastInterval: time.Second, + SlowInterval: time.Minute, + HistoryCapacity: 3, // history window = 3s + }, nil) + window := c.cfg.FastInterval * time.Duration(c.cfg.HistoryCapacity) + + base := time.Unix(1_700_000_000, 0) + + // tick simulates one fastTick's disk bookkeeping: add a sample for + // every key in present, then evict devices missing for longer than the + // history window. + tick := func(now time.Time, present map[string]struct{}) { + for key := range present { + rb, ok := c.diskHist[key] + if !ok { + rb = NewRingBuffer[HistoryPoint](c.cfg.HistoryCapacity) + c.diskHist[key] = rb + } + rb.Add(HistoryPoint{Timestamp: now, Value: 1}) + } + evictStaleSeries(c.diskHist, present, now, window) + } + + // Both devices present for the first tick. + tick(base, map[string]struct{}{"/mnt/usb": {}, "/mnt/flaky": {}}) + + // /mnt/flaky misses a single tick (e.g. a transient stat error) one + // second later; it must keep its history. + tick(base.Add(time.Second), map[string]struct{}{"/mnt/usb": {}}) + if _, ok := c.History().DiskUsedPercent["/mnt/flaky"]; !ok { + t.Fatal("expected device missing for a single tick to keep its history") + } + + // Time advances well past the history window with /mnt/flaky still + // absent (e.g. the USB drive was unplugged for good). + tick(base.Add(window+time.Second), map[string]struct{}{"/mnt/usb": {}}) + + hist := c.History() + if _, ok := hist.DiskUsedPercent["/mnt/flaky"]; ok { + t.Fatal("expected stale device series to be evicted once its newest sample exceeds the history window") + } + if _, ok := hist.DiskUsedPercent["/mnt/usb"]; !ok { + t.Fatal("expected continuously present device to be unaffected by eviction") + } +} + func TestCollector_Run_StopsOnContextCancel(t *testing.T) { c := New(Config{ FastInterval: 10 * time.Millisecond, diff --git a/internal/collector/ringbuffer.go b/internal/collector/ringbuffer.go index b6665f5..0fb6acd 100644 --- a/internal/collector/ringbuffer.go +++ b/internal/collector/ringbuffer.go @@ -68,3 +68,17 @@ func (r *RingBuffer[T]) Snapshot() []T { copy(out[n:], r.data[:r.next]) return out } + +// Newest returns the most recently added value and true, or the zero value +// and false if the buffer is empty. +func (r *RingBuffer[T]) Newest() (T, bool) { + r.mu.Lock() + defer r.mu.Unlock() + + if r.size == 0 { + var zero T + return zero, false + } + idx := (r.next - 1 + r.capacity) % r.capacity + return r.data[idx], true +} diff --git a/internal/collector/ringbuffer_test.go b/internal/collector/ringbuffer_test.go index f58c61c..d6514c9 100644 --- a/internal/collector/ringbuffer_test.go +++ b/internal/collector/ringbuffer_test.go @@ -118,3 +118,29 @@ func TestRingBuffer_Empty(t *testing.T) { t.Fatalf("expected empty snapshot, got %v", got) } } + +func TestRingBuffer_Newest_EmptyReturnsFalse(t *testing.T) { + rb := NewRingBuffer[int](3) + if _, ok := rb.Newest(); ok { + t.Fatal("expected Newest to report false on an empty buffer") + } +} + +func TestRingBuffer_Newest_ReturnsLastAdded(t *testing.T) { + rb := NewRingBuffer[int](3) + rb.Add(1) + rb.Add(2) + if got, ok := rb.Newest(); !ok || got != 2 { + t.Fatalf("Newest = (%v, %v), want (2, true)", got, ok) + } +} + +func TestRingBuffer_Newest_AfterWraparound(t *testing.T) { + rb := NewRingBuffer[int](3) + for i := 1; i <= 5; i++ { + rb.Add(i) + } + if got, ok := rb.Newest(); !ok || got != 5 { + t.Fatalf("Newest after wraparound = (%v, %v), want (5, true)", got, ok) + } +} From e5ee2a3a4672140abf61de842dac13cb1110b3a9 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 17 Aug 2026 18:37:58 +0000 Subject: [PATCH 2/4] Address review: test through fastTick, clamp eviction window, document rule - Rewrite the eviction test to drive fastTick itself instead of reimplementing its per-device bookkeeping around evictStaleSeries directly, so the real call sites (including the previously-uncovered rxHist/txHist eviction) are under test. Add a table-driven test for evictStaleSeries covering the present/within-window/at-boundary/ past-boundary cases and the zero-window no-op. - Derive the eviction window from a FastInterval clamped once at construction (Collector.fastInterval) instead of the raw config field, so a non-positive FastInterval (already defended against by Run's clampInterval for the ticker) can't silently disable eviction too. Added a regression test. - Document the eviction rule in ARCHITECTURE.md, including why it deliberately differs from the alert engine's pruneDisks (prune on first absence there vs. a full history window here). --- docs/ARCHITECTURE.md | 12 ++ internal/collector/collector.go | 19 +++- internal/collector/collector_test.go | 159 +++++++++++++++++++++------ 3 files changed, 153 insertions(+), 37 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 40b1675..2f84096 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -92,6 +92,18 @@ the first time a given device is seen. `RingBuffer[T]` (`ringbuffer.go`) is a fi capacity circular buffer with its own internal lock — safe to read concurrently with the collector's own tick, independent of the collector's outer `sync.RWMutex`. +Entries are also removed: `evictStaleSeries`, called from `fastTick` for `diskHist`, +`rxHist`, and `txHist`, deletes a device's series once its *newest* sample is older than +the retained history window (`FastInterval * HistoryCapacity`) — without this, a churning +`veth*` interface from container start/stop or a USB drive that gets unplugged would keep +its full ring buffer (and keep showing up in `GET /api/v1/metrics/history`) forever. This +deliberately differs from the alert engine's `pruneDisks` (see below), which drops state +on a single absent sample: `DiskCollector.Collect` already skips a mountpoint that +transiently fails to stat, so evicting history on the first miss would wipe a device's +history over a passing blip rather than an actual disappearance — eviction is tied to the +history window instead, so a one-tick miss is tolerated and only a sustained absence is +treated as gone. + `HistoryCapacity` (`config.Config.HistoryCapacity()`) is derived from `history_window_minutes / poll_interval_seconds`, not configured directly, and is capped at 1,000,000 points per series (`config.maxHistoryCapacity`) — `NewRingBuffer` allocates diff --git a/internal/collector/collector.go b/internal/collector/collector.go index 5c56000..e360372 100644 --- a/internal/collector/collector.go +++ b/internal/collector/collector.go @@ -97,6 +97,13 @@ type Collector struct { log *slog.Logger + // fastInterval is cfg.FastInterval clamped to a safe positive minimum + // (see clampInterval), computed once at construction so Run's ticker + // and fastTick's history-eviction window always agree, regardless of + // whether Run has been started yet (e.g. in tests calling fastTick + // directly). + fastInterval time.Duration + mu sync.RWMutex latest Snapshot cpuHist *RingBuffer[HistoryPoint] @@ -120,7 +127,7 @@ func New(cfg Config, log *slog.Logger) *Collector { if cfg.AlertsEnabled { alerts = alert.New(cfg.Thresholds, cfg.AlertFor) } - return &Collector{ + c := &Collector{ cfg: cfg, alerts: alerts, notifier: cfg.Notifier, @@ -147,6 +154,8 @@ func New(cfg Config, log *slog.Logger) *Collector { rxHist: make(map[string]*RingBuffer[HistoryPoint]), txHist: make(map[string]*RingBuffer[HistoryPoint]), } + c.fastInterval = c.clampInterval(cfg.FastInterval, "FastInterval") + return c } // Run collects an initial sample immediately, then continues sampling on @@ -166,11 +175,11 @@ func (c *Collector) Run(ctx context.Context) { // Defense in depth: a non-positive interval panics time.NewTicker. // config.Validate rejects such values at startup, but clamp here too so - // no future caller can crash the collector. - fastInterval := c.clampInterval(c.cfg.FastInterval, "FastInterval") + // no future caller can crash the collector. c.fastInterval was already + // clamped in New; only SlowInterval needs it here. slowInterval := c.clampInterval(c.cfg.SlowInterval, "SlowInterval") - fastTicker := time.NewTicker(fastInterval) + fastTicker := time.NewTicker(c.fastInterval) defer fastTicker.Stop() slowTicker := time.NewTicker(slowInterval) defer slowTicker.Stop() @@ -366,7 +375,7 @@ func (c *Collector) fastTick(ctx context.Context) { // tick: DiskCollector.Collect already skips mountpoints that fail to // stat, so a device missing for one tick must keep its history rather // than losing it immediately. - historyWindow := c.cfg.FastInterval * time.Duration(c.cfg.HistoryCapacity) + historyWindow := c.fastInterval * time.Duration(c.cfg.HistoryCapacity) evictStaleSeries(c.diskHist, diskKeys, now, historyWindow) evictStaleSeries(c.rxHist, netKeys, now, historyWindow) evictStaleSeries(c.txHist, netKeys, now, historyWindow) diff --git a/internal/collector/collector_test.go b/internal/collector/collector_test.go index 3ce585a..7185464 100644 --- a/internal/collector/collector_test.go +++ b/internal/collector/collector_test.go @@ -175,51 +175,146 @@ func TestCollector_Alerts_EvaluatedOnFastTick(t *testing.T) { } } -func TestCollector_History_EvictsStaleDeviceSeries(t *testing.T) { +func TestEvictStaleSeries(t *testing.T) { + const window = 3 * time.Second + now := time.Unix(1_700_000_000, 0) + + tests := []struct { + name string + present bool + age time.Duration + wantEvicted bool + }{ + {"present device is kept regardless of sample age", true, 10 * window, false}, + {"absent device within the window is kept", false, window - time.Second, false}, + {"absent device exactly at the window boundary is kept", false, window, false}, + {"absent device past the window boundary is evicted", false, window + time.Nanosecond, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + rb := NewRingBuffer[HistoryPoint](3) + rb.Add(HistoryPoint{Timestamp: now.Add(-tt.age), Value: 1}) + hist := map[string]*RingBuffer[HistoryPoint]{"dev": rb} + current := map[string]struct{}{} + if tt.present { + current["dev"] = struct{}{} + } + + evictStaleSeries(hist, current, now, window) + + _, ok := hist["dev"] + if ok == tt.wantEvicted { + t.Fatalf("evictStaleSeries: entry present = %v, want evicted = %v", ok, tt.wantEvicted) + } + }) + } +} + +func TestEvictStaleSeries_ZeroWindowDisablesEviction(t *testing.T) { + rb := NewRingBuffer[HistoryPoint](3) + rb.Add(HistoryPoint{Timestamp: time.Unix(0, 0), Value: 1}) + hist := map[string]*RingBuffer[HistoryPoint]{"dev": rb} + + evictStaleSeries(hist, map[string]struct{}{}, time.Now(), 0) + + if _, ok := hist["dev"]; !ok { + t.Fatal("expected a non-positive window to disable eviction") + } +} + +// TestCollector_FastTick_EvictsStaleDeviceSeries drives eviction through the +// production code path (fastTick), rather than calling evictStaleSeries +// directly, so the wiring in fastTick itself — not just the helper — is +// under test. +func TestCollector_FastTick_EvictsStaleDeviceSeries(t *testing.T) { c := New(Config{ FastInterval: time.Second, SlowInterval: time.Minute, HistoryCapacity: 3, // history window = 3s + NetworkEnabled: true, }, nil) - window := c.cfg.FastInterval * time.Duration(c.cfg.HistoryCapacity) - - base := time.Unix(1_700_000_000, 0) - - // tick simulates one fastTick's disk bookkeeping: add a sample for - // every key in present, then evict devices missing for longer than the - // history window. - tick := func(now time.Time, present map[string]struct{}) { - for key := range present { - rb, ok := c.diskHist[key] - if !ok { - rb = NewRingBuffer[HistoryPoint](c.cfg.HistoryCapacity) - c.diskHist[key] = rb - } - rb.Add(HistoryPoint{Timestamp: now, Value: 1}) - } - evictStaleSeries(c.diskHist, present, now, window) + + // A mountpoint/interface name the host running this test can never + // actually report, seeded with a sample far outside the history window. + const goneDevice = "pimonitor-test-gone" + stale := HistoryPoint{Timestamp: time.Now().Add(-time.Hour), Value: 1} + for _, m := range []map[string]*RingBuffer[HistoryPoint]{c.diskHist, c.rxHist, c.txHist} { + rb := NewRingBuffer[HistoryPoint](c.cfg.HistoryCapacity) + rb.Add(stale) + m[goneDevice] = rb } - // Both devices present for the first tick. - tick(base, map[string]struct{}{"/mnt/usb": {}, "/mnt/flaky": {}}) + c.fastTick(context.Background()) - // /mnt/flaky misses a single tick (e.g. a transient stat error) one - // second later; it must keep its history. - tick(base.Add(time.Second), map[string]struct{}{"/mnt/usb": {}}) - if _, ok := c.History().DiskUsedPercent["/mnt/flaky"]; !ok { - t.Fatal("expected device missing for a single tick to keep its history") + hist := c.History() + if _, ok := hist.DiskUsedPercent[goneDevice]; ok { + t.Fatal("expected fastTick to evict the stale disk series") + } + if _, ok := hist.NetworkRxBytesPerSec[goneDevice]; ok { + t.Fatal("expected fastTick to evict the stale rx series") } + if _, ok := hist.NetworkTxBytesPerSec[goneDevice]; ok { + t.Fatal("expected fastTick to evict the stale tx series") + } +} - // Time advances well past the history window with /mnt/flaky still - // absent (e.g. the USB drive was unplugged for good). - tick(base.Add(window+time.Second), map[string]struct{}{"/mnt/usb": {}}) +// TestCollector_FastTick_KeepsRecentlyMissingDeviceSeries checks the +// counterpart of the eviction rule through the same fastTick path: a device +// missing for a single tick must not lose its history, matching +// DiskCollector.Collect skipping mountpoints that transiently fail to stat. +func TestCollector_FastTick_KeepsRecentlyMissingDeviceSeries(t *testing.T) { + c := New(Config{ + FastInterval: time.Second, + SlowInterval: time.Minute, + HistoryCapacity: 3, // history window = 3s + NetworkEnabled: true, + }, nil) + + const flakyDevice = "pimonitor-test-flaky" + recent := HistoryPoint{Timestamp: time.Now(), Value: 1} + for _, m := range []map[string]*RingBuffer[HistoryPoint]{c.diskHist, c.rxHist, c.txHist} { + rb := NewRingBuffer[HistoryPoint](c.cfg.HistoryCapacity) + rb.Add(recent) + m[flakyDevice] = rb + } + + c.fastTick(context.Background()) hist := c.History() - if _, ok := hist.DiskUsedPercent["/mnt/flaky"]; ok { - t.Fatal("expected stale device series to be evicted once its newest sample exceeds the history window") + if _, ok := hist.DiskUsedPercent[flakyDevice]; !ok { + t.Fatal("expected fastTick to keep a disk series missing for only one tick") + } + if _, ok := hist.NetworkRxBytesPerSec[flakyDevice]; !ok { + t.Fatal("expected fastTick to keep an rx series missing for only one tick") } - if _, ok := hist.DiskUsedPercent["/mnt/usb"]; !ok { - t.Fatal("expected continuously present device to be unaffected by eviction") + if _, ok := hist.NetworkTxBytesPerSec[flakyDevice]; !ok { + t.Fatal("expected fastTick to keep a tx series missing for only one tick") + } +} + +// TestCollector_FastTick_EvictsWithNonPositiveFastInterval guards against +// the eviction window silently collapsing to zero (which disables eviction, +// see evictStaleSeries) for a Collector constructed with a non-positive +// FastInterval. Run's ticker already defends against this case via +// clampInterval; fastTick's history window must derive from the same +// clamped value rather than the raw, unclamped config field. +func TestCollector_FastTick_EvictsWithNonPositiveFastInterval(t *testing.T) { + c := New(Config{ + FastInterval: 0, // invalid; clamped to 1s + SlowInterval: time.Minute, + HistoryCapacity: 3, // clamped history window = 3s + }, nil) + + const goneDevice = "pimonitor-test-gone" + stale := HistoryPoint{Timestamp: time.Now().Add(-time.Hour), Value: 1} + rb := NewRingBuffer[HistoryPoint](c.cfg.HistoryCapacity) + rb.Add(stale) + c.diskHist[goneDevice] = rb + + c.fastTick(context.Background()) + + if _, ok := c.History().DiskUsedPercent[goneDevice]; ok { + t.Fatal("expected eviction to still apply when FastInterval is non-positive") } } From e3017294862d68f41191972991231eb7ae6d6894 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 17 Aug 2026 18:53:21 +0000 Subject: [PATCH 3/4] Address review: fix flaky test timing, reduce fastTick complexity - TestCollector_FastTick_KeepsRecentlyMissingDeviceSeries stamped its seeded sample with time.Now() and compared against a 3s window, but fastTick's own now is taken after collection that can legitimately run for several seconds on real Pi hardware (vcgencmd calls, statfsWithTimeout per mountpoint). Widen the window to 1h so the test no longer depends on tick duration. - SonarCloud flagged fastTick's cognitive complexity (25, limit 15). Extract the collect-and-log-errors prologue into collectFastTickSamples (returning a fastTickSamples struct) and the per-device history/eviction bookkeeping into recordDeviceHistory, leaving fastTick as orchestration only. No behavior change. --- internal/collector/collector.go | 198 ++++++++++++++++----------- internal/collector/collector_test.go | 2 +- 2 files changed, 119 insertions(+), 81 deletions(-) diff --git a/internal/collector/collector.go b/internal/collector/collector.go index e360372..bfc52ee 100644 --- a/internal/collector/collector.go +++ b/internal/collector/collector.go @@ -272,73 +272,149 @@ func (c *Collector) collectSysInfo() { c.latest.CPUCount = count } -func (c *Collector) fastTick(ctx context.Context) { - now := time.Now() +// fastTickSamples holds every metric fastTick reads before taking c.mu, +// plus each source's error, so history and alerts can tell a real zero +// value apart from a failed collection. +type fastTickSamples struct { + now time.Time + cpuUsage CPUUsage + cpuErr error + cpuFreq []CPUCoreFrequency + load LoadAverage + temp Temperature + gpuTemp *GPUTemperature + tempErr error + throttled *Throttled + mem Memory + swap Swap + memErr error + disks []Disk + diskErr error + netIfaces []NetworkInterface + uptimeSecs float64 +} - cpuUsage, cpuErr := c.cpu.Collect() - if cpuErr != nil { - c.log.Warn("cpu collection failed", "error", cpuErr) +// collectFastTickSamples gathers every metric source fastTick needs. Each +// source's error is logged here (but never aborts the tick) so a single +// failing collector (e.g. no thermal zone on non-Pi hardware) leaves just +// that field at its zero value rather than blocking the others. +func (c *Collector) collectFastTickSamples(ctx context.Context) fastTickSamples { + var s fastTickSamples + s.now = time.Now() + + s.cpuUsage, s.cpuErr = c.cpu.Collect() + if s.cpuErr != nil { + c.log.Warn("cpu collection failed", "error", s.cpuErr) } - cpuFreq, cpuFreqErr := c.cpuFreq.Collect() + var cpuFreqErr error + s.cpuFreq, cpuFreqErr = c.cpuFreq.Collect() if cpuFreqErr != nil { c.log.Warn("cpu frequency collection failed", "error", cpuFreqErr) } - load, err := c.loadAvg.Collect() + var err error + s.load, err = c.loadAvg.Collect() if err != nil { c.log.Warn("load average collection failed", "error", err) } - temp, gpuTemp, tempErr := c.temp.Collect(ctx) - if tempErr != nil { - c.log.Warn("temperature collection failed", "error", tempErr) + s.temp, s.gpuTemp, s.tempErr = c.temp.Collect(ctx) + if s.tempErr != nil { + c.log.Warn("temperature collection failed", "error", s.tempErr) } - throttled, throttledErr := c.throttled.Collect(ctx) - if throttledErr != nil { - c.log.Warn("throttled state collection failed", "error", throttledErr) + s.throttled, err = c.throttled.Collect(ctx) + if err != nil { + c.log.Warn("throttled state collection failed", "error", err) } - mem, swap, memErr := c.memory.Collect() - if memErr != nil { - c.log.Warn("memory collection failed", "error", memErr) + s.mem, s.swap, s.memErr = c.memory.Collect() + if s.memErr != nil { + c.log.Warn("memory collection failed", "error", s.memErr) } - disks, diskErr := c.disk.Collect() - if diskErr != nil { - c.log.Warn("disk collection failed", "error", diskErr) + s.disks, s.diskErr = c.disk.Collect() + if s.diskErr != nil { + c.log.Warn("disk collection failed", "error", s.diskErr) } - var netIfaces []NetworkInterface if c.cfg.NetworkEnabled { - netIfaces, err = c.network.Collect() + s.netIfaces, err = c.network.Collect() if err != nil { c.log.Warn("network collection failed", "error", err) } } - uptimeSecs, err := c.uptime.Collect() + s.uptimeSecs, err = c.uptime.Collect() if err != nil { c.log.Warn("uptime collection failed", "error", err) } + return s +} + +func (c *Collector) fastTick(ctx context.Context) { + s := c.collectFastTickSamples(ctx) c.mu.Lock() defer c.mu.Unlock() - c.latest.Timestamp = now - c.latest.UptimeSeconds = uptimeSecs - c.latest.CPU = cpuUsage - c.latest.CPUFrequency = cpuFreq - c.latest.Load = load - c.latest.Temperature = temp - c.latest.GPUTemperature = gpuTemp - c.latest.Throttled = throttled - c.latest.Memory = mem - c.latest.Swap = swap - c.latest.Disks = disks - c.latest.Network = netIfaces - - c.cpuHist.Add(HistoryPoint{Timestamp: now, Value: cpuUsage.OverallPercent}) - c.l1Hist.Add(HistoryPoint{Timestamp: now, Value: load.Load1}) - c.l5Hist.Add(HistoryPoint{Timestamp: now, Value: load.Load5}) - c.l15Hist.Add(HistoryPoint{Timestamp: now, Value: load.Load15}) - c.tempHist.Add(HistoryPoint{Timestamp: now, Value: temp.Celsius}) - c.memHist.Add(HistoryPoint{Timestamp: now, Value: mem.UsedPercent}) - c.swapHist.Add(HistoryPoint{Timestamp: now, Value: swap.UsedPercent}) + c.latest.Timestamp = s.now + c.latest.UptimeSeconds = s.uptimeSecs + c.latest.CPU = s.cpuUsage + c.latest.CPUFrequency = s.cpuFreq + c.latest.Load = s.load + c.latest.Temperature = s.temp + c.latest.GPUTemperature = s.gpuTemp + c.latest.Throttled = s.throttled + c.latest.Memory = s.mem + c.latest.Swap = s.swap + c.latest.Disks = s.disks + c.latest.Network = s.netIfaces + + c.cpuHist.Add(HistoryPoint{Timestamp: s.now, Value: s.cpuUsage.OverallPercent}) + c.l1Hist.Add(HistoryPoint{Timestamp: s.now, Value: s.load.Load1}) + c.l5Hist.Add(HistoryPoint{Timestamp: s.now, Value: s.load.Load5}) + c.l15Hist.Add(HistoryPoint{Timestamp: s.now, Value: s.load.Load15}) + c.tempHist.Add(HistoryPoint{Timestamp: s.now, Value: s.temp.Celsius}) + c.memHist.Add(HistoryPoint{Timestamp: s.now, Value: s.mem.UsedPercent}) + c.swapHist.Add(HistoryPoint{Timestamp: s.now, Value: s.swap.UsedPercent}) + + c.recordDeviceHistory(s.now, s.disks, s.netIfaces) + // Evaluate the freshly collected values against the alert thresholds. + // The engine has its own lock and never calls back into the collector, + // so doing this while c.mu is held cannot deadlock. Metrics whose + // collection failed this tick are flagged invalid so a bogus zero can't + // spuriously clear a real alert; the engine keeps their previous state. + if c.alerts != nil { + diskSamples := make([]alert.DiskSample, len(s.disks)) + for i, d := range s.disks { + diskSamples[i] = alert.DiskSample{Mountpoint: d.Mountpoint, UsedPercent: d.UsedPercent} + } + events := c.alerts.Evaluate(alert.Sample{ + Timestamp: s.now, + CPUPercent: s.cpuUsage.OverallPercent, + CPUValid: s.cpuErr == nil, + TemperatureC: s.temp.Celsius, + TemperatureValid: s.tempErr == nil, + MemoryPercent: s.mem.UsedPercent, + MemoryValid: s.memErr == nil, + SwapPercent: s.swap.UsedPercent, + SwapValid: s.memErr == nil, + Disks: diskSamples, + DisksValid: s.diskErr == nil, + }) + // Forward any transition events to the webhook notifier. Notify only + // enqueues (never blocks), so a slow webhook can't stall collection. + if c.notifier != nil && len(events) > 0 { + c.notifier.Notify(events) + } + } +} + +// recordDeviceHistory adds this tick's samples to the per-device history +// maps (diskHist, rxHist, txHist) and evicts entries for devices that have +// vanished (unplugged USB drive, torn-down veth interface, ...), or those +// maps would otherwise grow without bound as devices churn. A device is +// only evicted once its *newest* sample falls outside the retained history +// window, not merely for being absent from this single tick: +// DiskCollector.Collect already skips mountpoints that fail to stat, so a +// device missing for one tick must keep its history rather than losing it +// immediately. Called from fastTick while c.mu is already held. +func (c *Collector) recordDeviceHistory(now time.Time, disks []Disk, netIfaces []NetworkInterface) { diskKeys := make(map[string]struct{}, len(disks)) for _, d := range disks { rb, ok := c.diskHist[d.Mountpoint] @@ -367,48 +443,10 @@ func (c *Collector) fastTick(ctx context.Context) { netKeys[n.Name] = struct{}{} } - // Drop per-device series for mountpoints/interfaces that have vanished - // (unplugged USB drive, torn-down veth interface, ...), or diskHist/ - // rxHist/txHist would otherwise grow without bound as devices churn. A - // device is only evicted once its *newest* sample falls outside the - // retained history window, not merely for being absent from this single - // tick: DiskCollector.Collect already skips mountpoints that fail to - // stat, so a device missing for one tick must keep its history rather - // than losing it immediately. historyWindow := c.fastInterval * time.Duration(c.cfg.HistoryCapacity) evictStaleSeries(c.diskHist, diskKeys, now, historyWindow) evictStaleSeries(c.rxHist, netKeys, now, historyWindow) evictStaleSeries(c.txHist, netKeys, now, historyWindow) - - // Evaluate the freshly collected values against the alert thresholds. - // The engine has its own lock and never calls back into the collector, - // so doing this while c.mu is held cannot deadlock. Metrics whose - // collection failed this tick are flagged invalid so a bogus zero can't - // spuriously clear a real alert; the engine keeps their previous state. - if c.alerts != nil { - diskSamples := make([]alert.DiskSample, len(disks)) - for i, d := range disks { - diskSamples[i] = alert.DiskSample{Mountpoint: d.Mountpoint, UsedPercent: d.UsedPercent} - } - events := c.alerts.Evaluate(alert.Sample{ - Timestamp: now, - CPUPercent: cpuUsage.OverallPercent, - CPUValid: cpuErr == nil, - TemperatureC: temp.Celsius, - TemperatureValid: tempErr == nil, - MemoryPercent: mem.UsedPercent, - MemoryValid: memErr == nil, - SwapPercent: swap.UsedPercent, - SwapValid: memErr == nil, - Disks: diskSamples, - DisksValid: diskErr == nil, - }) - // Forward any transition events to the webhook notifier. Notify only - // enqueues (never blocks), so a slow webhook can't stall collection. - if c.notifier != nil && len(events) > 0 { - c.notifier.Notify(events) - } - } } // Alerts returns the current alert states and recent transition events. When diff --git a/internal/collector/collector_test.go b/internal/collector/collector_test.go index 7185464..decb192 100644 --- a/internal/collector/collector_test.go +++ b/internal/collector/collector_test.go @@ -266,7 +266,7 @@ func TestCollector_FastTick_KeepsRecentlyMissingDeviceSeries(t *testing.T) { c := New(Config{ FastInterval: time.Second, SlowInterval: time.Minute, - HistoryCapacity: 3, // history window = 3s + HistoryCapacity: 3600, // history window = 1h, far beyond any single tick's duration NetworkEnabled: true, }, nil) From 8a11900e2ffb78d742010231509f69a6eb4d90e0 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 17 Aug 2026 19:03:51 +0000 Subject: [PATCH 4/4] Restore 3s history window in KeepsRecentlyMissingDeviceSeries test The window had been widened to 1h on the premise that the test's margin must cover the tick's duration. It does not: fastTick captures its timestamp before any collection runs (s.now is the first statement of collectFastTickSamples), so the vcgencmd and statfs timeouts all elapse after that timestamp is taken and the gap between the test's seeded sample and fastTick's now is microseconds regardless of tick duration. Revert to the 3s window, which states the intent more directly and drops a comment that asserted a property of fastTick that is not true. --- internal/collector/collector_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/collector/collector_test.go b/internal/collector/collector_test.go index decb192..7185464 100644 --- a/internal/collector/collector_test.go +++ b/internal/collector/collector_test.go @@ -266,7 +266,7 @@ func TestCollector_FastTick_KeepsRecentlyMissingDeviceSeries(t *testing.T) { c := New(Config{ FastInterval: time.Second, SlowInterval: time.Minute, - HistoryCapacity: 3600, // history window = 1h, far beyond any single tick's duration + HistoryCapacity: 3, // history window = 3s NetworkEnabled: true, }, nil)