# 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 asyncio
import logging
import math
import re
import socket
from collections import defaultdict
from datetime import datetime, timezone

import httpx

from steward.core import prometheus

logger = logging.getLogger(__name__)


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


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


# The counter families the per-router figures are computed from. Across
# several Traefiks these add up (requests, bytes and latency buckets are each
# replica's share); everything else in an exposition is a gauge or a
# timestamp, which summing would turn into nonsense.
SUMMED = ("traefik_service_requests_total", "traefik_service_responses_bytes_total",
          "traefik_service_request_duration_seconds_bucket")


def _sum(scrapes: list[ParsedMetrics]) -> ParsedMetrics:
    out: ParsedMetrics = {}
    for scrape in scrapes:
        for name in SUMMED:
            into = out.setdefault(name, {})
            for labels, value in scrape.get(name, {}).items():
                into[labels] = into.get(labels, 0.0) + value
    return out


def combine(pairs: list[tuple[ParsedMetrics, ParsedMetrics | None, float]],
            ) -> tuple[ParsedMetrics, ParsedMetrics | None, float]:
    """One (current, previous, elapsed) for the router arithmetic out of each
    Traefik's (current, its own previous scrape, seconds since it).

    Only the Traefiks that have a baseline are added up, current and previous
    alike, so the sum's rise is the sum of each one's own rise: a replica seen
    for the first time contributes from its next scrape on, and is never read
    as traffic. When none has a baseline yet, every current comes back with no
    previous, so each router is listed at zero."""
    based = [(cur, prev, el) for cur, prev, el in pairs if prev is not None]
    if not based:
        return _sum([cur for cur, _, _ in pairs]), None, 0.0
    return (_sum([cur for cur, _, _ in based]), _sum([prev for _, prev, _ in based]),
            sum(el for _, _, el in based) / len(based))


def merge_global_stats(stats: list[dict]) -> dict:
    """Traefik's own figures across its replicas: connections and memory add
    up; reloads and the last reload are the newest any replica reports."""
    def values(key):
        return [s[key] for s in stats if s.get(key) is not None]
    conns, mem = values("open_conns_total"), values("process_memory_bytes")
    reloads, last = values("config_reloads_total"), values("config_last_reload_at")
    return {"open_conns_total": sum(conns) if conns else None,
            "config_reloads_total": max(reloads) if reloads else None,
            "config_last_reload_at": max(last) if last else None,
            "process_memory_bytes": sum(mem) if mem else None}


def unique_certs(scrapes: list[ParsedMetrics]) -> list[dict]:
    """The certificates any replica serves, each once (by serial)."""
    seen: dict[str, dict] = {}
    for scrape in scrapes:
        for cert in extract_certs(scrape):
            seen.setdefault(cert["serial"], cert)
    return list(seen.values())


# ──────────────────────────────────────────────────────────────────────────────
# Per-router metric computation
# ──────────────────────────────────────────────────────────────────────────────

def _rise(prev: dict[frozenset, float] | None, labels: frozenset, value: float) -> float:
    """How far a counter rose since the previous scrape. On the first scrape
    (`prev` None) nothing can be compared, so 0. A series the previous scrape
    didn't have started from 0 since then, so all of it counts: Traefik makes
    a router's first 502 series when the first 502 happens (#5711)."""
    if prev is None:
        return 0.0
    return max(0.0, value - prev.get(labels, 0.0))


