# 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 from steward.core import prometheus # ────────────────────────────────────────────────────────────────────────────── # Prometheus text format parser # ────────────────────────────────────────────────────────────────────────────── # Parsed sample: {metric_name: {frozenset(label_items): float}} ParsedMetrics = dict[str, dict[frozenset, float]] def parse_prometheus(text: str) -> ParsedMetrics: """The exposition keyed by sample name, then by its label set — the shape the per-router arithmetic below reads. Parsing is steward.core.prometheus.""" metrics: ParsedMetrics = {} for family in prometheus.parse(text).values(): for sample in family.samples: metrics.setdefault(sample.name, {})[frozenset(sample.labels)] = sample.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