bvandeusenandClaude Opus 5.5
aae61f1d2c
#5303 "Swarm page laid out by node: role, resources and the tasks placed on each" A verdict line first, then the nodes that are down or drained and the services short of replicas, then one card per node. A node with a Steward agent shows what its swarm tasks use against the machine's cores and memory, service by service. A node without one lists Swarm's placement, by count. The services table moved to the Docker overview, where it can be filtered. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
906 lines
42 KiB
Python
906 lines
42 KiB
Python
# plugins/docker/routes.py
|
||
from __future__ import annotations
|
||
import json
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from quart import (
|
||
Blueprint, current_app, jsonify, redirect, render_template, request,
|
||
session, url_for,
|
||
)
|
||
from sqlalchemy import func, select
|
||
|
||
from steward.auth.middleware import require_role
|
||
from steward.models.users import UserRole
|
||
from steward.models.hosts import Host
|
||
from steward.core.formatting import human_bytes, human_duration, seconds_since
|
||
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, failing_services, parse_tasks, replaced_tasks, services_problems_first,
|
||
task_summary, trouble_since,
|
||
)
|
||
from .swarm_nodes import node_cards
|
||
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,
|
||
)
|
||
|
||
docker_bp = Blueprint("docker", __name__, template_folder="templates")
|
||
|
||
|
||
def _error(status: int, code: str, detail: str | None = None):
|
||
body: dict = {"ok": False, "error": code}
|
||
if detail:
|
||
body["detail"] = detail
|
||
return jsonify(body), status
|
||
|
||
|
||
def _uptime(started_at: datetime | None) -> str | None:
|
||
"""How long a container has run ('3d 4h'), or None if unknown."""
|
||
secs = seconds_since(started_at)
|
||
return None if secs is None or secs < 0 else human_duration(secs)
|
||
|
||
|
||
def _sparkline(values: list[float], width: int = 80, height: int = 20) -> str:
|
||
if len(values) < 2:
|
||
return f'<svg width="{width}" height="{height}"></svg>'
|
||
mn, mx = min(values), max(values)
|
||
if mx == mn:
|
||
mx = mn + 1.0
|
||
step = width / (len(values) - 1)
|
||
pts = []
|
||
for i, v in enumerate(values):
|
||
x = i * step
|
||
y = height - (v - mn) / (mx - mn) * (height - 2) - 1
|
||
pts.append(f"{x:.1f},{y:.1f}")
|
||
poly = " ".join(pts)
|
||
return (
|
||
f'<svg width="{width}" height="{height}" viewBox="0 0 {width} {height}" '
|
||
f'style="vertical-align:middle;">'
|
||
f'<polyline points="{poly}" fill="none" stroke="#6060c0" stroke-width="1.5"/>'
|
||
f'</svg>'
|
||
)
|
||
|
||
|
||
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():
|
||
poll_interval = current_app.config.get("MONITORS_POLL_INTERVAL", 60)
|
||
# Show the Swarm link only when a manager has actually reported topology, so
|
||
# non-swarm installs aren't offered an always-empty page.
|
||
async with current_app.db_sessionmaker() as db:
|
||
has_swarm = (await db.execute(
|
||
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: 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.
|
||
docker_down = {d.host_id: d for d in (await db.execute(
|
||
select(DockerHostStatus).where(DockerHostStatus.status != "ok"))).scalars()}
|
||
ranked = await docker_svc.ranked_containers(
|
||
db, now, docker_svc.crash_loop_deaths(current_app.config))
|
||
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 = 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)
|
||
|
||
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)
|
||
|
||
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"], ctx["node_hostname"]))
|
||
for r in docker_svc.needs_attention(ranked)]
|
||
running = sum(r["running"] for r in container_rows)
|
||
return await render_template(
|
||
"docker/rows.html",
|
||
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,
|
||
"detail": d.detail, "since": d.since} for h, d in docker_down.items()),
|
||
key=lambda d: d["host_name"].lower()),
|
||
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(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")),
|
||
)
|
||
|
||
|
||
def _is_stale(c, now: datetime, stale_after: int, docker_down=()) -> bool:
|
||
"""A container whose last report is older than the agent's staleness
|
||
window, or whose host's Docker is not answering: what the page shows may
|
||
no longer be true."""
|
||
return (c.host_id in docker_down or c.scraped_at is None
|
||
or (now - c.scraped_at).total_seconds() > stale_after)
|
||
|
||
|
||
@docker_bp.get("/widget")
|
||
@require_role(UserRole.viewer)
|
||
async def widget():
|
||
"""HTMX dashboard widget: swarm services and containers, problems first,
|
||
with how many things failed in the last 24h (linked to the list)."""
|
||
show_stopped = request.args.get("show_stopped", "no") == "yes"
|
||
widget_id = request.args.get("wid", "0")
|
||
now = datetime.now(timezone.utc)
|
||
|
||
async with current_app.db_sessionmaker() as db:
|
||
all_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())
|
||
events = await docker_svc.failure_events(db, now - timedelta(hours=24))
|
||
hosts = await docker_svc.host_map(
|
||
db,
|
||
{c.host_id for c in all_containers}
|
||
| {n.host_id for n in nodes} | {s.host_id for s in services},
|
||
)
|
||
|
||
# Swarm tasks collapse into per-service rows (each replica tagged with its
|
||
# 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 = services_problems_first(build_swarm_services(
|
||
all_containers, services, node_hostname, host_name,
|
||
crash_loop_deaths=docker_svc.crash_loop_deaths(current_app.config))["services"])
|
||
failures = docker_svc.failures_24h(events, swarm_services, host_name)
|
||
|
||
standalone = [c for c in all_containers if not c.service_name]
|
||
running = [c for c in standalone if c.status == "running"]
|
||
|
||
def troubled(c) -> bool:
|
||
return ((c.status or "") == "restarting" or c.health == "unhealthy"
|
||
or (c.status != "running" and docker_svc.exit_is_failure(c.exit_code, c.oom_killed)))
|
||
|
||
# The widget-shape order (UI & Design ruling): what is wrong first, then
|
||
# the healthy rest one line each, collapsing past HEALTHY_SHOWN.
|
||
bad_containers = sorted((c for c in standalone if troubled(c)), key=lambda c: c.name)
|
||
healthy = ([{"kind": "service", "s": s} for s in swarm_services if s["state"] == "healthy"]
|
||
+ [{"kind": "container", "c": c} for c in sorted(running, key=lambda c: c.name)
|
||
if not troubled(c)]
|
||
+ ([{"kind": "stopped", "c": c} for c in sorted(standalone, key=lambda c: c.name)
|
||
if c.status != "running" and not troubled(c)] if show_stopped else []))
|
||
|
||
return await render_template(
|
||
"docker/widget.html",
|
||
bad_services=[s for s in swarm_services if s["state"] != "healthy"],
|
||
bad_containers=bad_containers,
|
||
healthy=healthy[:HEALTHY_SHOWN], healthy_more=max(0, len(healthy) - HEALTHY_SHOWN),
|
||
services_total=len(swarm_services),
|
||
services_healthy=sum(s["state"] == "healthy" for s in swarm_services),
|
||
running_count=len(running),
|
||
failures=failures,
|
||
host_name=host_name,
|
||
total_count=len(all_containers) + len(swarm_services),
|
||
widget_id=widget_id,
|
||
)
|
||
|
||
|
||
# Healthy services and containers listed on the dashboard widget before the
|
||
# rest collapse to "+N healthy"; the Docker page lists them all.
|
||
HEALTHY_SHOWN = 40
|
||
FAILURES_SHOWN = 5
|
||
|
||
|
||
CPU_AVERAGE_MINUTES = 5
|
||
|
||
|
||
@docker_bp.get("/widget/resources")
|
||
@require_role(UserRole.viewer)
|
||
async def widget_resources():
|
||
"""HTMX dashboard widget: the containers using the most CPU, averaged over
|
||
the last few minutes so the order holds still between polls, with their
|
||
CPU in cores and memory in bytes, each beside bars of the last few hours
|
||
scaled to the row's own peak (never below a floor, #5214)."""
|
||
limit = max(1, min(20, _arg_number("limit", 10, int)))
|
||
hours = max(1, min(24, _arg_number("hours", 6, int)))
|
||
cpu_floor = max(0.1, _arg_number("cpu_floor", 1.0, float))
|
||
mem_floor = max(0.1, _arg_number("mem_floor", 1.0, float))
|
||
widget_id = request.args.get("wid", "0")
|
||
now = datetime.now(timezone.utc)
|
||
since = now - timedelta(minutes=CPU_AVERAGE_MINUTES)
|
||
slots, width = docker_svc.bar_slots(now, hours)
|
||
|
||
async with current_app.db_sessionmaker() as db:
|
||
# Dedup before the limit, else swarm tasks reported by both managers can
|
||
# fill the list with duplicates of the same container.
|
||
running = dedup_by_container_id([
|
||
c for c in await docker_svc.list_containers(db) if c.status == "running"])
|
||
average = {(h, n): v for h, n, v in (await db.execute(
|
||
select(DockerMetric.host_id, DockerMetric.container_name, func.avg(DockerMetric.cpu_pct))
|
||
.where(DockerMetric.scraped_at >= since)
|
||
.group_by(DockerMetric.host_id, DockerMetric.container_name)
|
||
)).all()}
|
||
|
||
def cpu_of(c):
|
||
v = average.get((c.host_id, c.name))
|
||
return c.cpu_pct if v is None else float(v)
|
||
|
||
busiest = sorted(running, key=lambda c: (-(cpu_of(c) or 0), c.name))[:limit]
|
||
history = await docker_svc.recent_usage(
|
||
db, {(c.host_id, c.name) for c in busiest}, slots[0], width)
|
||
hosts = await docker_svc.host_map(db, {c.host_id for c in busiest})
|
||
capacity = await docker_svc.host_capacity(db, {c.host_id for c in busiest})
|
||
|
||
def bars(c, i, scale, floor):
|
||
seen = history.get((c.host_id, c.name), {})
|
||
values = [None if seen.get(t) is None or seen[t][i] is None else seen[t][i] * scale
|
||
for t in slots]
|
||
return docker_svc.scaled_bars(values, floor)
|
||
|
||
rows_data = [{
|
||
"c": c,
|
||
"label": docker_svc.display_name(c),
|
||
"host_name": hosts[c.host_id].name if c.host_id in hosts else c.host_id,
|
||
"usage": docker_svc.usage(c, *capacity.get(c.host_id, (None, None)), cpu_pct=cpu_of(c)),
|
||
"cpu_bars": bars(c, 0, 0.01, cpu_floor), # cpu_pct → cores
|
||
"mem_bars": bars(c, 1, 1.0, mem_floor * 2**30), # bytes; floor in GiB
|
||
} for c in busiest]
|
||
return await render_template(
|
||
"docker/widget_resources.html",
|
||
rows=rows_data,
|
||
minutes=CPU_AVERAGE_MINUTES,
|
||
hours=hours, cpu_floor=cpu_floor, mem_floor=mem_floor * 2**30,
|
||
widget_id=widget_id,
|
||
)
|
||
|
||
|
||
def _arg_number(key, default, kind):
|
||
"""A numeric query argument, or `default` when missing or not a number."""
|
||
try:
|
||
return kind(request.args.get(key, default))
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
@docker_bp.get("/host/<host_id>")
|
||
@require_role(UserRole.viewer)
|
||
async def host_panel(host_id: str):
|
||
"""Per-host Docker fragment embedded on the core Hosts hub via HTMX:
|
||
problems first (with exit code and since when), then running, then
|
||
stopped. Swarm tasks Swarm has already replaced are history and left out.
|
||
|
||
Renders nothing when the host reports no containers, so hosts without Docker
|
||
don't carry an empty card.
|
||
"""
|
||
now = datetime.now(timezone.utc)
|
||
async with current_app.db_sessionmaker() as db:
|
||
containers = await docker_svc.list_containers(db, host_id)
|
||
if not containers:
|
||
return ""
|
||
services = list((await db.execute(select(DockerSwarmService))).scalars())
|
||
nodes = list((await db.execute(select(DockerSwarmNode))).scalars())
|
||
failures = await docker_svc.recent_failures_by_container(db, now - timedelta(hours=24))
|
||
capacity = await docker_svc.host_capacity(db, {host_id})
|
||
|
||
replaced = replaced_tasks(containers, services)
|
||
node_hostname = {n.node_id: n.hostname for n in nodes if n.hostname}
|
||
hour_ago = now - timedelta(hours=1)
|
||
crash = docker_svc.crash_loop_deaths(current_app.config)
|
||
rows = []
|
||
for c in containers:
|
||
if (c.host_id, c.name) in replaced:
|
||
continue
|
||
events = failures.get((c.host_id, c.name), [])
|
||
recent = sum(1 for e in events if e.at >= hour_ago)
|
||
problems = docker_svc.container_problems(c, recent, bool(events), crash)
|
||
rows.append({
|
||
"c": c, "label": docker_svc.task_label(c, node_hostname), "problems": problems,
|
||
"since": docker_svc.problem_since(problems, events, hour_ago),
|
||
"usage": docker_svc.usage(c, *capacity.get(host_id, (None, None))),
|
||
"uptime": _uptime(c.started_at) if c.status == "running" else None,
|
||
})
|
||
shown, folded = host_panel_rows(rows)
|
||
running = sum(1 for r in rows if r["c"].status == "running")
|
||
return await render_template(
|
||
"docker/host_panel.html",
|
||
rows=shown, folded=folded, says=docker_svc.PROBLEM_SAYS, severity=docker_svc.PROBLEM_SEVERITY,
|
||
running=running, stopped=len(rows) - running,
|
||
problems=sum(1 for r in rows if r["problems"]),
|
||
replaced=len(replaced), busiest=HOST_PANEL_BUSIEST,
|
||
)
|
||
|
||
|
||
# Running containers shown before the rest fold away on a host's page (#5276).
|
||
HOST_PANEL_BUSIEST = 8
|
||
|
||
|
||
def host_panel_rows(rows: list) -> tuple[list, list]:
|
||
"""(shown, folded) for a host's Docker panel: every container with a
|
||
problem, worst first; then the busiest running ones by CPU. The other
|
||
running containers (busiest first) and the stopped ones (by name) fold
|
||
away, so a host with dozens of containers keeps a short card."""
|
||
def cpu(r):
|
||
return getattr(r["c"], "cpu_pct", None) or 0.0
|
||
problems = sorted((r for r in rows if r["problems"]), key=lambda r: (
|
||
docker_svc.PROBLEMS.index(r["problems"][0]), r["label"].lower()))
|
||
running = sorted((r for r in rows if not r["problems"] and r["c"].status == "running"),
|
||
key=lambda r: (-cpu(r), r["label"].lower()))
|
||
stopped = sorted((r for r in rows if not r["problems"] and r["c"].status != "running"),
|
||
key=lambda r: r["label"].lower())
|
||
return (problems + running[:HOST_PANEL_BUSIEST],
|
||
running[HOST_PANEL_BUSIEST:] + stopped)
|
||
|
||
|
||
# Glyph + colour per lifecycle event, so the timeline reads at a glance.
|
||
_EVENT_STYLE = {
|
||
"start": ("▲", "var(--green)"),
|
||
"restart": ("↻", "var(--yellow)"),
|
||
"stop": ("■", "var(--text-muted)"),
|
||
"die": ("✕", "var(--red)"),
|
||
"oom": ("☠", "var(--red)"),
|
||
"health_change": ("◐", "var(--orange)"),
|
||
}
|
||
|
||
|
||
@docker_bp.get("/container/<host_id>/<name>")
|
||
@require_role(UserRole.viewer)
|
||
async def container_detail(host_id: str, name: str):
|
||
"""Full detail page for one container: facts, resource history, lifecycle."""
|
||
async with current_app.db_sessionmaker() as db:
|
||
c = await db.get(DockerContainer, (host_id, name))
|
||
host = await db.get(Host, host_id)
|
||
events = await docker_svc.container_events(db, host_id, name) if c is not None else []
|
||
node_name = None
|
||
if c is not None and c.node_id:
|
||
node_name = (await db.execute(
|
||
select(DockerSwarmNode.hostname)
|
||
.where(DockerSwarmNode.node_id == c.node_id, DockerSwarmNode.hostname != "")
|
||
.limit(1)
|
||
)).scalar()
|
||
cpus, mem_total = (await docker_svc.host_capacity(db, [host_id])).get(host_id, (None, None)) \
|
||
if c is not None else (None, None)
|
||
|
||
if c is None:
|
||
return await render_template(
|
||
"docker/container_detail.html",
|
||
container=None, host_id=host_id, name=name,
|
||
), 404
|
||
|
||
return await render_template(
|
||
"docker/container_detail.html",
|
||
container=c,
|
||
host=host,
|
||
host_id=host_id,
|
||
name=name,
|
||
label=docker_svc.task_label(c, {c.node_id: node_name} if node_name else None),
|
||
node_name=node_name,
|
||
usage=docker_svc.usage(c, cpus, mem_total),
|
||
exit_failed=docker_svc.exit_is_failure(c.exit_code, c.oom_killed),
|
||
ports=json.loads(c.ports_json) if c.ports_json else [],
|
||
uptime=_uptime(c.started_at) if c.status == "running" else None,
|
||
net_rx=human_bytes(c.net_rx_bytes), net_tx=human_bytes(c.net_tx_bytes),
|
||
blk_read=human_bytes(c.blk_read_bytes), blk_write=human_bytes(c.blk_write_bytes),
|
||
events=[
|
||
{"event": e.event, "detail": e.detail, "at": e.at,
|
||
"glyph": _EVENT_STYLE.get(e.event, ("•", "var(--text-muted)"))[0],
|
||
"colour": _EVENT_STYLE.get(e.event, ("•", "var(--text-muted)"))[1]}
|
||
for e in events
|
||
],
|
||
current_range=request.args.get("range", DEFAULT_RANGE),
|
||
)
|
||
|
||
|
||
@docker_bp.get("/swarm")
|
||
@require_role(UserRole.viewer)
|
||
async def swarm():
|
||
"""The Swarm page, laid out by node (#5303): each node's role, whether it
|
||
takes work, what its containers use against what the machine has, and the
|
||
services placed on it; services short of replicas first.
|
||
|
||
Swarm services/nodes are cluster-global — every manager's API returns the
|
||
same list — but each manager reports them independently (rows keyed by
|
||
host_id). So we group the reporting managers into swarms (managers that share
|
||
any node_id are the same cluster) and dedup within each: one row per service
|
||
(by name) and per node (by node_id), keeping the freshest. Without this, two
|
||
managers in one cluster list every service/node twice.
|
||
"""
|
||
async with current_app.db_sessionmaker() as db:
|
||
services = list((await db.execute(
|
||
select(DockerSwarmService).order_by(DockerSwarmService.service_name)
|
||
)).scalars())
|
||
nodes = list((await db.execute(
|
||
select(DockerSwarmNode).order_by(
|
||
DockerSwarmNode.role, DockerSwarmNode.hostname)
|
||
)).scalars())
|
||
hosts = await docker_svc.host_map(db, {s.host_id for s in services} | {n.host_id for n in nodes})
|
||
containers = await docker_svc.list_containers(db)
|
||
# The Steward host on each node: the one whose agent reports its
|
||
# containers, else the host named as the node is.
|
||
node_host = {c.node_id: c.host_id for c in containers if c.node_id}
|
||
unmatched = {(n.hostname or "").lower() for n in nodes if n.node_id not in node_host} - {""}
|
||
if unmatched:
|
||
by_name = {h.name.lower(): h for h in (await db.execute(
|
||
select(Host).where(func.lower(Host.name).in_(unmatched)))).scalars()}
|
||
for n in nodes:
|
||
h = by_name.get((n.hostname or "").lower())
|
||
if n.node_id not in node_host and h is not None:
|
||
node_host[n.node_id] = h.id
|
||
hosts.update(await docker_svc.host_map(db, set(node_host.values()) - set(hosts)))
|
||
capacity = await docker_svc.host_capacity(db, set(node_host.values()))
|
||
|
||
# ── Group reporting managers into swarms by shared node membership ──────────
|
||
manager_ids = {s.host_id for s in services} | {n.host_id for n in nodes}
|
||
nodes_by_mgr: dict[str, set[str]] = {}
|
||
for n in nodes:
|
||
nodes_by_mgr.setdefault(n.host_id, set()).add(n.node_id)
|
||
|
||
parent = {m: m for m in manager_ids}
|
||
|
||
def _find(x: str) -> str:
|
||
while parent[x] != x:
|
||
parent[x] = parent[parent[x]]
|
||
x = parent[x]
|
||
return x
|
||
|
||
mgrs = list(manager_ids)
|
||
for i in range(len(mgrs)):
|
||
for j in range(i + 1, len(mgrs)):
|
||
a, b = mgrs[i], mgrs[j]
|
||
if nodes_by_mgr.get(a) and nodes_by_mgr.get(b) and (nodes_by_mgr[a] & nodes_by_mgr[b]):
|
||
parent[_find(a)] = _find(b)
|
||
|
||
swarm_members: dict[str, set[str]] = {}
|
||
for m in manager_ids:
|
||
swarm_members.setdefault(_find(m), set()).add(m)
|
||
|
||
# ── Dedup services (by name) + nodes (by node_id) within each swarm ─────────
|
||
swarms = []
|
||
for members in swarm_members.values():
|
||
svc_by_name: dict[str, DockerSwarmService] = {}
|
||
for s in services:
|
||
if s.host_id in members and (
|
||
s.service_name not in svc_by_name
|
||
or s.updated_at > svc_by_name[s.service_name].updated_at
|
||
):
|
||
svc_by_name[s.service_name] = s
|
||
node_by_id: dict[str, DockerSwarmNode] = {}
|
||
for n in nodes:
|
||
if n.host_id in members and (
|
||
n.node_id not in node_by_id
|
||
or n.updated_at > node_by_id[n.node_id].updated_at
|
||
):
|
||
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 svc_by_name.values():
|
||
tasks = parse_tasks(s)
|
||
summary = task_summary(tasks, now, deaths, node_names)
|
||
state = ("healthy" if s.running >= s.desired and s.desired > 0
|
||
else "degraded" if s.running > 0 else "down")
|
||
placement = []
|
||
for p in (json.loads(s.placement_json) if s.placement_json else []):
|
||
nid = p.get("node_id", "")
|
||
placement.append({
|
||
"node": node_names.get(nid, nid[:12] or "?"),
|
||
"running": p.get("running", 0),
|
||
})
|
||
svc_dicts.append({
|
||
"name": s.service_name, "mode": s.mode,
|
||
"running": s.running, "desired": s.desired, "image": s.image,
|
||
"healthy": s.running >= s.desired and s.desired > 0,
|
||
"degraded": 0 < s.running < s.desired,
|
||
"down": s.running == 0 and s.desired > 0,
|
||
"state": state,
|
||
"placement": placement,
|
||
**summary,
|
||
"trouble_since": trouble_since(tasks, state, summary["crash_looping"], now),
|
||
})
|
||
manager_names = sorted(hosts[m].name if m in hosts else m for m in members)
|
||
swarms.append({
|
||
"managers": manager_names,
|
||
"services_total": len(svc_dicts),
|
||
"services_healthy": sum(d["state"] == "healthy" for d in svc_dicts),
|
||
"troubled": failing_services(svc_dicts),
|
||
"nodes": node_cards(
|
||
node_by_id.values(), svc_by_name.values(), containers, node_host=node_host,
|
||
host_names={h.id: h.name for h in hosts.values()}, capacity=capacity,
|
||
service_state={d["name"]: d["state"] for d in svc_dicts}),
|
||
})
|
||
|
||
swarms.sort(key=lambda g: g["managers"][0].lower() if g["managers"] else "")
|
||
return await render_template("docker/swarm.html", swarms=swarms)
|
||
|
||
|
||
# How many images a host agent reports, largest first (host_agent agent.py
|
||
# _disk_images caps its list here); a list this long may be missing smaller ones.
|
||
IMAGES_REPORTED = 50
|
||
|
||
|
||
@docker_bp.get("/disk")
|
||
@require_role(UserRole.viewer)
|
||
async def disk():
|
||
"""Image/disk usage page: reclaimable space, per-image sizes, stopped count.
|
||
|
||
Admins can reclaim space per host via the prune buttons, which fire the
|
||
audited bundled prune playbook through Ansible (m78) — the collection agent
|
||
itself stays read-only.
|
||
"""
|
||
from steward.core.capabilities import has_capability
|
||
from steward.models.ansible_inventory import AnsibleTarget
|
||
|
||
async with current_app.db_sessionmaker() as db:
|
||
summaries = list((await db.execute(select(DockerDiskUsage))).scalars())
|
||
images = list((await db.execute(
|
||
select(DockerImage).order_by(DockerImage.size.desc())
|
||
)).scalars())
|
||
# Stopped-container count per host, surfaced alongside reclaimable space.
|
||
stopped_rows = (await db.execute(
|
||
select(DockerContainer.host_id)
|
||
.where(DockerContainer.status != "running")
|
||
)).scalars()
|
||
stopped_by_host: dict[str, int] = {}
|
||
for hid in stopped_rows:
|
||
stopped_by_host[hid] = stopped_by_host.get(hid, 0) + 1
|
||
hosts = await docker_svc.host_map(db, {s.host_id for s in summaries})
|
||
# Which hosts have a linked Ansible target — gates the prune buttons
|
||
# (a target is required to route the playbook run to that host).
|
||
host_ids = {s.host_id for s in summaries}
|
||
target_by_host: dict[str, str] = {}
|
||
if host_ids:
|
||
for hid, tid in (await db.execute(
|
||
select(AnsibleTarget.host_id, AnsibleTarget.id)
|
||
.where(AnsibleTarget.host_id.in_(host_ids))
|
||
)).all():
|
||
target_by_host[hid] = tid
|
||
|
||
images_by_host: dict[str, list] = {}
|
||
for im in images:
|
||
images_by_host.setdefault(im.host_id, []).append({
|
||
"repo_tag": None if im.repo_tag in ("<none>", "", None) else im.repo_tag,
|
||
"image_id": im.image_id, "size": human_bytes(im.size),
|
||
"shared": human_bytes(im.shared_size), "containers": im.containers,
|
||
"reclaimable": im.containers == 0,
|
||
})
|
||
|
||
host_groups = [
|
||
{
|
||
"host_id": s.host_id,
|
||
"host_name": hosts[s.host_id].name if s.host_id in hosts else s.host_id,
|
||
"as_of": s.updated_at,
|
||
"summary": {
|
||
"images_total": s.images_total, "images_active": s.images_active,
|
||
"images_size": human_bytes(s.images_size),
|
||
"images_reclaimable": human_bytes(s.images_reclaimable),
|
||
"layers_size": human_bytes(s.layers_size),
|
||
"containers_count": s.containers_count,
|
||
"containers_size": human_bytes(s.containers_size),
|
||
"volumes_count": s.volumes_count,
|
||
"volumes_size": human_bytes(s.volumes_size),
|
||
"build_cache_size": human_bytes(s.build_cache_size),
|
||
},
|
||
"stopped": stopped_by_host.get(s.host_id, 0),
|
||
"images": images_by_host.get(s.host_id, []),
|
||
# The agent reports the largest IMAGES_REPORTED images (agent._disk_images).
|
||
"images_capped": len(images_by_host.get(s.host_id, [])) >= IMAGES_REPORTED,
|
||
"has_target": s.host_id in target_by_host,
|
||
}
|
||
for s in summaries
|
||
]
|
||
host_groups.sort(key=lambda g: g["host_name"].lower())
|
||
return await render_template(
|
||
"docker/disk.html", host_groups=host_groups,
|
||
ansible_available=has_capability("ansible.run_playbook"),
|
||
)
|
||
|
||
|
||
@docker_bp.post("/disk/<host_id>/prune")
|
||
@require_role(UserRole.admin)
|
||
async def disk_prune(host_id: str):
|
||
"""Reclaim Docker disk on a host by firing the bundled prune playbook via
|
||
Ansible (audited, admin-gated) — the collection agent stays read-only (m78).
|
||
`target` selects the scope: containers | images | system.
|
||
"""
|
||
from steward.core.capabilities import has_capability, invoke_capability
|
||
from steward.ansible.sources import BUILTIN_SOURCE_NAME
|
||
from steward.models.ansible_inventory import AnsibleTarget
|
||
|
||
if not has_capability("ansible.run_playbook"):
|
||
return _error(400, "ansible_unavailable", "Ansible is not available")
|
||
|
||
form = await request.form
|
||
prune_target = (form.get("target", "") or "").strip()
|
||
if prune_target not in docker_svc.PRUNE_TARGETS:
|
||
return _error(400, "bad_target", "Unknown prune target")
|
||
|
||
async with current_app.db_sessionmaker() as db:
|
||
target = (await db.execute(
|
||
select(AnsibleTarget).where(AnsibleTarget.host_id == host_id)
|
||
)).scalar_one_or_none()
|
||
if target is None:
|
||
return _error(400, "no_target",
|
||
"Link an Ansible target to this host before pruning")
|
||
|
||
# extra_vars_map outranks the playbook's own `vars:` defaults.
|
||
extra_vars = docker_svc.prune_extra_vars(prune_target)
|
||
|
||
actor_role = UserRole(session.get("user_role", "viewer"))
|
||
run, _source, err = await invoke_capability(
|
||
"ansible.run_playbook", actor_role,
|
||
current_app._get_current_object(), # type: ignore[attr-defined]
|
||
source_name=BUILTIN_SOURCE_NAME,
|
||
playbook_path="maintenance/docker_prune.yml",
|
||
inventory_scope=f"steward:target:{target.id}",
|
||
params={"extra_vars_map": extra_vars},
|
||
triggered_by=session.get("user_id"),
|
||
)
|
||
if err:
|
||
return _error(400, "prune_failed", err)
|
||
return redirect(url_for("ansible.run_detail", run_id=run.id))
|
||
|
||
|
||
@docker_bp.get("/container/<host_id>/<name>/history")
|
||
@require_role(UserRole.viewer)
|
||
async def container_history(host_id: str, name: str):
|
||
"""HTMX fragment: CPU + memory sparklines for the selected time range."""
|
||
since, range_key = parse_range(request.args.get("range"))
|
||
async with current_app.db_sessionmaker() as db:
|
||
cpu_hist, mem_hist = await docker_svc.container_history(db, host_id, name, since)
|
||
return await render_template(
|
||
"docker/_container_history.html",
|
||
sparkline_cpu=_sparkline(cpu_hist, width=320, height=48),
|
||
sparkline_mem=_sparkline(mem_hist, width=320, height=48),
|
||
have_data=len(cpu_hist) >= 2,
|
||
range_key=range_key,
|
||
)
|
||
|
||
|
||
# Newest lines returned per fetch — the retained window is bounded by the ring,
|
||
# but a container can still hold thousands of lines; cap what one fetch renders.
|
||
_LOG_TAIL_DEFAULT = 500
|
||
|
||
|
||
@docker_bp.get("/container/<host_id>/<name>/logs")
|
||
@require_role(UserRole.viewer)
|
||
async def container_logs(host_id: str, name: str):
|
||
"""Full log-viewer page for one container (lines load via an HTMX fragment)."""
|
||
async with current_app.db_sessionmaker() as db:
|
||
host = await db.get(Host, host_id)
|
||
return await render_template(
|
||
"docker/container_logs.html", host=host, host_id=host_id, name=name,
|
||
)
|
||
|
||
|
||
@docker_bp.get("/container/<host_id>/<name>/logs/lines")
|
||
@require_role(UserRole.viewer)
|
||
async def container_logs_lines(host_id: str, name: str):
|
||
"""HTMX fragment: the recent retained log lines, stream + text filtered.
|
||
Polled every few seconds by the viewer for near-live follow."""
|
||
stream = request.args.get("stream", "all")
|
||
query = (request.args.get("q") or "").strip()
|
||
async with current_app.db_sessionmaker() as db:
|
||
rows = await docker_svc.query_logs(db, host_id, name, stream, query, _LOG_TAIL_DEFAULT)
|
||
return await render_template(
|
||
"docker/_container_logs_lines.html",
|
||
lines=[{"ts": r.ts, "stream": r.stream, "line": r.line} for r in rows],
|
||
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/<name>")
|
||
@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/<name>/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/<name>/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,
|
||
)
|