From 98440d8de6b002cf124bcb22a95088275b35c4aa Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 15 Oct 2026 14:36:10 -0400 Subject: [PATCH] fix(traefik): count each Traefik process only against its own last scrape (#5844) One metrics URL can reach several Traefik processes: a swarm service's replicas behind the routing mesh or its VIP, a Kubernetes Service's pods, a keepalived pair after failover, or the same Traefik after a restart. Each keeps its own counters, so comparing one with another read the second one's lifetime totals as new traffic: retired routers flashed 100% 502 and an idle router showed 834 req/s whenever the scrape moved to the other replica. Each scrape is now keyed by Traefik's process_start_time_seconds and compared only with the same process's previous scrape. A process seen for the first time (or not in the last hour) sets its baseline and counts nothing. Without a start time every scrape is one process, as before. Co-Authored-By: Claude Opus 5.5 --- docs/plugins/traefik.md | 2 + plugins/traefik/scheduler.py | 22 +++++----- plugins/traefik/scraper.py | 47 ++++++++++++++++++++ tests/plugins/traefik/test_baselines.py | 57 +++++++++++++++++++++++++ 4 files changed, 116 insertions(+), 12 deletions(-) create mode 100644 tests/plugins/traefik/test_baselines.py diff --git a/docs/plugins/traefik.md b/docs/plugins/traefik.md index f21c10b..04d97a3 100644 --- a/docs/plugins/traefik.md +++ b/docs/plugins/traefik.md @@ -14,6 +14,8 @@ On each scrape: Request rates and error rates are computed as deltas between the current and previous scrape, divided by elapsed time. Latency percentiles (p50, p95, p99) are approximated via linear interpolation over histogram buckets — these are estimates, not exact percentiles. They cover the responses since the previous scrape (each bucket's rise, summed over every status code and method), not every response since Traefik started. A scrape in which a router answered nothing records no latency, shown as "—"; one past the last bucket (5 s by default) reads as that bucket's edge. +When the metrics URL can reach more than one Traefik (a swarm service's replicas behind the routing mesh or its VIP, a Kubernetes Service's pods, a keepalived pair after a failover, or the same Traefik after a restart), each process keeps its own counters. Steward tells them apart by `process_start_time_seconds` and compares each scrape only with the same process's previous one. The first scrape from a process it hasn't seen in the last hour sets that process's baseline and counts nothing. Through a single address the figures are those of whichever replica answered, not the sum of all of them. Scraping every replica and adding them up is not supported yet. + --- ## Configuration diff --git a/plugins/traefik/scheduler.py b/plugins/traefik/scheduler.py index 4271ae5..8cbe5ec 100644 --- a/plugins/traefik/scheduler.py +++ b/plugins/traefik/scheduler.py @@ -12,9 +12,8 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -# Module-level state: previous scrape data for delta computation -_prev_metrics: dict = {} -_prev_time: float = 0.0 +# Each Traefik process's last scrape, the baseline its counters rise from (#5844). +_baselines = None # Router name -> host names from Traefik's API, refreshed each scrape (#5254). # None until the API has answered once, and again whenever it stops answering, @@ -58,11 +57,12 @@ def make_scrape_task(app: "Quart") -> ScheduledTask: async def _do_scrape(app: "Quart") -> None: """Fetch Traefik metrics, write to DB, and emit plugin_metrics.""" - global _prev_metrics, _prev_time, _router_domains, _error_page_services + global _baselines, _router_domains, _error_page_services from datetime import datetime, timezone from .models import TraefikCert, TraefikGlobalStat, TraefikMetric, TraefikStatusCode from .scraper import ( + Baselines, api_base, compute_global_stats, compute_router_metrics, @@ -93,16 +93,14 @@ async def _do_scrape(app: "Quart") -> None: except Exception as exc: logger.debug("Traefik middlewares not read (%s); keeping the last error-page services", exc) - now = time.monotonic() - elapsed = now - _prev_time if _prev_time else 60.0 - router_metrics = compute_router_metrics(current, _prev_metrics or None, elapsed) - global_stats = compute_global_stats(current, _prev_metrics or None) - status_codes = compute_status_codes(current, _prev_metrics or None) + if _baselines is None: + _baselines = Baselines() + previous, elapsed = _baselines.swap(current, time.monotonic()) + router_metrics = compute_router_metrics(current, previous, elapsed) + global_stats = compute_global_stats(current, previous) + status_codes = compute_status_codes(current, previous) cert_list = extract_certs(current) - _prev_metrics = current - _prev_time = now - scraped_at = datetime.now(timezone.utc) async with app.db_sessionmaker() as session: diff --git a/plugins/traefik/scraper.py b/plugins/traefik/scraper.py index e44facc..5b7911a 100644 --- a/plugins/traefik/scraper.py +++ b/plugins/traefik/scraper.py @@ -39,6 +39,53 @@ def parse_prometheus(text: str) -> ParsedMetrics: return metrics +# ────────────────────────────────────────────────────────────────────────────── +# Which Traefik answered (#5844) +# ────────────────────────────────────────────────────────────────────────────── +# +# One metrics URL can reach more than one Traefik: a swarm service's replicas +# behind the routing mesh or its VIP, a Kubernetes Service's pods, a keepalived +# pair after failover, or the same Traefik after a restart. Each process keeps +# its own counters, so comparing one process's scrape with another's reads the +# second one's lifetime totals as new traffic (a retired router flashing 100% +# 502, an idle one at 800 req/s). Each process is told apart by its start time, +# which Traefik has exposed since v2.0, and is only ever compared with itself. + +# A process not heard from for this long has restarted or gone; its baseline +# is dropped rather than kept forever. +BASELINE_TTL_SECONDS = 3600 + + +def instance_of(current: ParsedMetrics) -> str: + """The Traefik process a scrape came from: its process_start_time_seconds, + or "" when the exposition doesn't carry one (then every scrape is taken to + be the same process, as before #5844).""" + start = current.get("process_start_time_seconds", {}) + return f"{next(iter(start.values())):.2f}" if start else "" + + +class Baselines: + """Each Traefik process's last scrape and when it was taken.""" + + def __init__(self) -> None: + self._seen: dict[str, tuple[ParsedMetrics, float]] = {} + + def swap(self, current: ParsedMetrics, now: float) -> tuple[ParsedMetrics | None, float]: + """(the previous scrape of the process `current` came from, seconds + since it), and keep `current` as that process's newest. A process seen + for the first time has no previous scrape: (None, 0.0), so nothing is + counted until it has a baseline of its own.""" + key = instance_of(current) + prev = self._seen.get(key) + self._seen[key] = (current, now) + for k, (_, at) in list(self._seen.items()): + if now - at > BASELINE_TTL_SECONDS: + del self._seen[k] + if prev is None or now - prev[1] > BASELINE_TTL_SECONDS: + return None, 0.0 + return prev[0], now - prev[1] + + # ────────────────────────────────────────────────────────────────────────────── # Per-router metric computation # ────────────────────────────────────────────────────────────────────────────── diff --git a/tests/plugins/traefik/test_baselines.py b/tests/plugins/traefik/test_baselines.py new file mode 100644 index 0000000..000382d --- /dev/null +++ b/tests/plugins/traefik/test_baselines.py @@ -0,0 +1,57 @@ +"""#5844: one metrics URL can reach several Traefik processes (swarm replicas, +pods, a keepalived pair, a restart). Each is counted only against its own last +scrape, so one replica's lifetime totals never read as another's new traffic.""" +from plugins.traefik.scraper import BASELINE_TTL_SECONDS, Baselines, compute_router_metrics, instance_of + +REQ = "traefik_service_requests_total" + + +def _scrape(started: float | None, **counts): + """A scrape from the process started at `started`, with {code: total} for qbittorrent@file.""" + out = {REQ: {frozenset({"service": "qbittorrent@file", "code": c, "method": "GET"}.items()): v + for c, v in counts.items()}} + if started is not None: + out["process_start_time_seconds"] = {frozenset(): started} + return out + + +def test_each_process_is_compared_only_with_itself(): + b = Baselines() + a1, b1 = _scrape(1000.0, **{"200": 10}), _scrape(2000.0, **{"502": 336}) + assert b.swap(a1, 0.0) == (None, 0.0) + # The other replica answers: it has no baseline yet, so nothing is counted. + assert b.swap(b1, 60.0) == (None, 0.0) + # Back to the first: compared with its own scrape two minutes ago. + a2 = _scrape(1000.0, **{"200": 22}) + assert b.swap(a2, 120.0) == (a1, 120.0) + assert b.swap(_scrape(2000.0, **{"502": 336}), 180.0) == (b1, 120.0) + + +def test_landing_on_the_other_replica_no_longer_reads_its_totals_as_traffic(): + """The live symptom: an idle retired router flashing 336 502s, 100% 5xx.""" + b = Baselines() + b.swap(_scrape(1000.0, **{"200": 10}), 0.0) + previous, elapsed = b.swap(_scrape(2000.0, **{"502": 336}), 60.0) + got = compute_router_metrics(_scrape(2000.0, **{"502": 336}), previous, elapsed) + assert got["qbittorrent@file"]["request_rate"] == 0.0 + assert got["qbittorrent@file"]["error_rate_5xx_pct"] == 0.0 + + +def test_a_restarted_traefik_starts_a_new_baseline(): + b = Baselines() + b.swap(_scrape(1000.0, **{"200": 5000}), 0.0) + assert b.swap(_scrape(3000.0, **{"200": 3}), 60.0) == (None, 0.0) + + +def test_without_a_start_time_every_scrape_is_one_process_as_before(): + b = Baselines() + first = _scrape(None, **{"200": 1}) + assert instance_of(first) == "" + b.swap(first, 0.0) + assert b.swap(_scrape(None, **{"200": 2}), 60.0) == (first, 60.0) + + +def test_a_baseline_older_than_the_ttl_is_dropped_not_compared(): + b = Baselines() + b.swap(_scrape(1000.0, **{"200": 1}), 0.0) + assert b.swap(_scrape(1000.0, **{"200": 900}), BASELINE_TTL_SECONDS + 1) == (None, 0.0)