diff --git a/plugins/docker/ingest.py b/plugins/docker/ingest.py index 1b715d6..cad4a6b 100644 --- a/plugins/docker/ingest.py +++ b/plugins/docker/ingest.py @@ -73,6 +73,20 @@ async def persist_host_docker(session, host, snapshots) -> None: existing.ports_json = json.dumps(c.get("ports") or []) existing.started_at = _parse_started_at(c.get("started_at")) existing.scraped_at = latest_at + # Enrichment (agent ≥ 1.4.0; .get keeps older agents working — fields + # stay None/0 when absent from the payload). + existing.restart_count = c.get("restart_count", 0) or 0 + existing.health = c.get("health") + existing.exit_code = c.get("exit_code") + existing.oom_killed = bool(c.get("oom_killed", False)) + existing.compose_project = c.get("compose_project") + existing.service_name = c.get("service_name") + existing.task_id = c.get("task_id") + existing.node_id = c.get("node_id") + existing.net_rx_bytes = c.get("net_rx_bytes") + existing.net_tx_bytes = c.get("net_tx_bytes") + existing.blk_read_bytes = c.get("blk_read_bytes") + existing.blk_write_bytes = c.get("blk_write_bytes") # Alert pipeline — resource is host-scoped so containers of the same name # on different hosts don't collide in the metric/alert namespace. diff --git a/plugins/docker/migrations/versions/docker_003_container_enrichment.py b/plugins/docker/migrations/versions/docker_003_container_enrichment.py new file mode 100644 index 0000000..4a7f794 --- /dev/null +++ b/plugins/docker/migrations/versions/docker_003_container_enrichment.py @@ -0,0 +1,41 @@ +"""Docker container enrichment: health, exit/restart, grouping, I/O counters + +Adds the fields the agent (≥1.4.0) now reports per container beyond the basic +state: health status, exit code, OOM flag, compose/swarm grouping labels, and +cumulative network/block I/O counters. Additive columns — no data loss, so no +DROP+recreate needed here. + +Revision ID: docker_003_container_enrichment +Revises: docker_002_host_scoped +Create Date: 2026-06-19 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "docker_003_container_enrichment" +down_revision: Union[str, None] = "docker_002_host_scoped" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column("docker_containers", sa.Column("health", sa.String(16), nullable=True)) + op.add_column("docker_containers", sa.Column("exit_code", sa.Integer, nullable=True)) + op.add_column("docker_containers", + sa.Column("oom_killed", sa.Boolean, nullable=False, server_default=sa.false())) + op.add_column("docker_containers", sa.Column("compose_project", sa.String(255), nullable=True)) + op.add_column("docker_containers", sa.Column("service_name", sa.String(255), nullable=True)) + op.add_column("docker_containers", sa.Column("task_id", sa.String(64), nullable=True)) + op.add_column("docker_containers", sa.Column("node_id", sa.String(64), nullable=True)) + op.add_column("docker_containers", sa.Column("net_rx_bytes", sa.BigInteger, nullable=True)) + op.add_column("docker_containers", sa.Column("net_tx_bytes", sa.BigInteger, nullable=True)) + op.add_column("docker_containers", sa.Column("blk_read_bytes", sa.BigInteger, nullable=True)) + op.add_column("docker_containers", sa.Column("blk_write_bytes", sa.BigInteger, nullable=True)) + + +def downgrade() -> None: + for col in ("blk_write_bytes", "blk_read_bytes", "net_tx_bytes", "net_rx_bytes", + "node_id", "task_id", "service_name", "compose_project", + "oom_killed", "exit_code", "health"): + op.drop_column("docker_containers", col) diff --git a/plugins/docker/models.py b/plugins/docker/models.py index b966455..8fd2ae4 100644 --- a/plugins/docker/models.py +++ b/plugins/docker/models.py @@ -2,7 +2,9 @@ from __future__ import annotations import uuid from datetime import datetime, timezone -from sqlalchemy import DateTime, Float, ForeignKey, Index, Integer, String, Text +from sqlalchemy import ( + BigInteger, Boolean, DateTime, Float, ForeignKey, Index, Integer, String, Text, +) from sqlalchemy.orm import Mapped, mapped_column from steward.models.base import Base @@ -38,6 +40,23 @@ class DockerContainer(Base): default=lambda: datetime.now(timezone.utc), ) + # ── Enrichment (agent ≥ 1.4.0) ──────────────────────────────────────────── + # Health/exit/restart come from `docker inspect` (not the list endpoint); + # exit_code is only meaningful for stopped containers. + health: Mapped[str | None] = mapped_column(String(16), nullable=True) # healthy|unhealthy|starting + exit_code: Mapped[int | None] = mapped_column(Integer, nullable=True) + oom_killed: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + # Grouping: compose project + swarm placement, read off container labels. + compose_project: Mapped[str | None] = mapped_column(String(255), nullable=True) + service_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + task_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + node_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + # Cumulative-since-start I/O counters (BigInteger — they exceed 2^31 quickly). + net_rx_bytes: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + net_tx_bytes: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + blk_read_bytes: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + blk_write_bytes: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + class DockerMetric(Base): """Time-series CPU/memory per container — one row per sample per running diff --git a/plugins/docker/routes.py b/plugins/docker/routes.py index 5eaae65..a6b38f2 100644 --- a/plugins/docker/routes.py +++ b/plugins/docker/routes.py @@ -1,6 +1,8 @@ # plugins/docker/routes.py from __future__ import annotations import json +from datetime import datetime, timezone + from quart import Blueprint, current_app, render_template, request from sqlalchemy import select @@ -13,6 +15,25 @@ from .models import DockerContainer, DockerMetric docker_bp = Blueprint("docker", __name__, template_folder="templates") +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'' @@ -119,6 +140,7 @@ async def rows(): 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), }) diff --git a/plugins/docker/templates/docker/rows.html b/plugins/docker/templates/docker/rows.html index 686609e..6dbfca8 100644 --- a/plugins/docker/templates/docker/rows.html +++ b/plugins/docker/templates/docker/rows.html @@ -54,8 +54,24 @@
-
{{ c.name }}
-
{{ c.status }}
+
+ {{ c.name }} + {% if c.health == 'healthy' %}● + {% elif c.health == 'unhealthy' %}● + {% elif c.health == 'starting' %}◐{% endif %} +
+
+ {{ c.status }}{% if item.uptime %} · up {{ item.uptime }}{% endif %} + {% if c.status != 'running' and c.exit_code is not none and c.exit_code != 0 %} + · exit {{ c.exit_code }}{% if c.oom_killed %} (OOM){% endif %} + {% endif %} + {% if c.restart_count %} · ⟳{{ c.restart_count }}{% endif %} +
+ {% if c.service_name or c.compose_project %} +
+ {{ c.service_name or c.compose_project }} +
+ {% endif %}
diff --git a/plugins/host_agent/agent.py b/plugins/host_agent/agent.py index 3e8356d..227bc2e 100644 --- a/plugins/host_agent/agent.py +++ b/plugins/host_agent/agent.py @@ -17,9 +17,10 @@ import time import urllib.error import urllib.request from collections import deque +from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone -AGENT_VERSION = "1.3.0" +AGENT_VERSION = "1.4.0" # Default path to the local Docker Engine socket. Overridable via the # `docker_socket` config key; collection is silently skipped if it's absent or @@ -515,6 +516,117 @@ def _docker_ports(ports: list) -> list: return out +def _docker_grouping(labels: dict) -> dict: + """Pull compose project + swarm service/task/node out of container labels. + + These are set by Docker itself (compose / swarm), so reading them off the + container costs nothing extra and lets Steward group + place containers + without a manager-node query. + """ + labels = labels or {} + return { + "compose_project": labels.get("com.docker.compose.project"), + "service_name": labels.get("com.docker.swarm.service.name"), + "task_id": labels.get("com.docker.swarm.task.id"), + "node_id": labels.get("com.docker.swarm.node.id"), + } + + +def _docker_inspect_fields(insp: dict): + """Return (health, restart_count, exit_code, oom_killed) from an inspect. + + health is None for containers without a HEALTHCHECK (no State.Health). + """ + state = (insp or {}).get("State") or {} + health = (state.get("Health") or {}).get("Status") # healthy|unhealthy|starting + restart_count = insp.get("RestartCount", 0) if isinstance(insp, dict) else 0 + exit_code = state.get("ExitCode") + oom_killed = bool(state.get("OOMKilled", False)) + return health, restart_count, exit_code, oom_killed + + +def _docker_net_io(stats: dict): + """Return cumulative (net_rx, net_tx, blk_read, blk_write) bytes from stats. + + Counters are cumulative since container start (rates can be derived later); + block I/O comes from the cgroup io_service_bytes list, absent on some setups. + """ + net_rx = net_tx = blk_read = blk_write = 0 + for iface in (stats.get("networks") or {}).values(): + net_rx += iface.get("rx_bytes", 0) or 0 + net_tx += iface.get("tx_bytes", 0) or 0 + for e in ((stats.get("blkio_stats") or {}).get("io_service_bytes_recursive") or []): + op = (e.get("op") or "").lower() + if op == "read": + blk_read += e.get("value", 0) or 0 + elif op == "write": + blk_write += e.get("value", 0) or 0 + return net_rx, net_tx, blk_read, blk_write + + +def _collect_one_container(socket_path: str, c: dict) -> dict: + """Build one container's enriched record (runs in a worker thread). + + Does a per-container inspect (health/restart/exit/oom — only available there) + plus a stats read for running containers (cpu/mem/net/io). All best-effort: + a failed call just leaves the affected fields at their defaults. + """ + cid = c.get("Id", "") or "" + names = c.get("Names") or [] + name = names[0].lstrip("/") if names else cid[:12] + state = c.get("State", "unknown") + + # Inspect every container (incl. stopped) — exit codes + restart counts only + # live here, and stopped containers are exactly where exit_code matters. + health = exit_code = None + restart_count = 0 + oom_killed = False + try: + insp = _docker_request(socket_path, f"/containers/{cid}/json") + health, restart_count, exit_code, oom_killed = _docker_inspect_fields(insp) + except (OSError, ValueError): + pass + + cpu_pct = mem_usage = mem_limit = mem_pct = None + net_rx = net_tx = blk_read = blk_write = None + if state == "running": + try: + stats = _docker_request(socket_path, f"/containers/{cid}/stats?stream=false") + cpu_pct = _docker_cpu_pct(stats) + mem_usage, mem_limit, mem_pct = _docker_mem(stats) + net_rx, net_tx, blk_read, blk_write = _docker_net_io(stats) + except (OSError, ValueError): + pass + + created = c.get("Created") + started_at = ( + datetime.fromtimestamp(created, tz=timezone.utc).isoformat() + if isinstance(created, (int, float)) else None + ) + record = { + "name": name, + "container_id": cid[:12], + "image": c.get("Image", ""), + "status": state, + "cpu_pct": cpu_pct, + "mem_usage_bytes": mem_usage, + "mem_limit_bytes": mem_limit, + "mem_pct": mem_pct, + "ports": _docker_ports(c.get("Ports", [])), + "started_at": started_at, + "health": health, + "restart_count": restart_count, + "exit_code": exit_code, + "oom_killed": oom_killed, + "net_rx_bytes": net_rx, + "net_tx_bytes": net_tx, + "blk_read_bytes": blk_read, + "blk_write_bytes": blk_write, + } + record.update(_docker_grouping(c.get("Labels"))) + return record + + def collect_docker(socket_path: str) -> list: """Per-container state from the local Docker socket, or [] if unavailable. @@ -526,44 +638,16 @@ def collect_docker(socket_path: str) -> list: containers = _docker_request(socket_path, "/containers/json?all=true") except (OSError, ValueError): return [] - if not isinstance(containers, list): + if not isinstance(containers, list) or not containers: return [] - out = [] - for c in containers: - cid = c.get("Id", "") or "" - names = c.get("Names") or [] - name = names[0].lstrip("/") if names else cid[:12] - state = c.get("State", "unknown") - - cpu_pct = mem_usage = mem_limit = mem_pct = None - if state == "running": - try: - stats = _docker_request( - socket_path, f"/containers/{cid}/stats?stream=false") - cpu_pct = _docker_cpu_pct(stats) - mem_usage, mem_limit, mem_pct = _docker_mem(stats) - except (OSError, ValueError): - pass # stats unavailable for this container — leave gauges None - - created = c.get("Created") - started_at = ( - datetime.fromtimestamp(created, tz=timezone.utc).isoformat() - if isinstance(created, (int, float)) else None - ) - out.append({ - "name": name, - "container_id": cid[:12], - "image": c.get("Image", ""), - "status": state, - "cpu_pct": cpu_pct, - "mem_usage_bytes": mem_usage, - "mem_limit_bytes": mem_limit, - "mem_pct": mem_pct, - "ports": _docker_ports(c.get("Ports", [])), - "started_at": started_at, - }) - return out + # Each container needs an inspect (+ a stats read if running), and the stats + # call blocks ~1s while the daemon computes the cpu delta. Done serially that + # stretches the whole sample on a busy host, so fan the per-container work out + # over a small bounded thread pool (I/O-bound → threads are enough). + workers = min(8, len(containers)) + with ThreadPoolExecutor(max_workers=workers) as ex: + return list(ex.map(lambda c: _collect_one_container(socket_path, c), containers)) # ─── ring buffer ───────────────────────────────────────────────────────────── diff --git a/tests/integration/test_docker.py b/tests/integration/test_docker.py index 9271e3c..6c5c280 100644 --- a/tests/integration/test_docker.py +++ b/tests/integration/test_docker.py @@ -75,6 +75,11 @@ def test_persist_scopes_containers_by_host(app): "name": "web", "container_id": "abc123", "image": "nginx", "status": "running", "cpu_pct": 12.5, "mem_pct": 30.0, "mem_usage_bytes": 100, "mem_limit_bytes": 200, "ports": [], "started_at": None, + # enrichment (agent ≥ 1.4.0) + "health": "healthy", "restart_count": 2, "exit_code": 0, "oom_killed": False, + "compose_project": "stack", "service_name": "web", + "net_rx_bytes": 1000, "net_tx_bytes": 2000, + "blk_read_bytes": 4096, "blk_write_bytes": 8192, }])] async def _go(): @@ -101,9 +106,14 @@ def test_persist_scopes_containers_by_host(app): resources = {r[0] for r in (await s.execute(text( "SELECT DISTINCT resource_name FROM plugin_metrics " "WHERE source_module = 'docker'"))).all()} - return web_rows, metric_hosts, resources + # Enrichment columns persisted (check one host's row). + enrich = (await s.execute(text( + "SELECT health, restart_count, service_name, net_rx_bytes " + "FROM docker_containers WHERE name = 'web' LIMIT 1"))).first() + return web_rows, metric_hosts, resources, enrich - web_rows, metric_hosts, resources = asyncio.run(_go()) + web_rows, metric_hosts, resources, enrich = asyncio.run(_go()) assert web_rows == 2 # one 'web' per host, keyed by (host_id, name) assert metric_hosts == 2 # time-series rows scoped per host assert resources == {"alpha/web", "beta/web"} + assert enrich == ("healthy", 2, "web", 1000) # enrichment round-trips diff --git a/tests/plugins/host_agent/test_agent_docker.py b/tests/plugins/host_agent/test_agent_docker.py index 6ab4b7f..e734c19 100644 --- a/tests/plugins/host_agent/test_agent_docker.py +++ b/tests/plugins/host_agent/test_agent_docker.py @@ -68,6 +68,47 @@ def test_docker_ports_keeps_only_published(): ] +# ── enrichment helpers ──────────────────────────────────────────────────────── + +def test_docker_grouping_from_labels(): + g = a._docker_grouping({ + "com.docker.compose.project": "myproj", + "com.docker.swarm.service.name": "web", + "com.docker.swarm.task.id": "t123", + "com.docker.swarm.node.id": "n456", + }) + assert g == {"compose_project": "myproj", "service_name": "web", + "task_id": "t123", "node_id": "n456"} + + +def test_docker_grouping_empty(): + assert a._docker_grouping(None) == { + "compose_project": None, "service_name": None, "task_id": None, "node_id": None} + + +def test_docker_inspect_fields(): + insp = {"RestartCount": 3, + "State": {"Health": {"Status": "unhealthy"}, "ExitCode": 137, "OOMKilled": True}} + assert a._docker_inspect_fields(insp) == ("unhealthy", 3, 137, True) + + +def test_docker_inspect_fields_no_healthcheck(): + # Containers without a HEALTHCHECK have no State.Health → health is None. + health, restarts, exit_code, oom = a._docker_inspect_fields( + {"RestartCount": 0, "State": {"ExitCode": 0, "OOMKilled": False}}) + assert health is None and restarts == 0 and exit_code == 0 and oom is False + + +def test_docker_net_io_sums_counters(): + stats = { + "networks": {"eth0": {"rx_bytes": 1000, "tx_bytes": 2000}, + "eth1": {"rx_bytes": 5, "tx_bytes": 7}}, + "blkio_stats": {"io_service_bytes_recursive": [ + {"op": "Read", "value": 4096}, {"op": "Write", "value": 8192}]}, + } + assert a._docker_net_io(stats) == (1005, 2007, 4096, 8192) + + # ── collect_docker assembly + silent skip ───────────────────────────────────── def test_collect_docker_silent_skip_when_no_socket(): @@ -83,7 +124,11 @@ def test_collect_docker_assembles_container(monkeypatch): "State": "running", "Created": 1_700_000_000, "Ports": [{"PublicPort": 8080, "PrivatePort": 80, "Type": "tcp"}], + "Labels": {"com.docker.compose.project": "myproj", + "com.docker.swarm.service.name": "web"}, } + inspect = {"RestartCount": 2, + "State": {"Health": {"Status": "healthy"}, "ExitCode": 0, "OOMKilled": False}} stats = { "cpu_stats": {"cpu_usage": {"total_usage": 200_000_000}, "system_cpu_usage": 2_000_000_000, "online_cpus": 4}, @@ -91,10 +136,19 @@ def test_collect_docker_assembles_container(monkeypatch): "system_cpu_usage": 1_000_000_000}, "memory_stats": {"usage": 200 * 1024 * 1024, "limit": 1000 * 1024 * 1024, "stats": {"cache": 50 * 1024 * 1024}}, + "networks": {"eth0": {"rx_bytes": 1000, "tx_bytes": 2000}}, + "blkio_stats": {"io_service_bytes_recursive": [ + {"op": "Read", "value": 4096}, {"op": "Write", "value": 8192}]}, } def fake_request(socket_path, path, timeout=a.DOCKER_API_TIMEOUT): - return [container] if path.startswith("/containers/json") else stats + if path.startswith("/containers/json"): + return [container] + if "/stats" in path: + return stats + if path.endswith("/json"): # inspect + return inspect + raise AssertionError(f"unexpected path {path}") monkeypatch.setattr(a, "_docker_request", fake_request) out = a.collect_docker("/var/run/docker.sock") @@ -108,21 +162,34 @@ def test_collect_docker_assembles_container(monkeypatch): assert c["mem_pct"] == 15.0 assert c["ports"] == [{"host_port": 8080, "container_port": 80, "protocol": "tcp"}] assert c["started_at"].startswith("2023-11-14T") # epoch → ISO-8601 + # enrichment + assert c["health"] == "healthy" and c["restart_count"] == 2 + assert c["exit_code"] == 0 and c["oom_killed"] is False + assert c["net_rx_bytes"] == 1000 and c["net_tx_bytes"] == 2000 + assert c["blk_read_bytes"] == 4096 and c["blk_write_bytes"] == 8192 + assert c["compose_project"] == "myproj" and c["service_name"] == "web" def test_collect_docker_skips_stats_for_stopped(monkeypatch): container = {"Id": "x" * 16, "Names": ["/db"], "Image": "pg", "State": "exited", - "Created": None, "Ports": []} + "Created": None, "Ports": [], "Labels": {}} + inspect = {"RestartCount": 0, + "State": {"ExitCode": 137, "OOMKilled": True}} # no Health block def fake_request(socket_path, path, timeout=a.DOCKER_API_TIMEOUT): if path.startswith("/containers/json"): return [container] + if path.endswith("/json"): # inspect — fetched for stopped too (exit code) + return inspect raise AssertionError("stats must not be fetched for a stopped container") monkeypatch.setattr(a, "_docker_request", fake_request) out = a.collect_docker("/var/run/docker.sock") assert out[0]["status"] == "exited" assert out[0]["cpu_pct"] is None and out[0]["started_at"] is None + # inspect still gives us why it died + assert out[0]["exit_code"] == 137 and out[0]["oom_killed"] is True + assert out[0]["health"] is None # ── build_sample wiring ───────────────────────────────────────────────────────