[9.3](backport #49632) Fix check config and take over interaction (#49785)

When using Filestream's take_over feature with autodiscover, files were being
re-ingested from the beginning instead of continuing from the offset recorded
by the Log input.

Autodiscover validates each rendered configuration by instantiating the input
with a temporary, suffixed ID before starting it. Because take_over ran during
input initialisation, states were migrated to the temporary ID rather than the
real input ID. When the real input started, the Log input states had already
been consumed, so all files appeared new.

The fix moves the take_over migration step from input initialisation to input
start. This ensures that config validation (CheckConfig) never triggers state
migration, and only the input that actually runs performs the takeover.

Additionally, the Log input state is no longer deleted from the registry after
migration. Instead, Filestream checks whether it already holds a state for the
file before migrating, skipping the takeover if a state is found. This makes
the mechanism idempotent and removes reliance on the TTL=-2 heuristic that was
used to detect previously-migrated states.

Last, but not least, a few other issues in the TakeOver implementation
are also fixed:
- Incorrect resource release
- ephemeralStore is now locked throughout the whole TakeOver duration

GenAI-Assisted: Yes
Human-Reviewed: Yes
Tool: Claude-CLI, Model: Claude 4.6 Opus (Thinking)
Tool: Cursor-CLI, Model: GPT-5.3 Codex Extra High

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
(cherry picked from commit 8a648cf)

# Conflicts:
#	filebeat/input/filestream/internal/input-logfile/store.go

---------

Co-authored-by: Tiago Queiroz <tiago.queiroz@elastic.co>

Author: mergify[bot]

.buildkite/filebeat/filebeat-pipeline.yml

@@ -166,7 +166,9 @@ steps:
 
       - label: ":ubuntu: Filebeat: Go Integration Tests"
         command: |
+          source .buildkite/scripts/docker.sh
           cd filebeat
+          SNAPSHOT=true PACKAGES="docker" PLATFORMS="linux/amd64" mage -v package
           mage goIntegTest
         retry:
           automatic:

changelog/fragments/1769099693-filestream-takeover-autodiscover-reingest.yaml

@@ -0,0 +1,7 @@
+kind: bug-fix
+
+summary: Fix Filestream take_over causing file re-ingestion when used with autodiscover.
+
+component: filebeat
+
+issue: https://github.com/elastic/beats/issues/49579

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

@@ -19,6 +19,7 @@ package input_logfile
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/elastic/beats/v7/filebeat/input/filestream/internal/task"
@@ -32,14 +33,15 @@ import (
 type managedInput struct {
 	// id is the input ID, it is defined by setting 'id'
 	// in the input configuration
-	id               string
-	manager          *InputManager
-	ackCH            *updateChan
-	sourceIdentifier *SourceIdentifier
-	prospector       Prospector
-	harvester        Harvester
-	cleanTimeout     time.Duration
-	harvesterLimit   uint64
+	id                     string
+	manager                *InputManager
+	ackCH                  *updateChan
+	sourceIdentifier       *SourceIdentifier
+	previousSrcIdentifiers []*SourceIdentifier
+	prospector             Prospector
+	harvester              Harvester
+	cleanTimeout           time.Duration
+	harvesterLimit         uint64
 }
 
 // Name is required to implement the v2.Input interface
@@ -55,6 +57,10 @@ func (inp *managedInput) Run(
 	ctx input.Context,
 	pipeline beat.PipelineConnector,
 ) (err error) {
+
+	// Notify the manager the input has stopped, currently that is used to
+	// keep track of duplicated IDs
+	defer inp.manager.StopInput(inp.id)
 	ctx.UpdateStatus(status.Starting, "")
 	groupStore := inp.manager.getRetainedStore()
 	defer groupStore.Release()
@@ -86,18 +92,18 @@ func (inp *managedInput) Run(
 
 	prospectorStore := inp.manager.getRetainedStore()
 	defer prospectorStore.Release()
-	sourceStore := newSourceStore(prospectorStore, inp.sourceIdentifier, nil)
+	sourceStore := newSourceStore(prospectorStore, inp.sourceIdentifier, inp.previousSrcIdentifiers)
+
+	if err := inp.prospector.TakeOver(sourceStore, inp.sourceIdentifier.ID); err != nil {
+		return fmt.Errorf("prospector failed to take over states: %w", err)
+	}
 
 	// Mark it as running for now.
 	// Any errors encountered by harvester will change state to Degraded
 	ctx.UpdateStatus(status.Running, "")
 
 	inp.prospector.Run(ctx, sourceStore, hg)
 
-	// Notify the manager the input has stopped, currently that is used to
-	// keep track of duplicated IDs
-	inp.manager.StopInput(inp.id)
-
 	return nil
 }
 

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

@@ -277,14 +277,15 @@ func (cim *InputManager) Create(config *conf.C) (inp v2.Input, retErr error) {
 	}
 
 	return &managedInput{
-		manager:          cim,
-		ackCH:            cim.ackCH,
-		id:               settings.ID,
-		prospector:       prospector,
-		harvester:        harvester,
-		sourceIdentifier: srcIdentifier,
-		cleanTimeout:     settings.CleanInactive,
-		harvesterLimit:   settings.HarvesterLimit,
+		manager:                cim,
+		ackCH:                  cim.ackCH,
+		id:                     settings.ID,
+		prospector:             prospector,
+		harvester:              harvester,
+		sourceIdentifier:       srcIdentifier,
+		previousSrcIdentifiers: previousSrcIdentifiers,
+		cleanTimeout:           settings.CleanInactive,
+		harvesterLimit:         settings.HarvesterLimit,
 	}, nil
 }
 

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

@@ -53,6 +53,10 @@ func (m noopProspector) Init(_, _ StoreUpdater, _ func(Source) string) error {
 	return nil
 }
 
+func (m noopProspector) TakeOver(_ StoreUpdater, _ func(Source) string) error {
+	return nil
+}
+
 func (m noopProspector) Run(_ v2.Context, _ StateMetadataUpdater, _ HarvesterGroup) {}
 
 func (m noopProspector) Test() error {

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

@@ -26,11 +26,14 @@ import (
 // It also updates the statestore with the meta data of the running harvesters.
 type Prospector interface {
 	// Init updates the store before starting the prospector.
-	// It cleans up the store, migrates file identities and takes over
-	// states from log or other filestream inputs.
+	// It cleans up the store and migrates file identities.
 	// It receives two StoreUpdater: one global and another local
 	// to this prospector instance.
 	Init(local, global StoreUpdater, newID func(Source) string) error
+	// TakeOver migrates states from other inputs (Log input or other Filestream
+	// inputs) to this prospector's input. It must be called after Init and
+	// before Run, so that it is not triggered during CheckConfig validation.
+	TakeOver(local 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)

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

@@ -297,16 +297,14 @@ func (s *sourceStore) UpdateIdentifiers(getNewID func(v Value) (string, any)) {
 // Filestream inputs or the Log input. fn should return the new registry ID
 // and new CursorMeta. If fn returns an empty string, the entry is skipped.
 //
-// When fn returns a valid ID, the old resource is removed from both,
-// the in-memory and persistent store. The operations are synchronous.
-//
-// If the resource migrated was from the Log input, `TakeOver` will
-// remove it from the persistent store, however the Log input reigstrar
-// will write it back when Filebeat is shutting down. However,
-// there is a mechanism in place to detect this situation and avoid
-// migrating the same state over and over again.
-// See the comments on this method for more details.
+// When fn returns a valid ID:
+//   - If the old resource was from a Filestream input, it is removed from both
+//     the in-memory and disk store.
+//   - If it was from a Log input, it is left untouched.
 func (s *sourceStore) TakeOver(fn func(Value) (string, any)) {
+	// Lock the ephemeral store so we can migrate the states in one go
+	s.store.ephemeralStore.mu.Lock()
+	defer s.store.ephemeralStore.mu.Unlock()
 	matchPreviousFilestreamIDs := func(key string) bool {
 		for _, identifier := range s.identifiersToTakeOver {
 			if identifier.MatchesInput(key) {
@@ -353,44 +351,6 @@ func (s *sourceStore) TakeOver(fn func(Value) (string, any)) {
 					s.store.log.Errorf("cannot read Log input state: %s", err)
 					return true, nil
 				}
-				// That is a workaround for the problems with the
-				// Log input Registrar (`filebeat/registrar`) and the way it
-				// handles states.
-				// There are two problems:
-				//  - 1. The Log input store/registrar does not have an API for
-				//       removing states
-				//  - 2. When `registrar.Registrar` starts, it copies all states
-				//       belonging to the Log input from the disk store into
-				//       memory and when the Registrar is shutting down, it
-				//       writes all states to the disk. This all happens even
-				//       if no Log input was ever started.
-				// This means that no matter what we do here, the states from
-				// the Log input are always re-written to disk.
-				// See: filebeat/registrar/registrar.go, deferred statement on
-				// `Registrar.Run`.
-				//
-				// However, there is a "reset state" code, that runs
-				// during the Registrar initialisation and sets the
-				// TTL to -2, once the Log input havesting that file starts
-				// the TTL is set to -1 (never expires) or the configured
-				// value.
-				// See: filebeat/registrar/registrar.go (readStatesFrom) and
-				// filebeat/beater/filebeat.go (registrar.Start())
-				//
-				// This means that while the Log input is running and the file
-				// has been active at any moment during the Filebeat's execution
-				// the TTL is never going to be -2 during the shutdown.
-				//
-				// So, if TTL == -2, then in the previous run of Filebeat, there
-				// was no Log input using this state, which likely means, it is
-				// a state that has already been migrated to Filestream.
-				//
-				// The worst case that can happen is that we re-ingest the file
-				// once, which is still better than copying an old state with
-				// an incorrect offset every time Filebeat starts.
-				if st.TTL == -2 {
-					return true, nil
-				}
 				st.key = key
 				fromLogInput[key] = st
 			}
@@ -399,10 +359,6 @@ func (s *sourceStore) TakeOver(fn func(Value) (string, any)) {
 		})
 	}
 
-	// Lock the ephemeral store so we can migrate the states in one go
-	s.store.ephemeralStore.mu.Lock()
-	defer s.store.ephemeralStore.mu.Unlock()
-
 	// Migrate all states from the Filestream input
 	for k := range fromFilestreamInput {
 		res := s.store.ephemeralStore.unsafeFind(k, false)
@@ -418,13 +374,20 @@ func (s *sourceStore) TakeOver(fn func(Value) (string, any)) {
 			continue
 		}
 
+		// cleanup must be called on any "exit point" from this loop iteration.
+		// It is responsible for correctly releasing the locked resource.
+		cleanup := func() {
+			res.Release()
+			res.lock.Unlock()
+		}
+
 		newKey, updatedMeta := fn(res)
 		if len(newKey) > 0 {
 			// If the new key already exists in the store, do nothing.
 			// Unlock the resource and return
 			if res := s.store.ephemeralStore.unsafeFind(newKey, false); res != nil {
 				res.Release()
-				res.lock.Unlock()
+				cleanup()
 				continue
 			}
 
@@ -454,16 +417,24 @@ func (s *sourceStore) TakeOver(fn func(Value) (string, any)) {
 			s.store.log.Infof("migrated entry in registry from '%s' to '%s'. Cursor: %v", k, newKey, r.cursor)
 		}
 
-		res.Release()
-		res.lock.Unlock()
+		cleanup()
 	}
 
 	// Migrate all states from the Log input
 	for k, v := range fromLogInput {
 		newKey, updatedMeta := fn(v)
 		if len(newKey) > 0 {
-			// Find or create a resource. It should always create a new one.
 			res := s.store.ephemeralStore.unsafeFind(newKey, true)
+
+			// If the new key already exists in the store, the file has already
+			// been taken over. Skip it to avoid overwriting a valid Filestream
+			// state with a potentially stale Log input state.
+			if !res.IsNew() {
+				res.Release()
+				s.store.log.Infof("state for '%s' already exists as '%s', skipping takeover from Log input", k, newKey)
+				continue
+			}
+
 			res.cursorMeta = updatedMeta
 			// Convert the offset to the correct type
 			res.cursor = struct {
@@ -478,13 +449,6 @@ func (s *sourceStore) TakeOver(fn func(Value) (string, any)) {
 			// Update in-memory store
 			s.store.ephemeralStore.table[newKey] = res
 
-			// "remove" from the disk store.
-			// It will add a remove entry in the log file for this key, however
-			// the Registrar used by the Log input will write to disk all states
-			// it read when Filebeat was starting, thus "overriding" this delete.
-			// We keep it here because when we remove the Log input we will ensure
-			// the entry is actually remove from the disk store.
-			_ = s.store.persistentStore.Remove(k)
 			res.Release()
 			s.store.log.Infof("migrated entry in registry from '%s' to '%s'. Cursor: %v", k, newKey, res.cursor)
 		}

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

@@ -28,6 +28,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	inpFile "github.com/elastic/beats/v7/filebeat/input/file"
 	input "github.com/elastic/beats/v7/filebeat/input/v2"
 	"github.com/elastic/beats/v7/libbeat/statestore"
 	"github.com/elastic/beats/v7/libbeat/statestore/storetest"
@@ -456,6 +457,106 @@ func TestSourceStoreTakeOver(t *testing.T) {
 	checkEqualStoreState(t, want, backend.snapshot())
 }
 
+func TestSourceStoreTakeOverFromLogInput(t *testing.T) {
+	const (
+		logKey           = "filebeat::logs::native::inode:device"
+		filestreamNewKey = "filestream::input-id::native::inode:device"
+	)
+
+	// fn simulates what takeOverFn does: map the log key to a new Filestream key.
+	takeover := func(v Value) (string, any) {
+		if v.Key() == logKey {
+			return filestreamNewKey, testMeta{IdentifierName: "native"}
+		}
+		return "", nil
+	}
+
+	t.Run("log input state is left untouched", func(t *testing.T) {
+		backend := createSampleStore(t, nil)
+		require.NoError(t, backend.Store.Set(logKey, inpFile.State{
+			Source:         "/path/to/file",
+			Offset:         1234,
+			TTL:            -1,
+			IdentifierName: "native",
+		}), "populating test store")
+
+		s := testOpenStore(t, "filestream", backend)
+		defer s.Release()
+		store := &sourceStore{
+			identifier: &SourceIdentifier{"filestream::input-id::"},
+			store:      s,
+		}
+
+		store.TakeOver(takeover)
+
+		assert.NotNil(t, s.Get(filestreamNewKey), "Log input state must have been migrated to Filestream")
+		assert.NotNil(t, s.Get(logKey), "Log input state must left untouched")
+	})
+
+	t.Run("state with TTL=-2 is migrated", func(t *testing.T) {
+		backend := createSampleStore(t, nil)
+		require.NoError(t, backend.Store.Set(logKey, inpFile.State{
+			Source:         "/path/to/file",
+			Offset:         1234,
+			TTL:            -2, // previously this caused the state to be skipped
+			IdentifierName: "native",
+		}), "populating test store")
+
+		s := testOpenStore(t, "filestream", backend)
+		defer s.Release()
+		store := &sourceStore{
+			identifier: &SourceIdentifier{"filestream::input-id::"},
+			store:      s,
+		}
+
+		store.TakeOver(takeover)
+
+		s.ephemeralStore.mu.Lock()
+		_, migrated := s.ephemeralStore.table[filestreamNewKey]
+		s.ephemeralStore.mu.Unlock()
+
+		assert.True(t, migrated, "state with TTL=-2 must be migrated to Filestream")
+	})
+
+	t.Run("state is skipped when Filestream key already exists", func(t *testing.T) {
+		const existingOffset = int64(9999)
+
+		backend := createSampleStore(t, map[string]state{
+			filestreamNewKey: {
+				TTL:    60 * time.Second,
+				Cursor: map[string]any{"offset": existingOffset},
+			},
+		})
+		require.NoError(t, backend.Store.Set(logKey, inpFile.State{
+			Source:         "/path/to/file",
+			Offset:         1234,
+			TTL:            -1,
+			IdentifierName: "native",
+		}), "populating test store")
+
+		s := testOpenStore(t, "filestream", backend)
+		defer s.Release()
+		store := &sourceStore{
+			identifier: &SourceIdentifier{"filestream::input-id::"},
+			store:      s,
+		}
+
+		store.TakeOver(takeover)
+
+		// The pre-existing Filestream state must be unchanged.
+		s.ephemeralStore.mu.Lock()
+		res, exists := s.ephemeralStore.table[filestreamNewKey]
+		s.ephemeralStore.mu.Unlock()
+
+		require.True(t, exists, "Filestream key must still exist after TakeOver")
+		var cur struct {
+			Offset int64 `json:"offset"`
+		}
+		require.NoError(t, res.UnpackCursor(&cur), "unpacking cursor")
+		assert.Equal(t, existingOffset, cur.Offset, "existing Filestream cursor must not be overwritten by Log input state")
+	})
+}
+
 //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) {