[AGNTLOG-74] Fix Logs File Launcher Blocking AD (#36803)

<!--
* Contributors are encouraged to read our [CONTRIBUTING](/CONTRIBUTING.md) documentation.
* Both Contributor and Reviewer Checklists are available at https://datadoghq.dev/datadog-agent/guidelines/contributing/#pull-requests.
* The pull request:
  * Should only fix one issue or add one feature at a time.
  * Must update the test suite for the relevant functionality.
  * Should pass all status checks before being reviewed or merged.
* Commit titles should be prefixed with general area of pull request's change.
* Please fill the below sections if possible with relevant information or links.
-->
### What does this PR do?

Moves `FilesToTail()` to a go function to run concurrently alongside the rest of the launcher. This allows the launcher to run without being blocked when gathering file tailers for sources that can take a while to fetch (such as those on remote file shares).

### Motivation

In the previous implementation, the `scan` function was called synchronously by the file launcher. If the scan operation took a long time, this synchronous call would block the rest of the launcher leading to missed sourcs.. Now, the launcher will proceed while `FilesToTail()` runs in another thread so it won't miss sources anymore.

### Describe how you validated your changes
<!--
Validate your changes before merge, ensuring that:
* Your PR is tested by static / unit / integrations / e2e tests
* Your PR description details which e2e tests cover your changes, if any
* The PR description contains details of how you validated your changes. If you validated changes manually and not through automated tests, add context on why automated tests did not fit your changes validation.

If you want additional validation by a second person, you can ask reviewers to do it. Describe how to set up an environment for manual tests in the PR description. Manual validation is expected to happen on every commit before merge.

Any manual validation step should then map to an automated test. Manual validation should not substitute automation, minus exceptions not supported by test tooling yet.
-->

1. Benchmarked scan to confirm that increasing numbers of files were the culprit
2. Built and ran the agent

### Possible Drawbacks / Trade-offs

None that I can think of. 

### Additional Notes
<!--
* Anything else we should know when reviewing?
* Include benchmarking information here whenever possible.
* Include info about alternatives that were considered and why the proposed
  version was chosen.
-->

Would like more assistance QA'ing as I'm not very familiar with concurrency.

Author: soberpeach

comp/logs/agent/agentimpl/agent_test.go

@@ -104,6 +104,8 @@ func (suite *AgentTestSuite) SetupTest() {
 	suite.configOverrides["logs_config.run_path"] = suite.testDir
 	// Shorter grace period for tests.
 	suite.configOverrides["logs_config.stop_grace_period"] = 1
+	// Set a short scan period to allow it to run in the time period of the tcp and http tests
+	suite.configOverrides["logs_config.file_scan_period"] = 1
 
 	fakeTagger := taggerfxmock.SetupFakeTagger(suite.T())
 	suite.tagger = fakeTagger

pkg/logs/launchers/file/launcher.go

@@ -7,6 +7,7 @@
 package file
 
 import (
+	"context"
 	"regexp"
 	"slices"
 	"time"
@@ -52,10 +53,12 @@ type Launcher struct {
 	// set to true if we want to use `ContainersLogsDir` to validate that a new
 	// pod log file is being attached to the correct containerID.
 	// Feature flag defaulting to false, use `logs_config.validate_pod_container_id`.
-	validatePodContainerID bool
-	scanPeriod             time.Duration
-	flarecontroller        *flareController.FlareController
-	tagger                 tagger.Component
+	validatePodContainerID  bool
+	scanPeriod              time.Duration
+	flarecontroller         *flareController.FlareController
+	tagger                  tagger.Component
+	filesChan               chan []*tailer.File
+	filesTailedBetweenScans []*tailer.File
 	// Stores pertinent information about old tailer when rotation occurs and fingerprinting isn't possible
 	oldInfoMap    map[string]*oldTailerInfo
 	fingerprinter *tailer.Fingerprinter
@@ -92,6 +95,7 @@ func NewLauncher(tailingLimit int, tailerSleepDuration time.Duration, validatePo
 		scanPeriod:             scanPeriod,
 		flarecontroller:        flarecontroller,
 		tagger:                 tagger,
+		filesChan:              make(chan []*tailer.File, 1),
 		oldInfoMap:             make(map[string]*oldTailerInfo),
 		fingerprinter:          tailer.NewFingerprinter(fingerprintConfig),
 	}
@@ -121,17 +125,33 @@ func (s *Launcher) run() {
 		close(s.done)
 	}()
 
+	ctx, cancel := context.WithCancel(context.Background())
 	for {
 		select {
 		case source := <-s.addedSources:
 			s.addSource(source)
 		case source := <-s.removedSources:
 			s.removeSource(source)
 		case <-scanTicker.C:
+
+			activeSourcesCopy := make([]*sources.LogSource, len(s.activeSources))
+			copy(activeSourcesCopy, s.activeSources)
+
+			// Clear files tailed between scans before starting new FilesToTail
+			s.filesTailedBetweenScans = s.filesTailedBetweenScans[:0]
+
+			scanTicker.Stop()
+			go func() {
+				s.filesChan <- s.fileProvider.FilesToTail(ctx, s.validatePodContainerID, activeSourcesCopy, s.registry)
+			}()
+		case files := <-s.filesChan:
 			s.cleanUpRotatedTailers()
-			// check if there are new files to tail, tailers to stop and tailer to restart because of file rotation
-			s.scan()
+
+			s.resolveActiveTailers(files)
+			scanTicker.Reset(s.scanPeriod)
 		case <-s.stop:
+			// Cancel the context passed to fileProvider.FilesToTail
+			cancel()
 			// no more file should be tailed
 			s.cleanup()
 			return
@@ -158,12 +178,23 @@ func (s *Launcher) cleanup() {
 	s.oldInfoMap = make(map[string]*oldTailerInfo)
 }
 
-// scan checks all the files we're expected to tail, compares them to the currently tailed files,
-// and triggers the required updates.
-// For instance, when a file is logrotated, its tailer will keep tailing the rotated file.
-// The Scanner needs to stop that previous tailer, and start a new one for the new file.
-func (s *Launcher) scan() {
-	files := s.fileProvider.FilesToTail(s.validatePodContainerID, s.activeSources, s.registry)
+// resolveActiveTailers checks all the files we're expected to tail, compares them to the
+// currently tailed files, and triggers the required updates.  For instance,
+// when a file is logrotated, its tailer will keep tailing the rotated file.
+// The Scanner needs to stop that previous tailer, and start a new one for the
+// new file.
+func (s *Launcher) resolveActiveTailers(files []*tailer.File) {
+	// resolveActiveTailers() receives the files parameter from FilesToTail(),
+	// which is called in the main run loop of launcher. FilesToTail() is always
+	// executed concurrently.  It is therefore possible that addSource() can be
+	// called while FilesToTail() is still running. Since FilesToTail() is only
+	// passed a copy of activeSources it is possible that it would miss new
+	// sources added by addsource() therefore scan() would unschedule a tailer
+	// added during a concurrent scan.  In order to mitigate that possibility, any
+	// tailers started while FilesToTail() is running need to be merged with the
+	// result of FilesToTail() to prevent scan() from unscheudling them.
+	files = append(files, s.filesTailedBetweenScans...)
+	s.filesTailedBetweenScans = s.filesTailedBetweenScans[:0]
 	filesTailed := make(map[string]bool)
 	var allFiles []string
 
@@ -329,6 +360,7 @@ func (s *Launcher) launchTailers(source *sources.LogSource) {
 		log.Warnf("Could not collect files: %v", err)
 		return
 	}
+
 	for _, file := range files {
 		if s.tailers.Count() >= s.tailingLimit {
 			return
@@ -361,7 +393,10 @@ func (s *Launcher) launchTailers(source *sources.LogSource) {
 			source.Config.TailingMode = mode.String()
 		}
 
-		s.startNewTailer(file, mode, fingerprint)
+		newTailerStarted := s.startNewTailer(file, mode, fingerprint)
+		if newTailerStarted {
+			s.filesTailedBetweenScans = append(s.filesTailedBetweenScans, file)
+		}
 	}
 }
 

pkg/logs/launchers/file/launcher_test.go

@@ -8,6 +8,7 @@
 package file
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"strings"
@@ -173,7 +174,7 @@ func (suite *BaseLauncherTestSuite) SetupTest() {
 	suite.s.registry = auditorMock.NewMockRegistry()
 	suite.s.activeSources = append(suite.s.activeSources, suite.source)
 	status.InitStatus(cfg, util.CreateSources([]*sources.LogSource{suite.source}))
-	suite.s.scan()
+	suite.s.resolveActiveTailers(suite.s.fileProvider.FilesToTail(context.Background(), suite.s.validatePodContainerID, suite.s.activeSources, suite.s.registry))
 }
 
 func (suite *BaseLauncherTestSuite) TearDownTest() {
@@ -204,7 +205,7 @@ func (suite *BaseLauncherTestSuite) TestLauncherScanWithoutLogRotation() {
 	msg = <-suite.outputChan
 	suite.Equal("hello world", string(msg.GetContent()))
 
-	s.scan()
+	s.resolveActiveTailers(suite.s.fileProvider.FilesToTail(context.Background(), suite.s.validatePodContainerID, suite.s.activeSources, suite.s.registry))
 	newTailer, _ = s.tailers.Get(getScanKey(suite.testPath, suite.source))
 	// testing that launcher did not have to create a new tailer
 	suite.True(tailer == newTailer)
@@ -232,7 +233,7 @@ func (suite *BaseLauncherTestSuite) TestLauncherScanWithLogRotation() {
 	os.Rename(suite.testPath, suite.testRotatedPath)
 	f, err := os.Create(suite.testPath)
 	suite.Nil(err)
-	s.scan()
+	s.resolveActiveTailers(suite.s.fileProvider.FilesToTail(context.Background(), suite.s.validatePodContainerID, suite.s.activeSources, suite.s.registry))
 	newTailer, _ = s.tailers.Get(getScanKey(suite.testPath, suite.source))
 	suite.True(tailer != newTailer)
 
@@ -273,7 +274,7 @@ func (suite *BaseLauncherTestSuite) TestLauncherScanWithLogRotationAndChecksum_R
 	suite.Nil(err)
 	suite.Nil(suite.testFile.Sync())
 
-	s.scan()
+	s.resolveActiveTailers(suite.s.fileProvider.FilesToTail(context.Background(), suite.s.validatePodContainerID, suite.s.activeSources, suite.s.registry))
 
 	// Read message to confirm tailer is working
 	msg := <-suite.outputChan
@@ -310,7 +311,7 @@ func (suite *BaseLauncherTestSuite) TestLauncherScanWithLogRotationAndChecksum_R
 	suite.Nil(f.Sync())
 	defer f.Close()
 
-	s.scan()
+	s.resolveActiveTailers(suite.s.fileProvider.FilesToTail(context.Background(), suite.s.validatePodContainerID, suite.s.activeSources, suite.s.registry))
 
 	newTailer, _ := s.tailers.Get(getScanKey(suite.testPath, suite.source))
 	suite.True(tailer != newTailer, "A new tailer should have been created due to content change")
@@ -354,7 +355,7 @@ func (suite *BaseLauncherTestSuite) TestLauncherScanWithLogRotationAndChecksum_N
 	suite.Nil(err)
 	suite.Nil(suite.testFile.Sync())
 
-	s.scan()
+	s.resolveActiveTailers(suite.s.fileProvider.FilesToTail(context.Background(), suite.s.validatePodContainerID, suite.s.activeSources, suite.s.registry))
 
 	// Read message
 	msg := <-suite.outputChan
@@ -376,7 +377,7 @@ func (suite *BaseLauncherTestSuite) TestLauncherScanWithLogRotationAndChecksum_N
 	suite.False(didRotate, "Should not detect rotation when writing to the same file")
 
 	// Scan again - should not trigger any rotation logic
-	s.scan()
+	s.resolveActiveTailers(suite.s.fileProvider.FilesToTail(context.Background(), suite.s.validatePodContainerID, suite.s.activeSources, suite.s.registry))
 
 	// Verify the same tailer is still being used
 	newTailer, _ := s.tailers.Get(getScanKey(suite.testPath, suite.source))
@@ -409,7 +410,7 @@ func (suite *BaseLauncherTestSuite) TestLauncherScanWithLogRotationCopyTruncate(
 	suite.Nil(err)
 
 	suite.Nil(suite.testFile.Sync())
-	s.scan()
+	s.resolveActiveTailers(suite.s.fileProvider.FilesToTail(context.Background(), suite.s.validatePodContainerID, suite.s.activeSources, suite.s.registry))
 
 	newTailer, _ = s.tailers.Get(getScanKey(suite.testPath, suite.source))
 	suite.True(tailer != newTailer)
@@ -427,13 +428,13 @@ func (suite *BaseLauncherTestSuite) TestLauncherScanWithFileRemovedAndCreated()
 	// remove file
 	err = os.Remove(suite.testPath)
 	suite.Nil(err)
-	s.scan()
+	s.resolveActiveTailers(suite.s.fileProvider.FilesToTail(context.Background(), suite.s.validatePodContainerID, suite.s.activeSources, suite.s.registry))
 	suite.Equal(tailerLen-1, s.tailers.Count())
 
 	// create file
 	_, err = os.Create(suite.testPath)
 	suite.Nil(err)
-	s.scan()
+	s.resolveActiveTailers(suite.s.fileProvider.FilesToTail(context.Background(), suite.s.validatePodContainerID, suite.s.activeSources, suite.s.registry))
 	suite.Equal(tailerLen, s.tailers.Count())
 }
 
@@ -495,7 +496,8 @@ func runLauncherScanStartNewTailerTest(t *testing.T, testDirs []string) {
 		file.Close()
 
 		// test scan from beginning
-		launcher.scan()
+		launcher.resolveActiveTailers(launcher.fileProvider.FilesToTail(context.Background(), launcher.validatePodContainerID, launcher.activeSources, launcher.registry))
+
 		assert.Equal(t, 1, launcher.tailers.Count())
 		msg = <-outputChan
 		assert.Equal(t, "hello", string(msg.GetContent()))
@@ -538,7 +540,8 @@ func runLauncherScanStartNewTailerForEmptyFileTest(t *testing.T, testDirs []stri
 	_, err := os.Create(fmt.Sprintf("%s/test.log", testDir))
 	assert.Nil(t, err)
 
-	launcher.scan()
+	launcher.resolveActiveTailers(launcher.fileProvider.FilesToTail(context.Background(), launcher.validatePodContainerID, launcher.activeSources, launcher.registry))
+
 	assert.Equal(t, 0, launcher.tailers.Count())
 }
 
@@ -586,7 +589,8 @@ func runLauncherScanStartNewTailerWithOneLineTest(t *testing.T, testDirs []strin
 	file.Close()
 
 	// test scan from beginning
-	launcher.scan()
+	launcher.resolveActiveTailers(launcher.fileProvider.FilesToTail(context.Background(), launcher.validatePodContainerID, launcher.activeSources, launcher.registry))
+
 	assert.Equal(t, 1, launcher.tailers.Count())
 }
 
@@ -637,7 +641,8 @@ func runLauncherScanStartNewTailerWithLongLineTest(t *testing.T, testDirs []stri
 	file.Close()
 
 	// test scan from beginning
-	launcher.scan()
+	launcher.resolveActiveTailers(launcher.fileProvider.FilesToTail(context.Background(), launcher.validatePodContainerID, launcher.activeSources, launcher.registry))
+
 	assert.Equal(t, 1, launcher.tailers.Count())
 }
 
@@ -896,7 +901,7 @@ func runLauncherScanWithTooManyFilesTest(t *testing.T, testDirs []string) {
 	defer status.Clear()
 
 	// test at scan
-	launcher.scan()
+	launcher.resolveActiveTailers(launcher.fileProvider.FilesToTail(context.Background(), launcher.validatePodContainerID, launcher.activeSources, launcher.registry))
 	assert.Equal(t, 2, launcher.tailers.Count())
 	// Confirm that all of the files have been keepalive'd even if they are not tailed
 	assert.Equal(t, 3, len(launcher.registry.(*auditorMock.Registry).KeepAlives))
@@ -905,7 +910,7 @@ func runLauncherScanWithTooManyFilesTest(t *testing.T, testDirs []string) {
 	err = os.Remove(path)
 	assert.Nil(t, err)
 
-	launcher.scan()
+	launcher.resolveActiveTailers(launcher.fileProvider.FilesToTail(context.Background(), launcher.validatePodContainerID, launcher.activeSources, launcher.registry))
 	assert.Equal(t, 2, launcher.tailers.Count())
 }
 
@@ -1014,14 +1019,14 @@ func runLauncherScanRecentFilesWithRemovalTest(t *testing.T, testDirs []string)
 	launcher := createLauncher()
 	defer status.Clear()
 
-	launcher.scan()
+	launcher.resolveActiveTailers(launcher.fileProvider.FilesToTail(context.Background(), launcher.validatePodContainerID, launcher.activeSources, launcher.registry))
 	assert.Equal(t, 2, launcher.tailers.Count())
 	assert.True(t, launcher.tailers.Contains(path("1.log")))
 	assert.True(t, launcher.tailers.Contains(path("2.log")))
 
 	// When ... the newest file gets rm'd
 	rmFile("2.log")
-	launcher.scan()
+	launcher.resolveActiveTailers(launcher.fileProvider.FilesToTail(context.Background(), launcher.validatePodContainerID, launcher.activeSources, launcher.registry))
 
 	// Then the next 2 most recently modified should be tailed
 	assert.Equal(t, 2, launcher.tailers.Count())
@@ -1080,14 +1085,14 @@ func runLauncherScanRecentFilesWithNewFilesTest(t *testing.T, testDirs []string)
 	launcher := createLauncher()
 	defer status.Clear()
 
-	launcher.scan()
+	launcher.resolveActiveTailers(launcher.fileProvider.FilesToTail(context.Background(), launcher.validatePodContainerID, launcher.activeSources, launcher.registry))
 	assert.Equal(t, 2, launcher.tailers.Count())
 	assert.True(t, launcher.tailers.Contains(path("1.log")))
 	assert.True(t, launcher.tailers.Contains(path("2.log")))
 
 	// When ... a newer file appears
 	createFile("7.log", baseTime.Add(time.Second*8))
-	launcher.scan()
+	launcher.resolveActiveTailers(launcher.fileProvider.FilesToTail(context.Background(), launcher.validatePodContainerID, launcher.activeSources, launcher.registry))
 
 	// Then it should be tailed
 	assert.Equal(t, 2, launcher.tailers.Count())
@@ -1096,7 +1101,7 @@ func runLauncherScanRecentFilesWithNewFilesTest(t *testing.T, testDirs []string)
 
 	// When ... an even newer file appears
 	createFile("a.log", baseTime.Add(time.Second*10))
-	launcher.scan()
+	launcher.resolveActiveTailers(launcher.fileProvider.FilesToTail(context.Background(), launcher.validatePodContainerID, launcher.activeSources, launcher.registry))
 
 	// Then it should be tailed
 	assert.Equal(t, 2, launcher.tailers.Count())
@@ -1150,7 +1155,7 @@ func runLauncherFileRotationTest(t *testing.T, testDirs []string) {
 	launcher := createLauncher()
 	defer status.Clear()
 
-	launcher.scan()
+	launcher.resolveActiveTailers(launcher.fileProvider.FilesToTail(context.Background(), launcher.validatePodContainerID, launcher.activeSources, launcher.registry))
 	assert.Equal(t, 2, launcher.tailers.Count())
 	assert.Equal(t, 0, len(launcher.rotatedTailers))
 	assert.True(t, launcher.tailers.Contains(path("c.log")))
@@ -1168,7 +1173,7 @@ func runLauncherFileRotationTest(t *testing.T, testDirs []string) {
 	assert.Nil(t, err)
 	assert.True(t, didRotate)
 
-	launcher.scan()
+	launcher.resolveActiveTailers(launcher.fileProvider.FilesToTail(context.Background(), launcher.validatePodContainerID, launcher.activeSources, launcher.registry))
 	assert.Equal(t, launcher.tailers.Count(), 2)
 	assert.Equal(t, 1, len(launcher.rotatedTailers))
 	assert.True(t, launcher.tailers.Contains(path("c.log")))
@@ -1223,14 +1228,14 @@ func runLauncherFileDetectionSingleScanTest(t *testing.T, testDirs []string) {
 	launcher := createLauncher()
 	defer status.Clear()
 
-	launcher.scan()
+	launcher.resolveActiveTailers(launcher.fileProvider.FilesToTail(context.Background(), launcher.validatePodContainerID, launcher.activeSources, launcher.registry))
 	assert.Equal(t, 2, launcher.tailers.Count())
 	assert.True(t, launcher.tailers.Contains(path("a.log")))
 	assert.True(t, launcher.tailers.Contains(path("b.log")))
 
 	createFile("z.log")
 
-	launcher.scan()
+	launcher.resolveActiveTailers(launcher.fileProvider.FilesToTail(context.Background(), launcher.validatePodContainerID, launcher.activeSources, launcher.registry))
 	assert.Equal(t, launcher.tailers.Count(), 2)
 	assert.True(t, launcher.tailers.Contains(path("z.log")))
 	assert.True(t, launcher.tailers.Contains(path("b.log")))
@@ -1264,7 +1269,7 @@ func (suite *BaseLauncherTestSuite) TestLauncherDoesNotCreateTailerForTruncatedU
 	suite.Nil(err)
 	suite.Nil(suite.testFile.Sync())
 
-	s.scan()
+	s.resolveActiveTailers(suite.s.fileProvider.FilesToTail(context.Background(), suite.s.validatePodContainerID, suite.s.activeSources, suite.s.registry))
 
 	// Read message to confirm tailer is working
 	msg := <-suite.outputChan
@@ -1287,7 +1292,7 @@ func (suite *BaseLauncherTestSuite) TestLauncherDoesNotCreateTailerForTruncatedU
 	suite.True(didRotate, "Should detect rotation when file becomes empty (fingerprint = 0)")
 
 	// Now test the launcher's behavior: it should NOT create a new tailer for the undersized file
-	s.scan()
+	s.resolveActiveTailers(suite.s.fileProvider.FilesToTail(context.Background(), suite.s.validatePodContainerID, suite.s.activeSources, suite.s.registry))
 
 	// Verify no new tailer was created for the undersized file
 	// The old tailer should be removed but no new one should be created
@@ -1323,7 +1328,7 @@ func (suite *BaseLauncherTestSuite) TestLauncherDoesNotCreateTailerForRotatedUnd
 	suite.Nil(err)
 	suite.Nil(suite.testFile.Sync())
 
-	s.scan()
+	s.resolveActiveTailers(suite.s.fileProvider.FilesToTail(context.Background(), suite.s.validatePodContainerID, suite.s.activeSources, suite.s.registry))
 
 	// Read message to confirm tailer is working
 	msg := <-suite.outputChan
@@ -1350,7 +1355,7 @@ func (suite *BaseLauncherTestSuite) TestLauncherDoesNotCreateTailerForRotatedUnd
 	suite.True(didRotate, "Should detect rotation when original file is moved and new file is created")
 
 	// Now test the launcher's behavior: it should NOT create a new tailer for the undersized file
-	s.scan()
+	s.resolveActiveTailers(suite.s.fileProvider.FilesToTail(context.Background(), suite.s.validatePodContainerID, suite.s.activeSources, suite.s.registry))
 
 	// Verify no new tailer was created for the undersized file
 	// The old tailer should be removed but no new one should be created