bvandeusenandClaude Opus 5.5
29be9e3590
Service page: - The header carries the verdict, mode, stack, image and when it was reported. 'In trouble since' shows only while the service is in trouble; once recovered, 'Last failed X ago · its logs'. - The failure card shows only while the service is down, degraded or crash-looping. - The tile strip and the kv card are gone. - One Tasks table: running tasks first with CPU, memory and since when, then Swarm's record. The node is named as its Steward host throughout, and 'N failed in 24h' sits in the caption. Container page: - Same shape: header verdict, a 'Last exit' card only while it is stopped after a failing exit, set-up and I/O in one kv card, and lifecycle as a table. - A swarm task's breadcrumb goes through its service. Both pages: - Resource history is the one fragment (_usage_history.html): now and peak in cores and bytes, drawn with a floor of 1 core / 1 GiB (sparkline_svg floor=), so idle noise draws flat. - The container page's old % sparklines, its private _sparkline and the unused bucket()/container_history() helpers are removed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
676 lines
30 KiB
Python
676 lines
30 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, 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] = {}
|
||
# 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,
|
||
"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
|
||
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"),
|
||
"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)
|
||
|
||
|
||
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
|
||
|
||
|
||
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]
|
||
|
||
|
||
# A swarm task's container is named "<service>.<slot>.<task id>": slot is the
|
||
# replica number, or the node id for a global service; task ids are 25 chars.
|
||
_TASK_SUFFIX = re.compile(r"^(?P<slot>[A-Za-z0-9]+)\.(?P<task>[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."""
|
||
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))
|
||
|
||
|
||
async def disk_hosts(db, since: datetime, *, points: int = 60) -> list[dict]:
|
||
"""Each Docker host's disk as `docker system df` last said, most space
|
||
first, with its recorded total since `since` (#5304): the points, oldest
|
||
first, and the growth over them (disk_view.growth; None until there are
|
||
two). Hosts are named by their Steward host."""
|
||
from steward.core import metric_series
|
||
from .disk_view import DISK_PARTS, growth
|
||
from .models import DockerDiskUsage
|
||
summaries = list((await db.execute(select(DockerDiskUsage))).scalars())
|
||
hosts = await host_map(db, {s.host_id for s in summaries})
|
||
out = []
|
||
for s in summaries:
|
||
name = hosts[s.host_id].name if s.host_id in hosts else s.host_id
|
||
parts = {m: int(getattr(s, f) or 0) for m, f in DISK_PARTS.items()}
|
||
series = await metric_series.points(db, "docker", name, "disk_total_bytes", since, target=points)
|
||
out.append({"host_id": s.host_id, "host_name": name, "as_of": s.updated_at,
|
||
"total": sum(parts.values()), "parts": parts,
|
||
"reclaimable": int(s.images_reclaimable or 0), "points": series,
|
||
"growth": growth(series)})
|
||
return sorted(out, key=lambda h: (-h["total"], h["host_name"].lower()))
|