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

VihasMakwana · 2023-06-05T05:44:36Z

Description: Added a new feature gate that enables a thread pool mechanism to respect the poll_interval parameter

Current Scenario:

If a file takes longer than poll_interval to consume, the current implementation would block until it consumes entirely. In other words, it doesn't respect poll_interval.

Improvisation using thread pooling:

In a thread pool model, the backend will queue the files as it proceeds and won't wait for them to consume, all the reading will be asynchronous.

Link to tracking Issue: #18908

Testing: Nothing new added, existing ones are modified as per the feature gate

I will provide benchmarks in the comments.

VihasMakwana · 2023-06-09T08:50:52Z

will add a changelog entry, @djaglowski please review it!

djaglowski

@VihasMakwana, thanks for continuing on this. I still want to be very careful here but this is looking like a much simpler PR than we had before.

djaglowski · 2023-06-09T14:04:10Z

pkg/stanza/fileconsumer/file_test.go

+	if useThreadPool.IsEnabled() {
+		operator, emitCalls = buildTestManagerWithOptions(t, cfg, withReaderChan())
+	}


We could probably make better use of the options pattern here, since we are doing this everywhere.

Maybe the options can be set in TestMain and then we can always just call buildTestManager(t, cfg, options...)

pkg/stanza/fileconsumer/trie.go

djaglowski · 2023-06-09T14:07:32Z

pkg/stanza/fileconsumer/trie.go

