bvandeusenandClaude Opus 5.5
434127b3e1
Status repeated the Monitors page's counts, check strip and uptime. Monitors now has one header row and every column Status had: the last 30 checks, latest response with its trend, uptime over 24h / 7d / 30d (the range picker goes) and TLS days on every row. /status and /status/rows and the nav link are removed outright; the dashboard Overview (/status/widget) stays and its Details link goes to Monitors. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
266 lines
11 KiB
Python
266 lines
11 KiB
Python
"""Monitor queries and validation shared by the monitors pages and the MCP tools.
|
|
|
|
Each function takes an open session (or plain values), with no Quart request
|
|
context, so the routes and the /mcp tools (which run outside Quart's
|
|
pipeline) call the same code and refuse the same input for the same reasons.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from collections.abc import Mapping
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import func, select
|
|
|
|
from steward.core.formatting import status_class
|
|
from steward.core.status import _heartbeat_title, _last_n_results, _uptime_by_monitor
|
|
from steward.models.monitors import MONITOR_TYPE_VALUES, Monitor, MonitorResult, MonitorType
|
|
|
|
HTTP_METHODS = ("GET", "HEAD", "POST")
|
|
# The config keys each type takes (models/monitors.py documents their meaning).
|
|
CONFIG_KEYS: dict[str, tuple[str, ...]] = {
|
|
MonitorType.icmp.value: (),
|
|
MonitorType.tcp.value: ("port",),
|
|
MonitorType.dns.value: ("expected_ip",),
|
|
MonitorType.http.value: ("method", "expected_status", "content_match", "headers",
|
|
"timeout_seconds", "follow_redirects", "verify_ssl"),
|
|
}
|
|
|
|
|
|
def _whole(name: str, value, default: int, lo: int, hi: int | None = None) -> int:
|
|
if value is None or value == "":
|
|
return default
|
|
if isinstance(value, bool):
|
|
raise ValueError(f"{name} must be a whole number, got {value!r}.")
|
|
try:
|
|
n = int(value)
|
|
except (TypeError, ValueError):
|
|
raise ValueError(f"{name} must be a whole number, got {value!r}.") from None
|
|
if n < lo or (hi is not None and n > hi):
|
|
bounds = f"between {lo} and {hi}" if hi is not None else f"at least {lo}"
|
|
raise ValueError(f"{name} must be {bounds}, got {n}.")
|
|
return n
|
|
|
|
|
|
def _headers(value) -> dict:
|
|
if value is None or value == "":
|
|
return {}
|
|
if isinstance(value, str):
|
|
try:
|
|
value = json.loads(value)
|
|
except ValueError:
|
|
raise ValueError("headers must be a JSON object of name: value strings.") from None
|
|
if not isinstance(value, dict) or not all(isinstance(k, str) and isinstance(v, str)
|
|
for k, v in value.items()):
|
|
raise ValueError("headers must be an object of name: value strings.")
|
|
return value
|
|
|
|
|
|
def normalise_target(mtype: str, target: str) -> str:
|
|
"""For http, default a bare host to https://; others pass through."""
|
|
target = (target or "").strip()
|
|
if mtype == MonitorType.http.value and target and not target.startswith(("http://", "https://")):
|
|
return "https://" + target
|
|
return target
|
|
|
|
|
|
def build_config(mtype: str, opts: Mapping) -> dict:
|
|
"""The type-specific config from already-typed or form-string values.
|
|
|
|
Missing keys take their defaults; a key the type does not take, or a
|
|
value out of range, raises ValueError saying which and why.
|
|
"""
|
|
unknown = sorted(set(opts) - set(CONFIG_KEYS[mtype]))
|
|
if unknown:
|
|
allowed = ", ".join(CONFIG_KEYS[mtype]) or "no config"
|
|
raise ValueError(f"a {mtype} monitor does not take {', '.join(unknown)}; it takes {allowed}.")
|
|
if mtype == MonitorType.tcp.value:
|
|
return {"port": _whole("port", opts.get("port"), 80, 1, 65535)}
|
|
if mtype == MonitorType.dns.value:
|
|
return {"expected_ip": (str(opts.get("expected_ip") or "").strip() or None)}
|
|
if mtype == MonitorType.http.value:
|
|
method = str(opts.get("method") or "GET").strip().upper()
|
|
if method not in HTTP_METHODS:
|
|
raise ValueError(f"method {method!r} is not one of {', '.join(HTTP_METHODS)}.")
|
|
return {
|
|
"method": method,
|
|
"expected_status": _whole("expected_status", opts.get("expected_status"), 200, 100, 599),
|
|
"content_match": str(opts.get("content_match") or "").strip(),
|
|
"headers": _headers(opts.get("headers")),
|
|
"timeout_seconds": _whole("timeout_seconds", opts.get("timeout_seconds"), 10, 1, 60),
|
|
"follow_redirects": bool(opts.get("follow_redirects", True)),
|
|
"verify_ssl": bool(opts.get("verify_ssl", True)),
|
|
}
|
|
return {} # icmp
|
|
|
|
|
|
def monitor_fields(mtype: str, *, target: str, name: str = "", check_interval_seconds=0,
|
|
opts: Mapping | None = None) -> dict:
|
|
"""Validated column values for a new or edited monitor; ValueError says why not."""
|
|
mtype = (mtype or "").strip().lower()
|
|
if mtype not in MONITOR_TYPE_VALUES:
|
|
raise ValueError(f"type {mtype!r} is not one of {', '.join(MONITOR_TYPE_VALUES)}.")
|
|
target = normalise_target(mtype, target)
|
|
if not target:
|
|
raise ValueError("a monitor needs a target: an address, or a URL for http.")
|
|
if len(target) > 2048:
|
|
raise ValueError("target is longer than 2048 characters.")
|
|
return {
|
|
"type": mtype,
|
|
"target": target,
|
|
"name": ((name or "").strip() or target)[:128],
|
|
"config_json": json.dumps(build_config(mtype, opts or {})),
|
|
"check_interval_seconds": _whole("check_interval_seconds", check_interval_seconds, 0, 0),
|
|
}
|
|
|
|
|
|
async def check_name_free(db, mtype: str, name: str, exclude_id: str | None = None) -> None:
|
|
"""ValueError when another monitor of this type already has this name.
|
|
|
|
A monitor's checks are recorded, and its alert rules watch it, by
|
|
(type, name), so two monitors sharing both would share one history.
|
|
"""
|
|
q = select(Monitor.id).where(Monitor.type == mtype, Monitor.name == name)
|
|
if exclude_id:
|
|
q = q.where(Monitor.id != exclude_id)
|
|
if (await db.execute(q.limit(1))).first():
|
|
raise ValueError(f"another {mtype} monitor is already named {name!r}; its checks and "
|
|
"alert rules go by that name, so pick a different one.")
|
|
|
|
|
|
# Where each state sorts on the Monitors page: what needs looking at first.
|
|
STATE_ORDER = {"down": 0, "degraded": 1, "pending": 2, "up": 3, "paused": 4}
|
|
|
|
|
|
def row_state(monitor: Monitor, latest, uptime_pct, thresholds: dict) -> str:
|
|
"""paused, pending (never checked), down, degraded or up.
|
|
|
|
Degraded is up but worth a look: uptime over the window at or below its
|
|
warning cutoff, or the latest response time past its critical one
|
|
(Settings → Thresholds). A slow-but-tolerable reply is coloured, not sorted.
|
|
"""
|
|
if not monitor.enabled:
|
|
return "paused"
|
|
if latest is None:
|
|
return "pending"
|
|
if not latest.is_up:
|
|
return "down"
|
|
if (status_class(uptime_pct, "uptime", thresholds) in ("warn", "crit")
|
|
or status_class(latest.response_ms, "latency", thresholds) == "crit"):
|
|
return "degraded"
|
|
return "up"
|
|
|
|
|
|
def down_reason(latest) -> str | None:
|
|
"""Why a failed check failed, in a few words: the probe's error, else the
|
|
HTTP status, else a content mismatch; None when the result says nothing."""
|
|
if latest is None:
|
|
return None
|
|
if latest.error_msg:
|
|
return latest.error_msg
|
|
if latest.status_code:
|
|
return f"HTTP {latest.status_code}"
|
|
if latest.content_matched is False:
|
|
return "expected content not found"
|
|
return None
|
|
|
|
|
|
def problems_first(rows: list[dict]) -> list[dict]:
|
|
"""Down, then degraded, then pending, up and paused; by name within each."""
|
|
return sorted(rows, key=lambda d: (STATE_ORDER[d["state"]], d["monitor"].name.lower()))
|
|
|
|
|
|
async def monitor_rows(db, thresholds: dict):
|
|
"""Per-monitor state: latest result, heartbeat strip, latency trend,
|
|
uptime over 24h / 7d / 30d and TLS days, problems first.
|
|
|
|
Returns (rows, counts). `counts` has up (degraded included), degraded,
|
|
down, pending and paused. A monitor is degraded on its 24h uptime.
|
|
"""
|
|
from steward.core.status import sparkline_svg
|
|
monitors = list((await db.execute(select(Monitor).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)
|
|
data = []
|
|
for m in monitors:
|
|
rows = recent.get(m.id, [])
|
|
latest = rows[-1] if rows else None
|
|
windows = uptime.get(m.id, {})
|
|
tls_days = None
|
|
if latest and latest.tls_expires_at:
|
|
tls_days = (latest.tls_expires_at - now).total_seconds() / 86400.0
|
|
times = [r.response_ms for r in rows if r.response_ms is not None]
|
|
data.append({
|
|
"monitor": m,
|
|
"latest": latest,
|
|
"state": row_state(m, latest, windows.get("24h"), thresholds),
|
|
"heartbeat": [{"state": "up" if r.is_up else "down",
|
|
"title": _heartbeat_title(m.type, r)} for r in rows],
|
|
"spark": sparkline_svg(times, width=56, height=16, stroke="currentColor") if len(times) > 1 else "",
|
|
"uptime": windows,
|
|
"tls_days": tls_days,
|
|
"down_since": None,
|
|
"down_reason": down_reason(latest) if latest is not None and not latest.is_up else None,
|
|
})
|
|
# When each down monitor went down (a query per down monitor: few, and indexed).
|
|
for d in data:
|
|
if d["state"] == "down":
|
|
d["down_since"] = await down_since(db, d["monitor"].id)
|
|
counts = {k: sum(d["state"] == k for d in data) for k in STATE_ORDER}
|
|
counts["up"] += counts["degraded"]
|
|
return problems_first(data), counts
|
|
|
|
|
|
async def down_since(db, monitor_id: str):
|
|
"""When the monitor's current run of failures began, or None if its latest check is up.
|
|
|
|
The first failed check after the last success; if it has never succeeded,
|
|
its first check.
|
|
"""
|
|
last_up = (await db.execute(
|
|
select(func.max(MonitorResult.checked_at))
|
|
.where(MonitorResult.monitor_id == monitor_id, MonitorResult.is_up.is_(True))
|
|
)).scalar()
|
|
stmt = select(func.min(MonitorResult.checked_at)).where(MonitorResult.monitor_id == monitor_id)
|
|
if last_up is not None:
|
|
stmt = stmt.where(MonitorResult.checked_at > last_up)
|
|
return (await db.execute(stmt)).scalar()
|
|
|
|
|
|
async def monitor_states(db, monitors: list[Monitor]) -> dict[str, dict]:
|
|
"""{monitor_id: {latest, recent, uptime, down_since}} for the given monitors.
|
|
|
|
`recent` is the last HEARTBEAT_COUNT results oldest-first; `uptime` is
|
|
{"24h","7d","30d"} percentages; `down_since` is set only for a monitor
|
|
whose latest check failed.
|
|
"""
|
|
ids = [m.id for m in monitors]
|
|
recent = await _last_n_results(db, ids)
|
|
uptime = await _uptime_by_monitor(db, ids)
|
|
out: dict[str, dict] = {}
|
|
for m in monitors:
|
|
rows = recent.get(m.id, [])
|
|
latest = rows[-1] if rows else None
|
|
out[m.id] = {
|
|
"latest": latest,
|
|
"recent": rows,
|
|
"uptime": uptime.get(m.id, {}),
|
|
"down_since": None,
|
|
}
|
|
if latest is not None and not latest.is_up:
|
|
out[m.id]["down_since"] = await down_since(db, m.id)
|
|
return out
|
|
|
|
|
|
async def recent_failures(db, monitor_id: str, limit: int = 10) -> list[MonitorResult]:
|
|
"""The monitor's latest failed checks, newest first."""
|
|
return list((await db.execute(
|
|
select(MonitorResult)
|
|
.where(MonitorResult.monitor_id == monitor_id, MonitorResult.is_up.is_(False))
|
|
.order_by(MonitorResult.checked_at.desc())
|
|
.limit(limit)
|
|
)).scalars())
|