fix(docker): a restarted agent's re-sent log lines are stored once (#5809)
An agent keeps its log cursors in memory, so after a restart it seeds each container from its last 50 lines, which Steward already holds; docker_logs has no unique key, so every update or reboot doubled them. The ingest now reads what it already has for each container over the batch's own span, in one query, and skips those lines. A push of new lines overlaps nothing. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -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.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user