From 31e93e96cf26f006b806997ac7d9023879a07aa2 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 8 Oct 2026 08:20:09 -0400 Subject: [PATCH] fix(traefik): the rows page reads every router's history in one index-only query (#5355) #5355 "Traefik page doesn't load": /plugins/traefik/rows took 12-28 s. It ran a query per router ever seen, and each raw row in the range was a heap fetch. traefik_004 rebuilds the router index to cover the charted columns and indexes scraped_at for the request chart; router_history reads the current routers' buckets in one query. Co-Authored-By: Claude Opus 5.5 --- .../versions/traefik_004_covering_indexes.py | 34 ++++++++ plugins/traefik/routes.py | 79 ++++++++++--------- tests/integration/test_traefik_latest.py | 34 ++++++++ 3 files changed, 109 insertions(+), 38 deletions(-) create mode 100644 plugins/traefik/migrations/versions/traefik_004_covering_indexes.py diff --git a/plugins/traefik/migrations/versions/traefik_004_covering_indexes.py b/plugins/traefik/migrations/versions/traefik_004_covering_indexes.py new file mode 100644 index 0000000..23de1cf --- /dev/null +++ b/plugins/traefik/migrations/versions/traefik_004_covering_indexes.py @@ -0,0 +1,34 @@ +"""Traefik reads answered from the index alone + +The rows page and the request chart read every raw row in the chosen range. +With only (router_name, scraped_at) indexed, each of those rows was a heap +fetch: a random disk read on a database whose indexes no longer fit in +memory, which took the rows page to 12-28 s. Carrying the charted columns +in the index makes these index-only scans; scraped_at gets an index of its +own for the request chart, which filters on time across every router. + +Revision ID: traefik_004_covering_indexes +Revises: traefik_003_access_log +Create Date: 2026-10-08 +""" +from typing import Sequence, Union +from alembic import op + +revision: str = "traefik_004_covering_indexes" +down_revision: Union[str, None] = "traefik_003_access_log" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.drop_index("ix_traefik_metrics_router_scraped", "traefik_metrics") + op.create_index("ix_traefik_metrics_router_scraped", "traefik_metrics", ["router_name", "scraped_at"], + postgresql_include=["request_rate", "latency_p95_ms", "error_rate_5xx_pct"]) + op.create_index("ix_traefik_metrics_scraped", "traefik_metrics", ["scraped_at"], + postgresql_include=["request_rate", "error_rate_5xx_pct"]) + + +def downgrade() -> None: + op.drop_index("ix_traefik_metrics_scraped", "traefik_metrics") + op.drop_index("ix_traefik_metrics_router_scraped", "traefik_metrics") + op.create_index("ix_traefik_metrics_router_scraped", "traefik_metrics", ["router_name", "scraped_at"]) diff --git a/plugins/traefik/routes.py b/plugins/traefik/routes.py index 120dbda..a3b836d 100644 --- a/plugins/traefik/routes.py +++ b/plugins/traefik/routes.py @@ -180,44 +180,21 @@ async def rows(): now = datetime.now(timezone.utc) thresholds = current_app.config.get("THRESHOLDS", {}) + scrape = current_app.config["PLUGINS"]["traefik"]["scrape_interval_seconds"] + async with current_app.db_sessionmaker() as db: - # Every known router, with its most recent raw row - newest = {m.router_name: m for m in await latest_per_router(db)} - routers = sorted(newest) + # Every known router, with its most recent raw row; a router Traefik + # stopped reporting gets no sparklines, so its history is not read. + newest = await latest_per_router(db) + current, gone = gone_routers([{"name": m.router_name, "latest": m} for m in newest], scrape) - b_secs = bucket_seconds(since) - bucket_col = _time_bucket(TraefikMetric.scraped_at, b_secs) - - router_data = [] - for router in routers: - # Bucketed history for sparklines — always ~80 points regardless of range - result = await db.execute( - select( - func.avg(TraefikMetric.request_rate).label("request_rate"), - func.avg(TraefikMetric.latency_p95_ms).label("latency_p95_ms"), - func.avg(TraefikMetric.error_rate_5xx_pct).label("error_rate_5xx_pct"), - func.min(TraefikMetric.scraped_at).label("scraped_at"), - bucket_col, - ) - .where(TraefikMetric.router_name == router) - .where(TraefikMetric.scraped_at >= since) - .group_by(bucket_col) - .order_by(bucket_col) - ) - history = result.all() - - latest = newest[router] - - state, why = router_state(latest, thresholds) - router_data.append({ - "name": router, - "latest": latest, - "state": state, - "why": why, - "sparkline_req": _sparkline([r.request_rate for r in history]), - "sparkline_p95": _sparkline([r.latency_p95_ms for r in history]), - "sparkline_5xx": _sparkline([r.error_rate_5xx_pct for r in history]), - }) + history = await router_history(db, [r["name"] for r in current], since, bucket_seconds(since)) + for r in current: + points = history[r["name"]] + r["state"], r["why"] = router_state(r["latest"], thresholds) + r["sparkline_req"] = _sparkline([p.request_rate for p in points]) + r["sparkline_p95"] = _sparkline([p.latency_p95_ms for p in points]) + r["sparkline_5xx"] = _sparkline([p.error_rate_5xx_pct for p in points]) # Latest global stat (always most recent, not range-filtered) result = await db.execute( @@ -233,8 +210,6 @@ async def rows(): ) raw_certs = result.scalars().all() - scrape = current_app.config["PLUGINS"]["traefik"]["scrape_interval_seconds"] - current, gone = gone_routers(router_data, scrape) served, old = current_certs(raw_certs, scrape) domains = router_domains() @@ -382,6 +357,34 @@ def _time_bucket(column, seconds: int): func.to_timestamp(0)).label("bucket") +async def router_history(db, names: list[str], since, bucket_secs: int) -> dict[str, list]: + """{router: [rows of request_rate, latency_p95_ms, error_rate_5xx_pct]} + per `bucket_secs` window since `since`, oldest first: one query for every + router, answered from the covering (router_name, scraped_at) index + (traefik_004). It was a query per router, each row a heap read, which + took the rows page to 12-28 s.""" + history: dict[str, list] = {n: [] for n in names} + if not history: + return history + bucket_col = _time_bucket(TraefikMetric.scraped_at, bucket_secs) + result = await db.execute( + select( + TraefikMetric.router_name, + func.avg(TraefikMetric.request_rate).label("request_rate"), + func.avg(TraefikMetric.latency_p95_ms).label("latency_p95_ms"), + func.avg(TraefikMetric.error_rate_5xx_pct).label("error_rate_5xx_pct"), + bucket_col, + ) + .where(TraefikMetric.router_name.in_(names)) + .where(TraefikMetric.scraped_at >= since) + .group_by(TraefikMetric.router_name, bucket_col) + .order_by(TraefikMetric.router_name, bucket_col) + ) + for row in result.all(): + history[row.router_name].append(row) + return history + + async def request_history(db, since, bucket_seconds: int) -> list[dict]: """[{"at", "req_rate", "err_pct"}] per time bucket since `since`: each scrape summed over routers (the total req/s), averaged over the bucket; diff --git a/tests/integration/test_traefik_latest.py b/tests/integration/test_traefik_latest.py index 9ed2a0a..fa4508e 100644 --- a/tests/integration/test_traefik_latest.py +++ b/tests/integration/test_traefik_latest.py @@ -74,3 +74,37 @@ def test_the_request_chart_sums_routers_and_weights_the_5xx_share(app): got = asyncio.run(go()) # One bucket: totals 4 and 6 average to 5; 150 weighted 5xx over 10 req/s is 15%. assert got == [{"at": at, "req_rate": 5.0, "err_pct": 15.0}] + + +def test_router_history_reads_every_router_in_one_query_bucketed(app): + """#5214: the rows page read each router's history in its own query, a + heap fetch per row, and took 12-28 s. One query now buckets every named + router; a router not asked for is not read, and one with no rows in the + range still has an (empty) entry.""" + from plugins.traefik.models import TraefikMetric + from plugins.traefik.routes import router_history + + at = datetime(2098, 6, 1, 12, 0, tzinfo=timezone.utc) + timedelta(minutes=5 * (uuid.uuid4().int % 2_000)) + tag = uuid.uuid4().hex[:8] + a, b, c, idle = (f"{x}-{tag}@docker" for x in ("a", "b", "c", "idle")) + + def row(name, minute, rate, p95=2.0): + return TraefikMetric(router_name=name, scraped_at=at + timedelta(minutes=minute), request_rate=rate, + error_rate_4xx_pct=0.0, error_rate_5xx_pct=0.0, latency_p50_ms=1.0, + latency_p95_ms=p95, latency_p99_ms=3.0, response_bytes_rate=0.0) + + async def go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + s.add_all([row(a, 0, 1.0, 10.0), row(a, 1, 3.0, 30.0), row(a, 5, 7.0), + row(b, 0, 4.0), row(c, 0, 9.0), row(idle, -60, 5.0)]) + async with s.begin(): + return await router_history(s, [a, b, idle], at, 300) + + got = asyncio.run(go()) + assert set(got) == {a, b, idle} # c was not asked for + # a: minutes 0 and 1 share the first 5-minute bucket and average; minute 5 starts the next. + assert [r.request_rate for r in got[a]] == [2.0, 7.0] + assert got[a][0].latency_p95_ms == 20.0 + assert [r.request_rate for r in got[b]] == [4.0] + assert got[idle] == [] # its only row is before the range