diff --git a/docs/architecture.md b/docs/architecture.md index 7779cdf..fee63c6 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -101,6 +101,16 @@ Each rolled-up hour keeps three figures: A two-minute spike raises the peak and leaves the p95 low; an hour spent at that level raises both. Learned normals use this to tell a recurring spike from a held level. Hours rolled up before the p95 was kept have it NULL; read those as their average. +### Normals and anomalies + +A metric a plugin registers with `learn_normal()` gets a normal per hour of the day, then per hour of the week once three weeks of history exist (`metric_norms`, relearned hourly by `metric_norms`; `steward/core/norms.py`). + +Every minute, `metric_anomalies` averages each learned key's last 15 minutes and compares it with the normal for that hour. A level above the usual hold ×1.25, and above the metric's floor, opens an anomaly (`steward/core/anomalies.py`): +- `metric_anomalies` holds one row per item to review. A repeat while the row still waits joins it as another occurrence (`metric_anomaly_spans`). +- A row is `to_review` until someone marks it `normal` or `real` (Alerts → Anomalies, or MCP `review_anomaly`). An ended row nobody reviews within `anomalies.review_days` (14) becomes `lapsed`. +- Learning skips every hour an occurrence covers unless its anomaly was marked `normal`, so a new level is never learned without review. +- `data_cleanup` prunes reviewed and lapsed rows at `data.retention_days`; a row still to review is never pruned. + --- ## Database Session Pattern diff --git a/docs/core/mcp.md b/docs/core/mcp.md index f4411fb..e9300a5 100644 --- a/docs/core/mcp.md +++ b/docs/core/mcp.md @@ -75,6 +75,7 @@ Each role includes the tools of the roles above it in this list. | Hosts and metrics | `list_hosts`, `get_host`, `get_host_metrics`, `query_metrics` | | Monitors | `list_monitors`, `get_monitor` | | Alerts | `list_alerts`, `get_alert_history`, `list_alert_rules`, `get_alert_rule_options`, `list_maintenance_windows` | +| Metric anomalies | `list_anomalies` (metrics that left their normal for the hour, to review first) | | Automation | `list_playbooks`, `get_playbook`, `list_inventory`, `list_ansible_runs`, `get_ansible_run`, `wait_for_run`, `list_schedules` | | Audit | `list_audit_events` | | Docker plugin | `list_containers`, `get_container`, `list_swarm_services`, `search_container_logs`, `get_docker_disk_usage` | @@ -87,6 +88,7 @@ Each role includes the tools of the roles above it in this list. | Automation | `run_playbook`, `cancel_ansible_run`, `create_schedule`, `update_schedule`, `delete_schedule`, `update_group_members` (put hosts into an inventory group or take them out; a group alert rule follows within a minute) | | Monitors | `create_monitor`, `update_monitor`, `set_monitor_enabled` | | Alerts | `create_alert_rule`, `update_alert_rule`, `acknowledge_alert`, `send_test_alert`, `create_maintenance_window`, `end_maintenance_window` | +| Metric anomalies | `review_anomaly` (mark one normal, so its hours are learned, or real) | ### Admin: destructive and remediation diff --git a/steward/alerts/routes.py b/steward/alerts/routes.py index 4457968..cf73d44 100644 --- a/steward/alerts/routes.py +++ b/steward/alerts/routes.py @@ -2,6 +2,7 @@ from __future__ import annotations from datetime import datetime, timezone from quart import Blueprint, render_template, request, redirect, url_for, current_app, session from steward.core.audit import log_audit +from steward.core.settings import get_setting from steward.core import outbox from sqlalchemy import func, select from steward.auth.middleware import require_role @@ -279,6 +280,58 @@ async def alert_history(): return await render_template("alerts/history.html", **ctx) +@alerts_bp.get("/anomalies") +@require_role(UserRole.viewer) +async def anomalies(): + """Metric anomalies (#5740): those to review, newest occurrence first, + then the reviewed and lapsed ones under day dividers.""" + from steward.audit.service import mark_day_groups + from steward.core import anomalies as anom + from steward.core import formatting as fmt + async with current_app.db_sessionmaker() as db: + to_review = await anom.views(db, await anom.list_anomalies(db, (anom.TO_REVIEW,))) + past = await anom.views(db, await anom.list_anomalies( + db, (anom.NORMAL, anom.REAL, anom.LAPSED), limit=PAGE_SIZE * 4)) + days = int(await get_setting(db, "anomalies.review_days") or 14) + tz = fmt.zone(session.get("tz")) + to_review = [{"v": v} for v in to_review] + past = [{"v": v} for v in past] + at = lambda item: item["v"].anomaly.last_started_at # noqa: E731 + mark_day_groups(to_review, tz, at=at) + mark_day_groups(past, tz, at=at) + return await render_template("alerts/anomalies.html", to_review=to_review, past=past, + review_days=days) + + +@alerts_bp.post("/anomalies//review") +@require_role(UserRole.operator) +async def review_anomaly(anomaly_id: str): + """Mark an anomaly normal (its hours are learned) or real (they are not), + with an optional note. Audited.""" + from steward.core import anomalies as anom + form = await request.form + review = form.get("review", "") + username = session.get("username", "") + async with current_app.db_sessionmaker() as db: + async with db.begin(): + try: + a = await anom.mark(db, anomaly_id, review, username, form.get("note")) + except LookupError: + return "Not found", 404 + except ValueError as exc: + return f"Bad request: {exc}", 400 + detail = {"source": a.source_module, "resource": a.resource_name, + "metric": a.metric_name, "note": a.note} + await log_audit(current_app, session.get("user_id"), username, + f"metric_anomaly.marked_{review}", entity_type="metric_anomaly", + entity_id=anomaly_id, detail=detail) + back = form.get("next", "") + # Only back to a page of this site. + if not back.startswith("/") or back.startswith("//"): + back = url_for("alerts.anomalies") + return redirect(back) + + # How many of one rule's transitions the history verdict reads ("fired N # times, X in all"); a rule that flapped more often than this in the range # is summed over its newest ones. diff --git a/steward/alerts/service.py b/steward/alerts/service.py index f4ffd7f..2f28237 100644 --- a/steward/alerts/service.py +++ b/steward/alerts/service.py @@ -104,13 +104,17 @@ PLUGIN_PAGES = {"traefik": "/plugins/traefik/", "unifi": "/plugins/unifi/", async def resource_url(db, rule: AlertRule) -> str | None: - """Where the thing a rule watches is shown, or None when nothing shows it. + """Where the thing a rule watches is shown, or None when nothing shows it.""" + return await resource_url_for(db, rule.source_module, rule.resource_name) + + +async def resource_url_for(db, source: str, resource: str) -> str | None: + """Where a (source, resource) of the metric store is shown, or None. host_agent resources are the host's name, or ":"; docker's are "/"; a monitor's is its name under its type. """ from steward.models.hosts import Host - source, resource = rule.source_module, rule.resource_name if source == "host_agent": name = resource.split(":", 1)[0] host_id = (await db.execute( diff --git a/steward/app.py b/steward/app.py index 44500a2..335a941 100644 --- a/steward/app.py +++ b/steward/app.py @@ -449,6 +449,10 @@ def _register_core_tasks(app: Quart) -> None: from .core.norms import run_norms await run_norms(app) + async def run_metric_anomalies(): + from .core.anomalies import run_anomalies + await run_anomalies(app) + async def run_email_outbox(): from .core.outbox import send_due await send_due(app) @@ -520,6 +524,14 @@ def _register_core_tasks(app: Quart) -> None: interval_seconds=3600, run_on_startup=False, ), + # Each learned metric's last 15 minutes against its normal (#5740): + # opens, extends and ends anomalies, and lapses unreviewed ones. + ScheduledTask( + name="metric_anomalies", + coro_factory=run_metric_anomalies, + interval_seconds=60, + run_on_startup=False, + ), ]) # Pull Ansible git sources as each falls due. One task reads the sources diff --git a/steward/audit/service.py b/steward/audit/service.py index b627825..c4e8b98 100644 --- a/steward/audit/service.py +++ b/steward/audit/service.py @@ -265,13 +265,15 @@ def detail_facts(detail: dict, scopes: dict[str, str]) -> list[dict]: return facts -def mark_day_groups(items: list[dict], tz, group: str = "", now=None) -> str: +def mark_day_groups(items: list[dict], tz, group: str = "", now=None, + at=lambda item: item["event"].timestamp) -> str: """Give each newest-first entry its recency group (formatting.day_group, in the reader's zone) and whether it opens one, so the page draws a divider there. `group` is the one the previous page ended in: a group - split across pages is opened once. Returns the group this page ends in.""" + split across pages is opened once. `at` reads an entry's moment. + Returns the group this page ends in.""" for item in items: - item["group"] = day_group(item["event"].timestamp, tz, now) + item["group"] = day_group(at(item), tz, now) item["opens_group"] = item["group"] != group group = item["group"] return group diff --git a/steward/backup/contents.py b/steward/backup/contents.py index 3a16df0..3e4c5a5 100644 --- a/steward/backup/contents.py +++ b/steward/backup/contents.py @@ -25,6 +25,9 @@ KEPT: frozenset[str] = frozenset({ "unifi_known_clients", # History kept by the ruling. "audit_events", "alert_events", + # Metric anomalies and their reviews (#5740): what someone marked normal + # or real is a decision, kept like alert history. + "metric_anomalies", "metric_anomaly_spans", }) # Rows that are measurements, logs or in-flight work: left out. diff --git a/steward/core/anomalies.py b/steward/core/anomalies.py new file mode 100644 index 0000000..69dd6b5 --- /dev/null +++ b/steward/core/anomalies.py @@ -0,0 +1,316 @@ +"""A metric that departs from its normal is an anomaly to review (#5740). + +Every minute the last WINDOW of each learned key is averaged and compared +with the normal for the slot it falls in (steward.core.norms). A level above +the slot's usual hold by DEPART times, and by at least the metric's floor, +opens an occurrence; the occurrence ends when the level comes back. + +What the operator ruled (System 182 "Metric Norms and Anomalies"): +- A departure is a warning, marked as a metric anomaly, and stays "to review" + after it ends until someone marks it normal (its hours are learned) or real + (they are not). +- The list never becomes a backlog. An occurrence while the key's earlier + anomaly is still to review joins it, so one row covers every repeat. An + ended anomaly nobody reviewed within anomalies.review_days lapses: it leaves + the to-review list and stays in history as not reviewed. A lapsed anomaly + is not learned, as no normal is learned without review. +""" +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone + +from sqlalchemy import ARRAY, String, bindparam, case, delete, func, or_, select, text, update + +from steward.core.norms import LEARNED, Norm, norms_at +from steward.models.norms import MetricAnomaly, MetricAnomalySpan + +WINDOW = timedelta(minutes=15) # the level judged is this window's average +DEPART = 1.25 # times the slot's usual hold that counts as a departure +# An occurrence that comes back within one window of ending continues the +# same occurrence: a level wobbling around the line is one event, not ten. +REJOIN = WINDOW + +TO_REVIEW, NORMAL, REAL, LAPSED = "to_review", "normal", "real", "lapsed" +REVIEWS = (TO_REVIEW, NORMAL, REAL, LAPSED) +MARKS = (NORMAL, REAL) + + +def departs(level: float, norm: Norm, floor: float) -> bool: + """Whether `level` is out of `norm`: above its usual hold by DEPART times + and by at least `floor`, so a 0.1% → 0.3% share is never news.""" + return level > norm.usual_hold * DEPART and level - norm.usual_hold >= floor + + +# Each learned key reporting in the window, with the window's average. Keys +# come from plugin_metric_latest (one row each), and each average is a range +# read of the series index, so the pass never scans the raw table. +_LEVELS = text(""" +SELECT l.source_module, l.resource_name, l.metric_name, v.level +FROM plugin_metric_latest l +CROSS JOIN LATERAL ( + SELECT avg(p.value) AS level FROM plugin_metrics p + WHERE p.source_module = l.source_module AND p.resource_name = l.resource_name + AND p.metric_name = l.metric_name AND p.recorded_at >= :since + AND p.recorded_at <= :now +) v +WHERE l.source_module || '/' || l.metric_name = ANY(:keys) + AND l.recorded_at >= :since AND v.level IS NOT NULL +""").bindparams(bindparam("keys", type_=ARRAY(String))) + + +async def levels_now(session, now: datetime) -> dict[tuple[str, str, str], float]: + """{(source, resource, metric): its average over the last WINDOW}.""" + keys = [f"{s}/{m}" for s, m in LEARNED] + if not keys: + return {} + rows = await session.execute(_LEVELS, {"keys": keys, "since": now - WINDOW, "now": now}) + return {(s, r, m): float(level) for s, r, m, level in rows} + + +async def detect(session, now: datetime | None = None) -> dict: + """One pass: open, extend and close occurrences. Runs inside the caller's + transaction. Returns {"opened", "closed"} counts.""" + now = now or datetime.now(timezone.utc) + levels = await levels_now(session, now) + norms = await norms_at(session, list(levels), now) + open_now = { + (a.source_module, a.resource_name, a.metric_name): (span, a) + for span, a in (await session.execute( + select(MetricAnomalySpan, MetricAnomaly) + .join(MetricAnomaly, MetricAnomaly.id == MetricAnomalySpan.anomaly_id) + .where(MetricAnomalySpan.ended_at.is_(None)))).all() + } + opened = closed = 0 + for key, level in levels.items(): + norm = norms.get(key) + learned = LEARNED.get((key[0], key[2])) + out = norm is not None and learned is not None and departs(level, norm, learned.floor) + current = open_now.pop(key, None) + if current is not None: + span, anomaly = current + if out: + _raise_peak(anomaly, span, level, norm, now) + else: + _close(anomaly, span, now) + closed += 1 + elif out: + await _occur(session, key, level, norm, now) + opened += 1 + # An open occurrence whose key stopped reporting, or lost its normal, + # can't be judged any more: it ends now. + for span, anomaly in open_now.values(): + _close(anomaly, span, now) + closed += 1 + return {"opened": opened, "closed": closed} + + +def _raise_peak(anomaly: MetricAnomaly, span: MetricAnomalySpan, level: float, + norm: Norm, now: datetime) -> None: + span.peak_level = max(span.peak_level, level) + if level > anomaly.peak_level: + anomaly.peak_level, anomaly.peak_at = level, now + anomaly.usual_hold, anomaly.usual_peak = norm.usual_hold, norm.usual_peak + + +def _close(anomaly: MetricAnomaly, span: MetricAnomalySpan, now: datetime) -> None: + span.ended_at = now + anomaly.ended_at = now + + +async def _occur(session, key, level: float, norm: Norm, now: datetime) -> None: + """A departure on a key with no open occurrence: join the key's anomaly + still to review, or open a new one.""" + source, resource, metric = key + start = now - WINDOW + anomaly = (await session.execute( + select(MetricAnomaly) + .where(MetricAnomaly.source_module == source, MetricAnomaly.resource_name == resource, + MetricAnomaly.metric_name == metric, MetricAnomaly.review == TO_REVIEW) + .order_by(MetricAnomaly.last_started_at.desc()).limit(1))).scalar_one_or_none() + if anomaly is None: + anomaly = MetricAnomaly( + source_module=source, resource_name=resource, metric_name=metric, + started_at=start, last_started_at=start, ended_at=None, occurrences=1, + peak_level=level, peak_at=now, usual_hold=norm.usual_hold, + usual_peak=norm.usual_peak, review=TO_REVIEW) + session.add(anomaly) + await session.flush() + session.add(MetricAnomalySpan(anomaly_id=anomaly.id, started_at=start, peak_level=level)) + return + last = (await session.execute( + select(MetricAnomalySpan).where(MetricAnomalySpan.anomaly_id == anomaly.id) + .order_by(MetricAnomalySpan.started_at.desc()).limit(1))).scalar_one_or_none() + if last is not None and last.ended_at is not None and now - last.ended_at <= REJOIN: + last.ended_at = None + span = last + else: + span = MetricAnomalySpan(anomaly_id=anomaly.id, started_at=start, peak_level=level) + session.add(span) + anomaly.occurrences += 1 + anomaly.last_started_at = start + anomaly.ended_at = None + _raise_peak(anomaly, span, level, norm, now) + + +async def lapse(session, now: datetime, review_days: int) -> int: + """Ended anomalies still to review after `review_days` lapse. One still + going never does. reviewed_at records when it lapsed. Returns how many.""" + result = await session.execute( + update(MetricAnomaly) + .where(MetricAnomaly.review == TO_REVIEW, MetricAnomaly.ended_at.is_not(None), + MetricAnomaly.ended_at < now - timedelta(days=review_days)) + .values(review=LAPSED, reviewed_at=now)) + return result.rowcount or 0 + + +async def prune(session, before: datetime) -> int: + """Reviewed and lapsed anomalies whose last occurrence ended before + `before` are deleted (data.retention_days), their spans with them.""" + result = await session.execute( + delete(MetricAnomaly) + .where(MetricAnomaly.review != TO_REVIEW, MetricAnomaly.ended_at.is_not(None), + MetricAnomaly.ended_at < before)) + return result.rowcount or 0 + + +async def mark(session, anomaly_id: str, review: str, by: str, + note: str | None = None, now: datetime | None = None) -> MetricAnomaly: + """Mark an anomaly normal or real. A lapsed one may still be marked. + Raises LookupError when there is no such anomaly, ValueError on a bad mark.""" + if review not in MARKS: + raise ValueError(f"review must be one of {', '.join(MARKS)}") + anomaly = await session.get(MetricAnomaly, anomaly_id) + if anomaly is None: + raise LookupError(anomaly_id) + anomaly.review = review + anomaly.reviewed_by = by + anomaly.reviewed_at = now or datetime.now(timezone.utc) + anomaly.note = (note or "").strip() or None + return anomaly + + +async def run_anomalies(app) -> None: + """The minute pass: judge every learned key, then lapse what has waited + out the review window.""" + from steward.core.settings import get_setting + now = datetime.now(timezone.utc) + async with app.db_sessionmaker() as session: + async with session.begin(): + await detect(session, now) + days = int(await get_setting(session, "anomalies.review_days") or 14) + await lapse(session, now, max(1, days)) + + +# ── Reading them ───────────────────────────────────────────────────────────── + +@dataclass +class AnomalyView: + """One anomaly as the pages, the report and MCP show it.""" + anomaly: MetricAnomaly + last_span: MetricAnomalySpan | None + href: str | None = None + + @property + def ongoing(self) -> bool: + return self.anomaly.ended_at is None + + @property + def link(self) -> str | None: + """The resource's page over a range that takes in the latest + occurrence (the pages offer ranges back from now, not a window).""" + if not self.href: + return None + reach = datetime.now(timezone.utc) - self.anomaly.last_started_at + key = "24h" if reach < timedelta(hours=20) else "7d" if reach < timedelta(days=6) else "30d" + return f"{self.href}?range={key}" + + +def _ordered(): + """To review first, then the rest; newest occurrence first within each.""" + return (case((MetricAnomaly.review == TO_REVIEW, 0), else_=1), + MetricAnomaly.last_started_at.desc()) + + +async def views(session, rows: list[MetricAnomaly], *, links: bool = True) -> list[AnomalyView]: + """Each anomaly with its latest occurrence and the page that shows its + resource.""" + if not rows: + return [] + from steward.alerts.service import resource_url_for + latest = {} + for span in (await session.execute( + select(MetricAnomalySpan) + .where(MetricAnomalySpan.anomaly_id.in_([a.id for a in rows])) + .order_by(MetricAnomalySpan.started_at))).scalars(): + latest[span.anomaly_id] = span + hrefs: dict[tuple[str, str], str | None] = {} + out = [] + for a in rows: + key = (a.source_module, a.resource_name) + if links and key not in hrefs: + hrefs[key] = await resource_url_for(session, *key) + out.append(AnomalyView(a, latest.get(a.id), hrefs.get(key))) + return out + + +async def list_anomalies(session, reviews=None, *, since: datetime | None = None, + source: str | None = None, resource: str | None = None, + limit: int = 200) -> list[MetricAnomaly]: + """Anomalies in any of `reviews` (all when None), to review first, newest + occurrence first. `since` keeps those with an occurrence after it, or + still going.""" + q = select(MetricAnomaly) + if reviews: + q = q.where(MetricAnomaly.review.in_(list(reviews))) + if since is not None: + q = q.where(or_(MetricAnomaly.ended_at.is_(None), MetricAnomaly.ended_at >= since)) + if source: + q = q.where(MetricAnomaly.source_module == source) + if resource: + # host_agent's parts ("host:/var") belong to the host's page too. + q = q.where(or_(MetricAnomaly.resource_name == resource, + MetricAnomaly.resource_name.like(f"{resource}:%"))) + return list((await session.execute(q.order_by(*_ordered()).limit(limit))).scalars()) + + +async def open_anomalies(session) -> list[MetricAnomaly]: + """Anomalies with an occurrence going on now that still need attention: + to review, or marked real. One marked normal is the new normal being + learned, and is not a warning.""" + return list((await session.execute( + select(MetricAnomaly) + .where(MetricAnomaly.ended_at.is_(None), MetricAnomaly.review.in_((TO_REVIEW, REAL))) + .order_by(MetricAnomaly.last_started_at))).scalars()) + + +async def to_review_count(session) -> int: + return int((await session.execute( + select(func.count()).select_from(MetricAnomaly) + .where(MetricAnomaly.review == TO_REVIEW))).scalar_one()) + + +async def spans_for(session, source: str, resource: str, metric: str | None, + since: datetime) -> list[MetricAnomalySpan]: + """Occurrences on one resource (and metric) since `since`, for shading + its chart.""" + q = (select(MetricAnomalySpan) + .join(MetricAnomaly, MetricAnomaly.id == MetricAnomalySpan.anomaly_id) + .where(MetricAnomaly.source_module == source, MetricAnomaly.resource_name == resource, + or_(MetricAnomalySpan.ended_at.is_(None), MetricAnomalySpan.ended_at >= since))) + if metric: + q = q.where(MetricAnomaly.metric_name == metric) + return list((await session.execute(q.order_by(MetricAnomalySpan.started_at))).scalars()) + + +async def week_summary(session, since: datetime) -> dict: + """For the weekly report: anomalies with an occurrence since `since`, and + those that lapsed since then.""" + seen = await list_anomalies(session, since=since, limit=50) + lapsed = list((await session.execute( + select(MetricAnomaly).where(MetricAnomaly.review == LAPSED, + MetricAnomaly.reviewed_at >= since) + .order_by(MetricAnomaly.last_started_at.desc()).limit(50))).scalars()) + seen_ids = {a.id for a in seen} + return {"seen": seen, "lapsed": [a for a in lapsed if a.id not in seen_ids]} + diff --git a/steward/core/cleanup.py b/steward/core/cleanup.py index 6b9d0ec..6ad8621 100644 --- a/steward/core/cleanup.py +++ b/steward/core/cleanup.py @@ -8,6 +8,7 @@ from typing import TYPE_CHECKING from sqlalchemy import delete +from steward.core.anomalies import prune as prune_anomalies from steward.models.alerts import AlertEvent from steward.models.ansible import AnsibleRun from steward.models.audit import AuditEvent @@ -76,6 +77,11 @@ async def run_cleanup(app: "Quart") -> dict: removed.append([f"{table}: older than {days} days", result.rowcount or 0]) if result.rowcount: logger.info("Pruned %d rows from %s", result.rowcount, table) + # Anomalies go by when their last occurrence ended, and only once + # reviewed or lapsed: one still to review is never pruned (#5740). + pruned = await prune_anomalies(session, cutoff) + removed.append([f"metric_anomalies: reviewed, ended over {data_days} days ago", + pruned]) files = await asyncio.to_thread(prune_run_logs, ARTIFACT_DIR, cutoff) removed.append([f"Ansible run log files: older than {data_days} days", files]) diff --git a/steward/core/formatting.py b/steward/core/formatting.py index 8bc550b..699f661 100644 --- a/steward/core/formatting.py +++ b/steward/core/formatting.py @@ -207,6 +207,12 @@ STATES: dict[str, tuple[str, str]] = { "failed": ("crit", "Failed"), "interrupted": ("warn", "Interrupted"), "cancelled": ("dim", "Cancelled"), + # A metric anomaly (#5740), and where its review stands. + "anomaly": ("warn", "Metric anomaly"), + "to_review": ("warn", "To review"), + "normal": ("ok", "Marked normal"), + "real": ("crit", "Real problem"), + "lapsed": ("dim", "Not reviewed"), } diff --git a/steward/core/mail.py b/steward/core/mail.py index 720dbac..18533c8 100644 --- a/steward/core/mail.py +++ b/steward/core/mail.py @@ -258,6 +258,22 @@ def burst_email(cfg: dict, alerts: list[dict], to: Recipient) -> tuple[str, str, return f"[Steward] {len(facts)} alerts: {said}", text, html +def anomaly_facts(a: dict, tz: tzinfo) -> dict: + """A metric anomaly in the weekly report (#5740): its review as a badge, + and when it happened in the reader's zone.""" + from steward.core.formatting import status_info + tone, review = status_info(a["review"]) + if a["occurrences"] > 1: + last = (f"since {moment(a['last_started_at'], tz)}, ongoing" if a["ended_at"] is None + else f"{moment(a['last_started_at'], tz)} – {moment(a['ended_at'], tz)}") + when = f"{a['occurrences']} times since {moment(a['started_at'], tz)}, last {last}" + elif a["ended_at"] is None: + when = f"ongoing since {moment(a['started_at'], tz)}" + else: + when = f"{moment(a['started_at'], tz)} – {moment(a['ended_at'], tz)}" + return {"name": a["name"], "tone": tone, "review": review, "when": when, "level": a["level"]} + + def report_email(cfg: dict, data: dict, to: Recipient) -> tuple[str, str, str]: """(subject, text, html) for the weekly report built by reports.build_report_data, its dates in the recipient's zone.""" @@ -269,8 +285,12 @@ def report_email(cfg: dict, data: dict, to: Recipient) -> tuple[str, str, str]: else f"{start.day} {start:%b %Y} – {end.day} {end:%b %Y}") alerts = [{**a, "tone": status_info(a["state"])[0], "state": status_info(a["state"])[1]} for a in data["active_alerts"]] + found = data.get("anomalies") or {} + anomalies = {"seen": [anomaly_facts(a, tz) for a in found.get("seen", [])], + "lapsed": [anomaly_facts(a, tz) for a in found.get("lapsed", [])], + "to_review": found.get("to_review", 0)} base = (cfg.get("base_url") or "").rstrip("/") - text, html = _render("report", d=data, alerts=alerts, period=period, + text, html = _render("report", d=data, alerts=alerts, anomalies=anomalies, period=period, sent=moment(data["generated_at"], tz), base=base) return f"[Steward] Weekly report, {period}", text, html diff --git a/steward/core/norms.py b/steward/core/norms.py index 287b6f1..08103bd 100644 --- a/steward/core/norms.py +++ b/steward/core/norms.py @@ -24,6 +24,10 @@ The slots: Only metrics registered with learn_normal() are learned: a host agent alone reports thousands of per-core, per-disk and per-interface series, and most have no use for a normal. + +An hour an anomaly covers is learned only once someone marks the anomaly +normal (steward.core.anomalies): a level that jumps and stays is never +absorbed on its own, and a lapsed anomaly is never learned. """ from __future__ import annotations @@ -80,11 +84,12 @@ def slot_of(at: datetime, kind: str) -> int: # One statement learns every normal: the hours of each learned key (rolled up, -# plus the raw window not rolled up yet), which kind of slot its history -# supports, then each slot's usual hold and peak across past days, every hour -# counted in its own slot and its neighbours'. +# plus the raw window not rolled up yet) less those an anomaly covers that +# nobody marked normal (#5740), which kind of slot its history supports, then +# each slot's usual hold and peak across past days, every hour counted in its +# own slot and its neighbours'. _REBUILD = text(""" -WITH hours AS ( +WITH all_hours AS ( SELECT source_module AS s, resource_name AS r, metric_name AS m, bucket AS b, COALESCE(value_p95, value_avg) AS hold, value_max AS peak FROM plugin_metrics_hourly @@ -97,6 +102,16 @@ WITH hours AS ( AND source_module || '/' || metric_name = ANY(:keys) GROUP BY 1, 2, 3, 4 ), +hours AS ( + SELECT * FROM all_hours h + WHERE NOT EXISTS ( + SELECT 1 FROM metric_anomaly_spans sp + JOIN metric_anomalies a ON a.id = sp.anomaly_id + WHERE a.review <> 'normal' + AND a.source_module = h.s AND a.resource_name = h.r AND a.metric_name = h.m + AND sp.started_at < h.b + interval '1 hour' + AND COALESCE(sp.ended_at, :now) > h.b) +), keyed AS ( SELECT s, r, m, count(DISTINCT date_trunc('day', b)) AS days, min(b) AS first FROM hours GROUP BY s, r, m diff --git a/steward/core/reports.py b/steward/core/reports.py index 21634fb..481727e 100644 --- a/steward/core/reports.py +++ b/steward/core/reports.py @@ -106,6 +106,15 @@ async def build_report_data(app) -> dict: hosts_down = sorted(name for mid, name in enabled if latest.get(mid) and not latest[mid][-1].is_up) + # ── Metric anomalies this week, and those that lapsed (#5740) ───────── + from steward.core import anomalies as anom + week = await anom.week_summary(db, cutoff_7d) + anomalies = { + "seen": [_anomaly(a) for a in week["seen"]], + "lapsed": [_anomaly(a) for a in week["lapsed"]], + "to_review": await anom.to_review_count(db), + } + return { "generated_at": now, "uptime_summary": uptime_summary, @@ -113,6 +122,21 @@ async def build_report_data(app) -> dict: "alert_events": alert_events, "top_routers": top_routers, "hosts_down": hosts_down, + "anomalies": anomalies, + } + + +def _anomaly(a) -> dict: + """One anomaly for the report; its times are put in each reader's zone + when the email is written (mail.report_email).""" + from steward.core.formatting import metric_label, metric_value + return { + "name": f"{a.resource_name} — {metric_label(a.metric_name)}", + "review": a.review, + "started_at": a.started_at, "last_started_at": a.last_started_at, "ended_at": a.ended_at, + "occurrences": a.occurrences, + "level": (f"held {metric_value(a.metric_name, a.peak_level)}, usual " + f"{metric_value(a.metric_name, a.usual_hold)}"), } diff --git a/steward/core/settings.py b/steward/core/settings.py index 85bcad3..c80453b 100644 --- a/steward/core/settings.py +++ b/steward/core/settings.py @@ -114,6 +114,9 @@ DEFAULTS: dict[str, Any] = { # Seconds a host's clock may be from Steward's before it is called out # (milestone 490), either way. "thresholds.clock_warn_s": 30, "thresholds.clock_crit_s": 300, + # Days an ended metric anomaly waits to be reviewed before it lapses: + # it leaves the to-review list and is kept as not reviewed (#5740). + "anomalies.review_days": 14, # Docker time-series retention (rule 25 — tunable, no restart). Raw 30s # samples are heavy, so keep a short raw window then roll up to hourly # averages kept much longer; lifecycle events are light, keep a month. diff --git a/steward/hosts/routes.py b/steward/hosts/routes.py index a310cd7..657399c 100644 --- a/steward/hosts/routes.py +++ b/steward/hosts/routes.py @@ -1,13 +1,13 @@ from __future__ import annotations import asyncio import uuid -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from quart import Blueprint, make_response, render_template, request, redirect, url_for, current_app, session from sqlalchemy import select from steward.ansible import executor, sources as ansible_src from steward.alerts import service as alert_svc from steward.auth.middleware import require_role -from steward.core import renames +from steward.core import anomalies as anom, renames from steward.core.audit import audit_request, log_audit from steward.models.ansible import AnsibleRun, AnsibleRunStatus from steward.models.hosts import Host @@ -131,10 +131,14 @@ async def host_detail(host_id: str): linkable_targets = (await db.execute( select(AnsibleTarget).where(AnsibleTarget.host_id.is_(None)) .order_by(AnsibleTarget.name))).scalars().all() + # The host's metric anomalies, its disks and interfaces included (#5740). + anomalies = await anom.views(db, await anom.list_anomalies( + db, source="host_agent", resource=host.name, + since=datetime.now(timezone.utc) - timedelta(days=30), limit=10), links=False) return await render_template( "hosts/detail.html", - host=host, monitors=monitors, uptime=uptime, alerts=alerts, + host=host, monitors=monitors, uptime=uptime, alerts=alerts, anomalies=anomalies, linked_target=linked_target, linkable_targets=linkable_targets, ansible_sources=_ansible_source_names(), ) diff --git a/steward/mcp/tools/__init__.py b/steward/mcp/tools/__init__.py index 646c96e..c0fc013 100644 --- a/steward/mcp/tools/__init__.py +++ b/steward/mcp/tools/__init__.py @@ -12,9 +12,9 @@ each argument's empty value means, and what comes back. """ from __future__ import annotations -from . import alerts, audit, automation, hosts, metrics, monitors, overview, session +from . import alerts, anomalies, audit, automation, hosts, metrics, monitors, overview, session -_MODULES = (session, overview, hosts, monitors, alerts, metrics, audit, automation) +_MODULES = (session, overview, hosts, monitors, alerts, anomalies, metrics, audit, automation) def register_all(register_tool) -> None: diff --git a/steward/mcp/tools/anomalies.py b/steward/mcp/tools/anomalies.py new file mode 100644 index 0000000..8038394 --- /dev/null +++ b/steward/mcp/tools/anomalies.py @@ -0,0 +1,96 @@ +"""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 steward.core import anomalies as anom +from steward.mcp._context import current_caller +from steward.models.users import UserRole + +from ._audit import audit_mcp, mcp_username +from ._shape import bounded, clamp_limit, db_session, iso, rnd + +LIST_MAX = 200 + + +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_id: str, review: str, note: str = "") -> dict: + """Mark a metric anomaly "normal" or "real" (list_anomalies gives ids). + + 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: optional, kept with the review (say what it was). + A lapsed anomaly may still be marked. Audited as " via MCP". + """ + caller = current_caller() + async with db_session() as db, db.begin(): + 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.") from None + detail = {"source": a.source_module, "resource": a.resource_name, + "metric": a.metric_name, "note": a.note} + result = {"anomaly_id": a.id, "review": a.review, "reviewed_by": a.reviewed_by, + "reviewed_at": iso(a.reviewed_at), "note": a.note} + await audit_mcp(f"metric_anomaly.marked_{review}", "metric_anomaly", anomaly_id, detail) + return result + + +def register(register_tool) -> None: + register_tool(list_anomalies, role=UserRole.viewer, read_only=True) + register_tool(review_anomaly, role=UserRole.operator, read_only=False) diff --git a/steward/mcp/tools/overview.py b/steward/mcp/tools/overview.py index 827c11b..fb4f62b 100644 --- a/steward/mcp/tools/overview.py +++ b/steward/mcp/tools/overview.py @@ -7,6 +7,7 @@ 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 @@ -140,6 +141,22 @@ def run_findings(runs) -> list[dict]: 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()): @@ -162,12 +179,15 @@ async def get_overview() -> dict: 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, and anything enabled plugins add (unhealthy containers, - for one). Each finding carries the ids to follow and `next`, the call - that shows its evidence. + 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, and `active_maintenance` names windows - currently silencing alerts. `all_clear` is true when nothing is wrong. + `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() @@ -187,12 +207,15 @@ async def get_overview() -> dict: .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)) + + run_findings(runs) + + anomaly_findings(open_anomalies)) plugin_found, unavailable = await _plugin_findings(getattr(app, "mcp_server", None)) findings = rank(findings + plugin_found) @@ -208,6 +231,7 @@ async def get_overview() -> dict: "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), diff --git a/steward/migrations/versions/0046_metric_anomalies.py b/steward/migrations/versions/0046_metric_anomalies.py new file mode 100644 index 0000000..3b7b5d4 --- /dev/null +++ b/steward/migrations/versions/0046_metric_anomalies.py @@ -0,0 +1,55 @@ +"""Metrics that departed from their normal, to review (#5740). + +Revision ID: 0046_metric_anomalies +Revises: 0045_metric_norms +Create Date: 2026-10-14 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "0046_metric_anomalies" +down_revision: Union[str, None] = "0045_metric_norms" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "metric_anomalies", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("source_module", sa.String(64), nullable=False), + sa.Column("resource_name", sa.String(255), nullable=False), + sa.Column("metric_name", sa.String(128), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("last_started_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("ended_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("occurrences", sa.Integer, nullable=False), + sa.Column("peak_level", sa.Float, nullable=False), + sa.Column("peak_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("usual_hold", sa.Float, nullable=False), + sa.Column("usual_peak", sa.Float, nullable=False), + sa.Column("review", sa.String(12), nullable=False), + sa.Column("reviewed_by", sa.String(64), nullable=True), + sa.Column("reviewed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("note", sa.Text, nullable=True), + ) + op.create_index("ix_metric_anomalies_key", "metric_anomalies", + ["source_module", "resource_name", "metric_name"]) + op.create_index("ix_metric_anomalies_review_last", "metric_anomalies", + ["review", "last_started_at"]) + op.create_table( + "metric_anomaly_spans", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("anomaly_id", sa.String(36), + sa.ForeignKey("metric_anomalies.id", ondelete="CASCADE"), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("ended_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("peak_level", sa.Float, nullable=False), + ) + op.create_index("ix_metric_anomaly_spans_anomaly_id", "metric_anomaly_spans", ["anomaly_id"]) + + +def downgrade() -> None: + op.drop_table("metric_anomaly_spans") + op.drop_table("metric_anomalies") diff --git a/steward/models/__init__.py b/steward/models/__init__.py index fb5de64..d559ad3 100644 --- a/steward/models/__init__.py +++ b/steward/models/__init__.py @@ -6,7 +6,7 @@ from .password_resets import PasswordReset from .hosts import Host from .monitors import Monitor, MonitorResult, MonitorType from .metrics import PluginMetric -from .norms import MetricNorm +from .norms import MetricNorm, MetricAnomaly, MetricAnomalySpan from .alerts import AlertRule, AlertState, AlertEvent, AlertOperator, AlertStateEnum from .ansible import AnsibleRun, AnsibleRunStatus from .ansible_inventory import AnsibleTarget, AnsibleGroup, ansible_target_groups @@ -19,7 +19,7 @@ __all__ = [ "Base", "User", "ApiToken", "UserInvite", "PasswordReset", "Host", "Monitor", "MonitorResult", "MonitorType", - "PluginMetric", "MetricNorm", + "PluginMetric", "MetricNorm", "MetricAnomaly", "MetricAnomalySpan", "AlertRule", "AlertState", "AlertEvent", "AlertOperator", "AlertStateEnum", "AnsibleRun", "AnsibleRunStatus", "AnsibleTarget", "AnsibleGroup", "ansible_target_groups", diff --git a/steward/models/norms.py b/steward/models/norms.py index 13283df..5d2a965 100644 --- a/steward/models/norms.py +++ b/steward/models/norms.py @@ -1,8 +1,9 @@ from __future__ import annotations from datetime import datetime -from sqlalchemy import DateTime, Float, Integer, SmallInteger, String +from sqlalchemy import DateTime, Float, ForeignKey, Index, Integer, SmallInteger, String, Text from sqlalchemy.orm import Mapped, mapped_column from .base import Base +from steward.core.ids import time_ordered_id class MetricNorm(Base): @@ -25,3 +26,55 @@ class MetricNorm(Base): usual_peak: Mapped[float] = mapped_column(Float, nullable=False) days_seen: Mapped[int] = mapped_column(Integer, nullable=False) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + +class MetricAnomaly(Base): + """A metric that departed from its normal (#5740): one item to review. + + A departure that recurs while the item is still to review joins it as + another occurrence (a MetricAnomalySpan) rather than adding a row, so the + list stays one row per resource and metric. `review` is to_review until + someone marks it normal (its hours are learned) or real (they are not); + an ended item nobody reviewed within anomalies.review_days becomes lapsed, + and is not learned either.""" + __tablename__ = "metric_anomalies" + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=time_ordered_id) + source_module: Mapped[str] = mapped_column(String(64), nullable=False) + resource_name: Mapped[str] = mapped_column(String(255), nullable=False) + metric_name: Mapped[str] = mapped_column(String(128), nullable=False) + # The first occurrence's start, the latest's start, and the latest's end: + # NULL while an occurrence is still going. + started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + last_started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + ended_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + occurrences: Mapped[int] = mapped_column(Integer, nullable=False, default=1) + # The highest 15-minute level any occurrence held, and the normal for the + # hour it was reached in. + peak_level: Mapped[float] = mapped_column(Float, nullable=False) + peak_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + usual_hold: Mapped[float] = mapped_column(Float, nullable=False) + usual_peak: Mapped[float] = mapped_column(Float, nullable=False) + review: Mapped[str] = mapped_column(String(12), nullable=False, default="to_review") + reviewed_by: Mapped[str | None] = mapped_column(String(64), nullable=True) + reviewed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + note: Mapped[str | None] = mapped_column(Text, nullable=True) + + __table_args__ = ( + Index("ix_metric_anomalies_key", "source_module", "resource_name", "metric_name"), + Index("ix_metric_anomalies_review_last", "review", "last_started_at"), + ) + + +class MetricAnomalySpan(Base): + """One occurrence of an anomaly: when the metric was out of its normal. + Learning skips the hours a span covers unless its anomaly was marked + normal, and a resource's chart shades them.""" + __tablename__ = "metric_anomaly_spans" + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=time_ordered_id) + anomaly_id: Mapped[str] = mapped_column( + String(36), ForeignKey("metric_anomalies.id", ondelete="CASCADE"), nullable=False, index=True) + started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + ended_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + peak_level: Mapped[float] = mapped_column(Float, nullable=False) diff --git a/steward/settings/routes.py b/steward/settings/routes.py index ece099c..f9b795b 100644 --- a/steward/settings/routes.py +++ b/steward/settings/routes.py @@ -155,6 +155,7 @@ _RETENTION_FIELDS = [ ("docker_logs_retention_days", "docker.logs.retention_days"), ("metrics_raw_days", "metrics.retention.raw_days"), ("metrics_rollup_days", "metrics.retention.rollup_days"), + ("anomalies_review_days", "anomalies.review_days"), ] diff --git a/steward/status/routes.py b/steward/status/routes.py index 1b7363a..8255626 100644 --- a/steward/status/routes.py +++ b/steward/status/routes.py @@ -4,6 +4,7 @@ from sqlalchemy import func, select from steward.alerts.service import active_alerts, alert_label from steward.auth.middleware import require_role +from steward.core import anomalies as anom from steward.core.status import collect_status from steward.models.hosts import Host from steward.models.users import UserRole @@ -29,11 +30,16 @@ async def widget(): total_hosts = (await db.execute( select(func.count()).select_from(Host))).scalar() alerts = await active_alerts(db) + # Metric anomalies going on now are warnings; the ones waiting for a + # review are one line linking to the list (#5740). + anomalies = await anom.views(db, await anom.open_anomalies(db)) + to_review = await anom.to_review_count(db) up, down, pending = _summarize(entries) return await render_template( "status/widget.html", entries=entries, up=up, down=down, pending=pending, total_hosts=total_hosts, alerts=overview_alerts(alerts), + anomalies=anomalies, to_review=to_review, widget_id=request.args.get("wid", "0"), ) diff --git a/steward/templates/_anomaly.html b/steward/templates/_anomaly.html new file mode 100644 index 0000000..b233e85 --- /dev/null +++ b/steward/templates/_anomaly.html @@ -0,0 +1,72 @@ +{# Metric anomalies (#5740), drawn the same on Alerts → Anomalies, a resource's + own page and the overview widget. Import with: + {% from "_anomaly.html" import anomaly_row, anomaly_when with context %} + (with context: the review buttons read session.user_role). #} +{% from "_macros.html" import status_badge %} + +{# When it happened: "since …" while going, "from – to" once ended; a repeat + says how often and when the latest was. #} +{% macro anomaly_when(v) -%} +{%- set a = v.anomaly -%} +{%- if a.occurrences > 1 -%} + {{ a.occurrences }} times since {{ a.started_at|when }}, last + {% if v.ongoing %}since {{ a.last_started_at|when }}, ongoing{% else %}{{ a.last_started_at|when }} – {{ a.ended_at|when }}{% endif %} +{%- elif v.ongoing -%} + Ongoing since {{ a.started_at|when }} +{%- else -%} + {{ a.started_at|when }} – {{ a.ended_at|when }} +{%- endif -%} +{%- endmacro %} + +{# Its highest level against what is usual for that hour. #} +{% macro anomaly_level(a) -%} +Held {{ metric_value(a.metric_name, a.peak_level) }}; usual for that hour +{{ metric_value(a.metric_name, a.usual_hold) }}, peaking at {{ metric_value(a.metric_name, a.usual_peak) }} +{%- endmacro %} + +{# The review buttons, or how it was reviewed. `back` is the page to come + back to after marking. #} +{% macro anomaly_review(a, back=none) -%} +{%- set can_review = session.user_role in ('operator', 'admin') -%} +{%- if a.review == "normal" -%} + Marked normal by {{ a.reviewed_by or "someone" }}, {{ a.reviewed_at|when }}. +{%- elif a.review == "real" -%} + Real problem, marked by {{ a.reviewed_by or "someone" }}, {{ a.reviewed_at|when }}{% if a.note %}: {{ a.note }}{% else %}.{% endif %} +{%- elif a.review == "lapsed" -%} + Not reviewed within the review window; lapsed {{ a.reviewed_at|when }}. Its hours are not learned. +{%- endif -%} +{%- if can_review and a.review in ("to_review", "lapsed") %} +
+ {{ csrf_field() }} + {% if back %}{% endif %} + + + +
+{%- endif -%} +{%- endmacro %} + +{# One anomaly as a row: badge, resource and metric, when, level, review. #} +{% macro anomaly_row(v, back=none, show_resource=true) -%} +{%- set a = v.anomaly -%} +
+
+ {{ status_badge("anomaly") }} + {% if a.review != "to_review" %}{{ status_badge(a.review) }}{% endif %} + {% if show_resource %} + {% if v.link %}{{ a.resource_name }} + {% else %}{{ a.resource_name }}{% endif %} + {{ source_label(a.source_module) }} + {% endif %} + {{ metric_label(a.metric_name) }} +
+
+ {{ anomaly_when(v) }} · {{ anomaly_level(a) }} +
+
{{ anomaly_review(a, back) }}
+
+{%- endmacro %} diff --git a/steward/templates/_anomaly_card.html b/steward/templates/_anomaly_card.html new file mode 100644 index 0000000..d289fc6 --- /dev/null +++ b/steward/templates/_anomaly_card.html @@ -0,0 +1,14 @@ +{# A resource's metric anomalies (#5740), for its own page: `anomalies` are + anomalies.views() rows, to review first; `back` is this page's path, so a + review returns here. Shown only when there is one. #} +{% from "_anomaly.html" import anomaly_row with context %} +{% if anomalies %} +{% set going = anomalies | selectattr("ongoing") | list %} +
+

+ {% if going %}Out of its normal now{% else %}Metric anomalies, last 30 days{% endif %} +

+ {% for v in anomalies %}{{ anomaly_row(v, back=back, show_resource=v.anomaly.resource_name != base_resource) }}{% endfor %} + +
+{% endif %} diff --git a/steward/templates/alerts/anomalies.html b/steward/templates/alerts/anomalies.html new file mode 100644 index 0000000..60ee56f --- /dev/null +++ b/steward/templates/alerts/anomalies.html @@ -0,0 +1,36 @@ +{% extends "base.html" %} +{% from "_macros.html" import crumbs, page_header %} +{% from "_anomaly.html" import anomaly_row with context %} +{% block title %}Metric Anomalies — Steward{% endblock %} +{% block breadcrumb %}{{ crumbs([("Alerts", "/alerts/"), ("Anomalies", "")]) }}{% endblock %} +{% block content %} +{% call page_header("Metric anomalies", "A metric that left what is normal for it at that hour. Mark each normal, so its hours are learned, or a real problem, so they are not.") %} + History +{% endcall %} + +{% macro divided(items) %} +{% for item in items %} + {% if item.opens_group %}

{{ item.group }}

{% endif %} + {{ anomaly_row(item.v, back="/alerts/anomalies") }} +{% endfor %} +{% endmacro %} + +
+

+ {% if to_review %}{{ to_review|length }} to review{% else %}Nothing to review{% endif %} +

+

+ An ended anomaly nobody reviews within {{ review_days }} day{{ "s" if review_days != 1 }} lapses: + it leaves this list, stays below as not reviewed, and is never learned. + The window is in Settings → Thresholds. +

+ {{ divided(to_review) }} +
+ +{% if past %} +
+

Reviewed and lapsed

+ {{ divided(past) }} +
+{% endif %} +{% endblock %} diff --git a/steward/templates/alerts/list.html b/steward/templates/alerts/list.html index 338235c..41599cd 100644 --- a/steward/templates/alerts/list.html +++ b/steward/templates/alerts/list.html @@ -14,6 +14,7 @@ {% set rules_off = rules|rejectattr("enabled")|list|length %} {% call page_header("Alerts") %} History + Anomalies Maintenance {% if can_edit %}New rule{% endif %} {% endcall %} diff --git a/steward/templates/email/report.html b/steward/templates/email/report.html index c0ff029..a31d9ca 100644 --- a/steward/templates/email/report.html +++ b/steward/templates/email/report.html @@ -43,6 +43,23 @@ This week {{ d.alert_events.fired }} {{ "alert" if d.alert_events.fired == 1 els No monitors are enabled. {% endif %} +{% if anomalies.seen or anomalies.lapsed %} +{{ section("Metric anomalies this week", base ~ "/alerts/anomalies" if base else "") }} +{% call rows() %} +{% for a in anomalies.seen %} +{{ row([badge(a.tone, a.review), a.name ~ " — " ~ a.when, a.level]) }} +{% endfor %} +{% for a in anomalies.lapsed %} +{{ row([badge(a.tone, a.review), a.name ~ " — lapsed unreviewed; " ~ a.when, a.level]) }} +{% endfor %} +{% endcall %} +{% if anomalies.to_review %} + +{{ anomalies.to_review }} {{ "anomaly waits" if anomalies.to_review == 1 else "anomalies wait" }} to be marked normal or real. + +{% endif %} +{% endif %} + {% if d.top_routers %} {{ section("Busiest Traefik routers, last 24 hours", base ~ "/plugins/traefik/" if base else "") }} {% call rows() %} diff --git a/steward/templates/email/report.txt b/steward/templates/email/report.txt index f66602d..09f9207 100644 --- a/steward/templates/email/report.txt +++ b/steward/templates/email/report.txt @@ -21,6 +21,20 @@ UPTIME OVER 7 DAYS{% if base %} ({{ base }}/monitors/){% endif %} {% else %} No monitors are enabled. {% endfor %} +{% if anomalies.seen or anomalies.lapsed %} + +METRIC ANOMALIES THIS WEEK{% if base %} ({{ base }}/alerts/anomalies){% endif %} + +{% for a in anomalies.seen %} +- {{ a.review }}: {{ a.name }}, {{ a.when }} ({{ a.level }}) +{% endfor %} +{% for a in anomalies.lapsed %} +- {{ a.review }}: {{ a.name }}, lapsed unreviewed; {{ a.when }} ({{ a.level }}) +{% endfor %} +{% if anomalies.to_review %} +{{ anomalies.to_review }} {{ "anomaly waits" if anomalies.to_review == 1 else "anomalies wait" }} to be marked normal or real. +{% endif %} +{% endif %} {% if d.top_routers %} BUSIEST TRAEFIK ROUTERS, LAST 24 HOURS{% if base %} ({{ base }}/plugins/traefik/){% endif %} diff --git a/steward/templates/hosts/detail.html b/steward/templates/hosts/detail.html index 98bd7b1..3c2eb35 100644 --- a/steward/templates/hosts/detail.html +++ b/steward/templates/hosts/detail.html @@ -44,6 +44,9 @@ {% endif %} +{# ── Metric anomalies: out of what is normal for this host at that hour ───── #} +{% with back="/hosts/" ~ host.id, base_resource=host.name %}{% include "_anomaly_card.html" %}{% endwith %} + {# ── Live vitals strip (host_agent fragment) — full width on top, polled ────── #}
{% endmacro %} +
+

Metric anomalies

+

+ A metric that leaves what is normal for it at that hour is listed under + Alerts → Anomalies to review. One nobody marks normal or real within this + window, once it has ended, lapses: it leaves the list and is kept as not + reviewed, and its hours are never learned. +

+ {{ days("Review window", "anomalies_review_days", "anomalies.review_days", + "An anomaly still going never lapses.") }} +
+

Docker data retention

diff --git a/steward/templates/status/widget.html b/steward/templates/status/widget.html index 4b0bba6..fb6bd0e 100644 --- a/steward/templates/status/widget.html +++ b/steward/templates/status/widget.html @@ -3,6 +3,7 @@ widget is the full list of checks; this is the summary. #} {% from "_macros.html" import status_badge %} {% set firing = alerts | selectattr("label", "in", ["firing", "no_data"]) | list %} +{% set anomalies = anomalies or [] %}

{% if session.user_id %}{% endif %} @@ -24,7 +25,7 @@
{% set problems = entries | rejectattr("status", "equalto", "up") | list %} -{% if not problems and not alerts %} +{% if not problems and not alerts and not anomalies %}
{% if entries %}✓ All {{ up }} monitors are up and no alert is firing.{% else %}No monitors configured, and no alert is firing.{% endif %}
@@ -54,6 +55,18 @@ {% if e.since %}{% if e.since_at_least %}over {% endif %}{{ e.since|ago }}{% endif %}
{% endfor %} + {# Metric anomalies going on now (#5740): a warning each. #} + {% for v in anomalies[:4] %} + + {% endfor %} {# What didn't fit, each linking to its own list (#5510). #} {% set more_alerts, more_monitors = alerts|length - 6, problems|length - 8 %} {% if more_alerts > 0 or more_monitors > 0 %} @@ -65,3 +78,9 @@ {% endif %}
{% endif %} +{% if to_review %} +
+ {% if session.user_id %}{{ to_review }} metric anomal{{ "y" if to_review == 1 else "ies" }} to review + {% else %}{{ to_review }} metric anomal{{ "y" if to_review == 1 else "ies" }} to review{% endif %} +
+{% endif %} diff --git a/tests/core/test_anomalies.py b/tests/core/test_anomalies.py new file mode 100644 index 0000000..246483d --- /dev/null +++ b/tests/core/test_anomalies.py @@ -0,0 +1,148 @@ +"""Metric anomalies (#5740): what counts as a departure, and how one is shown +on Alerts → Anomalies, the Overview and the weekly report. The detection, +lapsing and learning passes run against Postgres in +tests/integration/test_anomalies.py.""" +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace +from zoneinfo import ZoneInfo + +from quart import render_template, session + +from steward.core import mail +from steward.core.anomalies import DEPART, AnomalyView, departs +from steward.core.norms import Norm +from tests.test_style_tokens import hex_colours + +NOW = datetime.now(timezone.utc) +NORM = Norm(kind="day", slot=14, usual_hold=10.0, usual_peak=20.0, days_seen=7) + + +def test_a_departure_is_above_the_usual_hold_by_the_ratio_and_the_floor(): + assert departs(50.0, NORM, floor=5) + assert not departs(10.0 * DEPART, NORM, floor=0) # at the line is not over it + assert not departs(14.0, NORM, floor=5) # over ×1.25, but by less than the floor + assert departs(15.5, NORM, floor=5) + assert not departs(0.3, Norm("day", 1, 0.1, 0.2, 7), floor=2) # 0.1% → 0.3% is never news + + +def _anomaly(**kw): + a = dict(id="a1", source_module="host_agent", resource_name="nas", metric_name="cpu_pct", + started_at=NOW - timedelta(hours=3), last_started_at=NOW - timedelta(hours=1), + ended_at=None, occurrences=1, peak_level=82.0, peak_at=NOW, usual_hold=30.0, + usual_peak=45.0, review="to_review", reviewed_by=None, reviewed_at=None, note=None) + a.update(kw) + return SimpleNamespace(**a) + + +def _view(href="/hosts/h1", **kw): + return AnomalyView(_anomaly(**kw), None, href) + + +def test_a_rows_link_opens_its_page_over_a_range_that_reaches_it(): + assert _view().link == "/hosts/h1?range=24h" + assert _view(last_started_at=NOW - timedelta(days=3)).link == "/hosts/h1?range=7d" + assert _view(last_started_at=NOW - timedelta(days=12)).link == "/hosts/h1?range=30d" + assert _view(href=None).link is None + + +async def _page(app, role, to_review, past): + async with app.test_request_context("/alerts/anomalies"): + session["user_id"], session["user_role"] = "u1", role + return await render_template("alerts/anomalies.html", review_days=14, + to_review=[{"v": v, "opens_group": i == 0, "group": "Today"} + for i, v in enumerate(to_review)], + past=[{"v": v, "opens_group": i == 0, "group": "Older"} + for i, v in enumerate(past)]) + + +async def test_the_page_lists_what_waits_for_review_first(app): + waiting = _view(resource_name="waiting-box") + done = _view(id="a2", resource_name="done-box", review="normal", reviewed_by="ann", + reviewed_at=NOW, ended_at=NOW - timedelta(minutes=5), + last_started_at=NOW - timedelta(minutes=30)) + page = await _page(app, "operator", [waiting], [done]) + assert "1 to review" in page + assert page.index("waiting-box") < page.index("done-box") + assert "Metric anomaly" in page and "Marked normal by ann" in page + assert "Ongoing since" in page and "usual for that hour" in page + assert 'action="/alerts/anomalies/a1/review"' in page # to review: buttons + assert 'action="/alerts/anomalies/a2/review"' not in page # reviewed: none + assert 'href="/hosts/h1?range=24h"' in page + assert ">Today<" in page and ">Older<" in page + assert not hex_colours(page), hex_colours(page) + + +async def test_a_viewer_sees_anomalies_but_cannot_mark_them(app): + page = await _page(app, "viewer", [_view()], []) + assert "nas" in page and "This is normal" not in page and "Real problem" not in page + + +async def test_a_repeat_says_how_often_and_when_the_latest_was(app): + v = _view(occurrences=5, ended_at=NOW - timedelta(minutes=10)) + page = await _page(app, "operator", [v], []) + assert "5 times since" in page and ", last" in page + + +async def test_a_quiet_page_says_nothing_waits(app): + page = await _page(app, "admin", [], []) + assert "Nothing to review" in page and "within 14 days" in page + + +async def _widget(app, logged_in, anomalies, to_review): + async with app.test_request_context("/status/widget"): + if logged_in: + session["user_id"] = "u1" + return await render_template("status/widget.html", entries=[], up=0, down=0, pending=0, + total_hosts=1, alerts=[], anomalies=anomalies, + to_review=to_review, widget_id="1") + + +async def test_the_overview_warns_of_an_open_anomaly_and_counts_those_to_review(app): + page = await _widget(app, True, [_view()], 3) + assert 'Metric anomaly' in page + assert "nas · CPU" in page + assert '3 metric anomalies to review' in page + assert "No monitors configured" not in page + shared = await _widget(app, False, [_view()], 1) + assert "1 metric anomaly to review" in shared and 'href="/alerts/anomalies"' not in shared + + +async def test_a_quiet_overview_has_no_review_line(app): + page = await _widget(app, True, [], 0) + assert "to review" not in page + + +REPORT = {"generated_at": datetime(2026, 10, 11, 12, 0, tzinfo=timezone.utc), + "uptime_summary": [], "active_alerts": [], "alert_events": {"fired": 0, "resolved": 0}, + "top_routers": [], "hosts_down": [], + "anomalies": { + "seen": [{"name": "nas — CPU", "review": "to_review", "occurrences": 2, + "started_at": datetime(2026, 10, 6, 7, 0, tzinfo=timezone.utc), + "last_started_at": datetime(2026, 10, 9, 7, 10, tzinfo=timezone.utc), + "ended_at": datetime(2026, 10, 9, 7, 40, tzinfo=timezone.utc), + "level": "held 82%, usual 30%"}], + "lapsed": [{"name": "web01 — Load", "review": "lapsed", "occurrences": 1, + "started_at": datetime(2026, 9, 20, 3, 0, tzinfo=timezone.utc), + "last_started_at": datetime(2026, 9, 20, 3, 0, tzinfo=timezone.utc), + "ended_at": datetime(2026, 9, 20, 4, 0, tzinfo=timezone.utc), + "level": "held 9.1, usual 2.0"}], + "to_review": 1}} + + +def test_the_weekly_report_lists_the_weeks_anomalies_in_the_readers_zone(): + to = mail.Recipient("ann@example.com", ZoneInfo("America/New_York"), False, "ann") + _, text, html = mail.report_email({"base_url": "https://steward.example"}, REPORT, to) + for part in (text, html): + assert "nas — CPU" in part and "2 times since Tue 6 Oct 2026 03:00 EDT" in part + assert "Fri 9 Oct 2026 03:10 EDT – Fri 9 Oct 2026 03:40 EDT" in part + assert "web01 — Load" in part and "lapsed unreviewed" in part + assert "1 anomaly waits to be marked normal or real" in part + assert "https://steward.example/alerts/anomalies" in part + assert "METRIC ANOMALIES THIS WEEK" in text + + +def test_a_week_without_anomalies_leaves_the_section_out(): + to = mail.Recipient("ann@example.com", timezone.utc, False, "ann") + quiet = {**REPORT, "anomalies": {"seen": [], "lapsed": [], "to_review": 0}} + _, text, html = mail.report_email({}, quiet, to) + assert "anomal" not in (text + html).lower() diff --git a/tests/integration/test_anomalies.py b/tests/integration/test_anomalies.py new file mode 100644 index 0000000..7d3a815 --- /dev/null +++ b/tests/integration/test_anomalies.py @@ -0,0 +1,267 @@ +"""Integration: a metric out of its normal is an anomaly to review (#5740). + +The operator's rulings (System 182): an anomaly stays "to review" until +someone marks it, even after it ends; a repeat joins the row still waiting; +an ended one nobody reviews lapses after the review window, an ongoing one +never; and only an anomaly marked normal is learned. +Requires STEWARD_DATABASE_URL. +""" +from __future__ import annotations + +import asyncio +import os +import uuid +from datetime import datetime, timedelta, timezone + +import pytest + +pytestmark = pytest.mark.integration + +METRIC = "cpu_pct" +NOW = datetime(2026, 10, 12, 10, 0, tzinfo=timezone.utc) + + +@pytest.fixture +def app(): + if not os.environ.get("STEWARD_DATABASE_URL"): + pytest.skip("needs Postgres") + from steward.app import create_app + return create_app(testing=False) + + +@pytest.fixture +def src(): + """A source of the test's own, learned for its run. A detection pass judges + every learned key, so another test's keys must not be learned here; within + a test, resources are reported at times far enough apart not to overlap.""" + from steward.core import norms + name = f"anom{uuid.uuid4().hex[:8]}" + norms.learn_normal(name, METRIC, 5) + yield name + norms.LEARNED.pop((name, METRIC), None) + + +def _norms(src: str, res: str, hold: float = 10.0, peak: float = 20.0): + """A learned day normal for every hour: holds `hold`, peaks at `peak`.""" + from steward.models.norms import MetricNorm + return [MetricNorm(source_module=src, resource_name=res, metric_name=METRIC, slot=h, + kind="day", usual_hold=hold, usual_peak=peak, days_seen=7, updated_at=NOW) + for h in range(24)] + + +async def _report(s, src: str, res: str, until: datetime, value: float): + """A sample a minute over the window before `until`, at `value`.""" + from steward.core.anomalies import WINDOW + from steward.models.metrics import PluginMetric, PluginMetricLatest + n = int(WINDOW.total_seconds() // 60) + s.add_all([PluginMetric(source_module=src, resource_name=res, metric_name=METRIC, + value=value, recorded_at=until - timedelta(minutes=m)) + for m in range(n)]) + await s.merge(PluginMetricLatest(source_module=src, resource_name=res, metric_name=METRIC, + value=value, recorded_at=until)) + + +def _run(app, src: str, res: str, steps, *, seed: bool = True): + """Seed the normal (unless `seed` is false), then for each (at, value) + report the window before `at` and run one detection pass at `at`. Returns + the key's anomalies, oldest first, with their spans.""" + from sqlalchemy import select + from steward.core.anomalies import detect + from steward.models.norms import MetricAnomaly, MetricAnomalySpan + + async def _go(): + async with app.db_sessionmaker() as s: + if seed: + async with s.begin(): + s.add_all(_norms(src, res)) + for at, value in steps: + async with s.begin(): + await _report(s, src, res, at, value) + async with s.begin(): + await detect(s, at) + rows = list((await s.execute( + select(MetricAnomaly).where(MetricAnomaly.resource_name == res) + .order_by(MetricAnomaly.started_at))).scalars()) + spans = {a.id: list((await s.execute( + select(MetricAnomalySpan).where(MetricAnomalySpan.anomaly_id == a.id) + .order_by(MetricAnomalySpan.started_at))).scalars()) for a in rows} + return rows, spans + + return asyncio.run(_go()) + + +def _res(tag: str) -> str: + return f"{tag}-{uuid.uuid4().hex[:8]}" + + +def test_a_level_within_its_normal_is_no_anomaly(app, src): + # Usual hold 10: 12 is under ×1.25, and 14 is over it but not by the floor of 5. + rows, _ = _run(app, src, _res("calm"), [(NOW, 12.0), (NOW + timedelta(minutes=30), 14.0)]) + assert rows == [] + + +def test_a_step_change_stays_open_and_to_review(app, src): + from steward.core.anomalies import TO_REVIEW + res = _res("step") + rows, spans = _run(app, src, res, [(NOW + timedelta(minutes=m), 50.0) for m in (0, 30, 60, 90)]) + (a,) = rows + assert a.review == TO_REVIEW and a.ended_at is None and a.occurrences == 1 + assert a.peak_level == 50.0 and a.usual_hold == 10.0 and a.usual_peak == 20.0 + assert a.started_at == NOW - timedelta(minutes=15) + assert len(spans[a.id]) == 1 and spans[a.id][0].ended_at is None + + +def test_one_that_ended_unwatched_is_still_to_review_and_a_repeat_joins_it(app, src): + from steward.core.anomalies import TO_REVIEW + res = _res("repeat") + back = NOW + timedelta(minutes=30) + rows, spans = _run(app, src, res, [ + (NOW, 60.0), (back, 10.0), # out, then back + (NOW + timedelta(hours=3), 40.0), # again, later + (NOW + timedelta(hours=3, minutes=30), 10.0), + (NOW + timedelta(hours=3, minutes=40), 45.0), # back out within a window + ]) + (a,) = rows + assert a.review == TO_REVIEW + # Two occurrences: the third departure came back within one window of + # the second ending, so it continues the second. + assert a.occurrences == 2 and len(spans[a.id]) == 2 + first, second = spans[a.id] + assert first.ended_at == back and first.peak_level == 60.0 + # The continuation's window still held some of the 10s, so the + # occurrence's peak stays its first level. + assert second.ended_at is None and second.peak_level == 40.0 + assert a.peak_level == 60.0 and a.last_started_at == second.started_at + + +def test_an_ended_anomaly_lapses_after_the_window_and_an_ongoing_one_never(app, src): + from sqlalchemy import select + from steward.core.anomalies import LAPSED, TO_REVIEW, detect, lapse + from steward.models.norms import MetricAnomaly + ended, going = _res("ended"), _res("going") + end = NOW + timedelta(minutes=30) + _run(app, src, ended, [(NOW, 60.0), (end, 10.0)]) + _run(app, src, going, [(NOW + timedelta(days=2), 60.0)]) + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + # The ongoing one keeps reporting high, so it stays open. + await _report(s, src, going, end + timedelta(days=30), 60.0) + await detect(s, end + timedelta(days=30)) + async with s.begin(): + await lapse(s, end + timedelta(days=13), 14) + mid = {a.resource_name: a.review for a in (await s.execute( + select(MetricAnomaly).where(MetricAnomaly.resource_name.in_((ended, going))) + )).scalars()} + async with s.begin(): + await lapse(s, end + timedelta(days=30), 14) + s.expire_all() + late = {a.resource_name: (a.review, a.ended_at) for a in (await s.execute( + select(MetricAnomaly).where(MetricAnomaly.resource_name.in_((ended, going))) + )).scalars()} + return mid, late + + mid, late = asyncio.run(_go()) + assert mid == {ended: TO_REVIEW, going: TO_REVIEW} + assert late[ended][0] == LAPSED + assert late[going] == (TO_REVIEW, None) + + +def test_a_repeat_after_review_is_a_new_row(app, src): + from steward.core.anomalies import REAL, TO_REVIEW, mark + res = _res("again") + rows, _ = _run(app, src, res, [(NOW, 60.0), (NOW + timedelta(minutes=30), 10.0)]) + + async def _mark(): + async with app.db_sessionmaker() as s: + async with s.begin(): + await mark(s, rows[0].id, REAL, "tester", "disk filled") + + asyncio.run(_mark()) + rows, _ = _run(app, src, res, [(NOW + timedelta(hours=5), 60.0)], seed=False) + assert [(a.review, a.occurrences) for a in rows] == [(REAL, 1), (TO_REVIEW, 1)] + assert rows[0].note == "disk filled" and rows[0].reviewed_by == "tester" + + +def test_to_review_is_listed_first(app, src): + from steward.core.anomalies import NORMAL, TO_REVIEW, list_anomalies, mark + older, newer = _res("older"), _res("newer") + old_rows, _ = _run(app, src, older, [(NOW, 60.0), (NOW + timedelta(minutes=30), 10.0)]) + new_rows, _ = _run(app, src, newer, [(NOW + timedelta(hours=2), 60.0), + (NOW + timedelta(hours=2, minutes=30), 10.0)]) + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + await mark(s, new_rows[0].id, NORMAL, "tester") + got = await list_anomalies(s, limit=1000) + return [a.resource_name for a in got if a.resource_name in (older, newer)] + + assert asyncio.run(_go()) == [older, newer] + assert old_rows[0].review == TO_REVIEW + + +@pytest.mark.parametrize("review,learned_hold", [ + ("normal", 90.0), ("real", 10.0), ("lapsed", 10.0), ("to_review", 10.0)]) +def test_only_an_anomaly_marked_normal_is_learned(app, src, review, learned_hold): + """Seven days at 10, but 14:00–15:00 held 90 on three of them, each under + one anomaly. Only when it is marked normal does 14:00 learn 90.""" + from steward.core.norms import norms_at, rebuild_norms + from steward.models.metrics import PluginMetricHourly + from steward.models.norms import MetricAnomaly, MetricAnomalySpan + res = _res(f"learn-{review}") + hot_days = (1, 2, 3) + rows = [] + for d in range(1, 8): + for h in range(24): + bucket = NOW.replace(hour=0) - timedelta(days=d) + timedelta(hours=h) + v = 90.0 if (d in hot_days and h == 14) else 10.0 + rows.append(PluginMetricHourly(source_module=src, resource_name=res, metric_name=METRIC, + bucket=bucket, value_avg=v, value_max=v, value_p95=v, + sample_count=60)) + first = NOW.replace(hour=14) - timedelta(days=max(hot_days)) + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + s.add_all(rows) + a = MetricAnomaly(source_module=src, resource_name=res, metric_name=METRIC, + started_at=first, last_started_at=first, occurrences=3, + ended_at=first + timedelta(hours=1), peak_level=90.0, + peak_at=first, usual_hold=10.0, usual_peak=10.0, review=review) + s.add(a) + await s.flush() + s.add_all([MetricAnomalySpan( + anomaly_id=a.id, started_at=NOW.replace(hour=14) - timedelta(days=d), + ended_at=NOW.replace(hour=15) - timedelta(days=d), peak_level=90.0) + for d in hot_days]) + async with s.begin(): + await rebuild_norms(s, NOW) + got = await norms_at(s, [(src, res, METRIC)], NOW.replace(hour=14)) + return got[(src, res, METRIC)] + + norm = asyncio.run(_go()) + assert norm.usual_hold == learned_hold + + +def test_cleanup_prunes_reviewed_anomalies_and_never_one_to_review(app, src): + from sqlalchemy import select + from steward.core.anomalies import REAL, mark, prune + from steward.models.norms import MetricAnomaly + kept, gone = _res("kept"), _res("gone") + _run(app, src, kept, [(NOW, 60.0), (NOW + timedelta(minutes=30), 10.0)]) + later = NOW + timedelta(days=1) + rows, _ = _run(app, src, gone, [(later, 60.0), (later + timedelta(minutes=30), 10.0)]) + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + await mark(s, rows[0].id, REAL, "tester") + async with s.begin(): + await prune(s, NOW + timedelta(days=2)) + return {a.resource_name for a in (await s.execute( + select(MetricAnomaly).where(MetricAnomaly.resource_name.in_((kept, gone))) + )).scalars()} + + assert asyncio.run(_go()) == {kept}