feat(anomalies): a metric out of its normal is an anomaly to review (#5740)
#5740 "A metric that departs from its normal is recorded as an anomaly to review, and is findable after it ends", milestone 513. - Every minute each learned key's last 15 minutes is judged against its hour's normal: above the usual hold x1.25 and by the metric's floor opens an occurrence, which ends when the level comes back. - metric_anomalies holds one row per item to review; a repeat while it still waits joins it as an occurrence (metric_anomaly_spans). - An ended row nobody reviews within anomalies.review_days (14, Settings -> Thresholds) lapses; an ongoing one never does. - Learning skips every hour an occurrence covers unless its anomaly was marked normal: no normal is learned without review. - Alerts -> Anomalies lists those to review first, then the reviewed and lapsed under day dividers; marking is operator/admin and audited. - The Overview widget and get_overview warn of open ones and count those to review; the host page shows its own; the weekly report lists the week's and the lapsed; MCP list_anomalies and review_anomaly. - Reviewed and lapsed rows are pruned at data.retention_days, and kept in backups like alert history. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -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]}
|
||||
|
||||
Reference in New Issue
Block a user