From a840d6f8232f743d9aab114de91bf442769a6898 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 18 Jun 2026 21:51:14 -0400 Subject: [PATCH] feat(docker): collect + persist /system/df image/disk usage (agent 1.6.0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Backend for the image/disk panel (milestone 77 #942). Agent gains collect_disk_usage() — one /system/df call (gated on containers existing, so Docker-less hosts pay nothing), surfacing reclaimable bytes (image size held by unreferenced images), layers/containers/volumes/build-cache sizes, and the top 50 images by size. Emitted as sample["docker_disk"]; host_agent ingest tracks the newest sample's copy and hands it to the docker capability as a 5th arg. New current-state tables docker_disk_usage (one row/host) + docker_images (per-host image rows), docker_007 migration; ingest upserts the summary and replaces the image set per host (stale images pruned). Unit tests for the df parsing/reclaimable math + build_sample gating; integration test for persistence + image-set replacement. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_016Jg27rgypiW2efULXJDtMC --- plugins/docker/ingest.py | 56 ++++++++++++- .../versions/docker_007_disk_usage.py | 55 +++++++++++++ plugins/docker/models.py | 50 ++++++++++++ plugins/host_agent/agent.py | 81 ++++++++++++++++++- plugins/host_agent/routes.py | 10 ++- tests/integration/test_docker.py | 54 +++++++++++++ tests/plugins/host_agent/test_agent_docker.py | 64 +++++++++++++++ 7 files changed, 367 insertions(+), 3 deletions(-) create mode 100644 plugins/docker/migrations/versions/docker_007_disk_usage.py diff --git a/plugins/docker/ingest.py b/plugins/docker/ingest.py index 1e57619..bf6726f 100644 --- a/plugins/docker/ingest.py +++ b/plugins/docker/ingest.py @@ -151,7 +151,57 @@ async def _persist_swarm(session, host, swarm: dict) -> None: await session.execute(stale_nodes) -async def persist_host_docker(session, host, snapshots, swarm=None) -> None: +async def _persist_disk(session, host, disk: dict) -> None: + """Upsert this host's /system/df summary + replace its image storage rows. + + Current-state (not time-series): the agent re-reports the full summary each + interval, so we overwrite the single summary row and replace the image set + (delete rows no longer present, upsert the rest), scoped to this host. + """ + from datetime import timezone + from .models import DockerDiskUsage, DockerImage + + now = datetime.now(timezone.utc) + + summary = await session.get(DockerDiskUsage, host.id) + if summary is None: + summary = DockerDiskUsage(host_id=host.id) + session.add(summary) + summary.layers_size = int(disk.get("layers_size") or 0) + summary.images_size = int(disk.get("images_size") or 0) + summary.images_reclaimable = int(disk.get("images_reclaimable") or 0) + summary.containers_size = int(disk.get("containers_size") or 0) + summary.volumes_size = int(disk.get("volumes_size") or 0) + summary.build_cache_size = int(disk.get("build_cache_size") or 0) + summary.images_total = int(disk.get("images_total") or 0) + summary.images_active = int(disk.get("images_active") or 0) + summary.containers_count = int(disk.get("containers_count") or 0) + summary.volumes_count = int(disk.get("volumes_count") or 0) + summary.updated_at = now + + images = disk.get("images") or [] + seen: set[str] = set() + for im in images: + iid = im.get("image_id") + if not iid or iid in seen: + continue + seen.add(iid) + row = await session.get(DockerImage, (host.id, iid)) + if row is None: + row = DockerImage(host_id=host.id, image_id=iid) + session.add(row) + row.repo_tag = (im.get("repo_tag") or "")[:512] + row.size = int(im.get("size") or 0) + row.shared_size = int(im.get("shared_size") or 0) + row.containers = int(im.get("containers") or 0) + row.updated_at = now + stale = delete(DockerImage).where(DockerImage.host_id == host.id) + if seen: + stale = stale.where(DockerImage.image_id.notin_(seen)) + await session.execute(stale) + + +async def persist_host_docker(session, host, snapshots, swarm=None, disk=None) -> None: """Upsert containers + time-series + lifecycle events + swarm for one host. `snapshots` is a list of (recorded_at: datetime, containers: list[dict]) — @@ -160,12 +210,16 @@ async def persist_host_docker(session, host, snapshots, swarm=None) -> None: DockerMetric rows; the newest snapshot drives current container state, the alert pipeline, and lifecycle-event derivation. `swarm` is the newest sample's swarm object (or None off managers) — persisted when present. + `disk` is the newest sample's /system/df summary (or None on Docker-less + hosts) — persisted when present. """ from steward.core.alerts import record_metric from .models import DockerContainer, DockerEvent, DockerMetric if swarm is not None: await _persist_swarm(session, host, swarm) + if disk is not None: + await _persist_disk(session, host, disk) if not snapshots: return diff --git a/plugins/docker/migrations/versions/docker_007_disk_usage.py b/plugins/docker/migrations/versions/docker_007_disk_usage.py new file mode 100644 index 0000000..f6f0cb5 --- /dev/null +++ b/plugins/docker/migrations/versions/docker_007_disk_usage.py @@ -0,0 +1,55 @@ +"""Docker disk usage + image storage tables + +Adds docker_disk_usage (one per-host /system/df summary row) and docker_images +(the heavy-hitter image storage records). Both current-state, host-scoped, +re-reported each interval. Additive create_table. + +Revision ID: docker_007_disk_usage +Revises: docker_006_metric_rollup +Create Date: 2026-06-19 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "docker_007_disk_usage" +down_revision: Union[str, None] = "docker_006_metric_rollup" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "docker_disk_usage", + sa.Column("host_id", sa.String(length=36), nullable=False), + sa.Column("layers_size", sa.BigInteger(), nullable=False, server_default="0"), + sa.Column("images_size", sa.BigInteger(), nullable=False, server_default="0"), + sa.Column("images_reclaimable", sa.BigInteger(), nullable=False, server_default="0"), + sa.Column("containers_size", sa.BigInteger(), nullable=False, server_default="0"), + sa.Column("volumes_size", sa.BigInteger(), nullable=False, server_default="0"), + sa.Column("build_cache_size", sa.BigInteger(), nullable=False, server_default="0"), + sa.Column("images_total", sa.Integer(), nullable=False, server_default="0"), + sa.Column("images_active", sa.Integer(), nullable=False, server_default="0"), + sa.Column("containers_count", sa.Integer(), nullable=False, server_default="0"), + sa.Column("volumes_count", sa.Integer(), nullable=False, server_default="0"), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["host_id"], ["hosts.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("host_id"), + ) + op.create_table( + "docker_images", + sa.Column("host_id", sa.String(length=36), nullable=False), + sa.Column("image_id", sa.String(length=64), nullable=False), + sa.Column("repo_tag", sa.String(length=512), nullable=False, server_default=""), + sa.Column("size", sa.BigInteger(), nullable=False, server_default="0"), + sa.Column("shared_size", sa.BigInteger(), nullable=False, server_default="0"), + sa.Column("containers", sa.Integer(), nullable=False, server_default="0"), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["host_id"], ["hosts.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("host_id", "image_id"), + ) + + +def downgrade() -> None: + op.drop_table("docker_images") + op.drop_table("docker_disk_usage") diff --git a/plugins/docker/models.py b/plugins/docker/models.py index cfbfee8..04cb377 100644 --- a/plugins/docker/models.py +++ b/plugins/docker/models.py @@ -189,6 +189,56 @@ class DockerSwarmService(Base): ) +class DockerDiskUsage(Base): + """Per-host Docker disk usage summary from `/system/df` (current state). + + One row per host (the agent re-reports the full summary each interval). + Reclaimable = bytes held by images no container references. Sizes are bytes + (BigInteger — image caches exceed 2^31 easily). + """ + __tablename__ = "docker_disk_usage" + + host_id: Mapped[str] = mapped_column( + String(36), ForeignKey("hosts.id", ondelete="CASCADE"), primary_key=True + ) + layers_size: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + images_size: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + images_reclaimable: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + containers_size: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + volumes_size: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + build_cache_size: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + images_total: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + images_active: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + containers_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + volumes_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, + default=lambda: datetime.now(timezone.utc), + ) + + +class DockerImage(Base): + """Per-host image storage record (the heavy hitters from `/system/df`). + + Keyed (host_id, image_id); `containers` is the reference count (0 ⇒ the + image's size is reclaimable). Replaced wholesale per host on each report. + """ + __tablename__ = "docker_images" + + host_id: Mapped[str] = mapped_column( + String(36), ForeignKey("hosts.id", ondelete="CASCADE"), primary_key=True + ) + image_id: Mapped[str] = mapped_column(String(64), primary_key=True) + repo_tag: Mapped[str] = mapped_column(String(512), nullable=False, default="") + size: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + shared_size: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + containers: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, + default=lambda: datetime.now(timezone.utc), + ) + + class DockerSwarmNode(Base): """A Swarm node as seen by a manager host: role / availability / status.""" __tablename__ = "docker_swarm_nodes" diff --git a/plugins/host_agent/agent.py b/plugins/host_agent/agent.py index d1acc08..98a237e 100644 --- a/plugins/host_agent/agent.py +++ b/plugins/host_agent/agent.py @@ -20,7 +20,7 @@ from collections import deque from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone -AGENT_VERSION = "1.5.0" +AGENT_VERSION = "1.6.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 @@ -761,6 +761,74 @@ def collect_swarm(socket_path: str): } +# ─── disk usage (image storage) ────────────────────────────────────────────── + + +def _disk_images(images: list) -> list: + """Normalise /system/df image rows → compact per-image storage records. + + Sorted largest-first and capped so a host with a huge image cache doesn't + bloat the payload; the panel only needs the heavy hitters. `containers` is + how many containers reference the image (0 ⇒ reclaimable). + """ + out = [] + for im in images: + if not isinstance(im, dict): + continue + tags = im.get("RepoTags") or [] + # Untagged/dangling images report [":"] or null. + tag = next((t for t in tags if t and t != ":"), None) + raw_id = (im.get("Id") or "").split(":", 1)[-1] + out.append({ + "image_id": raw_id[:12], + "repo_tag": tag or "", + "size": int(im.get("Size") or 0), + "shared_size": int(im.get("SharedSize") or 0) if im.get("SharedSize", -1) >= 0 else 0, + "containers": int(im.get("Containers") or 0) if (im.get("Containers") or 0) > 0 else 0, + }) + out.sort(key=lambda r: r["size"], reverse=True) + return out[:50] + + +def collect_disk_usage(socket_path: str): + """Docker disk usage from `/system/df`, or None if unavailable. + + One bounded API call (the daemon walks images/layers/volumes). Reclaimable = + bytes held by images no container references — the same intent as + `docker system df`'s RECLAIMABLE column. Silent-skip (None) on any transport + failure, same contract as collect_docker/collect_swarm. + """ + try: + df = _docker_request(socket_path, "/system/df") + except (OSError, ValueError): + return None + if not isinstance(df, dict): + return None + + images = [im for im in (df.get("Images") or []) if isinstance(im, dict)] + containers = [c for c in (df.get("Containers") or []) if isinstance(c, dict)] + volumes = [v for v in (df.get("Volumes") or []) if isinstance(v, dict)] + build_cache = [b for b in (df.get("BuildCache") or []) if isinstance(b, dict)] + + def _vol_size(v): + return int((v.get("UsageData") or {}).get("Size") or 0) + + return { + "layers_size": int(df.get("LayersSize") or 0), + "images_total": len(images), + "images_active": sum(1 for im in images if (im.get("Containers") or 0) > 0), + "images_size": sum(int(im.get("Size") or 0) for im in images), + "images_reclaimable": sum( + int(im.get("Size") or 0) for im in images if (im.get("Containers") or 0) <= 0), + "containers_count": len(containers), + "containers_size": sum(int(c.get("SizeRw") or 0) for c in containers), + "volumes_count": len(volumes), + "volumes_size": sum(_vol_size(v) for v in volumes if _vol_size(v) > 0), + "build_cache_size": sum(int(b.get("Size") or 0) for b in build_cache), + "images": _disk_images(images), + } + + # ─── ring buffer ───────────────────────────────────────────────────────────── @@ -861,6 +929,17 @@ def build_sample(mounts: list[str], state: dict, if swarm is not None: sample["swarm"] = swarm + # Image/disk usage — host-level docker fact (one /system/df call), omitted on + # Docker-less hosts. Only meaningful when there's docker here at all, so skip + # the call entirely when collect_docker found nothing. + if docker: + try: + disk = collect_disk_usage(docker_socket) + except Exception: + disk = None + if disk is not None: + sample["docker_disk"] = disk + return sample diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index 3e6c486..1b20286 100644 --- a/plugins/host_agent/routes.py +++ b/plugins/host_agent/routes.py @@ -199,6 +199,8 @@ async def ingest(): docker_snapshots: list[tuple[datetime, list]] = [] latest_swarm: dict | None = None latest_swarm_ts: datetime | None = None + latest_disk: dict | None = None + latest_disk_ts: datetime | None = None for sample in samples: try: recorded_at = _parse_ts(sample["ts"]) @@ -216,6 +218,11 @@ async def ingest(): if isinstance(swarm, dict) and ( latest_swarm_ts is None or recorded_at > latest_swarm_ts): latest_swarm, latest_swarm_ts = swarm, recorded_at + # Disk usage is current-state too — keep only the newest sample's. + disk = sample.get("docker_disk") + if isinstance(disk, dict) and ( + latest_disk_ts is None or recorded_at > latest_disk_ts): + latest_disk, latest_disk_ts = disk, recorded_at accepted += 1 if latest_ts is None or recorded_at > latest_ts: latest_ts = recorded_at @@ -228,7 +235,7 @@ async def ingest(): # (opportunistic synergy via the capability registry — no hard import, # no-op when docker is disabled). A failure here must never sink the # whole ingest, so the metrics above still land. - if docker_snapshots or latest_swarm is not None: + if docker_snapshots or latest_swarm is not None or latest_disk is not None: from steward.core.capabilities import has_capability, invoke_capability if has_capability("docker.persist_host_samples"): try: @@ -238,6 +245,7 @@ async def ingest(): await invoke_capability( "docker.persist_host_samples", UserRole.admin, session, host, docker_snapshots, latest_swarm, + latest_disk, ) except Exception: current_app.logger.exception( diff --git a/tests/integration/test_docker.py b/tests/integration/test_docker.py index 8e98265..1c4af60 100644 --- a/tests/integration/test_docker.py +++ b/tests/integration/test_docker.py @@ -230,6 +230,60 @@ def test_swarm_topology_persisted(app): assert node[0] == "manager" and node[2] == "ready" and node[3] is True +@_NEEDS_DB +def test_disk_usage_persisted(app): + """persist_host_docker stores the /system/df summary + image rows, and + replaces the image set on the next report (stale images pruned).""" + from sqlalchemy import text + from steward.models.hosts import Host + + persist = _persist_fn(app) + disk1 = { + "layers_size": 1000, "images_total": 2, "images_active": 1, + "images_size": 1000, "images_reclaimable": 700, + "containers_count": 2, "containers_size": 100, + "volumes_count": 1, "volumes_size": 40, "build_cache_size": 15, + "images": [ + {"image_id": "aaaaaaaaaaaa", "repo_tag": "app:1", "size": 300, + "shared_size": 50, "containers": 1}, + {"image_id": "bbbbbbbbbbbb", "repo_tag": "", "size": 700, + "shared_size": 0, "containers": 0}, + ], + } + disk2 = {**disk1, "images_reclaimable": 0, "images": [disk1["images"][0]]} + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(text("DELETE FROM docker_images")) + await s.execute(text("DELETE FROM docker_disk_usage")) + h = Host(id=str(uuid.uuid4()), name="dsk", address="10.6.6.6") + s.add(h) + await s.flush() + hid = h.id + await persist(s, h, [], None, disk1) + summary = (await s.execute(text( + "SELECT images_reclaimable, images_total, build_cache_size " + "FROM docker_disk_usage WHERE host_id=:h"), {"h": hid})).first() + img_count = (await s.execute(text( + "SELECT COUNT(*) FROM docker_images WHERE host_id=:h"), {"h": hid})).scalar() + # Re-report with one image dropped — the stale row must be pruned. + async with s.begin(): + await persist(s, await s.get(Host, hid), [], None, disk2) + after = (await s.execute(text( + "SELECT image_id FROM docker_images WHERE host_id=:h"), {"h": hid})).all() + reclaimable2 = (await s.execute(text( + "SELECT images_reclaimable FROM docker_disk_usage WHERE host_id=:h"), + {"h": hid})).scalar() + return summary, img_count, [r[0] for r in after], reclaimable2 + + summary, img_count, after_ids, reclaimable2 = asyncio.run(_go()) + assert summary[0] == 700 and summary[1] == 2 and summary[2] == 15 + assert img_count == 2 + assert after_ids == ["aaaaaaaaaaaa"] # the unreferenced image pruned on re-report + assert reclaimable2 == 0 # summary upserted in place + + @_NEEDS_DB def test_retention_rollup_and_prune(app): """Old raw metrics roll up to hourly averages then delete; stale rollup + diff --git a/tests/plugins/host_agent/test_agent_docker.py b/tests/plugins/host_agent/test_agent_docker.py index 67b3654..646f095 100644 --- a/tests/plugins/host_agent/test_agent_docker.py +++ b/tests/plugins/host_agent/test_agent_docker.py @@ -295,6 +295,7 @@ def test_collect_swarm_silent_skip_when_no_socket(): def test_build_sample_includes_docker_when_present(monkeypatch): monkeypatch.setattr(a, "collect_docker", lambda _s: [{"name": "web", "status": "running"}]) monkeypatch.setattr(a, "collect_swarm", lambda _s: None) + monkeypatch.setattr(a, "collect_disk_usage", lambda _s: None) sample = a.build_sample(["/"], {}, "/var/run/docker.sock") assert sample["docker"] == [{"name": "web", "status": "running"}] @@ -321,6 +322,69 @@ def test_build_sample_omits_swarm_off_manager(monkeypatch): assert "swarm" not in sample +# ── disk usage (/system/df) ───────────────────────────────────────────────── + +def test_disk_images_normalises_and_caps(): + images = [ + {"Id": "sha256:aaaaaaaaaaaa1111", "RepoTags": ["nginx:latest"], + "Size": 200, "SharedSize": 50, "Containers": 2}, + {"Id": "sha256:bbbbbbbbbbbb2222", "RepoTags": [":"], + "Size": 500, "SharedSize": -1, "Containers": 0}, + ] + out = a._disk_images(images) + # Sorted largest-first; dangling tag normalised; short id; SharedSize<0 → 0. + assert out[0]["repo_tag"] == "" and out[0]["size"] == 500 + assert out[0]["shared_size"] == 0 and out[0]["containers"] == 0 + assert out[1]["repo_tag"] == "nginx:latest" and out[1]["image_id"] == "aaaaaaaaaaaa" + + +def test_collect_disk_usage_computes_reclaimable(monkeypatch): + df = { + "LayersSize": 1000, + "Images": [ + {"Id": "sha256:a", "RepoTags": ["app:1"], "Size": 300, "Containers": 1}, + {"Id": "sha256:b", "RepoTags": None, "Size": 700, "Containers": 0}, + ], + "Containers": [{"Id": "c1", "SizeRw": 25}, {"Id": "c2", "SizeRw": 75}], + "Volumes": [{"Name": "v1", "UsageData": {"Size": 40}}, + {"Name": "v2", "UsageData": {"Size": -1}}], + "BuildCache": [{"Size": 10}, {"Size": 5}], + } + monkeypatch.setattr(a, "_docker_request", + lambda s, p, timeout=a.DOCKER_API_TIMEOUT: df) + out = a.collect_disk_usage("/var/run/docker.sock") + assert out["layers_size"] == 1000 + assert out["images_total"] == 2 and out["images_active"] == 1 + assert out["images_size"] == 1000 + assert out["images_reclaimable"] == 700 # only the unreferenced image + assert out["containers_count"] == 2 and out["containers_size"] == 100 + assert out["volumes_count"] == 2 and out["volumes_size"] == 40 # negative skipped + assert out["build_cache_size"] == 15 + assert len(out["images"]) == 2 + + +def test_collect_disk_usage_silent_skip_when_no_socket(): + assert a.collect_disk_usage("/nonexistent/does-not-exist.sock") is None + + +def test_build_sample_includes_disk_when_docker_present(monkeypatch): + monkeypatch.setattr(a, "collect_docker", lambda _s: [{"name": "web", "status": "running"}]) + monkeypatch.setattr(a, "collect_swarm", lambda _s: None) + monkeypatch.setattr(a, "collect_disk_usage", lambda _s: {"images_reclaimable": 700}) + sample = a.build_sample(["/"], {}, "/var/run/docker.sock") + assert sample["docker_disk"] == {"images_reclaimable": 700} + + +def test_build_sample_omits_disk_when_no_docker(monkeypatch): + # No containers → the /system/df call is skipped entirely (cost guard). The + # sentinel would land under docker_disk if the gate ever called it. + monkeypatch.setattr(a, "collect_docker", lambda _s: []) + monkeypatch.setattr(a, "collect_swarm", lambda _s: None) + monkeypatch.setattr(a, "collect_disk_usage", lambda _s: {"sentinel": True}) + sample = a.build_sample(["/"], {}, "/var/run/docker.sock") + assert "docker_disk" not in sample + + # ── config ──────────────────────────────────────────────────────────────────── def test_read_config_defaults_docker_socket(tmp_path):