diff --git a/plugins/traefik/routes.py b/plugins/traefik/routes.py index 3673bd6..d67fc25 100644 --- a/plugins/traefik/routes.py +++ b/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]: diff --git a/plugins/traefik/scheduler.py b/plugins/traefik/scheduler.py index af49c0a..50f8a47 100644 --- a/plugins/traefik/scheduler.py +++ b/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 ────────────────────────────────────────────────── diff --git a/tests/integration/test_traefik_latest.py b/tests/integration/test_traefik_latest.py index 312e332..f69cd1b 100644 --- a/tests/integration/test_traefik_latest.py +++ b/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):