bvandeusenandClaude Opus 5.5
81995051dc
The scraper folded traefik_service_requests_total{code} into two shares and
dropped the code, so a problem couldn't say whether it was scanners (404), a
login wall (401/403) or the app down (502/503/504). Each scrape now stores the
responses per router and code that occurred (traefik_status_codes, migration
traefik_006), pruned with the router metrics and left out of backups (System
180 ruling: Traefik history is not backed up). A problem card lists its codes
over the last 15 minutes, most first, or says none occurred, or that codes
aren't recorded yet.
A counter series new since the previous scrape now counts from 0 instead of
being skipped: Traefik creates a router's 502 series at its first 502, so a
first burst of 5xx read as 0% until it happened again.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
235 lines
8.8 KiB
Python
235 lines
8.8 KiB
Python
# 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, TraefikStatusCode
|
|
from .scraper import (
|
|
api_base,
|
|
compute_global_stats,
|
|
compute_router_metrics,
|
|
compute_status_codes,
|
|
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)
|
|
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"]),
|
|
]:
|
|
# 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"],
|
|
)
|