From cbbf186f171abb435ba447cdfa2a3fccfed0e004 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 14 Oct 2026 12:45:42 -0400 Subject: [PATCH] feat(metrics): each rolled-up hour keeps the level it held, its p95 (#5738) plugin_metrics_hourly gains value_p95 beside value_avg and value_max. A two-minute backup spike raises the hour's peak and leaves its p95 low; an hour spent at that level raises both. Learned normals (milestone 513) need that difference to call a recurring spike usual without calling the same level held at another hour usual too. Hours rolled up before this keep a NULL p95 and read as their average. Co-Authored-By: Claude Opus 5.5 --- docs/architecture.md | 16 +++++ steward/core/metrics_retention.py | 9 ++- .../migrations/versions/0044_hourly_p95.py | 26 ++++++++ steward/models/metrics.py | 4 ++ tests/integration/test_hourly_p95.py | 62 +++++++++++++++++++ 5 files changed, 114 insertions(+), 3 deletions(-) create mode 100644 steward/migrations/versions/0044_hourly_p95.py create mode 100644 tests/integration/test_hourly_p95.py diff --git a/docs/architecture.md b/docs/architecture.md index 4cb75f5..7779cdf 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -87,6 +87,22 @@ Core tasks registered in `app._task_registry`: --- +## Metric Store + +Every plugin and monitor records into one store keyed by (source, resource, metric): +- `plugin_metrics` holds raw samples for `metrics.retention.raw_days` (7 by default). +- `plugin_metric_latest` holds each key's newest value. +- `data_cleanup` rolls raw samples older than the raw window into `plugin_metrics_hourly`, kept for `metrics.retention.rollup_days` (90 by default). + +Each rolled-up hour keeps three figures: +- `value_avg`: the hour's average. +- `value_max`: its peak. +- `value_p95`: the level it held for all but its briefest 5%, about 3 minutes. + +A two-minute spike raises the peak and leaves the p95 low; an hour spent at that level raises both. Learned normals use this to tell a recurring spike from a held level. Hours rolled up before the p95 was kept have it NULL; read those as their average. + +--- + ## Database Session Pattern `app.db_sessionmaker` is an SQLAlchemy `async_sessionmaker`. All DB access follows this pattern: diff --git a/steward/core/metrics_retention.py b/steward/core/metrics_retention.py index 8be4085..6247ede 100644 --- a/steward/core/metrics_retention.py +++ b/steward/core/metrics_retention.py @@ -40,7 +40,7 @@ async def rollup_plugin_metrics( """Roll up + prune plugin_metrics. Returns a counts dict for logging. 1. Aggregate plugin_metrics older than the (hour-aligned) raw window into - plugin_metrics_hourly (avg/max per source/resource/metric/hour), upserting + plugin_metrics_hourly (avg/max/p95 per source/resource/metric/hour), upserting so a re-run is idempotent, then delete those raw rows. 2. Prune rolled-up rows older than the rollup window. """ @@ -67,6 +67,7 @@ async def rollup_plugin_metrics( hour.label("bucket"), func.avg(PluginMetric.value).label("value_avg"), func.max(PluginMetric.value).label("value_max"), + func.percentile_cont(0.95).within_group(PluginMetric.value).label("value_p95"), func.count().label("sample_count"), ) .where(PluginMetric.recorded_at < raw_cutoff) @@ -78,17 +79,19 @@ async def rollup_plugin_metrics( for r in (await session.execute(agg)).all(): avg_v = float(r.value_avg or 0.0) max_v = float(r.value_max or 0.0) + p95_v = float(r.value_p95) if r.value_p95 is not None else avg_v cnt = int(r.sample_count or 0) await session.execute( pg_insert(PluginMetricHourly) .values( source_module=r.source_module, resource_name=r.resource_name, metric_name=r.metric_name, bucket=r.bucket, - value_avg=avg_v, value_max=max_v, sample_count=cnt, + value_avg=avg_v, value_max=max_v, value_p95=p95_v, sample_count=cnt, ) .on_conflict_do_update( constraint="uq_plugin_metrics_hourly_bucket", - set_={"value_avg": avg_v, "value_max": max_v, "sample_count": cnt}, + set_={"value_avg": avg_v, "value_max": max_v, "value_p95": p95_v, + "sample_count": cnt}, ) ) rolled += 1 diff --git a/steward/migrations/versions/0044_hourly_p95.py b/steward/migrations/versions/0044_hourly_p95.py new file mode 100644 index 0000000..5456e1c --- /dev/null +++ b/steward/migrations/versions/0044_hourly_p95.py @@ -0,0 +1,26 @@ +"""Each hour of the metric rollup keeps the level it held, its 95th percentile, +beside its average and peak (#5738), so a learned normal can tell a brief +spike (a high peak, a low p95) from a level the metric stayed at. + +Revision ID: 0044_hourly_p95 +Revises: 0043_outbox_alert_words +Create Date: 2026-10-14 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "0044_hourly_p95" +down_revision: Union[str, None] = "0043_outbox_alert_words" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # Hours rolled up before this have no p95 to recover (their raw samples + # are gone), so they stay NULL and read as their average. + op.add_column("plugin_metrics_hourly", sa.Column("value_p95", sa.Float, nullable=True)) + + +def downgrade() -> None: + op.drop_column("plugin_metrics_hourly", "value_p95") diff --git a/steward/models/metrics.py b/steward/models/metrics.py index a84a751..b61d988 100644 --- a/steward/models/metrics.py +++ b/steward/models/metrics.py @@ -70,6 +70,10 @@ class PluginMetricHourly(Base): bucket: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) value_avg: Mapped[float] = mapped_column(Float, nullable=False, default=0.0) value_max: Mapped[float] = mapped_column(Float, nullable=False, default=0.0) + # The level the hour held for all but its briefest 5% (#5738): a spike of + # a minute or two raises value_max and leaves this low. NULL on hours + # rolled up before it was kept; read those as value_avg. + value_p95: Mapped[float | None] = mapped_column(Float, nullable=True) sample_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) __table_args__ = ( diff --git a/tests/integration/test_hourly_p95.py b/tests/integration/test_hourly_p95.py new file mode 100644 index 0000000..207cce6 --- /dev/null +++ b/tests/integration/test_hourly_p95.py @@ -0,0 +1,62 @@ +"""Integration: each rolled-up hour keeps the level it held (#5738). + +A learned normal has to tell a two-minute backup spike from an hour spent at +that level. Both have the same peak; only the hour's 95th percentile differs, +so the rollup must keep it beside the average and the 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) + + +def test_a_brief_spike_raises_the_peak_and_a_held_level_raises_the_p95(app): + from sqlalchemy import select + from steward.core.metrics_retention import rollup_plugin_metrics + from steward.models.metrics import PluginMetric, PluginMetricHourly + + now = datetime.now(timezone.utc) + res = f"p95-{uuid.uuid4().hex[:8]}" + spike_hour = (now - timedelta(days=10)).replace(minute=0, second=0, microsecond=0) + held_hour = spike_hour + timedelta(hours=1) + + def minute(hour, m, metric, value): + return PluginMetric(source_module="test", resource_name=res, metric_name=metric, + value=value, recorded_at=hour + timedelta(minutes=m)) + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + # One sample a minute. The spike hour sits at 10 and touches 95 + # for two minutes; the held hour sits at 90 bar a two-minute dip. + s.add_all([minute(spike_hour, m, "cpu_pct", 95.0 if m in (2, 3) else 10.0) + for m in range(60)]) + s.add_all([minute(held_hour, m, "cpu_pct", 10.0 if m in (2, 3) else 90.0) + for m in range(60)]) + async with s.begin(): + await rollup_plugin_metrics(s, raw_days=7, rollup_days=90, now=now) + rows = (await s.execute( + select(PluginMetricHourly) + .where(PluginMetricHourly.resource_name == res) + .order_by(PluginMetricHourly.bucket))).scalars().all() + return [(r.bucket, r.value_max, r.value_p95, r.value_avg) for r in rows] + + (b1, spike_max, spike_p95, _), (b2, held_max, held_p95, _) = asyncio.run(_go()) + assert (b1, b2) == (spike_hour, held_hour) + assert spike_max == 95.0 and abs(spike_p95 - 10.0) < 0.001 + assert held_max == 90.0 and abs(held_p95 - 90.0) < 0.001