fix(traefik): the newest scrape per router comes from plugin_metric_latest (#5355)

The phase log on the live instance: 'traefik rows: slow 7.4s (latest=7.04s
history=0.11s window=0.04s ...)'. latest_per_router walked traefik_metrics'
index one router at a time; each step lands on the router's oldest
entries, which retention deletes every minute (likely dead ones). The
scraper already records every router figure through MetricBatch, which
keeps one row per router and figure in plugin_metric_latest (#5208's
table, read as latest_by_resource does); latest_per_router reads that
now, and the router figures carry the scrape's own time.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

plugins/traefik/routes.py

@@ -4,7 +4,7 @@ from collections import defaultdict
 from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
 from quart import Blueprint, current_app, render_template, request
-from sqlalchemy import func, select, text
+from sqlalchemy import func, select
 import json
 import time
 
@@ -12,6 +12,7 @@ from steward.auth.middleware import require_role
 from steward.models.users import UserRole
 from steward.core.time_range import parse_range, DEFAULT_RANGE, bucket_seconds, subsample
 from steward.core.formatting import cert_left, status_class
+from steward.models.metrics import PluginMetricLatest
 from .models import TraefikAccessSummary, TraefikCert, TraefikGlobalStat, TraefikMetric
 from .scheduler import router_domains
 from .scraper import cert_names, covers
@@ -140,27 +141,30 @@ def split_routers(routers: list[dict]) -> tuple[list[dict], list[dict]]:
     return problems, healthy
 
 
-# Newest row per router. Postgres has no skip scan, so DISTINCT or max() GROUP
-# BY reads every row in retention; this walks the (router_name, scraped_at)
-# index one router at a time instead: a step per router, not per row.
-_LATEST_PER_ROUTER = text("""
-    WITH RECURSIVE names AS (
-        SELECT min(router_name) AS name FROM traefik_metrics
-        UNION ALL
-        SELECT (SELECT min(router_name) FROM traefik_metrics WHERE router_name > names.name)
-        FROM names WHERE names.name IS NOT NULL
-    )
-    SELECT m.* FROM names
-    JOIN LATERAL (
-        SELECT * FROM traefik_metrics WHERE router_name = names.name
-        ORDER BY scraped_at DESC LIMIT 1
-    ) m ON true
-""")
+# The figures the scraper records for every router (scheduler._do_scrape).
+ROUTER_METRICS = ("request_rate", "error_rate_4xx_pct", "error_rate_5xx_pct", "latency_p50_ms",
+                  "latency_p95_ms", "latency_p99_ms", "response_bytes_rate")
 
 
-async def latest_per_router(db) -> list:
-    return list((await db.execute(
-        select(TraefikMetric).from_statement(_LATEST_PER_ROUTER))).scalars().all())
+async def latest_per_router(db) -> list[SimpleNamespace]:
+    """Each router's newest scrape, read from plugin_metric_latest: the scraper
+    records every figure through MetricBatch, which keeps one row per router
+    and figure there (latest_by_resource's table, #5208). A walk of
+    traefik_metrics' index, one step per router, took 7 s on the live
+    instance (#5355); each step lands on the router's oldest entries, which
+    retention deletes every minute, so it likely waded through dead ones.
+
+    Each router reads as one scrape, with the attributes a TraefikMetric row
+    has; certificates, recorded under the same source, are left out."""
+    rows = (await db.execute(select(PluginMetricLatest).where(
+        PluginMetricLatest.source_module == "traefik",
+        PluginMetricLatest.metric_name.in_(ROUTER_METRICS)))).scalars()
+    by_router: dict[str, dict] = {}
+    for r in rows:
+        by_router.setdefault(r.resource_name, {})[r.metric_name] = r
+    return [SimpleNamespace(router_name=name, scraped_at=max(m.recorded_at for m in figures.values()),
+                            **{k: figures[k].value if k in figures else 0.0 for k in ROUTER_METRICS})
+            for name, figures in by_router.items() if "request_rate" in figures]
 
 
 def current_certs(certs, scrape_seconds: int) -> tuple[list, list]:

plugins/traefik/scheduler.py

@@ -115,11 +115,14 @@ async def _do_scrape(app: "Quart") -> None:
                     ("latency_p99_ms", metrics["latency_p99_ms"]),
                     ("response_bytes_rate", metrics["response_bytes_rate"]),
                 ]:
+                    # Stamped with the scrape, so a router's latest figures
+                    # (routes.latest_per_router) all carry the same moment.
                     batch.add(
                         source_module="traefik",
                         resource_name=router_name,
                         metric_name=metric_name,
                         value=value,
+                        recorded_at=scraped_at,
                     )
 
             # ── Global stats ──────────────────────────────────────────────────

tests/integration/test_traefik_latest.py

@@ -21,29 +21,42 @@ def app():
     return create_app(testing=False)
 
 
-def test_each_router_reads_as_its_newest_row(app):
-    from plugins.traefik.models import TraefikMetric
-    from plugins.traefik.routes import latest_per_router
+def test_each_router_reads_as_its_newest_scrape(app):
+    """#5355: the newest scrape per router comes from plugin_metric_latest,
+    which the scraper's MetricBatch keeps, not from a walk of traefik_metrics
+    (7 s on the live instance). A backlog arriving late doesn't move it back,
+    and a certificate's figure under the same source is not a router."""
+    from steward.core.alerts import MetricBatch
+    from plugins.traefik.routes import ROUTER_METRICS, latest_per_router
 
     now = datetime.now(timezone.utc).replace(microsecond=0)
     tag = uuid.uuid4().hex[:8]
-    a, b = f"a-{tag}@docker", f"b-{tag}@docker"
+    a, b, cert = f"a-{tag}@docker", f"b-{tag}@docker", f"cert-{tag}.example"
 
-    def row(name, ago, rate):
-        return TraefikMetric(router_name=name, scraped_at=now - timedelta(minutes=ago), request_rate=rate,
-                             error_rate_4xx_pct=0.0, error_rate_5xx_pct=0.0, latency_p50_ms=1.0,
-                             latency_p95_ms=2.0, latency_p99_ms=3.0, response_bytes_rate=0.0)
+    def scrape(batch, name, ago, rate):
+        for m in ROUTER_METRICS:
+            batch.add("traefik", name, m, rate if m == "request_rate" else 1.0,
+                      recorded_at=now - timedelta(minutes=ago))
 
     async def go():
         async with app.db_sessionmaker() as s:
+            for name, ago, rate in [(a, 3, 1.0), (a, 1, 2.0), (b, 5, 4.0), (a, 2, 3.0)]:
+                async with s.begin():
+                    batch = MetricBatch()
+                    scrape(batch, name, ago, rate)
+                    await batch.record(s)
             async with s.begin():
-                s.add_all([row(a, 3, 1.0), row(a, 1, 2.0), row(a, 2, 3.0), row(b, 5, 4.0)])
+                batch = MetricBatch()
+                batch.add("traefik", cert, "cert_expiry_days", 30.0, recorded_at=now)
+                await batch.record(s)
             async with s.begin():
                 return {m.router_name: m for m in await latest_per_router(s)}
 
     got = asyncio.run(go())
     assert got[a].request_rate == 2.0 and got[a].scraped_at == now - timedelta(minutes=1)
+    assert got[a].latency_p95_ms == 1.0 and got[a].response_bytes_rate == 1.0
     assert got[b].request_rate == 4.0
+    assert cert not in got
 
 
 def test_the_request_chart_sums_routers_and_weights_the_5xx_share(app):