[Filestream] migrate state from previous ID to current

CHANGELOG.next.asciidoc

@@ -413,6 +413,7 @@ otherwise no tag is added. {issue}42208[42208] {pull}42403[42403]
 - Publish events progressively in the Okta provider of the Entity Analytics input. {issue}40106[40106] {pull}42567[42567]
 - The journald input is now generally available. {pull}42107[42107]
 - Add metrics for number of events and pages published by HTTPJSON input. {issue}42340[42340] {pull}42442[42442]
+- Filestram input now supports migrating state when changing its ID, for that set `previous_ids`. {issue}42472[42472] {pull}42624[42624]
 
 *Auditbeat*
 

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

@@ -247,6 +247,13 @@ filebeat.inputs:
   # Change to true to enable this input configuration.
   enabled: false
 
+  # Previous IDs from this input. This allows migrating the state from files
+  # ingested when running this input with a different ID.
+  # Only the state of existing files currently being harvested by the
+  # input will be migrated.
+  #previous_ids:
+    #- "old-id"
+
   # Paths that should be crawled and fetched. Glob based paths.
   # To fetch all ".log" files from a specific level of subdirectories
   # /var/log/*/*.log can be used.

filebeat/docs/inputs/input-filestream-file-options.asciidoc

@@ -40,6 +40,32 @@ inputs that use duplicated IDs.
 WARNING: Duplicated IDs will lead to data duplication and some input
 instances will not produce any metrics.
 
+[float]
+[id="{beatname_lc}-input-{type}-previous_ids"]
+===== `previous_ids`
+
+This allows {beatname_uc} to migrate the state of the files harvested
+under a different ID before the input starts. Only the state of
+existing files currently being harvested by the input are migrated.
+
+The file state with the old ID will be removed from the registry after
+the migration.
+
+WARNING: Ensure there are no duplicated Filestream input IDs before
+using `previous_ids` and that different input instances do not share
+any `previous_ids`.
+
+["source","yaml",subs="attributes"]
+----
+{beatname_lc}.inputs:
+- type: {type}
+  id: "current-id"
+  paths:
+    - /var/log/*.log
+  previous_ids:
+    - "old-id"
+----
+
 [float]
 [[filestream-input-paths]]
 ===== `paths`

filebeat/filebeat.reference.yml

@@ -660,6 +660,13 @@ filebeat.inputs:
   # Change to true to enable this input configuration.
   enabled: false
 
+  # Previous IDs from this input. This allows migrating the state from files
+  # ingested when running this input with a different ID.
+  # Only the state of existing files currently being harvested by the
+  # input will be migrated.
+  #previous_ids:
+    #- "old-id"
+
   # Paths that should be crawled and fetched. Glob based paths.
   # To fetch all ".log" files from a specific level of subdirectories
   # /var/log/*/*.log can be used.

filebeat/input/filestream/config.go

@@ -36,6 +36,7 @@ type config struct {
 	Reader readerConfig `config:",inline"`
 
 	ID           string          `config:"id"`
+	PreviousIDs  []string        `config:"previous_ids"`
 	Paths        []string        `config:"paths"`
 	Close        closerConfig    `config:"close"`
 	FileWatcher  *conf.Namespace `config:"prospector"`

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

@@ -85,7 +85,7 @@
 
 	prospectorStore := inp.manager.getRetainedStore()
 	defer prospectorStore.Release()
-	sourceStore := newSourceStore(prospectorStore, inp.sourceIdentifier)
+	sourceStore := newSourceStore(prospectorStore, inp.sourceIdentifier, nil)
 
 	// Mark it as running for now.
 	// Any errors encountered by harverter will change state to Degraded
@@ -122,7 +122,7 @@
 			return
 		}
 
 		op := private[last].(*updateOp)
 		ch.Send(scheduledUpdate{op: op, n: n})
 	})
 }

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

