diff --git a/docs/architecture.md b/docs/architecture.md index 3bfeb32..d003bb8 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -105,7 +105,7 @@ A two-minute spike raises the peak and leaves the p95 low; an hour spent at that 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`): +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, has departed. It opens an anomaly once it is also past the usual peak for that hour, or held: the median of the last hour is over the same line. A burst of a few minutes within the usual peak lifts the 15-minute average and opens nothing (`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. diff --git a/steward/core/anomalies.py b/steward/core/anomalies.py index 64e137c..2c2dd3f 100644 --- a/steward/core/anomalies.py +++ b/steward/core/anomalies.py @@ -2,8 +2,12 @@ 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. +the slot's usual hold by DEPART times, and by at least the metric's floor, has +departed. That opens an occurrence once it is also outside what the normal +records in height or in time: higher than the slot's usual peak, or held, with +most of the last HELD above the line. A burst of a few minutes within the +usual peak lifts the average for a while and opens nothing. The occurrence ends +when the level comes back under the line. What the operator ruled (System 182 "Metric Norms and Anomalies"): - A departure is a warning, marked as a metric anomaly, and stays "to review" @@ -28,6 +32,7 @@ from steward.timeline.record import anomaly_changed 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 +HELD = timedelta(hours=1) # a level within the usual peak opens once the median of this is out # 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 @@ -118,6 +123,32 @@ async def levels_now(session, now: datetime) -> dict[tuple[str, str, str], float return {(s, r, m): float(level) for s, r, m, level in rows} +# The median, as a burst moves an average of the same stretch but not this. +_HELD_LEVELS = text(""" +SELECT k.s, k.r, k.m, v.held +FROM unnest(:sources, :resources, :metrics) AS k(s, r, m) +CROSS JOIN LATERAL ( + SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY p.value) AS held FROM plugin_metrics p + WHERE p.source_module = k.s AND p.resource_name = k.r AND p.metric_name = k.m + AND p.recorded_at >= :since AND p.recorded_at <= :now +) v +WHERE v.held IS NOT NULL +""").bindparams(bindparam("sources", type_=ARRAY(String)), bindparam("resources", type_=ARRAY(String)), + bindparam("metrics", type_=ARRAY(String))) + + +async def held_levels(session, keys: list[tuple[str, str, str]], now: datetime) -> dict[tuple[str, str, str], float]: + """{(source, resource, metric): the median of its last HELD}. Asked only of + keys that have departed and stayed within their usual peak, so the longer + read isn't paid for the rest.""" + if not keys: + return {} + rows = await session.execute(_HELD_LEVELS, { + "sources": [k[0] for k in keys], "resources": [k[1] for k in keys], "metrics": [k[2] for k in keys], + "since": now - HELD, "now": now}) + return {(s, r, m): float(held) for s, r, m, held 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.""" @@ -135,10 +166,13 @@ async def detect(session, now: datetime | None = None) -> dict: } opened = closed = 0 changed: list[MetricAnomaly] = [] # opened, joined or ended: their timeline entries follow + out_now = {key for key, level in levels.items() + if key in norms and departs(level, norms[key], LEARNED[(key[0], key[2])].least_departure)} + held = await held_levels(session, [key for key in out_now + if key not in open_now and levels[key] <= norms[key].usual_peak], now) 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.least_departure) + out = key in out_now current = open_now.pop(key, None) if current is not None: span, anomaly = current @@ -148,8 +182,11 @@ async def detect(session, now: datetime | None = None) -> dict: _close(anomaly, span, now) changed.append(anomaly) closed += 1 - elif out: - changed.append(await _occur(session, key, level, norm, now)) + elif out and level > norm.usual_peak: + changed.append(await _occur(session, key, level, norm, now, WINDOW)) + opened += 1 + elif out and departs(held.get(key, 0.0), norm, LEARNED[(key[0], key[2])].least_departure): + changed.append(await _occur(session, key, level, norm, now, HELD)) opened += 1 # An open occurrence whose key stopped reporting, lost its normal or fell # below its gate can't be judged any more: it ends now. @@ -175,11 +212,12 @@ def _close(anomaly: MetricAnomaly, span: MetricAnomalySpan, now: datetime) -> No anomaly.ended_at = now -async def _occur(session, key, level: float, norm: Norm, now: datetime) -> MetricAnomaly: +async def _occur(session, key, level: float, norm: Norm, now: datetime, judged_over: timedelta) -> MetricAnomaly: """A departure on a key with no open occurrence: join the key's anomaly - still to review, or open a new one. Returns the anomaly.""" + still to review, or open a new one, starting where the stretch it was + judged over starts. Returns the anomaly.""" source, resource, metric = key - start = now - WINDOW + start = now - judged_over anomaly = (await session.execute( select(MetricAnomaly) .where(MetricAnomaly.source_module == source, MetricAnomaly.resource_name == resource, diff --git a/steward/mcp/tools/anomalies.py b/steward/mcp/tools/anomalies.py index 5b6a6f0..2937f71 100644 --- a/steward/mcp/tools/anomalies.py +++ b/steward/mcp/tools/anomalies.py @@ -52,7 +52,8 @@ def _row(v: anom.AnomalyView) -> dict: 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). + more than the metric's floor), and either past how high it usually goes + (`usual_peak`) or held there for most of an hour. review: "to_review" (default) — waiting for someone to mark them; "normal" — accepted, their hours are learned; "real" — a real problem, not learned; @@ -263,8 +264,9 @@ def slot_label(kind: str, slot: int) -> str: 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).""" + """The 15-minute level that has departed in a slot whose usual hold is + `norm_hold` (anomalies.departs). It opens an anomaly past the usual peak, + or once held.""" return max(norm_hold * anom.DEPART, norm_hold + floor) @@ -287,8 +289,10 @@ async def get_metric_normal(source_module: str, resource: str, metric: str) -> d (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). + went), `days_seen`, and `flag_above`: the 15-minute average that has + departed there (usual hold ×1.25, and at least `floor` above it). A + departure opens an anomaly once it passes `usual_peak`, or when the median + of the last hour is over `flag_above` too. `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. diff --git a/tests/integration/test_anomalies.py b/tests/integration/test_anomalies.py index 09d9c34..f5411b1 100644 --- a/tests/integration/test_anomalies.py +++ b/tests/integration/test_anomalies.py @@ -104,6 +104,45 @@ def test_a_level_within_its_normal_is_no_anomaly(app, src): assert rows == [] +def _hour_of(app, src: str, res: str, *, level: float, for_minutes: int, then: float = 10.0): + """A detection pass after an hour reading `then`, with its last + `for_minutes` at `level`, against the normal that holds 10 and peaks at 20.""" + from sqlalchemy import select + from steward.core.anomalies import detect + from steward.models.metrics import PluginMetric, PluginMetricLatest + from steward.models.norms import MetricAnomaly + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + s.add_all(_norms(src, res)) + s.add_all([PluginMetric(source_module=src, resource_name=res, metric_name=METRIC, + value=level if m < for_minutes else then, + recorded_at=NOW - timedelta(minutes=m)) for m in range(60)]) + await s.merge(PluginMetricLatest(source_module=src, resource_name=res, metric_name=METRIC, + value=level, recorded_at=NOW)) + async with s.begin(): + await detect(s, NOW) + return list((await s.execute(select(MetricAnomaly).where(MetricAnomaly.resource_name == res))).scalars()) + return asyncio.run(_go()) + + +def test_a_burst_within_its_usual_peak_is_no_anomaly(app, src): + # 18 is over the line (15) and under the usual peak (20), for 15 minutes of the hour. + assert _hour_of(app, src, _res("burst"), level=18.0, for_minutes=15) == [] + + +def test_a_burst_past_its_usual_peak_is_an_anomaly(app, src): + (a,) = _hour_of(app, src, _res("tall"), level=30.0, for_minutes=15) + assert a.started_at == NOW - timedelta(minutes=15) + + +def test_a_level_within_its_usual_peak_is_an_anomaly_once_held(app, src): + # The system running at its peaked rate: never past the peak, but not a burst. + (a,) = _hour_of(app, src, _res("held"), level=18.0, for_minutes=45) + assert a.started_at == NOW - timedelta(hours=1) and a.peak_level == 18.0 + + def test_a_step_change_stays_open_and_to_review(app, src): from steward.core.anomalies import TO_REVIEW res = _res("step")