[pkg/stanza] Skip rotated file handles if truncated (#27064)

Fixes
#27037

.chloggen/pkg-stanza-lost-truncate.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. filelogreceiver)
+component: filelogreceiver
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix issue where truncated file could be read incorrectly.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [27037]
+
+# (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: []

pkg/stanza/fileconsumer/config.go

@@ -180,7 +180,7 @@ func (c Config) buildManager(logger *zap.SugaredLogger, emit emit.Callback, fact
  headerConfig: hCfg,
  },
  fileMatcher: fileMatcher,
- roller: newRoller(),
+ roller: newRoller(int(c.FingerprintSize)),
  pollInterval: c.PollInterval,
  maxBatchFiles: c.MaxConcurrentFiles / 2,
  maxBatches: c.MaxBatches,

pkg/stanza/fileconsumer/reader.go

@@ -42,6 +42,7 @@ type reader struct {
  *readerConfig
  *readerMetadata
  file *os.File
+ fileName string
  lineSplitFunc bufio.SplitFunc
  splitFunc bufio.SplitFunc
  decoder *decode.Decoder
@@ -129,10 +130,9 @@ func (r *reader) Delete() {
  if r.file == nil {
  return
  }
- f := r.file
  r.Close()
- if err := os.Remove(f.Name()); err != nil {
- r.Errorf("could not delete %s", f.Name())
+ if err := os.Remove(r.fileName); err != nil {
+ r.Errorf("could not delete %s", r.fileName)
  }
 }
 

pkg/stanza/fileconsumer/reader_factory.go

@@ -58,6 +58,7 @@ func (f *readerFactory) build(file *os.File, m *readerMetadata, lineSplitFunc bu
  readerConfig: f.readerConfig,
  readerMetadata: m,
  file: file,
+ fileName: file.Name(),
  SugaredLogger: f.SugaredLogger.With("path", file.Name()),
  decoder: decode.New(f.encoding),
  lineSplitFunc: lineSplitFunc,
@@ -82,12 +83,12 @@ func (f *readerFactory) build(file *os.File, m *readerMetadata, lineSplitFunc bu
  }
 
  // Resolve file name and path attributes
- resolved := file.Name()
+ resolved := r.fileName
 
  // Dirty solution, waiting for this permanent fix https://github.com/golang/go/issues/39786
  // EvalSymlinks on windows is partially working depending on the way you use Symlinks and Junctions
  if runtime.GOOS != "windows" {
- resolved, err = filepath.EvalSymlinks(file.Name())
+ resolved, err = filepath.EvalSymlinks(r.fileName)
  if err != nil {
  f.Errorf("resolve symlinks: %w", err)
  }
@@ -98,12 +99,12 @@ func (f *readerFactory) build(file *os.File, m *readerMetadata, lineSplitFunc bu
  }
 
  if f.readerConfig.includeFileName {
- r.FileAttributes[logFileName] = filepath.Base(file.Name())
+ r.FileAttributes[logFileName] = filepath.Base(r.fileName)
  } else if r.FileAttributes[logFileName] != nil {
  delete(r.FileAttributes, logFileName)
  }
  if f.readerConfig.includeFilePath {
- r.FileAttributes[logFilePath] = file.Name()
+ r.FileAttributes[logFilePath] = r.fileName
  } else if r.FileAttributes[logFilePath] != nil {
  delete(r.FileAttributes, logFilePath)
  }

pkg/stanza/fileconsumer/roller_other.go

@@ -9,14 +9,20 @@ package fileconsumer // import "github.com/open-telemetry/opentelemetry-collecto
 import (
  "context"
  "sync"
+
+ "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/fingerprint"
 )
 
 type detectLostFiles struct {
- oldReaders []*reader
+ oldReaders []*reader
+ fingerprintSize int // used when we check for truncation
 }
 
-func newRoller() roller {
- return &detectLostFiles{[]*reader{}}
+func newRoller(fingerprintSize int) roller {
+ return &detectLostFiles{
+ oldReaders: []*reader{},
+ fingerprintSize: fingerprintSize,
+ }
 }
 
 func (r *detectLostFiles) readLostFiles(ctx context.Context, newReaders []*reader) {
@@ -28,6 +34,21 @@ OUTER:
  if newReader.Fingerprint.StartsWith(oldReader.Fingerprint) {
  continue OUTER
  }
+ if oldReader.fileName == newReader.fileName {
+ // At this point, we know that the file has been rotated. However, we do not know
+ // if it was moved or truncated. If truncated, then both handles point to the same
+ // file, in which case we should only read from it using the new reader.
+
+ // We can detect truncation by recreating a fingerprint from the old handle.
+ // If it matches the old fingerprint, then we know that the file was moved,
+ // so we can consider the file lost and continue reading from the old handle.
+ // If there's an error reading a new fingerprint from the old handle, let's assume we can't
+ // read the rest of it anyways.
+ refreshedFingerprint, err := fingerprint.New(oldReader.file, r.fingerprintSize)
+ if err == nil && !refreshedFingerprint.StartsWith(oldReader.Fingerprint) {
+ continue OUTER
+ }
+ }
  }
  lostReaders = append(lostReaders, oldReader)
  }

pkg/stanza/fileconsumer/roller_windows.go

@@ -10,7 +10,7 @@ import "context"
 
 type closeImmediately struct{}
 
-func newRoller() roller {
+func newRoller(_ int) roller {
  return &closeImmediately{}
 }
 

pkg/stanza/fileconsumer/rotation_test.go

@@ -449,12 +449,10 @@ func TestTrackRotatedFilesLogOrder(t *testing.T) {
  originalFile.Close()
 
  newDir := fmt.Sprintf("%s%s", tempDir[:len(tempDir)-1], "_new/")
- err := os.Mkdir(newDir, 0777)
- require.NoError(t, err)
+ require.NoError(t, os.Mkdir(newDir, 0777))
  movedFileName := fmt.Sprintf("%s%s", newDir, "newfile.log")
 
- err = os.Rename(orginalName, movedFileName)
- require.NoError(t, err)
+ require.NoError(t, os.Rename(orginalName, movedFileName))
 
  newFile, err := os.OpenFile(orginalName, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
  require.NoError(t, err)
@@ -463,6 +461,84 @@ func TestTrackRotatedFilesLogOrder(t *testing.T) {
  waitForTokens(t, emitCalls, [][]byte{[]byte("testlog2"), []byte("testlog3")})
 }
 
+// When a file it rotated out of pattern via move/create, we should
+// detect that our old handle is still valid attempt to read from it.
+func TestRotatedOutOfPatternMoveCreate(t *testing.T) {
+ if runtime.GOOS == windowsOS {
+ t.Skip("Moving files while open is unsupported on Windows")
+ }
+ t.Parallel()
+
+ tempDir := t.TempDir()
+ cfg := NewConfig()
+ cfg.Include = append(cfg.Include, fmt.Sprintf("%s/*.log1", tempDir))
+ cfg.StartAt = "beginning"
+ operator, emitCalls := buildTestManager(t, cfg)
+ operator.persister = testutil.NewMockPersister("test")
+
+ originalFile := openTempWithPattern(t, tempDir, "*.log1")
+ originalFileName := originalFile.Name()
+
+ writeString(t, originalFile, "testlog1\n")
+ operator.poll(context.Background())
+ waitForToken(t, emitCalls, []byte("testlog1"))
+
+ // write more log, before next poll() begins
+ writeString(t, originalFile, "testlog2\n")
+
+ // move the file so it no longer matches
+ require.NoError(t, originalFile.Close())
+ require.NoError(t, os.Rename(originalFileName, originalFileName+".old"))
+
+ newFile := openFile(t, originalFileName)
+ _, err := newFile.Write([]byte("testlog4\ntestlog5\n"))
+ require.NoError(t, err)
+
+ // poll again
+ operator.poll(context.Background())
+
+ // expect remaining log from old file as well as all from new file
+ waitForTokens(t, emitCalls, [][]byte{[]byte("testlog2"), []byte("testlog4"), []byte("testlog5")})
+}
+
+// When a file it rotated out of pattern via copy/truncate, we should
+// detect that our old handle is stale and not attempt to read from it.
+func TestRotatedOutOfPatternCopyTruncate(t *testing.T) {
+ t.Parallel()
+
+ tempDir := t.TempDir()
+ cfg := NewConfig()
+ cfg.Include = append(cfg.Include, fmt.Sprintf("%s/*.log1", tempDir))
+ cfg.StartAt = "beginning"
+ operator, emitCalls := buildTestManager(t, cfg)
+ operator.persister = testutil.NewMockPersister("test")
+
+ originalFile := openTempWithPattern(t, tempDir, "*.log1")
+ writeString(t, originalFile, "testlog1\n")
+ operator.poll(context.Background())
+ waitForToken(t, emitCalls, []byte("testlog1"))
+
+ // write more log, before next poll() begins
+ writeString(t, originalFile, "testlog2\n")
+ // copy the file to another file i.e. rotate, out of pattern
+ newFile := openTempWithPattern(t, tempDir, "*.log2")
+ _, err := originalFile.Seek(0, 0)
+ require.NoError(t, err)
+ _, err = io.Copy(newFile, originalFile)
+ require.NoError(t, err)
+
+ _, err = originalFile.Seek(0, 0)
+ require.NoError(t, err)
+ require.NoError(t, originalFile.Truncate(0))
+ _, err = originalFile.Write([]byte("testlog4\ntestlog5\n"))
+ require.NoError(t, err)
+
+ // poll again
+ operator.poll(context.Background())
+
+ waitForTokens(t, emitCalls, [][]byte{[]byte("testlog4"), []byte("testlog5")})
+}
+
 // TruncateThenWrite tests that, after a file has been truncated,
 // any new writes are picked up
 func TestTruncateThenWrite(t *testing.T) {