[pkg/stanza] Fix issue where files were unnecessarily kept open on wi…

…ndows (open-telemetry#29514)

Resolves open-telemetry#29149

.chloggen/fileconsumer-fix-windows.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 files were unnecessarily kept open on Windows
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [29149]
+
+# (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/file.go

@@ -80,6 +80,7 @@ func (m *Manager) closePreviousFiles() {
  for _, r := range m.previousPollFiles {
  m.knownFiles = append(m.knownFiles, r.Close())
  }
+ m.previousPollFiles = nil
 }
 
 // Stop will stop the file monitoring process
@@ -159,11 +160,7 @@ func (m *Manager) consume(ctx context.Context, paths []string) {
  m.Debug("Consuming files", zap.Strings("paths", paths))
  readers := m.makeReaders(paths)
 
- // take care of files which disappeared from the pattern since the last poll cycle
- // this can mean either files which were removed, or rotated into a name not matching the pattern
- // we do this before reading existing files to ensure we emit older log lines before newer ones
- m.readLostFiles(ctx, readers)
- m.closePreviousFiles()
+ m.preConsume(ctx, readers)
 
  // read new readers to end
  var wg sync.WaitGroup
@@ -176,7 +173,7 @@ func (m *Manager) consume(ctx context.Context, paths []string) {
  }
  wg.Wait()
 
- m.previousPollFiles = readers
+ m.postConsume(readers)
 }
 
 func (m *Manager) makeFingerprint(path string) (*fingerprint.Fingerprint, *os.File) {

pkg/stanza/fileconsumer/file_other.go

@@ -13,8 +13,10 @@ import (
  "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/reader"
 )
 
-func (m *Manager) readLostFiles(ctx context.Context, newReaders []*reader.Reader) {
- // Detect files that have been rotated out of matching pattern
+// Take care of files which disappeared from the pattern since the last poll cycle
+// this can mean either files which were removed, or rotated into a name not matching the pattern
+// we do this before reading existing files to ensure we emit older log lines before newer ones
+func (m *Manager) preConsume(ctx context.Context, newReaders []*reader.Reader) {
  lostReaders := make([]*reader.Reader, 0, len(m.previousPollFiles))
 OUTER:
  for _, oldReader := range m.previousPollFiles {
@@ -48,3 +50,10 @@ OUTER:
  }
  lostWG.Wait()
 }
+
+// On non-windows platforms, we keep files open between poll cycles so that we can detect
+// and read "lost" files, which have been moved out of the matching pattern.
+func (m *Manager) postConsume(readers []*reader.Reader) {
+ m.closePreviousFiles()
+ m.previousPollFiles = readers
+}

pkg/stanza/fileconsumer/file_test.go

@@ -1071,6 +1071,7 @@ func TestFileBatchingRespectsStartAtEnd(t *testing.T) {
 
  operator, emitChan := buildTestManager(t, cfg)
  operator.persister = testutil.NewUnscopedMockPersister()
+ operator.movingAverageMatches = 10
 
  temps := make([]*os.File, 0, initFiles+moreFiles)
  for i := 0; i < initFiles; i++ {
@@ -1662,7 +1663,10 @@ func TestStalePartialFingerprintDiscarded(t *testing.T) {
  waitForToken(t, emitCalls, []byte(content))
  expectNoTokens(t, emitCalls)
  operator.wg.Wait()
- require.Len(t, operator.previousPollFiles, 1)
+ if runtime.GOOS != "windows" {
+ // On windows, we never keep files in previousPollFiles, so we don't expect to see them here
+ require.Len(t, operator.previousPollFiles, 1)
+ }
 
  // keep append data to file1 and file2
  newContent := "bbbbbbbbbbbb"
@@ -1675,3 +1679,22 @@ func TestStalePartialFingerprintDiscarded(t *testing.T) {
  waitForTokens(t, emitCalls, []byte(content), []byte(newContent1), []byte(newContent))
  operator.wg.Wait()
 }
+
+func TestWindowsFilesClosedImmediately(t *testing.T) {
+ t.Parallel()
+
+ tempDir := t.TempDir()
+ cfg := NewConfig().includeDir(tempDir)
+ cfg.StartAt = "beginning"
+ operator, emitCalls := buildTestManager(t, cfg)
+
+ temp := openTemp(t, tempDir)
+ writeString(t, temp, "testlog\n")
+ require.NoError(t, temp.Close())
+
+ operator.poll(context.Background())
+ waitForToken(t, emitCalls, []byte("testlog"))
+
+ // On Windows, poll should close the file after reading it. We can test this by trying to move it.
+ require.NoError(t, os.Rename(temp.Name(), temp.Name()+"_renamed"))
+}

pkg/stanza/fileconsumer/file_windows.go

@@ -12,6 +12,12 @@ import (
  "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/reader"
 )
 
-func (m *Manager) readLostFiles(ctx context.Context, newReaders []*reader.Reader) {
+func (m *Manager) preConsume(ctx context.Context, newReaders []*reader.Reader) {
  return
 }
+
+// On windows, we close files immediately after reading becauase they cannot be moved while open.
+func (m *Manager) postConsume(readers []*reader.Reader) {
+ m.previousPollFiles = readers
+ m.closePreviousFiles()
+}