From 4c84bc85601d1a35f18495e1952dca7ba20ee121 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 14 Oct 2026 12:48:44 -0400 Subject: [PATCH] feat(norms): Steward learns each metric's normal per hour of day, then week (#5739) A metric registered with learn_normal() gets, for every resource reporting it, a usual hold (the 90th percentile across past days of each hour's p95) and a usual peak for each slot: - an hour of the day once 5 days of history exist; - an hour of the week once history reaches back 3 weeks, so weekly jobs are learned too. Each slot takes in the hours either side, so a job that runs a little late stays normal. One SQL statement relearns everything hourly into metric_norms, from the rollup plus the raw window not rolled up yet; norms_at() reads the slot a moment falls in. Nothing is registered yet: Traefik (#5741) and hosts (#5742) opt in. Co-Authored-By: Claude Opus 5.5 --- steward/app.py | 13 ++ steward/backup/contents.py | 2 + steward/core/norms.py | 175 ++++++++++++++++++ .../migrations/versions/0045_metric_norms.py | 33 ++++ steward/models/__init__.py | 3 +- steward/models/norms.py | 27 +++ tests/core/test_norms.py | 27 +++ tests/integration/test_norms.py | 106 +++++++++++ 8 files changed, 385 insertions(+), 1 deletion(-) create mode 100644 steward/core/norms.py create mode 100644 steward/migrations/versions/0045_metric_norms.py create mode 100644 steward/models/norms.py create mode 100644 tests/core/test_norms.py create mode 100644 tests/integration/test_norms.py diff --git a/steward/app.py b/steward/app.py index 7829453..44500a2 100644 --- a/steward/app.py +++ b/steward/app.py @@ -445,6 +445,10 @@ def _register_core_tasks(app: Quart) -> None: from .backup.job import tick await tick(app) + async def run_metric_norms(): + from .core.norms import run_norms + await run_norms(app) + async def run_email_outbox(): from .core.outbox import send_due await send_due(app) @@ -507,6 +511,15 @@ def _register_core_tasks(app: Quart) -> None: interval_seconds=cleanup_interval, run_on_startup=False, ), + # Each learned metric's normal per hour (#5739). Hourly, since a slot + # gains one hour of history an hour; the table outlives a restart, so + # there is nothing to rebuild at startup. + ScheduledTask( + name="metric_norms", + coro_factory=run_metric_norms, + interval_seconds=3600, + run_on_startup=False, + ), ]) # Pull Ansible git sources as each falls due. One task reads the sources diff --git a/steward/backup/contents.py b/steward/backup/contents.py index a17ccdd..3a16df0 100644 --- a/steward/backup/contents.py +++ b/steward/backup/contents.py @@ -30,6 +30,8 @@ KEPT: frozenset[str] = frozenset({ # Rows that are measurements, logs or in-flight work: left out. EXCLUDED: frozenset[str] = frozenset({ "plugin_metrics", "plugin_metric_latest", "plugin_metrics_hourly", + # Learned from the metrics above, so relearned once they fill again. + "metric_norms", "monitor_results", "email_outbox", "email_outbox_events", "password_resets", # Run history; its logs live on disk, not in the database. diff --git a/steward/core/norms.py b/steward/core/norms.py new file mode 100644 index 0000000..287b6f1 --- /dev/null +++ b/steward/core/norms.py @@ -0,0 +1,175 @@ +"""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. +""" +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), 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 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 +), +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) diff --git a/steward/migrations/versions/0045_metric_norms.py b/steward/migrations/versions/0045_metric_norms.py new file mode 100644 index 0000000..593f979 --- /dev/null +++ b/steward/migrations/versions/0045_metric_norms.py @@ -0,0 +1,33 @@ +"""What each learned metric usually does in each time slot (#5739). + +Revision ID: 0045_metric_norms +Revises: 0044_hourly_p95 +Create Date: 2026-10-14 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "0045_metric_norms" +down_revision: Union[str, None] = "0044_hourly_p95" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "metric_norms", + sa.Column("source_module", sa.String(64), primary_key=True), + sa.Column("resource_name", sa.String(255), primary_key=True), + sa.Column("metric_name", sa.String(128), primary_key=True), + sa.Column("slot", sa.SmallInteger, primary_key=True), + sa.Column("kind", sa.String(8), nullable=False), + sa.Column("usual_hold", sa.Float, nullable=False), + sa.Column("usual_peak", sa.Float, nullable=False), + sa.Column("days_seen", sa.Integer, nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + ) + + +def downgrade() -> None: + op.drop_table("metric_norms") diff --git a/steward/models/__init__.py b/steward/models/__init__.py index a5ceb45..fb5de64 100644 --- a/steward/models/__init__.py +++ b/steward/models/__init__.py @@ -6,6 +6,7 @@ from .password_resets import PasswordReset from .hosts import Host from .monitors import Monitor, MonitorResult, MonitorType from .metrics import PluginMetric +from .norms import MetricNorm from .alerts import AlertRule, AlertState, AlertEvent, AlertOperator, AlertStateEnum from .ansible import AnsibleRun, AnsibleRunStatus from .ansible_inventory import AnsibleTarget, AnsibleGroup, ansible_target_groups @@ -18,7 +19,7 @@ __all__ = [ "Base", "User", "ApiToken", "UserInvite", "PasswordReset", "Host", "Monitor", "MonitorResult", "MonitorType", - "PluginMetric", + "PluginMetric", "MetricNorm", "AlertRule", "AlertState", "AlertEvent", "AlertOperator", "AlertStateEnum", "AnsibleRun", "AnsibleRunStatus", "AnsibleTarget", "AnsibleGroup", "ansible_target_groups", diff --git a/steward/models/norms.py b/steward/models/norms.py new file mode 100644 index 0000000..13283df --- /dev/null +++ b/steward/models/norms.py @@ -0,0 +1,27 @@ +from __future__ import annotations +from datetime import datetime +from sqlalchemy import DateTime, Float, Integer, SmallInteger, String +from sqlalchemy.orm import Mapped, mapped_column +from .base import Base + + +class MetricNorm(Base): + """What one metric usually does in one time slot (#5739), learned from + its own history by steward.core.norms.rebuild_norms. + + `kind` says what a slot is: "day" (0–23, the UTC hour of day) or "week" + (0–167, Monday 00:00 UTC first). A key has slots of one kind at a time. + The whole table is rebuilt each pass, so a row is never edited.""" + __tablename__ = "metric_norms" + + source_module: Mapped[str] = mapped_column(String(64), primary_key=True) + resource_name: Mapped[str] = mapped_column(String(255), primary_key=True) + metric_name: Mapped[str] = mapped_column(String(128), primary_key=True) + slot: Mapped[int] = mapped_column(SmallInteger, primary_key=True) + kind: Mapped[str] = mapped_column(String(8), nullable=False) + # The level the slot's hours usually held, and how high they usually + # peaked: the 90th percentile, across past days, of each hour's p95 and max. + usual_hold: Mapped[float] = mapped_column(Float, nullable=False) + usual_peak: Mapped[float] = mapped_column(Float, nullable=False) + days_seen: Mapped[int] = mapped_column(Integer, nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) diff --git a/tests/core/test_norms.py b/tests/core/test_norms.py new file mode 100644 index 0000000..fb7d733 --- /dev/null +++ b/tests/core/test_norms.py @@ -0,0 +1,27 @@ +"""Each metric's normal, by time slot (#5739): the slot arithmetic and the +registry. Learning itself runs in SQL; tests/integration/test_norms.py +covers it against Postgres.""" +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +from steward.core import norms + + +def test_a_day_slot_is_the_utc_hour(): + at = datetime(2026, 10, 14, 2, 30, tzinfo=timezone(timedelta(hours=-5))) # 07:30 UTC + assert norms.slot_of(at, "day") == 7 + + +def test_a_week_slot_counts_hours_from_monday_utc(): + monday = datetime(2026, 10, 12, 0, 10, tzinfo=timezone.utc) + assert norms.slot_of(monday, "week") == 0 + assert norms.slot_of(monday + timedelta(days=2, hours=2), "week") == 50 # Wed 02:10 + assert norms.slot_of(monday + timedelta(days=6, hours=23), "week") == 167 + + +def test_a_learned_metric_is_registered_once_with_its_floor(): + norms.learn_normal("test", "x_pct", 2) + norms.learn_normal("test", "x_pct", 3) + assert norms.LEARNED[("test", "x_pct")] == norms.Learned("test", "x_pct", 3.0) + del norms.LEARNED[("test", "x_pct")] diff --git a/tests/integration/test_norms.py b/tests/integration/test_norms.py new file mode 100644 index 0000000..47774f7 --- /dev/null +++ b/tests/integration/test_norms.py @@ -0,0 +1,106 @@ +"""Integration: Steward learns each metric's normal per hour (#5739). + +The operator's case (System 182): a nightly backup that runs hot at 02:00 +must read as normal at 02:00, and the same level at 14:00 must not, so the +normal has to be per hour and built from what each hour held, not its peak. +Requires STEWARD_DATABASE_URL. +""" +from __future__ import annotations + +import asyncio +import os +import uuid +from datetime import datetime, timedelta, timezone + +import pytest + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def app(): + if not os.environ.get("STEWARD_DATABASE_URL"): + pytest.skip("needs Postgres") + from steward.app import create_app + return create_app(testing=False) + + +@pytest.fixture +def learned(): + from steward.core import norms + norms.learn_normal("normtest", "cpu_pct", 5) + yield + norms.LEARNED.pop(("normtest", "cpu_pct"), None) + + +# Monday 00:00 UTC, so week slots are easy to read. +NOW = datetime(2026, 10, 12, 0, 0, tzinfo=timezone.utc) + + +def _hours(res: str, days: int, level): + """An hourly rollup row for each of the last `days` days' hours, held at + level(day, hour) -> (p95, max).""" + from steward.models.metrics import PluginMetricHourly + rows = [] + for d in range(1, days + 1): + for h in range(24): + bucket = NOW - timedelta(days=d) + timedelta(hours=h) + p95, peak = level(bucket, h) + rows.append(PluginMetricHourly(source_module="normtest", resource_name=res, + metric_name="cpu_pct", bucket=bucket, + value_avg=p95, value_max=peak, value_p95=p95, + sample_count=60)) + return rows + + +def _learn(app, rows): + from steward.core.norms import norms_at, rebuild_norms + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + s.add_all(rows) + async with s.begin(): + await rebuild_norms(s, NOW) + res = rows[0].resource_name + out = {} + for at in (NOW.replace(hour=2), NOW.replace(hour=3), NOW.replace(hour=14), + NOW + timedelta(days=1, hours=2)): + got = await norms_at(s, [("normtest", res, "cpu_pct")], at) + out[at] = got.get(("normtest", res, "cpu_pct")) + return out + + return asyncio.run(_go()) + + +def test_a_nightly_backup_is_normal_at_its_hour_and_not_at_another(app, learned): + res = f"night-{uuid.uuid4().hex[:8]}" + # Every night 02:00–03:00 holds 90%; the rest of the day holds 10% and + # spikes briefly to 95% (a high max, a low p95). + got = _learn(app, _hours(res, 7, lambda b, h: (90.0, 95.0) if h == 2 else (10.0, 95.0))) + at2, at3, at14 = (got[NOW.replace(hour=h)] for h in (2, 3, 14)) + assert at2.kind == "day" and at2.days_seen == 7 + assert at2.usual_hold == 90.0 + # 03:00 takes in 02:00 as a neighbour: a backup that runs late stays normal. + assert at3.usual_hold == 90.0 + # 14:00 holds 10: the spikes raised its peak, never its hold. + assert at14.usual_hold == 10.0 and at14.usual_peak == 95.0 + + +def test_too_little_history_learns_no_normal(app, learned): + res = f"young-{uuid.uuid4().hex[:8]}" + got = _learn(app, _hours(res, 4, lambda b, h: (10.0, 10.0))) + assert all(n is None for n in got.values()) + + +def test_three_weeks_of_history_learns_the_hour_of_the_week(app, learned): + res = f"weekly-{uuid.uuid4().hex[:8]}" + # A weekly job: Tuesdays 02:00 hold 80; every other hour holds 5. + got = _learn(app, _hours(res, 22, lambda b, h: (80.0, 80.0) if b.weekday() == 1 and h == 2 + else (5.0, 5.0))) + tuesday_2 = got[NOW + timedelta(days=1, hours=2)] + monday_2 = got[NOW.replace(hour=2)] + assert tuesday_2.kind == "week" and tuesday_2.slot == 26 and tuesday_2.usual_hold == 80.0 + # Monday 02:00 is far from Tuesday's job: on an hour-of-day normal it + # would have been learned as the job's level. + assert monday_2.usual_hold == 5.0