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 collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

from sqlalchemy import ARRAY, Boolean, DateTime, 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. It may
    be a function, read at each pass, for a floor that is a setting (#5742).

    `gate` is (metric, least): the resource is judged only while that metric
    of its own averages at least `least` over the window, as a router's
    error share means nothing over a handful of requests (#5741). `least`
    may be a function, read at each pass, for a minimum that is a setting. `since`:
    history before it is not learned from, for a metric whose meaning
    changed then. `parts=False` learns only whole resources, not their parts
    ("<host>:core3"): a host's CPU has a normal, each of its cores doesn't."""
    source_module: str
    metric_name: str
    floor: float | Callable[[], float]
    gate: tuple[str, float | Callable[[], float]] | None = None
    since: datetime | None = None
    parts: bool = True

    @property
    def least_departure(self) -> float:
        return float(self.floor() if callable(self.floor) else self.floor)

    @property
    def floor_name(self) -> str | None:
        """The NORM_FLOORS setting the floor is read from, if it is one."""
        return getattr(self.floor, "floor_name", None)


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


def learn_normal(source_module: str, metric_name: str, floor: float | Callable[[], float], *,
                 gate: tuple[str, float | Callable[[], float]] | None = None,
                 since: datetime | None = None, parts: bool = True) -> 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, floor if callable(floor) else float(floor), gate, since, parts)


def floor_setting(app, name: str) -> Callable[[], float]:
    """A floor read from Settings → Thresholds (norms.floor.<name>) at each
    pass, so a saved change applies at once. The function carries the
    floor's name (`floor_name`), so a reader can say which setting it is."""
    from steward.core.settings import floor_of

    def read() -> float:
        return floor_of(app.config.get("THRESHOLDS", {}), name)
    read.floor_name = name
    return read


@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 learned AS (
    SELECT key, since, parts
    FROM unnest(CAST(:keys AS text[]), CAST(:starts AS timestamptz[]), CAST(:parts AS boolean[]))
         AS k(key, since, parts)
),
all_hours AS (
    SELECT h.source_module AS s, h.resource_name AS r, h.metric_name AS m, h.bucket AS b,
           COALESCE(h.value_p95, h.value_avg) AS hold, h.value_max AS peak
    FROM plugin_metrics_hourly h
    JOIN learned k ON k.key = h.source_module || '/' || h.metric_name
    WHERE h.bucket >= k.since AND (k.parts OR strpos(h.resource_name, ':') = 0)
    UNION ALL
    SELECT p.source_module, p.resource_name, p.metric_name, date_trunc('hour', p.recorded_at),
           percentile_cont(0.95) WITHIN GROUP (ORDER BY p.value), max(p.value)
    FROM plugin_metrics p
    JOIN learned k ON k.key = p.source_module || '/' || p.metric_name
    WHERE p.recorded_at >= k.since AND p.recorded_at < :this_hour
      AND (k.parts OR strpos(p.resource_name, ':') = 0)
    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)),
               bindparam("starts", type_=ARRAY(DateTime(timezone=True))),
               bindparam("parts", type_=ARRAY(Boolean)))


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))
    if not LEARNED:
        return 0
    week_since = now - WEEK_WINDOW
    result = await session.execute(_REBUILD, {
        "keys": [f"{s}/{m}" for s, m in LEARNED],
        # Each key learns from WEEK_WINDOW back, or from its own `since`.
        "starts": [max(week_since, k.since) if k.since else week_since for k in LEARNED.values()],
        "parts": [k.parts for k in LEARNED.values()],
        "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)