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])}