# steward/core/status.py
"""Unified monitor-status aggregation.
A single readable "is everything up?" surface (the Status page + dashboard
widget) is assembled from heterogeneous monitor types — core ping/DNS, plus
any plugin that contributes (e.g. the http plugin). Rather than have the core
status page import plugin tables (plugins are optional and loaded late), each
monitor type registers a *status source*: an async callable(db) that returns a
list of normalised StatusEntry objects. Core registers its ping/DNS sources in
create_app; a plugin registers its own from setup().
"""
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 # "ping" | "dns" | "http" | ...
key: str # unique within kind (host_id / 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''
)
# ── Shared query helpers (host-keyed result tables: ping, dns) ────────────────
async def _last_n_by_host(db, model, ts_col, host_ids: list[str],
n: int = HEARTBEAT_COUNT) -> dict[str, list]:
"""Return {host_id: [rows]} of the last n results per host, oldest-first."""
if not host_ids:
return {}
rn = func.row_number().over(
partition_by=model.host_id, order_by=ts_col.desc()
).label("rn")
subq = select(model.id, rn).where(model.host_id.in_(host_ids)).subquery()
res = await db.execute(
select(model)
.join(subq, model.id == subq.c.id)
.where(subq.c.rn <= n)
.order_by(model.host_id, ts_col.asc())
)
out: dict[str, list] = {hid: [] for hid in host_ids}
for row in res.scalars():
out[row.host_id].append(row)
return out
async def _uptime_by_host(db, model, ts_col, status_col, up_value,
host_ids: list[str]) -> dict[str, dict]:
"""Per-host uptime % over 24h/7d/30d in one grouped query."""
if not host_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(
model.host_id,
func.count().label("total_30d"),
func.sum(case((status_col == up_value, 1), else_=0)).label("up_30d"),
func.sum(case((ts_col >= c7, 1), else_=0)).label("total_7d"),
func.sum(case((and_(ts_col >= c7, status_col == up_value), 1), else_=0)).label("up_7d"),
func.sum(case((ts_col >= c24, 1), else_=0)).label("total_24h"),
func.sum(case((and_(ts_col >= c24, status_col == up_value), 1), else_=0)).label("up_24h"),
)
.where(model.host_id.in_(host_ids))
.where(ts_col >= c30)
.group_by(model.host_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.host_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
# ── Core status sources: ping + DNS ───────────────────────────────────────────
async def ping_status_source(db) -> list[StatusEntry]:
from steward.models.hosts import Host
from steward.models.monitors import PingResult, PingStatus
hosts = (await db.execute(
select(Host).where(Host.ping_enabled.is_(True)).order_by(Host.name)
)).scalars().all()
ids = [h.id for h in hosts]
recent = await _last_n_by_host(db, PingResult, PingResult.probed_at, ids)
uptime = await _uptime_by_host(
db, PingResult, PingResult.probed_at, PingResult.status, PingStatus.up, ids
)
entries: list[StatusEntry] = []
for h in hosts:
rows = recent.get(h.id, [])
latest = rows[-1] if rows else None
heartbeat = [{
"state": "down" if r.status == PingStatus.down else "up",
"title": (
f"Down — {r.probed_at:%H:%M:%S} UTC" if r.status == PingStatus.down
else f"{r.response_time_ms:.0f} ms — {r.probed_at:%H:%M:%S} UTC"
if r.response_time_ms is not None else f"Up — {r.probed_at:%H:%M:%S} UTC"
),
} for r in rows]
status = "pending" if latest is None else (
"down" if latest.status == PingStatus.down else "up"
)
entries.append(StatusEntry(
kind="ping", key=h.id, name=h.name, target=h.address, status=status,
last_checked=latest.probed_at if latest else None,
uptime=uptime.get(h.id, {}), heartbeat=heartbeat,
latency_ms=latest.response_time_ms if latest else None,
spark=[r.response_time_ms for r in rows if r.response_time_ms is not None],
detail_url="/ping/",
))
return entries
async def dns_status_source(db) -> list[StatusEntry]:
from steward.models.hosts import Host
from steward.models.monitors import DnsResult, DnsStatus
hosts = (await db.execute(
select(Host).where(Host.dns_enabled.is_(True)).order_by(Host.name)
)).scalars().all()
ids = [h.id for h in hosts]
recent = await _last_n_by_host(db, DnsResult, DnsResult.resolved_at, ids)
uptime = await _uptime_by_host(
db, DnsResult, DnsResult.resolved_at, DnsResult.status, DnsStatus.resolved, ids
)
entries: list[StatusEntry] = []
for h in hosts:
rows = recent.get(h.id, [])
latest = rows[-1] if rows else None
heartbeat = [{
"state": "up" if r.status == DnsStatus.resolved else "down",
"title": (
f"{r.resolved_ip} — {r.resolved_at:%H:%M:%S} UTC" if r.status == DnsStatus.resolved
else f"Failed — {r.resolved_at:%H:%M:%S} UTC"
),
} for r in rows]
status = "pending" if latest is None else (
"up" if latest.status == DnsStatus.resolved else "down"
)
entries.append(StatusEntry(
kind="dns", key=h.id, name=h.name, target=h.address, status=status,
last_checked=latest.resolved_at if latest else None,
uptime=uptime.get(h.id, {}), heartbeat=heartbeat,
detail_url="/dns/",
))
return entries