FabledSteward/plugins/traefik/scraper.py

# plugins/traefik/scraper.py
"""Prometheus text-format scraper and per-service 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-service request rate (delta/elapsed), error rates (% 4xx, % 5xx),
   latency percentiles (p50, p95, p99) via linear histogram interpolation,
   and bandwidth (bytes/sec from response bytes counter delta)
4. Extracts TLS certificate expiry info
5. Computes process-level and config health stats
"""
from __future__ import annotations
import math
import re
from collections import defaultdict
from datetime import datetime, timezone

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-service derived metrics from two consecutive scrapes.

    Uses traefik_service_* metrics (available by default).

    Returns:
        {service_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,
            "latency_p95_ms": float,
            "latency_p99_ms": float,
            "response_bytes_rate": float,    # bytes/sec
        }}
    """
    prev = previous or {}
    elapsed = max(elapsed_seconds, 1.0)

    # ── Request counters ──────────────────────────────────────────────────────
    req_samples = current.get("traefik_service_requests_total", {})
    prev_req = prev.get("traefik_service_requests_total", {})

    service_total: dict[str, float] = defaultdict(float)
    service_4xx: dict[str, float] = defaultdict(float)
    service_5xx: dict[str, float] = defaultdict(float)

    for labels, value in req_samples.items():
        label_dict = dict(labels)
        service = label_dict.get("service", "")
        if not service:
            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)
        service_total[service] += delta
        if code.startswith("4"):
            service_4xx[service] += delta
        elif code.startswith("5"):
            service_5xx[service] += delta

    # ── Response bytes counters ───────────────────────────────────────────────
    bytes_samples = current.get("traefik_service_responses_bytes_total", {})
    prev_bytes = prev.get("traefik_service_responses_bytes_total", {})

    service_bytes: dict[str, float] = defaultdict(float)
    for labels, value in bytes_samples.items():
        service = dict(labels).get("service", "")
        if not service:
            continue
        prev_value = prev_bytes.get(labels, value)
        service_bytes[service] += max(0.0, value - prev_value)

    # ── Histogram buckets ─────────────────────────────────────────────────────
    hist_buckets = current.get("traefik_service_request_duration_seconds_bucket", {})

    # Collect all known services
    all_services: set[str] = set(service_total.keys()) | set(service_bytes.keys())
    for labels in hist_buckets:
        service = dict(labels).get("service", "")
        if service:
            all_services.add(service)

    result: dict[str, dict[str, float]] = {}

    for service in all_services:
        total = service_total.get(service, 0.0)
        result[service] = {
            "request_rate": total / elapsed,
            "error_rate_4xx_pct": (service_4xx.get(service, 0.0) / total * 100.0)
            if total > 0 else 0.0,
            "error_rate_5xx_pct": (service_5xx.get(service, 0.0) / total * 100.0)
            if total > 0 else 0.0,
            "latency_p50_ms": _percentile_ms(hist_buckets, service, 0.50),
            "latency_p95_ms": _percentile_ms(hist_buckets, service, 0.95),
            "latency_p99_ms": _percentile_ms(hist_buckets, service, 0.99),
            "response_bytes_rate": service_bytes.get(service, 0.0) / elapsed,
        }

    return result


def compute_global_stats(
    current: ParsedMetrics,
    previous: ParsedMetrics | None,
) -> dict[str, float | datetime | None]:
    """Extract process-level and config health metrics from a scrape.

    Returns:
        {
            "open_conns_total": float | None,
            "config_reloads_total": float | None,
            "config_last_reload_at": datetime | None,     # last successful reload
            "process_memory_bytes": float | None,
        }
    """
    stats: dict[str, float | datetime | None] = {
        "open_conns_total": None,
        "config_reloads_total": None,
        "config_last_reload_at": None,
        "process_memory_bytes": None,
    }

    # Open connections — sum across all entrypoints/protocols
    open_conns = current.get("traefik_open_connections", {})
    if open_conns:
        stats["open_conns_total"] = sum(open_conns.values())

    # Config reloads — sum across result labels (success + failure)
    reloads = current.get("traefik_config_reloads_total", {})
    if reloads:
        stats["config_reloads_total"] = sum(reloads.values())

    # The moment of the last successful reload, as a Unix timestamp: Traefik
    # names it "last_reload_success", but it is a time, not a 0/1 flag. Read
    # as a flag, every scrape said "Reload failed" (#5355).
    last_success = current.get("traefik_config_last_reload_success", {})
    if last_success:
        # Usually a single sample with no labels
        ts = next(iter(last_success.values()))
        if ts > 0:
            stats["config_last_reload_at"] = datetime.fromtimestamp(ts, timezone.utc)

    # Process RSS memory
    mem = current.get("process_resident_memory_bytes", {})
    if mem:
        stats["process_memory_bytes"] = next(iter(mem.values()))

    return stats


def extract_certs(current: ParsedMetrics) -> list[dict]:
    """Extract TLS certificate expiry info from traefik_tls_certs_not_after.

    Returns a list of dicts:
        [{"serial": str, "cn": str, "sans": str, "not_after": datetime}, ...]
    """
    certs = []
    samples = current.get("traefik_tls_certs_not_after", {})
    for labels, ts_value in samples.items():
        label_dict = dict(labels)
        serial = label_dict.get("serial", "")
        if not serial:
            continue
        try:
            not_after = datetime.fromtimestamp(ts_value, tz=timezone.utc)
        except (ValueError, OSError, OverflowError):
            continue
        certs.append({
            "serial": serial,
            "cn": label_dict.get("cn") or None,
            "sans": label_dict.get("sans") or None,
            "not_after": not_after,
        })
    return certs


def _percentile_ms(
    buckets: dict[frozenset, float],
    service: 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("service") != service:
            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
        if i == 0:
            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):
            return prev_le * 1000.0
        fraction = (target - prev_count) / (count - prev_count)
        return (prev_le + fraction * (le - prev_le)) * 1000.0

    return 0.0


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


# ──────────────────────────────────────────────────────────────────────────────
# Router domains (Traefik API)
# ──────────────────────────────────────────────────────────────────────────────
#
# The metrics name each router but not the hosts it answers for, so which
# certificate serves which router is only knowable from Traefik's API. Read
# once per scrape; when the API isn't reachable, routers simply have no
# domains and certificates are shown on their own (#5254).

_HOST_MATCHER = re.compile(r"Host(?:SNI)?\(([^)]*)\)")
_QUOTED = re.compile(r"[`\"']([^`\"']+)[`\"']")


def rule_hosts(rule: str) -> list[str]:
    """The host names a router rule matches: Host(`a`, `b`) and HostSNI(`c`)
    give ["a", "b", "c"]; HostRegexp and path-only rules give nothing."""
    hosts: list[str] = []
    for args in _HOST_MATCHER.findall(rule or ""):
        hosts.extend(h.strip().lower() for h in _QUOTED.findall(args) if h.strip() != "*")
    return hosts


def api_base(metrics_url: str, api_url: str = "") -> str:
    """The Traefik API's base: the configured api_url, else /api on the
    metrics URL's own host (the default "traefik" entrypoint serves both)."""
    if api_url:
        return api_url.rstrip("/")
    parts = httpx.URL(metrics_url)
    return str(parts.copy_with(path="/api", query=None, fragment=None)).rstrip("/")


