FabledSteward/steward/core/anomalies.py

"""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 _gate_open(learned, key: tuple[str, str, str], reported: dict) -> bool:
    """Whether `key` may be judged: ungated, or its resource's gate metric
    averaged at least the gate's least over the window."""
    if learned.gate is None:
        return True
    metric, least = learned.gate
    return reported.get((key[0], key[1], metric), 0.0) >= (least() if callable(least) else least)


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


def against(value: float, norm: Norm | None, floor: float, unit: str) -> tuple[bool, str]:
    """(departs, what is usual) for a page's reading past its fixed cutoff
    (#5741, #5742). Without a normal it departs, and the cutoff stands; within
    one it is usual for that resource, and the page says so. `unit` is "%"
    (one decimal) or " ms" (whole)."""
    if norm is None:
        return True, ""
    usual = f"{norm.usual_hold:.1f}{unit}" if unit == "%" else f"{norm.usual_hold:.0f}{unit}"
    return departs(value, norm, floor), f"usual {usual} at this hour"


def graded(value: float | None, kind: str, thresholds: dict, norm: Norm | None,
           floor: float, unit: str) -> tuple[str, str | None]:
    """(status class, why) for a behaviour reading on a page (#5742): graded
    against Settings → Thresholds, except that one past its cutoff but within
    its own normal for the hour is "ok", and `why` says what is usual. Only
    for behaviour (CPU, memory, load, latency); a failure keeps its cutoff
    whatever its history (System 182)."""
    from steward.core.formatting import status_class
    cls = status_class(value, kind, thresholds)
    if cls in ("warn", "crit") and norm is not None:
        out, said = against(value, norm, floor, unit)
        if not out:
            return "ok", said
    return cls, None


# 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 (strpos(l.resource_name, ':') = 0 OR NOT l.source_module || '/' || l.metric_name = ANY(:whole))
  AND l.recorded_at >= :since AND v.level IS NOT NULL
""").bindparams(bindparam("keys", type_=ARRAY(String)), bindparam("whole", 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}, for
    every learned metric and every metric one is gated on. A metric learned
    without its parts reads its whole resources only."""
    keys = sorted({f"{s}/{m}" for s, m in LEARNED}
                  | {f"{k.source_module}/{k.gate[0]}" for k in LEARNED.values() if k.gate})
    if not keys:
        return {}
    whole = sorted(f"{k.source_module}/{k.metric_name}" for k in LEARNED.values() if not k.parts)
    rows = await session.execute(_LEVELS, {"keys": keys, "whole": whole,
                                           "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)
    reported = await levels_now(session, now)
    levels = {key: level for key, level in reported.items()
              if (key[0], key[2]) in LEARNED and _gate_open(LEARNED[(key[0], key[2])], key, reported)}
    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.least_departure)
        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, lost its normal or fell
    # below its gate 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]}