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 e990578..bfc52ee 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() @@ -263,73 +272,150 @@ 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 +} + +// 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() - cpuUsage, cpuErr := c.cpu.Collect() - if cpuErr != nil { - c.log.Warn("cpu collection failed", "error", cpuErr) + 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] if !ok { @@ -337,7 +423,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,37 +440,13 @@ func (c *Collector) fastTick(ctx context.Context) { c.txHist[n.Name] = txRB } txRB.Add(HistoryPoint{Timestamp: now, Value: n.TxBytesPerSec}) + netKeys[n.Name] = struct{}{} } - // 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) - } - } + 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) } // Alerts returns the current alert states and recent transition events. When @@ -394,6 +458,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..7185464 100644 --- a/internal/collector/collector_test.go +++ b/internal/collector/collector_test.go @@ -175,6 +175,149 @@ func TestCollector_Alerts_EvaluatedOnFastTick(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) + + // 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 + } + + c.fastTick(context.Background()) + + 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") + } +} + +// 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[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.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") + } +} + 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) + } +}