diff --git a/plugins/docker/migrations/versions/docker_004_events_swarm.py b/plugins/docker/migrations/versions/docker_004_events_swarm.py new file mode 100644 index 0000000..2f236ca --- /dev/null +++ b/plugins/docker/migrations/versions/docker_004_events_swarm.py @@ -0,0 +1,77 @@ +"""Docker lifecycle events + swarm topology tables + +Adds the storage for milestone-77 monitoring depth that doesn't fit on the +per-container row: + + * docker_events — lifecycle (start/stop/die/oom/health_change) derived by + diffing consecutive host snapshots; host-scoped, indexed for both timeline + lookups (host_id, container_name, at) and retention pruning (at). + * docker_swarm_services / docker_swarm_nodes — manager-reported Swarm + topology (desired-vs-running replicas, node role/availability/status). + +All new tables — purely additive, so no DROP+recreate. `event`/`mode`/`role` +etc. are plain strings (no CHECK whitelist), matching how docker_containers +models `status`; keeps the value sets open without a migration per addition +(so rule 36 doesn't apply here). + +Revision ID: docker_004_events_swarm +Revises: docker_003_container_enrichment +Create Date: 2026-06-19 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "docker_004_events_swarm" +down_revision: Union[str, None] = "docker_003_container_enrichment" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "docker_events", + 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("event", sa.String(16), nullable=False), + sa.Column("detail", sa.String(255), nullable=True), + sa.Column("at", sa.DateTime(timezone=True), nullable=False), + ) + op.create_index("ix_docker_events_host_container_time", "docker_events", + ["host_id", "container_name", "at"]) + op.create_index("ix_docker_events_at", "docker_events", ["at"]) + + op.create_table( + "docker_swarm_services", + sa.Column("host_id", sa.String(36), + sa.ForeignKey("hosts.id", ondelete="CASCADE"), primary_key=True), + sa.Column("service_name", sa.String(255), primary_key=True), + sa.Column("mode", sa.String(16), nullable=False, server_default="replicated"), + sa.Column("desired", sa.Integer, nullable=False, server_default="0"), + sa.Column("running", sa.Integer, nullable=False, server_default="0"), + sa.Column("image", sa.String(512), nullable=True), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + ) + + op.create_table( + "docker_swarm_nodes", + sa.Column("host_id", sa.String(36), + sa.ForeignKey("hosts.id", ondelete="CASCADE"), primary_key=True), + sa.Column("node_id", sa.String(64), primary_key=True), + sa.Column("hostname", sa.String(255), nullable=False, server_default=""), + sa.Column("role", sa.String(16), nullable=False, server_default="worker"), + sa.Column("availability", sa.String(16), nullable=False, server_default="active"), + sa.Column("status", sa.String(16), nullable=False, server_default="unknown"), + sa.Column("leader", sa.Boolean, nullable=False, server_default=sa.false()), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + ) + + +def downgrade() -> None: + op.drop_table("docker_swarm_nodes") + op.drop_table("docker_swarm_services") + op.drop_index("ix_docker_events_at", table_name="docker_events") + op.drop_index("ix_docker_events_host_container_time", table_name="docker_events") + op.drop_table("docker_events") diff --git a/plugins/docker/models.py b/plugins/docker/models.py index 8fd2ae4..cf83bed 100644 --- a/plugins/docker/models.py +++ b/plugins/docker/models.py @@ -85,3 +85,89 @@ class DockerMetric(Base): Index("ix_docker_metrics_host_container_time", "host_id", "container_name", "scraped_at"), ) + + +class DockerEvent(Base): + """Lifecycle events derived by diffing consecutive host snapshots. + + The agent only reports current state, so Steward synthesises lifecycle by + comparing each new snapshot against the stored DockerContainer state for the + host: a container that appears → `start`; one that vanishes → `stop`; a + transition to a stopped status with a non-zero exit → `die`; an OOM-kill + flip → `oom`; a health status change → `health_change`. Scoped to the + reporting host (names are only unique within a host). `event` is a plain + string (no CHECK whitelist), matching how `status` is modelled on + DockerContainer — keeps the value set open without a migration per addition. + """ + __tablename__ = "docker_events" + + 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 + ) + container_name: Mapped[str] = mapped_column(String(255), nullable=False) + event: Mapped[str] = mapped_column(String(16), nullable=False) + # start | stop | die | oom | health_change + detail: Mapped[str | None] = mapped_column(String(255), nullable=True) + # e.g. "exit 137", "healthy→unhealthy", image on start + at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, + default=lambda: datetime.now(timezone.utc), + ) + + # Timeline lookups filter on (host_id, container_name) and sort by time; + # retention prunes by `at` alone — index both access paths. + __table_args__ = ( + Index("ix_docker_events_host_container_time", + "host_id", "container_name", "at"), + Index("ix_docker_events_at", "at"), + ) + + +class DockerSwarmService(Base): + """A Swarm service as seen by a manager host: desired vs running replicas. + + Manager-scoped — only a host that is a Swarm manager reports these, so a + single-manager cluster yields one row set keyed by that host. Service names + are unique within a swarm; we still key by (host_id, service_name) so two + managers reporting independently don't collide. + """ + __tablename__ = "docker_swarm_services" + + host_id: Mapped[str] = mapped_column( + String(36), ForeignKey("hosts.id", ondelete="CASCADE"), primary_key=True + ) + service_name: Mapped[str] = mapped_column(String(255), primary_key=True) + mode: Mapped[str] = mapped_column(String(16), nullable=False, default="replicated") + # replicated | global + 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) + 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" + + host_id: Mapped[str] = mapped_column( + String(36), ForeignKey("hosts.id", ondelete="CASCADE"), primary_key=True + ) + node_id: Mapped[str] = mapped_column(String(64), primary_key=True) + hostname: Mapped[str] = mapped_column(String(255), nullable=False, default="") + role: Mapped[str] = mapped_column(String(16), nullable=False, default="worker") + # manager | worker + availability: Mapped[str] = mapped_column(String(16), nullable=False, default="active") + # active | pause | drain + status: Mapped[str] = mapped_column(String(16), nullable=False, default="unknown") + # ready | down | unknown + leader: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, + default=lambda: datetime.now(timezone.utc), + ) diff --git a/tests/integration/test_docker.py b/tests/integration/test_docker.py index 6c5c280..2865d65 100644 --- a/tests/integration/test_docker.py +++ b/tests/integration/test_docker.py @@ -53,6 +53,28 @@ def test_docker_tables_are_host_scoped(app): asyncio.run(_go()) +@_NEEDS_DB +def test_events_and_swarm_tables_exist(app): + """docker_004 created the lifecycle + swarm topology tables with host_id FKs.""" + from sqlalchemy import text + + async def _go(): + async with app.db_sessionmaker() as s: + for tbl in ("docker_events", "docker_swarm_services", "docker_swarm_nodes"): + 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 cols, f"{tbl} missing entirely" + assert "host_id" in cols, f"{tbl} missing host_id" + # docker_events carries the lifecycle columns the ingest derivation writes. + ev_cols = {r[0] for r in (await s.execute(text( + "SELECT column_name FROM information_schema.columns " + "WHERE table_name = 'docker_events'"))).all()} + assert {"event", "container_name", "at", "detail"} <= ev_cols + + 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