205 lines
7.6 KiB
Python
205 lines
7.6 KiB
Python
"""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),
|
|
} |