[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/benchmark_test.go

@@ -4,10 +4,15 @@
 package fileconsumer
 
 import (
+	"context"
+	"fmt"
 	"os"
 	"path/filepath"
+	"strings"
+	"sync"
 	"testing"
 
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/fingerprint"
@@ -20,6 +25,10 @@ type fileInputBenchmark struct {
 	config func() *Config
 }
 
+type fileSizeBenchmark struct {
+	sizes [2]int
+}
+
 type benchFile struct {
 	*os.File
 	log func(int)
@@ -180,3 +189,66 @@ func BenchmarkFileInput(b *testing.B) {
 		})
 	}
 }
+
+func max(x, y int) int {
+	if x < y {
+		return y
+	}
+	return x
+}
+
+func (fileSize fileSizeBenchmark) createFiles(b *testing.B, rootDir string) {
+	// create 50 files, some with large file sizes, other's with rather small
+	getMessage := func(m int) string { return fmt.Sprintf("message %d", m) }
+	logs := make([]string, 0)
+	for i := 0; i < max(fileSize.sizes[0], fileSize.sizes[1]); i++ {
+		logs = append(logs, getMessage(i))
+	}
+
+	for i := 0; i < 50; i++ {
+		file := openFile(b, filepath.Join(rootDir, fmt.Sprintf("file_%s.log", uuid.NewString())))
+		file.WriteString(uuid.NewString() + strings.Join(logs[:fileSize.sizes[i%2]], "\n") + "\n")
+	}
+}
+
+func BenchmarkFileSizeVarying(b *testing.B) {
+	fileSize := fileSizeBenchmark{
+		sizes: [2]int{b.N * 5000, b.N * 10}, // Half the files will be huge, other half will be smaller
+	}
+	rootDir := b.TempDir()
+	cfg := NewConfig().includeDir(rootDir)
+	cfg.StartAt = "beginning"
+	cfg.MaxConcurrentFiles = 50
+	totalLogs := fileSize.sizes[0]*50 + fileSize.sizes[1]*50
+	emitCalls := make(chan *emitParams, totalLogs+10)
+
+	operator, _ := buildTestManager(b, cfg, withEmitChan(emitCalls), withReaderChan())
+	operator.persister = testutil.NewMockPersister("test")
+	defer func() {
+		require.NoError(b, operator.Stop())
+	}()
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		var once sync.Once
+		for i := 0; i < totalLogs; i++ {
+			once.Do(func() {
+				// Reset once we get the first log
+				b.ResetTimer()
+			})
+			<-emitCalls
+		}
+		// Stop the timer, as we're measuring log throughput
+		b.StopTimer()
+	}()
+	// create first set of files
+	fileSize.createFiles(b, rootDir)
+	operator.poll(context.Background())
+
+	// create new set of files, call poll() again
+	fileSize.createFiles(b, rootDir)
+	operator.poll(context.Background())
+
+	wg.Wait()
+}

pkg/stanza/fileconsumer/config.go

@@ -17,6 +17,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"
@@ -35,6 +36,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,
@@ -153,7 +161,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{
@@ -180,7 +188,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()