diff --git a/plugins/docker/ingest.py b/plugins/docker/ingest.py index 9efc563..027ce95 100644 --- a/plugins/docker/ingest.py +++ b/plugins/docker/ingest.py @@ -147,12 +147,36 @@ async def _persist_logs(session, host, log_batches) -> None: if not await get_setting(session, "docker.logs.enabled"): return exclude = set(await get_setting(session, "docker.logs.exclude") or ()) - for row in _log_rows(log_batches, host.id): - if row["container_name"] in exclude: + rows = [r for r in _log_rows(log_batches, host.id) if r["container_name"] not in exclude] + stored = await _stored_lines(session, host.id, rows) + for row in rows: + if (row["container_name"], row["ts"], row["stream"], row["line"]) in stored: continue session.add(DockerLog(**row)) +async def _stored_lines(session, host_id: str, rows: list) -> set: + """The lines of `rows` this host already has, as (container, ts, stream, + line). A restarted agent has lost its cursors and re-sends each + container's last lines (#5809); a push of new lines overlaps nothing, so + the one query, bounded per container to the batch's own span, finds none.""" + from sqlalchemy import and_, or_, select + from .models import DockerLog + + spans: dict[str, list] = {} + for r in rows: + span = spans.setdefault(r["container_name"], [r["ts"], r["ts"]]) + span[0], span[1] = min(span[0], r["ts"]), max(span[1], r["ts"]) + if not spans: + return set() + found = await session.execute( + select(DockerLog.container_name, DockerLog.ts, DockerLog.stream, DockerLog.line) + .where(DockerLog.host_id == host_id, or_(*( + and_(DockerLog.container_name == name, DockerLog.ts.between(lo, hi)) + for name, (lo, hi) in spans.items())))) + return {tuple(r) for r in found} + + async def _persist_swarm(session, host, swarm: dict) -> None: """Upsert this manager's swarm topology; drop rows no longer reported. diff --git a/tests/integration/test_docker.py b/tests/integration/test_docker.py index 80abc77..909efa5 100644 --- a/tests/integration/test_docker.py +++ b/tests/integration/test_docker.py @@ -221,6 +221,55 @@ def test_persist_logs_stores_lines(app): ] +@_NEEDS_DB +def test_persist_logs_skips_lines_a_restarted_agent_sends_again(app): + """#5809: a restarted agent has lost its cursors and re-sends each + container's last lines. Those are stored once; a new line, or another line + at the same instant, is still kept.""" + from types import SimpleNamespace + from sqlalchemy import text + from steward.models.hosts import Host + + persist = _persist_fn(app) + now = datetime.now(timezone.utc) + + def line(container, at, text_, stream="stdout"): + return {"container": container, "stream": stream, + "ts": f"2026-01-01T00:00:0{at}+00:00", "line": text_} + + first = [(now, [line("web", 1, "started"), line("web", 2, "listening"), + line("db", 1, "ready")])] + again = [(now, [line("web", 1, "started"), line("web", 2, "listening"), + line("web", 2, "worker 2 up"), line("web", 3, "GET /"), + line("db", 1, "ready"), line("db", 1, "ready", stream="stderr")])] + + hid = str(uuid.uuid4()) + host = SimpleNamespace(id=hid, name="loghost-again") # as below: no expiry across commits + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + s.add(Host(id=hid, name="loghost-again", address="10.0.0.19")) + await s.flush() + await persist(s, host, [], None, None, first) + async with s.begin(): + await persist(s, host, [], None, None, again) + rows = (await s.execute(text( + "SELECT container_name, stream, line FROM docker_logs " + "WHERE host_id = :h ORDER BY container_name, ts, stream, line"), + {"h": hid})).all() + return [tuple(r) for r in rows] + + assert asyncio.run(_go()) == [ + ("db", "stderr", "ready"), + ("db", "stdout", "ready"), + ("web", "stdout", "started"), + ("web", "stdout", "listening"), + ("web", "stdout", "worker 2 up"), + ("web", "stdout", "GET /"), + ] + + @_NEEDS_DB def test_persist_logs_respects_toggle_and_exclude(app): """Server-side controls (Settings): an excluded container's lines are dropped