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, has
departed. That opens an occurrence once it is also outside what the normal
records in height or in time: higher than the metric usually goes (Norm.
usual_height), or held, with most of the last HELD above the line. A burst of
a few minutes within that 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"
  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, field
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
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 its usual height 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

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, and the share stands
    for at least `fewest` events in it."""
    if learned.gate is None:
        return True
    metric, least = learned.gate
    rate = reported.get((key[0], key[1], metric), 0.0)
    if rate < (least() if callable(least) else least):
        return False
    if learned.fewest is None:
        return True
    fewest = learned.fewest() if callable(learned.fewest) else learned.fewest
    return rate * WINDOW.total_seconds() * reported.get(key, 0.0) / 100.0 >= fewest


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}


# 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."""
    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
    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_height], now)
    for key, level in levels.items():
        norm = norms.get(key)
        out = key in out_now
        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)
                changed.append(anomaly)
                closed += 1
        elif out and level > norm.usual_height:
            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.
    for span, anomaly in open_now.values():
        _close(anomaly, span, now)
        changed.append(anomaly)
        closed += 1
    for anomaly in changed:
        await anomaly_changed(session, anomaly)
    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, 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, starting where the stretch it was
    judged over starts. Returns the anomaly."""
    source, resource, metric = key
    start = now - judged_over
    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 anomaly
    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)
    return anomaly


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."""
    ids = list((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)
        .returning(MetricAnomaly.id)
        .execution_options(synchronize_session=False))).scalars())
    if ids:
        for anomaly in (await session.execute(
                select(MetricAnomaly).where(MetricAnomaly.id.in_(ids))
                .execution_options(populate_existing=True))).scalars():
            await anomaly_changed(session, anomaly)
    return len(ids)


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
    await anomaly_changed(session, anomaly)
    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
    # Open tickets that make it a known issue (tickets.service.known_issues).
    tickets: list = field(default_factory=list)

    @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)))
    from steward.tickets.service import known_issues
    known = await known_issues(session, {a.id: ([("anomaly", a.id)], a.resource_name) for a in rows})
    for v in out:
        v.tickets = known.get(v.anomaly.id, [])
    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]}