From 7b80552a7de876c85106b843cc53095f8d857708 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 18 Jun 2026 17:54:36 -0400 Subject: [PATCH] feat(docker): per-host collection via the host agent; drop central scrape MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Docker collection moves off the central single-socket scrape onto the host agent, giving Docker a real per-host dimension. The Steward host now reports its own containers like any other host, and same-named containers on different hosts no longer collide. - agent: stdlib UDS Docker client (AF_UNIX HTTP/1.1, Connection: close, chunked-aware), collect_docker() ports the cpu%/mem math; sample["docker"] added best-effort (silent-skip on absent/unreadable socket). AGENT_VERSION 1.2.0 → 1.3.0; optional docker_socket config key. - ingest: host_agent ingest hands per-host container snapshots to the docker plugin via a new "docker.persist_host_samples" capability (no hard import, no-op when docker disabled), inside a SAVEPOINT so a docker failure never sinks the host metrics. Resource names are host-scoped ("/"). - schema: docker_containers re-keyed (host_id, name); docker_metrics gains host_id; docker_002 migration DROP+recreates (dev-only, rule 122). - ui: Docker page + widgets grouped by host with host links; new per-host Docker panel embedded on the Hosts hub (gated on docker enabled via a new enabled_plugins template context). Replaces the SQLite-only strftime bucketing with DB-agnostic Python bucketing. - provisioning: install/provision playbooks add steward-agent to the docker group (best-effort) so the agent can read the socket. - removed central scrape: docker scheduler.py + scraper.py deleted; plugin.yaml socket_path/scrape_interval_seconds/include_stopped dropped (plugin 2.0.0). - tests: agent docker collector units (math, chunked decode, silent-skip, sample shape, config) + integration (host-scoped schema + persistence). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_016Jg27rgypiW2efULXJDtMC --- plugins/docker/__init__.py | 20 +- plugins/docker/ingest.py | 89 ++++++++ .../versions/docker_002_host_scoped.py | 94 +++++++++ plugins/docker/models.py | 27 ++- plugins/docker/plugin.yaml | 11 +- plugins/docker/routes.py | 192 ++++++++++++----- plugins/docker/scheduler.py | 93 -------- plugins/docker/scraper.py | 139 ------------ .../docker/templates/docker/host_panel.html | 22 ++ plugins/docker/templates/docker/rows.html | 32 ++- plugins/docker/templates/docker/widget.html | 13 +- .../templates/docker/widget_resources.html | 9 +- plugins/host_agent/agent.py | 199 +++++++++++++++++- plugins/host_agent/routes.py | 24 +++ plugins/index.yaml | 6 +- .../ansible/bundled/host_agent/install.yml | 17 ++ .../ansible/bundled/host_agent/provision.yml | 17 ++ steward/app.py | 12 +- steward/templates/hosts/detail.html | 5 + tests/integration/test_docker.py | 109 ++++++++++ tests/plugins/host_agent/test_agent_docker.py | 157 ++++++++++++++ 21 files changed, 964 insertions(+), 323 deletions(-) create mode 100644 plugins/docker/ingest.py create mode 100644 plugins/docker/migrations/versions/docker_002_host_scoped.py delete mode 100644 plugins/docker/scheduler.py delete mode 100644 plugins/docker/scraper.py create mode 100644 plugins/docker/templates/docker/host_panel.html create mode 100644 tests/integration/test_docker.py create mode 100644 tests/plugins/host_agent/test_agent_docker.py diff --git a/plugins/docker/__init__.py b/plugins/docker/__init__.py index d569556..4272f51 100644 --- a/plugins/docker/__init__.py +++ b/plugins/docker/__init__.py @@ -13,10 +13,26 @@ def setup(app: "Quart") -> None: _app = app from .models import DockerContainer, DockerMetric # noqa: F401 — register with Base + # Publish the per-host persist hook so the host agent can hand off the + # `docker` array from each sample without importing our models (opportunistic + # synergy — host_agent gates on has_capability and no-ops if we're disabled). + # required_role=viewer: this is a trusted server-side data-plane write driven + # by the authenticated agent ingest, not a user-facing privileged action. + from steward.core.capabilities import register_capability + from steward.models.users import UserRole + from .ingest import persist_host_docker + register_capability( + "docker.persist_host_samples", persist_host_docker, + label="Persist host Docker samples", + description="Store per-host container state + metrics pushed by the host agent.", + required_role=UserRole.viewer, + ) + def get_scheduled_tasks() -> list: - from .scheduler import make_task - return [make_task(_app)] + # Collection is now agent-driven (pushed to the host_agent ingest); the + # central socket scrape was removed. No periodic task to register. + return [] def get_blueprint(): diff --git a/plugins/docker/ingest.py b/plugins/docker/ingest.py new file mode 100644 index 0000000..1b715d6 --- /dev/null +++ b/plugins/docker/ingest.py @@ -0,0 +1,89 @@ +# plugins/docker/ingest.py +"""Persist host-scoped Docker samples pushed by the host agent. + +Published as the "docker.persist_host_samples" capability (see __init__.setup), +so the host_agent plugin can hand off the `docker` array from a sample WITHOUT +importing the docker models — the coupling is opportunistic and degrades to a +no-op when the docker plugin is disabled. Runs inside the caller's open +transaction (the ingest handler's session); never opens or commits its own. +""" +from __future__ import annotations + +import json +from datetime import datetime + + +def _parse_started_at(value) -> datetime | None: + """Parse the agent's ISO-8601 started_at string back to a datetime.""" + if not isinstance(value, str) or not value: + return None + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + + +async def persist_host_docker(session, host, snapshots) -> None: + """Upsert containers + append time-series for one host's docker snapshots. + + `snapshots` is a list of (recorded_at: datetime, containers: list[dict]) — + one entry per ingested sample carrying docker data (usually just one; more + when the agent flushes a backlog). Every snapshot contributes time-series + DockerMetric rows; the newest snapshot drives current container state and + the alert pipeline (record_metric writes "now", so feeding it stale buffered + snapshots would be misleading). + """ + from steward.core.alerts import record_metric + from .models import DockerContainer, DockerMetric + + if not snapshots: + return + ordered = sorted(snapshots, key=lambda s: s[0]) + latest_at, latest_containers = ordered[-1] + + # Time-series points for every snapshot (running containers with a CPU read). + for recorded_at, containers in ordered: + for c in containers: + if c.get("status") == "running" and c.get("cpu_pct") is not None: + session.add(DockerMetric( + host_id=host.id, + container_name=c["name"], + scraped_at=recorded_at, + cpu_pct=c["cpu_pct"], + mem_pct=c.get("mem_pct") or 0.0, + mem_usage_bytes=c.get("mem_usage_bytes") or 0, + )) + + # Current state + alerts from the newest snapshot only. + for c in latest_containers: + name = c.get("name") + if not name: + continue + existing = await session.get(DockerContainer, (host.id, name)) + if existing is None: + existing = DockerContainer(host_id=host.id, name=name) + session.add(existing) + existing.container_id = c.get("container_id", "") or "" + existing.image = c.get("image", "") or "" + existing.status = c.get("status", "unknown") or "unknown" + existing.cpu_pct = c.get("cpu_pct") + existing.mem_usage_bytes = c.get("mem_usage_bytes") + existing.mem_limit_bytes = c.get("mem_limit_bytes") + existing.mem_pct = c.get("mem_pct") + existing.ports_json = json.dumps(c.get("ports") or []) + existing.started_at = _parse_started_at(c.get("started_at")) + existing.scraped_at = latest_at + + # Alert pipeline — resource is host-scoped so containers of the same name + # on different hosts don't collide in the metric/alert namespace. + if c.get("status") == "running" and c.get("cpu_pct") is not None: + resource = f"{host.name}/{name}" + await record_metric( + session=session, source_module="docker", + resource_name=resource, metric_name="cpu_pct", value=c["cpu_pct"], + ) + if c.get("mem_pct") is not None: + await record_metric( + session=session, source_module="docker", + resource_name=resource, metric_name="mem_pct", value=c["mem_pct"], + ) diff --git a/plugins/docker/migrations/versions/docker_002_host_scoped.py b/plugins/docker/migrations/versions/docker_002_host_scoped.py new file mode 100644 index 0000000..aaefa83 --- /dev/null +++ b/plugins/docker/migrations/versions/docker_002_host_scoped.py @@ -0,0 +1,94 @@ +"""Docker collection goes per-host: add host_id, re-key by (host_id, name) + +Collection moved from the central single-socket scrape to the host agent, so +containers are now scoped to the host that reported them. docker_containers is +re-keyed (host_id, name) and docker_metrics gains host_id. + +Dev-only posture (rule 122): the old tables only ever held the Steward box's +own containers (a single global namespace), which are disposable — so this +DROP+recreates rather than backfilling a host_id onto orphan rows. + +Revision ID: docker_002_host_scoped +Revises: docker_001_initial +Create Date: 2026-06-18 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "docker_002_host_scoped" +down_revision: Union[str, None] = "docker_001_initial" +branch_labels: Union[str, Sequence[str], None] = None +# FK targets hosts.id (created in 0002_core_monitors) — make the edge explicit. +depends_on: Union[str, Sequence[str], None] = ("0002_core_monitors",) + + +def upgrade() -> None: + op.drop_table("docker_metrics") + op.drop_table("docker_containers") + + op.create_table( + "docker_containers", + sa.Column("host_id", sa.String(36), + sa.ForeignKey("hosts.id", ondelete="CASCADE"), primary_key=True), + sa.Column("name", sa.String(255), primary_key=True), + sa.Column("container_id", sa.String(64), nullable=False, server_default=""), + sa.Column("image", sa.String(512), nullable=False, server_default=""), + sa.Column("status", sa.String(32), nullable=False, server_default="unknown"), + sa.Column("cpu_pct", sa.Float, nullable=True), + sa.Column("mem_usage_bytes", sa.Integer, nullable=True), + sa.Column("mem_limit_bytes", sa.Integer, nullable=True), + sa.Column("mem_pct", sa.Float, nullable=True), + sa.Column("restart_count", sa.Integer, nullable=False, server_default="0"), + sa.Column("ports_json", sa.Text, nullable=False, server_default="[]"), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("scraped_at", sa.DateTime(timezone=True), nullable=False), + ) + + op.create_table( + "docker_metrics", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("host_id", sa.String(36), + sa.ForeignKey("hosts.id", ondelete="CASCADE"), nullable=False), + sa.Column("container_name", sa.String(255), nullable=False), + sa.Column("scraped_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("cpu_pct", sa.Float, nullable=False, server_default="0"), + sa.Column("mem_pct", sa.Float, nullable=False, server_default="0"), + sa.Column("mem_usage_bytes", sa.Integer, nullable=False, server_default="0"), + ) + op.create_index("ix_docker_metrics_host_id", "docker_metrics", ["host_id"]) + op.create_index("ix_docker_metrics_container_name", "docker_metrics", + ["container_name"]) + op.create_index("ix_docker_metrics_scraped_at", "docker_metrics", ["scraped_at"]) + op.create_index("ix_docker_metrics_host_container_time", "docker_metrics", + ["host_id", "container_name", "scraped_at"]) + + +def downgrade() -> None: + op.drop_table("docker_metrics") + op.drop_table("docker_containers") + + op.create_table( + "docker_containers", + sa.Column("name", sa.String(255), primary_key=True), + sa.Column("container_id", sa.String(64), nullable=False, server_default=""), + sa.Column("image", sa.String(512), nullable=False, server_default=""), + sa.Column("status", sa.String(32), nullable=False, server_default="unknown"), + sa.Column("cpu_pct", sa.Float, nullable=True), + sa.Column("mem_usage_bytes", sa.Integer, nullable=True), + sa.Column("mem_limit_bytes", sa.Integer, nullable=True), + sa.Column("mem_pct", sa.Float, nullable=True), + sa.Column("restart_count", sa.Integer, nullable=False, server_default="0"), + sa.Column("ports_json", sa.Text, nullable=False, server_default="[]"), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("scraped_at", sa.DateTime(timezone=True), nullable=False), + ) + op.create_table( + "docker_metrics", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("container_name", sa.String(255), nullable=False, index=True), + sa.Column("scraped_at", sa.DateTime(timezone=True), nullable=False, index=True), + sa.Column("cpu_pct", sa.Float, nullable=False, server_default="0"), + sa.Column("mem_pct", sa.Float, nullable=False, server_default="0"), + sa.Column("mem_usage_bytes", sa.Integer, nullable=False, server_default="0"), + ) diff --git a/plugins/docker/models.py b/plugins/docker/models.py index bec356b..b966455 100644 --- a/plugins/docker/models.py +++ b/plugins/docker/models.py @@ -2,16 +2,24 @@ from __future__ import annotations import uuid from datetime import datetime, timezone -from sqlalchemy import DateTime, Float, Integer, String, Text +from sqlalchemy import DateTime, Float, ForeignKey, Index, Integer, String, Text from sqlalchemy.orm import Mapped, mapped_column from steward.models.base import Base class DockerContainer(Base): - """Latest known state per container — upserted by name on each scrape.""" + """Latest known state per container, scoped to the host that reported it. + + Collection is per-host via the host agent, so container names are only + unique within a host — the natural key is (host_id, name). host_id is NOT + NULL: every container arrives through a host_agent ingest that resolves a + Host first. Deleting a host cascades its containers away. + """ __tablename__ = "docker_containers" - # Container name is the stable natural key; container_id changes on recreation + host_id: Mapped[str] = mapped_column( + String(36), ForeignKey("hosts.id", ondelete="CASCADE"), primary_key=True + ) name: Mapped[str] = mapped_column(String(255), primary_key=True) container_id: Mapped[str] = mapped_column(String(64), nullable=False, default="") image: Mapped[str] = mapped_column(String(512), nullable=False, default="") @@ -32,12 +40,16 @@ class DockerContainer(Base): class DockerMetric(Base): - """Time-series CPU/memory per container — one row per scrape per running container.""" + """Time-series CPU/memory per container — one row per sample per running + container, scoped to the reporting host.""" __tablename__ = "docker_metrics" id: Mapped[str] = mapped_column( String(36), primary_key=True, default=lambda: str(uuid.uuid4()) ) + host_id: Mapped[str] = mapped_column( + String(36), ForeignKey("hosts.id", ondelete="CASCADE"), nullable=False, index=True + ) container_name: Mapped[str] = mapped_column(String(255), nullable=False, index=True) scraped_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, @@ -47,3 +59,10 @@ class DockerMetric(Base): cpu_pct: Mapped[float] = mapped_column(Float, nullable=False, default=0.0) mem_pct: Mapped[float] = mapped_column(Float, nullable=False, default=0.0) mem_usage_bytes: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + + # Per-container history lookups filter on (host_id, container_name) then sort + # by time — a composite index keeps the rows() sparkline queries cheap. + __table_args__ = ( + Index("ix_docker_metrics_host_container_time", + "host_id", "container_name", "scraped_at"), + ) diff --git a/plugins/docker/plugin.yaml b/plugins/docker/plugin.yaml index 9d38f24..1237a62 100644 --- a/plugins/docker/plugin.yaml +++ b/plugins/docker/plugin.yaml @@ -1,6 +1,6 @@ name: docker -version: "1.0.0" -description: "Docker container status, resource usage, restart tracking via Docker socket" +version: "2.0.0" +description: "Per-host Docker container status + resource usage, collected by the host agent" author: "Steward" license: "MIT" min_app_version: "0.1.0" @@ -11,7 +11,6 @@ tags: - docker - infrastructure -config: - socket_path: /var/run/docker.sock - scrape_interval_seconds: 60 - include_stopped: false +# No config: collection is agent-driven (the host agent reads each host's local +# socket and pushes containers to the host_agent ingest). This plugin is pure +# presentation + storage, so there's no socket path or scrape interval to tune. diff --git a/plugins/docker/routes.py b/plugins/docker/routes.py index 41b123a..5eaae65 100644 --- a/plugins/docker/routes.py +++ b/plugins/docker/routes.py @@ -2,11 +2,12 @@ from __future__ import annotations import json from quart import Blueprint, current_app, render_template, request -from sqlalchemy import Integer, cast, func, select +from sqlalchemy import select from steward.auth.middleware import require_role from steward.models.users import UserRole -from steward.core.time_range import parse_range, DEFAULT_RANGE, bucket_seconds +from steward.models.hosts import Host +from steward.core.time_range import parse_range, DEFAULT_RANGE from .models import DockerContainer, DockerMetric docker_bp = Blueprint("docker", __name__, template_folder="templates") @@ -33,6 +34,38 @@ def _sparkline(values: list[float], width: int = 80, height: int = 20) -> str: ) +def _bucket(values: list[float], target: int = 40) -> list[float]: + """Bucket-average a series down to ~target points (DB-agnostic, in Python). + + Replaces the old SQL strftime() bucketing, which was SQLite-only and broke + on Postgres. Agent cadence is dense, so a multi-hour window is hundreds of + rows — averaging into a readable point count keeps the sparkline's shape. + """ + n = len(values) + if n <= target: + return values + size = (n + target - 1) // target + out: list[float] = [] + for i in range(0, n, size): + chunk = values[i:i + size] + out.append(sum(chunk) / len(chunk)) + return out + + +async def _container_history(db, host_id: str, name: str, since) -> tuple[list, list]: + """Return (cpu_series, mem_series) sparkline-ready for one host's container.""" + rows = (await db.execute( + select(DockerMetric.cpu_pct, DockerMetric.mem_pct) + .where(DockerMetric.host_id == host_id) + .where(DockerMetric.container_name == name) + .where(DockerMetric.scraped_at >= since) + .order_by(DockerMetric.scraped_at) + )).all() + cpu = _bucket([r.cpu_pct or 0.0 for r in rows]) + mem = _bucket([r.mem_pct or 0.0 for r in rows]) + return cpu, mem + + @docker_bp.get("/") @require_role(UserRole.viewer) async def index(): @@ -45,64 +78,64 @@ async def index(): ) +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() + } + + @docker_bp.get("/rows") @require_role(UserRole.viewer) async def rows(): - """HTMX fragment: container list with status and resource sparklines.""" + """HTMX fragment: containers grouped by host, with resource sparklines.""" since, range_key = parse_range(request.args.get("range")) - b_secs = bucket_seconds(since) - - bucket_col = ( - cast(func.strftime('%s', DockerMetric.scraped_at), Integer) / b_secs - ).label("bucket") async with current_app.db_sessionmaker() as db: - # All known containers ordered by running first, then name - result = await db.execute( - select(DockerContainer) - .order_by( - # running first + containers = list((await db.execute( + select(DockerContainer).order_by( (DockerContainer.status == "running").desc(), DockerContainer.name, ) - ) - containers = list(result.scalars()) + )).scalars()) + hosts = await _host_map(db, {c.host_id for c in containers}) - # Build per-container sparkline histories - histories: dict[str, list] = {} + # Group by host so each container is clearly attributed to the box it + # runs on (names are only unique within a host now). + groups: dict[str, dict] = {} for c in containers: - result = await db.execute( - select( - func.avg(DockerMetric.cpu_pct).label("cpu_pct"), - func.avg(DockerMetric.mem_pct).label("mem_pct"), - bucket_col, - ) - .where(DockerMetric.container_name == c.name) - .where(DockerMetric.scraped_at >= since) - .group_by(bucket_col) - .order_by(bucket_col) - ) - histories[c.name] = result.all() - - running = sum(1 for c in containers if c.status == "running") - stopped = len(containers) - running - - container_data = [] - for c in containers: - hist = histories.get(c.name, []) - ports = json.loads(c.ports_json) if c.ports_json else [] - container_data.append({ - "container": c, - "ports": ports, - "sparkline_cpu": _sparkline([r.cpu_pct or 0 for r in hist]), - "sparkline_mem": _sparkline([r.mem_pct or 0 for r in hist]), - }) + g = groups.get(c.host_id) + if g is None: + host = hosts.get(c.host_id) + g = groups[c.host_id] = { + "host": host, + "host_id": c.host_id, + "host_name": host.name if host else c.host_id, + "containers": [], "running": 0, "stopped": 0, + } + cpu_hist, mem_hist = await _container_history(db, c.host_id, c.name, since) + g["containers"].append({ + "container": c, + "ports": json.loads(c.ports_json) if c.ports_json else [], + "sparkline_cpu": _sparkline(cpu_hist), + "sparkline_mem": _sparkline(mem_hist), + }) + if c.status == "running": + g["running"] += 1 + else: + g["stopped"] += 1 + host_groups = sorted(groups.values(), key=lambda g: g["host_name"].lower()) + running = sum(g["running"] for g in host_groups) + total = len(containers) return await render_template( "docker/rows.html", - container_data=container_data, + host_groups=host_groups, running=running, - stopped=stopped, + stopped=total - running, + total=total, range_key=range_key, ) @@ -110,27 +143,38 @@ async def rows(): @docker_bp.get("/widget") @require_role(UserRole.viewer) async def widget(): - """HTMX dashboard widget: container status overview.""" + """HTMX dashboard widget: container status overview, grouped by host.""" show_stopped = request.args.get("show_stopped", "no") == "yes" widget_id = request.args.get("wid", "0") async with current_app.db_sessionmaker() as db: - result = await db.execute( - select(DockerContainer) - .order_by( + all_containers = list((await db.execute( + select(DockerContainer).order_by( (DockerContainer.status == "running").desc(), DockerContainer.name, ) - ) - all_containers = list(result.scalars()) + )).scalars()) + hosts = await _host_map(db, {c.host_id for c in all_containers}) running = [c for c in all_containers if c.status == "running"] stopped = [c for c in all_containers if c.status != "running"] display = all_containers if show_stopped else running + # Group for display so multi-host fleets read clearly; single-host stays flat. + groups: dict[str, dict] = {} + for c in display: + g = groups.setdefault(c.host_id, { + "host_id": c.host_id, + "host_name": hosts[c.host_id].name if c.host_id in hosts else c.host_id, + "containers": [], + }) + g["containers"].append(c) + host_groups = sorted(groups.values(), key=lambda g: g["host_name"].lower()) + return await render_template( "docker/widget.html", - containers=display, + host_groups=host_groups, + multi_host=len(host_groups) > 1, running_count=len(running), stopped_count=len(stopped), show_stopped=show_stopped, @@ -141,20 +185,54 @@ async def widget(): @docker_bp.get("/widget/resources") @require_role(UserRole.viewer) async def widget_resources(): - """HTMX dashboard widget: CPU + memory usage for running containers.""" + """HTMX dashboard widget: CPU + memory usage for the busiest containers.""" limit = max(1, min(20, int(request.args.get("limit", 10) or 10))) widget_id = request.args.get("wid", "0") async with current_app.db_sessionmaker() as db: - result = await db.execute( + containers = list((await db.execute( select(DockerContainer) .where(DockerContainer.status == "running") .order_by(DockerContainer.cpu_pct.desc().nullslast()) - ) - containers = list(result.scalars())[:limit] + )).scalars())[:limit] + hosts = await _host_map(db, {c.host_id for c in containers}) + rows_data = [ + {"c": c, "host_name": hosts[c.host_id].name if c.host_id in hosts else c.host_id} + for c in containers + ] return await render_template( "docker/widget_resources.html", - containers=containers, + rows=rows_data, + multi_host=len({c.host_id for c in containers}) > 1, widget_id=widget_id, ) + + +@docker_bp.get("/host/") +@require_role(UserRole.viewer) +async def host_panel(host_id: str): + """Per-host Docker fragment embedded on the core Hosts hub via HTMX. + + Renders nothing when the host reports no containers, so hosts without Docker + don't carry an empty card. + """ + async with current_app.db_sessionmaker() as db: + containers = list((await db.execute( + select(DockerContainer) + .where(DockerContainer.host_id == host_id) + .order_by( + (DockerContainer.status == "running").desc(), + DockerContainer.name, + ) + )).scalars()) + + if not containers: + return "" + running = sum(1 for c in containers if c.status == "running") + return await render_template( + "docker/host_panel.html", + containers=containers, + running=running, + stopped=len(containers) - running, + ) diff --git a/plugins/docker/scheduler.py b/plugins/docker/scheduler.py deleted file mode 100644 index 827cf5f..0000000 --- a/plugins/docker/scheduler.py +++ /dev/null @@ -1,93 +0,0 @@ -# plugins/docker/scheduler.py -from __future__ import annotations -import json -import logging -from datetime import datetime, timezone - -from steward.core.scheduler import ScheduledTask -from steward.core.alerts import record_metric - -logger = logging.getLogger(__name__) - - -def make_task(app) -> ScheduledTask: - interval = int( - app.config["PLUGINS"]["docker"].get("scrape_interval_seconds", 60) - ) - - async def scrape(): - await _do_scrape(app) - - return ScheduledTask( - name="docker_scrape", - coro_factory=scrape, - interval_seconds=interval, - run_on_startup=True, - ) - - -async def _do_scrape(app) -> None: - from .scraper import scrape_docker - from .models import DockerContainer, DockerMetric - - cfg = app.config["PLUGINS"]["docker"] - socket_path = cfg.get("socket_path", "/var/run/docker.sock") - include_stopped = bool(cfg.get("include_stopped", False)) - - try: - containers = await scrape_docker(socket_path, include_stopped) - except ConnectionError as exc: - logger.error("Docker scrape failed: %s", exc) - return - except Exception: - logger.exception("Docker scrape error") - return - - now = datetime.now(timezone.utc) - - async with app.db_sessionmaker() as session: - async with session.begin(): - for c in containers: - # Upsert container state - existing = await session.get(DockerContainer, c["name"]) - if existing is None: - existing = DockerContainer(name=c["name"]) - session.add(existing) - existing.container_id = c["container_id"] - existing.image = c["image"] - existing.status = c["status"] - existing.cpu_pct = c["cpu_pct"] - existing.mem_usage_bytes = c["mem_usage_bytes"] - existing.mem_limit_bytes = c["mem_limit_bytes"] - existing.mem_pct = c["mem_pct"] - existing.restart_count = c["restart_count"] - existing.ports_json = json.dumps(c["ports"]) - existing.started_at = c["started_at"] - existing.scraped_at = now - - # Time-series metric (running containers only) - if c["status"] == "running" and c["cpu_pct"] is not None: - session.add(DockerMetric( - container_name=c["name"], - scraped_at=now, - cpu_pct=c["cpu_pct"], - mem_pct=c["mem_pct"] or 0.0, - mem_usage_bytes=c["mem_usage_bytes"] or 0, - )) - - # Feed alert pipeline - await record_metric( - session=session, - source_module="docker", - resource_name=c["name"], - metric_name="cpu_pct", - value=c["cpu_pct"], - ) - if c["mem_pct"] is not None: - await record_metric( - session=session, - source_module="docker", - resource_name=c["name"], - metric_name="mem_pct", - value=c["mem_pct"], - ) diff --git a/plugins/docker/scraper.py b/plugins/docker/scraper.py deleted file mode 100644 index f4133ed..0000000 --- a/plugins/docker/scraper.py +++ /dev/null @@ -1,139 +0,0 @@ -# plugins/docker/scraper.py -"""Docker API client via Unix socket.""" -from __future__ import annotations -import asyncio -import logging -from datetime import datetime, timezone - -import httpx - -logger = logging.getLogger(__name__) - - -def _short_id(container_id: str) -> str: - return container_id[:12] if container_id else "" - - -def _calc_cpu_pct(stats: dict) -> float: - """Calculate CPU % from a Docker stats snapshot (one-shot).""" - try: - cpu = stats.get("cpu_stats", {}) - precpu = stats.get("precpu_stats", {}) - cpu_delta = ( - cpu["cpu_usage"]["total_usage"] - precpu["cpu_usage"]["total_usage"] - ) - sys_delta = cpu.get("system_cpu_usage", 0) - precpu.get("system_cpu_usage", 0) - num_cpus = cpu.get("online_cpus") or len( - cpu["cpu_usage"].get("percpu_usage") or [None] - ) - if sys_delta <= 0 or cpu_delta < 0: - return 0.0 - return round((cpu_delta / sys_delta) * num_cpus * 100.0, 2) - except (KeyError, TypeError, ZeroDivisionError): - return 0.0 - - -def _calc_mem(stats: dict) -> tuple[int, int, float]: - """Return (usage_bytes, limit_bytes, mem_pct) from Docker stats.""" - try: - mem = stats.get("memory_stats", {}) - usage = mem.get("usage", 0) - limit = mem.get("limit", 0) - # Subtract page cache for working set (cgroup v1: "cache", v2: "inactive_file") - cache = mem.get("stats", {}).get("cache", 0) or \ - mem.get("stats", {}).get("inactive_file", 0) - actual = max(0, usage - cache) - pct = round(actual / limit * 100.0, 2) if limit > 0 else 0.0 - return actual, limit, pct - except (KeyError, TypeError): - return 0, 0, 0.0 - - -def _parse_ports(ports: list) -> list[dict]: - """Normalise Docker port bindings to a compact list.""" - result = [] - for p in (ports or []): - if p.get("PublicPort"): - result.append({ - "host_port": p["PublicPort"], - "container_port": p["PrivatePort"], - "protocol": p.get("Type", "tcp"), - }) - return result - - -async def scrape_docker(socket_path: str, include_stopped: bool) -> list[dict]: - """ - Fetch all containers + resource stats from the Docker daemon. - Returns a list of normalised dicts ready for the scheduler to persist. - Raises ConnectionError if the socket is unreachable. - """ - transport = httpx.AsyncHTTPTransport(uds=socket_path) - try: - async with httpx.AsyncClient( - transport=transport, - base_url="http://docker", - timeout=10.0, - ) as client: - resp = await client.get( - "/containers/json", - params={"all": "true" if include_stopped else "false"}, - ) - resp.raise_for_status() - containers = resp.json() - - async def _get_stats(c: dict) -> tuple[dict, dict | None]: - if c.get("State") != "running": - return c, None - try: - r = await client.get( - f"/containers/{c['Id']}/stats", - params={"stream": "false", "one-shot": "true"}, - timeout=5.0, - ) - return c, r.json() - except Exception as exc: - logger.debug( - "Stats unavailable for %s: %s", _short_id(c.get("Id", "")), exc - ) - return c, None - - pairs = await asyncio.gather(*[_get_stats(c) for c in containers]) - - except httpx.ConnectError as exc: - raise ConnectionError( - f"Cannot connect to Docker socket at {socket_path}: {exc}" - ) from exc - - results = [] - for container, stats in pairs: - names = container.get("Names") or [] - name = names[0].lstrip("/") if names else _short_id(container.get("Id", "")) - - cpu_pct = mem_usage = mem_limit = mem_pct = None - if stats is not None: - cpu_pct = _calc_cpu_pct(stats) - mem_usage, mem_limit, mem_pct = _calc_mem(stats) - - # Created is a Unix epoch int from /containers/json - created_ts = container.get("Created") - started_at = ( - datetime.fromtimestamp(created_ts, tz=timezone.utc) - if isinstance(created_ts, (int, float)) else None - ) - - results.append({ - "name": name, - "container_id": _short_id(container.get("Id", "")), - "image": container.get("Image", ""), - "status": container.get("State", "unknown"), - "cpu_pct": cpu_pct, - "mem_usage_bytes": mem_usage, - "mem_limit_bytes": mem_limit, - "mem_pct": mem_pct, - "restart_count": 0, # would require /inspect; left as 0 for now - "ports": _parse_ports(container.get("Ports", [])), - "started_at": started_at, - }) - - return results diff --git a/plugins/docker/templates/docker/host_panel.html b/plugins/docker/templates/docker/host_panel.html new file mode 100644 index 0000000..1666174 --- /dev/null +++ b/plugins/docker/templates/docker/host_panel.html @@ -0,0 +1,22 @@ +{# docker/host_panel.html — per-host Docker fragment embedded on the Hosts hub #} +
+
+

