bvandeusenandClaude Opus 5.5
f3433fe3ce
Host CPU, memory and load per core, container CPU and memory, and every monitor's response time learn a normal per hour (milestone 513, System 182). A reading past its fixed cutoff but within its normal reads healthy and says what is usual at this hour; one well out of it keeps its grade and is a metric anomaly. Disk, a monitor down and an agent offline keep their cutoffs. - Each normal's floor, the smallest departure that counts, is a setting: Settings -> Thresholds -> Normals (norms.floor.*). Defaults: host CPU 10 points, host memory 5, load per core 10, container CPU 10 points of one core, container memory 5, monitor response 100 ms; Traefik's 2-point 4xx and 50 ms p95 floors move there from constants. learn_normal takes a floor function (floor_setting) read at each pass. - learn_normal(parts=False): a host learns, its cores don't. - Ingest records load_pct (load_1m / cores x 100), the figure pages grade. - Shared grading: anomalies.against / graded, worst_reading(normals=), hosts' host_grades, docker usage(thresholds=, norms=), monitors' latency_grade. Applied to host vitals, the Hosts widget, MCP overview, Docker usage, and the monitor list, widget and page. - The monitor and container pages list their anomalies. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
337 lines
15 KiB
Python
337 lines
15 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 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())
|