"""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 ipaddress import json from collections.abc import Mapping from datetime import datetime, timedelta, timezone from sqlalchemy import func, select from steward.core import anomalies as anom from steward.core.formatting import status_class from steward.core.norms import Norm, floor_setting, learn_normal, norms_at from steward.core.settings import floor_of 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", "resolver"), 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 _resolver(value) -> str | None: """A DNS server's IP address, or None for the container's own resolver.""" value = str(value or "").strip() if not value: return None try: return str(ipaddress.ip_address(value)) except ValueError: raise ValueError(f"resolver must be a DNS server's IP address (e.g. 1.1.1.1), got {value!r}.") from None 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), "resolver": _resolver(opts.get("resolver"))} 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 learn_response_times(app) -> None: """Each monitor's response time learns its normal (#5742), by type (the metric's source), with its floor from Settings → Thresholds. A monitor down keeps its fixed meaning whatever its history (System 182).""" for mtype in MONITOR_TYPE_VALUES: learn_normal(mtype, "response_ms", floor_setting(app, "monitor_ms")) async def latency_norms(db, monitors: list[Monitor], now: datetime | None = None) -> dict[str, Norm]: """{monitor_id: its response time's normal for this hour}; a monitor with too little history yet is absent, and graded on the fixed cutoffs.""" got = await norms_at(db, [(m.type, m.name, "response_ms") for m in monitors], now or datetime.now(timezone.utc)) return {m.id: got[(m.type, m.name, "response_ms")] for m in monitors if (m.type, m.name, "response_ms") in got} def latency_grade(latest, thresholds: dict, norm: Norm | None = None) -> tuple[str, str | None]: """(status class, why) for the latest response time: against the latency cutoffs, except within its own normal for the hour, which is "ok" and says what is usual.""" if latest is None or latest.response_ms is None: return "unknown", None return anom.graded(latest.response_ms, "latency", thresholds, norm, floor_of(thresholds, "monitor_ms"), " ms") def row_state(monitor: Monitor, latest, uptime_pct, thresholds: dict, norm: Norm | None = None) -> 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) and out of its normal (`norm`, #5742). 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 latency_grade(latest, thresholds, norm)[0] == "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) norms = await latency_norms(db, monitors, now) data = [] for m in monitors: rows = recent.get(m.id, []) latest = rows[-1] if rows else None windows = uptime.get(m.id, {}) latency, usual = latency_grade(latest, thresholds, norms.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, norms.get(m.id)), # The latest response time's grade, and what is usual when that # quieted it (#5742). "latency": latency, "latency_usual": usual, "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 response_history(db, monitor_id: str, now: datetime, hours: int = 24, points: int = 96) -> list[dict]: """The monitor's response time and failed checks over the last `hours`, in about `points` buckets, oldest first: [{at (epoch ms), ms, down}]. `ms` is the bucket's average (None when no check returned a time, as for DNS); `down` counts the failed checks in it.""" width = max(60, hours * 3600 // points) bucket = func.date_bin(func.make_interval(0, 0, 0, 0, 0, 0, width), MonitorResult.checked_at, func.to_timestamp(0)).label("bucket") rows = (await db.execute( select(bucket, func.avg(MonitorResult.response_ms), func.count().filter(MonitorResult.is_up.is_(False))) .where(MonitorResult.monitor_id == monitor_id, MonitorResult.checked_at >= now - timedelta(hours=hours)) .group_by(bucket).order_by(bucket) )).all() return [{"at": int(at.timestamp() * 1000), "ms": None if ms is None else round(float(ms), 1), "down": int(down)} for at, ms, down in rows] 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())