From 0945d78290b2382ac8ba8aedc1d40ff8100975ed Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 15 Oct 2026 13:17:34 -0400 Subject: [PATCH] feat(mcp): understand and judge a metric anomaly: what happened, what was usual, what else went on (#5837) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit get_anomaly gives each occurrence against what is usual at that hour, the metric around the newest one with its normal, what else happened on the resource's host around each occurrence (alert transitions, Ansible runs, other anomalies there and a count elsewhere, plus plugins' own events through a new mcp_events hook: the host agent's maintenance jobs and downtime, Docker's container starts, deaths and OOM kills), and how earlier anomalies on the same metric were judged. get_metric_normal reads the learned slots and the level that would be flagged. review_anomaly takes one id or several, all or nothing, with a note. get_anomaly_settings and (admin) set_anomaly_settings read and change the review window and each normal's floor, as Settings → Thresholds does. metric_series.points takes an optional `until`; core.alerts gains host_of_resource; floor_setting names its floor. Co-Authored-By: Claude Opus 5.5 --- docs/core/mcp.md | 14 +- docs/plugins/writing-a-plugin.md | 15 + plugins/docker/__init__.py | 5 + plugins/docker/mcp_tools.py | 14 +- plugins/host_agent/__init__.py | 5 + plugins/host_agent/mcp_tools.py | 29 +- steward/core/alerts.py | 12 +- steward/core/metric_series.py | 13 +- steward/core/norms.py | 14 +- steward/mcp/server.py | 11 +- steward/mcp/tools/_events.py | 170 ++++++++++ steward/mcp/tools/anomalies.py | 334 ++++++++++++++++++- tests/integration/test_mcp_anomaly_review.py | 174 ++++++++++ tests/mcp/test_anomaly_review.py | 105 ++++++ 14 files changed, 881 insertions(+), 34 deletions(-) create mode 100644 steward/mcp/tools/_events.py create mode 100644 tests/integration/test_mcp_anomaly_review.py create mode 100644 tests/mcp/test_anomaly_review.py diff --git a/docs/core/mcp.md b/docs/core/mcp.md index 62a47b1..72f2c9a 100644 --- a/docs/core/mcp.md +++ b/docs/core/mcp.md @@ -75,7 +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) | +| Metric anomalies | `list_anomalies` (metrics that left their normal for the hour, to review first), `get_anomaly` (one anomaly with each occurrence against what is usual at that hour, the metric around it, what else happened on its host — alerts, Ansible runs, container deaths and starts, maintenance, reboots, other anomalies — and how earlier ones on the same metric were judged), `get_metric_normal` (the learned normal slot by slot, and the level that would be flagged), `get_anomaly_settings` (the review window and each floor) | | 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` | @@ -88,11 +88,12 @@ 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) | +| Metric anomalies | `review_anomaly` (mark one or several normal, so their hours are learned, or real, with a note saying why) | ### Admin: destructive and remediation -`set_alert_action` (the playbook a rule runs when it fires), `delete_monitor`, +`set_anomaly_settings` (the anomaly review window and each normal's floor, as +Settings → Thresholds sets them), `set_alert_action` (the playbook a rule runs when it fires), `delete_monitor`, `delete_alert_rule`, `prune_docker`, `run_maintenance` (update a host now, rebooting only if the update needs it, as the host page's Update now does). @@ -150,7 +151,10 @@ Every tool follows these conventions (`steward/mcp/tools/_shape.py`): Core tools: add an `async def` to a module in `steward/mcp/tools/` and list it in that module's `register()` with its minimum role and `read_only`. A plugin -exposes `register_mcp_tools(register)`, and optionally `mcp_overview(db)` to -contribute findings to `get_overview`. See +exposes `register_mcp_tools(register)`, optionally `mcp_overview(db)` to +contribute findings to `get_overview`, and optionally +`mcp_events(db, host_id, host_name, since, until)`: what it knows happened on a +host in a window, as a list of `steward.mcp.tools._events.event(...)`, which +`get_anomaly` shows beside an anomaly. See [Writing a plugin](../plugins/writing-a-plugin.md). A tool that changes anything calls `audit_mcp(...)` from `steward/mcp/tools/_audit.py`. diff --git a/docs/plugins/writing-a-plugin.md b/docs/plugins/writing-a-plugin.md index 6089cc4..0921500 100644 --- a/docs/plugins/writing-a-plugin.md +++ b/docs/plugins/writing-a-plugin.md @@ -391,6 +391,21 @@ async def mcp_overview(db) -> list[dict]: - `severity` is `"critical"` (something is down or about to be) or `"warning"` (degraded, or heading that way). Report problems only; a healthy plugin returns `[]`. - It is called only while the plugin is enabled, with its own session. If it raises, the overview still returns, and names the plugin under `unavailable`. +### What happened on a host + +`get_anomaly` shows, beside a metric anomaly, what else happened on its host around each occurrence. A plugin that records events on a host adds them by exporting an optional `mcp_events(db, host_id, host_name, since, until)` that returns events built with `steward.mcp.tools._events.event`: + +```python +# plugins/myplugin/__init__.py +async def mcp_events(db, host_id, host_name, since, until) -> list[dict]: + from steward.mcp.tools._events import event + return [event(r.at, "backup_job", f"Backup {r.name}: {r.state}", until=r.finished_at) + for r in await _jobs_on(db, host_id, since, until)] +``` + +- Return what happened between `since` and `until`, oldest first: something that lasted gives `until`. Keep it to tens of rows; the caller keeps the first 40. +- It is called only while the plugin is enabled, with its own session. If it raises, the anomaly is still returned without the plugin's events. The host agent (maintenance jobs, reboots) and Docker (container starts, deaths, OOM kills) are the examples to read. + --- ## Shipping a Plugin diff --git a/plugins/docker/__init__.py b/plugins/docker/__init__.py index 367cdb0..7183969 100644 --- a/plugins/docker/__init__.py +++ b/plugins/docker/__init__.py @@ -74,6 +74,11 @@ async def mcp_overview(db) -> list[dict]: return await findings(db) +async def mcp_events(db, host_id: str, host_name: str, since, until) -> list[dict]: + from .mcp_tools import mcp_events as events + return await events(db, host_id, host_name, since, until) + + def get_nav() -> list[dict]: # Surfaces in the sidebar's Infrastructure group (fleet-wide container view). return [{"label": "Docker", "href": "/plugins/docker/"}] diff --git a/plugins/docker/mcp_tools.py b/plugins/docker/mcp_tools.py index 700e408..ec87c00 100644 --- a/plugins/docker/mcp_tools.py +++ b/plugins/docker/mcp_tools.py @@ -27,7 +27,7 @@ from steward.models.users import UserRole from . import service as docker_svc from .service import PROBLEM_SAYS, PROBLEM_SEVERITY, PROBLEMS, container_problems, problem_since from .dedup import dedup_by_container_id -from .models import DockerContainer, DockerDiskUsage, DockerSwarmNode, DockerSwarmService +from .models import DockerContainer, DockerDiskUsage, DockerEvent, DockerSwarmNode, DockerSwarmService from .swarm_view import build_swarm_services, replaced_tasks, task_time CRASH_LOOP_DEATHS = docker_svc.CRASH_LOOP_DEATHS @@ -751,6 +751,18 @@ async def prune_docker(host: str, what: str, check: bool = True) -> dict: "next": f'wait_for_run(run_id="{run.id}")'} +async def mcp_events(db, host_id: str, host_name: str, since, until) -> list[dict]: + """Container lifecycle events on a host in a window (get_anomaly): starts + (with the image, so a deploy shows), deaths with their exit, OOM kills, + restarts and health changes.""" + from steward.mcp.tools._events import event + rows = (await db.execute(select(DockerEvent).where( + DockerEvent.host_id == host_id, DockerEvent.at >= since, DockerEvent.at <= until) + .order_by(DockerEvent.at).limit(40))).scalars() + return [event(e.at, "container", f"{e.container_name} {e.event}" + (f" ({e.detail})" if e.detail else ""), + subject=f"{host_name}/{e.container_name}") for e in rows] + + def register_tools(register) -> None: for fn in (list_containers, get_container, list_swarm_services, search_container_logs, get_docker_disk_usage): diff --git a/plugins/host_agent/__init__.py b/plugins/host_agent/__init__.py index 9e3cc7a..370a7b6 100644 --- a/plugins/host_agent/__init__.py +++ b/plugins/host_agent/__init__.py @@ -62,6 +62,11 @@ async def mcp_overview(db) -> list[dict]: return await findings(db) +async def mcp_events(db, host_id: str, host_name: str, since, until) -> list[dict]: + from .mcp_tools import mcp_events as events + return await events(db, host_id, host_name, since, until) + + def get_alert_metrics() -> dict[str, list[str]]: """Host metrics ingest records for alert rules (resource = the host name; disk_used_pct per mount, resource ":"). seconds_since_seen diff --git a/plugins/host_agent/mcp_tools.py b/plugins/host_agent/mcp_tools.py index e8ae378..cc2879a 100644 --- a/plugins/host_agent/mcp_tools.py +++ b/plugins/host_agent/mcp_tools.py @@ -15,8 +15,8 @@ from steward.mcp.tools.overview import finding from steward.models.hosts import Host from steward.models.users import UserRole -from . import cadence, maintenance -from .models import (HostAgentRegistration, HostMaintenanceJob, HostMaintenancePlan, +from . import cadence, downtime, maintenance +from .models import (HostAgentRegistration, HostDowntime, HostMaintenanceJob, HostMaintenancePlan, HostMaintenanceRun) # A registration this old that has never checked in is a failed install, not one in progress. @@ -295,6 +295,31 @@ async def mcp_overview(db) -> list[dict]: return out +async def mcp_events(db, host_id: str, host_name: str, since, until) -> list[dict]: + """What the host agent knows happened on a host in a window (get_anomaly): + maintenance jobs, and downtime (reboots, agent stops, lost contact).""" + from steward.mcp.tools._events import event + out = [] + jobs = (await db.execute(select(HostMaintenanceJob).where( + HostMaintenanceJob.host_id == host_id, HostMaintenanceJob.created_at <= until, + (HostMaintenanceJob.finished_at.is_(None)) | (HostMaintenanceJob.finished_at >= since)) + .order_by(HostMaintenanceJob.created_at).limit(20))).scalars() + for job in jobs: + out.append(event(job.created_at, "maintenance", f"Maintenance: {maintenance.summary(job)}", + until=job.finished_at, ids={"job_id": job.id})) + now = now_utc() + rows = (await db.execute(select(HostDowntime).where( + HostDowntime.host_id == host_id, HostDowntime.started_at <= until, + (HostDowntime.ended_at.is_(None)) | (HostDowntime.ended_at >= since)) + .order_by(HostDowntime.started_at).limit(20))).scalars() + for row in rows: + seen = downtime.describe(row, now) + out.append(event(row.started_at, "downtime", + f"{seen['label']} ({int(seen['seconds'])} s{', still going' if seen['ongoing'] else ''})", + until=row.ended_at)) + return out + + def register_tools(register) -> None: register(list_agents, role=UserRole.viewer, read_only=True) register(list_maintenance_plans, role=UserRole.viewer, read_only=True) diff --git a/steward/core/alerts.py b/steward/core/alerts.py index 716db40..533369e 100644 --- a/steward/core/alerts.py +++ b/steward/core/alerts.py @@ -84,13 +84,19 @@ def is_heartbeat(rule) -> bool: _HEARTBEAT.source_module, _HEARTBEAT.metric_name) +def host_of_resource(source_module: str, resource: str) -> str | None: + """The host whose agent carries this series, or None for a source no + agent feeds (a monitor, a Traefik router).""" + host_of = _AGENT_FED.get(source_module) + return host_of(resource) if host_of is not None else None + + def host_behind(rule) -> str | None: """The host whose agent feeds `rule`, or None: a rule on another source, or the heartbeat itself (agent offline is the one that speaks).""" - host_of = _AGENT_FED.get(rule.source_module) - if host_of is None or is_heartbeat(rule): + if is_heartbeat(rule): return None - return host_of(rule.resource_name) + return host_of_resource(rule.source_module, rule.resource_name) async def offline_agents(session) -> set[str]: diff --git a/steward/core/metric_series.py b/steward/core/metric_series.py index 812234a..3548c4c 100644 --- a/steward/core/metric_series.py +++ b/steward/core/metric_series.py @@ -143,8 +143,9 @@ async def backfill_latest(sessionmaker, log) -> int: async def points(db, source_module: str, resource: str, metric: str, since: datetime, - *, target: int = 120) -> list[tuple[datetime, float]]: - """(time, average) points for one metric since `since`, oldest first. + *, target: int = 120, until: datetime | None = None) -> list[tuple[datetime, float]]: + """(time, average) points for one metric since `since` (up to `until`, + else now), oldest first. The part of the range older than the raw window comes from the hourly rollup; the rest is bucket-averaged in SQL (buckets no wider than an hour @@ -167,14 +168,17 @@ async def points(db, source_module: str, resource: str, metric: str, since: date PluginMetricHourly.resource_name == resource, PluginMetricHourly.metric_name == metric, PluginMetricHourly.bucket >= since, - PluginMetricHourly.bucket < raw_cutoff, + PluginMetricHourly.bucket < (min(raw_cutoff, until) if until else raw_cutoff), ) .order_by(PluginMetricHourly.bucket) ) out.extend((b, float(v)) for b, v in hourly) raw_since = max(since, raw_cutoff) - width_s = bucket_seconds(raw_since, target) + if until is not None and raw_since >= until: + return out + width_s = (bucket_seconds(raw_since, target) if until is None + else max(60, int((until - raw_since).total_seconds() / target))) if since < raw_cutoff: width_s = min(width_s, 3600) bucket = func.date_bin( @@ -189,6 +193,7 @@ async def points(db, source_module: str, resource: str, metric: str, since: date PluginMetric.resource_name == resource, PluginMetric.metric_name == metric, PluginMetric.recorded_at >= raw_since, + *((PluginMetric.recorded_at < until,) if until is not None else ()), ) .group_by(bucket) .order_by(bucket) diff --git a/steward/core/norms.py b/steward/core/norms.py index 4a79928..a6b2cd2 100644 --- a/steward/core/norms.py +++ b/steward/core/norms.py @@ -72,6 +72,11 @@ class Learned: def least_departure(self) -> float: return float(self.floor() if callable(self.floor) else self.floor) + @property + def floor_name(self) -> str | None: + """The NORM_FLOORS setting the floor is read from, if it is one.""" + return getattr(self.floor, "floor_name", None) + LEARNED: dict[tuple[str, str], Learned] = {} @@ -87,9 +92,14 @@ def learn_normal(source_module: str, metric_name: str, floor: float | Callable[[ def floor_setting(app, name: str) -> Callable[[], float]: """A floor read from Settings → Thresholds (norms.floor.) at each - pass, so a saved change applies at once.""" + pass, so a saved change applies at once. The function carries the + floor's name (`floor_name`), so a reader can say which setting it is.""" from steward.core.settings import floor_of - return lambda: floor_of(app.config.get("THRESHOLDS", {}), name) + + def read() -> float: + return floor_of(app.config.get("THRESHOLDS", {}), name) + read.floor_name = name + return read @dataclass(frozen=True) diff --git a/steward/mcp/server.py b/steward/mcp/server.py index ca25c19..962cb87 100644 --- a/steward/mcp/server.py +++ b/steward/mcp/server.py @@ -79,6 +79,9 @@ class StewardMCPServer(MCPServer): self.tool_plugins: dict[str, str] = {} # plugin name -> its `mcp_overview(db)`: findings it adds to get_overview. self.overview_sources: dict[str, object] = {} + # plugin name -> its `mcp_events(db, host_id, host_name, since, until)`: + # what it knows happened on a host, for get_anomaly. + self.event_sources: dict[str, object] = {} # ── registration ───────────────────────────────────────────────────────── def register_tool(self, fn, *, role: UserRole, read_only: bool, plugin: str | None = None) -> None: @@ -103,7 +106,9 @@ class StewardMCPServer(MCPServer): def register_plugin(self, name: str, module) -> None: """Add a plugin's tools via its optional `register_mcp_tools(register)`, - and its optional `mcp_overview(db)` as a get_overview findings source. + its optional `mcp_overview(db)` as a get_overview findings source, and + its optional `mcp_events(db, host_id, host_name, since, until)` as a + source of what happened on a host (get_anomaly). Idempotent per plugin (hot-reload drops its previous tools first) and tolerant: a raising hook is logged and leaves the plugin with no tools, @@ -114,6 +119,10 @@ class StewardMCPServer(MCPServer): overview = getattr(module, "mcp_overview", None) if callable(overview): self.overview_sources[name] = overview + self.event_sources.pop(name, None) + events = getattr(module, "mcp_events", None) + if callable(events): + self.event_sources[name] = events hook = getattr(module, "register_mcp_tools", None) if hook is None: return diff --git a/steward/mcp/tools/_events.py b/steward/mcp/tools/_events.py new file mode 100644 index 0000000..2e1ee18 --- /dev/null +++ b/steward/mcp/tools/_events.py @@ -0,0 +1,170 @@ +"""What else was going on: events on a host, or one resource, in a window +(#5837), so an anomaly comes with its circumstances. + +Core gathers alert transitions, Ansible runs and other metric anomalies. A +plugin adds its own through `mcp_events(db, host_id, host_name, since, until)` +in its __init__.py (the host agent its maintenance jobs and downtime, Docker +its container events); each returns a list of `event(...)`. +""" +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import datetime + +from sqlalchemy import and_, or_, select + +from steward.alerts.service import watches_host +from steward.core.alerts import host_of_resource +from steward.mcp._context import mcp_app +from steward.models.alerts import AlertEvent, AlertRule +from steward.models.ansible import AnsibleRun +from steward.models.ansible_inventory import AnsibleTarget +from steward.models.hosts import Host +from steward.models.monitors import Monitor +from steward.models.norms import MetricAnomaly, MetricAnomalySpan + +from ._shape import db_session, iso + +logger = logging.getLogger(__name__) + +PER_SOURCE = 40 # events any one source contributes to one window + + +def event(at: datetime | None, kind: str, what: str, *, until: datetime | None = None, + subject: str | None = None, ids: dict | None = None) -> dict: + """One thing that happened. `until` for something that lasted (a run, a + reboot); `subject` names what it happened to when that isn't the host.""" + return {"at": iso(at), "until": iso(until), "kind": kind, "what": what, + "subject": subject, "ids": ids or {}} + + +@dataclass(frozen=True) +class Place: + """Where a metric's resource lives: its host, when it has one.""" + source: str + resource: str + host_id: str | None = None + host_name: str | None = None + + +async def place_of(db, source: str, resource: str) -> Place: + """The host behind a resource: an agent-fed series names it ("nas:/", + "nas/web"), a monitor may be pinned to one; a Traefik router has none.""" + name = host_of_resource(source, resource) + host = None + if name: + host = (await db.execute(select(Host).where(Host.name == name).limit(1))).scalar_one_or_none() + else: + monitor = (await db.execute(select(Monitor).where( + Monitor.type == source, Monitor.name == resource).limit(1))).scalar_one_or_none() + if monitor is not None and monitor.host_id: + host = await db.get(Host, monitor.host_id) + return Place(source, resource, host.id if host else None, host.name if host else None) + + +def _overlaps(start, end): + """A row lasting from `start` to `end` (None: still going) overlaps the window.""" + def clause(since, until): + return and_(start <= until, or_(end.is_(None), end >= since)) + return clause + + +async def alert_events(db, place: Place, since: datetime, until: datetime) -> list[dict]: + """Alert transitions on the resource itself, or anything its host + carries: its agent series, its containers, its monitors.""" + monitor_keys = set() + if place.host_id: + monitor_keys = {(t, n) for t, n in (await db.execute( + select(Monitor.type, Monitor.name).where(Monitor.host_id == place.host_id))).all()} + rows = (await db.execute( + select(AlertEvent, AlertRule).join(AlertRule, AlertRule.id == AlertEvent.rule_id) + .where(AlertEvent.transitioned_at >= since, AlertEvent.transitioned_at <= until) + .order_by(AlertEvent.transitioned_at).limit(500))).all() + out = [] + for ev, rule in rows: + mine = (rule.source_module, rule.resource_name) == (place.source, place.resource) + if not mine and not (place.host_name and watches_host(rule, place.host_name, monitor_keys)): + continue + note = f" ({ev.note})" if ev.note else "" + out.append(event(ev.transitioned_at, "alert", f"{rule.name}: {ev.from_state} → {ev.to_state}{note}", + subject=rule.resource_name, ids={"rule_id": rule.id})) + return out[:PER_SOURCE] + + +async def ansible_runs(db, place: Place, since: datetime, until: datetime) -> list[dict]: + """Ansible runs that touched the host while the window was open (by the + hosts in their recap: a run still going hasn't said yet).""" + if not place.host_id: + return [] + names = {place.host_name} | set((await db.execute( + select(AnsibleTarget.name).where(AnsibleTarget.host_id == place.host_id))).scalars()) + runs = (await db.execute( + select(AnsibleRun).where(_overlaps(AnsibleRun.started_at, AnsibleRun.finished_at)(since, until)) + .order_by(AnsibleRun.started_at).limit(200))).scalars() + out = [] + for r in runs: + if not names & set(((r.results or {}).get("hosts") or {}).keys()): + continue + check = " in check mode" if (r.params or {}).get("check") else "" + out.append(event(r.started_at, "ansible_run", + f"Ansible {r.playbook_path}{check}: {r.status.value}", + until=r.finished_at, ids={"run_id": r.id})) + return out[:PER_SOURCE] + + +def _same_host(source: str, resource: str, place: Place) -> bool: + if (source, resource) == (place.source, place.resource): + return True + return bool(place.host_name) and host_of_resource(source, resource) == place.host_name + + +async def other_anomalies(db, place: Place, anomaly_id: str, since: datetime, + until: datetime) -> tuple[list[dict], int]: + """Other anomalies going on in the window: those on the same host as + events, and how many there were anywhere else (many at once points at + something shared: the network, storage, Steward itself).""" + rows = (await db.execute( + select(MetricAnomalySpan, MetricAnomaly) + .join(MetricAnomaly, MetricAnomaly.id == MetricAnomalySpan.anomaly_id) + .where(MetricAnomaly.id != anomaly_id, + _overlaps(MetricAnomalySpan.started_at, MetricAnomalySpan.ended_at)(since, until)) + .order_by(MetricAnomalySpan.started_at).limit(500))).all() + here, elsewhere = [], 0 + for span, a in rows: + if _same_host(a.source_module, a.resource_name, place): + here.append(event(span.started_at, "anomaly", + f"{a.resource_name} {a.metric_name} anomaly (peak {round(span.peak_level, 3)}, " + f"usual {round(a.usual_hold, 3)}; {a.review})", + until=span.ended_at, subject=a.resource_name, ids={"anomaly_id": a.id})) + else: + elsewhere += 1 + return here[:PER_SOURCE], elsewhere + + +async def plugin_events(place: Place, since: datetime, until: datetime) -> list[dict]: + """What enabled plugins know happened on the host (`mcp_events`), each + in its own session so one that fails can't spoil the others' reads.""" + server = getattr(mcp_app(), "mcp_server", None) + if server is None or not place.host_id: + return [] + out = [] + for plugin, source in sorted((getattr(server, "event_sources", None) or {}).items()): + if not server.plugin_enabled(plugin): + continue + try: + async with db_session() as db: + out.extend((await source(db, place.host_id, place.host_name, since, until))[:PER_SOURCE]) + except Exception: + logger.exception("MCP events source of plugin %r failed", plugin) + return out + + +async def around(db, place: Place, anomaly_id: str, since: datetime, + until: datetime) -> tuple[list[dict], int]: + """Everything above for one window, oldest first, and the count of + anomalies elsewhere.""" + anomalies, elsewhere = await other_anomalies(db, place, anomaly_id, since, until) + found = (await alert_events(db, place, since, until) + await ansible_runs(db, place, since, until) + + anomalies + await plugin_events(place, since, until)) + return sorted(found, key=lambda e: e["at"] or ""), elsewhere diff --git a/steward/mcp/tools/anomalies.py b/steward/mcp/tools/anomalies.py index 8038394..43178c4 100644 --- a/steward/mcp/tools/anomalies.py +++ b/steward/mcp/tools/anomalies.py @@ -2,14 +2,25 @@ 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.mcp._context import current_caller +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, rnd +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: @@ -68,29 +79,320 @@ async def list_anomalies(review: str = "to_review", limit: int = 50) -> dict: 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). +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: optional, kept with the review (say what it was). - A lapsed anomaly may still be marked. Audited as " via MCP". + 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: - 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 + 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: - register_tool(list_anomalies, role=UserRole.viewer, read_only=True) + 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) diff --git a/tests/integration/test_mcp_anomaly_review.py b/tests/integration/test_mcp_anomaly_review.py new file mode 100644 index 0000000..a259b9d --- /dev/null +++ b/tests/integration/test_mcp_anomaly_review.py @@ -0,0 +1,174 @@ +"""Integration (#5837): understanding and judging a metric anomaly through +MCP, against real Postgres. + +A host's CPU anomaly with two occurrences. Around the newest one, the host +had an alert, a container death, a reboot, an Ansible run and a memory +anomaly; an earlier anomaly on the same metric was marked normal with a note. +get_anomaly brings all of that back; review_anomaly marks two at once; +get_metric_normal reads the learned slots; set_anomaly_settings changes a +floor and the review window, and the running app follows. + +Plugin rows are seeded with plain SQL (a second import of a plugin's models +raises "Table already defined"). +""" +from __future__ import annotations + +import asyncio +import os +import uuid +from datetime import datetime, timedelta, timezone + +import pytest + +pytestmark = pytest.mark.integration + + +@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) + + +def test_an_anomaly_with_its_circumstances_judged_through_mcp(app): + from sqlalchemy import select, text + from steward.core.api_tokens import Caller + from steward.core.norms import slot_of + from steward.mcp import _context + from steward.mcp.tools import anomalies as tools + from steward.models.alerts import AlertEvent, AlertOperator, AlertRule + from steward.models.ansible import AnsibleRun, AnsibleRunStatus + from steward.models.audit import AuditEvent + from steward.models.hosts import Host + from steward.models.metrics import PluginMetric + from steward.models.norms import MetricAnomaly, MetricAnomalySpan, MetricNorm + from steward.models.users import User, UserRole + + tag = uuid.uuid4().hex[:8] + host_name, other_host = f"an-{tag}", f"an-other-{tag}" + now = datetime.now(timezone.utc).replace(second=0, microsecond=0) + new_start, new_end = now - timedelta(hours=3), now - timedelta(hours=2) + old_start, old_end = now - timedelta(hours=26), now - timedelta(hours=25, minutes=30) + ids = {} + + def anomaly(resource, metric, start, end, **kw): + return MetricAnomaly(source_module="host_agent", resource_name=resource, metric_name=metric, + started_at=start, last_started_at=start, ended_at=end, + occurrences=kw.pop("occurrences", 1), peak_level=kw.pop("peak", 90.0), peak_at=start, usual_hold=20.0, + usual_peak=35.0, **kw) + + async def _seed(): + async with app.db_sessionmaker() as s: + async with s.begin(): + user = User(username=f"an-{tag}", email=f"an-{tag}@test", password_hash="x", + role=UserRole.admin) + host, other = Host(name=host_name, address="10.9.9.7"), Host(name=other_host, address="10.9.9.8") + s.add_all([user, host, other]) + await s.flush() + ids.update(user=user.id, host=host.id) + + cpu = anomaly(host_name, "cpu_pct", old_start, new_end, occurrences=2, peak=93.0) + cpu.last_started_at = new_start + before = anomaly(host_name, "cpu_pct", now - timedelta(days=9), now - timedelta(days=9, hours=-1), + review="normal", reviewed_by="op", note="nightly backup", peak=70.0) + mem = anomaly(host_name, "mem_used_pct", new_start + timedelta(minutes=10), new_end, peak=95.0) + far = anomaly(other_host, "cpu_pct", new_start, new_end) + s.add_all([cpu, before, mem, far]) + await s.flush() + ids.update(cpu=cpu.id, before=before.id, mem=mem.id) + s.add_all([ + MetricAnomalySpan(anomaly_id=cpu.id, started_at=old_start, ended_at=old_end, peak_level=88.0), + MetricAnomalySpan(anomaly_id=cpu.id, started_at=new_start, ended_at=new_end, peak_level=93.0), + MetricAnomalySpan(anomaly_id=mem.id, started_at=new_start + timedelta(minutes=10), + ended_at=new_end, peak_level=95.0), + MetricAnomalySpan(anomaly_id=far.id, started_at=new_start, ended_at=new_end, peak_level=90.0), + MetricNorm(source_module="host_agent", resource_name=host_name, metric_name="cpu_pct", + slot=slot_of(new_start, "day"), kind="day", usual_hold=20.0, usual_peak=35.0, + days_seen=12, updated_at=now), + ]) + for minutes in range(-240, -60, 5): + at = now + timedelta(minutes=minutes) + s.add(PluginMetric(source_module="host_agent", resource_name=host_name, + metric_name="cpu_pct", recorded_at=at, + value=90.0 if new_start <= at < new_end else 18.0)) + + rule = AlertRule(name=f"CPU high {tag}", source_module="host_agent", resource_name=host_name, + metric_name="cpu_pct", operator=AlertOperator.gt, threshold=85, + created_by=user.id) + s.add(rule) + await s.flush() + s.add(AlertEvent(rule_id=rule.id, from_state="ok", to_state="firing", metric_value=91.0, + transitioned_at=new_start + timedelta(minutes=20))) + s.add(AnsibleRun(playbook_path="maintenance/backup.yml", source_name="steward-builtin", + status=AnsibleRunStatus.success, started_at=new_start - timedelta(minutes=10), + finished_at=new_start + timedelta(minutes=5), params={"check": False}, + results={"hosts": {host_name: {"ok": 3}}, "failures": []})) + await s.execute(text( + "INSERT INTO docker_events (id, host_id, container_name, event, detail, at) " + "VALUES (:id, :h, 'web', 'die', 'exit 137', :at)"), + {"id": str(uuid.uuid4()), "h": host.id, "at": new_start + timedelta(minutes=15)}) + await s.execute(text( + "INSERT INTO host_downtime (id, host_id, kind, started_at, ended_at, created_at) " + "VALUES (:id, :h, 'reboot', :s, :e, now())"), + {"id": str(uuid.uuid4()), "h": host.id, "s": new_end - timedelta(minutes=5), + "e": new_end + timedelta(minutes=1)}) + + async def _go(): + await _seed() + _context._app_ctx.set(app) + _context._caller_ctx.set(Caller(user_id=ids["user"], username=f"an-{tag}", + role=UserRole.admin, token_id=9)) + + got = await tools.get_anomaly(ids["cpu"]) + assert got["place"] == {"host": host_name, "host_id": ids["host"]} + assert got["occurrences_total"] == 2 and got["truncated"] is False + newest, oldest = got["occurrences"] + assert newest["held_s"] == 3600 and newest["peak_level"] == 93.0 + assert newest["usual"]["usual_hold"] == 20.0 and newest["usual"]["flag_above"] >= 25.0 + kinds = {e["kind"] for e in newest["around"]} + assert {"alert", "ansible_run", "container", "downtime", "anomaly"} <= kinds + whats = " | ".join(e["what"] for e in newest["around"]) + assert "web die (exit 137)" in whats and "maintenance/backup.yml" in whats + assert any(e["kind"] == "anomaly" and e["ids"]["anomaly_id"] == ids["mem"] for e in newest["around"]) + assert newest["anomalies_elsewhere"] >= 1 + assert not [e for e in oldest["around"] if e["kind"] in ("container", "ansible_run")] + points = got["series"]["points"] + assert max(p[1] for p in points) == 90.0 and min(p[1] for p in points) == 18.0 + assert any(p[2] == 20.0 for p in points) + [prior] = got["precedent"] + assert prior["anomaly_id"] == ids["before"] and prior["note"] == "nightly backup" + + normal = await tools.get_metric_normal("host_agent", host_name, "cpu_pct") + assert normal["learned"] is True and normal["kind"] == "day" + assert normal["floor_setting"] == "host_cpu_pct" + [slot] = normal["slots"] + assert slot["usual_hold"] == 20.0 and slot["label"].endswith(":00 UTC") + with pytest.raises(ValueError, match="doesn't learn a normal"): + await tools.get_metric_normal("host_agent", host_name, "nope_pct") + + with pytest.raises(ValueError, match="Nothing was marked"): + await tools.review_anomaly([ids["cpu"], "no-such-id"], "real") + marked = await tools.review_anomaly([ids["cpu"], ids["mem"]], "real", note="the backup job hung") + assert {m["anomaly_id"] for m in marked["marked"]} == {ids["cpu"], ids["mem"]} + assert marked["reviewed_by"] == f"an-{tag} via MCP" + + try: + changed = await tools.set_anomaly_settings(review_days=21, floors={"host_cpu_pct": 15}) + assert changed["review_days"] == 21 + floor = {f["name"]: f for f in changed["floors"]}["host_cpu_pct"] + assert floor["value"] == 15.0 and "host_agent/cpu_pct" in floor["metrics"] + assert app.config["THRESHOLDS"]["floors"]["host_cpu_pct"] == 15.0 + finally: + await tools.set_anomaly_settings(review_days=14, floors={"host_cpu_pct": 10}) + + async with app.db_sessionmaker() as s: + audits = list((await s.execute(select(AuditEvent.action).where( + AuditEvent.user_id == ids["user"]).order_by(AuditEvent.timestamp))).scalars()) + rows = {a.id: a for a in (await s.execute(select(MetricAnomaly).where( + MetricAnomaly.id.in_([ids["cpu"], ids["mem"]])))).scalars()} + assert audits.count("metric_anomaly.marked_real") == 2 and audits.count("settings.saved") == 2 + assert {a.review for a in rows.values()} == {"real"} + assert {a.note for a in rows.values()} == {"the backup job hung"} + + asyncio.run(_go()) diff --git a/tests/mcp/test_anomaly_review.py b/tests/mcp/test_anomaly_review.py new file mode 100644 index 0000000..ba281ce --- /dev/null +++ b/tests/mcp/test_anomaly_review.py @@ -0,0 +1,105 @@ +"""The anomaly review tools' pure parts and refusals (#5837): slot names, the +line an anomaly opens at, how far a series reaches, the plugin events hook, +and calls refused before anything is read or changed. + +get_anomaly's circumstances, a bulk review and a settings change run against +Postgres in tests/integration/test_mcp_anomaly_review.py. +""" +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace as NS + +import pytest + +from steward.core import alerts +from steward.mcp.tools import _events, anomalies +from steward.models.users import UserRole + +NOW = datetime(2026, 10, 15, 12, 0, tzinfo=timezone.utc) + + +def test_slots_are_named_as_a_reader_would(): + assert anomalies.slot_label("day", 3) == "03:00 UTC" + assert anomalies.slot_label("week", 0) == "Mon 00:00 UTC" + assert anomalies.slot_label("week", 6 * 24 + 23) == "Sun 23:00 UTC" + + +def test_an_anomaly_opens_above_the_larger_of_the_factor_and_the_floor(): + assert anomalies.flag_above(20.0, 10.0) == 30.0 # 20 + floor beats 20 × 1.25 + assert anomalies.flag_above(100.0, 10.0) == 125.0 # the factor beats the floor + + +def test_a_series_reaches_as_far_as_the_occurrence_lasted_within_one_to_six_hours(): + assert anomalies.series_pad(NOW - timedelta(minutes=20), NOW, NOW) == timedelta(hours=1) + assert anomalies.series_pad(NOW - timedelta(hours=3), NOW, NOW) == timedelta(hours=3) + assert anomalies.series_pad(NOW - timedelta(days=2), None, NOW) == timedelta(hours=6) + + +def test_an_event_says_when_what_and_until(): + e = _events.event(NOW, "downtime", "Rebooted (60 s)", until=NOW + timedelta(minutes=1)) + assert e == {"at": "2026-10-15T12:00:00Z", "until": "2026-10-15T12:01:00Z", "kind": "downtime", + "what": "Rebooted (60 s)", "subject": None, "ids": {}} + + +def test_a_resource_names_its_host_through_the_source_that_feeds_it(): + alerts.agent_fed("test_src", lambda r: r.split(":", 1)[0]) + try: + assert alerts.host_of_resource("test_src", "nas:/var") == "nas" + assert alerts.host_of_resource("no_such_source", "nas") is None + finally: + alerts._AGENT_FED.pop("test_src", None) + + +def test_same_host_covers_the_resource_itself_and_its_hosts_other_series(): + alerts.agent_fed("test_src", lambda r: r.split(":", 1)[0]) + try: + place = _events.Place("test_src", "nas", "h1", "nas") + assert _events._same_host("test_src", "nas:/var", place) + assert not _events._same_host("test_src", "other", place) + bare = _events.Place("traefik", "router-a") + assert _events._same_host("traefik", "router-a", bare) + assert not _events._same_host("traefik", "router-b", bare) + finally: + alerts._AGENT_FED.pop("test_src", None) + + +async def test_refusals_change_nothing(): + with pytest.raises(ValueError, match="at least one anomaly"): + await anomalies.review_anomaly([], "normal") + with pytest.raises(ValueError, match="At most"): + await anomalies.review_anomaly([f"id-{i}" for i in range(anomalies.REVIEW_MAX + 1)], "real") + with pytest.raises(ValueError, match="No floor named bogus"): + await anomalies.set_anomaly_settings(floors={"bogus": 1}) + with pytest.raises(ValueError, match="0 or more"): + await anomalies.set_anomaly_settings(floors={"host_cpu_pct": -1}) + with pytest.raises(ValueError, match="Nothing to change"): + await anomalies.set_anomaly_settings() + with pytest.raises(ValueError, match="doesn't learn a normal"): + await anomalies.get_metric_normal("no_source", "x", "no_metric") + + +def test_anomaly_tools_are_registered_with_their_roles(app): + mcp = app.mcp_server + for name, role, read_only in (("list_anomalies", UserRole.viewer, True), + ("get_anomaly", UserRole.viewer, True), + ("get_metric_normal", UserRole.viewer, True), + ("get_anomaly_settings", UserRole.viewer, True), + ("review_anomaly", UserRole.operator, False), + ("set_anomaly_settings", UserRole.admin, False)): + assert mcp.tool_roles[name] is role, name + assert mcp._tool_manager.get_tool(name).annotations.read_only_hint is read_only, name + + +def test_a_plugin_adds_events_through_its_hook(app): + async def events(db, host_id, host_name, since, until): + return [] + + mcp = app.mcp_server + mcp.register_plugin("events_test", NS(mcp_events=events)) + try: + assert mcp.event_sources["events_test"] is events + mcp.register_plugin("events_test", NS()) + assert "events_test" not in mcp.event_sources + finally: + mcp.event_sources.pop("events_test", None)