FabledSteward/plugins/traefik/scraper.py

# 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 math
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
        if math.isinf(le):
            # Percentile falls in the open (last_finite_le, +Inf] bucket;
            # cap at the last finite bucket upper bound.
            return prev_le * 1000.0
        fraction = (target - prev_count) / (count - prev_count)
        return (prev_le + fraction * (le - prev_le)) * 1000.0

    return 0.0  # unreachable: +Inf bucket always satisfies count >= target


# ──────────────────────────────────────────────────────────────────────────────
# 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)