FabledSteward/steward/core/norms.py

"""Each metric's normal, by time slot (#5739, milestone 513).

A fixed cutoff suits some resources and not others: a router that serves
error pages answers 4xx all day, and a host that backs up at 02:00 runs hot
for that hour every night. So Steward learns, for each metric that opts in,
what it usually does in each hour, and a reading is judged against the
normal for its own hour (System 182 "Metric Norms and Anomalies").

What a slot's normal is:
- usual_hold: the level the slot's hours usually held. Each hour's p95
  (plugin_metrics_hourly.value_p95, #5738), at its 90th percentile across
  past days. A two-minute spike doesn't raise it; an hour spent high does.
- usual_peak: the same for each hour's max, so a page can say how high the
  slot usually goes.

The slots:
- An hour of the day (24 slots) once a key has DAY_MIN_DAYS of history.
- An hour of the week (168 slots) once its history reaches back WEEK_MIN_DAYS,
  so a weekly job is learned too.
- Slots are UTC hours. Each takes in its neighbours (±NEIGHBOURS), so a job
  that starts a few minutes late, or an hour off across a daylight-saving
  change, stays inside its normal.

Only metrics registered with learn_normal() are learned: a host agent alone
reports thousands of per-core, per-disk and per-interface series, and most
have no use for a normal.

An hour an anomaly covers is learned only once someone marks the anomaly
normal (steward.core.anomalies): a level that jumps and stays is never
absorbed on its own, and a lapsed anomaly is never learned.
"""
from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

from sqlalchemy import ARRAY, String, bindparam, delete, select, text, tuple_

from steward.models.norms import MetricNorm

DAY_MIN_DAYS = 5        # days of history before an hour-of-day normal is trusted
WEEK_MIN_DAYS = 21      # history reaching back this far learns hour-of-week instead
DAY_WINDOW = timedelta(days=28)    # what an hour-of-day normal learns from
WEEK_WINDOW = timedelta(days=56)   # what an hour-of-week normal learns from: 8 of each slot
NEIGHBOURS = 1          # hours either side a slot takes in
USUAL = 0.9             # "usual" is this percentile across past days: one odd day in ten is ignored


@dataclass(frozen=True)
class Learned:
    """A metric that learns a normal. `floor` is the smallest departure,
    in the metric's own unit, worth calling an anomaly (#5740), so that
    0.1% → 0.3% of a share or 2 ms → 4 ms of latency is never news."""
    source_module: str
    metric_name: str
    floor: float


LEARNED: dict[tuple[str, str], Learned] = {}


def learn_normal(source_module: str, metric_name: str, floor: float) -> None:
    """Have Steward learn this metric's normal for every resource reporting
    it. A plugin calls this from setup(), as it calls prune_by_age."""
    LEARNED[(source_module, metric_name)] = Learned(source_module, metric_name, float(floor))


@dataclass(frozen=True)
class Norm:
    kind: str           # "day" or "week"
    slot: int
    usual_hold: float
    usual_peak: float
    days_seen: int


def slot_of(at: datetime, kind: str) -> int:
    """The slot `at` falls in: its UTC hour of the day (0–23), or its hour of
    the week counted from Monday 00:00 UTC (0–167)."""
    at = at.astimezone(timezone.utc)
    if kind == "week":
        return at.weekday() * 24 + at.hour
    return at.hour


