# plugins/docker/routes.py
from __future__ import annotations
import json
from datetime import datetime, timedelta, timezone
from quart import Blueprint, current_app, render_template, request
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.time_range import parse_range, DEFAULT_RANGE
from .models import (
DockerContainer, DockerDiskUsage, DockerEvent, DockerImage, DockerMetric,
DockerSwarmNode, DockerSwarmService,
)
docker_bp = Blueprint("docker", __name__, template_folder="templates")
def _human_bytes(n: int | None) -> str:
"""Compact binary size string (e.g. '1.4 GiB', '512 MiB', '0 B')."""
if n is None:
return "—"
size = float(n)
for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
if abs(size) < 1024.0 or unit == "TiB":
if unit == "B":
return f"{int(size)} {unit}"
return f"{size:.1f} {unit}"
size /= 1024.0
return f"{size:.1f} TiB"
def _human_uptime(started_at: datetime | None) -> str | None:
"""Compact 'how long running' string (e.g. '3d 4h', '5h 12m', '8m')."""
if started_at is None:
return None
if started_at.tzinfo is None:
started_at = started_at.replace(tzinfo=timezone.utc)
secs = int((datetime.now(timezone.utc) - started_at).total_seconds())
if secs < 0:
return None
d, rem = divmod(secs, 86400)
h, rem = divmod(rem, 3600)
m, _ = divmod(rem, 60)
if d:
return f"{d}d {h}h"
if h:
return f"{h}h {m}m"
return f"{m}m"
def _sparkline(values: list[float], width: int = 80, height: int = 20) -> str:
if len(values) < 2:
return f''
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''
)
def _bucket(values: list[float], target: int = 40) -> list[float]:
"""Bucket-average a series down to ~target points (DB-agnostic, in Python).
Replaces the old SQL strftime() bucketing, which was SQLite-only and broke
on Postgres. Agent cadence is dense, so a multi-hour window is hundreds of
rows — averaging into a readable point count keeps the sparkline's shape.
"""
n = len(values)
if n <= target:
return values
size = (n + target - 1) // target
out: list[float] = []
for i in range(0, n, size):
chunk = values[i:i + size]
out.append(sum(chunk) / len(chunk))
return out
async def _container_history(db, host_id: str, name: str, since) -> tuple[list, list]:
"""Return (cpu_series, mem_series) sparkline-ready for one host's container."""
rows = (await db.execute(
select(DockerMetric.cpu_pct, DockerMetric.mem_pct)
.where(DockerMetric.host_id == host_id)
.where(DockerMetric.container_name == name)
.where(DockerMetric.scraped_at >= since)
.order_by(DockerMetric.scraped_at)
)).all()
cpu = _bucket([r.cpu_pct or 0.0 for r in rows])
mem = _bucket([r.mem_pct or 0.0 for r in rows])
return cpu, mem
@docker_bp.get("/")
@require_role(UserRole.viewer)
async def index():
poll_interval = current_app.config.get("MONITORS_POLL_INTERVAL", 60)
current_range = request.args.get("range", DEFAULT_RANGE)
# 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
return await render_template(
"docker/index.html",
poll_interval=poll_interval,
current_range=current_range,
has_swarm=has_swarm,
has_disk=has_disk,
)
async def _host_map(db, host_ids: set[str]) -> dict[str, Host]:
if not host_ids:
return {}
return {
h.id: h for h in (await db.execute(
select(Host).where(Host.id.in_(host_ids)))).scalars().all()
}
@docker_bp.get("/rows")
@require_role(UserRole.viewer)
async def rows():
"""HTMX fragment: containers grouped by host, with resource sparklines."""
since, range_key = parse_range(request.args.get("range"))
async with current_app.db_sessionmaker() as db:
containers = list((await db.execute(
select(DockerContainer).order_by(
(DockerContainer.status == "running").desc(),
DockerContainer.name,
)
)).scalars())
hosts = await _host_map(db, {c.host_id for c in containers})
# Group by host so each container is clearly attributed to the box it
# runs on (names are only unique within a host now).
groups: dict[str, dict] = {}
for c in containers:
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,
}
cpu_hist, mem_hist = await _container_history(db, c.host_id, c.name, since)
g["containers"].append({
"container": c,
"ports": json.loads(c.ports_json) if c.ports_json else [],
"uptime": _human_uptime(c.started_at) if c.status == "running" else None,
"sparkline_cpu": _sparkline(cpu_hist),
"sparkline_mem": _sparkline(mem_hist),
})
if c.status == "running":
g["running"] += 1
else:
g["stopped"] += 1
# Sub-group each host's containers by compose project (or swarm service),
# preserving the running-first ordering. Containers with neither label 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 c.service_name 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())
running = sum(g["running"] for g in host_groups)
total = len(containers)
return await render_template(
"docker/rows.html",
host_groups=host_groups,
running=running,
stopped=total - running,
total=total,
range_key=range_key,
)
def _dedup_by_container_id(containers: list) -> list:
"""Collapse containers double-reported across hosts. Swarm-aware agents on
every manager list cluster-wide tasks, so the same container (identical
container_id) arrives once per manager and would otherwise be counted N times.
Keep the first occurrence per non-empty container_id — callers order
running-first, so a running report wins over a stale stopped one. Rows without
a container_id (older agents) can't be matched, so they're kept as-is."""
seen: set[str] = set()
out = []
for c in containers:
cid = (c.container_id or "").strip()
if cid:
if cid in seen:
continue
seen.add(cid)
out.append(c)
return out
@docker_bp.get("/widget")
@require_role(UserRole.viewer)
async def widget():
"""HTMX dashboard widget: container status overview, grouped by host."""
show_stopped = request.args.get("show_stopped", "no") == "yes"
widget_id = request.args.get("wid", "0")
since_24h = datetime.now(timezone.utc) - timedelta(hours=24)
async with current_app.db_sessionmaker() as db:
all_containers = _dedup_by_container_id(list((await db.execute(
select(DockerContainer).order_by(
(DockerContainer.status == "running").desc(),
DockerContainer.name,
)
)).scalars()))
# Recent failures are more actionable than a static "stopped" tally. Count
# 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)))
.where(DockerEvent.event.in_(("die", "oom")), DockerEvent.at >= since_24h)
)).scalar() or 0
hosts = await _host_map(db, {c.host_id for c in all_containers})
running = [c for c in all_containers if c.status == "running"]
display = all_containers if show_stopped else running
# Group for display so multi-host fleets read clearly; single-host stays flat.
groups: dict[str, dict] = {}
for c in display:
g = groups.setdefault(c.host_id, {
"host_id": c.host_id,
"host_name": hosts[c.host_id].name if c.host_id in hosts else c.host_id,
"containers": [],
})
g["containers"].append(c)
host_groups = sorted(groups.values(), key=lambda g: g["host_name"].lower())
return await render_template(
"docker/widget.html",
host_groups=host_groups,
multi_host=len(host_groups) > 1,
running_count=len(running),
failed_24h=failed_24h,
total_count=len(all_containers),
show_stopped=show_stopped,
widget_id=widget_id,
)
@docker_bp.get("/widget/resources")
@require_role(UserRole.viewer)
async def widget_resources():
"""HTMX dashboard widget: CPU + memory usage for the busiest containers."""
limit = max(1, min(20, int(request.args.get("limit", 10) or 10)))
widget_id = request.args.get("wid", "0")
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.
containers = _dedup_by_container_id(list((await db.execute(
select(DockerContainer)
.where(DockerContainer.status == "running")
.order_by(DockerContainer.cpu_pct.desc().nullslast())
)).scalars()))[:limit]
hosts = await _host_map(db, {c.host_id for c in containers})
rows_data = [
{"c": c, "host_name": hosts[c.host_id].name if c.host_id in hosts else c.host_id}
for c in containers
]
return await render_template(
"docker/widget_resources.html",
rows=rows_data,
multi_host=len({c.host_id for c in containers}) > 1,
widget_id=widget_id,
)
@docker_bp.get("/host/")
@require_role(UserRole.viewer)
async def host_panel(host_id: str):
"""Per-host Docker fragment embedded on the core Hosts hub via HTMX.
Renders nothing when the host reports no containers, so hosts without Docker
don't carry an empty card.
"""
async with current_app.db_sessionmaker() as db:
containers = list((await db.execute(
select(DockerContainer)
.where(DockerContainer.host_id == host_id)
.order_by(
(DockerContainer.status == "running").desc(),
DockerContainer.name,
)
)).scalars())
if not containers:
return ""
running = sum(1 for c in containers if c.status == "running")
return await render_template(
"docker/host_panel.html",
containers=containers,
running=running,
stopped=len(containers) - running,
)
# Glyph + colour per lifecycle event, so the timeline reads at a glance.
_EVENT_STYLE = {
"start": ("▲", "var(--green)"),
"stop": ("■", "var(--text-muted)"),
"die": ("✕", "var(--red)"),
"oom": ("☠", "var(--red)"),
"health_change": ("◐", "var(--orange)"),
}
@docker_bp.get("/container//")
@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 = []
if c is not None:
events = list((await db.execute(
select(DockerEvent)
.where(DockerEvent.host_id == host_id)
.where(DockerEvent.container_name == name)
.order_by(DockerEvent.at.desc())
.limit(50)
)).scalars())
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,
ports=json.loads(c.ports_json) if c.ports_json else [],
uptime=_human_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():
"""Swarm topology page: services with replica health, nodes, placement.
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 _host_map(db, {s.host_id for s in services} | {n.host_id for n in nodes})
# ── 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()}
svc_dicts = []
for s in sorted(svc_by_name.values(), key=lambda s: s.service_name):
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,
"placement": placement,
})
node_rows = sorted(
node_by_id.values(), key=lambda n: (n.role, (n.hostname or n.node_id).lower())
)
manager_names = sorted(hosts[m].name if m in hosts else m for m in members)
swarms.append({
"managers": manager_names,
"services": svc_dicts,
"nodes": node_rows,
})
swarms.sort(key=lambda g: g["managers"][0].lower() if g["managers"] else "")
return await render_template("docker/swarm.html", swarms=swarms)
@docker_bp.get("/disk")
@require_role(UserRole.viewer)
async def disk():
"""Image/disk usage page: reclaimable space, per-image sizes, stopped count.
Read-only — prune actions are deferred to the cleanup-actions milestone, so
this surfaces the numbers and notes where reclaim lives.
"""
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 _host_map(db, {s.host_id for s in summaries})
images_by_host: dict[str, list] = {}
for im in images:
images_by_host.setdefault(im.host_id, []).append({
"repo_tag": im.repo_tag, "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,
"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, []),
}
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)
@docker_bp.get("/container///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 _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,
)