Evict stale per-device history series in the collector

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

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,81 +272,160 @@ 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 {
 			rb = NewRingBuffer[HistoryPoint](c.cfg.HistoryCapacity)
 			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 {