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 <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
## Database Session Pattern
|
||||||
|
|
||||||
`app.db_sessionmaker` is an SQLAlchemy `async_sessionmaker`. All DB access follows this pattern:
|
`app.db_sessionmaker` is an SQLAlchemy `async_sessionmaker`. All DB access follows this pattern:
|
||||||
|
|||||||
@@ -40,7 +40,7 @@ async def rollup_plugin_metrics(
|
|||||||
"""Roll up + prune plugin_metrics. Returns a counts dict for logging.
|
"""Roll up + prune plugin_metrics. Returns a counts dict for logging.
|
||||||
|
|
||||||
1. Aggregate plugin_metrics older than the (hour-aligned) raw window into
|
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.
|
so a re-run is idempotent, then delete those raw rows.
|
||||||
2. Prune rolled-up rows older than the rollup window.
|
2. Prune rolled-up rows older than the rollup window.
|
||||||
"""
|
"""
|
||||||
@@ -67,6 +67,7 @@ async def rollup_plugin_metrics(
|
|||||||
hour.label("bucket"),
|
hour.label("bucket"),
|
||||||
func.avg(PluginMetric.value).label("value_avg"),
|
func.avg(PluginMetric.value).label("value_avg"),
|
||||||
func.max(PluginMetric.value).label("value_max"),
|
func.max(PluginMetric.value).label("value_max"),
|
||||||
|
func.percentile_cont(0.95).within_group(PluginMetric.value).label("value_p95"),
|
||||||
func.count().label("sample_count"),
|
func.count().label("sample_count"),
|
||||||
)
|
)
|
||||||
.where(PluginMetric.recorded_at < raw_cutoff)
|
.where(PluginMetric.recorded_at < raw_cutoff)
|
||||||
@@ -78,17 +79,19 @@ async def rollup_plugin_metrics(
|
|||||||
for r in (await session.execute(agg)).all():
|
for r in (await session.execute(agg)).all():
|
||||||
avg_v = float(r.value_avg or 0.0)
|
avg_v = float(r.value_avg or 0.0)
|
||||||
max_v = float(r.value_max 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)
|
cnt = int(r.sample_count or 0)
|
||||||
await session.execute(
|
await session.execute(
|
||||||
pg_insert(PluginMetricHourly)
|
pg_insert(PluginMetricHourly)
|
||||||
.values(
|
.values(
|
||||||
source_module=r.source_module, resource_name=r.resource_name,
|
source_module=r.source_module, resource_name=r.resource_name,
|
||||||
metric_name=r.metric_name, bucket=r.bucket,
|
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(
|
.on_conflict_do_update(
|
||||||
constraint="uq_plugin_metrics_hourly_bucket",
|
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
|
rolled += 1
|
||||||
|
|||||||
@@ -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")
|
||||||
@@ -70,6 +70,10 @@ class PluginMetricHourly(Base):
|
|||||||
bucket: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
|
bucket: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
|
||||||
value_avg: Mapped[float] = mapped_column(Float, nullable=False, default=0.0)
|
value_avg: Mapped[float] = mapped_column(Float, nullable=False, default=0.0)
|
||||||
value_max: 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)
|
sample_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
||||||
|
|
||||||
__table_args__ = (
|
__table_args__ = (
|
||||||
|
|||||||
@@ -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
|
||||||
Reference in New Issue
Block a user