diff --git a/docs/core/mcp.md b/docs/core/mcp.md index 72f2c9a..80ff24a 100644 --- a/docs/core/mcp.md +++ b/docs/core/mcp.md @@ -80,6 +80,7 @@ Each role includes the tools of the roles above it in this list. | Audit | `list_audit_events` | | Docker plugin | `list_containers`, `get_container`, `list_swarm_services`, `search_container_logs`, `get_docker_disk_usage` | | Host agent (built in, always on) | `list_agents`, `list_maintenance_plans` (each plan's group, weekly slot, next run and latest run: how far it got, who it skipped and why), `get_maintenance_job` (a host's update jobs or one job: state, packages updated, whether it rebooted, why it failed) | +| Traefik plugin | `list_routers` (each router graded as the Traefik page grades it, with the status codes behind a problem, what they mean and the swarm service behind it), `get_router` (one router's history, codes, normal and anomalies), `list_certificates` (served certificates, days left, the routers they cover), `get_traffic` (the access log's clients, countries and routers, and Traefik's own figures) | ### Operator: run playbooks, change monitoring diff --git a/plugins/traefik/__init__.py b/plugins/traefik/__init__.py index 06b0623..612b6f9 100644 --- a/plugins/traefik/__init__.py +++ b/plugins/traefik/__init__.py @@ -62,3 +62,13 @@ def get_alert_metrics() -> dict[str, list[str]]: return {"traefik": ["request_rate", "error_rate_4xx_pct", "error_rate_5xx_pct", "latency_p50_ms", "latency_p95_ms", "latency_p99_ms", "response_bytes_rate", "cert_expiry_days"]} + + +def register_mcp_tools(register) -> None: + from .mcp_tools import register_tools + register_tools(register) + + +async def mcp_overview(db) -> list[dict]: + from .mcp_tools import mcp_overview as findings + return await findings(db) diff --git a/plugins/traefik/mcp_tools.py b/plugins/traefik/mcp_tools.py new file mode 100644 index 0000000..bf1f3ca --- /dev/null +++ b/plugins/traefik/mcp_tools.py @@ -0,0 +1,304 @@ +"""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. +""" +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.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 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) -> 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). + + include_old: also list certificates Traefik stopped serving (renewed or + retired), which Steward keeps on record. + `warn_days` / `crit_days` are the Settings → Thresholds cutoffs + get_overview reports a served certificate at. + """ + 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", {}) + out = {"certificates": [_cert_view(c, now, routers_for(c)) for c in served], + "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).""" + 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 ""), + ids={"router": r["name"]}, next=f'get_router(router="{r["name"]}")')) + return out + + +async def mcp_overview(db) -> list[dict]: + thresholds, scrape = _cfg() + now = now_utc() + current, _gone = await _graded(db, now, thresholds, scrape) + served, _old = page.current_certs((await db.execute(select(TraefikCert))).scalars().all(), scrape) + return router_findings(current) + cert_findings(served, now, thresholds) + + +def register_tools(register) -> None: + for fn in (list_routers, get_router, list_certificates, get_traffic): + register(fn, role=UserRole.viewer, read_only=True) diff --git a/plugins/traefik/routes.py b/plugins/traefik/routes.py index 51c662e..2d034de 100644 --- a/plugins/traefik/routes.py +++ b/plugins/traefik/routes.py @@ -709,6 +709,42 @@ async def widget_request_chart(): ) +def summarize_access(rows, top: int = 10) -> dict: + """The access-log summaries in a range added up: totals, the internal and + external split, and the top client IPs, countries and routers.""" + total = sum(r.total_requests for r in rows) + external = sum(r.external_requests for r in rows) + ip_counts: dict = defaultdict(lambda: {"count": 0, "internal": True, "country": None}) + country_counts: dict[str, int] = defaultdict(int) + router_counts: dict[str, int] = defaultdict(int) + for row in rows: + for entry in json.loads(row.top_ips_json): + k = entry["ip"] + ip_counts[k]["count"] += entry["count"] + ip_counts[k]["internal"] = entry["internal"] + ip_counts[k]["country"] = entry.get("country") + for entry in json.loads(row.top_countries_json): + country_counts[entry["country"]] += entry["count"] + for entry in json.loads(row.top_routers_json): + router_counts[entry["router"]] += entry["count"] + + def ranked(counts: dict, label: str) -> list[dict]: + return sorted(({label: k, "count": v} for k, v in counts.items()), + key=lambda x: x["count"], reverse=True)[:top] + + return { + "total": total, + "internal": sum(r.internal_requests for r in rows), + "external": external, + "total_bytes": sum(r.total_bytes for r in rows), + "external_pct": (external / total * 100.0) if total else 0.0, + "top_ips": sorted(({"ip": k, **v} for k, v in ip_counts.items()), + key=lambda x: x["count"], reverse=True)[:top], + "top_countries": ranked(country_counts, "country"), + "top_routers": ranked(router_counts, "router"), + } + + @traefik_bp.get("/traffic") @require_role(UserRole.viewer) async def traffic(): @@ -732,38 +768,7 @@ async def traffic(): access_log_enabled=bool(al.get("enabled", False)), access_log_path=al.get("path", ""), ) - total = sum(r.total_requests for r in rows) - internal = sum(r.internal_requests for r in rows) - external = sum(r.external_requests for r in rows) - total_bytes = sum(r.total_bytes for r in rows) - - ip_counts: dict = defaultdict(lambda: {"count": 0, "internal": True, "country": None}) - country_counts: dict[str, int] = defaultdict(int) - router_counts: dict[str, int] = defaultdict(int) - - for row in rows: - for entry in json.loads(row.top_ips_json): - k = entry["ip"] - ip_counts[k]["count"] += entry["count"] - ip_counts[k]["internal"] = entry["internal"] - ip_counts[k]["country"] = entry.get("country") - for entry in json.loads(row.top_countries_json): - country_counts[entry["country"]] += entry["count"] - for entry in json.loads(row.top_routers_json): - router_counts[entry["router"]] += entry["count"] - - top_ips = sorted( - [{"ip": k, **v} for k, v in ip_counts.items()], - key=lambda x: x["count"], reverse=True - )[:10] - top_countries = sorted( - [{"country": k, "count": v} for k, v in country_counts.items()], - key=lambda x: x["count"], reverse=True - )[:10] - top_routers = sorted( - [{"router": k, "count": v} for k, v in router_counts.items()], - key=lambda x: x["count"], reverse=True - )[:10] + summary = summarize_access(rows) history = [ { @@ -774,19 +779,7 @@ async def traffic(): for r in rows ] - summary = { - "total": total, - "internal": internal, - "external": external, - "total_bytes": total_bytes, - "external_pct": (external / total * 100.0) if total else 0.0, - "top_ips": top_ips, - "top_countries": top_countries, - "top_routers": top_routers, - "sparkline_external": _sparkline( - [h["external_pct"] for h in subsample(history, 80)] - ), - } + summary["sparkline_external"] = _sparkline([h["external_pct"] for h in subsample(history, 80)]) return await render_template( "traefik/traffic.html", summary=summary, history=history, range_key=range_key diff --git a/tests/plugins/traefik/test_mcp_tools.py b/tests/plugins/traefik/test_mcp_tools.py new file mode 100644 index 0000000..6a13f70 --- /dev/null +++ b/tests/plugins/traefik/test_mcp_tools.py @@ -0,0 +1,74 @@ +"""The traefik plugin's MCP tools (#5838): which certificates and routers the +overview names, how a router is found by name, and what a row says.""" +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace as NS + +import pytest + +import plugins.traefik as traefik_plugin +from plugins.traefik import mcp_tools as t +from steward.models.users import UserRole + +NOW = datetime(2026, 10, 15, 12, 0, tzinfo=timezone.utc) +THRESHOLDS = {"tls_days": {"warn": 14, "crit": 3}} + + +def _cert(cn, days): + return NS(cn=cn, serial=f"s-{cn}", sans=None, not_after=NOW + timedelta(days=days), scraped_at=NOW) + + +def test_a_served_certificate_is_a_finding_inside_its_warning_days(): + found = {f["subject"]: f for f in t.cert_findings( + [_cert("fine.example", 40), _cert("soon.example", 10), _cert("now.example", 2), + _cert("gone.example", -1)], NOW, THRESHOLDS)} + assert set(found) == {"soon.example", "now.example", "gone.example"} + assert found["soon.example"]["severity"] == "warning" + assert found["now.example"]["severity"] == "critical" + assert "expired" in found["gone.example"]["summary"] + + +def test_only_a_failing_router_is_an_overview_finding(): + routers = [ + {"name": "app@swarm", "state": "crit", "why": "12.0% of responses are 5xx over the last 15 min", + "meaning": {"meaning": "Mostly 502: Traefik couldn't reach the app."}}, + {"name": "radarr@swarm", "state": "warn", "why": "7.2% of responses are 4xx"}, + {"name": "ok@swarm", "state": "ok", "why": None}, + ] + [f] = t.router_findings(routers) + assert f["severity"] == "critical" and f["subject"] == "app@swarm" + assert "Mostly 502" in f["summary"] and f["next"] == 'get_router(router="app@swarm")' + + +def test_a_router_is_found_with_or_without_its_provider(): + names = ["radarr@swarm", "sonarr@swarm", "api@internal", "web@docker", "web@file"] + assert t._match("radarr@swarm", names) == "radarr@swarm" + assert t._match("RADARR", names) == "radarr@swarm" + with pytest.raises(ValueError, match="matches web@docker, web@file"): + t._match("web", names) + with pytest.raises(ValueError, match="list_routers shows them"): + t._match("nope", names) + + +def test_a_row_names_codes_and_meaning_only_when_troubled(): + latest = NS(request_rate=2.5, error_rate_4xx_pct=7.25, error_rate_5xx_pct=0.0, latency_p50_ms=12.0, + latency_p95_ms=40.0, latency_p99_ms=80.0, scraped_at=NOW) + troubled = t._router_row({"name": "radarr@swarm", "state": "warn", "why": "4xx", "latest": latest, + "codes": [("404", 30)], "meaning": {"action": "tune"}, "service": "arr_radarr", + "normed": True}, {"radarr@swarm": {"cn": "radarr.example", "days": 50.0}}) + assert troubled["codes"] == [["404", 30]] and troubled["swarm_service"] == "arr_radarr" + assert troubled["graded_on_normal"] is True and troubled["certificate"]["days_left"] == 50.0 + healthy = t._router_row({"name": "ok@swarm", "state": "ok", "why": None, "latest": latest}, {}) + assert "codes" not in healthy and "meaning" not in healthy and "certificate" not in healthy + + +def test_the_plugin_registers_its_tools_for_viewers(): + got = {} + traefik_plugin.register_mcp_tools(lambda fn, role, read_only: got.update({fn.__name__: (role, read_only)})) + assert got == {name: (UserRole.viewer, True) + for name in ("list_routers", "get_router", "list_certificates", "get_traffic")} + assert callable(traefik_plugin.mcp_overview) + + +async def test_list_routers_refuses_an_unknown_state(): + with pytest.raises(ValueError, match="state must be one of"): + await t.list_routers(state="broken")