From 8941ed037201a946fa7ca6a6c8cef99c4bb1458d Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 6 Oct 2026 23:19:38 -0400 Subject: [PATCH] fix(metrics): seed the latest table in the background, not in the migration (#5208) Migration 0030 seeded plugin_metric_latest from the whole history: two index probes per distinct metric key. On the live instance that held the first boot for 7+ minutes with Steward down. The migration now only creates the table. Live sources fill it on their next sample, and backfill_latest runs after startup for resources that have history but no latest rows. Per resource it does one backward probe for the last sample, then takes the newest value per metric within the hour before it. Resources that already have rows are skipped, so later boots only walk the resource names. Co-Authored-By: Claude Opus 5.5 --- steward/app.py | 9 ++++ steward/core/metric_series.py | 51 +++++++++++++++++++ .../versions/0030_plugin_metric_latest.py | 30 ++--------- tests/integration/test_metric_latest.py | 36 +++++++++++++ 4 files changed, 100 insertions(+), 26 deletions(-) diff --git a/steward/app.py b/steward/app.py index d9356df..9ca2217 100644 --- a/steward/app.py +++ b/steward/app.py @@ -362,10 +362,19 @@ def create_app( if not testing: await _mark_interrupted_runs(app) asyncio.create_task(_run_scheduler(app)) + app._latest_backfill = asyncio.create_task(_backfill_latest(app)) return app +async def _backfill_latest(app: Quart) -> None: + from .core.metric_series import backfill_latest + try: + await backfill_latest(app.db_sessionmaker, app.logger) + except Exception: + app.logger.exception("latest metrics: backfill failed; quiet resources fill on their next sample") + + def _register_core_tasks(app: Quart) -> None: from .core.scheduler import ScheduledTask diff --git a/steward/core/metric_series.py b/steward/core/metric_series.py index c177bfc..812234a 100644 --- a/steward/core/metric_series.py +++ b/steward/core/metric_series.py @@ -91,6 +91,57 @@ async def latest_by_resource(db, source_module: str, resources, metrics=None, return out +_BACKFILL_SQL = text(""" + INSERT INTO plugin_metric_latest + (source_module, resource_name, metric_name, value, recorded_at) + SELECT DISTINCT ON (metric_name) source_module, resource_name, metric_name, + value, recorded_at + FROM plugin_metrics + WHERE source_module = :src AND resource_name = :res + AND recorded_at > :since AND recorded_at <= :last + ORDER BY metric_name, recorded_at DESC + ON CONFLICT DO NOTHING +""") + + +async def backfill_latest(sessionmaker, log) -> int: + """Fill plugin_metric_latest for resources that have history but no latest + row yet: the ones that stopped reporting before the table existed (#5208). + A live resource fills itself on its next sample. + + Runs in the background after startup so it never holds the boot. Per + resource it costs one backward probe for the last sample and a range over + the hour before it; a resource that already has latest rows is skipped, so + later boots only step through the resource names. Returns rows written. + """ + import time + started, written, filled = time.monotonic(), 0, 0 + async with sessionmaker() as db: + todo = [] + for src in await sources(db): + have = set((await db.execute( + select(PluginMetricLatest.resource_name) + .where(PluginMetricLatest.source_module == src).distinct() + )).scalars()) + todo += [(src, res) for res in await resources(db, src) if res not in have] + for src, res in todo: + async with sessionmaker() as db, db.begin(): + last = (await db.execute( + select(func.max(PluginMetric.recorded_at)).where( + PluginMetric.source_module == src, PluginMetric.resource_name == res) + )).scalar() + if last is None: + continue + result = await db.execute(_BACKFILL_SQL, { + "src": src, "res": res, "last": last, "since": last - timedelta(hours=1)}) + written += result.rowcount or 0 + filled += 1 + if filled: + log.info("latest metrics: backfilled %d values for %d quiet resources in %.1fs", + written, filled, time.monotonic() - started) + return written + + async def points(db, source_module: str, resource: str, metric: str, since: datetime, *, target: int = 120) -> list[tuple[datetime, float]]: """(time, average) points for one metric since `since`, oldest first. diff --git a/steward/migrations/versions/0030_plugin_metric_latest.py b/steward/migrations/versions/0030_plugin_metric_latest.py index b689b1f..c66985c 100644 --- a/steward/migrations/versions/0030_plugin_metric_latest.py +++ b/steward/migrations/versions/0030_plugin_metric_latest.py @@ -23,32 +23,10 @@ def upgrade() -> None: sa.Column("value", sa.Float, nullable=False), sa.Column("recorded_at", sa.DateTime(timezone=True), nullable=False), ) - # Seed from history without a full scan: step through the distinct keys of - # the (source, resource, metric, recorded_at) index one probe at a time, - # then take each key's newest row with a backward probe of the same index. - op.execute(""" - WITH RECURSIVE k AS ( - (SELECT source_module, resource_name, metric_name FROM plugin_metrics - ORDER BY source_module, resource_name, metric_name LIMIT 1) - UNION ALL - SELECT n.source_module, n.resource_name, n.metric_name - FROM k, LATERAL ( - SELECT p.source_module, p.resource_name, p.metric_name FROM plugin_metrics p - WHERE (p.source_module, p.resource_name, p.metric_name) - > (k.source_module, k.resource_name, k.metric_name) - ORDER BY p.source_module, p.resource_name, p.metric_name LIMIT 1 - ) n - ) - INSERT INTO plugin_metric_latest - (source_module, resource_name, metric_name, value, recorded_at) - SELECT k.source_module, k.resource_name, k.metric_name, l.value, l.recorded_at - FROM k, LATERAL ( - SELECT p.value, p.recorded_at FROM plugin_metrics p - WHERE p.source_module = k.source_module AND p.resource_name = k.resource_name - AND p.metric_name = k.metric_name - ORDER BY p.recorded_at DESC LIMIT 1 - ) l - """) + # Not seeded here: a seed over the whole history held the first boot for + # minutes. Live sources fill it on their next sample, and + # metric_series.backfill_latest fills the quiet ones in the background + # after startup. def downgrade() -> None: diff --git a/tests/integration/test_metric_latest.py b/tests/integration/test_metric_latest.py index add7a59..9bc60ce 100644 --- a/tests/integration/test_metric_latest.py +++ b/tests/integration/test_metric_latest.py @@ -148,3 +148,39 @@ def test_a_renamed_host_carries_its_latest_values(app): assert old_latest == {} assert new_latest["cpu_pct"][0] == 5.0 assert new_root["disk_used_pct"][0] == 60.0 + + +def test_the_backfill_fills_a_quiet_resource_once_and_leaves_live_ones_alone(app): + import logging + + from steward.core.alerts import record_metrics + from steward.core.metric_series import backfill_latest + + now = datetime.now(timezone.utc) + quiet, live = ("bf-quiet-" + uuid.uuid4().hex[:8]), ("bf-live-" + uuid.uuid4().hex[:8]) + went_quiet = now - timedelta(hours=3) + + async def go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + # History from before the table existed: rows with no latest. + s.add_all([ + _sample(quiet, "cpu_pct", 10.0, went_quiet - timedelta(minutes=30)), + _sample(quiet, "cpu_pct", 11.0, went_quiet), + _sample(quiet, "mem_used_pct", 50.0, went_quiet - timedelta(minutes=10)), + _sample(quiet, "load_1m", 9.9, went_quiet - timedelta(hours=2)), # outside the hour + ]) + async with s.begin(): + await record_metrics(s, [_sample(live, "cpu_pct", 70.0, now)]) + s.add(_sample(live, "cpu_pct", 1.0, now - timedelta(minutes=5))) + first = await backfill_latest(app.db_sessionmaker, logging.getLogger("test")) + second = await backfill_latest(app.db_sessionmaker, logging.getLogger("test")) + async with app.db_sessionmaker() as s: + return first, second, await _latest(s, quiet), await _latest(s, live) + + first, second, quiet_latest, live_latest = asyncio.run(go()) + assert first >= 2 and second == 0 + assert quiet_latest["cpu_pct"][0] == 11.0 + assert quiet_latest["mem_used_pct"][0] == 50.0 + assert "load_1m" not in quiet_latest + assert live_latest == {"cpu_pct": (70.0, live_latest["cpu_pct"][1])}