diff --git a/plugins/docker/ingest.py b/plugins/docker/ingest.py index 5262280..15ef48a 100644 --- a/plugins/docker/ingest.py +++ b/plugins/docker/ingest.py @@ -169,6 +169,7 @@ async def _persist_swarm(session, host, swarm: dict) -> None: row.running = int(s.get("running") or 0) row.image = s.get("image") row.placement_json = json.dumps(s.get("placement") or []) + row.tasks_json = json.dumps(s.get("tasks") or []) row.updated_at = now stale_services = delete(DockerSwarmService).where( DockerSwarmService.host_id == host.id) diff --git a/plugins/docker/mcp_tools.py b/plugins/docker/mcp_tools.py index c20dfe6..b8e4a3c 100644 --- a/plugins/docker/mcp_tools.py +++ b/plugins/docker/mcp_tools.py @@ -26,10 +26,10 @@ 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 +from .swarm_view import build_swarm_services, replaced_tasks, task_time PROBLEMS = ("restarting", "crash_looping", "unhealthy", "oom_killed", "exited_nonzero") -CRASH_LOOP_DEATHS = 3 # default; Settings → Thresholds sets it +CRASH_LOOP_DEATHS = docker_svc.CRASH_LOOP_DEATHS LOG_LINE_CHARS = 2000 @@ -61,7 +61,8 @@ def cut_middle(text: str, limit: int = LOG_LINE_CHARS) -> str: 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: +def container_row(c, host_names: dict, problems: list[str], now: datetime, + replaced: bool = False) -> dict: running = (c.status or "").lower() == "running" return { "container_name": c.name, @@ -82,6 +83,8 @@ def container_row(c, host_names: dict, problems: list[str], now: datetime) -> di "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, + # A swarm task Swarm has replaced: history, judged on its service. + "swarm_replaced": replaced if c.service_name else None, "reported_at": iso(c.scraped_at), "reported_age_s": age_s(c.scraped_at, now), } @@ -110,15 +113,41 @@ def container_findings(rows: list[dict]) -> list[dict]: return out +def task_error_text(err: dict | None) -> str: + """Swarm's own account of a task's end, as one clause.""" + if not err: + return "" + why = err.get("error") or err.get("message") or err.get("state") or "unknown" + code = f", exit {err['exit_code']}" if err.get("exit_code") not in (None, 0) else "" + where = f" on {err['node']}" if err.get("node") else "" + return f"{why}{code}{where}" + + def service_findings(services: list[dict]) -> list[dict]: + """Swarm services that are failing: crash-looping, under-replicated, or that + lost a task to an error in the last 24h. Each names Swarm's last task error.""" 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()")) + replicas = f"{s['running']} of {s['desired']} replicas running" + last = task_error_text(s.get("last_error")) + tail = f"; last task error: {last}" if last else "" + ids = {"service_name": s["name"]} + if s.get("crash_looping"): + out.append(finding( + "critical", "swarm_service_crash_looping", s["name"], + f"swarm service {s['name']} keeps failing: {s['failures_1h']} tasks failed in the " + f"last hour, {replicas}{tail}", ids=ids, next="list_swarm_services()")) + elif s["state"] != "healthy": + out.append(finding( + "critical" if s["state"] == "down" else "warning", f"swarm_service_{s['state']}", + s["name"], f"swarm service {s['name']} has {replicas}{tail}", + ids=ids, next="list_swarm_services()")) + elif s.get("failures_24h"): + n = s["failures_24h"] + out.append(finding( + "warning", "swarm_task_failed", s["name"], + f"swarm service {s['name']} had {n} failed task{'s' if n != 1 else ''} in the last " + f"24h and is now {replicas}{tail}", ids=ids, next="list_swarm_services()")) return out @@ -129,38 +158,49 @@ async def _host_names(db) -> dict[str, str]: 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) + return docker_svc.crash_loop_deaths(mcp_app().config) + + +async def _replaced(db, containers) -> set[tuple]: + services = list((await db.execute(select(DockerSwarmService))).scalars()) + return replaced_tasks(containers, services) async def _rows(db, now: datetime) -> list[dict]: + """Every container with its problems. A swarm task Swarm replaced has none: + its failure, if it was one, belongs to its service (service_findings).""" containers = dedup_by_container_id(await docker_svc.list_containers(db)) + replaced = await _replaced(db, containers) 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: + gone = (c.host_id, c.name) in replaced 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)) + problems = [] if gone else container_problems(c, recent, bool(events), deaths) + rows.append(container_row(c, host_names, problems, now, replaced=gone)) return rows -async def _swarm(db): +async def _swarm(db, now: datetime): 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) + built = build_swarm_services(containers, services, node_hostname, host_names, + now=now, crash_loop_deaths=_crash_loop_deaths()) 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) + now = now_utc() + rows = await _rows(db, now) + services, _, _ = await _swarm(db, now) return container_findings(rows) + service_findings(services) @@ -184,6 +224,11 @@ async def list_containers(host: str = "", status: str = "", problem: str = "", 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. + + A swarm task's container that Swarm has replaced (a rolling update, or a + retry after a failure) has `swarm_replaced: true` and no problems: Swarm + restarts a task as a new container, so failures are counted on the + service — list_swarm_services shows them. """ if problem and problem != "any" and problem not in PROBLEMS: raise ValueError(f"problem {problem!r} is not 'any' or one of {', '.join(PROBLEMS)}.") @@ -254,6 +299,7 @@ async def get_container(container: str, host: str = "", range: str = "24h", even now = now_utc() async with db_session() as db: c = await _pick_container(db, container, host) + gone = (c.host_id, c.name) in await _replaced(db, await docker_svc.list_containers(db)) 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) @@ -265,9 +311,9 @@ async def get_container(container: str, host: str = "", range: str = "24h", even ports = json.loads(c.ports_json or "[]") except ValueError: ports = [] + problems = [] if gone else container_problems(c, recent, bool(failures), _crash_loop_deaths()) return { - **container_row(c, host_names, - container_problems(c, recent, bool(failures), _crash_loop_deaths()), now), + **container_row(c, host_names, problems, now, replaced=gone), "container_id": c.container_id, "ports": ports, "net_rx_bytes": c.net_rx_bytes, "net_tx_bytes": c.net_tx_bytes, @@ -285,15 +331,32 @@ 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. + "down" (none). `replicas` are the service's current tasks: each names its + host and container status, and one marked `ghost` runs on a node with no + Steward agent, so Steward knows only how many tasks are placed there. + Tasks Swarm replaced (rolling updates, retries) are left out and counted + in `previous_tasks`. + + From Swarm's own task records (a manager's agent 1.8.0 or later; + `tasks_known: false` and nulls otherwise): + - `failures_1h` / `failures_24h`: tasks that failed or were rejected. + Swarm retries a failed task as a new one, so this, not any container's + restart count, is how a crash loop shows. Swarm keeps about 5 old tasks + per replica, which bounds the count. + - `crash_looping`: failures_1h reached the crash-loop count set under + Settings → Thresholds (default 3). + - `last_error`: the newest task that failed, was rejected, or carries an + error while Swarm retries it — `error` is Swarm's own text (e.g. "No + such image", "task: non-zero exit (1)"), with `exit_code`, `node`, `at`. + + Services with a problem come first. `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) + now = now_utc() async with db_session() as db: - services, nodes, host_names = await _swarm(db) + services, nodes, host_names = await _swarm(db, now) seen, node_rows = set(), [] for n in sorted(nodes, key=lambda n: n.updated_at, reverse=True): if n.node_id in seen: @@ -304,10 +367,26 @@ async def list_swarm_services(limit: int = 100) -> dict: "reported_by": host_names.get(n.host_id)}) node_rows.sort(key=lambda n: n["hostname"].lower()) order = {"down": 0, "degraded": 1, "healthy": 2} + + def _rank(s): + return (not s["crash_looping"], order.get(s["state"], 3), not s["failures_24h"], s["name"]) + + def _error_row(err): + if err is None: + return None + at = task_time(err) + return {"error": err["error"], "message": err["message"], "state": err["state"], + "exit_code": err["exit_code"], "node": err["node"], + "at": iso(at), "age_s": age_s(at, now) if at else None} + rows = [ { "service_name": s["name"], "mode": s["mode"], "state": s["state"], "running": s["running"], "desired": s["desired"], + "crash_looping": s["crash_looping"], + "failures_1h": s["failures_1h"], "failures_24h": s["failures_24h"], + "last_error": _error_row(s["last_error"]), + "previous_tasks": s["previous_tasks"], "tasks_known": s["tasks_known"], "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"], @@ -316,7 +395,7 @@ async def list_swarm_services(limit: int = 100) -> dict: for r in s["replicas"] ], } - for s in sorted(services, key=lambda s: (order.get(s["state"], 3), s["name"])) + for s in sorted(services, key=_rank) ] return {"nodes": node_rows, **bounded(rows, limit, key="services")} diff --git a/plugins/docker/migrations/versions/docker_010_swarm_tasks.py b/plugins/docker/migrations/versions/docker_010_swarm_tasks.py new file mode 100644 index 0000000..bf0e4ad --- /dev/null +++ b/plugins/docker/migrations/versions/docker_010_swarm_tasks.py @@ -0,0 +1,30 @@ +"""Docker swarm service tasks column + +Adds docker_swarm_services.tasks_json — each service's newest tasks as a +manager's agent (≥ 1.8.0) reports them: slot, node, desired and current state, +Swarm's error and message, exit code and time. Lets Steward tell a task Swarm +replaced from one that crashed, count a service's failures across task +retries, and say why its last task failed. Additive column; no DROP+recreate. + +Revision ID: docker_010_swarm_tasks +Revises: docker_009_container_logs +Create Date: 2026-10-06 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "docker_010_swarm_tasks" +down_revision: Union[str, None] = "docker_009_container_logs" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column("docker_swarm_services", + sa.Column("tasks_json", sa.Text, nullable=False, + server_default="[]")) + + +def downgrade() -> None: + op.drop_column("docker_swarm_services", "tasks_json") diff --git a/plugins/docker/models.py b/plugins/docker/models.py index 7082d20..257ea6e 100644 --- a/plugins/docker/models.py +++ b/plugins/docker/models.py @@ -223,6 +223,11 @@ class DockerSwarmService(Base): # A manager sees every task, so this captures cross-node placement that the # local docker_containers rows (this host only) can't. placement_json: Mapped[str] = mapped_column(Text, nullable=False, default="[]") + # The service's newest tasks, newest first, as the manager's agent shaped + # them: [{"task_id", "slot", "node_id", "desired_state", "state", "message", + # "error", "exit_code", "at"}], JSON. Swarm's own record of which tasks it + # replaced and why each one stopped; "[]" from agents older than 1.8.0. + tasks_json: Mapped[str] = mapped_column(Text, nullable=False, default="[]") updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc), diff --git a/plugins/docker/routes.py b/plugins/docker/routes.py index d930f63..226b31f 100644 --- a/plugins/docker/routes.py +++ b/plugins/docker/routes.py @@ -15,7 +15,7 @@ from steward.models.hosts import Host from steward.core.time_range import parse_range, DEFAULT_RANGE from . import service as docker_svc from .dedup import dedup_by_container_id -from .swarm_view import build_swarm_services +from .swarm_view import build_swarm_services, parse_tasks, task_summary from .models import ( DockerContainer, DockerDiskUsage, DockerEvent, DockerImage, DockerSwarmNode, DockerSwarmService, @@ -128,7 +128,9 @@ async def rows(): # from the managers' placement). Non-swarm containers keep the host view. node_hostname = {n.node_id: n.hostname for n in nodes if n.hostname} host_name = {h.id: h.name for h in hosts.values()} - swarm_view = build_swarm_services(containers, services, node_hostname, host_name) + swarm_view = build_swarm_services( + containers, services, node_hostname, host_name, + crash_loop_deaths=docker_svc.crash_loop_deaths(current_app.config)) standalone = [c for c in containers if not c.service_name] groups: dict[str, dict] = {} @@ -204,13 +206,13 @@ async def widget(): )).scalars())) services = list((await db.execute(select(DockerSwarmService))).scalars()) nodes = list((await db.execute(select(DockerSwarmNode))).scalars()) - # Recent failures are more actionable than a static "stopped" tally. Count + # Recent failures are more actionable than a static "stopped" tally. Read # DISTINCT container names so the two managers' duplicate die/oom events - # don't double it. - failed_24h = (await db.execute( - select(func.count(func.distinct(DockerEvent.container_name))) + # don't double it; swarm tasks are counted on their service below. + dead_names = list((await db.execute( + select(func.distinct(DockerEvent.container_name)) .where(DockerEvent.event.in_(("die", "oom")), DockerEvent.at >= since_24h) - )).scalar() or 0 + )).scalars()) hosts = await docker_svc.host_map( db, {c.host_id for c in all_containers} @@ -221,7 +223,10 @@ async def widget(): # host); non-swarm containers stay grouped by host. node_hostname = {n.node_id: n.hostname for n in nodes if n.hostname} host_name = {h.id: h.name for h in hosts.values()} - swarm_services = build_swarm_services(all_containers, services, node_hostname, host_name)["services"] + swarm_services = build_swarm_services( + all_containers, services, node_hostname, host_name, + crash_loop_deaths=docker_svc.crash_loop_deaths(current_app.config))["services"] + failed_24h = docker_svc.failed_24h(dead_names, swarm_services) standalone = [c for c in all_containers if not c.service_name] running = [c for c in standalone if c.status == "running"] @@ -413,6 +418,8 @@ async def swarm(): node_by_id[n.node_id] = n node_names = {nid: (nrow.hostname or nid) for nid, nrow in node_by_id.items()} + now = datetime.now(timezone.utc) + deaths = docker_svc.crash_loop_deaths(current_app.config) svc_dicts = [] for s in sorted(svc_by_name.values(), key=lambda s: s.service_name): placement = [] @@ -428,7 +435,10 @@ async def swarm(): "healthy": s.running >= s.desired and s.desired > 0, "degraded": 0 < s.running < s.desired, "down": s.running == 0 and s.desired > 0, + "state": ("healthy" if s.running >= s.desired and s.desired > 0 + else "degraded" if s.running > 0 else "down"), "placement": placement, + **task_summary(parse_tasks(s), now, deaths, node_names), }) node_rows = sorted( node_by_id.values(), key=lambda n: (n.role, (n.hostname or n.node_id).lower()) diff --git a/plugins/docker/service.py b/plugins/docker/service.py index aab5729..1cd985b 100644 --- a/plugins/docker/service.py +++ b/plugins/docker/service.py @@ -14,6 +14,37 @@ from steward.models.hosts import Host from .models import DockerContainer, DockerEvent, DockerLog, DockerMetric, DockerMetricHourly +CRASH_LOOP_DEATHS = 3 # default; Settings → Thresholds sets it + + +def crash_loop_deaths(config) -> int: + """Deaths within an hour that make a container or a swarm service a crash + loop, from Settings → Thresholds (app.config["THRESHOLDS"]).""" + th = (config.get("THRESHOLDS") or {}).get("crash_loop") or {} + return int(th.get("warn") or CRASH_LOOP_DEATHS) + + +def failed_24h(dead_names, swarm_services) -> int: + """Failures in the last 24h for the dashboard: distinct plain containers + that died or were OOM-killed, plus each swarm service's failed tasks. + + A swarm task's container dies whenever Swarm replaces it (a rolling update + kills it with 137), so its die event is not a failure. Services with + Swarm's task records count their failed tasks instead; a service without + them (an agent older than 1.8.0) falls back to its tasks' die events. + """ + by_prefix = {s["name"] + ".": s for s in swarm_services} + plain, fallback = 0, set() + for name in dead_names: + svc = next((s for p, s in by_prefix.items() if name.startswith(p)), None) + if svc is None: + plain += 1 + elif not svc["tasks_known"]: + fallback.add(name) + tasks = sum(s["failures_24h"] or 0 for s in swarm_services if s["tasks_known"]) + return plain + len(fallback) + tasks + + def bucket(values: list[float], target: int = 40) -> list[float]: """Bucket-average a series down to ~target points (DB-agnostic, in Python). diff --git a/plugins/docker/swarm_view.py b/plugins/docker/swarm_view.py index cfffa80..f2d779d 100644 --- a/plugins/docker/swarm_view.py +++ b/plugins/docker/swarm_view.py @@ -11,12 +11,28 @@ For every swarm service we list the real replicas we have detail for, then add "ghost" replicas for the remaining placement count on each node (agent-less nodes). Non-swarm containers pass through grouped by host, unchanged. +A third source, DockerSwarmService.tasks_json (agent ≥ 1.8.0), is Swarm's own +record of each service's newest tasks. It decides which task containers are +*superseded* — replaced by Swarm, so neither a replica nor a crash — and gives +each service its failures across task retries and the error of its last +failed task. Swarm restarts a task by creating a new one (a new container), so +failures can only be counted per service, never per container. + Kept model-free (no ORM imports) so it's unit-testable and safe to import in tests without the plugin-loader "table already defined" gotcha. """ from __future__ import annotations import json +from datetime import datetime, timedelta, timezone + +# Task states that mean the task ended in error: `failed` (its container exited +# non-zero or never started) and `rejected` (Swarm could not start it at all, +# e.g. no such image, a bad mount). +FAILED_TASK_STATES = ("failed", "rejected") +# Desired states of a task Swarm still means to run. Anything else (shutdown, +# remove, complete, orphaned) is a task Swarm has finished with. +_CURRENT_DESIRED = ("running", "ready", "") def _service_state(running: int, desired: int) -> str: @@ -36,30 +52,179 @@ def _placement(service) -> list[dict]: return [p for p in data if isinstance(p, dict)] -def build_swarm_services(containers, services, node_hostname: dict, host_name: dict) -> dict: - """Return {"services": [...], "standalone": [...]}. +def parse_tasks(service) -> list[dict]: + """A service row's task records, newest first; [] when its agent sent none.""" + raw = getattr(service, "tasks_json", None) or "[]" + try: + data = json.loads(raw) + except (ValueError, TypeError): + return [] + return [t for t in data if isinstance(t, dict)] - services[i] = {name, mode, desired, running, state, replicas:[...]} - real replica = {ghost:False, host, host_id, name, status, cpu_pct, mem_pct, - health, restart_count} - ghost replica = {ghost:True, host, count} # placement with no local row - standalone[i] = {host_id, host, containers:[...]} # non-swarm, by host + +def task_time(task: dict) -> datetime | None: + """A task's status timestamp. Docker writes nanoseconds; Python ≥ 3.11 keeps + the first six digits.""" + raw = task.get("at") or "" + try: + at = datetime.fromisoformat(raw.replace("Z", "+00:00")) + except ValueError: + return None + return at if at.tzinfo else at.replace(tzinfo=timezone.utc) + + +def task_failures(tasks: list[dict], since: datetime) -> int: + """How many tasks ended in error at or after `since`.""" + n = 0 + for t in tasks: + at = task_time(t) + if t.get("state") in FAILED_TASK_STATES and at is not None and at >= since: + n += 1 + return n + + +def last_task_error(tasks: list[dict]) -> dict | None: + """Why the service's newest troubled task is in trouble, from Swarm itself. + + The newest task that failed or was rejected, or that Swarm attached an error + to while still trying (a pending task with "no suitable node", say). None + when no retained task has a problem. """ + for t in tasks: # newest first + if t.get("state") in FAILED_TASK_STATES or t.get("error"): + return {"state": t.get("state"), "error": t.get("error") or None, + "message": t.get("message") or None, "exit_code": t.get("exit_code"), + "node_id": t.get("node_id") or None, "at": t.get("at") or None} + return None + + +def task_summary(tasks: list[dict], now: datetime, crash_loop_deaths: int, + node_hostname: dict) -> dict: + """What a service's task records say: tasks_known, failures_1h, + failures_24h (None without records), crash_looping, and last_error with + the node's hostname as `node`.""" + known = bool(tasks) + failures_1h = task_failures(tasks, now - timedelta(hours=1)) if known else None + last_error = last_task_error(tasks) + if last_error is not None: + nid = last_error["node_id"] + last_error["node"] = node_hostname.get(nid) or (nid[:12] if nid else None) + at = task_time(last_error) + last_error["age_s"] = max(0, int((now - at).total_seconds())) if at else None + return { + "tasks_known": known, + "failures_1h": failures_1h, + "failures_24h": task_failures(tasks, now - timedelta(hours=24)) if known else None, + "crash_looping": bool(known and failures_1h >= crash_loop_deaths), + "last_error": last_error, + } + + +def _slot_key(c) -> str: + """A task container's slot within its service: ".." for + replicated services, ".." for global ones.""" + svc, name = c.service_name, c.name or "" + rest = name[len(svc) + 1:] if name.startswith(svc + ".") else name + return rest.split(".", 1)[0] + + +def superseded_containers(swarm_cs, tasks_by_id: dict[str, dict]) -> set[tuple]: + """(host_id, name) of task containers Swarm has replaced. + + With the task on record, its desired state decides: Swarm sets a replaced + or failed task's desired state to shutdown. Without it (an older agent, or a + task older than the retained history), a container that is not running is + superseded when another task in the same slot is running or started later. + """ + out: set[tuple] = set() + by_slot: dict[tuple, list] = {} + for c in swarm_cs: + by_slot.setdefault((c.service_name, _slot_key(c)), []).append(c) + for c in swarm_cs: + task = tasks_by_id.get(getattr(c, "task_id", None) or "") + if task is not None: + if task.get("desired_state") not in _CURRENT_DESIRED: + out.add((c.host_id, c.name)) + continue + if (getattr(c, "status", "") or "").lower() == "running": + continue + mine = getattr(c, "started_at", None) + for other in by_slot[(c.service_name, _slot_key(c))]: + if other is c: + continue + theirs = getattr(other, "started_at", None) + if ((getattr(other, "status", "") or "").lower() == "running" + or (theirs is not None and (mine is None or theirs > mine))): + out.add((c.host_id, c.name)) + break + return out + + +def _newest_per_name(services) -> dict[str, object]: + """Every manager reports the same cluster-global service set; keep the most + recently updated row per name, so a manager on an older agent (no tasks) + doesn't shadow one that reports them.""" + by_name: dict[str, object] = {} + for s in services: + name = getattr(s, "service_name", None) + if not name: + continue + kept = by_name.get(name) + if kept is None: + by_name[name] = s + continue + a, b = getattr(s, "updated_at", None), getattr(kept, "updated_at", None) + if a is not None and (b is None or a > b): + by_name[name] = s + return by_name + + +def replaced_tasks(containers, services) -> set[tuple]: + """(host_id, name) of the task containers Swarm has replaced, from the + container rows and the managers' service rows. What the container views use + to leave history out of their problem lists.""" + swarm_cs = [c for c in containers if getattr(c, "service_name", None)] + tasks_by_id = {t["task_id"]: t for s in _newest_per_name(services).values() + for t in parse_tasks(s) if t.get("task_id")} + return superseded_containers(swarm_cs, tasks_by_id) + + +def build_swarm_services(containers, services, node_hostname: dict, host_name: dict, + *, now: datetime | None = None, crash_loop_deaths: int = 3) -> dict: + """Return {"services": [...], "standalone": [...], "superseded": {(host_id, name)}}. + + services[i] = {name, mode, desired, running, state, replicas:[...], + previous_tasks, tasks_known, failures_1h, failures_24h, + crash_looping, last_error} + real replica = {ghost:False, host, host_id, name, status, cpu_pct, mem_pct, + health, restart_count} # current tasks only + ghost replica = {ghost:True, host, count} # placement with no local row + previous_tasks = task containers Swarm replaced, left out of replicas + failures_* = tasks that failed or were rejected in the window (None + without task data); crash_looping = failures_1h ≥ the + Settings → Thresholds crash-loop count + last_error = last_task_error(), plus `node` (the node's hostname) + standalone[i] = {host_id, host, containers:[...]} # non-swarm, by host + superseded = the task containers Swarm replaced, so callers can skip them + """ + now = now or datetime.now(timezone.utc) containers = list(containers) swarm_cs = [c for c in containers if getattr(c, "service_name", None)] standalone_cs = [c for c in containers if not getattr(c, "service_name", None)] - # Dedup service rows by name (every manager reports the same cluster-global - # set); keep the first — placement is identical across managers. - svc_by_name: dict[str, object] = {} - for s in services: - name = getattr(s, "service_name", None) - if name and name not in svc_by_name: - svc_by_name[name] = s + svc_by_name = _newest_per_name(services) + tasks_by_svc = {name: parse_tasks(s) for name, s in svc_by_name.items()} + tasks_by_id = {t["task_id"]: t for ts in tasks_by_svc.values() for t in ts + if t.get("task_id")} + superseded = superseded_containers(swarm_cs, tasks_by_id) - # Real containers grouped by service name. + # Current task containers grouped by service name; replaced ones are counted. cs_by_service: dict[str, list] = {} + previous: dict[str, int] = {} for c in swarm_cs: + if (c.host_id, c.name) in superseded: + previous[c.service_name] = previous.get(c.service_name, 0) + 1 + continue cs_by_service.setdefault(c.service_name, []).append(c) def _host_of(c): @@ -124,6 +289,8 @@ def build_swarm_services(containers, services, node_hostname: dict, host_name: d "desired": desired, "state": _service_state(int(running or 0), int(desired or 0)), "replicas": replicas, + "previous_tasks": previous.get(name, 0), + **task_summary(tasks_by_svc.get(name, []), now, crash_loop_deaths, node_hostname), }) # Non-swarm containers grouped by host, running-first order preserved. @@ -137,4 +304,4 @@ def build_swarm_services(containers, services, node_hostname: dict, host_name: d g["containers"].append(c) standalone = sorted(groups.values(), key=lambda g: g["host"].lower()) - return {"services": out_services, "standalone": standalone} + return {"services": out_services, "standalone": standalone, "superseded": superseded} diff --git a/plugins/docker/templates/docker/_swarm_failures.html b/plugins/docker/templates/docker/_swarm_failures.html new file mode 100644 index 0000000..97cece0 --- /dev/null +++ b/plugins/docker/templates/docker/_swarm_failures.html @@ -0,0 +1,31 @@ +{#- One line on a swarm service's task failures, from Swarm's own task records + (agent ≥ 1.8.0): a crash loop or the failures of the last 24h, and how the + newest troubled task ended, in Swarm's words. Expects `s` with state, + crash_looping, failures_1h, failures_24h and last_error; `style` is + optional extra CSS for the wrapper. Renders nothing for a healthy service + with no recent failures. -#} +{%- macro ago(sec) -%} + {%- if sec < 90 -%}{{ sec }}s ago + {%- elif sec < 5400 -%}{{ sec // 60 }}m ago + {%- elif sec < 172800 -%}{{ sec // 3600 }}h ago + {%- else -%}{{ sec // 86400 }}d ago{%- endif -%} +{%- endmacro -%} +{%- set e = s.last_error -%} +{%- if s.crash_looping or s.failures_24h or (e and s.state != 'healthy') %} +
+ {% if s.crash_looping %} + Crash-looping · {{ s.failures_1h }} tasks failed in the last hour + {% elif s.failures_24h %} + {{ s.failures_24h }} task{{ 's' if s.failures_24h != 1 }} failed in the last 24h + {% endif %} + {% if e %} + + Last task {{ e.state }}: + {{ e.error or e.message or e.state }} + {%- if e.exit_code %} · exit {{ e.exit_code }}{% endif %} + {%- if e.node %} · {{ e.node }}{% endif %} + {%- if e.age_s is not none %} · {{ ago(e.age_s) }}{% endif %} + + {% endif %} +
+{%- endif %} diff --git a/plugins/docker/templates/docker/rows.html b/plugins/docker/templates/docker/rows.html index 1d04717..8748207 100644 --- a/plugins/docker/templates/docker/rows.html +++ b/plugins/docker/templates/docker/rows.html @@ -26,9 +26,10 @@ -{# ── Swarm services: collapsed per-service, each replica labelled with its host. - Ghost replicas (tasks on nodes with no Steward agent) come from the managers' - placement data so the picture is cluster-complete. ──────────────────────── #} +{# ── Swarm services: collapsed per-service, each current replica labelled with + its host (tasks Swarm replaced are left out). Ghost replicas (tasks on nodes + with no Steward agent) come from the managers' placement data so the picture + is cluster-complete. A failing service names Swarm's last task error. ───── #} {% if swarm_services %}
@@ -46,6 +47,7 @@ {{ s.running }}/{{ s.desired }} running
+ {% with style="padding-left:1.1rem;margin-bottom:0.35rem;" %}{% include "docker/_swarm_failures.html" %}{% endwith %}
{% for r in s.replicas %} {% if r.ghost %} diff --git a/plugins/docker/templates/docker/swarm.html b/plugins/docker/templates/docker/swarm.html index 7fdbf44..b26704c 100644 --- a/plugins/docker/templates/docker/swarm.html +++ b/plugins/docker/templates/docker/swarm.html @@ -22,7 +22,7 @@ - + @@ -39,9 +39,14 @@ + {% else %} - + {% endfor %}
ServiceModeReplicasImagePlacementServiceModeReplicasImagePlacementTask failures
{% for p in s.placement %}{{ p.node }} ({{ p.running }}){% if not loop.last %}, {% endif %}{% else %}—{% endfor %} + {% include "docker/_swarm_failures.html" %} + {%- if not s.tasks_known %}— + {%- elif not (s.crash_looping or s.failures_24h or (s.last_error and s.state != 'healthy')) %}none in 24h{% endif %} +
No services in this swarm.
No services in this swarm.
diff --git a/plugins/docker/templates/docker/widget.html b/plugins/docker/templates/docker/widget.html index ad86814..9a80a8b 100644 --- a/plugins/docker/templates/docker/widget.html +++ b/plugins/docker/templates/docker/widget.html @@ -8,7 +8,7 @@ {{ running_count }} running
-
+
{{ failed_24h }} failed (24h)
@@ -24,6 +24,7 @@
{{ s.name }} + {% if s.crash_looping %}crash-looping{% endif %} {{ s.running }}/{{ s.desired }}
diff --git a/plugins/host_agent/agent.py b/plugins/host_agent/agent.py index 653f073..9659e29 100644 --- a/plugins/host_agent/agent.py +++ b/plugins/host_agent/agent.py @@ -20,7 +20,7 @@ from collections import deque from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone -AGENT_VERSION = "1.7.0" +AGENT_VERSION = "1.8.0" # Default path to the local Docker Engine socket. Overridable via the # `docker_socket` config key; collection is silently skipped if it's absent or @@ -874,14 +874,46 @@ def _swarm_is_manager(info: dict) -> bool: return bool((info or {}).get("Swarm", {}).get("ControlAvailable")) +# Newest tasks kept per service. Swarm retains about 5 per slot by default, so +# this covers a replicated service's recent history and a crash loop's deaths +# without shipping a global service's whole fleet every sample. +SWARM_TASKS_PER_SERVICE = 10 +_TASK_TEXT_CHARS = 300 + + +def _swarm_task(t: dict) -> dict: + """One task as Steward stores it: where it ran, what Swarm wants of it, + what it is doing, and why it stopped (Err/Message, exit code).""" + status = t.get("Status") or {} + cstatus = status.get("ContainerStatus") or {} + exit_code = cstatus.get("ExitCode") + return { + "task_id": t.get("ID", "") or "", + "slot": t.get("Slot"), + "node_id": t.get("NodeID", "") or "", + "desired_state": (t.get("DesiredState") or "").lower(), + "state": (status.get("State") or "").lower(), + "message": (status.get("Message") or "")[:_TASK_TEXT_CHARS], + "error": (status.get("Err") or "")[:_TASK_TEXT_CHARS], + "exit_code": exit_code if isinstance(exit_code, int) else None, + "at": status.get("Timestamp") or t.get("UpdatedAt") or "", + } + + def _swarm_services(services: list, tasks: list) -> list: """Roll desired-vs-running replicas per service up from the task list. Replica health isn't on the service object — it's derived by counting tasks: `running` = tasks currently in state running; `desired` = the service's configured replica count (replicated) or the count of tasks the scheduler - wants running (global, which has no fixed number). + wants running (global, which has no fixed number). Each service also + carries its newest tasks (`tasks`), so Steward can tell a task Swarm + replaced from one that crashed, and say why the last one failed. """ + tasks_by_svc: dict = {} + for t in (tasks or []): + if t.get("ServiceID"): + tasks_by_svc.setdefault(t["ServiceID"], []).append(_swarm_task(t)) running_by_svc: dict = {} desired_by_svc: dict = {} # only used for global-mode services placement: dict = {} # svc_id -> {node_id: running_count} @@ -921,6 +953,8 @@ def _swarm_services(services: list, tasks: list) -> list: # sees every task, even ones whose containers live on other hosts). "placement": [{"node_id": n, "running": cnt} for n, cnt in sorted(placement.get(sid, {}).items())], + "tasks": sorted(tasks_by_svc.get(sid, []), key=lambda t: t["at"], + reverse=True)[:SWARM_TASKS_PER_SERVICE], }) return out diff --git a/tests/integration/test_docker.py b/tests/integration/test_docker.py index 48b7479..6be8047 100644 --- a/tests/integration/test_docker.py +++ b/tests/integration/test_docker.py @@ -7,6 +7,7 @@ containers on different hosts no longer collide. Requires STEWARD_DATABASE_URL. from __future__ import annotations import asyncio +import json import os import uuid from datetime import datetime, timezone @@ -391,7 +392,11 @@ def test_swarm_topology_persisted(app): swarm = { "services": [{"service_name": "web", "mode": "replicated", "desired": 3, "running": 2, "image": "nginx", - "placement": [{"node_id": "n1", "running": 2}]}], + "placement": [{"node_id": "n1", "running": 2}], + "tasks": [{"task_id": "t1", "slot": 3, "node_id": "n1", + "desired_state": "shutdown", "state": "failed", + "message": "started", "error": "task: non-zero exit (1)", + "exit_code": 1, "at": "2026-10-06T13:19:51Z"}]}], "nodes": [{"node_id": "n1", "hostname": "mgr", "role": "manager", "availability": "active", "status": "ready", "leader": True}], } @@ -408,7 +413,7 @@ def test_swarm_topology_persisted(app): await persist(s, h, [], swarm) hid = h.id svc = (await s.execute(text( - "SELECT mode, desired, running, image, placement_json " + "SELECT mode, desired, running, image, placement_json, tasks_json " "FROM docker_swarm_services WHERE host_id = :h"), {"h": hid})).first() node = (await s.execute(text( "SELECT role, availability, status, leader " @@ -418,6 +423,8 @@ def test_swarm_topology_persisted(app): svc, node = asyncio.run(_go()) assert svc[0] == "replicated" and svc[1] == 3 and svc[2] == 2 and svc[3] == "nginx" assert "n1" in svc[4] # placement JSON carries the node id + (task,) = json.loads(svc[5]) # docker_010: the manager's task records survive + assert task["error"] == "task: non-zero exit (1)" and task["slot"] == 3 assert node[0] == "manager" and node[2] == "ready" and node[3] is True diff --git a/tests/integration/test_mcp_plugin_tools.py b/tests/integration/test_mcp_plugin_tools.py index 6e7affb..5f561c0 100644 --- a/tests/integration/test_mcp_plugin_tools.py +++ b/tests/integration/test_mcp_plugin_tools.py @@ -9,6 +9,7 @@ from __future__ import annotations import asyncio import importlib +import json import os import uuid from datetime import datetime, timedelta, timezone @@ -57,14 +58,18 @@ def test_docker_and_agent_tools_over_seeded_rows(app): hid, hid2 = str(uuid.uuid4()), str(uuid.uuid4()) host_name, quiet_host = f"dock-{tag}", f"quiet-{tag}" web, db_, job, svc = f"web-{tag}", f"db-{tag}", f"job-{tag}", f"svc-{tag}" + replaced = f"{svc}.1.old" + + def _ago(minutes): + return (now - timedelta(minutes=minutes)).isoformat().replace("+00:00", "Z") container_sql = text(""" INSERT INTO docker_containers (host_id, name, container_id, image, status, cpu_pct, mem_pct, mem_usage_bytes, mem_limit_bytes, restart_count, ports_json, started_at, scraped_at, health, - exit_code, oom_killed, compose_project, service_name) + exit_code, oom_killed, compose_project, service_name, task_id) VALUES (:h, :n, :cid, 'img:1', :st, 5.0, 10.0, 1000, 10000, :rc, '[]', :started, :now, - :health, :exit, false, :proj, :svc) + :health, :exit, false, :proj, :svc, :task) """) async def _seed(): @@ -73,20 +78,23 @@ def test_docker_and_agent_tools_over_seeded_rows(app): s.add(Host(id=hid, name=host_name, address="10.8.8.8")) s.add(Host(id=hid2, name=quiet_host, address="10.8.8.9")) async with s.begin(): - for name, st, health, exit_code, proj, service in ( - (web, "running", "healthy", None, f"stack-{tag}", None), - (db_, "running", "unhealthy", None, f"stack-{tag}", None), - (job, "exited", None, 1, None, None), - (f"{svc}.1.abc", "running", None, None, None, svc), + for name, st, health, exit_code, proj, service, task in ( + (web, "running", "healthy", None, f"stack-{tag}", None, None), + (db_, "running", "unhealthy", None, f"stack-{tag}", None, None), + (job, "exited", None, 1, None, None, None), + (f"{svc}.1.abc", "running", None, None, None, svc, f"abc-{tag}"), + # The task a rolling update replaced: killed (137), not crashed. + (replaced, "exited", None, 137, None, svc, f"old-{tag}"), ): await s.execute(container_sql, { "h": hid, "n": name, "cid": uuid.uuid4().hex, "st": st, "rc": 0, "started": now - timedelta(hours=2), "now": now, "health": health, - "exit": exit_code, "proj": proj, "svc": service}) - await s.execute(text( - "INSERT INTO docker_events (id, host_id, container_name, event, detail, at) " - "VALUES (:id, :h, :n, 'die', 'exit 1', :at)"), - {"id": str(uuid.uuid4()), "h": hid, "n": job, "at": now - timedelta(minutes=10)}) + "exit": exit_code, "proj": proj, "svc": service, "task": task}) + for name in (job, replaced): + await s.execute(text( + "INSERT INTO docker_events (id, host_id, container_name, event, detail, at) " + "VALUES (:id, :h, :n, 'die', 'exit 1', :at)"), + {"id": str(uuid.uuid4()), "h": hid, "n": name, "at": now - timedelta(minutes=10)}) for i, (stream, line) in enumerate(( ("stdout", "starting up"), ("stdout", "connected to db"), ("stderr", "ERROR: 100% of disk_used"), ("stdout", "retrying"), @@ -104,8 +112,18 @@ def test_docker_and_agent_tools_over_seeded_rows(app): "t": now - timedelta(minutes=30 - i * 10), "c": cpu}) await s.execute(text( "INSERT INTO docker_swarm_services (host_id, service_name, mode, desired, running, " - "placement_json, updated_at) VALUES (:h, :n, 'replicated', 3, 1, '[]', :now)"), - {"h": hid, "n": svc, "now": now}) + "placement_json, tasks_json, updated_at) " + "VALUES (:h, :n, 'replicated', 3, 1, '[]', :tasks, :now)"), + {"h": hid, "n": svc, "now": now, "tasks": json.dumps([ + {"task_id": f"abc-{tag}", "slot": 1, "node_id": f"node-{tag}", + "desired_state": "running", "state": "running", "message": "started", + "error": "", "exit_code": None, "at": _ago(9)}, + {"task_id": f"bad-{tag}", "slot": 2, "node_id": f"node-{tag}", + "desired_state": "shutdown", "state": "failed", "message": "started", + "error": "task: non-zero exit (1)", "exit_code": 1, "at": _ago(10)}, + {"task_id": f"old-{tag}", "slot": 1, "node_id": f"node-{tag}", + "desired_state": "shutdown", "state": "shutdown", "message": "shutdown", + "error": "", "exit_code": 137, "at": _ago(11)}])}) await s.execute(text( "INSERT INTO docker_swarm_nodes (host_id, node_id, hostname, role, availability, " "status, leader, updated_at) VALUES (:h, :nid, :hn, 'manager', 'active', 'ready', " @@ -154,7 +172,7 @@ def test_docker_and_agent_tools_over_seeded_rows(app): crashed = await dtools.get_container(job, host=host_name) assert [e["event"] for e in crashed["events"]] == ["die"] # A unique part of a long swarm task name is enough. - assert (await dtools.get_container(f"{svc}.1"))["swarm_service"] == svc + assert (await dtools.get_container(f"{svc}.1.ab"))["swarm_service"] == svc with pytest.raises(ValueError, match="list_containers"): await dtools.get_container(f"nope-{tag}") @@ -180,6 +198,15 @@ def test_docker_and_agent_tools_over_seeded_rows(app): sw = await dtools.list_swarm_services(limit=500) s = next(x for x in sw["services"] if x["service_name"] == svc) assert (s["state"], s["running"], s["desired"]) == ("degraded", 1, 3) + # Swarm's task records: the replaced task is history, the failed one counts. + assert [r["container_name"] for r in s["replicas"]] == [f"{svc}.1.abc"] + assert s["previous_tasks"] == 1 and s["tasks_known"] is True + assert (s["failures_1h"], s["failures_24h"], s["crash_looping"]) == (1, 1, False) + assert s["last_error"]["error"] == "task: non-zero exit (1)" + assert s["last_error"]["node"] == host_name and s["last_error"]["age_s"] >= 600 + old = (await dtools.list_containers(host=host_name, status="exited", limit=500))["containers"] + old = next(c for c in old if c["container_name"] == replaced) + assert old["swarm_replaced"] is True and old["problems"] == [] assert any(n["node_id"] == f"node-{tag}" and n["leader"] for n in sw["nodes"]) disk = (await dtools.get_docker_disk_usage(host=host_name))["hosts"] assert disk[0]["images_reclaimable_bytes"] == 3000 @@ -191,10 +218,23 @@ def test_docker_and_agent_tools_over_seeded_rows(app): assert ("container_unhealthy", db_) in found assert ("container_exited_nonzero", job) in found assert ("swarm_service_degraded", svc) in found + assert not any(subject == replaced for _, subject in found) assert not any(subject == web for _, subject in found) assert ("agent_never_reported", quiet_host) in agent_found assert not any(subject == host_name for _, subject in agent_found) + # ── the web pages say the same ───────────────────────────────────── + client = app.test_client() + async with client.session_transaction() as sess: + sess["user_id"], sess["user_role"], sess["username"] = "u", "viewer", "viewer" + topo = await (await client.get("/plugins/docker/swarm")).get_data(as_text=True) + assert svc in topo and "1 task failed in the last 24h" in topo + assert "task: non-zero exit (1)" in topo + cards = await (await client.get("/plugins/docker/rows")).get_data(as_text=True) + assert f"{svc}.1.abc" in cards and replaced not in cards + assert "task: non-zero exit (1)" in cards + assert (await client.get("/plugins/docker/widget")).status_code == 200 + # ── host agents ──────────────────────────────────────────────────── agents = {a["host_name"]: a for a in (await atools.list_agents())["agents"]} assert agents[host_name]["state"] == "reporting" and agents[host_name]["agent_version"] == "1.5.0" diff --git a/tests/plugins/docker/test_mcp_tools.py b/tests/plugins/docker/test_mcp_tools.py index c696fdf..1c6351a 100644 --- a/tests/plugins/docker/test_mcp_tools.py +++ b/tests/plugins/docker/test_mcp_tools.py @@ -73,6 +73,40 @@ def test_service_findings_by_state(): assert got == {"b": "warning", "c": "critical"} +def test_service_findings_name_the_crash_loop_and_swarms_error(): + err = {"error": "task: non-zero exit (1)", "message": "started", "state": "failed", + "exit_code": 1, "node": "docker03"} + svcs = [{"name": "db", "state": "down", "running": 0, "desired": 1, "crash_looping": True, + "failures_1h": 4, "failures_24h": 5, "last_error": err}, + {"name": "app", "state": "down", "running": 0, "desired": 1, "crash_looping": False, + "failures_1h": 0, "failures_24h": 0, + "last_error": {"error": "No such image: app:dev", "state": "rejected", + "exit_code": None, "node": None, "message": None}}, + {"name": "web", "state": "healthy", "running": 2, "desired": 2, "crash_looping": False, + "failures_1h": 0, "failures_24h": 1, "last_error": err}, + {"name": "ok", "state": "healthy", "running": 1, "desired": 1, "crash_looping": False, + "failures_1h": 0, "failures_24h": 0, "last_error": None}] + got = {f["subject"]: f for f in t.service_findings(svcs)} + assert set(got) == {"db", "app", "web"} + db = got["db"] + assert db["kind"] == "swarm_service_crash_looping" and db["severity"] == "critical" + assert "4 tasks failed in the last hour" in db["summary"] + assert "last task error: task: non-zero exit (1), exit 1 on docker03" in db["summary"] + # Down without a loop still names why: here Swarm could not even start it. + assert got["app"]["kind"] == "swarm_service_down" + assert got["app"]["summary"].endswith("last task error: No such image: app:dev") + # Recovered, but a task died in the last day: worth knowing, not critical. + assert got["web"]["kind"] == "swarm_task_failed" and got["web"]["severity"] == "warning" + assert "had 1 failed task in the last 24h" in got["web"]["summary"] + + +def test_a_replaced_swarm_task_is_marked_and_judged_on_its_service(): + task = _c(name="app.1.old", status="exited", exit_code=137, service_name="app") + row = t.container_row(task, {}, [], NOW, replaced=True) + assert row["swarm_replaced"] is True and row["problems"] == [] + assert t.container_row(_c(), {}, [], NOW)["swarm_replaced"] is None # not a swarm task + + def test_plugin_exposes_the_mcp_hooks(): reg = MagicMock() docker_plugin.register_mcp_tools(reg) diff --git a/tests/plugins/docker/test_swarm_view.py b/tests/plugins/docker/test_swarm_view.py index 19a7d12..b982d51 100644 --- a/tests/plugins/docker/test_swarm_view.py +++ b/tests/plugins/docker/test_swarm_view.py @@ -1,22 +1,40 @@ """Unit tests for the swarm-aware container view builder (pure, no DB/ORM).""" +import json +from datetime import datetime, timedelta, timezone from types import SimpleNamespace -from plugins.docker.swarm_view import build_swarm_services +from plugins.docker.swarm_view import build_swarm_services, last_task_error, task_failures NODE_HOSTNAME = {"nA": "docker01", "nB": "docker02", "nC": "docker03"} HOST_NAME = {"h1": "docker01", "h2": "docker02"} # agents run on docker01/02 only -def _c(host_id, name, *, service=None, node=None, status="running", cpu=1.0): +NOW = datetime(2026, 10, 6, 13, 30, tzinfo=timezone.utc) + + +def _c(host_id, name, *, service=None, node=None, status="running", cpu=1.0, + task=None, started=None): return SimpleNamespace(host_id=host_id, name=name, service_name=service, node_id=node, status=status, cpu_pct=cpu, mem_pct=2.0, - health=None, restart_count=0, container_id=name) + health=None, restart_count=0, container_id=name, + task_id=task, started_at=started) -def _svc(name, running, desired, placement, mode="replicated"): - import json +def _svc(name, running, desired, placement, mode="replicated", tasks=None, updated=None): return SimpleNamespace(service_name=name, running=running, desired=desired, - mode=mode, placement_json=json.dumps(placement)) + mode=mode, placement_json=json.dumps(placement), + tasks_json=json.dumps(tasks or []), updated_at=updated) + + +def _task(tid, *, state, desired="shutdown", ago_min=5, error="", exit_code=None, node="nA", + slot=1): + at = (NOW - timedelta(minutes=ago_min)).isoformat().replace("+00:00", "Z") + return {"task_id": tid, "slot": slot, "node_id": node, "desired_state": desired, + "state": state, "message": "", "error": error, "exit_code": exit_code, "at": at} + + +def _view(containers, services): + return build_swarm_services(containers, services, NODE_HOSTNAME, HOST_NAME, now=NOW) def test_real_replicas_plus_agentless_ghost(): @@ -77,3 +95,80 @@ def test_down_service_state(): services = [_svc("idle", 0, 2, [])] view = build_swarm_services([], services, NODE_HOSTNAME, HOST_NAME) assert view["services"][0]["state"] == "down" + + +# ── task awareness (agent ≥ 1.8.0) ─────────────────────────────────────────── + +def test_a_task_swarm_replaced_is_not_a_replica(): + # A stop-first redeploy: the old task was killed (exit 137) and Swarm set its + # desired state to shutdown. It is history, not a dead replica. + containers = [_c("h1", "app.1.old", service="app", node="nA", status="exited", task="old"), + _c("h2", "app.1.new", service="app", node="nB", task="new")] + tasks = [_task("new", state="running", desired="running", ago_min=1, node="nB"), + _task("old", state="shutdown", ago_min=2, exit_code=137)] + view = _view(containers, [_svc("app", 1, 1, [{"node_id": "nB", "running": 1}], tasks=tasks)]) + (svc,) = view["services"] + assert [r["name"] for r in svc["replicas"]] == ["app.1.new"] + assert svc["previous_tasks"] == 1 and svc["state"] == "healthy" + assert view["superseded"] == {("h1", "app.1.old")} + # A deliberate shutdown is not a failure. + assert svc["failures_24h"] == 0 and svc["last_error"] is None and not svc["crash_looping"] + + +def test_without_task_data_the_slot_decides(): + # An older agent sends no tasks: a stopped container is replaced when its slot + # has a running task, or one that started later. + t0 = NOW - timedelta(hours=2) + containers = [_c("h1", "lib.1.a", service="lib", status="exited", started=t0), + _c("h1", "lib.1.b", service="lib", status="exited", started=t0 + timedelta(minutes=5)), + _c("h1", "lib.1.c", service="lib", started=t0 + timedelta(minutes=10)), + _c("h1", "lib.2.d", service="lib", status="exited", started=t0)] + view = _view(containers, [_svc("lib", 1, 2, [])]) + (svc,) = view["services"] + # Slot 2's only task is stopped with nothing newer: still the current task. + assert sorted(r["name"] for r in svc["replicas"]) == ["lib.1.c", "lib.2.d"] + assert svc["previous_tasks"] == 2 + assert svc["tasks_known"] is False and svc["failures_1h"] is None + assert svc["crash_looping"] is False + + +def test_a_crash_loop_is_counted_across_task_retries(): + # Each retry is a new task and a new container that dies once; only the + # service sees the loop. + tasks = [_task("t5", state="starting", desired="running", ago_min=0)] + tasks += [_task(f"t{i}", state="failed", ago_min=10 * i, exit_code=1, + error="task: non-zero exit (1)") for i in range(1, 5)] + tasks.append(_task("t9", state="failed", ago_min=180, exit_code=1)) + view = build_swarm_services([], [_svc("db", 0, 1, [], tasks=tasks)], NODE_HOSTNAME, HOST_NAME, + now=NOW, crash_loop_deaths=3) + (svc,) = view["services"] + assert svc["failures_1h"] == 4 and svc["failures_24h"] == 5 + assert svc["crash_looping"] is True and svc["state"] == "down" + err = svc["last_error"] + assert err["error"] == "task: non-zero exit (1)" and err["exit_code"] == 1 + assert err["node"] == "docker01" and err["state"] == "failed" + # The threshold comes from Settings → Thresholds. + lax = build_swarm_services([], [_svc("db", 0, 1, [], tasks=tasks)], NODE_HOSTNAME, HOST_NAME, + now=NOW, crash_loop_deaths=5) + assert lax["services"][0]["crash_looping"] is False + + +def test_a_rejected_or_stuck_task_explains_itself(): + rejected = [_task("t1", state="rejected", desired="running", error="No such image: codex:dev")] + assert last_task_error(rejected)["error"] == "No such image: codex:dev" + # A task Swarm is still trying to place carries its reason too. + pending = [_task("t2", state="pending", desired="running", error="no suitable node")] + assert last_task_error(pending)["state"] == "pending" + assert last_task_error([_task("t3", state="running", desired="running")]) is None + assert task_failures(rejected, NOW - timedelta(hours=1)) == 1 + + +def test_the_newest_manager_report_wins(): + # Two managers report the same service; the one updated last (here: the one + # with task data) is used. + old = _svc("web", 1, 1, [], updated=NOW - timedelta(minutes=5)) + new = _svc("web", 1, 1, [], tasks=[_task("t1", state="failed", ago_min=3)], + updated=NOW - timedelta(seconds=30)) + (svc,) = _view([], [old, new])["services"] + assert svc["tasks_known"] is True and svc["failures_1h"] == 1 + diff --git a/tests/plugins/host_agent/test_agent_docker.py b/tests/plugins/host_agent/test_agent_docker.py index 10ed4c4..4eb0591 100644 --- a/tests/plugins/host_agent/test_agent_docker.py +++ b/tests/plugins/host_agent/test_agent_docker.py @@ -226,6 +226,8 @@ def test_swarm_services_rolls_up_replicas(): "Status": {"State": "running"}}, ] out = {s["service_name"]: s for s in a._swarm_services(services, tasks)} + assert [t["state"] for t in out["web"].pop("tasks")] == ["running", "running", "failed"] + assert len(out["agent"].pop("tasks")) == 2 assert out["web"] == {"service_name": "web", "mode": "replicated", "desired": 3, "running": 2, "image": "nginx:1", "placement": [{"node_id": "n1", "running": 2}]} @@ -236,6 +238,33 @@ def test_swarm_services_rolls_up_replicas(): {"node_id": "n2", "running": 1}]} +def test_swarm_task_keeps_why_it_stopped(): + t = a._swarm_task({ + "ID": "t1", "Slot": 1, "NodeID": "n2", "DesiredState": "Shutdown", + "Status": {"State": "failed", "Timestamp": "2026-10-06T13:19:51.5Z", + "Message": "started", "Err": "task: non-zero exit (1)" + "x" * 400, + "ContainerStatus": {"ContainerID": "c1", "ExitCode": 1}}}) + assert t["task_id"] == "t1" and t["slot"] == 1 and t["node_id"] == "n2" + assert t["desired_state"] == "shutdown" and t["state"] == "failed" and t["exit_code"] == 1 + assert t["at"] == "2026-10-06T13:19:51.5Z" and t["message"] == "started" + assert t["error"].startswith("task: non-zero exit (1)") and len(t["error"]) == 300 + # A task that never got a container has no exit code. + bare = a._swarm_task({"ID": "t2", "DesiredState": "running", + "Status": {"State": "rejected", "Err": "No such image: x"}}) + assert bare["exit_code"] is None and bare["slot"] is None and bare["error"] == "No such image: x" + + +def test_swarm_services_keeps_the_newest_tasks(): + services = [{"ID": "s", "Spec": {"Name": "db", "Mode": {"Replicated": {"Replicas": 1}}}}] + tasks = [{"ID": f"t{i}", "ServiceID": "s", "Slot": 1, "DesiredState": "shutdown", + "Status": {"State": "failed", "Timestamp": f"2026-10-06T13:{i:02d}:00Z"}} + for i in range(a.SWARM_TASKS_PER_SERVICE + 5)] + (svc,) = a._swarm_services(services, tasks) + ids = [t["task_id"] for t in svc["tasks"]] + newest = a.SWARM_TASKS_PER_SERVICE + 4 + assert len(ids) == a.SWARM_TASKS_PER_SERVICE and ids[0] == f"t{newest}" + + def test_swarm_nodes_normalises_fields(): nodes = [ {"ID": "n1", "Description": {"Hostname": "mgr"},