From 77f9271c9dbf79091cb214e35b293de7a61d9b4e Mon Sep 17 00:00:00 2001 From: Paulo Dias Date: Thu, 23 Oct 2025 21:39:26 +0100 Subject: [PATCH 1/2] [pkg/stanza] Fix collector stops reading new file after copytruncate rotation Signed-off-by: Paulo Dias --- .chloggen/fix_43693.yaml | 27 +++++++++++++++++++++++++++ pkg/stanza/fileconsumer/file.go | 12 +++++++++++- 2 files changed, 38 insertions(+), 1 deletion(-) create mode 100644 .chloggen/fix_43693.yaml diff --git a/.chloggen/fix_43693.yaml b/.chloggen/fix_43693.yaml new file mode 100644 index 0000000000000..e0e75600c9cef --- /dev/null +++ b/.chloggen/fix_43693.yaml @@ -0,0 +1,27 @@ +# Use this changelog template to create an entry for release notes. + +# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix' +change_type: "bug_fix" + +# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog) +component: pkg/stanza + +# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`). +note: "Improve logic to detect copytruncate rotations and reset reader offsets to ensure continuous log collection." + +# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists. +issues: [43693] + +# (Optional) One or more lines of additional information to render under the primary note. +# These lines will be padded with 2 spaces and then inserted directly into the document. +# Use pipe (|) for multiline entries. +subtext: + +# If your change doesn't affect end users or the exported elements of any package, +# you should instead start your pull request title with [chore] or use the "Skip Changelog" label. +# Optional: The change log or logs in which this entry should be included. +# e.g. '[user]' or '[user, api]' +# Include 'user' if the change is relevant to end users. +# Include 'api' if there is a change to a library API. +# Default: '[user]' +change_logs: [user] diff --git a/pkg/stanza/fileconsumer/file.go b/pkg/stanza/fileconsumer/file.go index 2da1c723766bb..62bbb530e0a3e 100644 --- a/pkg/stanza/fileconsumer/file.go +++ b/pkg/stanza/fileconsumer/file.go @@ -289,7 +289,17 @@ func (m *Manager) newReader(ctx context.Context, file *os.File, fp *fingerprint. zap.String("rotated_path", file.Name())) } } - return m.readerFactory.NewReaderFromMetadata(file, oldReader.Close()) + // Close old reader and adjust metadata if file was copy-truncated. + md := oldReader.Close() + if info, err := file.Stat(); err == nil && md.Offset > info.Size() { + m.set.Logger.Debug("File has been rotated(truncated). Resetting offset to 0", + zap.String("path", file.Name()), + zap.Int64("stored_offset", md.Offset), + zap.Int64("current_file_size", info.Size()), + ) + md.Offset = 0 + } + return m.readerFactory.NewReaderFromMetadata(file, md) } // Check for closed files for match From 94f819d05d2187455de6a2cbd04c21908b4bf1ed Mon Sep 17 00:00:00 2001 From: Paulo Dias Date: Thu, 23 Oct 2025 22:13:55 +0100 Subject: [PATCH 2/2] feat: add tests Signed-off-by: Paulo Dias --- pkg/stanza/fileconsumer/file_test.go | 57 ++++++++++++++++++++++++++++ 1 file changed, 57 insertions(+) diff --git a/pkg/stanza/fileconsumer/file_test.go b/pkg/stanza/fileconsumer/file_test.go index f2ef4b312dd9c..184b3c3edc9ee 100644 --- a/pkg/stanza/fileconsumer/file_test.go +++ b/pkg/stanza/fileconsumer/file_test.go @@ -4,6 +4,7 @@ package fileconsumer import ( + "bytes" "compress/gzip" "context" "fmt" @@ -1664,3 +1665,59 @@ func TestArchive(t *testing.T) { sink.ExpectCalls(t, log3, log4) } + +func TestCopyTruncateResetsOffsetOnRestart_IdenticalFirstKB(t *testing.T) { + t.Parallel() + + line := string(bytes.Repeat([]byte("a"), 1024)) // identical 1024B lines + + tempDir := t.TempDir() + cfg := NewConfig().includeDir(tempDir) + cfg.StartAt = "beginning" + cfg.FingerprintSize = 1000 // identical prefix across rotations + + // Manager #1 (manual polling, no background goroutine) + op1, sink1 := testManager(t, cfg) + op1.persister = testutil.NewUnscopedMockPersister() + + // Create file and write 20 lines + log := filetest.OpenTemp(t, tempDir) + for range 20 { + filetest.WriteString(t, log, line+"\n") + } + + // First poll: read the existing 20 lines + op1.poll(t.Context()) + for range 20 { + sink1.ExpectToken(t, []byte(line)) + } + + // Simulate copytruncate + rotated := log.Name() + ".1" + origData, err := os.ReadFile(log.Name()) + require.NoError(t, err) + require.NoError(t, os.WriteFile(rotated, origData, 0o600)) + require.NoError(t, log.Truncate(0)) + _, err = log.Seek(0, 0) + require.NoError(t, err) + for range 10 { + filetest.WriteString(t, log, line+"\n") + } + + // Persist metadata as if we were running; then stop op1. + // (poll() already saves checkpoints when persister is set.) + // Ensure any internal rotations are finalized. + op1.poll(t.Context()) + require.NoError(t, op1.Stop()) + + // Manager #2 (manual polling) resumes from persisted metadata + op2, sink2 := testManager(t, cfg) + op2.persister = op1.persister + + // On poll, offset > current size is detected and reset to 0. + op2.poll(t.Context()) + for range 10 { + sink2.ExpectToken(t, []byte(line)) + } + sink2.ExpectNoCalls(t) +}