Add Linux server monitor dashboard
This commit is contained in:
yhlhoh committed
1 parent
0446648ed1
commit
315cb36a21
19 files changed
+5997
-1
No files matched your search
@@ -0,0 +1,205 @@
|
||||
"""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),
|
||||
}
|
||||
Reference in new issue
Block a user