bvandeusenandClaude Opus 5.5
2b97c36fd4
The host agent writes each maintenance job once it ends, and each stretch a host was away (a reboot, a crash, lost contact, the agent restarting), kept up to date until it is back. Docker judges each snapshot: a unit on a new image, or a container recreated, is a deploy; deaths past the Thresholds crash-loop count are one crash-loop entry that later deaths extend; single restarts are recorded but hidden. An anomaly has one entry whose type follows its review: only those judged real are shown by default. A source whose record changes keeps one entry for it (record_or_update). Plugins register the records their entries point to, so an aged-out job or downtime reads "Details no longer kept". The first fill runs once per source (timeline.filled), the host agent's jobs and downtime included. get_anomaly's "what else was going on" reads the timeline for the host or resource instead of querying alerts, runs, maintenance, downtime and container deaths itself; Docker's own source keeps only starts, stops and health changes. A new kind, Observed, is for what Steward saw happen with no rule or person behind it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
185 lines
10 KiB
Python
185 lines
10 KiB
Python
"""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
|
|
from steward.timeline.record import from_alert, from_run, record
|
|
|
|
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()
|
|
fired = AlertEvent(rule_id=rule.id, from_state="ok", to_state="firing", metric_value=91.0,
|
|
transitioned_at=new_start + timedelta(minutes=20))
|
|
run = 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": []})
|
|
s.add_all([fired, run])
|
|
await s.flush()
|
|
# The timeline, as the alert, the run and the plugins write it.
|
|
await from_alert(s, rule, fired)
|
|
await from_run(s, run)
|
|
record(s, type="docker.restarted", at=new_start + timedelta(minutes=15),
|
|
summary=f"web on {host_name}: died (exit 137)", hosts=[host_name],
|
|
subject=f"{host_name}/web")
|
|
record(s, type="host.rebooted", at=new_end - timedelta(minutes=5),
|
|
until=new_end + timedelta(minutes=1), summary=f"{host_name}: rebooted for 6m",
|
|
hosts=[host_name], subject=host_name)
|
|
# A container event the timeline doesn't hold, from Docker's own source.
|
|
await s.execute(text(
|
|
"INSERT INTO docker_events (id, host_id, container_name, event, detail, at) "
|
|
"VALUES (:id, :h, 'web', 'health_change', 'healthy→unhealthy', :at)"),
|
|
{"id": str(uuid.uuid4()), "h": host.id, "at": new_start + timedelta(minutes=16)})
|
|
|
|
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.fired", "ansible.unchanged", "docker.restarted", "host.rebooted", "container",
|
|
"anomaly"} <= kinds
|
|
whats = " | ".join(e["what"] for e in newest["around"])
|
|
assert "died (exit 137)" in whats and "maintenance/backup.yml" in whats
|
|
assert "web health_change (healthy→unhealthy)" 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.unchanged")]
|
|
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())
|