def compute_status_codes(current: ParsedMetrics, previous: ParsedMetrics | None) -> dict[str, dict[str, int]]:
    """{service: {code: responses since the previous scrape}} for each 4xx and
    5xx code that occurred, from traefik_service_requests_total{code} (#5711).
    Codes that didn't occur are left out; the first scrape has none."""
    prev = None if previous is None else previous.get("traefik_service_requests_total", {})
    out: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
    for labels, value in current.get("traefik_service_requests_total", {}).items():
        label_dict = dict(labels)
        service, code = label_dict.get("service", ""), label_dict.get("code", "")
        if not service or code[:1] not in ("4", "5"):
            continue
        count = round(_rise(prev, labels, value))
        if count:
            out[service][code] += count
    return {service: dict(codes) for service, codes in out.items()}


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 | None,  # None: no response since the last scrape
            "latency_p95_ms": float | None,
            "latency_p99_ms": float | None,
            "response_bytes_rate": float,    # bytes/sec
        }}
    """
    first = not previous
    prev = previous or {}
    elapsed = max(elapsed_seconds, 1.0)

    # ── Request counters ──────────────────────────────────────────────────────
    req_samples = current.get("traefik_service_requests_total", {})
    prev_req = None if first else 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", "")
        delta = _rise(prev_req, labels, 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 = None if first else 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
        service_bytes[service] += _rise(prev_bytes, labels, value)

    # ── Histogram buckets ─────────────────────────────────────────────────────
    hist_buckets = current.get("traefik_service_request_duration_seconds_bucket", {})
    prev_hist = None if first else prev.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)
        latency = service_buckets(hist_buckets, prev_hist, service)
        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(latency, 0.50),
            "latency_p95_ms": _percentile_ms(latency, 0.95),
            "latency_p99_ms": _percentile_ms(latency, 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 service_buckets(current: dict[frozenset, float], previous: dict[frozenset, float] | None,
                    service: str) -> list[tuple[float, float]]:
    """The service's latency histogram since the previous scrape, as sorted
    (le, responses) pairs: each bucket's rise, summed over every label set
    (code, method, protocol) that shares its `le`. Empty on the first scrape.

    Traefik's buckets count from its start, so reading them as they stand
    gave a percentile of every response since then, a figure that barely
    moved (nextcloud's flat 100 ms p95); and walking every label set's
    buckets as one list stopped at whichever came first (#5741)."""
    if previous is None:
        return []
    by_le: dict[float, float] = defaultdict(float)
    for labels, count in current.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)
        except ValueError:
            continue
        by_le[le] += _rise(previous, labels, count)
    return sorted(by_le.items())


def _percentile_ms(buckets: list[tuple[float, float]], pct: float) -> float | None:
    """Linear interpolation of a percentile of `buckets` (service_buckets'),
    in ms; None when no response arrived since the previous scrape, as an
    idle router has no latency to report. A percentile past the last finite
    bucket reads as that bucket's edge: at least that long."""
    total = buckets[-1][1] if buckets else 0.0   # the +Inf bucket counts every response
    if total <= 0.0:
        return None
    target = total * pct
    for i, (le, count) in enumerate(buckets):
        if count < target:
            continue
        if i == 0:
            return le * (target / count) * 1000.0
        prev_le, prev_count = buckets[i - 1]
        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 None


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


async def _addresses(host: str, port: int) -> list[str]:
    """Every address `host` resolves to, in the order given, each once."""
    infos = await asyncio.get_running_loop().getaddrinfo(host, port, type=socket.SOCK_STREAM)
    return list(dict.fromkeys(info[4][0] for info in infos))


async def fetch_instances(metrics_url: str) -> list[ParsedMetrics]:
    """A scrape of every Traefik the metrics URL's host name resolves to.

    A name for several replicas (a swarm service's `tasks.<service>`, a
    Kubernetes headless Service, round-robin DNS) is scraped address by
    address, each with the original Host header and TLS name, so every
    replica's traffic is counted (#5844). A name with one address is one
    scrape, as before. Two answers from the same process (localhost over IPv4
    and IPv6) are kept once. A replica that doesn't answer is left out with a
    warning; none answering raises."""
    url = httpx.URL(metrics_url)
    host = url.host
    port = url.port or (443 if url.scheme == "https" else 80)
    try:
        addresses = await _addresses(host, port)
    except OSError:
        addresses = [host]
    if len(addresses) <= 1:
        return [await fetch_metrics(metrics_url)]

    async def one(address: str) -> ParsedMetrics:
        target = url.copy_with(host=address)
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.get(target, headers={"Host": url.netloc.decode()},
                                        extensions={"sni_hostname": host})
            response.raise_for_status()
        return parse_prometheus(response.text)

    results = await asyncio.gather(*(one(a) for a in addresses), return_exceptions=True)
    scrapes: dict[str, ParsedMetrics] = {}
    for address, result in zip(addresses, results):
        if isinstance(result, BaseException):
            logger.warning("Traefik at %s (%s) didn't answer: %s", address, host, result)
            continue
        key = instance_of(result) or address
        scrapes.setdefault(key, result)
    if not scrapes:
        raise RuntimeError(f"no Traefik behind {host} answered ({len(addresses)} addresses)")
    return list(scrapes.values())


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


async def fetch_error_page_services(base: str) -> frozenset[str]:
    """The services Traefik's `errors` middlewares send error pages to, named
    as the metrics name them ("error-pages-service@swarm"). Such a service
    answers with the code of the error it covers for: its 4xx and 5xx are
    other routers' errors, already graded on them."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(f"{base}/http/middlewares", params={"per_page": 1000})
        response.raise_for_status()
        return error_page_services(response.json())


def error_page_services(middlewares: list[dict]) -> frozenset[str]:
    """fetch_error_page_services' reading of /api/http/middlewares: a service
    named without a provider belongs to its middleware's provider."""
    out = set()
    for m in middlewares:
        service = (m.get("errors") or {}).get("service")
        if not service:
            continue
        if "@" not in service:
            provider = m.get("provider") or (m.get("name", "").partition("@")[2])
            service = f"{service}@{provider}" if provider else service
        out.add(service)
    return frozenset(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