"""Background sampler, history ring buffers and SSE fan-out.""" from __future__ import annotations import json import logging import queue import threading import time from collections import deque from typing import Any, Dict, List, Optional from .metrics import SystemCollector logger = logging.getLogger("servermon.service") HISTORY_WINDOWS = ("cpu", "mem", "load", "rx", "tx", "disk_r", "disk_w") class MonitorService: """Samples the metrics source on a fixed cadence and serves immutable JSON. All HTTP/SSE readers share the latest serialized snapshot, so displaying the dashboard from several screens only costs one sampler. """ def __init__( self, source: Any, interval: float = 1.0, history_limit: int = 120, display_points: int = 60, ) -> None: self.source = source self.interval = max(0.2, float(interval)) self.history_limit = max(10, int(history_limit)) self.display_points = max(10, min(display_points, self.history_limit)) self._lock = threading.RLock() self._stop = threading.Event() self._thread: Optional[threading.Thread] = None self._current: Optional[Dict[str, Any]] = None self._json = "{}" self._history: Dict[str, deque] = { key: deque(maxlen=self.history_limit) for key in HISTORY_WINDOWS } self._subscribers: set[queue.Queue] = set() self._samples = 0 self._last_sample_ts: Optional[float] = None self._last_error: Optional[str] = None self._last_error_log = 0.0 self.started_at = time.time() self._bootstrapped = False # ------------------------------------------------------------------ # Lifecycle # ------------------------------------------------------------------ def bootstrap(self) -> None: """Populate the first snapshot (and demo history) before serving HTTP.""" if self._bootstrapped: return is_demo = self.source.__class__.__name__ == "DemoSource" if is_demo: # A full trend makes the panel useful immediately in demo mode. for _ in range(self.display_points): self._record(self.source.snapshot(), broadcast=False) else: self._record(self.source.snapshot(), broadcast=False) self._bootstrapped = True def start(self) -> None: if self._thread and self._thread.is_alive(): return if not self._bootstrapped: self.bootstrap() self._stop.clear() self._thread = threading.Thread(target=self._run, name="sampler", daemon=True) self._thread.start() def stop(self) -> None: self._stop.set() if self._thread: self._thread.join(timeout=3.0) def _run(self) -> None: next_run = time.monotonic() while not self._stop.is_set(): next_run += self.interval delay = next_run - time.monotonic() if delay > 0: self._stop.wait(delay) else: # The machine slept or sampling was slow; do not burst to catch up. next_run = time.monotonic() if self._stop.is_set(): break try: self.collect_once() except Exception as exc: # pragma: no cover - defensive self._last_error = f"{type(exc).__name__}: {exc}" now = time.monotonic() if now - self._last_error_log > 30.0: logger.exception("metrics sample failed") self._last_error_log = now def collect_once(self) -> Dict[str, Any]: snapshot = self.source.snapshot() self._record(snapshot, broadcast=True) return snapshot # ------------------------------------------------------------------ # Snapshot storage # ------------------------------------------------------------------ def _record(self, snapshot: Dict[str, Any], broadcast: bool) -> None: cpu = snapshot.get("cpu") or {} memory = snapshot.get("memory") or {} load = snapshot.get("load") or {} network = snapshot.get("network") or {} disk_io = snapshot.get("disk_io") or {} values = { "cpu": float(cpu.get("percent") or 0.0), "mem": float(memory.get("percent") or 0.0), "load": float(load.get("load1") or 0.0), "rx": float(network.get("total_rx_bps") or 0.0), "tx": float(network.get("total_tx_bps") or 0.0), "disk_r": float(disk_io.get("read_bps") or 0.0), "disk_w": float(disk_io.get("write_bps") or 0.0), } with self._lock: for key, value in values.items(): self._history[key].append(round(value, 1) if key != "load" else round(value, 2)) history = { key: [round(v, 1) if key != "load" else round(v, 2) for v in list(values_deque)[-self.display_points:]] for key, values_deque in self._history.items() } payload = dict(snapshot) payload["history"] = history payload["meta"] = { "sample_interval": self.interval, "history_points": self.display_points, "history_seconds": int(self.display_points * self.interval), "demo": bool((snapshot.get("host") or {}).get("demo")), } encoded = json.dumps(payload, ensure_ascii=False, separators=(",", ":")) self._current = payload self._json = encoded self._samples += 1 self._last_sample_ts = time.time() subscribers = list(self._subscribers) if broadcast: for subscriber in subscribers: try: if subscriber.full(): try: subscriber.get_nowait() except queue.Empty: pass subscriber.put_nowait(encoded) except (queue.Full, ValueError): pass def snapshot(self) -> Optional[Dict[str, Any]]: with self._lock: return self._current def snapshot_json(self) -> str: with self._lock: return self._json def wait_ready(self, timeout: float = 5.0) -> bool: deadline = time.monotonic() + timeout while time.monotonic() < deadline: if self._current is not None: return True time.sleep(0.02) return self._current is not None # ------------------------------------------------------------------ # SSE subscribers # ------------------------------------------------------------------ def subscribe(self) -> queue.Queue: subscriber: queue.Queue = queue.Queue(maxsize=2) with self._lock: self._subscribers.add(subscriber) return subscriber def unsubscribe(self, subscriber: queue.Queue) -> None: with self._lock: self._subscribers.discard(subscriber) def health(self) -> Dict[str, Any]: with self._lock: age = time.time() - self._last_sample_ts if self._last_sample_ts else None return { "ok": self._current is not None and (age is None or age < self.interval * 10), "samples": self._samples, "subscribers": len(self._subscribers), "last_sample_age": round(age, 2) if age is not None else None, "last_error": self._last_error, "interval": self.interval, "uptime_seconds": round(time.time() - self.started_at, 1), }