bvandeusenandClaude Opus 5.5
398312b8be
A Traefik that hasn't restarted keeps reporting the certificate it replaced, so codex.fabledsword.com's old one (21 days left) was listed beside its renewal (81 days) and would have raised an expiry warning on 2026-10-22. A reported certificate whose every name a later-expiring one covers is now old, on the page, the widget, the overview and MCP. list_certificates takes `limit` (default 100) with total/truncated, and says how many served certificates no HTTP router uses (`unrouted`). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
354 lines
17 KiB
Python
354 lines
17 KiB
Python
"""The traefik plugin's MCP tools and its get_overview findings (#5838).
|
|
|
|
Registered through register_mcp_tools() in __init__.py, so they exist only
|
|
while the plugin is enabled. Routers are graded exactly as the Traefik page
|
|
grades them (routes.py): over the last 15 minutes, the 5xx share against its
|
|
fixed cutoff, the 4xx share and p95 against the router's own normal where it
|
|
has one; neither share over fewer requests than Settings → Thresholds asks.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import timedelta
|
|
|
|
from sqlalchemy import select
|
|
|
|
from steward.core import anomalies as anom
|
|
from steward.core.norms import norms_at
|
|
from steward.core.time_range import bucket_seconds
|
|
from steward.mcp._context import mcp_app
|
|
from steward.mcp.tools._shape import (
|
|
age_s, bounded, clamp_limit, db_session, iso, lookback, now_utc, rnd, summarize,
|
|
)
|
|
from steward.mcp.tools.overview import finding
|
|
from steward.models.users import UserRole
|
|
|
|
from . import meaning
|
|
from . import routes as page
|
|
from .models import TraefikAccessSummary, TraefikCert, TraefikGlobalStat
|
|
from .scheduler import error_page_services, router_domains
|
|
from .scraper import cert_names, covers
|
|
|
|
STATES = ("crit", "warn", "ok")
|
|
|
|
|
|
def _cfg():
|
|
app = mcp_app()
|
|
return app.config.get("THRESHOLDS", {}), int(app.config["PLUGINS"]["traefik"]["scrape_interval_seconds"])
|
|
|
|
|
|
async def _graded(db, now, thresholds: dict, scrape: int) -> tuple[list[dict], list[dict]]:
|
|
"""(current, gone) routers, graded and explained as the page does them."""
|
|
newest = await page.latest_per_router(db)
|
|
current, gone = page.gone_routers([{"name": m.router_name, "latest": m} for m in newest], scrape)
|
|
names = [r["name"] for r in current]
|
|
page.grade_routers(current, await page.recent_window(db, names, now), thresholds,
|
|
await page.router_norms(db, names, now), error_page_services())
|
|
troubled = [r for r in current if r["state"] != "ok"]
|
|
codes = await page.status_codes(db, [r["name"] for r in troubled], now - page.STATE_WINDOW)
|
|
for r in current:
|
|
r["codes"] = codes.get(r["name"], [])
|
|
if troubled:
|
|
page.explain_problems(troubled, await page.swarm_service_names(db))
|
|
return current, gone
|
|
|
|
|
|
def _cert_view(c, now, routers: list[str] | None) -> dict:
|
|
return {"cn": c.cn or c.serial, "serial": c.serial,
|
|
"names": sorted(cert_names(c.cn, c.sans)),
|
|
"not_after": iso(c.not_after),
|
|
"days_left": rnd((c.not_after - now).total_seconds() / 86400.0, 1),
|
|
"routers": routers}
|
|
|
|
|
|
def _router_row(r: dict, certs: dict) -> dict:
|
|
latest = r["latest"]
|
|
row = {
|
|
"router": r["name"], "state": r["state"], "why": r["why"],
|
|
"requests_per_s": rnd(latest.request_rate, 3),
|
|
"error_4xx_pct": rnd(latest.error_rate_4xx_pct, 2),
|
|
"error_5xx_pct": rnd(latest.error_rate_5xx_pct, 2),
|
|
"p50_ms": rnd(latest.latency_p50_ms, 1), "p95_ms": rnd(latest.latency_p95_ms, 1),
|
|
"p99_ms": rnd(latest.latency_p99_ms, 1),
|
|
"graded_on_normal": r.get("normed", False),
|
|
"last_scraped_at": iso(latest.scraped_at),
|
|
}
|
|
if r["state"] != "ok":
|
|
row["codes"] = [[c, n] for c, n in r.get("codes", [])]
|
|
row["meaning"] = r.get("meaning")
|
|
row["swarm_service"] = r.get("service")
|
|
if r["name"] in certs:
|
|
row["certificate"] = {"cn": certs[r["name"]]["cn"], "days_left": rnd(certs[r["name"]]["days"], 1)}
|
|
return row
|
|
|
|
|
|
async def list_routers(state: str = "", limit: int = 100) -> dict:
|
|
"""Traefik's routers, graded as the Traefik page grades them: problems
|
|
first ("crit" before "warn"), then healthy ones, each by name.
|
|
|
|
state: "crit" (5xx share past its cutoff: failing), "warn" (p95 or 4xx
|
|
share past its cutoff and outside the router's usual for this hour),
|
|
"ok"; empty for all.
|
|
limit: default 100, max 500.
|
|
|
|
The grade is over the last 15 minutes, weighted by traffic; the figures
|
|
shown are the latest scrape. A troubled router carries `codes` (the 4xx
|
|
and 5xx codes it answered in that window, most first, e.g. 404 means
|
|
scanners, 502/503 the app unreachable), `meaning` (what the codes point
|
|
to, whether it is usually a fault, and `action`: "act" on the app or
|
|
"tune" the cutoff), and `swarm_service` (the service behind it, for
|
|
list_swarm_services and search_container_logs). `graded_on_normal` says
|
|
the 4xx/p95 were judged against the router's own normal. `certificate`
|
|
is the soonest-expiring served certificate for its hosts. `gone` lists
|
|
routers Traefik stopped reporting.
|
|
"""
|
|
if state and state not in STATES:
|
|
raise ValueError(f"state must be one of {', '.join(STATES)}, or empty for all.")
|
|
thresholds, scrape = _cfg()
|
|
now = now_utc()
|
|
async with db_session() as db:
|
|
current, gone = await _graded(db, now, thresholds, scrape)
|
|
served, _old = page.current_certs(
|
|
(await db.execute(select(TraefikCert))).scalars().all(), scrape)
|
|
domains = router_domains() or {}
|
|
certs = page.certs_by_router([r["name"] for r in current], domains, served, now)
|
|
problems, healthy = page.split_routers(current)
|
|
rows = [_router_row(r, certs) for r in problems + healthy if not state or r["state"] == state]
|
|
error_5xx, error_4xx = page.error_cutoffs(thresholds)
|
|
out = bounded(rows, clamp_limit(limit, 100, 500), key="routers",
|
|
more="Pass a larger `limit`, or a `state`.")
|
|
out["cutoffs"] = {"error_5xx_pct": error_5xx, "error_4xx_pct": error_4xx,
|
|
"min_requests_15m": page.min_requests(thresholds)}
|
|
out["gone"] = [{"router": r["name"], "last_scraped_at": iso(r["latest"].scraped_at)} for r in gone[:20]]
|
|
return out
|
|
|
|
|
|
def _match(want: str, names: list[str]) -> str:
|
|
want = (want or "").strip()
|
|
for test in (lambda n: n == want, lambda n: n.lower() == want.lower(),
|
|
lambda n: n.split("@")[0].lower() == want.lower()):
|
|
found = [n for n in names if test(n)]
|
|
if len(found) == 1:
|
|
return found[0]
|
|
if len(found) > 1:
|
|
raise ValueError(f"{want!r} matches {', '.join(sorted(found))}; name one with its @provider.")
|
|
raise ValueError(f"No router {want!r}; list_routers shows them.")
|
|
|
|
|
|
async def get_router(router: str, range: str = "24h") -> dict:
|
|
"""One router: its grade now, its history over `range`, the status codes
|
|
it answered, what is normal for it at this hour, and its anomalies.
|
|
|
|
router: its name, with or without the @provider ("radarr@swarm",
|
|
"radarr").
|
|
range: "1h", "6h", "24h" (default), "7d" or "30d".
|
|
|
|
`series` summarizes requests/s, p95 ms and the 4xx and 5xx shares over the
|
|
range (min/avg/max/last, ≤60 points each). `codes` are the 4xx and 5xx
|
|
codes answered over the range, most first. `normal` is its learned usual
|
|
for this hour (4xx share, p95), null until it has enough history.
|
|
`anomalies` are its metric anomalies of the last 30 days (get_anomaly
|
|
explains one).
|
|
"""
|
|
since, key = lookback(range)
|
|
thresholds, scrape = _cfg()
|
|
now = now_utc()
|
|
async with db_session() as db:
|
|
current, gone = await _graded(db, now, thresholds, scrape)
|
|
everyone = {r["name"]: r for r in current + gone}
|
|
name = _match(router, list(everyone))
|
|
r = everyone[name]
|
|
history = (await page.router_history(db, [name], since, max(60, int((now - since).total_seconds() // 60))))[name]
|
|
codes = (await page.status_codes(db, [name], since)).get(name, [])
|
|
norms = await norms_at(db, [("traefik", name, m) for m in page.NORMED], now)
|
|
found = await anom.list_anomalies(db, source="traefik", resource=name,
|
|
since=now - timedelta(days=30), limit=10)
|
|
if name not in {c["name"] for c in current}:
|
|
r = {**r, "state": "gone", "why": "Traefik no longer reports this router"}
|
|
|
|
def series(attr):
|
|
return summarize([(p.bucket, float(getattr(p, attr))) for p in history if getattr(p, attr) is not None])
|
|
|
|
out = {
|
|
"router": name, "state": r["state"], "why": r["why"],
|
|
"meaning": r.get("meaning"), "swarm_service": r.get("service"),
|
|
"range": key,
|
|
"series": {"requests_per_s": series("request_rate"), "p95_ms": series("latency_p95_ms"),
|
|
"error_4xx_pct": series("error_rate_4xx_pct"),
|
|
"error_5xx_pct": series("error_rate_5xx_pct")},
|
|
"codes": [[c, n] for c, n in codes],
|
|
"normal": {m: {"usual_hold": rnd(n.usual_hold, 3), "usual_peak": rnd(n.usual_peak, 3),
|
|
"kind": n.kind, "days_seen": n.days_seen}
|
|
for (_s, _r, m), n in norms.items()} or None,
|
|
"anomalies": [{"anomaly_id": a.id, "metric": a.metric_name, "review": a.review,
|
|
"started_at": iso(a.started_at), "ongoing": a.ended_at is None,
|
|
"peak_level": rnd(a.peak_level, 3), "usual_hold": rnd(a.usual_hold, 3)}
|
|
for a in found],
|
|
}
|
|
if r["state"] == "ok" and r.get("why"):
|
|
out["note"] = r["why"]
|
|
return out
|
|
|
|
|
|
async def list_certificates(include_old: bool = False, limit: int = 100) -> dict:
|
|
"""The TLS certificates Traefik serves, soonest expiry first: names
|
|
covered, expiry, days left, and the routers whose hosts each covers
|
|
(null when Traefik's API couldn't say which hosts a router serves; []
|
|
when no current HTTP router serves its names, as for a certificate
|
|
Traefik holds for a retired service, or one only a TCP router uses).
|
|
|
|
include_old: also list certificates Traefik stopped serving (renewed or
|
|
retired; a renewed one counts as old though Traefik still reports
|
|
it), which Steward keeps on record.
|
|
limit: certificates to return, default 100, max 500.
|
|
`warn_days` / `crit_days` are the Settings → Thresholds cutoffs
|
|
get_overview reports a served certificate at. `unrouted` counts the
|
|
served certificates with no router.
|
|
"""
|
|
thresholds, scrape = _cfg()
|
|
now = now_utc()
|
|
async with db_session() as db:
|
|
certs = (await db.execute(select(TraefikCert).order_by(TraefikCert.not_after))).scalars().all()
|
|
newest = await page.latest_per_router(db)
|
|
served, old = page.current_certs(certs, scrape)
|
|
domains = router_domains()
|
|
names = [m.router_name for m in newest]
|
|
|
|
def routers_for(c):
|
|
if domains is None:
|
|
return None
|
|
covered = cert_names(c.cn, c.sans)
|
|
return sorted({n for n in names for h in page.router_hosts(n, domains) if covers(covered, h)})
|
|
|
|
tls = thresholds.get("tls_days", {})
|
|
views = [_cert_view(c, now, routers_for(c)) for c in served]
|
|
out = bounded(views, clamp_limit(limit, 100, 500), key="certificates",
|
|
more="Pass a larger `limit`.")
|
|
out.update({"unrouted": sum(1 for v in views if v["routers"] == []),
|
|
"warn_days": tls.get("warn", 14), "crit_days": tls.get("crit", 3)})
|
|
if include_old:
|
|
out["old"] = [_cert_view(c, now, None) | {"last_seen_at": iso(c.scraped_at)} for c in old[:50]]
|
|
return out
|
|
|
|
|
|
async def get_traffic(range: str = "24h") -> dict:
|
|
"""Who reached Traefik over `range`, from its access log: requests in
|
|
total, internal (private networks) and external, bytes, and the top
|
|
client IPs (with country when the GeoIP database is set), countries and
|
|
routers. Also Traefik's own process figures (open connections, memory,
|
|
when its configuration last loaded).
|
|
|
|
range: "1h", "6h", "24h" (default), "7d" or "30d".
|
|
`access_log: false` means the access log isn't being read (Settings →
|
|
Plugins → Traefik), so there are no traffic figures.
|
|
"""
|
|
since, key = lookback(range)
|
|
now = now_utc()
|
|
async with db_session() as db:
|
|
rows = (await db.execute(select(TraefikAccessSummary)
|
|
.where(TraefikAccessSummary.period_end >= since)
|
|
.order_by(TraefikAccessSummary.period_end))).scalars().all()
|
|
stat = (await db.execute(select(TraefikGlobalStat)
|
|
.order_by(TraefikGlobalStat.scraped_at.desc()).limit(1))).scalar_one_or_none()
|
|
cfg = mcp_app().config["PLUGINS"]["traefik"].get("access_log", {})
|
|
out = {"range": key, "access_log": bool(isinstance(cfg, dict) and cfg.get("enabled")),
|
|
"traffic": None, "traefik": None}
|
|
if rows:
|
|
summary = page.summarize_access(rows)
|
|
summary["external_pct"] = rnd(summary["external_pct"], 1)
|
|
summary["from"], summary["to"] = iso(rows[0].period_start), iso(rows[-1].period_end)
|
|
out["traffic"] = summary
|
|
if stat is not None:
|
|
out["traefik"] = {"open_connections": stat.open_conns_total, "memory_bytes": stat.process_memory_bytes,
|
|
"config_reloads": stat.config_reloads_total,
|
|
"config_loaded_at": iso(stat.config_last_reload_at),
|
|
"config_loaded_age_s": age_s(stat.config_last_reload_at, now),
|
|
"scraped_at": iso(stat.scraped_at)}
|
|
return out
|
|
|
|
|
|
# ── get_overview ─────────────────────────────────────────────────────────────
|
|
|
|
def cert_findings(certs, now, thresholds: dict) -> list[dict]:
|
|
"""A served certificate within its warning days (critical within its
|
|
critical days, or expired)."""
|
|
tls = thresholds.get("tls_days", {})
|
|
warn, crit = float(tls.get("warn", 14)), float(tls.get("crit", 3))
|
|
out = []
|
|
for c in certs:
|
|
days = (c.not_after - now).total_seconds() / 86400.0
|
|
if days > warn:
|
|
continue
|
|
name = c.cn or c.serial
|
|
left = "expired" if days < 0 else f"expires in {days:.1f} days"
|
|
out.append(finding("critical" if days <= crit else "warning", "traefik_cert_expiring", name,
|
|
f"TLS certificate {name} served by Traefik {left} ({iso(c.not_after)})",
|
|
ids={"serial": c.serial}, next="list_certificates()"))
|
|
return out
|
|
|
|
|
|
def router_findings(routers: list[dict]) -> list[dict]:
|
|
"""A router answering 5xx past its cutoff: failing. 4xx and latency are
|
|
left to the metric anomalies, which judge them against the router's
|
|
normal (System 182). `since` is when the failing began, as the page's
|
|
verdict says it (r["since"], None when too recent for a bucket to show)."""
|
|
out = []
|
|
for r in routers:
|
|
if r["state"] != "crit":
|
|
continue
|
|
said = (r.get("meaning") or {}).get("meaning")
|
|
out.append(finding("critical", "traefik_router_failing", r["name"],
|
|
f"Traefik router {r['name']}: {r['why']}" + (f". {said}" if said else ""),
|
|
since=r.get("since"), ids={"router": r["name"]},
|
|
next=f'get_router(router="{r["name"]}")'))
|
|
return out
|
|
|
|
|
|
async def failing_since(db, routers: list[dict], now, thresholds: dict) -> None:
|
|
"""Sets r["since"] on each failing router from its last day of history,
|
|
as the dashboard widget reads it (#5860)."""
|
|
failing = [r["name"] for r in routers if r["state"] == "crit"]
|
|
if not failing:
|
|
return
|
|
since = now - page.WIDGET_SINCE
|
|
history = await page.router_history(db, failing, since, bucket_seconds(since))
|
|
for r in routers:
|
|
if r["name"] in history:
|
|
r["since"], _before = page.problem_since(history[r["name"]], thresholds)
|
|
|
|
|
|
async def mcp_overview(db) -> list[dict]:
|
|
thresholds, scrape = _cfg()
|
|
now = now_utc()
|
|
current, _gone = await _graded(db, now, thresholds, scrape)
|
|
await failing_since(db, current, now, thresholds)
|
|
served, _old = page.current_certs((await db.execute(select(TraefikCert))).scalars().all(), scrape)
|
|
return router_findings(current) + cert_findings(served, now, thresholds)
|
|
|
|
|
|
async def service_events(db, router: str, since, until) -> list[dict]:
|
|
"""What happened to the containers of the swarm service behind a router
|
|
(starts with their image, deaths, OOM kills, health changes), on every
|
|
host: a router anomaly has no host of its own to read events by (#5862).
|
|
[] without the Docker plugin, or when no single service matches."""
|
|
from steward.core.plugin_manager import loaded_plugin_submodule
|
|
from steward.mcp.tools._events import event
|
|
from steward.models.hosts import Host
|
|
models = loaded_plugin_submodule("docker", "models")
|
|
if models is None:
|
|
return []
|
|
service = meaning.swarm_service(router, await page.swarm_service_names(db))
|
|
if service is None:
|
|
return []
|
|
e = models.DockerEvent
|
|
rows = (await db.execute(
|
|
select(e, Host.name).join(Host, Host.id == e.host_id)
|
|
.where(e.container_name.startswith(f"{service}.", autoescape=True), e.at >= since, e.at <= until)
|
|
.order_by(e.at).limit(500))).all()
|
|
return [event(ev.at, "container", f"{ev.container_name} {ev.event}" + (f" ({ev.detail})" if ev.detail else ""),
|
|
subject=f"{host}/{ev.container_name}") for ev, host in rows]
|
|
|
|
|
|
def register_tools(register) -> None:
|
|
for fn in (list_routers, get_router, list_certificates, get_traffic):
|
|
register(fn, role=UserRole.viewer, read_only=True)
|