# One statement learns every normal: the hours of each learned key (rolled up,
# plus the raw window not rolled up yet) less those an anomaly covers that
# nobody marked normal (#5740), which kind of slot its history supports, then
# each slot's usual hold and peak across past days, every hour counted in its
# own slot and its neighbours'.
_REBUILD = text("""
WITH all_hours AS (
    SELECT source_module AS s, resource_name AS r, metric_name AS m, bucket AS b,
           COALESCE(value_p95, value_avg) AS hold, value_max AS peak
    FROM plugin_metrics_hourly
    WHERE bucket >= :week_since AND source_module || '/' || metric_name = ANY(:keys)
    UNION ALL
    SELECT source_module, resource_name, metric_name, date_trunc('hour', recorded_at),
           percentile_cont(0.95) WITHIN GROUP (ORDER BY value), max(value)
    FROM plugin_metrics
    WHERE recorded_at >= :week_since AND recorded_at < :this_hour
      AND source_module || '/' || metric_name = ANY(:keys)
    GROUP BY 1, 2, 3, 4
),
hours AS (
    SELECT * FROM all_hours h
    WHERE NOT EXISTS (
        SELECT 1 FROM metric_anomaly_spans sp
        JOIN metric_anomalies a ON a.id = sp.anomaly_id
        WHERE a.review <> 'normal'
          AND a.source_module = h.s AND a.resource_name = h.r AND a.metric_name = h.m
          AND sp.started_at < h.b + interval '1 hour'
          AND COALESCE(sp.ended_at, :now) > h.b)
),
keyed AS (
    SELECT s, r, m, count(DISTINCT date_trunc('day', b)) AS days, min(b) AS first
    FROM hours GROUP BY s, r, m
),
kinds AS (
    SELECT s, r, m, days, CASE WHEN first <= :week_ready THEN 'week' ELSE 'day' END AS kind
    FROM keyed WHERE days >= CAST(:day_min_days AS int)
),
slotted AS (
    SELECT h.s, h.r, h.m, k.kind, k.days, h.hold, h.peak,
           CASE k.kind
               WHEN 'week' THEN (EXTRACT(ISODOW FROM h.b AT TIME ZONE 'UTC')::int - 1) * 24
                                + EXTRACT(HOUR FROM h.b AT TIME ZONE 'UTC')::int
               ELSE EXTRACT(HOUR FROM h.b AT TIME ZONE 'UTC')::int
           END AS slot,
           CASE k.kind WHEN 'week' THEN 168 ELSE 24 END AS slots
    FROM hours h JOIN kinds k ON k.s = h.s AND k.r = h.r AND k.m = h.m
    WHERE k.kind = 'week' OR h.b >= :day_since
)
INSERT INTO metric_norms (source_module, resource_name, metric_name, slot, kind,
                          usual_hold, usual_peak, days_seen, updated_at)
SELECT s, r, m, (slot + n.off + slots) % slots, kind,
       percentile_cont(CAST(:usual AS float8)) WITHIN GROUP (ORDER BY hold),
       percentile_cont(CAST(:usual AS float8)) WITHIN GROUP (ORDER BY peak),
       days, :now
FROM slotted CROSS JOIN generate_series(-CAST(:neighbours AS int), CAST(:neighbours AS int)) AS n(off)
GROUP BY s, r, m, (slot + n.off + slots) % slots, kind, days
""").bindparams(bindparam("keys", type_=ARRAY(String)))


async def rebuild_norms(session, now: datetime | None = None) -> int:
    """Relearn every normal from history, replacing the last pass's. Runs
    inside the caller's transaction. Returns how many slot rows it wrote."""
    now = now or datetime.now(timezone.utc)
    await session.execute(delete(MetricNorm))
    keys = [f"{s}/{m}" for s, m in LEARNED]
    if not keys:
        return 0
    result = await session.execute(_REBUILD, {
        "keys": keys,
        "week_since": now - WEEK_WINDOW,
        "day_since": now - DAY_WINDOW,
        "this_hour": now.replace(minute=0, second=0, microsecond=0),
        "week_ready": now - timedelta(days=WEEK_MIN_DAYS),
        "day_min_days": DAY_MIN_DAYS,
        "usual": USUAL,
        "neighbours": NEIGHBOURS,
        "now": now,
    })
    return result.rowcount or 0


async def norms_at(session, keys: list[tuple[str, str, str]],
                   at: datetime) -> dict[tuple[str, str, str], Norm]:
    """{(source, resource, metric): its normal for the slot `at` falls in}.
    A key with no normal yet (too little history, or not learned) is absent,
    and its caller falls back to the fixed cutoffs."""
    if not keys:
        return {}
    day, week = slot_of(at, "day"), slot_of(at, "week")
    rows = (await session.execute(
        select(MetricNorm)
        .where(tuple_(MetricNorm.source_module, MetricNorm.resource_name,
                      MetricNorm.metric_name).in_(keys))
        .where(((MetricNorm.kind == "day") & (MetricNorm.slot == day))
               | ((MetricNorm.kind == "week") & (MetricNorm.slot == week)))
    )).scalars().all()
    return {(n.source_module, n.resource_name, n.metric_name):
            Norm(n.kind, n.slot, n.usual_hold, n.usual_peak, n.days_seen) for n in rows}


async def run_norms(app) -> None:
    """The hourly pass: relearn every normal in one transaction."""
    async with app.db_sessionmaker() as session:
        async with session.begin():
            await rebuild_norms(session)