diff --git a/plugins/docker/__init__.py b/plugins/docker/__init__.py index 5e09d1d..36403c4 100644 --- a/plugins/docker/__init__.py +++ b/plugins/docker/__init__.py @@ -49,6 +49,16 @@ def get_blueprint(): return docker_bp +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) + + def get_nav() -> list[dict]: # Surfaces in the sidebar's Infrastructure group (fleet-wide container view). return [{"label": "Docker", "href": "/plugins/docker/"}] diff --git a/plugins/docker/mcp_tools.py b/plugins/docker/mcp_tools.py new file mode 100644 index 0000000..2f2a797 --- /dev/null +++ b/plugins/docker/mcp_tools.py @@ -0,0 +1,418 @@ +"""The docker plugin's MCP tools and its get_overview findings. + +Registered through register_mcp_tools() in __init__.py, so they exist only +while the plugin is enabled. Reads go through service.py, the same functions +the plugin's pages use. +""" +from __future__ import annotations + +import json +from datetime import datetime, timedelta + +from sqlalchemy import select + +from steward.mcp._context import mcp_app +from steward.mcp.tools._shape import ( + age_s, bounded, clamp_limit, db_session, iso, lookback, now_utc, pick, rnd, summarize, +) +from steward.mcp.tools.overview import finding +from steward.models.hosts import Host +from steward.models.users import UserRole + +from . import service as docker_svc +from .dedup import dedup_by_container_id +from .models import DockerDiskUsage, DockerSwarmNode, DockerSwarmService +from .swarm_view import build_swarm_services + +PROBLEMS = ("restarting", "crash_looping", "unhealthy", "oom_killed", "exited_nonzero") +CRASH_LOOP_DEATHS = 3 # default; Settings → Thresholds sets it +LOG_LINE_CHARS = 2000 + + +# ── pure helpers ───────────────────────────────────────────────────────────── + +def container_problems(c, failures_1h: int, failed_24h: bool, + crash_loop_deaths: int = CRASH_LOOP_DEATHS) -> list[str]: + """What is wrong with one container, most severe first; [] when nothing is.""" + status = (c.status or "").lower() + out = [] + if status == "restarting": + out.append("restarting") + if failures_1h >= crash_loop_deaths: + out.append("crash_looping") + if status == "running" and (c.health or "") == "unhealthy": + out.append("unhealthy") + if c.oom_killed: + out.append("oom_killed") + if status in ("exited", "dead") and (c.exit_code or 0) != 0 and failed_24h: + out.append("exited_nonzero") + return out + + +def cut_middle(text: str, limit: int = LOG_LINE_CHARS) -> str: + """Keep both ends of an over-long line and say how much of the middle went.""" + if text is None or len(text) <= limit: + return text + keep = limit // 2 + return f"{text[:keep]} [… {len(text) - 2 * keep} characters omitted …] {text[-keep:]}" + + +def container_row(c, host_names: dict, problems: list[str], now: datetime) -> dict: + running = (c.status or "").lower() == "running" + return { + "container_name": c.name, + "host_id": c.host_id, + "host_name": host_names.get(c.host_id), + "image": c.image, + "status": c.status, + "health": c.health, + "problems": problems, + "exit_code": c.exit_code, + "oom_killed": bool(c.oom_killed), + "restart_count": c.restart_count, + "cpu_pct": rnd(c.cpu_pct), + "mem_pct": rnd(c.mem_pct), + "mem_usage_bytes": c.mem_usage_bytes, + "mem_limit_bytes": c.mem_limit_bytes, + "started_at": iso(c.started_at), + "uptime_s": age_s(c.started_at, now) if running and c.started_at else None, + "compose_project": c.compose_project, + "swarm_service": c.service_name, + "reported_at": iso(c.scraped_at), + "reported_age_s": age_s(c.scraped_at, now), + } + + +_SEVERITY = {"restarting": "critical", "crash_looping": "critical", "unhealthy": "critical", + "oom_killed": "warning", "exited_nonzero": "warning"} +_SAYS = {"restarting": "is restarting", "crash_looping": "keeps dying", + "unhealthy": "fails its healthcheck", "oom_killed": "was OOM-killed", + "exited_nonzero": "exited with an error"} + + +def container_findings(rows: list[dict]) -> list[dict]: + out = [] + for r in rows: + if not r["problems"]: + continue + worst = r["problems"][0] + where = f" on {r['host_name']}" if r["host_name"] else "" + detail = f" (exit {r['exit_code']})" if r["exit_code"] not in (None, 0) else "" + out.append(finding( + _SEVERITY[worst], f"container_{worst}", r["container_name"], + f"container {r['container_name']}{where} {' and '.join(_SAYS[p] for p in r['problems'])}{detail}", + ids={"host_id": r["host_id"], "container_name": r["container_name"]}, + next=f'get_container(container="{r["container_name"]}", host="{r["host_id"]}")')) + return out + + +def service_findings(services: list[dict]) -> list[dict]: + out = [] + for s in services: + if s["state"] == "healthy": + continue + out.append(finding( + "critical" if s["state"] == "down" else "warning", f"swarm_service_{s['state']}", + s["name"], f"swarm service {s['name']} has {s['running']} of {s['desired']} replicas running", + ids={"service_name": s["name"]}, next="list_swarm_services()")) + return out + + +# ── loading ────────────────────────────────────────────────────────────────── + +async def _host_names(db) -> dict[str, str]: + return dict((await db.execute(select(Host.id, Host.name))).all()) + + +def _crash_loop_deaths() -> int: + th = (mcp_app().config.get("THRESHOLDS") or {}).get("crash_loop") or {} + return int(th.get("warn") or CRASH_LOOP_DEATHS) + + +async def _rows(db, now: datetime) -> list[dict]: + containers = dedup_by_container_id(await docker_svc.list_containers(db)) + failures = await docker_svc.recent_failures_by_container(db, now - timedelta(hours=24)) + hour_ago = now - timedelta(hours=1) + host_names = await _host_names(db) + deaths = _crash_loop_deaths() + rows = [] + for c in containers: + events = failures.get((c.host_id, c.name), []) + recent = sum(1 for e in events if e.at >= hour_ago) + rows.append(container_row(c, host_names, container_problems(c, recent, bool(events), deaths), now)) + return rows + + +async def _swarm(db): + containers = dedup_by_container_id(await docker_svc.list_containers(db)) + services = list((await db.execute(select(DockerSwarmService))).scalars()) + nodes = list((await db.execute(select(DockerSwarmNode))).scalars()) + host_names = await _host_names(db) + node_hostname = {n.node_id: n.hostname for n in nodes if n.hostname} + built = build_swarm_services(containers, services, node_hostname, host_names) + return built["services"], nodes, host_names + + +async def mcp_overview(db) -> list[dict]: + """get_overview findings: failing containers and under-replicated swarm services.""" + rows = await _rows(db, now_utc()) + services, _, _ = await _swarm(db) + return container_findings(rows) + service_findings(services) + + +# ── tools ──────────────────────────────────────────────────────────────────── + +async def list_containers(host: str = "", status: str = "", problem: str = "", + compose_project: str = "", limit: int = 100) -> dict: + """Docker containers across every host that runs a Steward agent with Docker. + + host: a host's name or id (list_hosts); empty for all hosts. + status: Docker's state — "running", "exited", "restarting", "paused", + "created" or "dead"; empty for all. + problem: "any" for containers with something wrong, or one of + "restarting", "crash_looping" (died repeatedly within the last hour; + the count is set under Settings → Thresholds, default 3), "unhealthy" + (failing its healthcheck), "oom_killed", "exited_nonzero" (died with an + error in the last 24h); empty for no filter. + compose_project: only containers from this Compose project. + limit: default 100, max 500. + + Containers with problems come first. Each row's `problems` lists what is + wrong, most severe first. `reported_age_s` is how old the agent's last + report is, so a large value means the row may be stale. + """ + if problem and problem != "any" and problem not in PROBLEMS: + raise ValueError(f"problem {problem!r} is not 'any' or one of {', '.join(PROBLEMS)}.") + limit = clamp_limit(limit, 100, 500) + now = now_utc() + async with db_session() as db: + rows = await _rows(db, now) + if host: + hosts = list((await db.execute(select(Host))).scalars()) + h = pick(host, hosts, kind="host", list_tool="list_hosts") + rows = [r for r in rows if r["host_id"] == h.id] + if status: + rows = [r for r in rows if (r["status"] or "").lower() == status.lower()] + if problem == "any": + rows = [r for r in rows if r["problems"]] + elif problem: + rows = [r for r in rows if problem in r["problems"]] + if compose_project: + rows = [r for r in rows if r["compose_project"] == compose_project] + rows.sort(key=lambda r: (not r["problems"], (r["status"] or "") != "running", + (r["host_name"] or "").lower(), r["container_name"])) + return bounded(rows, limit, key="containers", + more="Pass a larger `limit` (max 500), or filter by `host`, `status` or `problem`.") + + +async def _pick_container(db, container: str, host: str): + containers = await docker_svc.list_containers(db) + if host: + hosts = list((await db.execute(select(Host))).scalars()) + h = pick(host, hosts, kind="host", list_tool="list_hosts") + containers = [c for c in containers if c.host_id == h.id] + want = (container or "").strip() + if not want: + raise ValueError("pass a container name; list_containers lists them.") + same = [c for c in containers if c.name == want] or \ + [c for c in containers if c.name.lower() == want.lower()] + if len(same) == 1: + return same[0] + if same: + raise ValueError(f"{len(same)} hosts run a container named {want!r}; pass `host` " + f"(one of: {', '.join(sorted(c.host_id for c in same))}).") + # One more chance: a unique container whose name contains it (swarm task + # names look like "stack_svc.1.k2j…"). + partial = [c for c in containers if want.lower() in c.name.lower()] + if len(partial) == 1: + return partial[0] + hint = (f" Containers whose names contain it: {', '.join(sorted(c.name for c in partial)[:10])}." + if partial else "") + raise ValueError(f"no container named {want!r}{' on that host' if host else ''}.{hint} " + "list_containers lists them.") + + +async def get_container(container: str, host: str = "", range: str = "24h", events: int = 20) -> dict: + """One container in depth: state, resources over time and lifecycle events. + + container: the container's name, as list_containers shows. A unique part of + a name also works (swarm task names are long). + host: the host's name or id. Needed only when several hosts run a + container with this name. + range: the window for `cpu_pct_history` and `mem_pct_history` (min, avg, + max, p95, last and a ≤60-point series) — "1h", "6h", "24h" (default), + "7d" or "30d". `cpu_pct` and `mem_pct` are the latest reported values. + events: how many lifecycle events (start, stop, die, oom, health_change) + to return, newest first (default 20, max 100). + """ + since, key = lookback(range) + events = clamp_limit(events, 20, 100) + now = now_utc() + async with db_session() as db: + c = await _pick_container(db, container, host) + failures = (await docker_svc.recent_failures_by_container( + db, now - timedelta(hours=24))).get((c.host_id, c.name), []) + host_names = await _host_names(db) + cpu = await docker_svc.container_points(db, c.host_id, c.name, "cpu_pct", since) + mem = await docker_svc.container_points(db, c.host_id, c.name, "mem_pct", since) + evs = await docker_svc.container_events(db, c.host_id, c.name, events) + recent = sum(1 for e in failures if e.at >= now - timedelta(hours=1)) + try: + ports = json.loads(c.ports_json or "[]") + except ValueError: + ports = [] + return { + **container_row(c, host_names, + container_problems(c, recent, bool(failures), _crash_loop_deaths()), now), + "container_id": c.container_id, + "ports": ports, + "net_rx_bytes": c.net_rx_bytes, "net_tx_bytes": c.net_tx_bytes, + "blk_read_bytes": c.blk_read_bytes, "blk_write_bytes": c.blk_write_bytes, + "range": key, + "cpu_pct_history": summarize(cpu), + "mem_pct_history": summarize(mem), + "events": [{"event": e.event, "detail": e.detail, "at": iso(e.at), "age_s": age_s(e.at, now)} + for e in evs], + "next": f'search_container_logs(container="{c.name}", host="{c.host_id}")', + } + + +async def list_swarm_services(limit: int = 100) -> dict: + """Docker Swarm services with their replicas per node, and the swarm's nodes. + + `state` is "healthy" (all desired replicas running), "degraded" (some) or + "down" (none). Each replica names its host and container status; a replica + marked `ghost` runs on a node with no Steward agent, so Steward knows only + how many tasks are placed there. `nodes` lists every swarm node once, with + its role (manager/worker), availability and status. + limit: default 100, max 500. + """ + limit = clamp_limit(limit, 100, 500) + async with db_session() as db: + services, nodes, host_names = await _swarm(db) + seen, node_rows = set(), [] + for n in sorted(nodes, key=lambda n: n.updated_at, reverse=True): + if n.node_id in seen: + continue + seen.add(n.node_id) + node_rows.append({"node_id": n.node_id, "hostname": n.hostname, "role": n.role, + "leader": n.leader, "availability": n.availability, "status": n.status, + "reported_by": host_names.get(n.host_id)}) + node_rows.sort(key=lambda n: n["hostname"].lower()) + order = {"down": 0, "degraded": 1, "healthy": 2} + rows = [ + { + "service_name": s["name"], "mode": s["mode"], "state": s["state"], + "running": s["running"], "desired": s["desired"], + "replicas": [ + {"ghost": True, "host": r["host"], "count": r["count"]} if r["ghost"] else + {"ghost": False, "host": r["host"], "host_id": r["host_id"], "container_name": r["name"], + "status": r["status"], "health": r["health"], "restart_count": r["restart_count"], + "cpu_pct": rnd(r["cpu_pct"]), "mem_pct": rnd(r["mem_pct"])} + for r in s["replicas"] + ], + } + for s in sorted(services, key=lambda s: (order.get(s["state"], 3), s["name"])) + ] + return {"nodes": node_rows, **bounded(rows, limit, key="services")} + + +async def search_container_logs(container: str = "", host: str = "", text: str = "", + range: str = "1h", stream: str = "", before: str = "", + context: int = 0, limit: int = 50) -> dict: + """Search the container log lines Steward has retained, newest first. + + container: one container's name (as list_containers shows); empty searches + every container (narrow with `host`). + host: a host's name or id; empty for all hosts. + text: case-insensitive substring to match; empty matches every line. + range: how far back — "1h" (default), "6h", "24h", "7d" or "30d". Logs are + kept per container up to a size and age limit, so older lines may be gone. + stream: "stdout" or "stderr"; empty for both. + before: an ISO time; returns only lines older than it. Pass a response's + `next_before` to page further back. + context: lines of surrounding log (0-5) to include with each match, from + the same container. Only with `container` set. + limit: matches to return, default 50, max 200. + + Lines longer than 2000 characters keep both ends, and the middle is + replaced by a marker saying how much was cut. + """ + since, key = lookback(range) + limit = clamp_limit(limit, 50, 200) + context = max(0, min(int(context or 0), 5)) + if context and not container: + raise ValueError("context needs `container`: it shows lines around a match within one container's log.") + if stream and stream not in ("stdout", "stderr"): + raise ValueError(f"stream {stream!r} is not 'stdout' or 'stderr'.") + before_dt = None + if before: + try: + before_dt = datetime.fromisoformat(before.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError(f"before {before!r} is not an ISO-8601 time; pass a response's next_before.") from exc + async with db_session() as db: + host_id = name = None + if container: + c = await _pick_container(db, container, host) + host_id, name = c.host_id, c.name + elif host: + hosts = list((await db.execute(select(Host))).scalars()) + host_id = pick(host, hosts, kind="host", list_tool="list_hosts").id + rows = await docker_svc.search_logs(db, since=since, host_id=host_id, name=name, text=text, + stream=stream, before=before_dt, limit=limit + 1) + more = len(rows) > limit + rows = rows[:limit] + host_names = await _host_names(db) + matches = [] + for r in rows: + m = {"at": iso(r.ts), "container_name": r.container_name, "host_id": r.host_id, + "host_name": host_names.get(r.host_id), "stream": r.stream, "line": cut_middle(r.line)} + if context: + pre, post = await docker_svc.log_context(db, r.host_id, r.container_name, r.ts, context) + m["before"] = [cut_middle(x.line) for x in pre] + m["after"] = [cut_middle(x.line) for x in post] + matches.append(m) + out = {"range": key, "matches": matches, "returned": len(matches), "truncated": more} + if more: + # Full precision: `at` is cut to the second for reading, and a cursor + # cut the same way would skip lines logged earlier in that second. + out["next_before"] = rows[-1].ts.isoformat() + out["more"] = "More matches are older; pass next_before as `before`, or narrow with `text`." + return out + + +async def get_docker_disk_usage(host: str = "") -> dict: + """Docker disk usage per host: images, containers, volumes and build cache, in bytes. + + host: a host's name or id; empty for every host. `images_reclaimable_bytes` + is what pruning unused images would free. An admin can prune from + Steward's Docker → Disk page. + """ + async with db_session() as db: + summaries = list((await db.execute(select(DockerDiskUsage))).scalars()) + host_names = await _host_names(db) + if host: + hosts = list((await db.execute(select(Host))).scalars()) + hid = pick(host, hosts, kind="host", list_tool="list_hosts").id + summaries = [s for s in summaries if s.host_id == hid] + rows = [ + { + "host_id": s.host_id, "host_name": host_names.get(s.host_id), + "images_total": s.images_total, "images_active": s.images_active, + "images_bytes": s.images_size, "images_reclaimable_bytes": s.images_reclaimable, + "containers_count": s.containers_count, "containers_bytes": s.containers_size, + "volumes_count": s.volumes_count, "volumes_bytes": s.volumes_size, + "build_cache_bytes": s.build_cache_size, + "reported_at": iso(s.updated_at), "reported_age_s": age_s(s.updated_at), + } + for s in summaries + ] + rows.sort(key=lambda r: -(r["images_reclaimable_bytes"] or 0)) + return {"hosts": rows, "total": len(rows)} + + +def register_tools(register) -> None: + for fn in (list_containers, get_container, list_swarm_services, + search_container_logs, get_docker_disk_usage): + register(fn, role=UserRole.viewer, read_only=True) diff --git a/plugins/docker/service.py b/plugins/docker/service.py index c8936ca..90e3bf9 100644 --- a/plugins/docker/service.py +++ b/plugins/docker/service.py @@ -6,11 +6,12 @@ Quart's pipeline) call the same code. """ from __future__ import annotations -from sqlalchemy import select +from sqlalchemy import func, select +from steward.core.time_range import bucket_seconds from steward.models.hosts import Host -from .models import DockerContainer, DockerEvent, DockerLog, DockerMetric +from .models import DockerContainer, DockerEvent, DockerLog, DockerMetric, DockerMetricHourly def bucket(values: list[float], target: int = 40) -> list[float]: @@ -97,3 +98,88 @@ async def query_logs(db, host_id: str, name: str, stream: str, query: str, stmt = stmt.where(DockerLog.line.ilike(f"%{escape_like(query)}%", escape="\\")) stmt = stmt.order_by(DockerLog.ts.desc(), DockerLog.id.desc()).limit(limit) return (await db.execute(stmt)).all() + + +CONTAINER_SERIES = ("cpu_pct", "mem_pct", "mem_usage_bytes") + + +async def container_points(db, host_id: str, name: str, metric: str, since, + target: int = 120) -> list[tuple]: + """(time, average) points of one container metric since `since`, oldest first. + + Raw samples are bucket-averaged in SQL; the part of the range older than + the oldest raw sample (already rolled up and pruned by retention) comes + from the hourly table. + """ + if metric not in CONTAINER_SERIES: + raise ValueError(f"metric must be one of {', '.join(CONTAINER_SERIES)}") + raw_col = getattr(DockerMetric, metric) + width_s = bucket_seconds(since, target) + rbucket = func.date_bin( + func.make_interval(0, 0, 0, 0, 0, 0, width_s), DockerMetric.scraped_at, func.to_timestamp(0), + ).label("bucket") + raw = (await db.execute( + select(rbucket, func.avg(raw_col)) + .where(DockerMetric.host_id == host_id, DockerMetric.container_name == name, + DockerMetric.scraped_at >= since) + .group_by(rbucket).order_by(rbucket) + )).all() + oldest_raw = raw[0][0] if raw else None + hourly_stmt = ( + select(DockerMetricHourly.bucket, getattr(DockerMetricHourly, metric)) + .where(DockerMetricHourly.host_id == host_id, + DockerMetricHourly.container_name == name, + DockerMetricHourly.bucket >= since) + .order_by(DockerMetricHourly.bucket) + ) + if oldest_raw is not None: + hourly_stmt = hourly_stmt.where(DockerMetricHourly.bucket < oldest_raw) + hourly = (await db.execute(hourly_stmt)).all() + return [(t, float(v)) for t, v in hourly] + [(t, float(v)) for t, v in raw] + + +async def search_logs(db, *, since, host_id: str | None = None, name: str | None = None, + text: str = "", stream: str = "", before=None, limit: int = 50) -> list: + """Retained log lines matching a filter, newest first (limit + 1 fetched by callers + that need to know whether more exist).""" + stmt = (select(DockerLog.id, DockerLog.host_id, DockerLog.container_name, + DockerLog.ts, DockerLog.stream, DockerLog.line) + .where(DockerLog.ts >= since)) + if host_id: + stmt = stmt.where(DockerLog.host_id == host_id) + if name: + stmt = stmt.where(DockerLog.container_name == name) + if stream in ("stdout", "stderr"): + stmt = stmt.where(DockerLog.stream == stream) + if text: + stmt = stmt.where(DockerLog.line.ilike(f"%{escape_like(text)}%", escape="\\")) + if before is not None: + stmt = stmt.where(DockerLog.ts < before) + stmt = stmt.order_by(DockerLog.ts.desc(), DockerLog.id.desc()).limit(limit) + return list((await db.execute(stmt)).all()) + + +async def log_context(db, host_id: str, name: str, ts, n: int) -> tuple[list, list]: + """Up to n lines either side of `ts` in one container's log, each oldest first.""" + cols = (DockerLog.id, DockerLog.ts, DockerLog.stream, DockerLog.line) + base = select(*cols).where(DockerLog.host_id == host_id, DockerLog.container_name == name) + before = (await db.execute( + base.where(DockerLog.ts < ts).order_by(DockerLog.ts.desc(), DockerLog.id.desc()).limit(n) + )).all() + after = (await db.execute( + base.where(DockerLog.ts > ts).order_by(DockerLog.ts, DockerLog.id).limit(n) + )).all() + return list(reversed(before)), list(after) + + +async def recent_failures_by_container(db, since) -> dict[tuple[str, str], list]: + """{(host_id, container_name): [die/oom events since `since`, newest first]}.""" + rows = (await db.execute( + select(DockerEvent) + .where(DockerEvent.event.in_(("die", "oom")), DockerEvent.at >= since) + .order_by(DockerEvent.at.desc()) + )).scalars() + out: dict[tuple[str, str], list] = {} + for e in rows: + out.setdefault((e.host_id, e.container_name), []).append(e) + return out diff --git a/plugins/host_agent/__init__.py b/plugins/host_agent/__init__.py index 6a7b229..a0f2b18 100644 --- a/plugins/host_agent/__init__.py +++ b/plugins/host_agent/__init__.py @@ -21,3 +21,13 @@ def get_scheduled_tasks() -> list: def get_blueprint(): from .routes import host_agent_bp return host_agent_bp + + +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/host_agent/mcp_tools.py b/plugins/host_agent/mcp_tools.py new file mode 100644 index 0000000..77d8c98 --- /dev/null +++ b/plugins/host_agent/mcp_tools.py @@ -0,0 +1,81 @@ +"""The host_agent plugin's MCP tools and its get_overview findings.""" +from __future__ import annotations + +from datetime import timedelta + +from sqlalchemy import select + +from steward.hosts import service as host_svc +from steward.mcp._context import mcp_app +from steward.mcp.tools._shape import age_s, db_session, iso, now_utc +from steward.mcp.tools.overview import finding +from steward.models.hosts import Host +from steward.models.users import UserRole + +from .models import HostAgentRegistration + +# A registration this old that has never checked in is a failed install, not one in progress. +PENDING_GRACE = timedelta(minutes=30) + + +def agent_state(reg, now, stale_after_s: int) -> str: + """The agent's state: reporting, stale (stopped checking in) or pending (never has).""" + if reg.last_seen_at is None: + return "pending" + return "reporting" if (now - reg.last_seen_at).total_seconds() <= stale_after_s else "stale" + + +async def _agents(db): + return (await db.execute( + select(HostAgentRegistration, Host) + .join(Host, Host.id == HostAgentRegistration.host_id) + .order_by(Host.name) + )).all() + + +async def list_agents() -> dict: + """Every host agent Steward has registered, with its version and whether it is reporting. + + `state` is "reporting", "stale" (it checked in before but its last + check-in is older than the stale window, `stale_after_s`) or "pending" + (registered, never checked in, usually an install that is still running + or that failed). Each row carries the agent version, kernel, distro, arch + and the IP the agent reports. get_host shows an agent's latest metrics. + """ + now = now_utc() + stale_after = host_svc.agent_stale_after_seconds(mcp_app().config) + async with db_session() as db: + rows = await _agents(db) + agents = [ + { + "host_id": h.id, "host_name": h.name, + "state": agent_state(reg, now, stale_after), + "last_seen_at": iso(reg.last_seen_at), "last_seen_age_s": age_s(reg.last_seen_at, now), + "agent_version": reg.agent_version, "kernel": reg.kernel, "distro": reg.distro, + "arch": reg.arch, "reported_ip": reg.host_ip, + "registered_at": iso(reg.created_at), + } + for reg, h in rows + ] + counts = {s: sum(1 for a in agents if a["state"] == s) for s in ("reporting", "stale", "pending")} + return {"stale_after_s": stale_after, "counts": counts, "agents": agents, "total": len(agents)} + + +async def mcp_overview(db) -> list[dict]: + """Agents registered more than PENDING_GRACE ago that have never checked in. + + Agents that stopped reporting are already flagged by the core overview + from their metrics, so they are not repeated here. + """ + now = now_utc() + return [ + finding("warning", "agent_never_reported", h.name, + "host agent was registered but has never checked in; the install may have failed", + since=reg.created_at, ids={"host_id": h.id}, next="list_agents()") + for reg, h in await _agents(db) + if reg.last_seen_at is None and now - reg.created_at > PENDING_GRACE + ] + + +def register_tools(register) -> None: + register(list_agents, role=UserRole.viewer, read_only=True) diff --git a/steward/core/settings.py b/steward/core/settings.py index 3469b56..6ce99b7 100644 --- a/steward/core/settings.py +++ b/steward/core/settings.py @@ -79,6 +79,10 @@ DEFAULTS: dict[str, Any] = { "thresholds.load_warn": 80, "thresholds.load_crit": 100, "thresholds.temp_warn": 70, "thresholds.temp_crit": 85, "thresholds.uptime_warn": 99.0, "thresholds.uptime_crit": 95.0, + # Days left on an HTTPS monitor's certificate (lower is worse), and how + # many container deaths within an hour count as crash-looping. + "thresholds.tls_warn_days": 14, "thresholds.tls_crit_days": 3, + "thresholds.crash_loop_deaths": 3, # Docker time-series retention (rule 25 — tunable, no restart). Raw 30s # samples are heavy, so keep a short raw window then roll up to hourly # averages kept much longer; lifecycle events are light, keep a month. @@ -379,6 +383,8 @@ def to_thresholds_cfg(settings: dict[str, Any]) -> dict: "temp": {"warn": g("thresholds.temp_warn", 70), "crit": g("thresholds.temp_crit", 85), "dir": "high"}, "latency": {"warn": g("ping.threshold.good_ms", 50), "crit": g("ping.threshold.warn_ms", 200), "dir": "high"}, "uptime": {"warn": g("thresholds.uptime_warn", 99.0), "crit": g("thresholds.uptime_crit", 95.0), "dir": "low"}, + "tls_days": {"warn": g("thresholds.tls_warn_days", 14), "crit": g("thresholds.tls_crit_days", 3), "dir": "low"}, + "crash_loop": {"warn": g("thresholds.crash_loop_deaths", 3), "crit": g("thresholds.crash_loop_deaths", 3), "dir": "high"}, } diff --git a/steward/mcp/tools/overview.py b/steward/mcp/tools/overview.py index 761986c..1724cfd 100644 --- a/steward/mcp/tools/overview.py +++ b/steward/mcp/tools/overview.py @@ -19,7 +19,6 @@ from ._shape import age_s, db_session, iso, now_utc, rnd logger = logging.getLogger(__name__) -TLS_WARN_DAYS = 14 FINDINGS_CAP = 50 _SEVERITY = {"critical": 0, "warning": 1} @@ -39,7 +38,11 @@ def rank(findings: list[dict]) -> list[dict]: -(f.get("age_s") or 0), f.get("subject", ""))) -def monitor_findings(monitors, states: dict, host_names: dict, now) -> list[dict]: +def monitor_findings(monitors, states: dict, host_names: dict, now, + tls: dict | None = None) -> list[dict]: + """Failing monitors, and HTTPS certificates inside the TLS window + (`tls` is THRESHOLDS["tls_days"]: warn/crit days left).""" + tls = tls or {"warn": 14, "crit": 3} out = [] for m in monitors: if not m.enabled: @@ -58,9 +61,9 @@ def monitor_findings(monitors, states: dict, host_names: dict, now) -> list[dict next=f'get_monitor(monitor="{m.id}")')) elif latest.tls_expires_at is not None: days = (latest.tls_expires_at - now).total_seconds() / 86400.0 - if days <= TLS_WARN_DAYS: + if days <= tls["warn"]: out.append(finding( - "critical" if days <= 3 else "warning", "tls_expiring", m.name, + "critical" if days <= tls["crit"] else "warning", "tls_expiring", m.name, f"TLS certificate for {m.target} expires in {days:.1f} days", ids=ids, next=f'get_monitor(monitor="{m.id}")')) return out @@ -151,9 +154,9 @@ async def get_overview() -> dict: Returns `findings` ranked critical first, then longest-standing first: failing monitors (with the error and since when), firing, pending and - still-breaching acknowledged alerts, TLS certificates expiring within - 14 days, host agents that stopped reporting, hosts over the CPU, memory or - disk thresholds set under Settings → Thresholds, Ansible runs that failed + still-breaching acknowledged alerts, TLS certificates near expiry, host + agents that stopped reporting, hosts over their CPU, memory or disk + limits (all cutoffs are set under Settings → Thresholds), Ansible runs that failed in the last 24h, and anything enabled plugins add (unhealthy containers, for one). Each finding carries the ids to follow and `next`, the call that shows its evidence. @@ -180,9 +183,10 @@ async def get_overview() -> dict: )).scalars()) windows = await alert_svc.list_maintenance_windows(db) - findings = (monitor_findings(monitors, states, host_names, now) + thresholds = app.config.get("THRESHOLDS") or {} + findings = (monitor_findings(monitors, states, host_names, now, thresholds.get("tls_days")) + alert_findings(active) - + agent_findings(hosts, agent, app.config.get("THRESHOLDS") or {}) + + agent_findings(hosts, agent, thresholds) + run_findings(runs)) plugin_found, unavailable = await _plugin_findings(getattr(app, "mcp_server", None)) findings = rank(findings + plugin_found) diff --git a/steward/settings/routes.py b/steward/settings/routes.py index 03b1c24..de1de43 100644 --- a/steward/settings/routes.py +++ b/steward/settings/routes.py @@ -126,6 +126,8 @@ _THRESHOLD_FIELDS = [ ("temp_warn", "thresholds.temp_warn", False), ("temp_crit", "thresholds.temp_crit", False), ("uptime_warn", "thresholds.uptime_warn", True), ("uptime_crit", "thresholds.uptime_crit", True), ("latency_warn", "ping.threshold.good_ms", False), ("latency_crit", "ping.threshold.warn_ms", False), + ("tls_warn", "thresholds.tls_warn_days", False), ("tls_crit", "thresholds.tls_crit_days", False), + ("crash_loop_deaths", "thresholds.crash_loop_deaths", False), ] # Docker time-series retention windows (days). Read fresh by the cleanup task, diff --git a/steward/templates/settings/thresholds.html b/steward/templates/settings/thresholds.html index 000b6f2..1e5ff12 100644 --- a/steward/templates/settings/thresholds.html +++ b/steward/templates/settings/thresholds.html @@ -40,6 +40,18 @@ {{ pair("Temperature", "temp_warn", "thresholds.temp_warn", "temp_crit", "thresholds.temp_crit", "°C", "Hottest reported sensor per host.") }} {{ pair("Ping latency", "latency_warn", "ping.threshold.good_ms", "latency_crit", "ping.threshold.warn_ms", "ms", "Round-trip time. Shared with the Ping widget's good/warn colouring.") }} {{ pair("Uptime / SLA", "uptime_warn", "thresholds.uptime_warn", "uptime_crit", "thresholds.uptime_crit", "%", "Lower is worse — amber below warn, red below crit.", step="0.1") }} + {{ pair("TLS certificate expiry", "tls_warn", "thresholds.tls_warn_days", "tls_crit", "thresholds.tls_crit_days", "days left", "HTTPS monitors whose certificate expires within these many days are flagged. Lower is worse.") }} + +