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 <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
@@ -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")
|
||||
@@ -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",
|
||||
|
||||
@@ -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)
|
||||
@@ -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")]
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user