Files
FabledSteward/plugins/traefik/routes.py
T
bvandeusenandClaude Opus 5.5 c3ba850a9c
CI / lint (push) Successful in 2s
CI / integration (push) Successful in 51s
CI / unit (push) Successful in 1m18s
CI / publish (push) Successful in 14s
fix(traefik): bucket history with date_bin; integer-epoch division never grouped scrapes (#5256)
CI 8551 integration: "integer out of range" for a 2099 timestamp.

The SQL in the log showed a second problem: `CAST(epoch AS INTEGER) / 60`
compiled to `/ CAST(60 AS NUMERIC)`, i.e. decimal division. Every scrape got
its own "bucket", so the request chart and the Traefik page's sparklines
never averaged anything. The int4 cast would also overflow in 2038.

Both now use date_bin from the epoch, as host_agent/metrics_query.py does.
The integration test adds a second scrape 30 s later in the same minute and
checks that the bucket averages the totals and weights the 5xx share over
both scrapes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-07 13:33:38 -04:00

532 lines
22 KiB
Python

# plugins/traefik/routes.py
from __future__ import annotations
from collections import defaultdict
from datetime import datetime, timedelta, timezone
from quart import Blueprint, current_app, render_template, request
from sqlalchemy import func, select, text
import json
from steward.auth.middleware import require_role
from steward.models.users import UserRole
from steward.core.time_range import parse_range, DEFAULT_RANGE, bucket_seconds, subsample
from steward.core.formatting import cert_left, status_class
from .models import TraefikAccessSummary, TraefikCert, TraefikGlobalStat, TraefikMetric
from .scheduler import router_domains
from .scraper import cert_names, covers
traefik_bp = Blueprint("traefik", __name__, template_folder="templates")
def _sparkline(values: list[float], width: int = 80, height: int = 20) -> str:
"""Generate an inline SVG sparkline from a list of float values."""
if len(values) < 2:
return f'<svg width="{width}" height="{height}"></svg>'
mn, mx = min(values), max(values)
if mx == mn:
mx = mn + 1.0
step = width / (len(values) - 1)
pts = []
for i, v in enumerate(values):
x = i * step
y = height - (v - mn) / (mx - mn) * (height - 2) - 1
pts.append(f"{x:.1f},{y:.1f}")
poly = " ".join(pts)
return (
f'<svg width="{width}" height="{height}" viewBox="0 0 {width} {height}" '
f'style="vertical-align:middle;">'
f'<polyline points="{poly}" fill="none" stroke="#6060c0" stroke-width="1.5"/>'
f'</svg>'
)
# A router serving any real share of 5xx is failing; a 4xx share above this
# is worth a look (clients hitting missing routes, auth failures).
ERROR_5XX_PCT = 0.1
ERROR_4XX_PCT = 1.0
_GRADE_ORDER = {"crit": 0, "warn": 1, "ok": 2}
def router_state(latest, thresholds: dict) -> tuple[str, str | None]:
"""(grade, why) for a router's latest scrape: crit while it serves 5xx,
warn on a p95 past the latency critical cutoff (Settings → Thresholds)
or a high 4xx share, otherwise ok with nothing to say."""
if (latest.error_rate_5xx_pct or 0) > ERROR_5XX_PCT:
return "crit", f"{latest.error_rate_5xx_pct:.1f}% of responses are 5xx"
p95 = latest.latency_p95_ms
if p95 is not None and status_class(p95, "latency", thresholds) == "crit":
return "warn", f"p95 latency is {p95:.0f} ms"
if (latest.error_rate_4xx_pct or 0) > ERROR_4XX_PCT:
return "warn", f"{latest.error_rate_4xx_pct:.1f}% of responses are 4xx"
return "ok", None
# A router missing from this many scrapes in a row is no longer reported:
# Traefik dropped it (a removed service), so its last state is history.
GONE_AFTER_SCRAPES = 3
def gone_routers(routers: list[dict], scrape_seconds: int) -> tuple[list[dict], list[dict]]:
"""(current, gone): a router whose newest row trails the newest scrape of
any router by GONE_AFTER_SCRAPES intervals is gone, newest-seen first."""
if not routers:
return [], []
newest = max(r["latest"].scraped_at for r in routers)
cutoff = newest - timedelta(seconds=GONE_AFTER_SCRAPES * max(int(scrape_seconds), 1))
current = [r for r in routers if r["latest"].scraped_at >= cutoff]
gone = sorted((r for r in routers if r["latest"].scraped_at < cutoff),
key=lambda r: r["latest"].scraped_at, reverse=True)
return current, gone
def split_routers(routers: list[dict]) -> tuple[list[dict], list[dict]]:
"""(problems, healthy): failing routers before slow or noisy ones, each
group by name; healthy routers by name."""
problems = sorted((r for r in routers if r["state"] != "ok"),
key=lambda r: (_GRADE_ORDER[r["state"]], r["name"].lower()))
healthy = sorted((r for r in routers if r["state"] == "ok"), key=lambda r: r["name"].lower())
return problems, healthy
# Newest row per router. Postgres has no skip scan, so DISTINCT or max() GROUP
# BY reads every row in retention; this walks the (router_name, scraped_at)
# index one router at a time instead: a step per router, not per row.
_LATEST_PER_ROUTER = text("""
WITH RECURSIVE names AS (
SELECT min(router_name) AS name FROM traefik_metrics
UNION ALL
SELECT (SELECT min(router_name) FROM traefik_metrics WHERE router_name > names.name)
FROM names WHERE names.name IS NOT NULL
)
SELECT m.* FROM names
JOIN LATERAL (
SELECT * FROM traefik_metrics WHERE router_name = names.name
ORDER BY scraped_at DESC LIMIT 1
) m ON true
""")
async def latest_per_router(db) -> list:
return list((await db.execute(
select(TraefikMetric).from_statement(_LATEST_PER_ROUTER))).scalars().all())
def current_certs(certs, scrape_seconds: int) -> tuple[list, list]:
"""(served, old): a certificate is served while Traefik still reports it.
Rows are upserted by serial and never deleted, so a renewed or retired
certificate lingers; one missing from the last GONE_AFTER_SCRAPES
scrapes is old (#5254)."""
certs = list(certs)
if not certs:
return [], []
newest = max(c.scraped_at for c in certs)
cutoff = newest - timedelta(seconds=GONE_AFTER_SCRAPES * max(int(scrape_seconds), 1))
return ([c for c in certs if c.scraped_at >= cutoff],
sorted((c for c in certs if c.scraped_at < cutoff), key=lambda c: c.scraped_at, reverse=True))
def router_hosts(name: str, domains: dict[str, list[str]]) -> list[str]:
"""The hosts a metrics router name serves. The API names a router
"app@docker"; metrics may carry it as "websecure-app@docker", one per
entrypoint, so a name not found as-is is tried without that prefix."""
if name in domains:
return domains[name]
for api_name, hosts in domains.items():
if name.endswith("-" + api_name):
return hosts
return []
def certs_by_router(routers: list[str], domains: dict[str, list[str]], certs, now) -> dict[str, dict]:
"""{router: {"cn", "days"}}: the soonest-expiring served certificate valid
for one of the router's hosts. A router with no hosts, or none covered,
is absent."""
out: dict[str, dict] = {}
names = [(c, cert_names(c.cn, c.sans)) for c in certs]
for router in routers:
for host in router_hosts(router, domains):
for c, cn in names:
if covers(cn, host):
days = (c.not_after - now).total_seconds() / 86400.0
if router not in out or days < out[router]["days"]:
out[router] = {"cn": c.cn or host, "days": days}
return out
def _access_log() -> dict:
"""The plugin's access_log settings ({} when unset or malformed)."""
cfg = current_app.config["PLUGINS"]["traefik"].get("access_log", {})
return cfg if isinstance(cfg, dict) else {}
@traefik_bp.get("/")
@require_role(UserRole.viewer)
async def index():
poll_interval = current_app.config.get("MONITORS_POLL_INTERVAL", 60)
access_log_enabled = bool(_access_log().get("enabled", False))
current_range = request.args.get("range", DEFAULT_RANGE)
return await render_template(
"traefik/index.html",
poll_interval=poll_interval,
access_log_enabled=access_log_enabled,
current_range=current_range,
)
@traefik_bp.get("/rows")
@require_role(UserRole.viewer)
async def rows():
"""HTMX fragment: service cards with sparklines scoped to selected time range."""
since, range_key = parse_range(request.args.get("range"))
now = datetime.now(timezone.utc)
thresholds = current_app.config.get("THRESHOLDS", {})
async with current_app.db_sessionmaker() as db:
# Every known router, with its most recent raw row
newest = {m.router_name: m for m in await latest_per_router(db)}
routers = sorted(newest)
b_secs = bucket_seconds(since)
bucket_col = _time_bucket(TraefikMetric.scraped_at, b_secs)
router_data = []
for router in routers:
# Bucketed history for sparklines — always ~80 points regardless of range
result = await db.execute(
select(
func.avg(TraefikMetric.request_rate).label("request_rate"),
func.avg(TraefikMetric.latency_p95_ms).label("latency_p95_ms"),
func.avg(TraefikMetric.error_rate_5xx_pct).label("error_rate_5xx_pct"),
func.min(TraefikMetric.scraped_at).label("scraped_at"),
bucket_col,
)
.where(TraefikMetric.router_name == router)
.where(TraefikMetric.scraped_at >= since)
.group_by(bucket_col)
.order_by(bucket_col)
)
history = result.all()
latest = newest[router]
state, why = router_state(latest, thresholds)
router_data.append({
"name": router,
"latest": latest,
"state": state,
"why": why,
"sparkline_req": _sparkline([r.request_rate for r in history]),
"sparkline_p95": _sparkline([r.latency_p95_ms for r in history]),
"sparkline_5xx": _sparkline([r.error_rate_5xx_pct for r in history]),
})
# Latest global stat (always most recent, not range-filtered)
result = await db.execute(
select(TraefikGlobalStat)
.order_by(TraefikGlobalStat.scraped_at.desc())
.limit(1)
)
global_stat = result.scalar_one_or_none()
# All certs, sorted by soonest expiry
result = await db.execute(
select(TraefikCert).order_by(TraefikCert.not_after)
)
raw_certs = result.scalars().all()
scrape = current_app.config["PLUGINS"]["traefik"]["scrape_interval_seconds"]
current, gone = gone_routers(router_data, scrape)
served, old = current_certs(raw_certs, scrape)
domains = router_domains()
def cert_row(c):
names = cert_names(c.cn, c.sans)
return {
"serial": c.serial,
"cn": c.cn or c.serial,
"sans": [n for n in names if n != c.cn],
"not_after": c.not_after,
"days_remaining": (c.not_after - now).total_seconds() / 86400.0,
# The routers whose hosts it covers; None when Traefik's API
# couldn't say which hosts each router serves.
"routers": None if domains is None else sorted(
{r["name"].split("@")[0] for r in current
for h in router_hosts(r["name"], domains) if covers(names, h)}),
}
certs = [cert_row(c) for c in served]
old_certs = [cert_row(c) for c in old]
problems, healthy = split_routers(current)
return await render_template(
"traefik/rows.html",
problems=problems,
healthy=healthy,
gone=gone,
error_4xx=ERROR_4XX_PCT,
error_5xx=ERROR_5XX_PCT,
global_stat=global_stat,
certs=certs,
old_certs=old_certs,
range_key=range_key,
)
@traefik_bp.get("/widget")
@require_role(UserRole.viewer)
async def widget():
"""The dashboard Routers widget (#5254), in the widget-shape order: a
verdict, routers in trouble with why, then the busiest healthy ones under
one header. Only routers and certificates Traefik still reports. A
certificate's expiry shows on the router it serves, and makes that router
a problem once it is inside the Settings → Thresholds TLS window."""
now = datetime.now(timezone.utc)
thresholds = current_app.config.get("THRESHOLDS", {})
scrape = current_app.config["PLUGINS"]["traefik"]["scrape_interval_seconds"]
async with current_app.db_sessionmaker() as db:
latest = await latest_per_router(db)
certs = (await db.execute(select(TraefikCert))).scalars().all()
current, _gone = gone_routers([{"name": m.router_name, "latest": m} for m in latest], scrape)
served, _old = current_certs(certs, scrape)
domains = router_domains()
by_router = certs_by_router([r["name"] for r in current], domains or {}, served, now)
for r in current:
r["state"], r["why"] = router_state(r["latest"], thresholds)
r["cert"] = by_router.get(r["name"])
if r["cert"]:
grade = status_class(r["cert"]["days"], "tls_days", thresholds)
r["cert"]["grade"] = grade
if _GRADE_ORDER.get(grade, 2) < _GRADE_ORDER[r["state"]]:
left = cert_left(r["cert"]["days"]) # "12 days", "expires today", "expired …"
r["state"], r["why"] = grade, (
f"certificate expires in {left}" if left[0].isdigit() else f"certificate {left}")
problems, healthy = split_routers(current)
healthy.sort(key=lambda r: -(r["latest"].request_rate or 0))
# With no router domains to go on, certificates can't sit on their
# routers: the expiring ones are listed after the routers instead.
loose = [] if domains is not None else sorted(
({"cn": c.cn or c.serial, "days": (c.not_after - now).total_seconds() / 86400.0} for c in served),
key=lambda c: c["days"])
loose = [c for c in loose if status_class(c["days"], "tls_days", thresholds) in ("warn", "crit")]
return await render_template(
"traefik/widget.html",
problems=problems, healthy=healthy[:ROUTERS_SHOWN],
healthy_more=max(0, len(healthy) - ROUTERS_SHOWN),
total=len(current), loose_certs=loose, domains_known=domains is not None,
)
# Healthy routers on the dashboard widget (the busiest); the page lists all.
ROUTERS_SHOWN = 6
@traefik_bp.get("/widget/access_log")
@require_role(UserRole.viewer)
async def widget_access_log():
"""HTMX dashboard widget: traffic origin summary with IP filter."""
from collections import defaultdict
ip_filter = request.args.get("ip_filter", "all")
limit = max(1, min(50, int(request.args.get("limit", 10) or 10)))
widget_id = request.args.get("wid", "0")
async with current_app.db_sessionmaker() as db:
result = await db.execute(
select(TraefikAccessSummary)
.order_by(TraefikAccessSummary.period_end.desc())
.limit(12) # last ~1 hour of 5-min windows
)
rows = result.scalars().all()
if not rows:
return await render_template("traefik/widget_access_log.html",
summary=None, widget_id=widget_id)
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)
ip_counts: dict = defaultdict(lambda: {"count": 0, "internal": True})
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.get("internal", True)
all_ips = sorted(
[{"ip": k, **v} for k, v in ip_counts.items()],
key=lambda x: x["count"], reverse=True,
)
if ip_filter == "internal":
all_ips = [e for e in all_ips if e["internal"]]
elif ip_filter == "external":
all_ips = [e for e in all_ips if not e["internal"]]
summary = {
"total": total,
"internal": internal,
"external": external,
"external_pct": (external / total * 100.0) if total else 0.0,
"top_ips": all_ips[:limit],
"ip_filter": ip_filter,
}
return await render_template("traefik/widget_access_log.html",
summary=summary, widget_id=widget_id)
def _time_bucket(column, seconds: int):
"""The start of the `seconds`-wide window a timestamp falls in, as
host_agent/metrics_query.py buckets. Integer epoch / int rendered as
numeric division (no two scrapes shared a bucket) and overflowed int4
after 2038 (#5256)."""
return func.date_bin(func.make_interval(0, 0, 0, 0, 0, 0, int(seconds)), column,
func.to_timestamp(0)).label("bucket")
async def request_history(db, since, bucket_seconds: int) -> list[dict]:
"""[{"at", "req_rate", "err_pct"}] per time bucket since `since`: each
scrape summed over routers (the total req/s), averaged over the bucket;
the 5xx share weighted by each router's traffic, so a quiet router's
errors don't count as much as a busy one's."""
per_scrape = (
select(
TraefikMetric.scraped_at.label("at"),
func.sum(TraefikMetric.request_rate).label("req"),
func.sum(TraefikMetric.request_rate * TraefikMetric.error_rate_5xx_pct).label("err_weighted"),
)
.where(TraefikMetric.scraped_at >= since)
.group_by(TraefikMetric.scraped_at)
.subquery()
)
bucket_col = _time_bucket(per_scrape.c.at, bucket_seconds)
rows = (await db.execute(
select(
func.avg(per_scrape.c.req).label("req_rate"),
func.sum(per_scrape.c.err_weighted).label("err_weighted"),
func.sum(per_scrape.c.req).label("req_sum"),
func.min(per_scrape.c.at).label("at"),
bucket_col,
)
.group_by(bucket_col)
.order_by(bucket_col)
)).all()
return [{"at": r.at, "req_rate": round(r.req_rate or 0, 3),
"err_pct": round((r.err_weighted or 0) / r.req_sum, 2) if r.req_sum else 0.0}
for r in rows]
@traefik_bp.get("/widget/request_chart")
@require_role(UserRole.viewer)
async def widget_request_chart():
"""HTMX dashboard widget: requests per second across all routers, and the
share of them answered 5xx, over the last few hours (#5256). Each scrape
is summed over routers first (the total), then averaged per bucket; the
5xx share is weighted by each router's traffic, not averaged per router."""
hours = max(1, min(24, int(request.args.get("hours", 6) or 6)))
widget_id = request.args.get("wid", "0")
since = datetime.now(timezone.utc) - timedelta(hours=hours)
b_secs = max(60, (hours * 3600) // 80) # ~80 buckets
async with current_app.db_sessionmaker() as db:
history = await request_history(db, since, b_secs)
labels = [int(r["at"].timestamp() * 1000) for r in history] # epoch ms; the browser shows local time
req_rates = [r["req_rate"] for r in history]
err_pcts = [r["err_pct"] for r in history]
return await render_template(
"traefik/widget_request_chart.html",
labels_json=json.dumps(labels),
req_rate_json=json.dumps(req_rates),
error_rate_json=json.dumps(err_pcts),
widget_id=widget_id,
hours=hours,
now_req=req_rates[-1] if req_rates else None,
peak_req=max(req_rates) if req_rates else None,
peak_err=max(err_pcts) if err_pcts else None,
)
@traefik_bp.get("/traffic")
@require_role(UserRole.viewer)
async def traffic():
"""HTMX fragment: traffic origin summary scoped to selected time range."""
since, range_key = parse_range(request.args.get("range"))
async with current_app.db_sessionmaker() as db:
result = await db.execute(
select(TraefikAccessSummary)
.where(TraefikAccessSummary.period_end >= since)
.order_by(TraefikAccessSummary.period_end.asc())
# No LIMIT — all summaries in range needed for accurate totals;
# sparkline subsamples to 80 points afterwards
)
rows = result.scalars().all()
if not rows:
al = _access_log()
return await render_template(
"traefik/traffic.html", summary=None, history=[], range_key=range_key,
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]
history = [
{
"period_end": r.period_end,
"external_pct": (r.external_requests / r.total_requests * 100.0)
if r.total_requests else 0.0,
}
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)]
),
}
return await render_template(
"traefik/traffic.html", summary=summary, history=history, range_key=range_key
)