From 45565b2c01a7c7bd156b93b467267698e967ebd3 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sun, 19 Jul 2026 18:43:14 -0400 Subject: [PATCH] feat(docker): docker_logs table + docker_009 migration [M79 step 2] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dedicated host-scoped table for pushed container log lines (one row per line), chained after docker_008. host_id FK CASCADE; twin indexes — (host_id, container_name, ts) for the viewer, ts alone for the age-cutoff prune — mirroring docker_events. Integration schema-shape test asserts columns + both indexes. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01CAGR73DUowdVFVvYzLXC5C --- .../versions/docker_009_container_logs.py | 48 +++++++++++++++++++ plugins/docker/models.py | 37 ++++++++++++++ tests/integration/test_docker.py | 22 +++++++++ 3 files changed, 107 insertions(+) create mode 100644 plugins/docker/migrations/versions/docker_009_container_logs.py diff --git a/plugins/docker/migrations/versions/docker_009_container_logs.py b/plugins/docker/migrations/versions/docker_009_container_logs.py new file mode 100644 index 0000000..773128b --- /dev/null +++ b/plugins/docker/migrations/versions/docker_009_container_logs.py @@ -0,0 +1,48 @@ +"""Docker container logs table + +Adds docker_logs — one row per container log line, pushed by the host agent and +folded into its metrics push. Time-series, host-scoped (container names are only +unique within a host). Bounded by a per-container size+age ring in retention, so +a chatty container keeps a shorter window rather than growing without limit. +Additive create_table + twin indexes (viewer lookup by (host, container, ts); +age-cutoff prune by ts). + +Revision ID: docker_009_container_logs +Revises: docker_008_bigint_mem +Create Date: 2026-07-19 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "docker_009_container_logs" +down_revision: Union[str, None] = "docker_008_bigint_mem" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "docker_logs", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("host_id", sa.String(length=36), nullable=False), + sa.Column("container_name", sa.String(length=255), nullable=False), + sa.Column("ts", sa.DateTime(timezone=True), nullable=False), + sa.Column("stream", sa.String(length=8), nullable=False, server_default="stdout"), + sa.Column("line", sa.Text(), nullable=False, server_default=""), + sa.ForeignKeyConstraint(["host_id"], ["hosts.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("ix_docker_logs_host_id", "docker_logs", ["host_id"]) + op.create_index("ix_docker_logs_container_name", "docker_logs", ["container_name"]) + op.create_index("ix_docker_logs_host_container_time", + "docker_logs", ["host_id", "container_name", "ts"]) + op.create_index("ix_docker_logs_ts", "docker_logs", ["ts"]) + + +def downgrade() -> None: + op.drop_index("ix_docker_logs_ts", table_name="docker_logs") + op.drop_index("ix_docker_logs_host_container_time", table_name="docker_logs") + op.drop_index("ix_docker_logs_container_name", table_name="docker_logs") + op.drop_index("ix_docker_logs_host_id", table_name="docker_logs") + op.drop_table("docker_logs") diff --git a/plugins/docker/models.py b/plugins/docker/models.py index 60f625d..7082d20 100644 --- a/plugins/docker/models.py +++ b/plugins/docker/models.py @@ -163,6 +163,43 @@ class DockerEvent(Base): ) +class DockerLog(Base): + """Container log lines pushed by the host agent — one row per line. + + Time-series, scoped to the reporting host (container names are only unique + within a host, same identity as docker_metrics). The agent tails each running + container incrementally and folds new lines into its metrics push; `ts` is the + line's own Docker timestamp. Bounded by a per-container size+age ring + (retention), so a chatty container just keeps a shorter window rather than + growing without limit. + """ + __tablename__ = "docker_logs" + + 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) + ts: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, + default=lambda: datetime.now(timezone.utc), + ) + stream: Mapped[str] = mapped_column(String(8), nullable=False, default="stdout") + # stdout | stderr + line: Mapped[str] = mapped_column(Text, nullable=False, default="") + + # Viewer filters on (host_id, container_name) and sorts by time; the ring + # prune walks the same key newest-first. A second index on `ts` alone serves + # the age-cutoff delete. Twin-index idiom, mirroring docker_events. + __table_args__ = ( + Index("ix_docker_logs_host_container_time", + "host_id", "container_name", "ts"), + Index("ix_docker_logs_ts", "ts"), + ) + + class DockerSwarmService(Base): """A Swarm service as seen by a manager host: desired vs running replicas. diff --git a/tests/integration/test_docker.py b/tests/integration/test_docker.py index 38bc9db..0210cd0 100644 --- a/tests/integration/test_docker.py +++ b/tests/integration/test_docker.py @@ -75,6 +75,28 @@ def test_events_and_swarm_tables_exist(app): asyncio.run(_go()) +@_NEEDS_DB +def test_docker_logs_table_shape(app): + """docker_009 created docker_logs: host-scoped, with the twin indexes the + viewer (host, container, ts) and the age-cutoff prune (ts) rely on.""" + from sqlalchemy import text + + async def _go(): + async with app.db_sessionmaker() as s: + cols = {r[0] for r in (await s.execute(text( + "SELECT column_name FROM information_schema.columns " + "WHERE table_name = 'docker_logs'"))).all()} + assert cols, "docker_logs missing entirely" + assert {"id", "host_id", "container_name", "ts", "stream", "line"} <= cols + idx = {r[0] for r in (await s.execute(text( + "SELECT indexname FROM pg_indexes " + "WHERE tablename = 'docker_logs'"))).all()} + assert "ix_docker_logs_host_container_time" in idx + assert "ix_docker_logs_ts" in idx + + 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