@@ -156,7 +156,9 @@ func (cim *InputManager) Create(config *conf.C) (v2.Input, error) {
 	}
 
 	settings := struct {
+		// All those values are duplicated from the Filestream configuration
 		ID                 string        `config:"id"`
+		PreviousIDs        []string      `config:"previous_ids"`
 		CleanInactive      time.Duration `config:"clean_inactive"`
 		HarvesterLimit     uint64        `config:"harvester_limit"`
 		AllowIDDuplication bool          `config:"allow_deprecated_id_duplication"`
@@ -219,25 +221,38 @@ func (cim *InputManager) Create(config *conf.C) (v2.Input, error) {
 		return nil, errNoInputRunner
 	}
 
-	sourceIdentifier, err := newSourceIdentifier(cim.Type, settings.ID)
+	srcIdentifier, err := newSourceIdentifier(cim.Type, settings.ID)
 	if err != nil {
 		return nil, fmt.Errorf("error while creating source identifier for input: %w", err)
 	}
 
+	var previousSrcIdentifiers []*sourceIdentifier
+	for _, id := range settings.PreviousIDs {
+		si, err := newSourceIdentifier(cim.Type, id)
+		if err != nil {
+			return nil,
+				fmt.Errorf(
+					"[ID: %q] error while creating source identifier for previous ID %q: %w",
+					settings.ID, id, err)
+		}
+
+		previousSrcIdentifiers = append(previousSrcIdentifiers, si)
+	}
+
 	pStore := cim.getRetainedStore()
 	defer pStore.Release()
 
-	prospectorStore := newSourceStore(pStore, sourceIdentifier)
+	prospectorStore := newSourceStore(pStore, srcIdentifier, previousSrcIdentifiers)
 
 	// create a store with the deprecated global ID. This will be used to
 	// migrate the entries in the registry to use the new input ID.
 	globalIdentifier, err := newSourceIdentifier(cim.Type, "")
 	if err != nil {
 		return nil, fmt.Errorf("cannot create global identifier for input: %w", err)
 	}
-	globalStore := newSourceStore(pStore, globalIdentifier)
+	defaultIDStore := newSourceStore(pStore, globalIdentifier, nil)
 
-	err = prospector.Init(prospectorStore, globalStore, sourceIdentifier.ID)
+	err = prospector.Init(prospectorStore, defaultIDStore, srcIdentifier.ID)
 	if err != nil {
 		return nil, err
 	}
@@ -249,7 +264,7 @@ func (cim *InputManager) Create(config *conf.C) (v2.Input, error) {
 		metricsID:        metricsID,
 		prospector:       prospector,
 		harvester:        harvester,
-		sourceIdentifier: sourceIdentifier,
+		sourceIdentifier: srcIdentifier,
 		cleanTimeout:     settings.CleanInactive,
 		harvesterLimit:   settings.HarvesterLimit,
 	}, nil

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

@@ -47,7 +47,7 @@ func (s *testSource) Name() string {
 
 type noopProspector struct{}
 
-func (m noopProspector) Init(_, _ ProspectorCleaner, _ func(Source) string) error {
+func (m noopProspector) Init(_, _ StoreUpdater, _ func(Source) string) error {
 	return nil
 }
 

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

@@ -25,8 +25,10 @@ import (
 // based on the retrieved information about the configured paths.
 // It also updates the statestore with the meta data of the running harvesters.
 type Prospector interface {
-	// Init runs the cleanup processes before starting the prospector.
-	Init(c, g ProspectorCleaner, newID func(Source) string) error
+	// Init updates the store before starting the prospector
+	// it receives two StoreUpdater: one global and another local
+	// to this prospector instance.
+	Init(local, global StoreUpdater, newID func(Source) string) error
 	// Run starts the event loop and handles the incoming events
 	// either by starting/stopping a harvester, or updating the statestore.
 	Run(input.Context, StateMetadataUpdater, HarvesterGroup)
@@ -48,14 +50,19 @@ type StateMetadataUpdater interface {
 	ResetCursor(s Source, cur interface{}) error
 }
 
-// ProspectorCleaner cleans the state store before it starts running.
-type ProspectorCleaner interface {
+// StoreUpdater allows manipulation of the state store
+type StoreUpdater interface {
 	// CleanIf removes an entry if the function returns true
 	CleanIf(func(v Value) bool)
 	// UpdateIdentifiers updates ID in the registry.
 	// The function passed to UpdateIdentifiers must return an empty string if the key
 	// remains the same.
 	UpdateIdentifiers(func(v Value) (string, interface{}))
+	// CopyStatesFromPreviousIDs copies the state from a previous input ID to the
+	// current input ID.
+	// The function passed to CopyStatesFromPreviousIDs must return an empty
+	// string if the entry should not be copied.
+	CopyStatesFromPreviousIDs(func(v Value) (string, interface{}))
 }
 
 // Value contains the cursor metadata.

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

@@ -36,8 +36,9 @@ import (
 // sourceStore is a store which can access resources using the Source
 // from an input.
 type sourceStore struct {
-	identifier *sourceIdentifier
-	store      *store
+	identifier          *sourceIdentifier
+	previousIdentifiers []*sourceIdentifier
+	store               *store
 }
 
 // store encapsulates the persistent store and the in memory state store, that
@@ -163,10 +164,11 @@ func openStore(log *logp.Logger, statestore StateStore, prefix string) (*store,
 	}, nil
 }
 
-func newSourceStore(s *store, identifier *sourceIdentifier) *sourceStore {
+func newSourceStore(s *store, identifier *sourceIdentifier, previousIdentifiers []*sourceIdentifier) *sourceStore {
 	return &sourceStore{
-		store:      s,
-		identifier: identifier,
+		store:               s,
+		identifier:          identifier,
+		previousIdentifiers: previousIdentifiers,
 	}
 }
 
@@ -212,9 +214,10 @@ func (s *sourceStore) CleanIf(pred func(v Value) bool) {
 	}
 }
 
-// UpdateIdentifiers copies an existing resource to a new ID and marks the previous one
-// for removal.
-func (s *sourceStore) UpdateIdentifiers(getNewID func(v Value) (string, interface{})) {
+type newIDFn func(v Value) (string, interface{})
+type matchesInputFn func(string) bool
+
+func (s *sourceStore) updateIdentifiers(matchesInput matchesInputFn, getNewID newIDFn) {
 	s.store.ephemeralStore.mu.Lock()
 	defer s.store.ephemeralStore.mu.Unlock()
 
@@ -234,7 +237,7 @@ func (s *sourceStore) UpdateIdentifiers(getNewID func(v Value) (string, interfac
 			continue
 		}
 
-		if !s.identifier.MatchesInput(key) {
+		if !matchesInput(key) {
 			continue
 		}
 
@@ -274,6 +277,26 @@ func (s *sourceStore) UpdateIdentifiers(getNewID func(v Value) (string, interfac
 	}
 }
 
+// UpdateIdentifiers copies an existing resource to a new ID and marks the previous one
+// for removal.
+func (s *sourceStore) UpdateIdentifiers(getNewID func(v Value) (string, interface{})) {
+	s.updateIdentifiers(s.identifier.MatchesInput, getNewID)
+}
+
+func (s *sourceStore) CopyStatesFromPreviousIDs(getNewID func(v Value) (string, interface{})) {
+	matchPreviousIdentifier := func(key string) bool {
+		for _, identifier := range s.previousIdentifiers {
+			if identifier.MatchesInput(key) {
+				return true
+			}
+		}
+
+		return false
+	}
+
+	s.updateIdentifiers(matchPreviousIdentifier, getNewID)
+}
+
 func (s *store) Retain() { s.refCount.Retain() }
 func (s *store) Release() {
 	if s.refCount.Release() {

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

@@ -20,6 +20,7 @@ package input_logfile
 
 import (
 	"errors"
+	"strings"
 	"testing"
 	"time"
 
@@ -358,7 +359,10 @@ func TestSourceStore_UpdateIdentifiers(t *testing.T) {
 		})
 		s := testOpenStore(t, "test", backend)
 		defer s.Release()
-		store := &sourceStore{&sourceIdentifier{"test"}, s}
+		store := &sourceStore{
+			identifier: &sourceIdentifier{"test"},
+			store:      s,
+		}
 
 		store.UpdateIdentifiers(func(v Value) (string, interface{}) {
 			var m testMeta
@@ -400,6 +404,68 @@ func TestSourceStore_UpdateIdentifiers(t *testing.T) {
 	})
 }
 
+func TestSourceStoreCopyStatesFromPreviousIDs(t *testing.T) {
+	backend := createSampleStore(t, map[string]state{
+		"filestream::previous-id::key1": { // Active resource
+			TTL:  60 * time.Second,
+			Meta: testMeta{IdentifierName: "test-file-identity"},
+		},
+		"filestream::another-input::key2": { // Active resource from another input
+			TTL:  60 * time.Second,
+			Meta: testMeta{IdentifierName: "test-file-identity"},
+		},
+	})
+	s := testOpenStore(t, "filestream", backend)
+	defer s.Release()
+	store := &sourceStore{
+		identifier:          &sourceIdentifier{"filestream::current-id::"},
+		previousIdentifiers: []*sourceIdentifier{&sourceIdentifier{"filestream::previous-id::"}},
+		store:               s,
+	}
+
+	store.CopyStatesFromPreviousIDs(func(v Value) (string, interface{}) {
+		r, ok := v.(*resource)
+		if !ok {
+			t.Fatalf("expecting v of type '*input_logfile.resource', got '%T' instead", v)
+		}
+
+		var m testMeta
+		err := v.UnpackCursorMeta(&m)
+		if err != nil {
+			t.Fatalf("cannot unpack meta: %v", err)
+		}
+
+		newID := strings.ReplaceAll(r.key, "previous-id", "current-id")
+
+		return newID, m
+	})
+
+	// The persistentStore is a mock that does not consider if a state has
+	// been removed before returning it, thus allowing us to get Updated
+	// timestamp from when the resource was deleted.
+	var deletedState state
+	s.persistentStore.Get("filestream::previous-id::key1", &deletedState)
+
+	want := map[string]state{
+		"filestream::previous-id::key1": { // old resource is deleted, TTL must be zero
+			Updated: deletedState.Updated,
+			TTL:     0 * time.Second,
+			Meta:    map[string]interface{}{"identifiername": "test-file-identity"},
+		},
+		"filestream::another-input::key2": { // Unchanged
+			TTL:  60 * time.Second,
+			Meta: map[string]interface{}{"identifiername": "test-file-identity"},
+		},
+		"filestream::current-id::key1": { // Updated resource
+			Updated: s.Get("filestream::current-id::key1").internalState.Updated,
+			TTL:     60 * time.Second,
+			Meta:    map[string]interface{}{"identifiername": "test-file-identity"},
+		},
+	}
+
+	checkEqualStoreState(t, want, backend.snapshot())
+}
+
 //nolint:dupl // Test code won't be refactored on this commit
 func TestSourceStore_CleanIf(t *testing.T) {
 	t.Run("entries are cleaned when function returns true", func(t *testing.T) {
@@ -413,7 +479,10 @@ func TestSourceStore_CleanIf(t *testing.T) {
 		})
 		s := testOpenStore(t, "test", backend)
 		defer s.Release()
-		store := &sourceStore{&sourceIdentifier{"test"}, s}
+		store := &sourceStore{
+			identifier: &sourceIdentifier{"test"},
+			store:      s,
+		}
 
 		store.CleanIf(func(_ Value) bool {
 			return true
@@ -445,7 +514,10 @@ func TestSourceStore_CleanIf(t *testing.T) {
 		})
 		s := testOpenStore(t, "test", backend)
 		defer s.Release()
-		store := &sourceStore{&sourceIdentifier{"test"}, s}
+		store := &sourceStore{
+			identifier: &sourceIdentifier{"test"},
+			store:      s,
+		}
 
 		store.CleanIf(func(v Value) bool {
 			return false