perf(metrics): a day of one metric reads from the index alone, built after boot (#5490)
#5490 "A day of one metric reads fast even cold". A cold 24 h series read measured 18.3 s against 63 ms warm (#5485): every sample's row sat on its own heap page. ix_plugin_metrics_series_value carries `value` (INCLUDE), so the read is an index-only scan; it replaces the plain series index from 0023. It is built once, with CREATE INDEX CONCURRENTLY from a task started after the app is serving, never in a migration, so the boot isn't held while it builds over the live table (#5220) and ingest and pages carry on. A build cut short by a restart leaves an INVALID index; the next boot drops and rebuilds it. Later boots find it present and only check. Postgres keeps it current on each insert, as it did the index it replaces. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -194,5 +194,5 @@ Defined in `steward/models/alerts.py`:
|
||||
- **`alert_rules`** — one row per configured rule
|
||||
- **`alert_states`** — one row per rule, tracks current state and consecutive failure count
|
||||
- **`alert_events`** — append-only log of all state transitions and notification outcomes
|
||||
- **`plugin_metrics`** — all metric values written by any monitor or plugin
|
||||
- **`plugin_metrics`** — all metric values written by any monitor or plugin. Its series index carries the value (`ix_plugin_metrics_series_value`), so a day of one metric is read from the index alone even when nothing is cached; Steward builds it in the background after startup (`core/metric_index.py`), never during a migration
|
||||
- **`plugin_metric_latest`** — the newest sample of each (source, resource, metric). Pages that show "what it is now" read this instead of searching history; retention drops a row once its sample leaves the raw window
|
||||
|
||||
@@ -381,10 +381,20 @@ def create_app(
|
||||
await _mark_interrupted_runs(app)
|
||||
asyncio.create_task(_run_scheduler(app))
|
||||
app._latest_backfill = asyncio.create_task(_backfill_latest(app))
|
||||
app._series_index = asyncio.create_task(_ensure_series_index(app))
|
||||
|
||||
return app
|
||||
|
||||
|
||||
async def _ensure_series_index(app: Quart) -> None:
|
||||
from .core.metric_index import ensure_series_index
|
||||
try:
|
||||
await ensure_series_index(app._db_engine, app.logger)
|
||||
except Exception:
|
||||
# Reads still work on the plain index; the next boot tries again.
|
||||
app.logger.exception("metric index: build failed; series reads use the plain index")
|
||||
|
||||
|
||||
async def _backfill_latest(app: Quart) -> None:
|
||||
from .core.metric_series import backfill_latest
|
||||
try:
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
"""The series index on plugin_metrics, built in the background after boot (#5490).
|
||||
|
||||
A read of one metric over a day (a rule's preview, query_metrics, a chart)
|
||||
walks the (source, resource, metric, recorded_at) index; without the value in
|
||||
the index, each sample is then fetched from its own heap page, interleaved
|
||||
with every other series, so a cold day measured 18.3 s against 63 ms warm
|
||||
(#5485). With `value` INCLUDEd the read is an index-only scan over adjacent
|
||||
index pages.
|
||||
|
||||
The index is built with CREATE INDEX CONCURRENTLY from a task started after
|
||||
the app is serving, never in a migration: a build over the live table takes
|
||||
minutes on slow storage, and a migration would hold the boot with Steward
|
||||
down for all of it (#5220). Ingest and pages keep working while it builds.
|
||||
Once it is valid the index it replaces is dropped. Later boots find it
|
||||
present and only check.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
from sqlalchemy import text
|
||||
|
||||
SERIES_INDEX = "ix_plugin_metrics_series_value"
|
||||
# The plain series index the covering one replaces (migration 0023).
|
||||
REPLACES = "ix_plugin_metrics_module_resource_metric_recorded"
|
||||
|
||||
_STATE = text("""
|
||||
SELECT i.indisvalid FROM pg_class c JOIN pg_index i ON i.indexrelid = c.oid
|
||||
WHERE c.relname = :name
|
||||
""")
|
||||
|
||||
|
||||
async def ensure_series_index(engine, log) -> str:
|
||||
"""Make sure the covering series index exists and is valid, then drop the
|
||||
one it replaces. Returns "present" when it already was, else "built".
|
||||
|
||||
A build a restart interrupted leaves an INVALID index behind; that one is
|
||||
dropped and built again. Runs outside a transaction, as CONCURRENTLY
|
||||
requires."""
|
||||
async with engine.connect() as conn:
|
||||
conn = await conn.execution_options(isolation_level="AUTOCOMMIT")
|
||||
valid = (await conn.execute(_STATE, {"name": SERIES_INDEX})).scalar()
|
||||
outcome = "present"
|
||||
if valid is not True:
|
||||
if valid is False:
|
||||
log.warning("metric index: %s was left half-built; dropping it to build again",
|
||||
SERIES_INDEX)
|
||||
await conn.execute(text(f"DROP INDEX CONCURRENTLY IF EXISTS {SERIES_INDEX}"))
|
||||
log.info("metric index: building %s in the background; pages and ingest carry on",
|
||||
SERIES_INDEX)
|
||||
started = time.monotonic()
|
||||
await conn.execute(text(
|
||||
f"CREATE INDEX CONCURRENTLY IF NOT EXISTS {SERIES_INDEX} ON plugin_metrics "
|
||||
"(source_module, resource_name, metric_name, recorded_at) INCLUDE (value)"))
|
||||
log.info("metric index: built %s in %.0fs", SERIES_INDEX, time.monotonic() - started)
|
||||
outcome = "built"
|
||||
if (await conn.execute(_STATE, {"name": REPLACES})).scalar() is not None:
|
||||
await conn.execute(text(f"DROP INDEX CONCURRENTLY IF EXISTS {REPLACES}"))
|
||||
log.info("metric index: dropped %s, which %s replaces", REPLACES, SERIES_INDEX)
|
||||
return outcome
|
||||
@@ -25,8 +25,12 @@ class PluginMetric(Base):
|
||||
__table_args__ = (
|
||||
Index("ix_plugin_metrics_module_resource_recorded",
|
||||
"source_module", "resource_name", "recorded_at"),
|
||||
Index("ix_plugin_metrics_module_resource_metric_recorded",
|
||||
"source_module", "resource_name", "metric_name", "recorded_at"),
|
||||
# One series over a time range, read from the index alone: `value` is
|
||||
# carried in it. Built after boot by core/metric_index.py, not by a
|
||||
# migration, since a build over the live table takes minutes (#5490).
|
||||
Index("ix_plugin_metrics_series_value",
|
||||
"source_module", "resource_name", "metric_name", "recorded_at",
|
||||
postgresql_include=["value"]),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
"""Integration: the covering series index on plugin_metrics (#5490), built
|
||||
after boot against Postgres. It replaces the plain series index, a later boot
|
||||
only checks, and a day of one series is then read from the index alone."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
|
||||
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_the_series_index_is_built_once_and_replaces_the_plain_one(app):
|
||||
from sqlalchemy import text
|
||||
from steward.core.metric_index import REPLACES, SERIES_INDEX, ensure_series_index
|
||||
|
||||
log = logging.getLogger("test")
|
||||
|
||||
async def _indexes():
|
||||
async with app.db_sessionmaker() as s:
|
||||
return dict((await s.execute(text(
|
||||
"SELECT indexname, indexdef FROM pg_indexes WHERE tablename = 'plugin_metrics'"))).all())
|
||||
|
||||
async def _go():
|
||||
first = await ensure_series_index(app._db_engine, log)
|
||||
got = await _indexes()
|
||||
assert SERIES_INDEX in got and REPLACES not in got
|
||||
assert "INCLUDE (value)" in got[SERIES_INDEX]
|
||||
assert await ensure_series_index(app._db_engine, log) == "present"
|
||||
assert first in ("built", "present")
|
||||
|
||||
# One series over a day is answered from the index alone.
|
||||
async with app.db_sessionmaker() as s:
|
||||
await s.execute(text("SET enable_seqscan = off"))
|
||||
await s.execute(text("SET enable_bitmapscan = off"))
|
||||
plan = "\n".join(r[0] for r in (await s.execute(text(
|
||||
"EXPLAIN SELECT date_trunc('hour', recorded_at), avg(value) FROM plugin_metrics "
|
||||
"WHERE source_module = 'host_agent' AND resource_name = 'x' "
|
||||
"AND metric_name = 'cpu_pct' AND recorded_at >= now() - interval '1 day' "
|
||||
"GROUP BY 1"))).all())
|
||||
assert f"Index Only Scan using {SERIES_INDEX}" in plan, plan
|
||||
|
||||
asyncio.run(_go())
|
||||
Reference in New Issue
Block a user