FabledSteward/plugins/traefik/scheduler.py

# plugins/traefik/scheduler.py
from __future__ import annotations
import logging
import time
from typing import TYPE_CHECKING

from datetime import datetime
from steward.core.scheduler import ScheduledTask

if TYPE_CHECKING:
    from quart import Quart

logger = logging.getLogger(__name__)

# Module-level state: previous scrape data for delta computation
_prev_metrics: dict = {}
_prev_time: float = 0.0

# Router name -> host names from Traefik's API, refreshed each scrape (#5254).
# None until the API has answered once, and again whenever it stops answering,
# so readers can tell "no domains" from "don't know".
_router_domains: dict[str, list[str]] | None = None


def router_domains() -> dict[str, list[str]] | None:
    return _router_domains


# The services Traefik's errors middlewares send error pages to (#5780), as
# last read from its API. Kept through an API blip: that config rarely moves,
# and dropping it would grade the error pages as failing for that while.
_error_page_services: frozenset[str] = frozenset()


def error_page_services() -> frozenset[str]:
    return _error_page_services


# Access log state
_geoip_db = None
_access_log_last_tick: "datetime | None" = None


def make_scrape_task(app: "Quart") -> ScheduledTask:
    """Return a ScheduledTask that scrapes Traefik metrics on the configured interval."""
    interval = app.config["PLUGINS"]["traefik"]["scrape_interval_seconds"]

    async def scrape() -> None:
        await _do_scrape(app)

    return ScheduledTask(
        name="traefik_scrape",
        coro_factory=scrape,
        interval_seconds=int(interval),
        run_on_startup=True,
    )


async def _do_scrape(app: "Quart") -> None:
    """Fetch Traefik metrics, write to DB, and emit plugin_metrics."""
    global _prev_metrics, _prev_time, _router_domains, _error_page_services

    from datetime import datetime, timezone
    from .models import TraefikCert, TraefikGlobalStat, TraefikMetric, TraefikStatusCode
    from .scraper import (
        api_base,
        compute_global_stats,
        compute_router_metrics,
        compute_status_codes,
        extract_certs,
        fetch_metrics,
        fetch_error_page_services,
        fetch_router_domains,
    )
    from steward.core.alerts import MetricBatch

    metrics_url: str = app.config["PLUGINS"]["traefik"]["metrics_url"]

    try:
        current = await fetch_metrics(metrics_url)
    except Exception:
        logger.exception("Traefik scrape failed (url=%s)", metrics_url)
        return
    cfg = app.config["PLUGINS"]["traefik"]
    try:
        _router_domains = await fetch_router_domains(api_base(metrics_url, cfg.get("api_url") or ""))
    except Exception as exc:
        if _router_domains is not None:
            logger.warning("Traefik API stopped answering (%s); certificates shown without routers", exc)
        _router_domains = None
    try:
        _error_page_services = await fetch_error_page_services(api_base(metrics_url, cfg.get("api_url") or ""))
    except Exception as exc:
        logger.debug("Traefik middlewares not read (%s); keeping the last error-page services", exc)

    now = time.monotonic()
    elapsed = now - _prev_time if _prev_time else 60.0
    router_metrics = compute_router_metrics(current, _prev_metrics or None, elapsed)
    global_stats = compute_global_stats(current, _prev_metrics or None)
    status_codes = compute_status_codes(current, _prev_metrics or None)
    cert_list = extract_certs(current)

    _prev_metrics = current
    _prev_time = now

    scraped_at = datetime.now(timezone.utc)

    async with app.db_sessionmaker() as session:
        async with session.begin():
            batch = MetricBatch()
            # ── Per-service metrics ───────────────────────────────────────────
            for router_name, metrics in router_metrics.items():
                row = TraefikMetric(
                    router_name=router_name,
                    scraped_at=scraped_at,
                    request_rate=metrics["request_rate"],
                    error_rate_4xx_pct=metrics["error_rate_4xx_pct"],
                    error_rate_5xx_pct=metrics["error_rate_5xx_pct"],
                    latency_p50_ms=metrics["latency_p50_ms"],
                    latency_p95_ms=metrics["latency_p95_ms"],
                    latency_p99_ms=metrics["latency_p99_ms"],
                    response_bytes_rate=metrics["response_bytes_rate"],
                )
                session.add(row)

                for metric_name, value in [
                    ("request_rate", metrics["request_rate"]),
                    ("error_rate_4xx_pct", metrics["error_rate_4xx_pct"]),
                    ("error_rate_5xx_pct", metrics["error_rate_5xx_pct"]),
                    ("latency_p50_ms", metrics["latency_p50_ms"]),
                    ("latency_p95_ms", metrics["latency_p95_ms"]),
                    ("latency_p99_ms", metrics["latency_p99_ms"]),
                    ("response_bytes_rate", metrics["response_bytes_rate"]),
                ]:
                    if value is None:       # no responses since the last scrape: no latency
                        continue
                    # Stamped with the scrape, so a router's latest figures
                    # (routes.latest_per_router) all carry the same moment.
                    batch.add(
                        source_module="traefik",
                        resource_name=router_name,
                        metric_name=metric_name,
                        value=value,
                        recorded_at=scraped_at,
                    )

            # ── 4xx and 5xx responses by code (#5711) ─────────────────────────
            for router_name, codes in status_codes.items():
                for code, count in codes.items():
                    session.add(TraefikStatusCode(router_name=router_name, scraped_at=scraped_at,
                                                  code=code, count=count))

            # ── Global stats ──────────────────────────────────────────────────
            session.add(TraefikGlobalStat(scraped_at=scraped_at, **global_stats))

            # ── TLS certs (upsert by serial) ──────────────────────────────────
            for cert in cert_list:
                existing = await session.get(TraefikCert, cert["serial"])
                if existing:
                    existing.cn = cert["cn"]
                    existing.sans = cert["sans"]
                    existing.not_after = cert["not_after"]
                    existing.scraped_at = scraped_at
                else:
                    session.add(TraefikCert(
                        serial=cert["serial"],
                        cn=cert["cn"],
                        sans=cert["sans"],
                        not_after=cert["not_after"],
                        scraped_at=scraped_at,
                    ))

                # Emit cert_expiry_days for alert rules
                days_remaining = (cert["not_after"] - scraped_at).total_seconds() / 86400.0
                batch.add(
                    source_module="traefik",
                    resource_name=cert["cn"] or cert["serial"],
                    metric_name="cert_expiry_days",
                    value=max(0.0, days_remaining),
                )

            await batch.record(session)

    logger.debug(
        "Traefik scrape: %d router(s), %d cert(s)",
        len(router_metrics),
        len(cert_list),
    )


