# 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 # 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 from datetime import datetime, timezone from .models import TraefikCert, TraefikGlobalStat, TraefikMetric from .scraper import ( api_base, compute_global_stats, compute_router_metrics, extract_certs, fetch_metrics, 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 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) 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"]), ]: # 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, ) # ── 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"], )