Docker

+ + {{ running }} running{% if stopped %} · {{ stopped }} stopped{% endif %} + All → + +
+
+ {% for c in containers %} +
+ + {{ c.name }} + + {% if c.cpu_pct is not none %}{{ "%.1f" | format(c.cpu_pct) }}%{% endif %} + {% if c.mem_pct is not none %} · {{ "%.1f" | format(c.mem_pct) }}%{% endif %} + +
+ {% endfor %} +
+
diff --git a/plugins/docker/templates/docker/rows.html b/plugins/docker/templates/docker/rows.html index 3b6c84e..686609e 100644 --- a/plugins/docker/templates/docker/rows.html +++ b/plugins/docker/templates/docker/rows.html @@ -1,4 +1,4 @@ -{# docker/rows.html — HTMX fragment for the Docker main page #} +{# docker/rows.html — HTMX fragment for the Docker main page, grouped by host #} {# ── Summary strip ─────────────────────────────────────────────────────────── #}
@@ -12,13 +12,28 @@
Total
- {{ container_data | length }} + {{ total }} +
+
+
Hosts
+ {{ host_groups | length }}
-{# ── Container table ─────────────────────────────────────────────────────── #} -{% if container_data %} -
+{# ── Container tables, one per host ───────────────────────────────────────── #} +{% if host_groups %} +{% for g in host_groups %} +
+
+ {% if g.host %} + {{ g.host.name }} + {% else %} + {{ g.host_name }} + {% endif %} + + {{ g.running }} running{% if g.stopped %} · {{ g.stopped }} stopped{% endif %} + +
@@ -32,7 +47,7 @@ - {% for item in container_data %} + {% for item in g.containers %} {% set c = item.container %}
@@ -77,10 +92,13 @@
+{% endfor %} {% else %}
- No containers found. Make sure the Docker socket is accessible and the plugin is configured correctly. + No containers reported yet. Containers are collected by the Steward host agent — + deploy the agent to a host running Docker (Hosts → the host → agent panel) and + its containers will appear here under that host.
{% endif %} diff --git a/plugins/docker/templates/docker/widget.html b/plugins/docker/templates/docker/widget.html index 061df4b..9a8a5a4 100644 --- a/plugins/docker/templates/docker/widget.html +++ b/plugins/docker/templates/docker/widget.html @@ -1,6 +1,6 @@ -{# docker/widget.html — dashboard widget: container status overview #} -{% if not containers and running_count == 0 and stopped_count == 0 %} -

No containers found.

+{# docker/widget.html — dashboard widget: container status overview, by host #} +{% if running_count == 0 and stopped_count == 0 %} +

No containers reported yet.

{% else %}
@@ -16,8 +16,12 @@ {% endif %}
+{% for g in host_groups %} +{% if multi_host %} +
{{ g.host_name }}
+{% endif %}
- {% for c in containers %} + {% for c in g.containers %}
@@ -31,5 +35,6 @@
{% endfor %}
+{% endfor %} {% endif %} diff --git a/plugins/docker/templates/docker/widget_resources.html b/plugins/docker/templates/docker/widget_resources.html index 39c90dc..3358bc3 100644 --- a/plugins/docker/templates/docker/widget_resources.html +++ b/plugins/docker/templates/docker/widget_resources.html @@ -1,13 +1,14 @@ -{# docker/widget_resources.html — dashboard widget: CPU + memory bars for running containers #} -{% if not containers %} +{# docker/widget_resources.html — dashboard widget: CPU + memory bars, busiest containers #} +{% if not rows %}
No running containers.
{% else %}
- {% for c in containers %} + {% for r in rows %} + {% set c = r.c %}
- {{ c.name }} + {{ c.name }}{% if multi_host %} · {{ r.host_name }}{% endif %} {% if c.cpu_pct is not none %}CPU {{ "%.1f" | format(c.cpu_pct) }}%{% endif %} diff --git a/plugins/host_agent/agent.py b/plugins/host_agent/agent.py index f78cef0..3e8356d 100644 --- a/plugins/host_agent/agent.py +++ b/plugins/host_agent/agent.py @@ -19,7 +19,12 @@ import urllib.request from collections import deque from datetime import datetime, timezone -AGENT_VERSION = "1.2.0" +AGENT_VERSION = "1.3.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 +# unreadable, so this never needs setting on a Docker-less host (zero-config). +DEFAULT_DOCKER_SOCKET = "/var/run/docker.sock" class ConfigError(Exception): @@ -62,6 +67,7 @@ def read_config(path: str) -> dict: raise ConfigError(f"{path}: missing required key(s): {', '.join(missing)}") cfg.setdefault("interval_seconds", 30) + cfg.setdefault("docker_socket", DEFAULT_DOCKER_SOCKET) return cfg @@ -386,6 +392,180 @@ def _rates(cur: dict, prev: dict, dt: float) -> dict: return out +# ─── docker ────────────────────────────────────────────────────────────────── +# +# Per-host container collection over the local Docker socket. stdlib-only: a +# minimal HTTP/1.1 client over an AF_UNIX socket. We send `Connection: close` +# so the daemon closes the socket at end of response and we can read to EOF; +# Docker still frames the body with Transfer-Encoding: chunked, so we de-chunk +# when that header is present rather than assume Content-Length. + +DOCKER_API_TIMEOUT = 5.0 + + +def _dechunk(body: bytes) -> bytes: + """Decode an HTTP/1.1 chunked-transfer body into the raw payload.""" + out = bytearray() + while body: + line, sep, rest = body.partition(b"\r\n") + if not sep: + break + try: + size = int(line.split(b";", 1)[0], 16) # ignore chunk extensions + except ValueError: + break + if size == 0: + break + out += rest[:size] + body = rest[size + 2:] # skip the chunk data and its trailing CRLF + return bytes(out) + + +def _docker_request(socket_path: str, path: str, timeout: float = DOCKER_API_TIMEOUT): + """GET `path` from the Docker Engine API over a Unix socket; return JSON. + + Raises OSError on any socket/transport problem (absent socket, permission + denied, non-200) and ValueError on a non-JSON body — both are caught by the + caller and treated as "no docker here", so collection degrades silently. + """ + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.settimeout(timeout) + try: + sock.connect(socket_path) + req = ( + "GET " + path + " HTTP/1.1\r\n" + "Host: docker\r\n" + "Accept: application/json\r\n" + "Connection: close\r\n" + "\r\n" + ) + sock.sendall(req.encode("ascii")) + chunks = [] + while True: + buf = sock.recv(65536) + if not buf: + break + chunks.append(buf) + finally: + sock.close() + + raw = b"".join(chunks) + head, _, body = raw.partition(b"\r\n\r\n") + header_text = head.decode("latin-1") + status_line = header_text.split("\r\n", 1)[0] + parts = status_line.split(None, 2) # "HTTP/1.1 200 OK" + status = int(parts[1]) if len(parts) >= 2 and parts[1].isdigit() else 0 + if status != 200: + raise OSError(f"docker API {path} returned {status}") + if "transfer-encoding: chunked" in header_text.lower(): + body = _dechunk(body) + return json.loads(body.decode("utf-8")) + + +def _docker_cpu_pct(stats: dict) -> float: + """CPU % from a Docker stats snapshot (delta of container vs system CPU). + + Ported verbatim from the old central scraper's math; correct as long as both + cpu_stats and precpu_stats are populated (we request without `one-shot` so + the daemon fills precpu over one cycle). + """ + try: + cpu = stats.get("cpu_stats", {}) + precpu = stats.get("precpu_stats", {}) + cpu_delta = ( + cpu["cpu_usage"]["total_usage"] - precpu["cpu_usage"]["total_usage"] + ) + sys_delta = cpu.get("system_cpu_usage", 0) - precpu.get("system_cpu_usage", 0) + num_cpus = cpu.get("online_cpus") or len( + cpu["cpu_usage"].get("percpu_usage") or [None] + ) + if sys_delta <= 0 or cpu_delta < 0: + return 0.0 + return round((cpu_delta / sys_delta) * num_cpus * 100.0, 2) + except (KeyError, TypeError, ZeroDivisionError): + return 0.0 + + +def _docker_mem(stats: dict): + """Return (usage_bytes, limit_bytes, mem_pct); working set excludes cache.""" + try: + mem = stats.get("memory_stats", {}) + usage = mem.get("usage", 0) + limit = mem.get("limit", 0) + # Subtract page cache for working set (cgroup v1 "cache", v2 "inactive_file"). + cache = mem.get("stats", {}).get("cache", 0) or \ + mem.get("stats", {}).get("inactive_file", 0) + actual = max(0, usage - cache) + pct = round(actual / limit * 100.0, 2) if limit > 0 else 0.0 + return actual, limit, pct + except (KeyError, TypeError): + return 0, 0, 0.0 + + +def _docker_ports(ports: list) -> list: + """Normalise Docker port bindings to a compact published-ports list.""" + out = [] + for p in (ports or []): + if p.get("PublicPort"): + out.append({ + "host_port": p["PublicPort"], + "container_port": p.get("PrivatePort"), + "protocol": p.get("Type", "tcp"), + }) + return out + + +def collect_docker(socket_path: str) -> list: + """Per-container state from the local Docker socket, or [] if unavailable. + + Absent socket / permission denied / non-Docker endpoint all yield [] — the + agent silently reports no containers rather than erroring, so a Docker-less + host needs no configuration. + """ + try: + containers = _docker_request(socket_path, "/containers/json?all=true") + except (OSError, ValueError): + return [] + if not isinstance(containers, list): + 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 + + # ─── ring buffer ───────────────────────────────────────────────────────────── @@ -408,11 +588,14 @@ class RingBuffer: # ─── payload ───────────────────────────────────────────────────────────────── -def build_sample(mounts: list[str], state: dict) -> dict: +def build_sample(mounts: list[str], state: dict, + docker_socket: str = DEFAULT_DOCKER_SOCKET) -> dict: """Collect one full sample. Partial samples allowed if a collector fails. `state` carries the previous network/disk counters + monotonic timestamp so throughput rates can be derived from deltas; it is mutated in place. + `docker_socket` is probed best-effort — the `docker` key is omitted entirely + when no containers are found, so non-Docker hosts add nothing to the payload. """ sample: dict = {"ts": datetime.now(timezone.utc).isoformat()} try: @@ -466,6 +649,14 @@ def build_sample(mounts: list[str], state: dict) -> dict: for n, (rd, wr) in _rates(disk_raw, state.get("disk", {}), dt).items() ] state["net"], state["disk"], state["mono"] = net_raw, disk_raw, now_mono + + try: + docker = collect_docker(docker_socket) + except Exception: + docker = [] + if docker: + sample["docker"] = docker + return sample @@ -547,6 +738,7 @@ def main_loop(conf_path: str) -> int: metadata = collect_metadata() hostname = cfg.get("hostname") or socket.gethostname() mounts = cfg.get("mounts") or default_mounts() + docker_socket = cfg.get("docker_socket") or DEFAULT_DOCKER_SOCKET buffer = RingBuffer(maxlen=20) backoff = 0 # Carries previous net/disk counters + monotonic ts for rate computation. @@ -560,13 +752,14 @@ def main_loop(conf_path: str) -> int: try: cfg = read_config(conf_path) mounts = cfg.get("mounts") or default_mounts() + docker_socket = cfg.get("docker_socket") or DEFAULT_DOCKER_SOCKET metadata = collect_metadata() # refresh host_ip/distro on reload _log("INFO", "config reloaded") except ConfigError as e: _log("ERROR", f"reload failed: {e}") _reload_requested = False - sample = build_sample(mounts, rate_state) + sample = build_sample(mounts, rate_state, docker_socket) buffered = buffer.drain() payload = build_payload( samples=buffered + [sample], diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index d8c66b1..9670569 100644 --- a/plugins/host_agent/routes.py +++ b/plugins/host_agent/routes.py @@ -196,6 +196,7 @@ async def ingest(): accepted = 0 latest_ts: datetime | None = None + docker_snapshots: list[tuple[datetime, list]] = [] for sample in samples: try: recorded_at = _parse_ts(sample["ts"]) @@ -204,6 +205,9 @@ async def ingest(): metrics = _expand_sample_to_metrics(sample, host.name, recorded_at) for m in metrics: session.add(m) + docker = sample.get("docker") + if isinstance(docker, list) and docker: + docker_snapshots.append((recorded_at, docker)) accepted += 1 if latest_ts is None or recorded_at > latest_ts: latest_ts = recorded_at @@ -212,6 +216,26 @@ async def ingest(): await session.rollback() return _error(400, "malformed_payload", "no valid samples") + # Hand host-scoped container data to the docker plugin if it's enabled + # (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: + from steward.core.capabilities import has_capability, invoke_capability + if has_capability("docker.persist_host_samples"): + try: + # SAVEPOINT so a docker-side failure rolls back only the + # docker writes, leaving the host metrics intact to commit. + async with session.begin_nested(): + await invoke_capability( + "docker.persist_host_samples", UserRole.admin, + session, host, docker_snapshots, + ) + except Exception: + current_app.logger.exception( + "host_agent ingest: docker persist failed for host=%s", + host.name) + md = payload.get("metadata") or {} changed = False if latest_ts and (reg.last_seen_at is None or latest_ts > reg.last_seen_at): diff --git a/plugins/index.yaml b/plugins/index.yaml index 0198bc3..3d5fe8e 100644 --- a/plugins/index.yaml +++ b/plugins/index.yaml @@ -14,14 +14,14 @@ updated: "2026-03-22" plugins: - name: docker - version: "1.0.0" - description: "Docker container status, resource usage, and restart tracking via Docker socket" + version: "2.0.0" + description: "Per-host Docker container status + resource usage, collected by the Steward host agent" author: "Steward" license: "MIT" min_app_version: "0.1.0" repository_url: "https://git.fabledsword.com/bvandeusen/Steward-plugins" homepage: "https://git.fabledsword.com/bvandeusen/Steward-plugins/src/branch/main/docker" - download_url: "https://git.fabledsword.com/bvandeusen/Steward-plugins/releases/download/docker-v1.0.0/docker.zip" + download_url: "https://git.fabledsword.com/bvandeusen/Steward-plugins/releases/download/docker-v2.0.0/docker.zip" checksum_sha256: "" tags: - containers diff --git a/steward/ansible/bundled/host_agent/install.yml b/steward/ansible/bundled/host_agent/install.yml index a1074d2..ce50939 100644 --- a/steward/ansible/bundled/host_agent/install.yml +++ b/steward/ansible/bundled/host_agent/install.yml @@ -29,6 +29,23 @@ shell: /usr/sbin/nologin create_home: false + # Let the agent read the local Docker socket so it can report this host's + # containers (the per-host docker collection path). Best-effort: skipped on + # hosts without Docker. Restart picks up the new supplementary group. + - name: Detect docker group + ansible.builtin.command: getent group docker + register: _docker_grp + changed_when: false + failed_when: false + + - name: Add steward-agent to the docker group + ansible.builtin.user: + name: steward-agent + groups: docker + append: true + when: _docker_grp.rc == 0 + notify: restart steward-agent + - name: Create agent directory ansible.builtin.file: path: /usr/local/lib/steward-agent diff --git a/steward/ansible/bundled/host_agent/provision.yml b/steward/ansible/bundled/host_agent/provision.yml index 09b50ec..21b62ac 100644 --- a/steward/ansible/bundled/host_agent/provision.yml +++ b/steward/ansible/bundled/host_agent/provision.yml @@ -85,6 +85,23 @@ shell: /usr/sbin/nologin create_home: false + # Let the agent read the local Docker socket so it can report this host's + # containers (the per-host docker collection path). Best-effort: skipped on + # hosts without Docker. Restart picks up the new supplementary group. + - name: Detect docker group + ansible.builtin.command: getent group docker + register: _docker_grp + changed_when: false + failed_when: false + + - name: Add steward-agent to the docker group + ansible.builtin.user: + name: steward-agent + groups: docker + append: true + when: _docker_grp.rc == 0 + notify: restart steward-agent + - name: Create agent directory ansible.builtin.file: path: /usr/local/lib/steward-agent diff --git a/steward/app.py b/steward/app.py index 7410bbb..29b26f4 100644 --- a/steward/app.py +++ b/steward/app.py @@ -167,7 +167,17 @@ def create_app( @app.context_processor def _inject_plugin_failures(): from .core.plugin_manager import get_plugin_failures - return {"plugin_failures": get_plugin_failures()} + # enabled_plugins lets templates gate cross-plugin embeds (e.g. the + # Hosts hub only fetches the docker fragment when docker is enabled), + # avoiding a 404 to a route whose blueprint isn't registered. + enabled_plugins = { + name for name, cfg in (app.config.get("PLUGINS") or {}).items() + if isinstance(cfg, dict) and cfg.get("enabled") + } + return { + "plugin_failures": get_plugin_failures(), + "enabled_plugins": enabled_plugins, + } # ── 11. Share-token middleware ───────────────────────────────────────────── @app.before_request diff --git a/steward/templates/hosts/detail.html b/steward/templates/hosts/detail.html index 9651b43..81fb23e 100644 --- a/steward/templates/hosts/detail.html +++ b/steward/templates/hosts/detail.html @@ -122,6 +122,11 @@
Loading agent…
+{# ── Docker (docker plugin fragment; renders nothing if the host has none) ──── #} +{% if "docker" in enabled_plugins %} +
+{% endif %} + {# ── Ansible ──────────────────────────────────────────────────────────────── #}

Ansible

diff --git a/tests/integration/test_docker.py b/tests/integration/test_docker.py new file mode 100644 index 0000000..9271e3c --- /dev/null +++ b/tests/integration/test_docker.py @@ -0,0 +1,109 @@ +"""Integration: per-host Docker schema + host-scoped persistence. + +Validates the docker_002 migration (host_id columns + composite key) applied and +that persist_host_docker scopes containers/metrics by host so same-named +containers on different hosts no longer collide. Requires STEWARD_DATABASE_URL. +""" +from __future__ import annotations + +import asyncio +import os +import uuid +from datetime import datetime, timezone + +import pytest + +pytestmark = pytest.mark.integration + +_NEEDS_DB = pytest.mark.skipif( + not os.environ.get("STEWARD_DATABASE_URL"), + reason="integration test needs a live Postgres (STEWARD_DATABASE_URL)", +) + + +@pytest.fixture +def app(): + if not os.environ.get("STEWARD_DATABASE_URL"): + pytest.skip("needs Postgres") + from steward.app import create_app + return create_app(testing=False) + + +@_NEEDS_DB +def test_docker_tables_are_host_scoped(app): + from sqlalchemy import text + + async def _go(): + async with app.db_sessionmaker() as s: + # Both tables carry a host_id column after docker_002. + for tbl in ("docker_containers", "docker_metrics"): + cols = {r[0] for r in (await s.execute(text( + "SELECT column_name FROM information_schema.columns " + "WHERE table_name = :t"), {"t": tbl})).all()} + assert "host_id" in cols, f"{tbl} missing host_id" + # docker_containers is keyed by (host_id, name), not name alone. + pk_cols = {r[0] for r in (await s.execute(text( + "SELECT a.attname FROM pg_index i " + "JOIN pg_attribute a ON a.attrelid = i.indrelid " + " AND a.attnum = ANY(i.indkey) " + "WHERE i.indrelid = 'docker_containers'::regclass AND i.indisprimary" + ))).all()} + assert pk_cols == {"host_id", "name"} + + asyncio.run(_go()) + + +def _persist_fn(app): + """Resolve persist_host_docker via the registered capability if the docker + plugin is loaded, else import it directly (the import is safe only when the + plugin is NOT loaded — otherwise its models are already mapped).""" + from steward.core.capabilities import has_capability, get_capability + if has_capability("docker.persist_host_samples"): + return get_capability("docker.persist_host_samples").fn + from plugins.docker.ingest import persist_host_docker + return persist_host_docker + + +@_NEEDS_DB +def test_persist_scopes_containers_by_host(app): + from sqlalchemy import text + from steward.models.hosts import Host + + persist = _persist_fn(app) + now = datetime.now(timezone.utc) + snapshot = [(now, [{ + "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, + }])] + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(text("DELETE FROM docker_metrics")) + await s.execute(text("DELETE FROM docker_containers")) + await s.execute(text( + "DELETE FROM plugin_metrics WHERE source_module = 'docker'")) + h1 = Host(id=str(uuid.uuid4()), name="alpha", address="10.0.0.1") + h2 = Host(id=str(uuid.uuid4()), name="beta", address="10.0.0.2") + s.add_all([h1, h2]) + await s.flush() + # Same container name on two hosts — must not collide now. + await persist(s, h1, snapshot) + await persist(s, h2, snapshot) + + web_rows = (await s.execute(text( + "SELECT COUNT(*) FROM docker_containers WHERE name = 'web'"))).scalar() + metric_hosts = (await s.execute(text( + "SELECT COUNT(DISTINCT host_id) FROM docker_metrics " + "WHERE container_name = 'web'"))).scalar() + # Alert pipeline received host-scoped resource names. + 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 + + web_rows, metric_hosts, resources = 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"} diff --git a/tests/plugins/host_agent/test_agent_docker.py b/tests/plugins/host_agent/test_agent_docker.py new file mode 100644 index 0000000..6ab4b7f --- /dev/null +++ b/tests/plugins/host_agent/test_agent_docker.py @@ -0,0 +1,157 @@ +# tests/plugins/host_agent/test_agent_docker.py +"""Unit tests for the agent's stdlib Docker collector. + +No socket / no network — the UDS request layer is monkeypatched so we exercise +the cpu/mem math, chunked-body decoding, sample assembly, and the silent-skip +contract in isolation. +""" +from plugins.host_agent import agent as a + + +# ── chunked transfer decoding ───────────────────────────────────────────────── + +def test_dechunk_reassembles_body(): + chunked = b"4\r\nWiki\r\n5\r\npedia\r\n0\r\n\r\n" + assert a._dechunk(chunked) == b"Wikipedia" + + +def test_dechunk_ignores_chunk_extensions(): + assert a._dechunk(b"3;name=v\r\nabc\r\n0\r\n\r\n") == b"abc" + + +# ── cpu / mem math (ported from the old central scraper) ────────────────────── + +def test_docker_cpu_pct(): + stats = { + "cpu_stats": { + "cpu_usage": {"total_usage": 200_000_000}, + "system_cpu_usage": 2_000_000_000, + "online_cpus": 4, + }, + "precpu_stats": { + "cpu_usage": {"total_usage": 100_000_000}, + "system_cpu_usage": 1_000_000_000, + }, + } + # cpu_delta=1e8, sys_delta=1e9 → 0.1 * 4 cpus * 100 = 40.0% + assert a._docker_cpu_pct(stats) == 40.0 + + +def test_docker_cpu_pct_guards_bad_stats(): + assert a._docker_cpu_pct({}) == 0.0 + # No forward progress in system time → 0, never a divide error. + assert a._docker_cpu_pct({ + "cpu_stats": {"cpu_usage": {"total_usage": 5}, "system_cpu_usage": 10}, + "precpu_stats": {"cpu_usage": {"total_usage": 5}, "system_cpu_usage": 10}, + }) == 0.0 + + +def test_docker_mem_subtracts_cache(): + stats = {"memory_stats": { + "usage": 200 * 1024 * 1024, + "limit": 1000 * 1024 * 1024, + "stats": {"cache": 50 * 1024 * 1024}, + }} + usage, limit, pct = a._docker_mem(stats) + assert usage == 150 * 1024 * 1024 # working set = usage − cache + assert limit == 1000 * 1024 * 1024 + assert pct == 15.0 + + +def test_docker_ports_keeps_only_published(): + ports = [ + {"PublicPort": 8080, "PrivatePort": 80, "Type": "tcp"}, + {"PrivatePort": 5432, "Type": "tcp"}, # unpublished → dropped + ] + assert a._docker_ports(ports) == [ + {"host_port": 8080, "container_port": 80, "protocol": "tcp"}, + ] + + +# ── collect_docker assembly + silent skip ───────────────────────────────────── + +def test_collect_docker_silent_skip_when_no_socket(): + # Absent socket path must degrade to [] (zero-config on Docker-less hosts). + assert a.collect_docker("/nonexistent/does-not-exist.sock") == [] + + +def test_collect_docker_assembles_container(monkeypatch): + container = { + "Id": "abc123def4567890", + "Names": ["/web"], + "Image": "nginx:latest", + "State": "running", + "Created": 1_700_000_000, + "Ports": [{"PublicPort": 8080, "PrivatePort": 80, "Type": "tcp"}], + } + stats = { + "cpu_stats": {"cpu_usage": {"total_usage": 200_000_000}, + "system_cpu_usage": 2_000_000_000, "online_cpus": 4}, + "precpu_stats": {"cpu_usage": {"total_usage": 100_000_000}, + "system_cpu_usage": 1_000_000_000}, + "memory_stats": {"usage": 200 * 1024 * 1024, "limit": 1000 * 1024 * 1024, + "stats": {"cache": 50 * 1024 * 1024}}, + } + + def fake_request(socket_path, path, timeout=a.DOCKER_API_TIMEOUT): + return [container] if path.startswith("/containers/json") else stats + + monkeypatch.setattr(a, "_docker_request", fake_request) + out = a.collect_docker("/var/run/docker.sock") + assert len(out) == 1 + c = out[0] + assert c["name"] == "web" + assert c["container_id"] == "abc123def456" # 12-char short id + assert c["image"] == "nginx:latest" + assert c["status"] == "running" + assert c["cpu_pct"] == 40.0 + 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 + + +def test_collect_docker_skips_stats_for_stopped(monkeypatch): + container = {"Id": "x" * 16, "Names": ["/db"], "Image": "pg", "State": "exited", + "Created": None, "Ports": []} + + def fake_request(socket_path, path, timeout=a.DOCKER_API_TIMEOUT): + if path.startswith("/containers/json"): + return [container] + 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 + + +# ── build_sample wiring ─────────────────────────────────────────────────────── + +def test_build_sample_includes_docker_when_present(monkeypatch): + monkeypatch.setattr(a, "collect_docker", lambda _s: [{"name": "web", "status": "running"}]) + sample = a.build_sample(["/"], {}, "/var/run/docker.sock") + assert sample["docker"] == [{"name": "web", "status": "running"}] + + +def test_build_sample_omits_docker_when_empty(monkeypatch): + monkeypatch.setattr(a, "collect_docker", lambda _s: []) + sample = a.build_sample(["/"], {}, "/var/run/docker.sock") + assert "docker" not in sample + + +# ── config ──────────────────────────────────────────────────────────────────── + +def test_read_config_defaults_docker_socket(tmp_path): + cfg_file = tmp_path / "agent.conf" + cfg_file.write_text("url = https://steward.example\ntoken = abc\n") + cfg = a.read_config(str(cfg_file)) + assert cfg["docker_socket"] == a.DEFAULT_DOCKER_SOCKET + + +def test_read_config_honours_explicit_docker_socket(tmp_path): + cfg_file = tmp_path / "agent.conf" + cfg_file.write_text( + "url = https://steward.example\ntoken = abc\n" + "docker_socket = /run/user/1000/docker.sock\n") + cfg = a.read_config(str(cfg_file)) + assert cfg["docker_socket"] == "/run/user/1000/docker.sock"