async def fetch_router_domains(base: str) -> dict[str, list[str]]:
    """{router name: host names} for every enabled HTTP and TCP router."""
    out: dict[str, list[str]] = {}
    async with httpx.AsyncClient(timeout=10.0) as client:
        for kind in ("http", "tcp"):
            response = await client.get(f"{base}/{kind}/routers", params={"per_page": 1000})
            if kind == "tcp" and response.status_code == 404:
                continue
            response.raise_for_status()
            for r in response.json():
                if r.get("status", "enabled") != "enabled" or not r.get("name"):
                    continue
                hosts = rule_hosts(r.get("rule", ""))
                if hosts:
                    out.setdefault(r["name"], []).extend(hosts)
    return out


def cert_names(cn: str | None, sans: str | None) -> set[str]:
    """Every name a certificate covers: its CN and SANs, lower-cased."""
    names = {n.strip().lower() for n in (sans or "").split(",") if n.strip()}
    if cn:
        names.add(cn.strip().lower())
    return names


def covers(names: set[str], host: str) -> bool:
    """Whether a certificate with these names is valid for host, a wildcard
    "*.example.com" covering one label: "a.example.com", not "a.b.example.com"."""
    if host in names:
        return True
    head, _, rest = host.partition(".")
    return bool(rest) and f"*.{rest}" in names