bvandeusenandClaude Opus 5.5
5569f9934e
#5740 "A metric that departs from its normal is recorded as an anomaly to review, and is findable after it ends", milestone 513. - Every minute each learned key's last 15 minutes is judged against its hour's normal: above the usual hold x1.25 and by the metric's floor opens an occurrence, which ends when the level comes back. - metric_anomalies holds one row per item to review; a repeat while it still waits joins it as an occurrence (metric_anomaly_spans). - An ended row nobody reviews within anomalies.review_days (14, Settings -> Thresholds) lapses; an ongoing one never does. - Learning skips every hour an occurrence covers unless its anomaly was marked normal: no normal is learned without review. - Alerts -> Anomalies lists those to review first, then the reviewed and lapsed under day dividers; marking is operator/admin and audited. - The Overview widget and get_overview warn of open ones and count those to review; the host page shows its own; the weekly report lists the week's and the lapsed; MCP list_anomalies and review_anomaly. - Reviewed and lapsed rows are pruned at data.retention_days, and kept in backups like alert history. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
255 lines
11 KiB
Python
255 lines
11 KiB
Python
"""get_overview: what is wrong right now, ranked, across everything Steward watches."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import timedelta
|
|
|
|
from sqlalchemy import select
|
|
|
|
from steward.alerts import service as alert_svc
|
|
from steward.core import anomalies as anom
|
|
from steward.hosts import service as host_svc
|
|
from steward.mcp._context import mcp_app
|
|
from steward.models.ansible import AnsibleRun, AnsibleRunStatus
|
|
from steward.models.hosts import Host
|
|
from steward.models.monitors import Monitor
|
|
from steward.models.users import UserRole
|
|
from steward.monitors import service as monitor_svc
|
|
|
|
from ._shape import age_s, db_session, iso, now_utc, rnd
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
FINDINGS_CAP = 50
|
|
_SEVERITY = {"critical": 0, "warning": 1}
|
|
|
|
|
|
def finding(severity: str, kind: str, subject: str, summary: str, *, since=None,
|
|
ids: dict | None = None, next: str | None = None) -> dict:
|
|
"""One overview finding. Plugins' `mcp_overview(db)` return a list of these."""
|
|
return {
|
|
"severity": severity, "kind": kind, "subject": subject, "summary": summary,
|
|
"since": iso(since), "age_s": age_s(since), "ids": ids or {}, "next": next,
|
|
}
|
|
|
|
|
|
def rank(findings: list[dict]) -> list[dict]:
|
|
"""Critical first; within a severity, the longest-standing first."""
|
|
return sorted(findings, key=lambda f: (_SEVERITY.get(f.get("severity"), 2),
|
|
-(f.get("age_s") or 0), f.get("subject", "")))
|
|
|
|
|
|
def monitor_findings(monitors, states: dict, host_names: dict, now,
|
|
tls: dict | None = None) -> list[dict]:
|
|
"""Failing monitors, and HTTPS certificates inside the TLS window
|
|
(`tls` is THRESHOLDS["tls_days"]: warn/crit days left)."""
|
|
tls = tls or {"warn": 14, "crit": 3}
|
|
out = []
|
|
for m in monitors:
|
|
if not m.enabled:
|
|
continue
|
|
st = states.get(m.id) or {}
|
|
latest = st.get("latest")
|
|
if latest is None:
|
|
continue
|
|
where = f" on {host_names[m.host_id]}" if m.host_id in host_names else ""
|
|
ids = {"monitor_id": m.id, "host_id": m.host_id}
|
|
if not latest.is_up:
|
|
out.append(finding(
|
|
"critical", "monitor_down", m.name,
|
|
f"{m.type} check of {m.target}{where} is failing: {latest.error_msg or 'down'}",
|
|
since=st.get("down_since") or latest.checked_at, ids=ids,
|
|
next=f'get_monitor(monitor="{m.id}")'))
|
|
elif latest.tls_expires_at is not None:
|
|
days = (latest.tls_expires_at - now).total_seconds() / 86400.0
|
|
if days <= tls["warn"]:
|
|
out.append(finding(
|
|
"critical" if days <= tls["crit"] else "warning", "tls_expiring", m.name,
|
|
f"TLS certificate for {m.target} expires in {days:.1f} days",
|
|
ids=ids, next=f'get_monitor(monitor="{m.id}")'))
|
|
return out
|
|
|
|
|
|
def alert_findings(active) -> list[dict]:
|
|
out = []
|
|
for state, rule in active:
|
|
s = state.state.value
|
|
if s == "firing":
|
|
sev, what = "critical", "is firing"
|
|
elif s == "pending":
|
|
sev, what = "warning", "is pending (breaching, not yet for long enough to fire)"
|
|
elif s == "acknowledged":
|
|
sev, what = "warning", "is acknowledged but still breaching"
|
|
else:
|
|
s, sev, what = "no_data", "warning", "has stopped reporting"
|
|
if state.no_data_since is not None and s != "no_data":
|
|
what += f" and has stopped reporting (no data since {iso(state.no_data_since)})"
|
|
out.append(finding(
|
|
sev, f"alert_{s}", rule.name,
|
|
f"{rule.metric_name} {rule.operator.value} {rule.threshold:g} on "
|
|
f"{rule.source_module}/{rule.resource_name} {what}",
|
|
since=state.no_data_since if s == "no_data" else state.last_transition_at,
|
|
ids={"rule_id": rule.id},
|
|
next=f'get_alert_history(rule="{rule.id}")'))
|
|
return out
|
|
|
|
|
|
_AGENT_CHECKS = (("cpu_pct", "cpu", "CPU"), ("mem_used_pct", "mem", "memory"),
|
|
("disk_used_pct_worst", "disk", "fullest disk"))
|
|
|
|
|
|
def agent_findings(hosts, agent: dict, thresholds: dict) -> list[dict]:
|
|
out = []
|
|
for h in hosts:
|
|
a = agent.get(h.name)
|
|
if not a:
|
|
continue
|
|
ids = {"host_id": h.id}
|
|
if not a.get("fresh"):
|
|
out.append(finding(
|
|
"warning", "agent_stale", h.name,
|
|
"host agent has stopped reporting; its metrics are out of date",
|
|
since=a.get("last_at"), ids=ids, next=f'get_host(host="{h.id}")'))
|
|
continue # stale values would only add noise
|
|
for metric, kind, label in _AGENT_CHECKS:
|
|
value = a.get(metric)
|
|
th = thresholds.get(kind) or {}
|
|
if value is None or th.get("warn") is None:
|
|
continue
|
|
if th.get("crit") is not None and value >= th["crit"]:
|
|
sev = "critical"
|
|
elif value >= th["warn"]:
|
|
sev = "warning"
|
|
else:
|
|
continue
|
|
out.append(finding(
|
|
sev, f"host_{kind}_high", h.name,
|
|
f"{label} at {rnd(value, 1)}% (warn {th['warn']}%, crit {th.get('crit')}%)",
|
|
ids=ids, next=f'get_host_metrics(host="{h.id}", metric="{metric}")'))
|
|
return out
|
|
|
|
|
|
def run_findings(runs) -> list[dict]:
|
|
out = []
|
|
for r in runs:
|
|
failures = ((r.results or {}).get("failures") or [])[:3]
|
|
detail = f": {'; '.join(str(f)[:200] for f in failures)}" if failures else ""
|
|
out.append(finding(
|
|
"warning", f"ansible_run_{r.status.value}", r.playbook_path,
|
|
f"Ansible run of {r.playbook_path} ({r.source_name}) {r.status.value}{detail}",
|
|
since=r.finished_at or r.started_at, ids={"run_id": r.id}))
|
|
return out
|
|
|
|
|
|
def anomaly_findings(open_now) -> list[dict]:
|
|
"""Metric anomalies going on now (#5740): a warning each."""
|
|
from steward.core.formatting import metric_label, metric_value
|
|
out = []
|
|
for a in open_now:
|
|
out.append(finding(
|
|
"warning", "metric_anomaly", f"{a.resource_name} {metric_label(a.metric_name)}",
|
|
f"Metric anomaly: {metric_label(a.metric_name)} on {a.resource_name} held "
|
|
f"{metric_value(a.metric_name, a.peak_level)} where "
|
|
f"{metric_value(a.metric_name, a.usual_hold)} is usual for that hour"
|
|
+ (f" ({a.occurrences} times since {iso(a.started_at)})" if a.occurrences > 1 else ""),
|
|
since=a.last_started_at, ids={"anomaly_id": a.id},
|
|
next=f'list_anomalies(review="{a.review}")'))
|
|
return out
|
|
|
|
|
|
async def _plugin_findings(server) -> tuple[list[dict], list[str]]:
|
|
found, unavailable = [], []
|
|
for plugin, source in sorted((getattr(server, "overview_sources", None) or {}).items()):
|
|
if not server.plugin_enabled(plugin):
|
|
continue
|
|
try:
|
|
async with db_session() as db:
|
|
found.extend(await source(db))
|
|
except Exception:
|
|
logger.exception("MCP overview source of plugin %r failed", plugin)
|
|
unavailable.append(plugin)
|
|
return found, unavailable
|
|
|
|
|
|
async def get_overview() -> dict:
|
|
"""What is wrong right now, across everything Steward watches. Start here.
|
|
|
|
Returns `findings` ranked critical first, then longest-standing first:
|
|
failing monitors (with the error and since when), firing, pending and
|
|
still-breaching acknowledged alerts, TLS certificates near expiry, host
|
|
agents that stopped reporting, hosts over their CPU, memory or disk
|
|
limits (all cutoffs are set under Settings → Thresholds), Ansible runs that failed
|
|
in the last 24h, metrics going on now well above what is normal for them
|
|
at this hour (metric anomalies), and anything enabled plugins add
|
|
(unhealthy containers, for one). Each finding carries the ids to follow
|
|
and `next`, the call that shows its evidence.
|
|
|
|
`counts` sizes the fleet; `counts.anomalies_to_review` is how many metric
|
|
anomalies wait for someone to mark them normal or real (list_anomalies).
|
|
`active_maintenance` names windows currently silencing alerts.
|
|
`all_clear` is true when nothing is wrong.
|
|
"""
|
|
app = mcp_app()
|
|
now = now_utc()
|
|
async with db_session() as db:
|
|
hosts = list((await db.execute(select(Host).order_by(Host.name))).scalars())
|
|
host_names = {h.id: h.name for h in hosts}
|
|
monitors = list((await db.execute(select(Monitor).order_by(Monitor.name))).scalars())
|
|
states = await monitor_svc.monitor_states(db, [m for m in monitors if m.enabled])
|
|
active = await alert_svc.active_alerts(db)
|
|
agent = await host_svc.agent_overview_by_host(
|
|
db, [h.name for h in hosts], host_svc.agent_stale_after_seconds(app.config))
|
|
runs = list((await db.execute(
|
|
select(AnsibleRun)
|
|
.where(AnsibleRun.status.in_((AnsibleRunStatus.failed, AnsibleRunStatus.interrupted)),
|
|
AnsibleRun.started_at >= now - timedelta(hours=24))
|
|
.order_by(AnsibleRun.started_at.desc())
|
|
.limit(20)
|
|
)).scalars())
|
|
windows = await alert_svc.list_maintenance_windows(db)
|
|
open_anomalies = await anom.open_anomalies(db)
|
|
to_review = await anom.to_review_count(db)
|
|
|
|
thresholds = app.config.get("THRESHOLDS") or {}
|
|
findings = (monitor_findings(monitors, states, host_names, now, thresholds.get("tls_days"))
|
|
+ alert_findings(active)
|
|
+ agent_findings(hosts, agent, thresholds)
|
|
+ run_findings(runs)
|
|
+ anomaly_findings(open_anomalies))
|
|
plugin_found, unavailable = await _plugin_findings(getattr(app, "mcp_server", None))
|
|
findings = rank(findings + plugin_found)
|
|
|
|
down = sum(1 for m in monitors if m.enabled and (st := states.get(m.id))
|
|
and st["latest"] is not None and not st["latest"].is_up)
|
|
out = {
|
|
"generated_at": iso(now),
|
|
"all_clear": not findings,
|
|
"counts": {
|
|
"hosts": len(hosts),
|
|
"monitors": len(monitors),
|
|
"monitors_down": down,
|
|
"alerts_active": len(active),
|
|
"critical": sum(1 for f in findings if f["severity"] == "critical"),
|
|
"warning": sum(1 for f in findings if f["severity"] == "warning"),
|
|
"anomalies_to_review": to_review,
|
|
},
|
|
"active_maintenance": [
|
|
{"window_id": w.id, "name": w.name, "end_at": iso(w.end_at),
|
|
"scope": {"source_module": w.scope_source, "resource": w.scope_resource}}
|
|
for w in windows if w.start_at <= now <= w.end_at
|
|
],
|
|
"findings": findings[:FINDINGS_CAP],
|
|
"truncated": len(findings) > FINDINGS_CAP,
|
|
}
|
|
if out["truncated"]:
|
|
out["more"] = (f"{len(findings) - FINDINGS_CAP} lower-ranked findings left out; "
|
|
"list_monitors(state=\"down\"), list_alerts and list_hosts show everything.")
|
|
if unavailable:
|
|
out["unavailable"] = [f"{p} plugin's findings could not be read (see Steward's log)"
|
|
for p in unavailable]
|
|
return out
|
|
|
|
|
|
def register(register_tool) -> None:
|
|
register_tool(get_overview, role=UserRole.viewer, read_only=True)
|