From c336f1f7745f233336621c167205a8c52b11f4ae Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 17 Mar 2026 21:37:34 -0400 Subject: [PATCH] feat: traefik prometheus scraper (parse, compute rates and percentiles) --- plugins/traefik/scraper.py | 189 +++++++++++++++++++++++++++++++++++++ 1 file changed, 189 insertions(+) create mode 100644 plugins/traefik/scraper.py diff --git a/plugins/traefik/scraper.py b/plugins/traefik/scraper.py new file mode 100644 index 0000000..2703b39 --- /dev/null +++ b/plugins/traefik/scraper.py @@ -0,0 +1,189 @@ +# plugins/traefik/scraper.py +"""Prometheus text-format scraper and per-router metrics calculator for Traefik. + +Traefik exposes Prometheus metrics at /metrics. This module: +1. Fetches the endpoint via httpx +2. Parses the Prometheus text format (counters + histograms) +3. Computes per-router request rate (delta/elapsed), error rates (% 4xx, % 5xx), + and latency percentiles (p50, p95, p99) via linear histogram interpolation +""" +from __future__ import annotations +import re +from collections import defaultdict + +import httpx + + +# ────────────────────────────────────────────────────────────────────────────── +# Prometheus text format parser +# ────────────────────────────────────────────────────────────────────────────── + +# Parsed sample: {metric_name: {frozenset(label_items): float}} +ParsedMetrics = dict[str, dict[frozenset, float]] + +_SAMPLE_RE = re.compile( + r'^([a-zA-Z_:][a-zA-Z0-9_:]*)(?:\{([^}]*)\})?\s+' + r'([+-]?(?:\d+\.?\d*|\.\d+)(?:[eE][+-]?\d+)?|[+-]?Inf|NaN)' +) +_LABEL_RE = re.compile(r'([a-zA-Z_][a-zA-Z0-9_]*)="([^"\\]*(?:\\.[^"\\]*)*)"') + + +def parse_prometheus(text: str) -> ParsedMetrics: + """Parse Prometheus text exposition format into nested dicts. + + Returns: {metric_name: {frozenset({(label, value), ...}): float}} + """ + metrics: ParsedMetrics = {} + for line in text.splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + m = _SAMPLE_RE.match(line) + if not m: + continue + name, labels_str, value_str = m.groups() + try: + value = float(value_str) + except ValueError: + continue + labels = frozenset( + (lm.group(1), lm.group(2)) + for lm in _LABEL_RE.finditer(labels_str or "") + ) + metrics.setdefault(name, {})[labels] = value + return metrics + + +# ────────────────────────────────────────────────────────────────────────────── +# Per-router metric computation +# ────────────────────────────────────────────────────────────────────────────── + +def compute_router_metrics( + current: ParsedMetrics, + previous: ParsedMetrics | None, + elapsed_seconds: float, +) -> dict[str, dict[str, float]]: + """Compute per-router derived metrics from two consecutive scrapes. + + Args: + current: Parsed metrics from the latest scrape. + previous: Parsed metrics from the prior scrape (None on first call). + elapsed_seconds: Time between scrapes (for rate calculation). + + Returns: + {router_name: { + "request_rate": float, # requests/sec + "error_rate_4xx_pct": float, # % 4xx of total + "error_rate_5xx_pct": float, # % 5xx of total + "latency_p50_ms": float, # approx ms + "latency_p95_ms": float, + "latency_p99_ms": float, + }} + """ + prev = previous or {} + elapsed = max(elapsed_seconds, 1.0) + + # ── Request counters ────────────────────────────────────────────────────── + req_samples = current.get("traefik_router_requests_total", {}) + prev_req = prev.get("traefik_router_requests_total", {}) + + router_total: dict[str, float] = defaultdict(float) + router_4xx: dict[str, float] = defaultdict(float) + router_5xx: dict[str, float] = defaultdict(float) + + for labels, value in req_samples.items(): + label_dict = dict(labels) + router = label_dict.get("router", "") + if not router: + continue + code = label_dict.get("code", "") + prev_value = prev_req.get(labels, value) # no delta on first scrape + delta = max(0.0, value - prev_value) + router_total[router] += delta + if code.startswith("4"): + router_4xx[router] += delta + elif code.startswith("5"): + router_5xx[router] += delta + + # ── Histogram buckets ───────────────────────────────────────────────────── + hist_buckets = current.get("traefik_router_request_duration_seconds_bucket", {}) + + # Collect all known routers (from both counter and histogram data) + all_routers: set[str] = set(router_total.keys()) + for labels in hist_buckets: + router = dict(labels).get("router", "") + if router: + all_routers.add(router) + + result: dict[str, dict[str, float]] = {} + + for router in all_routers: + total = router_total.get(router, 0.0) + result[router] = { + "request_rate": total / elapsed, + "error_rate_4xx_pct": (router_4xx.get(router, 0.0) / total * 100.0) + if total > 0 else 0.0, + "error_rate_5xx_pct": (router_5xx.get(router, 0.0) / total * 100.0) + if total > 0 else 0.0, + "latency_p50_ms": _percentile_ms(hist_buckets, router, 0.50), + "latency_p95_ms": _percentile_ms(hist_buckets, router, 0.95), + "latency_p99_ms": _percentile_ms(hist_buckets, router, 0.99), + } + + return result + + +def _percentile_ms( + buckets: dict[frozenset, float], + router: str, + pct: float, +) -> float: + """Linear interpolation of a Prometheus histogram percentile, returned in ms.""" + router_buckets: list[tuple[float, float]] = [] + for labels, count in buckets.items(): + label_dict = dict(labels) + if label_dict.get("router") != router: + continue + le_str = label_dict.get("le", "") + try: + le = float("inf") if le_str == "+Inf" else float(le_str) + router_buckets.append((le, count)) + except ValueError: + continue + + if not router_buckets: + return 0.0 + + router_buckets.sort(key=lambda t: t[0]) + total = router_buckets[-1][1] # +Inf bucket count == total requests + if total == 0.0: + return 0.0 + + target = total * pct + for i, (le, count) in enumerate(router_buckets): + if count < target: + continue + # Found the bucket that first reaches target count + if i == 0: + # Interpolate from 0 to first bucket upper bound + ratio = target / count if count > 0 else 0.0 + return le * ratio * 1000.0 + prev_le, prev_count = router_buckets[i - 1] + if count == prev_count: + return prev_le * 1000.0 + fraction = (target - prev_count) / (count - prev_count) + return (prev_le + fraction * (le - prev_le)) * 1000.0 + + return router_buckets[-2][0] * 1000.0 # last finite bucket upper bound + + +# ────────────────────────────────────────────────────────────────────────────── +# HTTP fetch +# ────────────────────────────────────────────────────────────────────────────── + +async def fetch_metrics(metrics_url: str) -> ParsedMetrics: + """Fetch and parse Prometheus metrics from the given URL.""" + async with httpx.AsyncClient(timeout=10.0) as client: + response = await client.get(metrics_url) + response.raise_for_status() + return parse_prometheus(response.text)