[pkg/stanza/fileconsumer] Add ability to read files asynchronously

.chloggen/add-threadpool-featuregate.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: 'enhancement'
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: fileconsumer
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Added a new feature gate that enables a thread pool mechanism to respect the poll_interval parameter.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [18908]
+
+# (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

@@ -16,6 +16,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/fingerprint"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/header"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/splitter"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/trie"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/matcher"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
@@ -33,6 +34,13 @@ var allowFileDeletion = featuregate.GlobalRegistry().MustRegister(
 	featuregate.WithRegisterReferenceURL("https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/16314"),
 )
 
+var useThreadPool = featuregate.GlobalRegistry().MustRegister(
+	"filelog.useThreadPool",
+	featuregate.StageAlpha,
+	featuregate.WithRegisterDescription("When enabled, log collection switches to a thread pool model, respecting the `poll_interval` config."),
+	// featuregate.WithRegisterReferenceURL("https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/16314"),
+)
+
 var AllowHeaderMetadataParsing = featuregate.GlobalRegistry().MustRegister(
 	"filelog.allowHeaderMetadataParsing",
 	featuregate.StageBeta,
@@ -151,7 +159,7 @@ func (c Config) buildManager(logger *zap.SugaredLogger, emit emit.Callback, fact
 		return nil, err
 	}
 
-	return &Manager{
+	manager := Manager{
 		SugaredLogger: logger.With("component", "fileconsumer"),
 		cancel:        func() {},
 		readerFactory: readerFactory{
@@ -178,7 +186,12 @@ func (c Config) buildManager(logger *zap.SugaredLogger, emit emit.Callback, fact
 		deleteAfterRead: c.DeleteAfterRead,
 		knownFiles:      make([]*reader, 0, 10),
 		seenPaths:       make(map[string]struct{}, 100),
-	}, nil
+	}
+	if useThreadPool.IsEnabled() {
+		manager.readerChan = make(chan readerWrapper, c.MaxConcurrentFiles)
+		manager.trie = trie.NewTrie()
+	}
+	return &manager, nil
 }
 
 func (c Config) validate() error {

pkg/stanza/fileconsumer/file.go

@@ -15,6 +15,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/fingerprint"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/trie"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/matcher"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
 )
@@ -45,6 +46,16 @@ type Manager struct {
 	seenPaths  map[string]struct{}
 
 	currentFps []*fingerprint.Fingerprint
+
+	// Following fields are used only when useThreadPool is enabled
+	workerWg       sync.WaitGroup
+	knownFilesLock sync.RWMutex
+
+	readerChan chan readerWrapper
+	trieLock   sync.RWMutex
+
+	// TRIE - this data structure stores the fingerprint of the files which are currently being consumed
+	trie *trie.Trie
 }
 
 func (m *Manager) Start(persister operator.Persister) error {
@@ -61,6 +72,11 @@ func (m *Manager) Start(persister operator.Persister) error {
 		m.Warnw("finding files", "error", err.Error())
 	}
 
+	// If useThreadPool is enabled, kick off the worker threads
+	if useThreadPool.IsEnabled() {
+		m.kickoffThreads(ctx)
+	}
+
 	// Start polling goroutine
 	m.startPoller(ctx)
 
@@ -71,6 +87,10 @@ func (m *Manager) Start(persister operator.Persister) error {
 func (m *Manager) Stop() error {
 	m.cancel()
 	m.wg.Wait()
+	if useThreadPool.IsEnabled() {
+		m.shutdownThreads()
+	}
+
 	m.roller.cleanup()
 	for _, reader := range m.knownFiles {
 		reader.Close()
@@ -95,14 +115,22 @@ func (m *Manager) startPoller(ctx context.Context) {
 				return
 			case <-globTicker.C:
 			}
-
 			m.poll(ctx)
 		}
 	}()
 }
 
 // poll checks all the watched paths for new entries
 func (m *Manager) poll(ctx context.Context) {
+	if useThreadPool.IsEnabled() {
+		m.pollConcurrent(ctx)
+	} else {
+		m.pollRegular(ctx)
+	}
+}
+
+// poll checks all the watched paths for new entries
+func (m *Manager) pollRegular(ctx context.Context) {
 	// Increment the generation on all known readers
 	// This is done here because the next generation is about to start
 	for i := 0; i < len(m.knownFiles); i++ {
@@ -134,6 +162,18 @@ func (m *Manager) poll(ctx context.Context) {
 	m.consume(ctx, matches)
 }
 
+func (m *Manager) readToEnd(ctx context.Context, r *reader) bool {
+	r.ReadToEnd(ctx)
+	if m.deleteAfterRead && r.eof {
+		r.Close()
+		if err := os.Remove(r.file.Name()); err != nil {
+			m.Errorf("could not delete %s", r.file.Name())
+		}
+		return true
+	}
+	return false
+}
+
 func (m *Manager) consume(ctx context.Context, paths []string) {
 	m.Debug("Consuming files")
 	readers := make([]*reader, 0, len(paths))
@@ -154,14 +194,7 @@ func (m *Manager) consume(ctx context.Context, paths []string) {
 		wg.Add(1)
 		go func(r *reader) {
 			defer wg.Done()
-			r.ReadToEnd(ctx)
-			// Delete a file if deleteAfterRead is enabled and we reached the end of the file
-			if m.deleteAfterRead && r.eof {
-				r.Close()
-				if err := os.Remove(r.file.Name()); err != nil {
-					m.Errorf("could not delete %s", r.file.Name())
-				}
-			}
+			m.readToEnd(ctx, r)
 		}(r)
 	}
 	wg.Wait()

pkg/stanza/fileconsumer/file_test.go

@@ -26,6 +26,20 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/testutil"
 )
 
+func TestMain(m *testing.M) {
+	// Run once with thread pool featuregate enabled
+	featuregate.GlobalRegistry().Set(useThreadPool.ID(), true) //nolint:all
+	if code := m.Run(); code > 0 {
+		os.Exit(code)
+	}
+	featuregate.GlobalRegistry().Set(useThreadPool.ID(), false) //nolint:all
+
+	// Run once with thread pool featuregate disabled
+	if code := m.Run(); code > 0 {
+		os.Exit(code)
+	}
+}
+
 func TestCleanStop(t *testing.T) {
 	t.Parallel()
 	t.Skip(`Skipping due to goroutine leak in opencensus.
@@ -78,10 +92,10 @@ func TestAddFileFields(t *testing.T) {
 // AddFileResolvedFields tests that the `log.file.name_resolved` and `log.file.path_resolved` fields are included
 // when IncludeFileNameResolved and IncludeFilePathResolved are set to true
 func TestAddFileResolvedFields(t *testing.T) {
+	t.Parallel()
 	if runtime.GOOS == windowsOS {
 		t.Skip("Windows symlinks usage disabled for now. See https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/21088")
 	}
-	t.Parallel()
 
 	tempDir := t.TempDir()
 	cfg := NewConfig().includeDir(tempDir)
@@ -445,7 +459,7 @@ func TestReadNewLogs(t *testing.T) {
 	tempDir := t.TempDir()
 	cfg := NewConfig().includeDir(tempDir)
 	cfg.StartAt = "beginning"
-	operator, emitCalls := buildTestManager(t, cfg)
+	operator, emitCalls := buildTestManager(t, cfg, withReaderChan())
 	operator.persister = testutil.NewMockPersister("test")
 
 	// Poll once so we know this isn't a new file
@@ -473,7 +487,7 @@ func TestReadExistingAndNewLogs(t *testing.T) {
 	tempDir := t.TempDir()
 	cfg := NewConfig().includeDir(tempDir)
 	cfg.StartAt = "beginning"
-	operator, emitCalls := buildTestManager(t, cfg)
+	operator, emitCalls := buildTestManager(t, cfg, withReaderChan())
 	operator.persister = testutil.NewMockPersister("test")
 
 	// Start with a file with an entry in it, and expect that entry
@@ -497,7 +511,7 @@ func TestStartAtEnd(t *testing.T) {
 
 	tempDir := t.TempDir()
 	cfg := NewConfig().includeDir(tempDir)
-	operator, emitCalls := buildTestManager(t, cfg)
+	operator, emitCalls := buildTestManager(t, cfg, withReaderChan())
 	operator.persister = testutil.NewMockPersister("test")
 
 	temp := openTemp(t, tempDir)
@@ -525,7 +539,7 @@ func TestStartAtEndNewFile(t *testing.T) {
 	tempDir := t.TempDir()
 	cfg := NewConfig().includeDir(tempDir)
 	cfg.StartAt = "beginning"
-	operator, emitCalls := buildTestManager(t, cfg)
+	operator, emitCalls := buildTestManager(t, cfg, withReaderChan())
 	operator.persister = testutil.NewMockPersister("test")
 
 	operator.poll(context.Background())
@@ -642,7 +656,7 @@ func TestSplitWrite(t *testing.T) {
 	tempDir := t.TempDir()
 	cfg := NewConfig().includeDir(tempDir)
 	cfg.StartAt = "beginning"
-	operator, emitCalls := buildTestManager(t, cfg)
+	operator, emitCalls := buildTestManager(t, cfg, withReaderChan())
 	operator.persister = testutil.NewMockPersister("test")
 
 	temp := openTemp(t, tempDir)
@@ -662,7 +676,7 @@ func TestIgnoreEmptyFiles(t *testing.T) {
 	tempDir := t.TempDir()
 	cfg := NewConfig().includeDir(tempDir)
 	cfg.StartAt = "beginning"
-	operator, emitCalls := buildTestManager(t, cfg)
+	operator, emitCalls := buildTestManager(t, cfg, withReaderChan())
 	operator.persister = testutil.NewMockPersister("test")
 
 	temp := openTemp(t, tempDir)
@@ -967,6 +981,9 @@ func TestManyLogsDelivered(t *testing.T) {
 
 func TestFileBatching(t *testing.T) {
 	t.Parallel()
+	if useThreadPool.IsEnabled() {
+		t.Skip(`Skipping for thread pool feature gate, as there's no concept of batching for thread pool`)
+	}
 
 	files := 100
 	linesPerFile := 10
@@ -983,7 +1000,7 @@ func TestFileBatching(t *testing.T) {
 	cfg.MaxConcurrentFiles = maxConcurrentFiles
 	cfg.MaxBatches = maxBatches
 	emitCalls := make(chan *emitParams, files*linesPerFile)
-	operator, _ := buildTestManager(t, cfg, withEmitChan(emitCalls))
+	operator, _ := buildTestManager(t, cfg, withEmitChan(emitCalls), withReaderChan())
 	operator.persister = testutil.NewMockPersister("test")
 
 	core, observedLogs := observer.New(zap.DebugLevel)
@@ -1339,7 +1356,8 @@ func TestDeleteAfterRead(t *testing.T) {
 	cfg.StartAt = "beginning"
 	cfg.DeleteAfterRead = true
 	emitCalls := make(chan *emitParams, totalLines)
-	operator, _ := buildTestManager(t, cfg, withEmitChan(emitCalls))
+	operator, _ := buildTestManager(t, cfg, withEmitChan(emitCalls), withReaderChan())
+	operator.persister = testutil.NewMockPersister("test")
 
 	operator.poll(context.Background())
 	actualTokens = append(actualTokens, waitForNTokens(t, emitCalls, totalLines)...)
@@ -1353,6 +1371,9 @@ func TestDeleteAfterRead(t *testing.T) {
 }
 
 func TestMaxBatching(t *testing.T) {
+	if useThreadPool.IsEnabled() {
+		t.Skip(`Skipping for thread pool feature gate, as there's no concept of batching for thread pool`)
+	}
 	t.Parallel()
 
 	files := 50
@@ -1370,7 +1391,7 @@ func TestMaxBatching(t *testing.T) {
 	cfg.MaxConcurrentFiles = maxConcurrentFiles
 	cfg.MaxBatches = maxBatches
 	emitCalls := make(chan *emitParams, files*linesPerFile)
-	operator, _ := buildTestManager(t, cfg, withEmitChan(emitCalls))
+	operator, _ := buildTestManager(t, cfg, withEmitChan(emitCalls), withReaderChan())
 	operator.persister = testutil.NewMockPersister("test")
 
 	core, observedLogs := observer.New(zap.DebugLevel)
@@ -1486,7 +1507,7 @@ func TestDeleteAfterRead_SkipPartials(t *testing.T) {
 	cfg.StartAt = "beginning"
 	cfg.DeleteAfterRead = true
 	emitCalls := make(chan *emitParams, longFileLines+1)
-	operator, _ := buildTestManager(t, cfg, withEmitChan(emitCalls))
+	operator, _ := buildTestManager(t, cfg, withEmitChan(emitCalls), withReaderChan())
 	operator.persister = testutil.NewMockPersister("test")
 
 	shortFile := openTemp(t, tempDir)
@@ -1526,6 +1547,8 @@ func TestDeleteAfterRead_SkipPartials(t *testing.T) {
 
 	// Stop consuming before long file has been fully consumed
 	cancel()
+	operator.cancel()
+	operator.workerWg.Wait()
 	wg.Wait()
 
 	// short file was fully consumed and should have been deleted
@@ -1609,6 +1632,9 @@ func TestHeaderPersistanceInHeader(t *testing.T) {
 	// one poll operation occurs between now and when we stop.
 	op1.poll(context.Background())
 
+	// for threadpool, as the poll is asynchronous, allow it to complete one poll cycle
+	time.Sleep(500 * time.Millisecond)
+
 	require.NoError(t, op1.Stop())
 
 	writeString(t, temp, "|headerField2: headerValue2\nlog line\n")
@@ -1636,7 +1662,7 @@ func TestStalePartialFingerprintDiscarded(t *testing.T) {
 	cfg := NewConfig().includeDir(tempDir)
 	cfg.FingerprintSize = 18
 	cfg.StartAt = "beginning"
-	operator, emitCalls := buildTestManager(t, cfg)
+	operator, emitCalls := buildTestManager(t, cfg, withReaderChan())
 	operator.persister = testutil.NewMockPersister("test")
 
 	// Both of they will be include