# steward/core/status.py
"""Unified monitor-status aggregation.
A single readable "is everything up?" surface (the Status page + dashboard
widget) is assembled from heterogeneous monitors. The core unified Monitor
entity (ping/dns/http) contributes via `monitor_status_source`; plugins may
register their own sources from setup(). Each source is an async callable(db)
returning normalised StatusEntry objects so the Status page never imports
plugin tables directly.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Awaitable, Callable
from sqlalchemy import and_, case, func, select
logger = logging.getLogger(__name__)
HEARTBEAT_COUNT = 30 # number of recent checks shown in the heartbeat bar
@dataclass
class StatusEntry:
"""One monitored thing, normalised across monitor types for display."""
kind: str # "icmp" | "tcp" | "dns" | "http" | ...
key: str # unique within kind (monitor id)
name: str
target: str = "" # address / URL shown as subtitle
status: str = "pending" # "up" | "down" | "pending"
last_checked: datetime | None = None
uptime: dict[str, float | None] = field(default_factory=dict) # 24h/7d/30d
heartbeat: list[dict] = field(default_factory=list) # oldest-first {state,title}
latency_ms: float | None = None
spark: list[float] = field(default_factory=list) # response series
spark_svg: str = "" # filled by the route via sparkline_svg()
tls_days: float | None = None # days until TLS expiry (http only)
detail_url: str | None = None
StatusSource = Callable[[object], Awaitable[list[StatusEntry]]]
_SOURCES: list[StatusSource] = []
def register_status_source(fn: StatusSource) -> None:
"""Register a status source. Idempotent on the same callable."""
if fn not in _SOURCES:
_SOURCES.append(fn)
def clear_status_sources() -> None:
"""Reset the registry (used by tests)."""
_SOURCES.clear()
async def collect_status(db) -> list[StatusEntry]:
"""Gather entries from every registered source, down-first then by name.
A failing source is logged and skipped so one broken plugin can't blank
the whole Status page.
"""
entries: list[StatusEntry] = []
for src in list(_SOURCES):
try:
entries.extend(await src(db))
except Exception:
logger.exception("status source %r failed", getattr(src, "__name__", src))
# Down first (most urgent), then pending, then up; alphabetical within.
order = {"down": 0, "pending": 1, "up": 2}
entries.sort(key=lambda e: (order.get(e.status, 3), e.kind, e.name.lower()))
return entries
def sparkline_svg(values: list[float], width: int = 80, height: int = 20,
stroke: str = "#6060c0") -> str:
"""Tiny inline SVG polyline for a response-time series."""
vals = [v for v in values if v is not None]
if len(vals) < 2:
return f''
mn, mx = min(vals), max(vals)
if mx == mn:
mx = mn + 1.0
step = width / (len(vals) - 1)
pts = []
for i, v in enumerate(vals):
x = i * step
y = height - (v - mn) / (mx - mn) * (height - 2) - 1
pts.append(f"{x:.1f},{y:.1f}")
poly = " ".join(pts)
return (
f''
)
# ── Unified monitor status source ─────────────────────────────────────────────
async def _last_n_results(db, monitor_ids: list[str],
n: int = HEARTBEAT_COUNT) -> dict[str, list]:
"""Return {monitor_id: [MonitorResult]} of the last n results, oldest-first."""
from steward.models.monitors import MonitorResult
if not monitor_ids:
return {}
rn = func.row_number().over(
partition_by=MonitorResult.monitor_id, order_by=MonitorResult.checked_at.desc()
).label("rn")
subq = select(MonitorResult.id, rn).where(
MonitorResult.monitor_id.in_(monitor_ids)
).subquery()
res = await db.execute(
select(MonitorResult)
.join(subq, MonitorResult.id == subq.c.id)
.where(subq.c.rn <= n)
.order_by(MonitorResult.monitor_id, MonitorResult.checked_at.asc())
)
out: dict[str, list] = {mid: [] for mid in monitor_ids}
for row in res.scalars():
out[row.monitor_id].append(row)
return out
async def _uptime_by_monitor(db, monitor_ids: list[str]) -> dict[str, dict]:
"""Per-monitor uptime % over 24h/7d/30d in one grouped query."""
from steward.models.monitors import MonitorResult
if not monitor_ids:
return {}
now = datetime.now(timezone.utc)
c30, c7, c24 = now - timedelta(days=30), now - timedelta(days=7), now - timedelta(hours=24)
res = await db.execute(
select(
MonitorResult.monitor_id,
func.count().label("total_30d"),
func.sum(case((MonitorResult.is_up.is_(True), 1), else_=0)).label("up_30d"),
func.sum(case((MonitorResult.checked_at >= c7, 1), else_=0)).label("total_7d"),
func.sum(case((and_(MonitorResult.checked_at >= c7, MonitorResult.is_up.is_(True)), 1), else_=0)).label("up_7d"),
func.sum(case((MonitorResult.checked_at >= c24, 1), else_=0)).label("total_24h"),
func.sum(case((and_(MonitorResult.checked_at >= c24, MonitorResult.is_up.is_(True)), 1), else_=0)).label("up_24h"),
)
.where(MonitorResult.monitor_id.in_(monitor_ids))
.where(MonitorResult.checked_at >= c30)
.group_by(MonitorResult.monitor_id)
)
def _pct(up, total):
return round(float(up) / float(total) * 100, 2) if total else None
out: dict[str, dict] = {}
for r in res:
out[r.monitor_id] = {
"24h": _pct(r.up_24h, r.total_24h),
"7d": _pct(r.up_7d, r.total_7d),
"30d": _pct(r.up_30d, r.total_30d),
}
return out
def _heartbeat_title(mtype: str, r) -> str:
"""Type-aware tooltip for one heartbeat cell."""
ts = f"{r.checked_at:%H:%M:%S} UTC"
if not r.is_up:
return f"{(r.error_msg or 'Down')} — {ts}"
if mtype == "http":
code = r.status_code or "—"
ms = f"{r.response_ms:.0f} ms" if r.response_ms is not None else ""
return f"{code} · {ms} — {ts}".replace(" · — ", " — ")
if mtype == "dns":
return f"{r.resolved_ip or 'resolved'} — {ts}"
if r.response_ms is not None:
return f"{r.response_ms:.0f} ms — {ts}"
return f"Up — {ts}"
async def monitor_status_source(db) -> list[StatusEntry]:
"""Contribute every enabled Monitor (all types) to the Status page."""
from steward.models.monitors import Monitor
monitors = list((await db.execute(
select(Monitor).where(Monitor.enabled.is_(True)).order_by(Monitor.name)
)).scalars())
ids = [m.id for m in monitors]
recent = await _last_n_results(db, ids)
uptime = await _uptime_by_monitor(db, ids)
now = datetime.now(timezone.utc)
entries: list[StatusEntry] = []
for m in monitors:
rows = recent.get(m.id, [])
latest = rows[-1] if rows else None
status = "pending" if latest is None else ("up" if latest.is_up else "down")
tls_days = None
if latest and latest.tls_expires_at:
tls_days = (latest.tls_expires_at - now).total_seconds() / 86400.0
entries.append(StatusEntry(
kind=m.type, key=m.id, name=m.name, target=m.target, status=status,
last_checked=latest.checked_at if latest else None,
uptime=uptime.get(m.id, {}),
heartbeat=[{"state": "up" if r.is_up else "down",
"title": _heartbeat_title(m.type, r)} for r in rows],
latency_ms=latest.response_ms if latest else None,
spark=[r.response_ms for r in rows if r.response_ms is not None],
tls_days=tls_days, detail_url="/monitors/",
))
return entries