alex/manage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(monitoring): decommission legacy SSH-scraping poller (slice 1)

Browse Source 
The 2026-06-16/17 observability update externalised metrics to
Prometheus + node_exporter + Grafana, but the legacy Manage-side
SSH-scraping monitor was never removed. It duplicated the new stack,
ran SSH df on every machine every 300s, and fed nothing (its UI was
deleted in e2ad731). This slice decommissions the duplication.

Removed (backend):
- services/monitoring_poller.py (MonitoringPoller) — entire file
- services/monitoring_actions.py (disk_space, run_machine_operation,
  poll_machine_snapshot, build_machine_client) — entire file;
  run_machine_operation had only 2 callers (the poller + /disk), both gone
- tests/test_monitoring_actions.py
- endpoints: POST /api/monitoring/poller, GET /machines/{id}/actions,
  GET /disk (and the now-dead _resolve_machine helper)
- lifespan wiring (main.py), dependency wrapper (dependencies.py),
  poller.start()/kick() from machine save (routers/settings.py)
- SettingsStore: monitoring_machine_actions table CREATE + 2 indexes +
  record/list/prune_machine_actions methods; DROP TABLE IF EXISTS on
  startup cleans existing DBs (user-approved)
- config knobs: monitoring_poll_interval_seconds,
  monitoring_poll_initial_delay_seconds, monitoring_action_retention_days
- test_api.py: TestMonitoring._ensure_machine + test_disk

Kept (fits the new model): /machines, /prometheus-targets, /alerts,
/alertmanager-status, /alertmanager-webhook; the disk_usage JOB template
(manual on-demand, not monitoring); node_exporter_* machine fields
(they point Prometheus at the right host).

