diff --git a/plugins/docker/ingest.py b/plugins/docker/ingest.py index cad4a6b..1e57619 100644 --- a/plugins/docker/ingest.py +++ b/plugins/docker/ingest.py @@ -2,16 +2,23 @@ """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. +so the host_agent plugin can hand off a sample's `docker` array (and, on a swarm +manager, its `swarm` object) 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 SAVEPOINT); +never opens or commits its own. """ from __future__ import annotations import json from datetime import datetime +from sqlalchemy import delete, select + + +# Stopped/terminal container states (anything not "running"/"paused"/"restarting"). +_STOPPED_STATES = {"exited", "dead", "stopped"} + def _parse_started_at(value) -> datetime | None: """Parse the agent's ISO-8601 started_at string back to a datetime.""" @@ -23,18 +30,142 @@ def _parse_started_at(value) -> datetime | None: return None -async def persist_host_docker(session, host, snapshots) -> None: - """Upsert containers + append time-series for one host's docker snapshots. +def _derive_events(old_state: dict, new_containers: list) -> list: + """Diff a fresh snapshot against stored per-container state → lifecycle events. + + Pure function (no DB) so it's unit-testable. `old_state` maps name → the + previously stored {status, health, oom_killed, exit_code}; `new_containers` + is the newest snapshot's list of container dicts. Returns a list of + (container_name, event, detail) tuples: + + start — a container transitions into running (or a genuinely new + container appears already running) + stop — running → not-running, or a running container is removed + die — same as stop but with a non-zero exit code (abnormal) + oom — OOMKilled flips False→True + health_change — HEALTHCHECK status string changes + + The CALLER skips this entirely on a host's first-ever snapshot (empty + old_state) so the baseline doesn't emit a spurious "start" per existing + container — a later-appearing container still gets one because by then + old_state is populated and that container's prior entry is simply absent. + """ + events: list = [] + new_by_name = {c["name"]: c for c in new_containers if c.get("name")} + + for name, c in new_by_name.items(): + new_status = (c.get("status") or "").lower() + new_running = new_status == "running" + new_oom = bool(c.get("oom_killed")) + new_health = c.get("health") + new_exit = c.get("exit_code") + old = old_state.get(name) + + if old is None: + # Newly observed container — only a start is meaningful (we have no + # prior state to diff a death against). + if new_running: + events.append((name, "start", c.get("image") or None)) + continue + + old_running = (old.get("status") or "").lower() == "running" + if new_running and not old_running: + events.append((name, "start", c.get("image") or None)) + elif old_running and not new_running: + if new_exit not in (None, 0): + events.append((name, "die", f"exit {new_exit}")) + else: + events.append((name, "stop", None)) + + if new_oom and not bool(old.get("oom_killed")): + events.append((name, "oom", None)) + + if new_health != old.get("health") and (new_health or old.get("health")): + events.append((name, "health_change", + f"{old.get('health') or '?'}→{new_health or '?'}")) + + # A running container that vanished from the listing entirely (removed). + for name, old in old_state.items(): + if name not in new_by_name and (old.get("status") or "").lower() == "running": + events.append((name, "stop", "removed")) + + return events + + +async def _persist_swarm(session, host, swarm: dict) -> None: + """Upsert this manager's swarm topology; drop rows no longer reported. + + Current-state tables (not time-series): a manager re-reports its full + services/nodes set every sample, so we upsert what's present and delete what + isn't (scoped to this host so two managers don't clobber each other). + """ + from datetime import timezone + from .models import DockerSwarmNode, DockerSwarmService + + now = datetime.now(timezone.utc) + + services = swarm.get("services") or [] + seen_services: set[str] = set() + for s in services: + name = s.get("service_name") + if not name: + continue + seen_services.add(name) + row = await session.get(DockerSwarmService, (host.id, name)) + if row is None: + row = DockerSwarmService(host_id=host.id, service_name=name) + session.add(row) + row.mode = s.get("mode") or "replicated" + row.desired = int(s.get("desired") or 0) + row.running = int(s.get("running") or 0) + row.image = s.get("image") + row.placement_json = json.dumps(s.get("placement") or []) + row.updated_at = now + stale_services = delete(DockerSwarmService).where( + DockerSwarmService.host_id == host.id) + if seen_services: + stale_services = stale_services.where( + DockerSwarmService.service_name.notin_(seen_services)) + await session.execute(stale_services) + + nodes = swarm.get("nodes") or [] + seen_nodes: set[str] = set() + for n in nodes: + nid = n.get("node_id") + if not nid: + continue + seen_nodes.add(nid) + row = await session.get(DockerSwarmNode, (host.id, nid)) + if row is None: + row = DockerSwarmNode(host_id=host.id, node_id=nid) + session.add(row) + row.hostname = n.get("hostname") or "" + row.role = n.get("role") or "worker" + row.availability = n.get("availability") or "active" + row.status = n.get("status") or "unknown" + row.leader = bool(n.get("leader", False)) + row.updated_at = now + stale_nodes = delete(DockerSwarmNode).where(DockerSwarmNode.host_id == host.id) + if seen_nodes: + stale_nodes = stale_nodes.where(DockerSwarmNode.node_id.notin_(seen_nodes)) + await session.execute(stale_nodes) + + +async def persist_host_docker(session, host, snapshots, swarm=None) -> None: + """Upsert containers + time-series + lifecycle events + swarm for one host. `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). + one entry per ingested sample carrying docker data (usually one; more when + the agent flushes a backlog). Every snapshot contributes time-series + 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. """ from steward.core.alerts import record_metric - from .models import DockerContainer, DockerMetric + from .models import DockerContainer, DockerEvent, DockerMetric + + if swarm is not None: + await _persist_swarm(session, host, swarm) if not snapshots: return @@ -54,6 +185,24 @@ async def persist_host_docker(session, host, snapshots) -> None: mem_usage_bytes=c.get("mem_usage_bytes") or 0, )) + # Snapshot of stored per-container state BEFORE the upsert overwrites it — + # used to diff lifecycle events. Skip event derivation on the host's first + # snapshot (empty state) so we don't emit a start per pre-existing container. + old_rows = (await session.execute( + select(DockerContainer).where(DockerContainer.host_id == host.id) + )).scalars().all() + old_state = { + r.name: {"status": r.status, "health": r.health, + "oom_killed": r.oom_killed, "exit_code": r.exit_code} + for r in old_rows + } + if old_state: + for name, event, detail in _derive_events(old_state, latest_containers): + session.add(DockerEvent( + host_id=host.id, container_name=name, + event=event, detail=detail, at=latest_at, + )) + # Current state + alerts from the newest snapshot only. for c in latest_containers: name = c.get("name") @@ -90,8 +239,8 @@ async def persist_host_docker(session, host, snapshots) -> None: # Alert pipeline — resource is host-scoped so containers of the same name # on different hosts don't collide in the metric/alert namespace. + resource = f"{host.name}/{name}" 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"], @@ -101,3 +250,20 @@ async def persist_host_docker(session, host, snapshots) -> None: session=session, source_module="docker", resource_name=resource, metric_name="mem_pct", value=c["mem_pct"], ) + # Restart count is alertable regardless of state (crash-looping matters + # most while the container is down/restarting). + await record_metric( + session=session, source_module="docker", + resource_name=resource, metric_name="restart_count", + value=float(c.get("restart_count", 0) or 0), + ) + # Health → 1.0/0.0 only for containers that actually define a HEALTHCHECK + # (health is None otherwise — recording 0 would false-alarm every plain + # container). + health = c.get("health") + if health in ("healthy", "unhealthy", "starting"): + await record_metric( + session=session, source_module="docker", + resource_name=resource, metric_name="is_healthy", + value=1.0 if health == "healthy" else 0.0, + ) diff --git a/plugins/docker/migrations/versions/docker_005_swarm_placement.py b/plugins/docker/migrations/versions/docker_005_swarm_placement.py new file mode 100644 index 0000000..0e90bd5 --- /dev/null +++ b/plugins/docker/migrations/versions/docker_005_swarm_placement.py @@ -0,0 +1,29 @@ +"""Docker swarm service placement column + +Adds docker_swarm_services.placement_json — the task→node placement of a +service's running replicas, captured from the agent's swarm payload (a manager +sees every task, so this records cross-node placement the local container rows +can't). Additive column; no DROP+recreate. + +Revision ID: docker_005_swarm_placement +Revises: docker_004_events_swarm +Create Date: 2026-06-19 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "docker_005_swarm_placement" +down_revision: Union[str, None] = "docker_004_events_swarm" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column("docker_swarm_services", + sa.Column("placement_json", sa.Text, nullable=False, + server_default="[]")) + + +def downgrade() -> None: + op.drop_column("docker_swarm_services", "placement_json") diff --git a/plugins/docker/models.py b/plugins/docker/models.py index cf83bed..fb3dd63 100644 --- a/plugins/docker/models.py +++ b/plugins/docker/models.py @@ -145,6 +145,10 @@ class DockerSwarmService(Base): desired: Mapped[int] = mapped_column(Integer, nullable=False, default=0) running: Mapped[int] = mapped_column(Integer, nullable=False, default=0) image: Mapped[str | None] = mapped_column(String(512), nullable=True) + # task→node placement of running replicas: [{"node_id", "running"}], JSON. + # A manager sees every task, so this captures cross-node placement that the + # local docker_containers rows (this host only) can't. + placement_json: Mapped[str] = mapped_column(Text, nullable=False, default="[]") updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc), diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index 9670569..3e6c486 100644 --- a/plugins/host_agent/routes.py +++ b/plugins/host_agent/routes.py @@ -197,6 +197,8 @@ async def ingest(): accepted = 0 latest_ts: datetime | None = None docker_snapshots: list[tuple[datetime, list]] = [] + latest_swarm: dict | None = None + latest_swarm_ts: datetime | None = None for sample in samples: try: recorded_at = _parse_ts(sample["ts"]) @@ -208,6 +210,12 @@ async def ingest(): docker = sample.get("docker") if isinstance(docker, list) and docker: docker_snapshots.append((recorded_at, docker)) + # Swarm is current-state, not time-series — keep only the newest + # sample's topology (a manager re-reports it every interval). + swarm = sample.get("swarm") + if isinstance(swarm, dict) and ( + latest_swarm_ts is None or recorded_at > latest_swarm_ts): + latest_swarm, latest_swarm_ts = swarm, recorded_at accepted += 1 if latest_ts is None or recorded_at > latest_ts: latest_ts = recorded_at @@ -220,7 +228,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: + if docker_snapshots or latest_swarm is not None: from steward.core.capabilities import has_capability, invoke_capability if has_capability("docker.persist_host_samples"): try: @@ -229,7 +237,7 @@ async def ingest(): async with session.begin_nested(): await invoke_capability( "docker.persist_host_samples", UserRole.admin, - session, host, docker_snapshots, + session, host, docker_snapshots, latest_swarm, ) except Exception: current_app.logger.exception( diff --git a/steward/alerts/routes.py b/steward/alerts/routes.py index f1396d4..621c58a 100644 --- a/steward/alerts/routes.py +++ b/steward/alerts/routes.py @@ -24,7 +24,9 @@ METRIC_CATALOG: dict[str, list[str]] = { "traefik": ["request_rate", "error_rate", "latency_p50_ms", "latency_p95_ms", "latency_p99_ms", "response_bytes_rate", "cert_expiry_days"], "unifi": ["is_up", "latency_ms", "total_clients"], - "docker": ["cpu_pct", "mem_pct"], + # restart_count = cumulative restarts (alert on crash-looping); is_healthy = + # 1.0 healthy / 0.0 unhealthy from the container HEALTHCHECK (alert on <1). + "docker": ["cpu_pct", "mem_pct", "restart_count", "is_healthy"], "host_agent": [ "cpu_pct", "mem_used_pct", "mem_available_bytes", "swap_used_bytes", "disk_used_pct_worst", "load_1m", "load_5m", "load_15m", "uptime_secs", diff --git a/tests/integration/test_docker.py b/tests/integration/test_docker.py index 2865d65..21accd5 100644 --- a/tests/integration/test_docker.py +++ b/tests/integration/test_docker.py @@ -139,3 +139,83 @@ def test_persist_scopes_containers_by_host(app): assert metric_hosts == 2 # time-series rows scoped per host assert resources == {"alpha/web", "beta/web"} assert enrich == ("healthy", 2, "web", 1000) # enrichment round-trips + + +@_NEEDS_DB +def test_lifecycle_events_derived_across_snapshots(app): + from sqlalchemy import text + from steward.models.hosts import Host + + persist = _persist_fn(app) + t1 = datetime(2026, 6, 19, 12, 0, 0, tzinfo=timezone.utc) + t2 = datetime(2026, 6, 19, 12, 0, 30, tzinfo=timezone.utc) + running = {"name": "web", "container_id": "abc", "image": "nginx", + "status": "running", "cpu_pct": 5.0, "mem_pct": 10.0, + "restart_count": 0, "exit_code": None, "oom_killed": False, + "health": "healthy"} + died = {**running, "status": "exited", "cpu_pct": None, + "exit_code": 137, "oom_killed": True, "health": None} + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(text("DELETE FROM docker_events")) + await s.execute(text("DELETE FROM docker_metrics")) + await s.execute(text("DELETE FROM docker_containers")) + h = Host(id=str(uuid.uuid4()), name="evt", address="10.9.9.9") + s.add(h) + await s.flush() + # Baseline snapshot establishes state — must NOT emit a start. + await persist(s, h, [(t1, [running])]) + # Second snapshot: the container OOM-died. + await persist(s, h, [(t2, [died])]) + hid = h.id + rows = (await s.execute(text( + "SELECT event, detail FROM docker_events WHERE host_id = :h " + "ORDER BY event"), {"h": hid})).all() + return [(r[0], r[1]) for r in rows] + + events = asyncio.run(_go()) + # baseline emitted nothing; the death emitted both die (exit 137) and oom. + assert ("die", "exit 137") in events + assert ("oom", None) in events + assert not any(e[0] == "start" for e in events) + + +@_NEEDS_DB +def test_swarm_topology_persisted(app): + from sqlalchemy import text + from steward.models.hosts import Host + + persist = _persist_fn(app) + swarm = { + "services": [{"service_name": "web", "mode": "replicated", + "desired": 3, "running": 2, "image": "nginx", + "placement": [{"node_id": "n1", "running": 2}]}], + "nodes": [{"node_id": "n1", "hostname": "mgr", "role": "manager", + "availability": "active", "status": "ready", "leader": True}], + } + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(text("DELETE FROM docker_swarm_services")) + await s.execute(text("DELETE FROM docker_swarm_nodes")) + h = Host(id=str(uuid.uuid4()), name="mgr", address="10.8.8.8") + s.add(h) + await s.flush() + # No containers in this sample — swarm must still persist. + await persist(s, h, [], swarm) + hid = h.id + svc = (await s.execute(text( + "SELECT mode, desired, running, image, placement_json " + "FROM docker_swarm_services WHERE host_id = :h"), {"h": hid})).first() + node = (await s.execute(text( + "SELECT role, availability, status, leader " + "FROM docker_swarm_nodes WHERE host_id = :h"), {"h": hid})).first() + return svc, node + + svc, node = asyncio.run(_go()) + assert svc[0] == "replicated" and svc[1] == 3 and svc[2] == 2 and svc[3] == "nginx" + assert "n1" in svc[4] # placement JSON carries the node id + assert node[0] == "manager" and node[2] == "ready" and node[3] is True diff --git a/tests/plugins/docker/__init__.py b/tests/plugins/docker/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/plugins/docker/test_ingest_events.py b/tests/plugins/docker/test_ingest_events.py new file mode 100644 index 0000000..b5b50e8 --- /dev/null +++ b/tests/plugins/docker/test_ingest_events.py @@ -0,0 +1,67 @@ +"""Unit tests for the docker plugin's lifecycle-event derivation. + +_derive_events is a pure function (no DB), and ingest.py imports its models +lazily inside functions, so importing it here doesn't register ORM tables — +safe for the no-DB unit lane. +""" +from plugins.docker.ingest import _derive_events + + +def _c(name, status, **kw): + return {"name": name, "status": status, **kw} + + +def test_start_on_running_after_stopped(): + old = {"web": {"status": "exited", "health": None, "oom_killed": False, "exit_code": 0}} + events = _derive_events(old, [_c("web", "running", image="nginx")]) + assert events == [("web", "start", "nginx")] + + +def test_start_for_new_container_when_state_exists(): + # old_state is non-empty (host already seen), a fresh container appears running. + old = {"other": {"status": "running", "health": None, "oom_killed": False, "exit_code": None}} + events = _derive_events(old, [ + _c("other", "running"), + _c("new", "running", image="redis"), + ]) + assert ("new", "start", "redis") in events + # the unchanged 'other' produces nothing + assert all(e[0] != "other" for e in events) + + +def test_stop_on_clean_exit(): + old = {"job": {"status": "running", "health": None, "oom_killed": False, "exit_code": None}} + events = _derive_events(old, [_c("job", "exited", exit_code=0)]) + assert events == [("job", "stop", None)] + + +def test_die_on_nonzero_exit(): + old = {"job": {"status": "running", "health": None, "oom_killed": False, "exit_code": None}} + events = _derive_events(old, [_c("job", "exited", exit_code=137)]) + assert events == [("job", "die", "exit 137")] + + +def test_oom_on_flip_to_true(): + old = {"hog": {"status": "running", "health": None, "oom_killed": False, "exit_code": None}} + # OOM-killed → also no longer running with a non-zero exit, so die + oom both fire. + events = _derive_events(old, [_c("hog", "exited", exit_code=137, oom_killed=True)]) + assert ("hog", "oom", None) in events + assert ("hog", "die", "exit 137") in events + + +def test_health_change_emitted(): + old = {"svc": {"status": "running", "health": "healthy", "oom_killed": False, "exit_code": None}} + events = _derive_events(old, [_c("svc", "running", health="unhealthy")]) + assert events == [("svc", "health_change", "healthy→unhealthy")] + + +def test_removed_running_container_stops(): + old = {"gone": {"status": "running", "health": None, "oom_killed": False, "exit_code": None}} + events = _derive_events(old, []) # vanished from the listing entirely + assert events == [("gone", "stop", "removed")] + + +def test_no_event_when_unchanged(): + old = {"web": {"status": "running", "health": "healthy", "oom_killed": False, "exit_code": None}} + events = _derive_events(old, [_c("web", "running", health="healthy")]) + assert events == []