+
+package fileconsumer
+
+type Trie struct {


I think we need thorough tests for the trie itself. What do you think about adding the trie and dedicated tests to fileconsumer/internal in a separate PR?

Yeah, we can do that. Will reduce the the time complexity of this PR

djaglowski · 2023-06-09T14:09:40Z

pkg/stanza/operator/helper/flusher.go

+	f.rwLock.Lock()
+	defer f.rwLock.Unlock()


We'll have to ensure this doesn't impact performance when the gate is not enabled. If benchmarks can show it's not an issue, that's fine but otherwise can we just check the gate?

djaglowski · 2023-06-09T14:10:32Z

receiver/filelogreceiver/filelog_test.go

+	if rt.enableThreadPool {
+		t.Cleanup(func() {
+			require.NoError(t, featuregate.GlobalRegistry().Set("filelog.useThreadPool", false))
+		})
+		require.NoError(t, featuregate.GlobalRegistry().Set("filelog.useThreadPool", true))
+	}


Does this test require its own management of the gate? Isn't it covered in TestMain?

yes because both of them are separate packages. TestMain will only cover the fileconsumer package.

djaglowski · 2023-06-09T14:12:55Z

pkg/stanza/fileconsumer/file.go

@@ -77,6 +97,13 @@ func (m *Manager) Start(persister operator.Persister) error {
 func (m *Manager) Stop() error {
 	m.cancel()
 	m.wg.Wait()
+	if useThreadPool.IsEnabled() {
+		close(m.readerChan)


In Start, it's possible that we return before creating the channel, so we need to check if the channel is nil. This can crash the collector from an otherwise recoverable situation.

pkg/stanza/fileconsumer/file_threadpool.go

djaglowski · 2023-06-09T14:16:44Z

pkg/stanza/fileconsumer/file_threadpool.go

+	// Get the list of paths on disk
+	matches := m.finder.FindFiles()
+	m.consumeConcurrent(ctx, matches)
+	m.clearCurrentFingerprints()


I think I asked you this before but I can't recall. Why can we do this in an asynchronous situation?

djaglowski · 2023-06-09T14:17:55Z

pkg/stanza/fileconsumer/file_threadpool.go

+		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())
+			}
+		} else {
+			// Save off any files that were not fully read or if deleteAfterRead is disabled
+			m.saveCurrentConcurrent(r)
+		}


Is it possible to deduplicate this code?

djaglowski · 2023-06-09T14:18:18Z

pkg/stanza/fileconsumer/file_threadpool.go

+	if _, ok := m.seenPaths[filePath]; !ok {
+		if m.readerFactory.fromBeginning {
+			m.Infow("Started watching file", "path", filePath)
+		} else {
+			m.Infow("Started watching file from end. To read preexisting logs, configure the argument 'start_at' to 'beginning'", "path", filePath)
+		}
+		m.seenPaths[filePath] = struct{}{}
+	}
+	file, err := os.Open(filePath) // #nosec - operator must read in files defined by user
+	if err != nil {
+		m.Debugf("Failed to open file", zap.Error(err))
+		return nil, nil
+	}
+	fp, err := m.readerFactory.newFingerprint(file)
+	if err != nil {
+		m.Errorw("Failed creating fingerprint", zap.Error(err))
+		return nil, nil
+	}
+	// Exclude any empty fingerprints or duplicate fingerprints to avoid doubling up on copy-truncate files
+
+	if len(fp.FirstBytes) == 0 {
+		if err = file.Close(); err != nil {
+			m.Errorf("problem closing file", "file", file.Name())
+		}
+		return nil, nil
+	}


I think all of this is duplicated as well. Can we extract it somehow?

Co-authored-by: Daniel Jaglowski <[email protected]>

djaglowski · 2023-06-22T14:57:01Z

@VihasMakwana, #23415 is merged, please rebase.

…regate

github-actions · 2023-07-10T05:21:22Z

This PR was marked stale due to lack of activity. It will be closed in 14 days.

github-actions · 2023-07-25T05:19:23Z

Closed as inactive. Feel free to reopen if this PR is still being worked on.

VihasMakwana · 2023-07-28T15:01:13Z

@djaglowski lets' keep this one closed, will reopen a fresh PR after we merge our trie's PR

Description: Add Trie data structure and keep it separate from PR #23056 Testing: Relevant test cases added --------- Co-authored-by: Dan Jaglowski <[email protected]>

Fixes open-telemetry#17846

1e71ef1

VihasMakwana requested a review from a team June 5, 2023 05:44

VihasMakwana requested a review from djaglowski as a code owner June 5, 2023 05:44

VihasMakwana marked this pull request as draft June 5, 2023 05:44

github-actions bot assigned TylerHelmuth Jun 5, 2023

github-actions bot added pkg/stanza receiver/filelog labels Jun 5, 2023

djaglowski changed the title ~~feature: Add a new feature gate~~ [pkg/stanza/fileconsumer] Add ability to read files asynchronously Jun 6, 2023

VihasMakwana marked this pull request as ready for review June 9, 2023 08:50

djaglowski reviewed Jun 9, 2023

View reviewed changes

VihasMakwana and others added 3 commits June 12, 2023 10:56

Update copyright header to new format

64901f2

Co-authored-by: Daniel Jaglowski <[email protected]>

Remove unnecessary export

5b3f2da

Co-authored-by: Daniel Jaglowski <[email protected]>

Address review comments

06d65ba

VihasMakwana mentioned this pull request Jun 15, 2023

[chore][pkg/stanza/fileconsumer] Add utility functions #23415

Merged

VihasMakwana requested a review from djaglowski June 19, 2023 06:48

VihasMakwana force-pushed the filelogreceiver_featuregate branch from 4ff27c6 to b501c5d Compare June 19, 2023 06:49

Merge branch 'main' into filelogreceiver_featuregate

935a08f

VihasMakwana force-pushed the filelogreceiver_featuregate branch from b501c5d to 935a08f Compare June 19, 2023 06:50

Move saveReaders to another go routine.

d105d85

VihasMakwana mentioned this pull request Jun 25, 2023

[pkg/stanza/fileconsumer] Add trie and test cases #23665

Closed

Merge remote-tracking branch 'origin/main' into filelogreceiver_featu…

039a6f2

…regate

VihasMakwana force-pushed the filelogreceiver_featuregate branch from 10e0001 to 039a6f2 Compare June 25, 2023 14:21

Vihas Splunk added 2 commits June 25, 2023 21:15

Update test cases

29f5dc9

Add license

2bbcc66

VihasMakwana force-pushed the filelogreceiver_featuregate branch 2 times, most recently from 9696336 to a989864 Compare June 25, 2023 16:18

FIx rotation bug

1fa0744

VihasMakwana force-pushed the filelogreceiver_featuregate branch from a989864 to 1fa0744 Compare June 25, 2023 16:22

github-actions bot added the Stale label Jul 10, 2023

github-actions bot closed this Jul 25, 2023

VihasMakwana mentioned this pull request Aug 7, 2023

[pkg/stanza/fileconsumer] Add trie and test cases #24982

Merged

h0cheung mentioned this pull request Aug 20, 2023

New component: k8slog receiver #23339

Closed

2 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

VihasMakwana commented Jun 5, 2023

VihasMakwana commented Jun 9, 2023

djaglowski left a comment

djaglowski Jun 9, 2023

djaglowski Jun 9, 2023

VihasMakwana Jun 12, 2023

djaglowski Jun 9, 2023

VihasMakwana Jun 12, 2023

djaglowski Jun 9, 2023

VihasMakwana Jun 12, 2023

djaglowski Jun 9, 2023

djaglowski Jun 9, 2023

djaglowski Jun 9, 2023

djaglowski Jun 9, 2023

djaglowski commented Jun 22, 2023

github-actions bot commented Jul 10, 2023

github-actions bot commented Jul 25, 2023

VihasMakwana commented Jul 28, 2023

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

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

Conversation

VihasMakwana commented Jun 5, 2023

VihasMakwana commented Jun 9, 2023

djaglowski left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

djaglowski commented Jun 22, 2023

github-actions bot commented Jul 10, 2023

github-actions bot commented Jul 25, 2023

VihasMakwana commented Jul 28, 2023