bvandeusenandClaude Opus 5.5
d4974ddf6c
Docker panel (#5276): problems first, then the 8 busiest running containers by CPU; the other running and the stopped fold behind one "+N more running · M stopped" line. Containers that are not swarm tasks read the way the Docker page reads them: CI jobs as "CI: workflow · job", buildx builders as "BuildKit builder · name". Monitors card (#5277): the edit/delete icon buttons and the inline add form are gone. Each row links to its monitor's page; "Add monitor" opens the add form filed under this host, starting from its address, with the host in its breadcrumbs and Cancel back to it. Agent card (#5278): Full metrics stays in view; Update, Rotate token, Unregister agent (was "Remove") and Re-provision sit under "Manage agent". Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
596 lines
26 KiB
Python
596 lines
26 KiB
Python
"""Docker queries shared by the plugin's pages and its MCP tools.
|
||
|
||
Each function takes an open session and returns rows or plain data, with no
|
||
Quart request context, so the routes and the /mcp tools (which run outside
|
||
Quart's pipeline) call the same code.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from sqlalchemy import func, select
|
||
|
||
from steward.core.time_range import bucket_seconds
|
||
from steward.models.hosts import Host
|
||
|
||
from .dedup import dedup_by_container_id
|
||
from .models import (
|
||
DockerContainer, DockerEvent, DockerLog, DockerMetric, DockerMetricHourly, DockerSwarmService,
|
||
)
|
||
from .swarm_view import replaced_tasks
|
||
|
||
|
||
CRASH_LOOP_DEATHS = 3 # default; Settings → Thresholds sets it
|
||
|
||
|
||
def crash_loop_deaths(config) -> int:
|
||
"""Deaths within an hour that make a container or a swarm service a crash
|
||
loop, from Settings → Thresholds (app.config["THRESHOLDS"])."""
|
||
th = (config.get("THRESHOLDS") or {}).get("crash_loop") or {}
|
||
return int(th.get("warn") or CRASH_LOOP_DEATHS)
|
||
|
||
|
||
# Exit codes a deliberate stop leaves: SIGTERM (143), and SIGKILL (137) when
|
||
# the process outlives the stop timeout. Without an OOM kill they are someone
|
||
# stopping the container, not the container failing.
|
||
STOP_EXITS = (137, 143)
|
||
|
||
|
||
def exit_is_failure(exit_code, oom_killed: bool = False) -> bool:
|
||
"""Whether a container's exit is a failure: an OOM kill, or a non-zero
|
||
exit other than a stop's."""
|
||
return bool(oom_killed) or exit_code not in (None, 0, *STOP_EXITS)
|
||
|
||
|
||
def _event_exit(detail) -> int | None:
|
||
"""The exit code in a die event's detail ("exit 1"), or None."""
|
||
text = (detail or "").strip()
|
||
if text.startswith("exit "):
|
||
try:
|
||
return int(text[5:])
|
||
except ValueError:
|
||
return None
|
||
return None
|
||
|
||
|
||
def event_is_failure(e) -> bool:
|
||
"""Whether a die/oom event is a failure. An oom always is; a die is unless
|
||
its exit is a stop's (or clean)."""
|
||
if e.event == "oom":
|
||
return True
|
||
return e.event == "die" and exit_is_failure(_event_exit(e.detail))
|
||
|
||
|
||
def failures_24h(events, swarm_services, host_name: dict) -> list[dict]:
|
||
"""What failed in the last 24h, newest first, for the dashboard count and
|
||
the Docker page's list of it: one row per plain container that died or was
|
||
OOM-killed, and one per swarm service with failed tasks.
|
||
|
||
`events` are die/oom events of the last 24h. A swarm task's container dies
|
||
whenever Swarm replaces it, so a service with Swarm's task records counts
|
||
its failed tasks instead of its containers' deaths; a service without them
|
||
(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}.
|
||
"""
|
||
by_prefix = {s["name"] + ".": s for s in swarm_services}
|
||
rows: dict[tuple, dict] = {}
|
||
# Each manager reports every swarm task, so one task's death can arrive
|
||
# once per manager: a service counts each dead task container once.
|
||
task_deaths: dict[tuple, set] = {}
|
||
for e in sorted(events, key=lambda e: e.at, reverse=True):
|
||
if not event_is_failure(e):
|
||
continue
|
||
svc = next((s for p, s in by_prefix.items() if e.container_name.startswith(p)), None)
|
||
if svc is not None and svc["tasks_known"]:
|
||
continue
|
||
kind, name = ("service", svc["name"]) if svc else ("container", e.container_name)
|
||
key = (kind, name) if svc else (kind, e.host_id, name)
|
||
if svc is not None:
|
||
seen = task_deaths.setdefault(key, set())
|
||
if e.container_name in seen:
|
||
continue
|
||
seen.add(e.container_name)
|
||
row = rows.get(key)
|
||
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}
|
||
else:
|
||
row["count"] += 1
|
||
from .swarm_view import task_time
|
||
for s in swarm_services:
|
||
if s["tasks_known"] and s["failures_24h"]:
|
||
err = s["last_error"] or {}
|
||
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")}
|
||
epoch = datetime.min.replace(tzinfo=timezone.utc)
|
||
return sorted(rows.values(), key=lambda r: r["at"] or epoch, reverse=True)
|
||
|
||
|
||
def task_label(c, node_hostname: dict | None = None) -> str:
|
||
"""How a container reads to a person. A swarm task is its service and
|
||
slot: "codex_db · 1" for a replica, "agent · node-3" for a global
|
||
service's task on a node. Anything else as display_name reads it."""
|
||
from .swarm_view import _slot_key
|
||
svc = getattr(c, "service_name", None)
|
||
if not svc:
|
||
return display_name(c)
|
||
slot = _slot_key(c)
|
||
if slot.isdigit():
|
||
return f"{svc} · {slot}"
|
||
host = (node_hostname or {}).get(slot) or (node_hostname or {}).get(getattr(c, "node_id", None))
|
||
return f"{svc} · {host}" if host else svc
|
||
|
||
|
||
_CI_JOB = re.compile(
|
||
r"^(?:GITEA|FORGEJO)-ACTIONS-TASK-\d+-WORKFLOW-(?P<workflow>.+?)-JOB-(?P<job>.+?)(?:-[0-9a-f]{16,})?$")
|
||
# docker buildx's container driver names its node "buildx_buildkit_<builder><n>".
|
||
_BUILDKIT = re.compile(r"^buildx_buildkit_(?P<builder>.+?)\d*$")
|
||
|
||
|
||
def display_name(c) -> str:
|
||
"""A container's name as a list can show it in one short line (#5253): a
|
||
swarm task by its service (the host says where it runs), a Gitea or
|
||
Forgejo Actions job container as "CI: <workflow> · <job>", a buildx
|
||
builder as "BuildKit builder · <name>", anything else by its own name."""
|
||
svc = getattr(c, "service_name", None)
|
||
if svc:
|
||
return svc
|
||
m = _CI_JOB.match(c.name or "")
|
||
if m:
|
||
return f"CI: {m['workflow']} · {m['job']}"
|
||
m = _BUILDKIT.match(c.name or "")
|
||
if m:
|
||
return f"BuildKit builder · {m['builder']}"
|
||
return c.name
|
||
|
||
|
||
def usage(c, cpus: int | None, mem_total: int | None, cpu_pct: float | None = None) -> dict:
|
||
"""A container's CPU and memory in terms a reader can weigh.
|
||
|
||
Docker's CPU% is per core (100% = one core busy), so it is shown as cores
|
||
and graded as a share of the host's CPUs when the host's core count is
|
||
known. Docker's memory limit is the host's RAM when none is set, so memory
|
||
is shown in bytes, against a limit only when one is actually set; mem_pct
|
||
is then a share of that limit, otherwise of the host's memory. `cpu_pct`
|
||
replaces the container's last sample (an average, say) when given.
|
||
"""
|
||
cpu = getattr(c, "cpu_pct", None) if cpu_pct is None else cpu_pct
|
||
limit = getattr(c, "mem_limit_bytes", None)
|
||
limited = bool(limit and mem_total and limit < mem_total * 0.95)
|
||
return {
|
||
"cores": None if cpu is None else cpu / 100.0,
|
||
"cpu_share": None if cpu is None or not cpus else cpu / cpus,
|
||
"cpus": cpus,
|
||
"mem_bytes": getattr(c, "mem_usage_bytes", None),
|
||
"mem_limit": limit if limited else None,
|
||
"mem_pct": getattr(c, "mem_pct", None),
|
||
"mem_of": "its limit" if limited else ("the host's memory" if mem_total else None),
|
||
}
|
||
|
||
|
||
async def host_capacity(db, host_ids) -> dict[str, tuple[int | None, int | None]]:
|
||
"""{host_id: (CPU cores, memory bytes)} from the host agent's latest
|
||
readings: cores counted from its per-core CPU series, memory from
|
||
available ÷ (1 − used%). (None, None) for a host without an agent."""
|
||
from sqlalchemy import or_
|
||
from steward.models.metrics import PluginMetricLatest
|
||
hosts = await host_map(db, set(host_ids))
|
||
if not hosts:
|
||
return {}
|
||
names = {h.name: hid for hid, h in hosts.items()}
|
||
rows = (await db.execute(
|
||
select(PluginMetricLatest.resource_name, PluginMetricLatest.metric_name,
|
||
PluginMetricLatest.value)
|
||
.where(PluginMetricLatest.source_module == "host_agent",
|
||
or_(PluginMetricLatest.resource_name.in_(list(names)),
|
||
*(PluginMetricLatest.resource_name.like(escape_like(n) + ":core%", escape="\\")
|
||
for n in names)))
|
||
)).all()
|
||
cores: dict[str, int] = {}
|
||
mem: dict[str, dict] = {}
|
||
for res, metric, value in rows:
|
||
if res in names:
|
||
mem.setdefault(res, {})[metric] = value
|
||
elif metric == "cpu_pct":
|
||
host = res.rsplit(":core", 1)[0]
|
||
cores[host] = cores.get(host, 0) + 1
|
||
out = {}
|
||
for name, hid in names.items():
|
||
m = mem.get(name, {})
|
||
avail, used = m.get("mem_available_bytes"), m.get("mem_used_pct")
|
||
total = int(avail / (1 - used / 100)) if avail and used is not None and used < 99.5 else None
|
||
out[hid] = (cores.get(name), total)
|
||
return out
|
||
|
||
|
||
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
|
||
|
||
|
||
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()
|
||
}
|
||
|
||
|
||
def escape_like(text: str) -> str:
|
||
"""Make `text` match literally inside a LIKE/ILIKE pattern (escape char \\)."""
|
||
return text.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
||
|
||
|
||
async def list_containers(db, host_id: str | None = None) -> list[DockerContainer]:
|
||
"""Containers, running first then by name; one host's when host_id is given."""
|
||
stmt = select(DockerContainer)
|
||
if host_id is not None:
|
||
stmt = stmt.where(DockerContainer.host_id == host_id)
|
||
stmt = stmt.order_by((DockerContainer.status == "running").desc(), DockerContainer.name)
|
||
return list((await db.execute(stmt)).scalars())
|
||
|
||
|
||
async def container_events(db, host_id: str, name: str, limit: int = 50) -> list[DockerEvent]:
|
||
"""A container's lifecycle events (start/stop/die/oom/health), newest first."""
|
||
return list((await db.execute(
|
||
select(DockerEvent)
|
||
.where(DockerEvent.host_id == host_id)
|
||
.where(DockerEvent.container_name == name)
|
||
.order_by(DockerEvent.at.desc())
|
||
.limit(limit)
|
||
)).scalars())
|
||
|
||
|
||
async def query_logs(db, host_id: str, name: str, stream: str, query: str,
|
||
limit: int) -> list:
|
||
"""The recent retained lines for one container, newest-first, optionally
|
||
filtered by stream and a case-insensitive substring.
|
||
|
||
Newest-first is deliberate: the viewer replaces the list on each poll, which
|
||
resets scroll to the top — so the latest lines stay visible without any
|
||
scroll handling, and older lines are a scroll away.
|
||
"""
|
||
stmt = (select(DockerLog.ts, DockerLog.stream, DockerLog.line)
|
||
.where(DockerLog.host_id == host_id)
|
||
.where(DockerLog.container_name == name))
|
||
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()
|
||
|
||
|
||
CONTAINER_SERIES = ("cpu_pct", "mem_pct", "mem_usage_bytes")
|
||
|
||
|
||
async def container_points(db, host_id: str, name: str, metric: str, since,
|
||
target: int = 120) -> list[tuple]:
|
||
"""(time, average) points of one container metric since `since`, oldest first.
|
||
|
||
Raw samples are bucket-averaged in SQL; the part of the range older than
|
||
the oldest raw sample (already rolled up and pruned by retention) comes
|
||
from the hourly table.
|
||
"""
|
||
if metric not in CONTAINER_SERIES:
|
||
raise ValueError(f"metric must be one of {', '.join(CONTAINER_SERIES)}")
|
||
raw_col = getattr(DockerMetric, metric)
|
||
width_s = bucket_seconds(since, target)
|
||
rbucket = func.date_bin(
|
||
func.make_interval(0, 0, 0, 0, 0, 0, width_s), DockerMetric.scraped_at, func.to_timestamp(0),
|
||
).label("bucket")
|
||
raw = (await db.execute(
|
||
select(rbucket, func.avg(raw_col))
|
||
.where(DockerMetric.host_id == host_id, DockerMetric.container_name == name,
|
||
DockerMetric.scraped_at >= since)
|
||
.group_by(rbucket).order_by(rbucket)
|
||
)).all()
|
||
oldest_raw = raw[0][0] if raw else None
|
||
hourly_stmt = (
|
||
select(DockerMetricHourly.bucket, getattr(DockerMetricHourly, metric))
|
||
.where(DockerMetricHourly.host_id == host_id,
|
||
DockerMetricHourly.container_name == name,
|
||
DockerMetricHourly.bucket >= since)
|
||
.order_by(DockerMetricHourly.bucket)
|
||
)
|
||
if oldest_raw is not None:
|
||
hourly_stmt = hourly_stmt.where(DockerMetricHourly.bucket < oldest_raw)
|
||
hourly = (await db.execute(hourly_stmt)).all()
|
||
return [(t, float(v)) for t, v in hourly] + [(t, float(v)) for t, v in raw]
|
||
|
||
|
||
# Busiest Containers bars (#5214): the recent past in BARS equal slices.
|
||
BARS = 24
|
||
|
||
|
||
def bar_slots(now: datetime, hours: int, bars: int = BARS) -> tuple[list[datetime], int]:
|
||
"""The start of each of the last `bars` buckets covering `hours`, oldest
|
||
first, aligned as date_bin aligns them (to the epoch), and their width in
|
||
seconds. The newest bucket is the one `now` falls in."""
|
||
width = max(60, hours * 3600 // bars)
|
||
last = int(now.timestamp()) // width * width
|
||
return [datetime.fromtimestamp(last - (bars - 1 - i) * width, timezone.utc)
|
||
for i in range(bars)], width
|
||
|
||
|
||
async def recent_usage(db, keys, since: datetime, width: int) -> dict:
|
||
"""{(host_id, container_name): {bucket start: (cpu_pct, mem_usage_bytes)}}
|
||
averaged per bucket since `since`, for the given containers, in one query."""
|
||
keys = set(keys)
|
||
if not keys:
|
||
return {}
|
||
bucket = func.date_bin(
|
||
func.make_interval(0, 0, 0, 0, 0, 0, int(width)), DockerMetric.scraped_at, func.to_timestamp(0),
|
||
).label("bucket")
|
||
rows = (await db.execute(
|
||
select(DockerMetric.host_id, DockerMetric.container_name, bucket,
|
||
func.avg(DockerMetric.cpu_pct), func.avg(DockerMetric.mem_usage_bytes))
|
||
.where(DockerMetric.scraped_at >= since,
|
||
DockerMetric.host_id.in_({h for h, _ in keys}),
|
||
DockerMetric.container_name.in_({n for _, n in keys}))
|
||
.group_by(DockerMetric.host_id, DockerMetric.container_name, bucket)
|
||
)).all()
|
||
out: dict = {}
|
||
for host_id, name, at, cpu, mem in rows:
|
||
if (host_id, name) in keys:
|
||
out.setdefault((host_id, name), {})[at] = (
|
||
None if cpu is None else float(cpu), None if mem is None else float(mem))
|
||
return out
|
||
|
||
|
||
def scaled_bars(values: list, floor: float) -> dict:
|
||
"""Bar heights in % of a roof: the row's own peak, but never below `floor`,
|
||
so a container idling at 0.05 cores draws flat and its 0.05 → 0.10 wobble
|
||
doesn't look like a 1.5 → 3 core spike. None (no samples) stays a gap."""
|
||
present = [v for v in values if v is not None]
|
||
peak = max(present) if present else None
|
||
roof = max(peak or 0.0, floor)
|
||
return {"bars": [None if v is None else round(100.0 * v / roof, 1) for v in values],
|
||
"peak": peak, "roof": roof}
|
||
|
||
|
||
async def search_logs(db, *, since, host_id: str | None = None, name: str | None = None,
|
||
text: str = "", stream: str = "", before=None, limit: int = 50) -> list:
|
||
"""Retained log lines matching a filter, newest first (limit + 1 fetched by callers
|
||
that need to know whether more exist)."""
|
||
stmt = (select(DockerLog.id, DockerLog.host_id, DockerLog.container_name,
|
||
DockerLog.ts, DockerLog.stream, DockerLog.line)
|
||
.where(DockerLog.ts >= since))
|
||
if host_id:
|
||
stmt = stmt.where(DockerLog.host_id == host_id)
|
||
if name:
|
||
stmt = stmt.where(DockerLog.container_name == name)
|
||
if stream in ("stdout", "stderr"):
|
||
stmt = stmt.where(DockerLog.stream == stream)
|
||
if text:
|
||
stmt = stmt.where(DockerLog.line.ilike(f"%{escape_like(text)}%", escape="\\"))
|
||
if before is not None:
|
||
stmt = stmt.where(DockerLog.ts < before)
|
||
stmt = stmt.order_by(DockerLog.ts.desc(), DockerLog.id.desc()).limit(limit)
|
||
return list((await db.execute(stmt)).all())
|
||
|
||
|
||
async def log_context(db, host_id: str, name: str, ts, n: int) -> tuple[list, list]:
|
||
"""Up to n lines either side of `ts` in one container's log, each oldest first."""
|
||
cols = (DockerLog.id, DockerLog.ts, DockerLog.stream, DockerLog.line)
|
||
base = select(*cols).where(DockerLog.host_id == host_id, DockerLog.container_name == name)
|
||
before = (await db.execute(
|
||
base.where(DockerLog.ts < ts).order_by(DockerLog.ts.desc(), DockerLog.id.desc()).limit(n)
|
||
)).all()
|
||
after = (await db.execute(
|
||
base.where(DockerLog.ts > ts).order_by(DockerLog.ts, DockerLog.id).limit(n)
|
||
)).all()
|
||
return list(reversed(before)), list(after)
|
||
|
||
|
||
async def recent_failures_by_container(db, since) -> dict[tuple[str, str], list]:
|
||
"""{(host_id, container_name): [die/oom events since `since`, newest first]},
|
||
leaving out a die that was a stop (event_is_failure)."""
|
||
rows = (await db.execute(
|
||
select(DockerEvent)
|
||
.where(DockerEvent.event.in_(("die", "oom")), DockerEvent.at >= since)
|
||
.order_by(DockerEvent.at.desc())
|
||
)).scalars()
|
||
out: dict[tuple[str, str], list] = {}
|
||
for e in rows:
|
||
if event_is_failure(e):
|
||
out.setdefault((e.host_id, e.container_name), []).append(e)
|
||
return out
|
||
|
||
|
||
async def failure_events(db, since) -> list:
|
||
"""Die and OOM events since `since`, newest first; failures_24h() decides
|
||
which of them are failures."""
|
||
return list((await db.execute(
|
||
select(DockerEvent)
|
||
.where(DockerEvent.event.in_(("die", "oom")), DockerEvent.at >= since)
|
||
.order_by(DockerEvent.at.desc())
|
||
)).scalars())
|
||
|
||
|
||
async def retained_names(db, want: str, *, exact: bool, host_id: str | None = None,
|
||
limit: int = 20) -> list[tuple[str, str, datetime]]:
|
||
"""(host_id, container_name, last seen) of containers Steward still holds
|
||
logs or events for, newest first — including ones Docker has removed.
|
||
|
||
exact: the name itself, from both logs and events. Otherwise names that
|
||
contain `want` (case-insensitive), from events only: a substring match
|
||
over every retained log line's name would scan the log table, and a
|
||
container Steward watched die has a die or stop event.
|
||
"""
|
||
seen: dict[tuple[str, str], datetime] = {}
|
||
sources = ((DockerLog, DockerLog.ts), (DockerEvent, DockerEvent.at)) if exact \
|
||
else ((DockerEvent, DockerEvent.at),)
|
||
for model, at in sources:
|
||
match = (model.container_name == want if exact
|
||
else model.container_name.ilike(f"%{escape_like(want)}%", escape="\\"))
|
||
stmt = (select(model.host_id, model.container_name, func.max(at))
|
||
.where(match).group_by(model.host_id, model.container_name))
|
||
if host_id:
|
||
stmt = stmt.where(model.host_id == host_id)
|
||
for hid, name, last in (await db.execute(stmt)).all():
|
||
key = (hid, name)
|
||
if key not in seen or last > seen[key]:
|
||
seen[key] = last
|
||
rows = sorted(((h, n, t) for (h, n), t in seen.items()), key=lambda r: r[2], reverse=True)
|
||
return rows[:limit]
|
||
|
||
|
||
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."""
|
||
keys = set(keys)
|
||
if not keys:
|
||
return {}
|
||
rows = (await db.execute(
|
||
select(DockerEvent)
|
||
.where(DockerEvent.event == "health_change",
|
||
DockerEvent.detail.like("%→unhealthy"),
|
||
DockerEvent.container_name.in_({name for _, name in keys}))
|
||
.order_by(DockerEvent.at.desc())
|
||
)).scalars()
|
||
out: dict[tuple[str, str], datetime] = {}
|
||
for e in rows:
|
||
key = (e.host_id, e.container_name)
|
||
if key in keys and key not in out:
|
||
out[key] = e.at
|
||
return out
|
||
|
||
|
||
PRUNE_TARGETS = ("containers", "images", "system")
|
||
|
||
|
||
def prune_extra_vars(prune_target: str) -> dict:
|
||
"""Extra-vars for the bundled docker_prune playbook. "Prune unused images"
|
||
means ALL unused (docker image prune -a), not just dangling; the full system
|
||
prune stays conservative (-f, no -a) per the operator's choice (m78)."""
|
||
extra_vars: dict = {"prune_target": prune_target}
|
||
if prune_target == "images":
|
||
extra_vars["prune_all_images"] = True
|
||
return extra_vars
|
||
|
||
|
||
# ── what is wrong with a container ───────────────────────────────────────────
|
||
# Shared by the Docker page's "Needs attention" and the MCP tools, so both
|
||
# rank the same containers the same way.
|
||
|
||
PROBLEMS = ("restarting", "crash_looping", "unhealthy", "oom_killed", "exited_nonzero")
|
||
|
||
def container_problems(c, failures_1h: int, failed_24h: bool,
|
||
crash_loop_deaths: int = CRASH_LOOP_DEATHS) -> list[str]:
|
||
"""What is wrong with one container, most severe first; [] when nothing is."""
|
||
status = (c.status or "").lower()
|
||
out = []
|
||
if status == "restarting":
|
||
out.append("restarting")
|
||
if failures_1h >= crash_loop_deaths:
|
||
out.append("crash_looping")
|
||
if status == "running" and (c.health or "") == "unhealthy":
|
||
out.append("unhealthy")
|
||
if c.oom_killed:
|
||
out.append("oom_killed")
|
||
if status in ("exited", "dead") and exit_is_failure(c.exit_code) and failed_24h:
|
||
out.append("exited_nonzero")
|
||
return out
|
||
|
||
|
||
def problem_since(problems: list[str], deaths: list, hour_ago: datetime,
|
||
unhealthy_at: datetime | None = None) -> datetime | None:
|
||
"""When a container's worst problem began; None when the retained events
|
||
cannot say. `deaths` are its die/oom events of the last 24h, newest first.
|
||
|
||
A restart or crash loop dates from its first death in the last hour, an
|
||
OOM kill or error exit from the death that ended it, and a failing
|
||
healthcheck from when its health last turned unhealthy.
|
||
"""
|
||
if not problems:
|
||
return None
|
||
worst = problems[0]
|
||
if worst in ("restarting", "crash_looping"):
|
||
in_hour = [e.at for e in deaths if e.at >= hour_ago]
|
||
return min(in_hour) if in_hour else (deaths[0].at if deaths else None)
|
||
if worst == "unhealthy":
|
||
return unhealthy_at
|
||
if worst == "oom_killed":
|
||
ooms = [e.at for e in deaths if e.event == "oom"]
|
||
return ooms[0] if ooms else (deaths[0].at if deaths else None)
|
||
return deaths[0].at if deaths else None
|
||
|
||
|
||
PROBLEM_SEVERITY = {"restarting": "critical", "crash_looping": "critical", "unhealthy": "critical",
|
||
"oom_killed": "warning", "exited_nonzero": "warning"}
|
||
PROBLEM_SAYS = {"restarting": "is restarting", "crash_looping": "keeps dying",
|
||
"unhealthy": "fails its healthcheck", "oom_killed": "was OOM-killed",
|
||
"exited_nonzero": "exited with an error"}
|
||
|
||
|
||
|
||
async def ranked_containers(db, now: datetime, crash_loop_deaths: int) -> list[dict]:
|
||
"""Every container (deduplicated) with what is wrong with it, most severe
|
||
first: {container, problems, since, replaced}. A swarm task Swarm has
|
||
replaced has no problems; its failure belongs to its service."""
|
||
containers = dedup_by_container_id(await list_containers(db))
|
||
services = list((await db.execute(select(DockerSwarmService))).scalars())
|
||
replaced = replaced_tasks(containers, services)
|
||
failures = await recent_failures_by_container(db, now - timedelta(hours=24))
|
||
hour_ago = now - timedelta(hours=1)
|
||
sick = await unhealthy_since(
|
||
db, [(c.host_id, c.name) for c in containers if (c.health or "") == "unhealthy"])
|
||
out = []
|
||
for c in containers:
|
||
key = (c.host_id, c.name)
|
||
gone = key in replaced
|
||
events = failures.get(key, [])
|
||
recent = sum(1 for e in events if e.at >= hour_ago)
|
||
problems = [] if gone else container_problems(c, recent, bool(events), crash_loop_deaths)
|
||
out.append({"container": c, "problems": problems, "replaced": gone,
|
||
"since": problem_since(problems, events, hour_ago, sick.get(key))})
|
||
return out
|
||
|
||
|
||
def needs_attention(ranked: list[dict]) -> list[dict]:
|
||
"""The containers with something wrong, worst problem first, then by host
|
||
and name."""
|
||
return sorted((r for r in ranked if r["problems"]),
|
||
key=lambda r: (PROBLEMS.index(r["problems"][0]), r["container"].host_id, r["container"].name))
|
||
|
||
|
||
def stale_after_seconds(config) -> int:
|
||
"""How old a container's last report may be before the page calls it
|
||
stale: the host agent's own staleness window, since the agent reports it."""
|
||
return int(((config.get("PLUGINS") or {}).get("host_agent") or {}).get("stale_after_seconds", 180))
|