[bug][filestream] - Fix filebeat GC cleanup bug (#40258)

* chore: initial commit

* chore: remove default 30mins

* fix: set -1 for default

* fix: typo

* fix: remove validator

* fix: typo

* chore: update reference.yml

* chore: add changelog

* fix: update reference.yml

* fix: address review comments

* fix: typo

(cherry picked from commit 1ca6e2f)

CHANGELOG.next.asciidoc

@@ -44,6 +44,9 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Update Salesforce module to use new Salesforce input. {pull}37509[37509]
 - Tag events that come from a filestream in "take over" mode. {pull}39828[39828]
 - Fix high IO and handling of a corrupted registry log file. {pull}35893[35893]
+- Filebeat, when running with Elastic-Agent, reports status for Filestream input. {pull}40121[40121]
+- Implement Elastic Agent status and health reporting for Winlog Filebeat input. {pull}40163[40163]
+- Fix filestream's registry GC: registry entries will never be removed if clean_inactive is set to "-1". {pull}40258[40258]
 
 *Heartbeat*
 

filebeat/_meta/config/filebeat.inputs.reference.yml.tmpl

@@ -432,7 +432,7 @@ filebeat.inputs:
 
   # Files for the modification data is older than clean_inactive the state from the registry is removed
   # By default this is disabled.
-  #clean_inactive: 0
+  #clean_inactive: -1
 
   # Removes the state for files which cannot be found on disk anymore immediately
   #clean_removed: true

filebeat/filebeat.reference.yml

@@ -839,7 +839,7 @@ filebeat.inputs:
 
   # Files for the modification data is older than clean_inactive the state from the registry is removed
   # By default this is disabled.
-  #clean_inactive: 0
+  #clean_inactive: -1
 
   # Removes the state for files which cannot be found on disk anymore immediately
   #clean_removed: true

filebeat/input/filestream/config.go

@@ -33,12 +33,14 @@ import (
 type config struct {
 	Reader readerConfig `config:",inline"`
 
-	ID             string             `config:"id"`
-	Paths          []string           `config:"paths"`
-	Close          closerConfig       `config:"close"`
-	FileWatcher    *conf.Namespace    `config:"prospector"`
-	FileIdentity   *conf.Namespace    `config:"file_identity"`
-	CleanInactive  time.Duration      `config:"clean_inactive" validate:"min=0"`
+	ID           string          `config:"id"`
+	Paths        []string        `config:"paths"`
+	Close        closerConfig    `config:"close"`
+	FileWatcher  *conf.Namespace `config:"prospector"`
+	FileIdentity *conf.Namespace `config:"file_identity"`
+
+	// -1 means that registry will never be cleaned
+	CleanInactive  time.Duration      `config:"clean_inactive" validate:"min=-1"`
 	CleanRemoved   bool               `config:"clean_removed"`
 	HarvesterLimit uint32             `config:"harvester_limit" validate:"min=0"`
 	IgnoreOlder    time.Duration      `config:"ignore_older"`
@@ -98,7 +100,7 @@ func defaultConfig() config {
 		Reader:         defaultReaderConfig(),
 		Paths:          []string{},
 		Close:          defaultCloserConfig(),
-		CleanInactive:  0,
+		CleanInactive:  -1,
 		CleanRemoved:   true,
 		HarvesterLimit: 0,
 		IgnoreOlder:    0,

filebeat/input/filestream/input.go

@@ -74,10 +74,11 @@ func Plugin(log *logp.Logger, store loginp.StateStore) input.Plugin {
 		Info:       "filestream input",
 		Doc:        "The filestream input collects logs from the local filestream service",
 		Manager: &loginp.InputManager{
-			Logger:     log,
-			StateStore: store,
-			Type:       pluginName,
-			Configure:  configure,
+			Logger:              log,
+			StateStore:          store,
+			Type:                pluginName,
+			Configure:           configure,
+			DefaultCleanTimeout: -1,
 		},
 	}
 }

filebeat/input/filestream/internal/input-logfile/clean.go

@@ -130,5 +130,7 @@ func checkCleanResource(started, now time.Time, resource *resource) bool {
 		reference = started
 	}
 
-	return reference.Add(ttl).Before(now) && resource.stored
+	// if ttl is negative, we never delete the entry
+	// else check for time elapsed
+	return ttl >= 0 && reference.Add(ttl).Before(now) && resource.stored
 }

filebeat/input/filestream/internal/input-logfile/clean_test.go

@@ -159,4 +159,24 @@ func TestGCStore(t *testing.T) {
 		want := map[string]state{}
 		checkEqualStoreState(t, want, backend.snapshot())
 	})
+
+	t.Run("state never removed with ttl=-1", func(t *testing.T) {
+
+		// keep started as a large value
+		started := time.Now().Add(-1 * time.Hour * 24 * 356) // cleanup process is running for a while already
+
+		initState := map[string]state{
+			"test::key": {
+				TTL:     -1,
+				Updated: started,
+			},
+		}
+
+		backend := createSampleStore(t, initState)
+		store := testOpenStore(t, "test", backend)
+		defer store.Release()
+
+		gcStore(logp.NewLogger("test"), started, store)
+		checkEqualStoreState(t, initState, backend.snapshot())
+	})
 }

filebeat/input/filestream/internal/input-logfile/manager.go

@@ -94,9 +94,6 @@ type StateStore interface {
 
 func (cim *InputManager) init() error {
 	cim.initOnce.Do(func() {
-		if cim.DefaultCleanTimeout <= 0 {
-			cim.DefaultCleanTimeout = 30 * time.Minute
-		}
 
 		log := cim.Logger.With("input_type", cim.Type)
 

x-pack/filebeat/filebeat.reference.yml

@@ -2541,7 +2541,7 @@ filebeat.inputs:
 
   # Files for the modification data is older than clean_inactive the state from the registry is removed
   # By default this is disabled.
-  #clean_inactive: 0
+  #clean_inactive: -1
 
   # Removes the state for files which cannot be found on disk anymore immediately
   #clean_removed: true