"""Metric anomaly tools (#5740): metrics that left what is normal for them at that hour, and marking each normal or real.""" from __future__ import annotations from datetime import timedelta from sqlalchemy import select from steward.core import anomalies as anom from steward.core import metric_series from steward.core.norms import LEARNED, norms_at, slot_of from steward.core.settings import (DEFAULTS, NORM_FLOORS, floor_of, get_all_settings, set_setting, to_thresholds_cfg) from steward.mcp._context import current_caller, mcp_app from steward.models.norms import MetricAnomaly, MetricAnomalySpan, MetricNorm from steward.models.users import UserRole from . import _events as events from ._audit import audit_mcp, mcp_username from ._shape import bounded, clamp_limit, db_session, iso, now_utc, rnd LIST_MAX = 200 REVIEW_MAX = 100 # anomalies one review_anomaly call may mark def _row(v: anom.AnomalyView) -> dict: a, last = v.anomaly, v.last_span return { "anomaly_id": a.id, "source_module": a.source_module, "resource": a.resource_name, "metric": a.metric_name, "review": a.review, "ongoing": v.ongoing, "started_at": iso(a.started_at), "ended_at": iso(a.ended_at), "occurrences": a.occurrences, "last_occurrence": {"started_at": iso(last.started_at), "ended_at": iso(last.ended_at), "peak_level": rnd(last.peak_level, 3)} if last else None, "peak_level": rnd(a.peak_level, 3), "peak_at": iso(a.peak_at), "usual_hold": rnd(a.usual_hold, 3), "usual_peak": rnd(a.usual_peak, 3), "reviewed_by": a.reviewed_by, "reviewed_at": iso(a.reviewed_at), "note": a.note, "page": v.href, } async def list_anomalies(review: str = "to_review", limit: int = 50) -> dict: """Metric anomalies: a learned metric whose 15-minute level went well above what is normal for it at that hour (its usual hold ×1.25, and by more than the metric's floor). review: "to_review" (default) — waiting for someone to mark them; "normal" — accepted, their hours are learned; "real" — a real problem, not learned; "lapsed" — ended and nobody reviewed them within the review window (Settings → Thresholds), so they are not learned; "" for all of them. One row covers every repeat on a resource and metric while it waits for review: `occurrences` counts them and `last_occurrence` is the latest. `peak_level` is the highest 15-minute level any occurrence held, against `usual_hold` (the level that hour usually holds) and `usual_peak` (how high it usually goes), all in the metric's own unit. `ongoing` is true while an occurrence is still going. To-review rows come first, newest occurrence first. review_anomaly marks one. """ if review and review not in anom.REVIEWS: raise ValueError(f"review must be one of {', '.join(anom.REVIEWS)}, or empty for all") limit = clamp_limit(limit, 50, LIST_MAX) async with db_session() as db: rows = await anom.list_anomalies(db, (review,) if review else None, limit=LIST_MAX) views = await anom.views(db, rows) to_review = await anom.to_review_count(db) out = bounded([_row(v) for v in views], limit, key="anomalies", more="Pass a larger `limit`, or a `review` state.") out["to_review"] = to_review return out async def review_anomaly(anomaly_ids: list[str] | str, review: str, note: str = "") -> dict: """Mark metric anomalies "normal" or "real": one id, or several judged the same way (list_anomalies gives ids; get_anomaly shows the evidence). normal: this level is usual for this resource at that time; its hours are learned into the metric's normal, so it stops being flagged once learned. real: it was a real problem; its hours stay out of what is learned. note: say why (what it was, what you checked); kept with each review, so the next look at this resource and metric sees the precedent. A lapsed anomaly may still be marked. All or nothing: an unknown id marks none. Audited as " via MCP", one row per anomaly. """ ids = [anomaly_ids] if isinstance(anomaly_ids, str) else list(anomaly_ids or []) ids = list(dict.fromkeys(i.strip() for i in ids if i and i.strip())) if not ids: raise ValueError("Name at least one anomaly id; list_anomalies shows them.") if len(ids) > REVIEW_MAX: raise ValueError(f"At most {REVIEW_MAX} anomalies at once.") caller = current_caller() marked = [] async with db_session() as db, db.begin(): for anomaly_id in ids: try: a = await anom.mark(db, anomaly_id, review, mcp_username(caller.username), note) except LookupError: raise ValueError(f"No anomaly {anomaly_id!r}; list_anomalies shows them. " "Nothing was marked.") from None marked.append(({"source": a.source_module, "resource": a.resource_name, "metric": a.metric_name, "note": a.note}, {"anomaly_id": a.id, "resource": a.resource_name, "metric": a.metric_name, "review": a.review})) reviewed_by, reviewed_at, kept = a.reviewed_by, a.reviewed_at, a.note for (detail, row) in marked: await audit_mcp(f"metric_anomaly.marked_{review}", "metric_anomaly", row["anomaly_id"], detail) return {"marked": [row for _d, row in marked], "review": review, "reviewed_by": reviewed_by, "reviewed_at": iso(reviewed_at), "note": kept} # ── One anomaly, understood ───────────────────────────────────────────────── AROUND = timedelta(minutes=30) # what else was going on: this long either side of an occurrence SPANS_MAX = 20 async def _norm_map(db, key: tuple[str, str, str]) -> tuple[str | None, dict[int, MetricNorm]]: """The key's learned slots, {slot: row}, of the kind it uses.""" rows = list((await db.execute(select(MetricNorm).where( MetricNorm.source_module == key[0], MetricNorm.resource_name == key[1], MetricNorm.metric_name == key[2]))).scalars()) kind = "week" if any(r.kind == "week" for r in rows) else "day" if rows else None return kind, {r.slot: r for r in rows if r.kind == kind} def _usual_at(kind, slots: dict, at) -> MetricNorm | None: return slots.get(slot_of(at, kind)) if kind else None def series_pad(started_at, ended_at, now) -> timedelta: """How far either side of an occurrence its series reaches: as long as the occurrence, at least an hour and at most six.""" held = (ended_at or now) - started_at return min(max(held, timedelta(hours=1)), timedelta(hours=6)) async def get_anomaly(anomaly_id: str, occurrences: int = 5) -> dict: """One metric anomaly with what is needed to judge it: each occurrence against what is usual at that hour, the metric around it, what else was going on, and how earlier ones on the same metric were judged. occurrences: how many of its occurrences to show, newest first (default 5, max 20). - `anomaly`: the list_anomalies row, including `usual_hold` and `usual_peak` as they stood when it was first flagged. - `place`: the host the resource lives on (`host`, `host_id`), when it has one: an agent series, a container, a monitor pinned to a host. - `occurrences[]`: `started_at`, `ended_at`, `held_s`, `peak_level`, and `usual` for that hour as Steward has learned it now (`usual_hold`, `usual_peak`, `flag_above`), plus `around`: alert transitions, Ansible runs, other anomalies on the host, and what plugins saw (container starts, deaths and OOM kills; maintenance jobs; reboots and agent stops), from 30 minutes before to 30 minutes after; and `anomalies_elsewhere`, how many anomalies on other hosts overlapped it. Many at once points at something shared (network, storage, Steward). - `series`: the newest shown occurrence with time either side (as long as it lasted, 1 to 6 hours), as [time, average, usual_hold at that hour] points. - `precedent`: earlier anomalies on this resource and metric, how they were judged and the note left. Judge it with review_anomaly: "normal" when it is how this resource behaves at that time (its hours are then learned), "real" when it was a problem. Say why in the note. """ limit = clamp_limit(occurrences, 5, SPANS_MAX) now = now_utc() async with db_session() as db: a = await db.get(MetricAnomaly, (anomaly_id or "").strip()) if a is None: raise ValueError(f"No anomaly {anomaly_id!r}; list_anomalies shows them.") [view] = await anom.views(db, [a]) key = (a.source_module, a.resource_name, a.metric_name) spans = list((await db.execute( select(MetricAnomalySpan).where(MetricAnomalySpan.anomaly_id == a.id) .order_by(MetricAnomalySpan.started_at.desc()))).scalars()) kind, slots = await _norm_map(db, key) learned = LEARNED.get((a.source_module, a.metric_name)) floor = learned.least_departure if learned else None place = await events.place_of(db, a.source_module, a.resource_name) shown = [] for span in spans[:limit]: usual = _usual_at(kind, slots, span.started_at) near, elsewhere = await events.around(db, place, a.id, span.started_at - AROUND, (span.ended_at or now) + AROUND) shown.append({ "started_at": iso(span.started_at), "ended_at": iso(span.ended_at), "ongoing": span.ended_at is None, "held_s": int(((span.ended_at or now) - span.started_at).total_seconds()), "peak_level": rnd(span.peak_level, 3), "usual": ({"slot": slot_label(kind, usual.slot), "usual_hold": rnd(usual.usual_hold, 3), "usual_peak": rnd(usual.usual_peak, 3), "flag_above": rnd(flag_above(usual.usual_hold, floor), 3) if floor is not None else None} if usual else None), "around": near, "anomalies_elsewhere": elsewhere, }) series = None if spans: newest = spans[0] pad = series_pad(newest.started_at, newest.ended_at, now) frm, to = newest.started_at - pad, min((newest.ended_at or now) + pad, now) pts = await metric_series.points(db, *key, frm, target=60, until=to) series = {"from": iso(frm), "to": iso(to), "columns": ["at", "value", "usual_hold"], "points": [[iso(t), rnd(v, 3), rnd(u.usual_hold, 3) if (u := _usual_at(kind, slots, t)) else None] for t, v in pts]} earlier = list((await db.execute( select(MetricAnomaly).where( MetricAnomaly.source_module == a.source_module, MetricAnomaly.resource_name == a.resource_name, MetricAnomaly.metric_name == a.metric_name, MetricAnomaly.id != a.id) .order_by(MetricAnomaly.last_started_at.desc()).limit(10))).scalars()) return { "anomaly": _row(view), "place": {"host": place.host_name, "host_id": place.host_id}, "floor": rnd(floor, 3) if floor is not None else None, "depart_factor": anom.DEPART, "occurrences": shown, "occurrences_total": len(spans), "truncated": len(spans) > limit, "series": series, "precedent": [{"anomaly_id": e.id, "review": e.review, "note": e.note, "reviewed_by": e.reviewed_by, "started_at": iso(e.started_at), "ended_at": iso(e.ended_at), "occurrences": e.occurrences, "peak_level": rnd(e.peak_level, 3), "usual_hold": rnd(e.usual_hold, 3)} for e in earlier], "next": (f'get_metric_normal(source_module="{a.source_module}", resource="{a.resource_name}", ' f'metric="{a.metric_name}")'), } # ── The learned normal ────────────────────────────────────────────────────── _DAYS = ("Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun") def slot_label(kind: str, slot: int) -> str: """A slot as a reader names it: "03:00 UTC", or "Sun 03:00 UTC".""" if kind == "week": return f"{_DAYS[slot // 24]} {slot % 24:02d}:00 UTC" return f"{slot:02d}:00 UTC" def flag_above(norm_hold: float, floor: float) -> float: """The 15-minute level that would open an anomaly in a slot whose usual hold is `norm_hold` (anomalies.departs).""" return max(norm_hold * anom.DEPART, norm_hold + floor) def _learned_or_refuse(source: str, metric: str): learned = LEARNED.get((source, metric)) if learned is None: known = ", ".join(sorted(f"{s}/{m}" for s, m in LEARNED)) raise ValueError(f"{source}/{metric} doesn't learn a normal. These do: {known}.") return learned async def get_metric_normal(source_module: str, resource: str, metric: str) -> dict: """What Steward has learned is normal for one resource's metric, slot by slot: the yardstick an anomaly is judged against. source_module, resource, metric: as list_anomalies or query_metrics name them (e.g. "host_agent", "VDNT-Docker03", "cpu_pct"). `kind` is "day" (24 UTC hours, once 5 days of history exist) or "week" (168 hours from Monday 00:00 UTC, once 21 days do). Each slot gives `usual_hold` (the level its hours usually held: each hour's p95, at the 90th percentile across past days), `usual_peak` (how high they usually went), `days_seen`, and `flag_above`: the 15-minute average that would open an anomaly there (usual hold ×1.25, and at least `floor` above it). `floor` is the smallest departure worth calling an anomaly, and `floor_setting` the Settings → Thresholds floor it comes from. Hours an anomaly covers are learned only once someone marks it normal. `learned: false` with no slots means too little history yet; the fixed cutoffs judge it meanwhile. """ learned = _learned_or_refuse(source_module, metric) async with db_session() as db: rows = list((await db.execute( select(MetricNorm) .where(MetricNorm.source_module == source_module, MetricNorm.resource_name == resource, MetricNorm.metric_name == metric) .order_by(MetricNorm.kind, MetricNorm.slot))).scalars()) now = now_utc() current = (await norms_at(db, [(source_module, resource, metric)], now)).get( (source_module, resource, metric)) floor = learned.least_departure kinds = {r.kind for r in rows} kind = "week" if "week" in kinds else "day" if rows else None slots = [{"slot": r.slot, "label": slot_label(r.kind, r.slot), "usual_hold": rnd(r.usual_hold, 3), "usual_peak": rnd(r.usual_peak, 3), "days_seen": r.days_seen, "flag_above": rnd(flag_above(r.usual_hold, floor), 3)} for r in rows if r.kind == kind] out = { "source_module": source_module, "resource": resource, "metric": metric, "learned": bool(slots), "kind": kind, "floor": rnd(floor, 3), "floor_setting": learned.floor_name, "only_while": ({"metric": learned.gate[0], "at_least": rnd(learned.gate[1]() if callable(learned.gate[1]) else learned.gate[1], 3)} if learned.gate else None), "now": ({"slot": current.slot, "label": slot_label(current.kind, current.slot), "usual_hold": rnd(current.usual_hold, 3), "usual_peak": rnd(current.usual_peak, 3), "flag_above": rnd(flag_above(current.usual_hold, floor), 3)} if current else None), "slots": slots, "updated_at": iso(max((r.updated_at for r in rows), default=None)), } return out # ── Sensitivity (Settings → Thresholds) ───────────────────────────────────── def _floors_view(thresholds: dict) -> list[dict]: uses: dict[str, list[str]] = {} for (source, metric), learned in sorted(LEARNED.items()): if learned.floor_name: uses.setdefault(learned.floor_name, []).append(f"{source}/{metric}") return [{"name": name, "value": floor_of(thresholds, name), "default": float(DEFAULTS[f"norms.floor.{name}"]), "metrics": uses.get(name, [])} for name in NORM_FLOORS] async def get_anomaly_settings() -> dict: """How sensitive anomaly detection is, as Settings → Thresholds sets it. `review_days`: an ended anomaly nobody reviewed within this many days lapses (leaves the to-review list, kept as not reviewed, never learned). `floors`: per kind of metric, the smallest departure from its normal worth calling an anomaly, in the metric's own unit (% points, ms), with the metrics each one applies to. A level must also be ×`depart_factor` its slot's usual hold, averaged over `window_minutes`. Raising a floor flags less; lowering it flags more. set_anomaly_settings (admin) changes the review window and floors. """ async with db_session() as db: settings = await get_all_settings(db) return {"review_days": int(settings.get("anomalies.review_days") or 14), "depart_factor": anom.DEPART, "window_minutes": int(anom.WINDOW.total_seconds() // 60), "floors": _floors_view(to_thresholds_cfg(settings))} async def set_anomaly_settings(review_days: int = 0, floors: dict[str, float] | None = None) -> dict: """Change how sensitive anomaly detection is, as Settings → Thresholds does. Takes effect on the next minute's pass; audited. review_days: days an ended anomaly waits for review before it lapses (at least 1); 0 leaves it as it is. floors: {name: value} for any of get_anomaly_settings' floor names, in that floor's unit, 0 or more. A floor not named is left as it is. Returns the settings as they now stand. """ floors = dict(floors or {}) unknown = sorted(set(floors) - set(NORM_FLOORS)) if unknown: raise ValueError(f"No floor named {', '.join(unknown)}; they are {', '.join(NORM_FLOORS)}.") changes: dict = {} for name, value in floors.items(): try: value = float(value) except (TypeError, ValueError): raise ValueError(f"Floor {name} must be a number.") from None if value < 0 or value != value: raise ValueError(f"Floor {name} must be 0 or more.") changes[f"norms.floor.{name}"] = value if review_days: if int(review_days) < 1: raise ValueError("review_days is at least 1.") changes["anomalies.review_days"] = int(review_days) if not changes: raise ValueError("Nothing to change: pass review_days or floors.") app = mcp_app() async with db_session() as db: async with db.begin(): before = await get_all_settings(db) for key, value in changes.items(): await set_setting(db, key, value) fresh = await get_all_settings(db) app.config["THRESHOLDS"] = to_thresholds_cfg(fresh) await audit_mcp("settings.saved", "settings", "anomalies", {"section": "anomalies", "changes": {k: {"from": before.get(k), "to": v} for k, v in changes.items()}}) return await get_anomaly_settings() def register(register_tool) -> None: for fn in (list_anomalies, get_anomaly, get_metric_normal, get_anomaly_settings): register_tool(fn, role=UserRole.viewer, read_only=True) register_tool(review_anomaly, role=UserRole.operator, read_only=False) register_tool(set_anomaly_settings, role=UserRole.admin, read_only=False)