diff --git a/plugins/docker/overview.py b/plugins/docker/overview.py new file mode 100644 index 0000000..90d33b1 --- /dev/null +++ b/plugins/docker/overview.py @@ -0,0 +1,150 @@ +# plugins/docker/overview.py +"""The Docker page's two tables, as plain rows (#5299, #5301): one row per +swarm service and one per standalone container, with the search, stack, host +and status filters and the column sort the page offers. Model-free, so it is +unit-tested without a database; routes.rows() feeds it. + +A stack is how the operator thinks of a group of services. Swarm names a +stack's services "_", so a service's stack is the part of its +name before the first "_"; a standalone container's is its compose project. +Stacks filter the list rather than section it, as host groups do on Hosts. +""" +from __future__ import annotations + +from datetime import datetime + +from .swarm_view import service_trouble + +# Status filter values and their labels; NO_STACK picks rows with no stack. +DOCKER_STATES = {"problem": "Needs attention", "healthy": "Healthy", "stopped": "Stopped"} +NO_STACK = "__none__" +# Each sort and whether it starts descending: the busiest first; "up" starts +# with the shortest uptime, the thing that most recently restarted. +DOCKER_SORTS = {"status": False, "name": False, "stack": False, "cpu": True, "mem": True, + "up": False} + + +def stack_of(service_name: str) -> str | None: + """The stack a swarm service was deployed in: "arr" for "arr_radarr"; + None for a name with no "_" (a service created on its own).""" + stack, sep, rest = service_name.partition("_") + return stack if sep and stack and rest else None + + +def _seconds_up(started_at, now: datetime) -> float | None: + return None if started_at is None else max((now - started_at).total_seconds(), 0.0) + + +def service_row(s: dict, now: datetime) -> dict: + """A built swarm service (swarm_view.build_swarm_services) as a table row. + CPU and memory add up the running replicas; "up" is the newest replica's + start, so a service that just restarted a task reads as just restarted.""" + reals = [r for r in s["replicas"] if not r["ghost"]] + running = [r for r in reals if (r["status"] or "").lower() == "running"] + cpu = [r["cpu_pct"] for r in running if r.get("cpu_pct") is not None] + mem = [r["mem_bytes"] for r in running if r.get("mem_bytes") is not None] + starts = [r["started_at"] for r in running if r.get("started_at")] + started = max(starts) if starts else None + return { + "kind": "service", "name": s["name"], "stack": stack_of(s["name"]), + "mode": s["mode"], + "image": s.get("image") or next((r["image"] for r in reals if r.get("image")), None), + "running": s["running"], "desired": s["desired"], "state": s["state"], + "trouble": service_trouble(s), + "nodes": sorted({r["host"] for r in reals}, key=str.lower), + "ghost_nodes": sorted({r["host"] for r in s["replicas"] if r["ghost"]}, key=str.lower), + "cpu_cores": sum(cpu) / 100 if cpu else None, + "mem_bytes": sum(mem) if mem else None, + "started_at": started, "up_s": _seconds_up(started, now), + "unhealthy": any(r.get("health") == "unhealthy" for r in reals), + "failures_24h": s.get("failures_24h"), "crash_looping": s.get("crash_looping"), + "last_error": s.get("last_error"), "trouble_since": s.get("trouble_since"), + } + + +def container_trouble(c, *, stale: bool, exit_failed: bool) -> int | None: + """How bad a standalone container is, on service_trouble's scale: 1 + unhealthy or restarting, 2 died with a failing exit, 3 its report is + stale; None when nothing is wrong (a cleanly stopped one included).""" + if c.health == "unhealthy" or c.status == "restarting": + return 1 + if c.status != "running" and exit_failed: + return 2 + if stale: + return 3 + return None + + +def container_row(c, *, label: str, host: str, usage: dict, ports: list, stale: bool, + exit_failed: bool, now: datetime) -> dict: + """A standalone container as a table row; `usage` is service.usage().""" + running = c.status == "running" + return { + "kind": "container", "name": label, "container_name": c.name, "host": host, + "host_id": c.host_id, "stack": c.compose_project or None, "image": c.image, + "status": c.status, "health": c.health, "running": running, "ports": ports, + "usage": usage, "cpu_cores": usage.get("cores") if running else None, + "mem_bytes": usage.get("mem_bytes") if running else None, + "started_at": c.started_at if running else None, + "up_s": _seconds_up(c.started_at, now) if running else None, + "exit_code": c.exit_code, "oom_killed": c.oom_killed, "exit_failed": exit_failed, + "restart_count": c.restart_count or 0, "stale": stale, "scraped_at": c.scraped_at, + "trouble": container_trouble(c, stale=stale, exit_failed=exit_failed), + } + + +def _hosts(row: dict) -> list[str]: + return row["nodes"] + row["ghost_nodes"] if row["kind"] == "service" else [row["host"]] + + +def _stopped(row: dict) -> bool: + return (row["running"] == 0) if row["kind"] == "service" else not row["running"] + + +def filter_rows(rows: list[dict], *, q: str = "", stack: str = "", host: str = "", + state: str = "") -> list[dict]: + """Rows matching a search (name, container name, image or stack, any + case), a stack (or NO_STACK), a host or node, and a Status filter; an + empty value doesn't filter.""" + q = (q or "").strip().lower() + out = [] + for r in rows: + if q and not any(q in (v or "").lower() for v in + (r["name"], r.get("container_name"), r["image"], r["stack"])): + continue + if (stack == NO_STACK and r["stack"]) or (stack and stack != NO_STACK and r["stack"] != stack): + continue + if host and host not in _hosts(r): + continue + if state == "problem" and r["trouble"] is None: + continue + if state == "healthy" and (r["trouble"] is not None or _stopped(r)): + continue + if state == "stopped" and not _stopped(r): + continue + out.append(r) + return out + + +def sort_rows(rows: list[dict], sort: str, desc: bool) -> list[dict]: + """Rows in a column's order. Ties keep name order; a row with no value for + the column goes last whichever way it runs. "status" is problems first, + worst first, then running, then stopped.""" + by_name = sorted(rows, key=lambda r: r["name"].lower()) + if sort == "name": + return list(reversed(by_name)) if desc else by_name + if sort == "status": + return sorted(by_name, key=lambda r: (4 if r["trouble"] is None else r["trouble"], + _stopped(r)), reverse=desc) + value = {"stack": lambda r: (r["stack"] or "").lower() or None, + "cpu": lambda r: r["cpu_cores"], "mem": lambda r: r["mem_bytes"], + "up": lambda r: r["up_s"]}[sort] + have = [r for r in by_name if value(r) is not None] + return sorted(have, key=value, reverse=desc) + [r for r in by_name if value(r) is None] + + +def filter_options(services: list[dict], containers: list[dict]) -> dict: + """The stack and host choices the filter bar offers, from every row.""" + rows = services + containers + return {"stacks": sorted({r["stack"] for r in rows if r["stack"]}, key=str.lower), + "hosts": sorted({h for r in rows for h in _hosts(r)}, key=str.lower)} diff --git a/plugins/docker/routes.py b/plugins/docker/routes.py index 4db1636..95eb331 100644 --- a/plugins/docker/routes.py +++ b/plugins/docker/routes.py @@ -20,6 +20,10 @@ from .swarm_view import ( build_swarm_services, failing_services, parse_tasks, replaced_tasks, services_problems_first, task_summary, trouble_since, ) +from .overview import ( + DOCKER_SORTS, DOCKER_STATES, NO_STACK, container_row, filter_rows, service_row, sort_rows, + stack_of, +) from .models import ( DockerContainer, DockerDiskUsage, DockerHostStatus, DockerImage, DockerMetric, DockerSwarmNode, DockerSwarmService, @@ -62,6 +66,37 @@ def _sparkline(values: list[float], width: int = 80, height: int = 20) -> str: ) +def _filters(args) -> dict: + """The Docker page's search, filters and sort from the query string + (#5299): q, stack, host, state, sort, dir; anything unknown is dropped.""" + sort = args.get("sort") if args.get("sort") in DOCKER_SORTS else "status" + desc = {"desc": True, "asc": False}.get(args.get("dir"), DOCKER_SORTS[sort]) + return {"q": (args.get("q") or "").strip(), "stack": args.get("stack", ""), + "host": args.get("host", ""), + "state": args.get("state") if args.get("state") in DOCKER_STATES else "", + "sort": sort, "dir": "desc" if desc else "asc"} + + +def _page_url(f: dict, **change) -> str: + """The Docker page with some filters changed; defaults are left out.""" + from urllib.parse import urlencode + want = {**f, **change} + if want["dir"] == ("desc" if DOCKER_SORTS[want["sort"]] else "asc"): + want["dir"] = "" + if want["sort"] == "status": + want["sort"] = "" + query = urlencode({k: v for k, v in want.items() if v}) + return "/plugins/docker/" + ("?" + query if query else "") + + +def _sort_url(f: dict, key: str) -> str: + """A column header's link: its own default direction, or the other way + when it is already the sort.""" + if key == f["sort"]: + return _page_url(f, sort=key, dir="asc" if f["dir"] == "desc" else "desc") + return _page_url(f, sort=key, dir="desc" if DOCKER_SORTS[key] else "asc") + + @docker_bp.get("/") @require_role(UserRole.viewer) async def index(): @@ -73,21 +108,60 @@ async def index(): select(DockerSwarmService.host_id).limit(1))).first() is not None has_disk = (await db.execute( select(DockerDiskUsage.host_id).limit(1))).first() is not None + choices = await _filter_choices(db) return await render_template( "docker/index.html", poll_interval=poll_interval, has_swarm=has_swarm, has_disk=has_disk, + f=_filters(request.args), choices=choices, states=DOCKER_STATES, no_stack=NO_STACK, ) +async def _filter_choices(db) -> dict: + """The stacks and hosts the filter bar offers: every swarm service's stack + and compose project, and every host or node that runs something.""" + names = set((await db.execute(select(DockerSwarmService.service_name).distinct())).scalars()) + names |= {n for n in (await db.execute( + select(DockerContainer.service_name).distinct())).scalars() if n} + stacks = {stack_of(n) for n in names} | {p for p in (await db.execute( + select(DockerContainer.compose_project).distinct())).scalars() if p} + host_ids = set((await db.execute(select(DockerContainer.host_id).distinct())).scalars()) + hosts = {h.name for h in (await docker_svc.host_map(db, host_ids)).values()} + hosts |= {n for n in (await db.execute(select(DockerSwarmNode.hostname).distinct())).scalars() if n} + return {"stacks": sorted((s for s in stacks if s), key=str.lower), + "hosts": sorted(hosts, key=str.lower)} + + +async def _swarm_context(db, now: datetime) -> dict: + """What the Docker page and a service's page both build from: containers, + swarm services and nodes, host names, and the swarm view (each service with + its current replicas, Swarm's task records and their failures).""" + containers = await docker_svc.list_containers(db) + services = list((await db.execute(select(DockerSwarmService))).scalars()) + nodes = list((await db.execute(select(DockerSwarmNode))).scalars()) + hosts = await docker_svc.host_map( + db, {c.host_id for c in containers} | {n.host_id for n in nodes} + | {s.host_id for s in services}) + 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, now=now, + crash_loop_deaths=docker_svc.crash_loop_deaths(current_app.config)) + return {"containers": containers, "services": services, "nodes": nodes, "hosts": hosts, + "node_hostname": node_hostname, "host_name": host_name, "swarm_view": swarm_view} + + @docker_bp.get("/rows") @require_role(UserRole.viewer) async def rows(): - """HTMX fragment: what needs attention first, then swarm services - (collapsed, cluster-complete), then non-swarm containers grouped by host.""" + """HTMX fragment: a verdict line and the problems that aren't zero, what + needs attention, what failed in 24h (each with its logs), then the swarm + services and the standalone containers as two tables that share the + filter bar's search, filters and sort (#5298–#5301).""" now = datetime.now(timezone.utc) stale_after = docker_svc.stale_after_seconds(current_app.config) + f = _filters(request.args) async with current_app.db_sessionmaker() as db: # Hosts whose Docker is not answering: their rows are what it last said. @@ -95,84 +169,43 @@ async def rows(): select(DockerHostStatus).where(DockerHostStatus.status != "ok"))).scalars()} ranked = await docker_svc.ranked_containers( db, now, docker_svc.crash_loop_deaths(current_app.config)) - containers = await docker_svc.list_containers(db) - services = list((await db.execute(select(DockerSwarmService))).scalars()) - nodes = list((await db.execute(select(DockerSwarmNode))).scalars()) - hosts = await docker_svc.host_map( - db, - {c.host_id for c in containers} - | {n.host_id for n in nodes} | {s.host_id for s in services} | set(docker_down), - ) - - # Swarm tasks collapse into per-service rows (each replica labelled with - # its host, plus "ghost" replicas for tasks on agent-less nodes, drawn - # 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} + ctx = await _swarm_context(db, now) + containers, hosts = ctx["containers"], ctx["hosts"] + hosts.update(await docker_svc.host_map(db, set(docker_down) - set(hosts))) host_name = {h.id: h.name for h in hosts.values()} - swarm_view = build_swarm_services( - containers, services, node_hostname, host_name, - crash_loop_deaths=docker_svc.crash_loop_deaths(current_app.config)) + swarm_view = ctx["swarm_view"] capacity = await docker_svc.host_capacity(db, {c.host_id for c in containers}) failures = docker_svc.failures_24h( await docker_svc.failure_events(db, now - timedelta(hours=24)), swarm_view["services"], host_name) - standalone = [c for c in containers if not c.service_name] - groups: dict[str, dict] = {} - for c in standalone: - g = groups.get(c.host_id) - if g is None: - host = hosts.get(c.host_id) - g = groups[c.host_id] = { - "host": host, - "host_id": c.host_id, - "host_name": host.name if host else c.host_id, - "containers": [], "running": 0, "stopped": 0, - } - g["containers"].append({ - "container": c, - "ports": json.loads(c.ports_json) if c.ports_json else [], - "uptime": _uptime(c.started_at) if c.status == "running" else None, - "stale": _is_stale(c, now, stale_after, docker_down), - "usage": docker_svc.usage(c, *capacity.get(c.host_id, (None, None))), - "exit_failed": docker_svc.exit_is_failure(c.exit_code, c.oom_killed), - }) - if c.status == "running": - g["running"] += 1 - else: - g["stopped"] += 1 + service_rows = [service_row(s, now) for s in swarm_view["services"]] + standalone = [c for c in containers if not c.service_name] + container_rows = [ + container_row( + c, label=docker_svc.display_name(c), host=host_name.get(c.host_id, c.host_id), + usage=docker_svc.usage(c, *capacity.get(c.host_id, (None, None))), + ports=json.loads(c.ports_json) if c.ports_json else [], + stale=_is_stale(c, now, stale_after, docker_down), + exit_failed=docker_svc.exit_is_failure(c.exit_code, c.oom_killed), now=now) + for c in standalone] + desc = f["dir"] == "desc" + shown_services = sort_rows(filter_rows(service_rows, q=f["q"], stack=f["stack"], host=f["host"], + state=f["state"]), f["sort"], desc) + shown_containers = sort_rows(filter_rows(container_rows, q=f["q"], stack=f["stack"], + host=f["host"], state=f["state"]), f["sort"], desc) - # Sub-group each host's containers by compose project, preserving the - # running-first ordering. Containers with no project fall into an unlabelled - # bucket rendered flat, so plain hosts look unchanged. - for g in groups.values(): - subs: dict[str, dict] = {} - order: list[str] = [] - for item in g["containers"]: - c = item["container"] - label = c.compose_project or None - key = label or "" - if key not in subs: - subs[key] = {"label": label, "containers": []} - order.append(key) - subs[key]["containers"].append(item) - g["subgroups"] = [subs[k] for k in order] - g["grouped"] = any(s["label"] for s in g["subgroups"]) - - host_groups = sorted(groups.values(), key=lambda g: g["host_name"].lower()) - swarm_services = services_problems_first(swarm_view["services"]) - standalone_running = sum(g["running"] for g in host_groups) + failing = failing_services(swarm_view["services"]) attention = [dict(r, stale=_is_stale(r["container"], now, stale_after, docker_down), - label=docker_svc.task_label(r["container"], node_hostname)) + label=docker_svc.task_label(r["container"], ctx["node_hostname"])) for r in docker_svc.needs_attention(ranked)] - failing = failing_services(swarm_services) + running = sum(r["running"] for r in container_rows) return await render_template( "docker/rows.html", - host_groups=host_groups, - swarm_services=swarm_services, - failing_services=failing, - failures=failures, - services_healthy=sum(s["state"] == "healthy" for s in swarm_services), + services=shown_services, containers=shown_containers, + services_total=len(service_rows), containers_total=len(container_rows), + services_healthy=sum(s["state"] == "healthy" for s in service_rows), + failing_services=failing, failures=failures, docker_hosts=len({c.host_id for c in containers}), docker_down=sorted( ({"host_id": h, "host_name": hosts[h].name if h in hosts else h, "status": d.status, @@ -181,9 +214,10 @@ async def rows(): attention=attention, says=docker_svc.PROBLEM_SAYS, severity=docker_svc.PROBLEM_SEVERITY, unhealthy=sum("unhealthy" in r["problems"] for r in attention), restarting=sum(bool({"restarting", "crash_looping"} & set(r["problems"])) for r in attention), - stale=sum(1 for g in host_groups for i in g["containers"] if i["stale"]), - running=standalone_running, - stopped=len(standalone) - standalone_running, + stale=sum(r["stale"] for r in container_rows), + running=running, stopped=len(container_rows) - running, + f=f, sort_url=lambda key: _sort_url(f, key), page_url=lambda **kw: _page_url(f, **kw), + filtered=any(f[k] for k in ("q", "stack", "host", "state")), ) @@ -743,3 +777,113 @@ async def container_logs_lines(host_id: str, name: str): name=name, tail=_LOG_TAIL_DEFAULT, filtered=bool(query) or stream in ("stdout", "stderr"), ) + + +# ── A swarm service's own page (#5302) ─────────────────────────────────────── + +def _task_label(slot: str, task_id: str) -> str: + """How a task reads beside its log lines: its slot and the start of its id.""" + return f"{slot} · {task_id[:10]}" + + +def _task_rows(tasks: list[dict], node_hostname: dict) -> list[dict]: + """Swarm's records of a service's newest tasks, newest first, as the page + lists them: where each ran, what state it reached, and why it ended.""" + from .swarm_view import FAILED_TASK_STATES, task_time + out = [] + for t in tasks: + nid = t.get("node_id") or "" + out.append({ + "task_id": t.get("task_id") or "", "slot": str(t.get("slot") or ""), + "node": node_hostname.get(nid) or (nid[:12] if nid else "—"), + "state": t.get("state") or "unknown", "desired": t.get("desired_state") or "", + "exit_code": t.get("exit_code"), "why": t.get("error") or t.get("message") or "", + "at": task_time(t), "failed": t.get("state") in FAILED_TASK_STATES, + "current": t.get("state") == "running" and t.get("desired_state") == "running", + }) + return out + + +@docker_bp.get("/service/") +@require_role(UserRole.viewer) +async def service_detail(name: str): + """One swarm service: where its replicas run and what they use, its image, + Swarm's record of its recent tasks, its CPU and memory over time across + every replica, and the logs of every task it has run — newest first, each + line marked with its task. `?task=` opens the logs on one task, which is + where a failure's "Logs →" lands.""" + now = datetime.now(timezone.utc) + async with current_app.db_sessionmaker() as db: + ctx = await _swarm_context(db, now) + task_containers = await docker_svc.service_containers(db, name) + s = next((s for s in ctx["swarm_view"]["services"] if s["name"] == name), None) + capacity = await docker_svc.host_capacity( + db, {r["host_id"] for r in (s["replicas"] if s else []) if not r["ghost"]}) + if s is None and not task_containers: + return await render_template("docker/service.html", name=name, service=None), 404 + + newest = max((v for v in ctx["services"] if v.service_name == name), + key=lambda v: v.updated_at, default=None) + tasks = _task_rows(parse_tasks(newest) if newest is not None else [], ctx["node_hostname"]) + by_key = {(c.host_id, c.name): c for c in ctx["containers"]} + replicas = [] + for r in (s["replicas"] if s else []): + c = None if r["ghost"] else by_key.get((r["host_id"], r["name"])) + replicas.append(dict(r, usage=docker_svc.usage(c, *capacity.get(r["host_id"], (None, None))) + if c is not None else None, + uptime=_uptime(c.started_at) if c is not None and c.status == "running" else None)) + # Every task Steward holds logs or events for, newest known first. + known = {t["task_id"]: _task_label(t["slot"], t["task_id"]) for t in tasks if t["task_id"]} + for _, cname in task_containers: + slot, tid = docker_svc.task_of(name, cname) + known.setdefault(tid, _task_label(slot, tid)) + focus = request.args.get("task", "") + return await render_template( + "docker/service.html", name=name, service=s, + row=service_row(s, now) if s else None, svc=newest, tasks=tasks, replicas=replicas, + task_choices=list(known.items()), focus_task=focus if focus in known else "", + stack=stack_of(name), current_range=request.args.get("range", DEFAULT_RANGE), + ) + + +@docker_bp.get("/service//history") +@require_role(UserRole.viewer) +async def service_history_fragment(name: str): + """HTMX fragment: the service's CPU (cores) and memory, summed across its + replicas, over the selected range, with the latest and the peak.""" + from steward.core.status import sparkline_svg + since, range_key = parse_range(request.args.get("range")) + async with current_app.db_sessionmaker() as db: + series = await docker_svc.service_history( + db, await docker_svc.service_containers(db, name), since) + cpu = [c / 100 for _, c, _ in series] + mem = [m for _, _, m in series] + return await render_template( + "docker/_service_history.html", have_data=len(series) >= 2, range_key=range_key, + cpu_now=cpu[-1] if cpu else None, cpu_peak=max(cpu) if cpu else None, + mem_now=mem[-1] if mem else None, mem_peak=max(mem) if mem else None, + spark_cpu=sparkline_svg(cpu, height=48, stroke="currentColor", fill_width=True), + spark_mem=sparkline_svg(mem, height=48, stroke="currentColor", fill_width=True), + ) + + +@docker_bp.get("/service//logs/lines") +@require_role(UserRole.viewer) +async def service_logs_lines(name: str): + """HTMX fragment: the newest retained lines across every task of the + service, stream, text and task filtered, each marked with its task.""" + stream = request.args.get("stream", "all") + query = (request.args.get("q") or "").strip() + task = request.args.get("task") or None + async with current_app.db_sessionmaker() as db: + rows = await docker_svc.query_service_logs( + db, await docker_svc.service_containers(db, name), stream, query, _LOG_TAIL_DEFAULT, + task=task) + lines = [] + for r in rows: + slot, tid = docker_svc.task_of(name, r.container_name) or ("?", r.container_name) + lines.append({"ts": r.ts, "stream": r.stream, "line": r.line, "task": _task_label(slot, tid)}) + return await render_template( + "docker/_container_logs_lines.html", lines=lines, name=name, tail=_LOG_TAIL_DEFAULT, + filtered=bool(query) or bool(task) or stream in ("stdout", "stderr"), service=True, + ) diff --git a/plugins/docker/service.py b/plugins/docker/service.py index 8d98931..fd68276 100644 --- a/plugins/docker/service.py +++ b/plugins/docker/service.py @@ -73,7 +73,8 @@ def failures_24h(events, swarm_services, host_name: dict) -> list[dict]: (an agent older than 1.8.0) falls back to its tasks' deaths. Each row: {kind: "container"|"service", name, host_id, host, count, at, - detail}. + detail, task_id}; task_id is the failed swarm task, so its logs can be + found (None for a container, or a service without task records). """ by_prefix = {s["name"] + ".": s for s in swarm_services} rows: dict[tuple, dict] = {} @@ -97,7 +98,9 @@ def failures_24h(events, swarm_services, host_name: dict) -> list[dict]: if row is None: rows[key] = {"kind": kind, "name": name, "host_id": e.host_id, "host": host_name.get(e.host_id, e.host_id), "count": 1, "at": e.at, - "detail": "OOM-killed" if e.event == "oom" else e.detail} + "detail": "OOM-killed" if e.event == "oom" else e.detail, + "task_id": (task_of(svc["name"], e.container_name) or (None, None))[1] + if svc else None} else: row["count"] += 1 from .swarm_view import task_time @@ -107,7 +110,8 @@ def failures_24h(events, swarm_services, host_name: dict) -> list[dict]: rows[("service", s["name"])] = { "kind": "service", "name": s["name"], "host_id": None, "host": err.get("node"), "count": s["failures_24h"], "at": task_time(err) if err else None, - "detail": err.get("error") or err.get("message") or err.get("state")} + "detail": err.get("error") or err.get("message") or err.get("state"), + "task_id": err.get("task_id")} epoch = datetime.min.replace(tzinfo=timezone.utc) return sorted(rows.values(), key=lambda r: r["at"] or epoch, reverse=True) @@ -470,6 +474,92 @@ async def retained_names(db, want: str, *, exact: bool, host_id: str | None = No return rows[:limit] +# A swarm task's container is named "..": slot is the +# replica number, or the node id for a global service; task ids are 25 chars. +_TASK_SUFFIX = re.compile(r"^(?P[A-Za-z0-9]+)\.(?P[a-z0-9]{25})$") + + +def task_of(service: str, container_name: str) -> tuple[str, str] | None: + """(slot, task id) when `container_name` is one of `service`'s task + containers, else None — so "arr" does not claim "arr_radarr.1.…".""" + if not container_name.startswith(service + "."): + return None + m = _TASK_SUFFIX.match(container_name[len(service) + 1:]) + return (m["slot"], m["task"]) if m else None + + +async def service_containers(db, service: str) -> list[tuple[str, str]]: + """(host_id, container_name) of every task container of a swarm service + that Steward still holds anything for: the live ones, and the ones Docker + removed that left events (each task container has a start event). The + page's logs and history read across all of them (#5302).""" + pattern = escape_like(service + ".") + "%" + found: set[tuple[str, str]] = set() + for model, col in ((DockerContainer, DockerContainer.name), + (DockerEvent, DockerEvent.container_name)): + rows = (await db.execute( + select(model.host_id, col).where(col.like(pattern, escape="\\")).distinct())).all() + found.update((h, n) for h, n in rows if task_of(service, n)) + return sorted(found) + + +async def query_service_logs(db, containers, stream: str, query: str, limit: int, + task: str | None = None) -> list: + """The newest retained lines across a service's task containers, newest + first, each with its container name so the page can say which task wrote + it. `task` keeps one task's lines (a failure's "Logs →").""" + from sqlalchemy import tuple_ + pairs = [(h, n) for h, n in containers if task is None or n.endswith("." + task)] + if not pairs: + return [] + stmt = (select(DockerLog.ts, DockerLog.stream, DockerLog.line, DockerLog.container_name) + .where(tuple_(DockerLog.host_id, DockerLog.container_name).in_(pairs))) + if stream in ("stdout", "stderr"): + stmt = stmt.where(DockerLog.stream == stream) + if query: + 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() + + +def sum_series(samples, since: datetime, until: datetime, points: int = 60) -> list: + """[(bucket start, cpu %, memory bytes)] summed across containers: each + container's samples are averaged within a bucket, then the containers are + added, so two replicas read as their combined load. `samples` are + (container_name, at, cpu_pct, mem_bytes); buckets with no sample are left + out rather than read as zero.""" + width = max((until - since).total_seconds() / points, 60.0) + acc: dict[int, dict[str, list]] = {} + for name, at, cpu, mem in samples: + i = int((at - since).total_seconds() // width) + if i < 0: + continue + per = acc.setdefault(i, {}).setdefault(name, [0.0, 0, 0]) + per[0] += cpu or 0.0 + per[1] += mem or 0 + per[2] += 1 + out = [] + for i in sorted(acc): + cpu = sum(c / n for c, _, n in acc[i].values()) + mem = sum(m / n for _, m, n in acc[i].values()) + out.append((since + timedelta(seconds=i * width), cpu, int(mem))) + return out + + +async def service_history(db, containers, since: datetime) -> list: + """sum_series over a service's task containers' samples since `since`.""" + from sqlalchemy import tuple_ + if not containers: + return [] + rows = (await db.execute( + select(DockerMetric.container_name, DockerMetric.scraped_at, + DockerMetric.cpu_pct, DockerMetric.mem_usage_bytes) + .where(tuple_(DockerMetric.host_id, DockerMetric.container_name).in_(list(containers)), + DockerMetric.scraped_at >= since) + .order_by(DockerMetric.scraped_at))).all() + return sum_series(rows, since, datetime.now(timezone.utc)) + + async def unhealthy_since(db, keys) -> dict[tuple[str, str], datetime]: """{(host_id, container_name): when it last turned unhealthy} for the given containers, from their health_change events; absent when none is retained.""" diff --git a/plugins/docker/swarm_view.py b/plugins/docker/swarm_view.py index 01047a9..984852a 100644 --- a/plugins/docker/swarm_view.py +++ b/plugins/docker/swarm_view.py @@ -94,7 +94,8 @@ def last_task_error(tasks: list[dict]) -> dict | None: 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} + "node_id": t.get("node_id") or None, "at": t.get("at") or None, + "task_id": t.get("task_id") or None, "slot": t.get("slot")} return None @@ -233,11 +234,12 @@ def build_swarm_services(containers, services, node_hostname: dict, host_name: d *, 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:[...], + services[i] = {name, mode, image, 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 + mem_bytes, started_at, image, 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 @@ -288,6 +290,9 @@ def build_swarm_services(containers, services, node_hostname: dict, host_name: d "status": getattr(c, "status", "unknown"), "cpu_pct": getattr(c, "cpu_pct", None), "mem_pct": getattr(c, "mem_pct", None), + "mem_bytes": getattr(c, "mem_usage_bytes", None), + "started_at": getattr(c, "started_at", None), + "image": getattr(c, "image", None) or None, "health": getattr(c, "health", None), "restart_count": getattr(c, "restart_count", 0) or 0, }) @@ -329,6 +334,7 @@ def build_swarm_services(containers, services, node_hostname: dict, host_name: d out_services.append({ "name": name, "mode": getattr(svc, "mode", "replicated") if svc is not None else "replicated", + "image": getattr(svc, "image", None) if svc is not None else None, "running": running, "desired": desired, "state": state, diff --git a/plugins/docker/templates/docker/_container_logs_lines.html b/plugins/docker/templates/docker/_container_logs_lines.html index 4e67b45..6dd2d50 100644 --- a/plugins/docker/templates/docker/_container_logs_lines.html +++ b/plugins/docker/templates/docker/_container_logs_lines.html @@ -1,9 +1,11 @@ {# docker/_container_logs_lines.html — log-line fragment, HTMX-polled (m79). - {{ l.line }} is auto-escaped by Jinja, so log content can't inject markup. #} + {{ l.line }} is auto-escaped by Jinja, so log content can't inject markup. + A service's lines (#5302) carry the task each came from. #} {% if lines %} {% for l in lines %}
{{ l.ts.strftime("%m-%d %H:%M:%S") }} + {% if l.task %}{{ l.task }}{% endif %} {{ l.stream }} {{ l.line }}
@@ -14,7 +16,7 @@ No lines match the current filter. {% else %} No logs collected yet for {{ name }}. Lines appear within a few - seconds of the container writing to stdout/stderr — unless it's on the log + seconds of {{ "one of its tasks" if service else "the container" }} writing to stdout/stderr — unless it's on the log exclude list or log collection is turned off in Settings → Thresholds & Retention. {% endif %} diff --git a/plugins/docker/templates/docker/_service_history.html b/plugins/docker/templates/docker/_service_history.html new file mode 100644 index 0000000..c79b204 --- /dev/null +++ b/plugins/docker/templates/docker/_service_history.html @@ -0,0 +1,22 @@ +{# docker/_service_history.html — a service's CPU and memory summed across its + replicas over the selected range (#5302): the latest and the peak, over a + line that spans its half. #} +{% from "docker/_usage.html" import cores %} +{% if have_data %} +
+
+
CPU
+
{{ cores(cpu_now) }} now · peak {{ cores(cpu_peak) }}
+
{{ spark_cpu|safe }}
+
+
+
Memory
+
{{ mem_now|bytes }} now · peak {{ mem_peak|bytes }}
+
{{ spark_mem|safe }}
+
+
+{% else %} +

+ Not enough samples in the last {{ range_key }} yet. History fills in as the agents report its replicas. +

+{% endif %} diff --git a/plugins/docker/templates/docker/index.html b/plugins/docker/templates/docker/index.html index d662bc3..1df9229 100644 --- a/plugins/docker/templates/docker/index.html +++ b/plugins/docker/templates/docker/index.html @@ -1,5 +1,8 @@ {% extends "base.html" %} {% from "_macros.html" import page_header %} +{# The Docker page. The filter bar sits outside the polled fragment, so a poll + never resets what is being typed; the fragment reads the bar on every poll, + and a change re-fetches it and puts the filters in the URL (#5299). #} {% block title %}Docker — Steward{% endblock %} {% block content %} {% call page_header("Docker", "What needs attention first, then every service and container.") %} @@ -7,17 +10,53 @@ {% if has_disk %}Disk{% endif %} {% endcall %} + +

Loading…

-{# A link to a section (the dashboard's "failed (24h)" → #failures) lands - before the rows exist; scroll to it once, when they first arrive. #}