Gate: backend pytest 173 passed; ruff clean.
This commit is contained in:
Developer
2026-06-17 20:48:56 +00:00
parent 1c29299e8c
commit 08a3b616f6
11 changed files with 113 additions and 691 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/media_library_viewer_api/services/monitoring_actions.py
-208
View File
@@ -1,208 +0,0 @@
"""Shared monitoring action helpers.
The router and the background poller both use these helpers so machine
operations are recorded consistently whether they were triggered by a user
request or by the backend's scheduled polling loop.
"""
from __future__ import annotations
import json
import logging
import shlex
import time
import uuid
from typing import Any, Callable
from fastapi import HTTPException
from media_library_viewer_api.clients.local import LocalCommandClient
from media_library_viewer_api.clients.ssh import RemoteSSHClient
from media_library_viewer_api.config import get_settings
from media_library_viewer_api.observability import record_ssh_command
from media_library_viewer_api.services.settings_store import SettingsStore
logger = logging.getLogger(__name__)
def build_machine_client(machine: dict[str, Any], store: SettingsStore):
"""Build the appropriate command client for a machine definition."""
mode = str(machine.get("mode") or "local").strip().lower()
if mode == "local":
return LocalCommandClient()
key_directory = str(machine.get("key_directory") or "").strip()
key_name = str(machine.get("key_name") or "").strip()
key_path = f"{key_directory}/{key_name}" if key_directory and key_name else None
private_key = str(machine.get("ssh_private_key") or "")
passphrase = str(machine.get("ssh_private_key_passphrase") or "")
ssh_key_id = str(machine.get("ssh_key_id") or "").strip()
if ssh_key_id:
ssh_key = store.get_ssh_key(ssh_key_id)
if ssh_key:
private_key = str(ssh_key.get("private_key") or private_key)
passphrase = str(ssh_key.get("passphrase") or passphrase)
settings = get_settings()
return RemoteSSHClient(
host=str(machine.get("host") or ""),
username=str(machine.get("username") or ""),
port=int(machine.get("port") or 22),
key_filename=key_path,
private_key=private_key or None,
private_key_passphrase=passphrase or None,
password=str(machine.get("password") or "") or None,
known_hosts_path=str(settings.ssh_known_hosts_file),
)
def disk_space(client: Any, path: str = "/") -> dict[str, Any]:
"""Return df information for the filesystem containing ``path``.
Works against any client with a ``run`` method (local shell or SSH).
"""
command = (
"df -P -B1 -- " + shlex.quote(path or "/") + " | awk 'NR==2 {printf "
'"{\\"filesystem\\":\\"%s\\",\\"size\\":%s,"'
'"\\"used\\":%s,\\"available\\":%s,"'
'"\\"used_pct\\":\\"%s\\",\\"mount\\":\\"%s\\"}", "'
"$1,$2,$3,$4,$5,$6}'"
)
logger.debug("Reading disk space for path=%s", path)
result = client.run(command, timeout=20)
if result.exit_status != 0 or not result.stdout.strip():
logger.warning("Failed to read disk space for %s: %s", path, result.stderr or result.stdout)
raise RuntimeError(result.stderr or result.stdout or "failed to read disk space")
data = json.loads(result.stdout)
logger.info("Disk space path=%s mount=%s used_pct=%s", path, data.get("mount"), data.get("used_pct"))
return data
def summarize_operation_result(action: str, result: Any) -> str:
"""Turn an operation result into a compact human-readable summary."""
if result is None:
return action
if isinstance(result, str):
text = result.strip().splitlines()[0] if result.strip() else action
return text[:200]
if isinstance(result, list):
return f"{action}: {len(result)} item(s)"
if isinstance(result, dict):
if action.startswith("disk lookup"):
used_pct = result.get("used_pct")
mount = result.get("mount") or result.get("filesystem")
return f"disk {mount or ''} used {used_pct or '?'}".strip()
if "message" in result and isinstance(result["message"], str):
return result["message"][:200]
return json_compact(result)
return action
def json_compact(value: Any) -> str:
try:
text = json.dumps(value, sort_keys=True, default=str)
return text[:200]
except Exception:
return str(value)[:200]
def run_machine_operation(
machine: dict[str, Any],
store: SettingsStore,
action: str,
callback: Callable[[Any], Any],
*,
summarize: Callable[[Any], str] | None = None,
request_id: str = "",
raise_http: bool = True,
client: Any | None = None,
) -> Any:
"""Run a machine operation, record history, and optionally raise on failure."""
started = time.perf_counter()
if client is None:
client = build_machine_client(machine, store)
try:
result = callback(client)
duration_ms = int((time.perf_counter() - started) * 1000)
record_ssh_command(
machine_id=machine.get("id") or "unknown",
action=action,
status="ok",
duration_seconds=duration_ms / 1000.0,
)
store.record_machine_action(
machine,
action,
"ok",
duration_ms=duration_ms,
request_id=request_id,
message=(summarize(result) if summarize else summarize_operation_result(action, result)),
)
return result
except HTTPException:
raise
except Exception as exc: # pragma: no cover - transport/network fallback
duration_ms = int((time.perf_counter() - started) * 1000)
record_ssh_command(
machine_id=machine.get("id") or "unknown",
action=action,
status="error",
duration_seconds=duration_ms / 1000.0,
)
logger.exception(
"Monitoring %s failed machine_id=%s machine_name=%s",
action,
machine["id"],
machine["name"],
)
error_text = str(exc)
store.record_machine_action(
machine,
action,
"error",
duration_ms=duration_ms,
request_id=request_id,
error=error_text,
)
if not raise_http:
return None
status_code = 503 if machine.get("mode") == "local" else 502
raise HTTPException(
status_code=status_code,
detail=f"{machine['name']}: {action} failed: {exc}",
) from exc
def poll_machine_snapshot(
machine: dict[str, Any],
store: SettingsStore,
*,
metrics_limit: int = 70_000,
request_id: str | None = None,
) -> dict[str, Any]:
"""Collect a backend-scheduled snapshot for a machine.
The legacy POSIX collector has been removed; this now records a lightweight
disk-space lookup on the same schedule so action history stays useful.
"""
request_id = request_id or f"poll:{machine.get('id') or uuid.uuid4().hex}"
results: dict[str, Any] = {"request_id": request_id, "machine_id": machine.get("id"), "actions": []}
client = build_machine_client(machine, store)
settings = get_settings()
path = str(machine.get("media_root") or settings.media_root or "/")
disk = run_machine_operation(
machine,
store,
f"disk lookup for {path}",
lambda client: disk_space(client, path),
request_id=request_id,
raise_http=False,
client=client,
)
results["disk_mount"] = (disk or {}).get("mount") if isinstance(disk, dict) else None
results["actions"].append("disk lookup")
return results
backend/src/media_library_viewer_api/services/monitoring_poller.py
-197
View File
@@ -1,197 +0,0 @@
"""Background poller for monitoring machine snapshots.
The poller runs entirely inside the backend. It periodically reads the defined
machines, collects a small snapshot from each enabled machine over SSH or local
shell execution, and stores the resulting history rows in the settings DB.
This keeps the Monitoring page populated without any daemon or agent running on
the remote machines.
"""
from __future__ import annotations
import logging
import threading
import time
from dataclasses import dataclass
from typing import Any
from media_library_viewer_api.config import get_settings
from media_library_viewer_api.services.monitoring_actions import poll_machine_snapshot
from media_library_viewer_api.services.settings_store import SettingsStore, get_settings_store
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class PollerConfig:
interval_seconds: int = 300
initial_delay_seconds: int = 20
metrics_limit: int = 70_000
retention_days: int = 30
class MonitoringPoller:
"""Single-worker background poller for monitoring snapshots."""
def __init__(self) -> None:
self._thread: threading.Thread | None = None
self._stop_event = threading.Event()
self._lock = threading.Lock()
self._last_run_at: float | None = None
self._last_success_at: float | None = None
self._last_error: str = ""
self._last_cycle_ms: int | None = None
self._poll_count = 0
self._error_count = 0
def _config(self) -> PollerConfig:
settings = get_settings()
return PollerConfig(
interval_seconds=max(30, int(getattr(settings, "monitoring_poll_interval_seconds", 300) or 300)),
initial_delay_seconds=max(0, int(getattr(settings, "monitoring_poll_initial_delay_seconds", 20) or 20)),
metrics_limit=70_000,
retention_days=max(1, int(getattr(settings, "monitoring_action_retention_days", 30) or 30)),
)
def start(self) -> None:
"""Start the background worker if it is not already running."""
with self._lock:
if self._thread and self._thread.is_alive():
return
self._stop_event.clear()
self._thread = threading.Thread(target=self._run, name="monitoring-poller", daemon=True)
self._thread.start()
logger.info("Monitoring poller started")
def kick(self) -> None:
"""Run one immediate snapshot cycle in the background."""
store = get_settings_store()
config = self._config()
threading.Thread(
target=self._run_cycle,
args=(store, config),
name="monitoring-poller-kick",
daemon=True,
).start()
logger.info("Monitoring poller kick requested")
def stop(self, timeout: float = 5.0) -> None:
"""Stop the worker thread and wait briefly for shutdown."""
with self._lock:
thread = self._thread
if not thread:
return
self._stop_event.set()
thread.join(timeout=timeout)
if thread.is_alive():
logger.warning("Monitoring poller did not stop within %.1fs", timeout)
else:
logger.info("Monitoring poller stopped")
with self._lock:
if self._thread is thread:
self._thread = None
def status(self) -> dict[str, Any]:
"""Return a small status snapshot for diagnostics and tests."""
with self._lock:
return {
"worker_running": bool(self._thread and self._thread.is_alive()),
"stop_requested": self._stop_event.is_set(),
"last_run_at": self._last_run_at,
"last_success_at": self._last_success_at,
"last_error": self._last_error,
"last_cycle_ms": self._last_cycle_ms,
"poll_count": self._poll_count,
"error_count": self._error_count,
}
def snapshot(self) -> dict[str, Any]:
"""Return status plus the active polling configuration."""
data = self.status()
config = self._config()
data.update(
{
"interval_seconds": config.interval_seconds,
"initial_delay_seconds": config.initial_delay_seconds,
"retention_days": config.retention_days,
}
)
return data
def _run_cycle(self, store: SettingsStore, config: PollerConfig) -> None:
start = time.perf_counter()
machines = store.list_machines()
enabled = [machine for machine in machines if machine.get("enabled")]
logger.info("Monitoring poll cycle starting enabled_machines=%s", len(enabled))
cycle_errors = 0
for machine in enabled:
if self._stop_event.is_set():
break
try:
snapshot = poll_machine_snapshot(
machine,
store,
metrics_limit=config.metrics_limit,
request_id=f"poll:{machine['id']}:{int(time.time())}",
)
logger.info(
"Monitoring poll snapshot machine_id=%s request_id=%s disk_mount=%s actions=%s",
machine["id"],
snapshot.get("request_id"),
snapshot.get("disk_mount"),
snapshot.get("actions"),
)
except Exception:
cycle_errors += 1
logger.exception(
"Monitoring poll snapshot failed machine_id=%s machine_name=%s", machine["id"], machine["name"]
)
retention_seconds = config.retention_days * 24 * 60 * 60
cutoff_ts = int(time.time()) - retention_seconds
removed = store.prune_machine_actions(cutoff_ts)
if removed:
logger.info("Pruned %s old monitoring action rows older than %s", removed, cutoff_ts)
duration_ms = int((time.perf_counter() - start) * 1000)
with self._lock:
self._last_run_at = time.time()
self._last_cycle_ms = duration_ms
self._poll_count += 1
if cycle_errors:
self._error_count += cycle_errors
self._last_error = f"{cycle_errors} machine(s) failed"
else:
self._last_success_at = self._last_run_at
self._last_error = ""
logger.info(
"Monitoring poll cycle complete enabled_machines=%s errors=%s duration_ms=%s removed_rows=%s",
len(enabled),
cycle_errors,
duration_ms,
removed,
)
def _run(self) -> None:
config = self._config()
if config.initial_delay_seconds:
logger.info("Monitoring poller initial delay=%ss", config.initial_delay_seconds)
if self._stop_event.wait(config.initial_delay_seconds):
return
store = get_settings_store()
while not self._stop_event.is_set():
try:
self._run_cycle(store, config)
except Exception:
with self._lock:
self._last_error = "poller cycle failed"
self._error_count += 1
logger.exception("Monitoring poller cycle failed")
if self._stop_event.wait(config.interval_seconds):
break
_MONITORING_POLLER = MonitoringPoller()
def get_monitoring_poller() -> MonitoringPoller:
return _MONITORING_POLLER
backend/src/media_library_viewer_api/services/settings_store.py
+4 -113
View File
@@ -85,25 +85,10 @@ class SettingsStore:
"""
)
conn.execute("CREATE INDEX IF NOT EXISTS idx_monitoring_machines_mode ON monitoring_machines(mode)")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS monitoring_machine_actions (
id TEXT PRIMARY KEY,
machine_id TEXT NOT NULL,
machine_name TEXT NOT NULL,
mode TEXT NOT NULL,
action TEXT NOT NULL,
status TEXT NOT NULL,
created_at INTEGER NOT NULL,
duration_ms INTEGER NOT NULL,
request_id TEXT NOT NULL,
message TEXT NOT NULL,
error TEXT NOT NULL,
stdout_tail TEXT NOT NULL,
stderr_tail TEXT NOT NULL
)
"""
)
# The legacy SSH-scraping monitor (MonitoringPoller) was decommissioned;
# metrics now live in Prometheus/node_exporter/Grafana. Drop the orphan
# table on startup so existing databases get a clean slate.
conn.execute("DROP TABLE IF EXISTS monitoring_machine_actions")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS ssh_keys (
@@ -178,18 +163,6 @@ class SettingsStore:
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_dashboard_shortcuts_type ON dashboard_shortcuts(shortcut_type)"
)
conn.execute(
"""
CREATE INDEX IF NOT EXISTS idx_monitoring_machine_actions_machine_time
ON monitoring_machine_actions(machine_id, created_at DESC)
"""
)
conn.execute(
"""
CREATE INDEX IF NOT EXISTS idx_monitoring_machine_actions_action_status
ON monitoring_machine_actions(action, status)
"""
)
conn.execute("""
CREATE TABLE IF NOT EXISTS backup_jobs (
id TEXT PRIMARY KEY,
@@ -563,88 +536,6 @@ class SettingsStore:
with self.connect() as conn:
conn.execute("DELETE FROM monitoring_machines WHERE id = ?", (machine_id,))
def record_machine_action(
self,
machine: dict[str, Any],
action: str,
status: str,
*,
duration_ms: int,
request_id: str = "",
message: str = "",
error: str = "",
stdout_tail: str = "",
stderr_tail: str = "",
) -> None:
"""Store a compact action history row for a machine operation."""
self.init_schema()
now = int(time.time())
with self.connect() as conn:
conn.execute(
"""
INSERT INTO monitoring_machine_actions
(
id, machine_id, machine_name, mode, action, status,
created_at, duration_ms, request_id, message, error,
stdout_tail, stderr_tail
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
uuid.uuid4().hex,
str(machine.get("id") or ""),
str(machine.get("name") or ""),
str(machine.get("mode") or "local"),
action,
status,
now,
duration_ms,
request_id,
message,
error,
stdout_tail,
stderr_tail,
),
)
def list_machine_actions(
self,
machine_id: str,
*,
limit: int = 20,
action: str | None = None,
status: str | None = None,
) -> list[dict[str, Any]]:
self.init_schema()
clauses = ["machine_id = ?"]
params: list[Any] = [machine_id]
if action:
clauses.append("action = ?")
params.append(action)
if status:
clauses.append("status = ?")
params.append(status)
sql = (
"SELECT machine_id, machine_name, mode, action, status, "
"created_at, duration_ms, request_id, message, error, stdout_tail, stderr_tail "
f"FROM monitoring_machine_actions WHERE {' AND '.join(clauses)} "
"ORDER BY created_at DESC LIMIT ?"
)
params.append(max(1, min(int(limit), 200)))
with self.connect() as conn:
rows = conn.execute(sql, params).fetchall()
return [dict(row) for row in rows]
def prune_machine_actions(self, older_than_ts: int) -> int:
"""Delete action history rows older than the given timestamp."""
self.init_schema()
with self.connect() as conn:
cur = conn.execute(
"DELETE FROM monitoring_machine_actions WHERE created_at < ?",
(int(older_than_ts),),
)
return int(cur.rowcount or 0)
@staticmethod
def _private_key_summary(private_key: str) -> dict[str, str]:
if not private_key: