Support both take over configurations

CHANGELOG.next.asciidoc

@@ -438,7 +438,7 @@ otherwise no tag is added. {issue}42208[42208] {pull}42403[42403]
 - Journald `include_matches.match` now accepts `+` to represent a logical disjunction (OR) {issue}40185[40185] {pull}42517[42517]
 - 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 take over now supports taking over states from other Filestream inputs and dynamic loading of inputs (autodiscover and Elastic-Agent). {issue}42472[42472] {issue}42884[42884] {pull}42624[42624]
+- Filestram take over now supports taking over states from other Filestream inputs and dynamic loading of inputs (autodiscover and Elastic-Agent). There is a new syntax for the configuration, but the previous one can still be used. {issue}42472[42472] {issue}42884[42884] {pull}42624[42624]
 - Add `etw` input fallback to attach an already existing session. {pull}42847[42847]
 - Update CEL mito extensions to v1.17.0. {pull}42851[42851]
 - Winlog input now can report its status to Elastic-Agent {pull}43089[43089]

docs/reference/filebeat/filebeat-input-filestream.md

@@ -354,13 +354,16 @@ The `take over` mode can work correctly only if the source (taken from) inputs a
 `take_over.enabled: true` requires the `filestream` to have a unique ID.
 ::::
 
-
 This `take over` mode was created to enable smooth migration from
 deprecated `log` inputs to the new `filestream` inputs and to allow
 changing `filestream` input IDs without data re-ingestion.
 
 See [*Migrate `log` input configurations to `filestream`*](/reference/filebeat/migrate-to-filestream.md) for more details about the migration process.
 
+The previous configuration format `take_over: true`, while
+discouraged, is still supported to migrate state from the `log` input
+to `filestream`.
+
 ::::{warning}
 The `take over` mode is still in beta, however, it should be generally safe to use.
 ::::

filebeat/input/filestream/config.go

@@ -49,8 +49,10 @@ type config struct {
 	IgnoreOlder    time.Duration      `config:"ignore_older"`
 	IgnoreInactive ignoreInactiveType `config:"ignore_inactive"`
 	Rotation       *conf.Namespace    `config:"rotation"`
-	TakeOver       takeOverConfig     `config:"take_over"`
 
+	// TakeOver is parsed independently from the rest of this struct, see
+	// 'GetTakeOverConfig' on internal/input-logfile/manager.go
+	TakeOver takeOverConfig `config:"-"`
 	// AllowIDDuplication is used by InputManager.Create
 	// (see internal/input-logfile/manager.go).
 	AllowIDDuplication bool `config:"allow_deprecated_id_duplication"`

filebeat/input/filestream/config_test.go

@@ -228,7 +228,7 @@
 				require.NoError(t, err, "could not create input configuration")
 				inputs = append(inputs, cfg)
 			}
 			err := logp.DevelopmentSetup(logp.ToObserverOutput())
 			require.NoError(t, err, "could not setup log for development")
 
 			err = ValidateInputIDs(inputs, logp.L())
@@ -239,3 +239,87 @@
 		})
 	}
 }
+
+func TestTakeOverCfg(t *testing.T) {
+	testCases := map[string]struct {
+		cfgYAML     string
+		takeOverCfg takeOverConfig
+		expectErr   bool
+	}{
+		"legacy mode enabled": {
+			cfgYAML: `
+              take_over: true`,
+			takeOverCfg: takeOverConfig{
+				Enabled: true,
+			},
+		},
+		"legacy mode disabled": {
+			cfgYAML: `
+              take_over: false`,
+			takeOverCfg: takeOverConfig{
+				Enabled: false,
+			},
+		},
+		"new mode enabled": {
+			cfgYAML: `
+              take_over:
+                enabled: true`,
+			takeOverCfg: takeOverConfig{
+				Enabled: true,
+			},
+		},
+		"new mode disabled": {
+			cfgYAML: `
+              take_over:
+                enabled: false`,
+			takeOverCfg: takeOverConfig{
+				Enabled: false,
+			},
+		},
+		"new mode with IDs": {
+			cfgYAML: `
+              take_over:
+                enabled: false
+                from_ids: ["foo", "bar"]`,
+			takeOverCfg: takeOverConfig{
+				Enabled: false,
+				FromIDs: []string{"foo", "bar"},
+			},
+		},
+		"take_over not defined": {
+			cfgYAML:   "",
+			expectErr: false,
+		},
+		"invalid config": {
+			cfgYAML:   "take_over.enabled: 42",
+			expectErr: true,
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			// It is required to have 'paths' set, so set it here for all tests
+			cfg := conf.MustNewConfigFrom(tc.cfgYAML)
+			err := cfg.SetChild("paths", -1, conf.MustNewConfigFrom(`["foo"]`))
+			if err != nil {
+				t.Fatalf("cannot set 'paths' in config: %s", err)
+			}
+
+			_, inp, err := configure(cfg, logp.NewNopLogger())
+			if tc.expectErr {
+				require.Error(t, err, "expecting error when parsing config")
+				require.Nil(t, inp, "returned filestream must be nil on error")
+				return
+			} else {
+				require.NoError(t, err, "expecting the config to be successfully parsed")
+			}
+
+			f, ok := inp.(*filestream)
+			if !ok {
+				t.Fatalf("expecting type filestream, got %T", inp)
+			}
+
+			assert.Equal(t, tc.takeOverCfg, f.takeOver, "take over config does not match")
+		})
+	}
+}

filebeat/input/filestream/input.go

@@ -91,6 +91,13 @@
 		return nil, nil, err
 	}
 
+	takeOverEnabled, fromIDs, err := loginp.GetTakeOverConfig(cfg, log)
+	if err != nil {
+		return nil, nil, err
+	}
+	config.TakeOver.Enabled = takeOverEnabled
+	config.TakeOver.FromIDs = fromIDs
+
 	prospector, err := newProspector(config, log)
 	if err != nil {
 		return nil, nil, fmt.Errorf("cannot create prospector: %w", err)
@@ -393,7 +400,7 @@
 			continue
 		}
 
 		metrics.BytesProcessed.Add(uint64(message.Bytes))
 
 		// add "take_over" tag if `take_over` is set to true
 		if inp.takeOver.Enabled {

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

@@ -131,7 +131,7 @@
 	if err != nil {
 		store.Release()
 		cim.shutdown()
 		return fmt.Errorf("Can not start registry cleanup process: %w", err)
 	}
 
 	return nil
@@ -156,9 +156,9 @@
 		HarvesterLimit     uint64        `config:"harvester_limit"`
 		AllowIDDuplication bool          `config:"allow_deprecated_id_duplication"`
 		TakeOver           struct {
-			Enabled bool     `config:"enabled"`
-			FromIDs []string `config:"from_ids"`
-		} `config:"take_over"`
+			Enabled bool
+			FromIDs []string
+		} `config:"-"`
 	}{
 		CleanInactive: cim.DefaultCleanTimeout,
 	}
@@ -167,6 +167,13 @@
 		return nil, err
 	}
 
+	takeOverEnabled, fromIDs, err := GetTakeOverConfig(config, cim.Logger)
+	if err != nil {
+		return nil, err
+	}
+	settings.TakeOver.Enabled = takeOverEnabled
+	settings.TakeOver.FromIDs = fromIDs
+
 	if settings.ID == "" {
 		cim.Logger.Warn("filestream input without ID is discouraged, please add an ID and restart Filebeat")
 	}
@@ -325,3 +332,46 @@
 func (i *sourceIdentifier) MatchesInput(id string) bool {
 	return strings.HasPrefix(id, i.prefix)
 }
+
+// GetTakeOverConfig returns the take over configuration as two independent
+// values. In the YAML they're defined as 'take_over.enabled' and
+// 'take_over.from_ids', respectively.
+// It can handle both formats: the single boolean (`take_over: true|false`) and
+// the object (show above). On error false, nil and the error are returned
+func GetTakeOverConfig(cfg *conf.C, logger *logp.Logger) (bool, []string, error) {
+	// This is never going to return an error because the config path
+	// is a single element. Anyways, we still handle it.
+	hasTakeOver, err := cfg.Has("take_over", -1)
+	if err != nil {
+		return false, nil, fmt.Errorf("cannot assert if 'take_over' is present: %w", err)
+	}
+	if hasTakeOver {
+		legacyEnabled, legacyErr := cfg.Bool("take_over", -1)
+		if legacyErr != nil {
+			// Try again with the new type
+			takeOverCfg, err := cfg.Child("take_over", -1)
+			// if there is an error now, then the config is definitely invalid
+			if err != nil {
+				return false, nil, errors.New("cannot parse the 'take_over' field")
+			}
+
+			// This is copied from filebeat/input/filestream/config.go
+			takeOver := struct {
+				Enabled bool     `config:"enabled"`
+				FromIDs []string `config:"from_ids"`
+			}{}
+			if err := takeOverCfg.Unpack(&takeOver); err != nil {
+				return false, nil, err
+			}
+
+			return takeOver.Enabled, takeOver.FromIDs, nil
+		} else {
+			if legacyEnabled {
+				logger.Warn("using 'take_over: true' is deprecated, use the new format: 'take_over.enabled: true'")
+			}
+			return legacyEnabled, nil, nil
+		}
+	}
+
+	return false, nil, nil
+}

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

@@ -434,3 +434,65 @@ func newBufferLogger() (*logp.Logger, *bytes.Buffer) {
 	}))
 	return log, buf
 }
+
+func TestGetTakeOverConfig(t *testing.T) {
+	testCases := map[string]struct {
+		cfgYAML   string
+		enabled   bool
+		fromIDs   []string
+		expectErr bool
+	}{
+		"legacy mode enabled": {
+			cfgYAML: `
+              take_over: true`,
+			enabled: true,
+		},
+		"legacy mode disabled": {
+			cfgYAML: `
+              take_over: false`,
+			enabled: false,
+		},
+		"new mode enabled": {
+			cfgYAML: `
+              take_over:
+                enabled: true`,
+			enabled: true,
+		},
+		"new mode disabled": {
+			cfgYAML: `
+              take_over:
+                enabled: false`,
+			enabled: false,
+		},
+		"new mode with IDs": {
+			cfgYAML: `
+              take_over:
+                enabled: true
+                from_ids: ["foo", "bar"]`,
+			enabled: true,
+			fromIDs: []string{"foo", "bar"},
+		},
+		"take_over not defined": {
+			cfgYAML:   "",
+			expectErr: false,
+		},
+		"invalid config": {
+			cfgYAML:   "take_over.enabled: 42",
+			expectErr: true,
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			enabled, fromIDs, err := GetTakeOverConfig(config.MustNewConfigFrom(tc.cfgYAML), logp.NewNopLogger())
+			if tc.expectErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+
+			assert.Equal(t, tc.enabled, enabled, "wrong value for enabled")
+			assert.Equal(t, tc.fromIDs, fromIDs, "wrong value for from_ids")
+		})
+	}
+}

filebeat/input/v2/compat/compat_test.go

@@ -110,7 +110,6 @@ paths:
 prospector:
   scanner:
     symlinks: true
-take_over: true
 type: test
 `, inputID)
 

filebeat/tests/integration/filestream_test.go

@@ -589,6 +589,11 @@
 			"expected-registry-happy-path.json"),
 		"Entries in the registry are different from the expectation",
 	)
+
+	deprecationLog := "using 'take_over: true' is deprecated, use the new format: 'take_over.enabled: true'"
+	if filebeat.LogContains(deprecationLog) {
+		t.Fatalf("deprecation log %q must not be present when using the new syntax", deprecationLog)
+	}
 }
 
 func TestFilestreamTakeOverFromLogInput(t *testing.T) {
@@ -655,6 +660,11 @@
 			"expected-registry-happy-path-log-input.json"),
 		"Entries in the registry are different from the expectation",
 	)
+
+	deprecationLog := "using 'take_over: true' is deprecated, use the new format: 'take_over.enabled: true'"
+	if !filebeat.LogContains(deprecationLog) {
+		t.Fatalf("did not find the deprecation log %q", deprecationLog)
+	}
 }
 
 func requireRegistryEntryRemoved(t *testing.T, workDir, identity string) {
@@ -731,7 +741,7 @@
 func waitForEOF(t *testing.T, filebeat *integration.BeatProc, files []string) {
 	for _, path := range files {
 		if runtime.GOOS == "windows" {
 			path = strings.Replace(path, `\`, `\\`, -1)
 		}
 		eofMsg := fmt.Sprintf("End of file reached: %s; Backoff now.", path)
 
@@ -750,7 +760,7 @@
 func waitForDidnotChange(t *testing.T, filebeat *integration.BeatProc, files []string) {
 	for _, path := range files {
 		if runtime.GOOS == "windows" {
 			path = strings.Replace(path, `\`, `\\`, -1)
 		}
 		eofMsg := fmt.Sprintf("File didn't change: %s", path)
 

filebeat/tests/integration/testdata/take-over/happy-path-log-input.yml

@@ -4,8 +4,7 @@ filebeat.inputs:
       - {{.testdata}}/take-over/*.log
 {{ if .takeOver }}
     id: take-over-from-log-input
-    take_over:
-      enabled: true
+    take_over: true
     file_identity.fingerprint: ~
     prospector:
       scanner: