[reckless-sherixx]

Summary

Container logs exported over OTLP were lost whenever the agent, the logs endpoint or its backend restarted:

• readers started at the end of each file and saved no position;
• a refused connection dropped the batch with no retry;
• the 2048-record queue dropped records when full;
• nothing was flushed on exit.

With this change, delivery is at least once across those restarts. On a local test (100 lines/s, files rotating every 40 s), stopping the agent (SIGTERM, then three SIGKILLs), restarting the collector and restarting the backend for 60 s lost 0 of 40,377 lines (the build before this change lost 3,772 of 26,760). A 330 s backend outage lost 0 of 38,789 (before: 5,436 lost). Duplicates, counted on raw rows as count() - uniqExact(line id) on a plain MergeTree table: 1,432, all from the two SIGKILLs that landed 4.9 s and 9.0 s after a positions save (457 and 975 lines). Every other scenario, including a SIGKILL 1.0 s after a save, gave 0.

• Positions (LOGS_POSITIONS_FILE). Each file's position is saved every 10 s and on SIGTERM, and only up to lines whose records the endpoint accepted, using the offset of each message's last line from logparser. After a restart a reader resumes there, and a truncated file starts at 0.
• First run. On the agent's first run on a node, files present at startup are read from their end, as before. Files found after the initial container scan (a new container, or a crash-loop restart's new file) are read from 0. Once a positions file exists, any unknown file is read from 0.
• Stale positions (LOGS_POSITIONS_MAX_AGE, default 24h, 0 disables). A positions file saved longer ago than this is ignored and the first-run rule applies, so a long downtime does not send a burst of old lines.
• Rotation. The old file is read to its end before the new one, and acks are tracked across any number of rotations. A file rotated while the agent was down is found again by its inode and finished first; its saved position is kept while the file is there under its rotated name, so a reader that starts late (containers are found as the eBPF programs load) still finds it. A reader paused by backpressure while its file rotates more than once reads the rotated files the kubelet has not gzipped yet before the new one.
• Retries. Only known-transient errors are retried until accepted: connection and timeout errors, 5xx, 408 and 429. Anything else (another 4xx, a response that is not OTLP) drops the batch.
• Backpressure. Readers stop reading while 2048 records or 16 MiB are waiting, so emitting never blocks (it runs under the parser lock that /metrics also takes). journald, which can't be paused, is dropped past 4096 waiting records, with a warning at most once a minute.
• SIGTERM. Stop reading, drop new journald records, deliver what was read (up to 15 s), then save.
• Metric. node_agent_log_records_dropped_total{reason} counts dropped records: backpressure, shutdown, rejected.

Limits: the kubelet gzips every rotated file but the newest, and gzipped files are not read. A paused reader therefore absorbs at most two rotations of a container's log file (about 20 MiB at the default 10 MiB); files gzipped before it reaches them are skipped with a warning naming their count. After a restart, files rotated between the saved one and the current one are not read either, also with a warning.

Without LOGS_ENDPOINT, nothing changes except that rotation now reads the old file to its end.

Depends on nudgebee/logparser#36, now merged; go.mod pins its merge commit. The chart half (positions hostPath, a logs exporter that waits for ClickHouse) is nudgebee/k8s-agent#678.

Test plan

• go test -race ./logs/ and go vet ./logs/ ./flags/ ./containers/ . on Linux. The tests cover positions, the first-run rule, stale positions, rotation (live, several before an ack, and while down), delivery, the shutdown drain with journald active, error classification, retry, the read gate and the dropped metric.
• On a two-node kind cluster with the chart from feat(chart): hold container logs back instead of dropping them when ClickHouse is down k8s-agent#678 and numbered generators (20 and 100 lines/s, one rotating every 100 s), counted in ClickHouse: baseline, SIGKILL, the agent down across a rotation (reader starting ~30 s after the agent), and ClickHouse down while gen-fast rotated twice all lost 0 lines with 0 duplicates; the build before the two rotation fixes lost 4,808 and 20,663 lines in the last two, silently. With three rotations during the pause, only the gzipped file's 10,000 lines were lost, and the agent logged it. TestSaveKeepsRotatedFileForItsReader, TestPausedReaderReadsFilesRotatedMeanwhile and TestPausedReaderWarnsAboutGzippedFiles fail without these fixes.
• End to end in local docker: ClickHouse 24.12 and collector-contrib 0.157 with the chart's rendered config. The real log path read a CRI-format generator that numbers every message, and each line was counted in ClickHouse on raw rows; OPTIMIZE TABLE ... FINAL left the duplicate count unchanged.

Scenario	Lost	Duplicates
SIGTERM	0	0
SIGKILL 0.97 s after a save	0	0
SIGKILL 4.88 s after a save	0	457
SIGKILL 9.01 s after a save	0	975
Collector restart (15 s down)	0	0
ClickHouse restart (60 s down)	0	0
ClickHouse down 330 s	0	0

[gemini-code-assist]

Code Review

This pull request introduces a robust log delivery tracking and checkpointing mechanism to ensure at-least-once delivery of container logs across agent restarts and collector outages. It adds a logs positions file to save how far each container log file has been delivered, tracks pending records, retries exports on transient failures, and handles log rotation gracefully by reading old files to their end. The review feedback highlights two key improvements in logs/positions.go: first, avoiding holding a mutex lock during blocking disk I/O operations inside save() to prevent performance degradation and blocking tail readers; second, delaying setting p.existed = true until after the positions file is successfully unmarshalled to prevent log duplication storms in case of file corruption.

[mayankpande88]

Solid design. I checked the assumptions it rests on against the SDK and they hold: the batch processor exports in emit order, WithExportTimeout(0) means no timeout (not an expired context), the queue is 4x the read gate so emitting can't block, and nothing else emits through the provider or calls ForceFlush. A few things before merge:

Blocker

• go.mod pins nudgebee/logparser#36's branch commit; re-pin to its merge commit once it lands.

Medium

• Shutdown drain never converges while journald logs (delivery.drain, logs.Shutdown). stopReading pauses only file readers; journald records (src == nil) keep being emitted, and since the batcher exports about once a second, pending is rarely 0. On any node with active systemd-unit logging, shutdown waits the full 15 s and then logs a false "N log records were not delivered". Suggest dropping src == nil records once stopping is set, or having drain wait only for records emitted before the grace period ends.
• First run on a node still starts new containers at the end (positions.startOffset). existed is fixed when the file is loaded, so for the agent's whole first lifetime on a node every container started after the agent (including each crash-loop restart's new log file) is read from its end and its startup lines are lost. After one agent restart the same case reads from 0. Consistent would be: files present at startup start at the end, files found later start at 0 (e.g. once the initial container scan is done).

Low

• Two rotations before the first ack (Source.newFile). prev is only recorded when nil, so if the reader moves A → B → C before any of B's records are acknowledged (likely during a collector outage), B's acks match neither generation and are discarded. A crash in that window resumes at A, then reads C from 0, so B is lost. It falls under the documented "less than one file per outage" limit, but the loss is silent.
• permanent() defaults to retry-forever. It classifies by matching the error text, and any unrecognised error retries indefinitely, e.g. a 200 whose body isn't valid protobuf (endpoint pointed at the wrong server): the batch is re-sent, and duplicated, forever while every reader stays paused. Retrying only known-transient errors (url.Error, 5xx, 408, 429) and dropping the rest would be safer.
• Stale positions file: after a long agent downtime, or the logs endpoint being disabled and re-enabled, every container's current file is re-read from 0. Backpressure bounds memory, but the backend gets a burst of old-timestamped records. Consider ignoring a positions file older than some max age.
• journald drops past maxPendingUnpaused increment dropped but are neither logged nor exported; a metric would make them visible.

Question: the test reports 0 duplicates with SIGKILL. With positions saved every 10 s at 100 lines/s, a SIGKILL should re-send up to ~1,000 lines. How were duplicates counted? If the backend deduplicated them, the number should say so.

[reckless-sherixx]

@mayankpande88 thanks for the review. All items are addressed in c063599 (the single branch commit, amended). go vet ./logs/ ./flags/ ./containers/ . and go test -race -count=1 ./logs/ pass on Linux (25 tests, no races). Each behaviour fix has a test that failed before the fix.

1. Blocker, logparser pin. Re-pinned to the merge commit of logparser#36: v0.0.0-20261022152836-24156609a9ba (24156609a9ba is the squash merge and the current head of logparser main). go.sum changes only the two logparser lines; go mod tidy leaves both files unchanged. No replace directive.

2. Medium, shutdown drain with journald. Took your first suggestion: once stopping is set, emit drops records with src == nil and counts them as dropped with reason shutdown. drain now waits only for records already pending plus file records flushed during the grace period. Test: TestDrainEndsWhileUnpausedSourceKeepsEmitting emits a journald record on every export. Before: drain timed out after 1 s with "1 records still pending". After: it returns in 0.01 s. Shutdown drops are counted, not logged, so a normal SIGTERM does not warn.

3. Medium, first run. Implemented the rule you described: files present at startup start at the end, files found later start at 0. The signal is the end of the initial container scan. Tracer.Run sends a start event for every running process into the registry's event channel, and handleEvents creates each container and its reader in channel order. After Run returns, the registry sends an internal marker event on the same channel, and handling it calls logs.InitialScanDone(). So the marker is handled only after every container found at startup has its reader. Tests: TestStartOffset (first run, found after the scan: from 0; before the fix it gave the file size) and TestFirstRunReadsFilesFoundAfterTheInitialScanFromTheirStart (before the fix, the startup lines of a new file were missing). Remaining gap: a container that starts in the few seconds before the marker is handled is still read from its end.

4. Low, several rotations before the first ack. Source now keeps a list of files, oldest first, instead of one prev. An ack finds its generation anywhere in the list, moves that file's position and drops older files. The saved position stays in the oldest file with undelivered lines. Test: TestSeveralRotationsBeforeDelivery (A, B, C, then an ack for B). Before, B's ack was discarded. On restart, B is still skipped: the reader finishes A and then reads C. That loss is no longer silent. The positions file records how many files were skipped, and the agent logs a warning on restart. Tests: TestSavedPositionCountsSkippedFiles and TestRestartWarnsAboutSkippedFiles. Fully recovering B would need every unfinished file saved and a queue of files in the reader; I left that out of this PR.

5. Low, retry only transient errors. Replaced permanent() with transient(). Retried: *url.Error (refused, reset, DNS, timeouts), 5xx, 408, 429, and the OTLP client's own 429/503 retry error. Everything else drops the batch with reason rejected and an error log, including other 4xx and a 200 whose body is not OTLP. Tests: TestTransient (replaces TestPermanent) and TestExporterDropsBatchOnUnreadableResponse. Against a server answering 200 with HTML, the old code re-sent the batch and kept it pending; it is now sent once and dropped. One trade-off: TLS certificate errors are also *url.Error, so they keep retrying.

6. Low, stale positions file. New flag LOGS_POSITIONS_MAX_AGE / --logs-positions-max-age, default 24h, 0 disables. Each save now writes saved_at. A file without it (written by an earlier build) is aged by its mtime, and a saved_at in the future is never stale. A stale file is ignored, with an info log, and the first-run rule from item 3 applies. Why 24h: positions are saved every 10 s while the agent runs, even during a collector outage. So the file only ages while the agent itself is down, and upgrades, reboots and overnight maintenance still resume. Tests: TestStalePositionsFileIsIgnored and TestStalePositionsFileWithoutSaveTime.

7. Low, journald drops. New counter node_agent_log_records_dropped_total{reason} with backpressure, shutdown and rejected. It is registered next to the other agent self-metrics, so it carries machine_id and system_uuid. A backpressure drop also logs a warning at most once a minute. Tests: TestUnpausedSourceIsDroppedWhenBackedUp, TestDrainEndsWhileUnpausedSourceKeepsEmitting and TestExporterRetriesUnreachableCollector each check their reason's count.

8. Question, duplicates. The 0 was a timing result, not deduplication.

   • How it was counted: raw otel_logs rows, no FINAL, DISTINCT or GROUP BY. The generator numbers every message, and duplicates were rows minus distinct numbers. The table is the exporter's default plain MergeTree with non_replicated_deduplication_window = 0, so nothing deduplicates on insert.
   • Why it showed 0: the harness killed the agent 30 s after starting it. That is exactly three save intervals, so the SIGKILL always landed 0.5 to 2.5 s after a positions save. That is before the next batch (about every 5 s) was acknowledged, so nothing had to be re-sent.
   • Re-run: I aimed each SIGKILL at a measured delay D after a save. D is the container's exit time minus the positions.json mtime.

   SIGKILL	Duplicates	Bound (D x 100 + one 512 batch)	Lost
   D = 0.97 s	0	609	0
   D = 4.88 s	457	1,000	0
   D = 9.01 s	975	1,413	0

   Each re-sent range starts at the line right after the offset saved in positions.json and ends before the kill. SIGTERM, a collector restart and a ClickHouse restart gave 0 duplicates and 0 lost. The run total was 40,377 lines, 0 lost, 1,432 duplicate rows, and OPTIMIZE TABLE otel_logs FINAL left 1,432 unchanged. A separate 330 s ClickHouse outage lost 0 of 38,789 with 0 duplicates. The PR body now has these numbers.

   Counting query (STOPPED is the generator's last number):

   WITH toUInt64(extract(Body, 'seq=([0-9]+)')) AS s
   SELECT count() AS rows, uniqExact(s) AS distinct_seq, rows - distinct_seq AS duplicates,
          STOPPED - uniqExactIf(s, s BETWEEN 1 AND STOPPED) AS missing
   FROM otel_logs WHERE Body LIKE 'seq=%'