def make_access_log_task(app: "Quart") -> ScheduledTask:
    """Return a ScheduledTask that processes new Traefik access log lines."""
    global _geoip_db
    cfg = app.config["PLUGINS"]["traefik"].get("access_log", {})
    if not isinstance(cfg, dict):
        cfg = {}
    log_path: str = cfg.get("path", "/var/log/traefik/access.log")
    geoip_path: str = cfg.get("geoip_db", "")

    from .access_log import open_geoip
    _geoip_db = open_geoip(geoip_path or None)
    if _geoip_db:
        logger.info("Traefik access log: GeoIP database loaded from %s", geoip_path)
    else:
        logger.info("Traefik access log: GeoIP not configured, country lookup disabled")

    interval = app.config["PLUGINS"]["traefik"]["scrape_interval_seconds"]

    async def process() -> None:
        await _do_access_log(app, log_path)

    return ScheduledTask(
        name="traefik_access_log",
        coro_factory=process,
        interval_seconds=int(interval),
        run_on_startup=True,
    )


async def _do_access_log(app: "Quart", log_path: str) -> None:
    """Read new access log lines, aggregate, and persist a summary."""
    global _access_log_last_tick

    from datetime import timezone
    from .access_log import read_new_lines, aggregate
    from .models import TraefikAccessSummary

    now = datetime.now(timezone.utc)
    period_start = _access_log_last_tick or now
    _access_log_last_tick = now

    entries = read_new_lines(log_path)
    summary = aggregate(entries, geoip_db=_geoip_db)

    if summary is None:
        return

    async with app.db_sessionmaker() as session:
        async with session.begin():
            session.add(TraefikAccessSummary(
                period_start=period_start,
                period_end=now,
                **summary,
            ))

    logger.debug(
        "Traefik access log: %d requests (%d internal, %d external)",
        summary["total_requests"],
        summary["internal_requests"],
        summary["external_requests"],
    )