feat(docker): collect + persist /system/df image/disk usage (agent 1.6.0)

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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016Jg27rgypiW2efULXJDtMC

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 "<none>")[: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

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="<none>"),
+        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")

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="<none>")
+    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"

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 ["<none>:<none>"] or null.
+        tag = next((t for t in tags if t and t != "<none>:<none>"), None)
+        raw_id = (im.get("Id") or "").split(":", 1)[-1]
+        out.append({
+            "image_id": raw_id[:12],
+            "repo_tag": tag or "<none>",
+            "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
 
 

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(

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": "<none>", "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 +

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": ["<none>:<none>"],
+         "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"] == "<none>" 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):