[Filebeat] Filestream running as Log input under Elastic Agent or feature flag

.buildkite/filebeat/filebeat-pipeline.yml

@@ -243,6 +243,7 @@ steps:
         command: |
           cd filebeat
           mage pythonIntegTest
+          mage PythonIntegTestLogAsFilestream
         retry:
           automatic:
             - limit: 1

docs/reference/filebeat/exported-fields-log.md

@@ -27,6 +27,22 @@ Contains log file lines.
     required: False
 
 
+**`log.file.device_id`**
+:   The device ID used for the log file, this is used by the 'native' file identity.
+
+    type: keyword
+
+    required: False
+
+
+**`log.file.inode`**
+:   The inode of the log file, this is used by the 'native' file identity.
+
+    type: long
+
+    required: False
+
+
 **`stream`**
 :   Log stream when reading container logs, can be 'stdout' or 'stderr'
 

filebeat/_meta/fields.common.yml

@@ -16,6 +16,20 @@
       description: >
         The file offset the reported line starts at.
 
+    - name: log.file.device_id
+      type: keyword
+      required: false
+      description: >-
+        The device ID used for the log file, this is used by the
+        'native' file identity.
+
+    - name: log.file.inode
+      type: long
+      required: false
+      description: >-
+        The inode of the log file, this is used by the 'native' file
+        identity.
+
     - name: stream
       type: keyword
       required: false

filebeat/input/default-inputs/inputs.go

@@ -20,6 +20,7 @@ package inputs
 import (
 	"github.com/elastic/beats/v7/filebeat/input/filestream"
 	"github.com/elastic/beats/v7/filebeat/input/kafka"
+	"github.com/elastic/beats/v7/filebeat/input/logv2"
 	"github.com/elastic/beats/v7/filebeat/input/net/tcp"
 	"github.com/elastic/beats/v7/filebeat/input/net/udp"
 	"github.com/elastic/beats/v7/filebeat/input/unix"
@@ -44,5 +45,6 @@ func genericInputs(log *logp.Logger, components statestore.States) []v2.Plugin {
 		tcp.Plugin(),
 		udp.Plugin(),
 		unix.Plugin(),
+		logv2.PluginV2(log, components),
 	}
 }

filebeat/input/log/config.go

@@ -158,7 +158,7 @@
 
 	// Input
 	if c.Type == harvester.LogType && len(c.Paths) == 0 {
 		return fmt.Errorf("No paths were defined for input")
 	}
 
 	if c.CleanInactive != 0 && c.IgnoreOlder == 0 {
@@ -172,12 +172,12 @@
 	// Harvester
 	if c.JSON != nil && len(c.JSON.MessageKey) == 0 &&
 		c.Multiline != nil {
 		return fmt.Errorf("When using the JSON decoder and multiline together, you need to specify a message_key value")
 	}
 
 	if c.JSON != nil && len(c.JSON.MessageKey) == 0 &&
 		(len(c.IncludeLines) > 0 || len(c.ExcludeLines) > 0) {
 		return fmt.Errorf("When using the JSON decoder and line filtering together, you need to specify a message_key value")
 	}
 
 	if c.ScanSort != "" {
@@ -196,7 +196,7 @@
 	return nil
 }
 
 // checkUnsupportedParams checks if unsupported/deprecated/discouraged paramaters are set and logs a warning
 func (c config) checkUnsupportedParams(logger *logp.Logger) {
 	if c.ScanSort != "" {
 		logger.Warn(cfgwarn.Experimental("scan_sort is used."))
@@ -232,10 +232,22 @@
 	for _, path := range c.Paths {
 		pathAbs, err := filepath.Abs(path)
 		if err != nil {
 			return fmt.Errorf("Failed to get the absolute path for %s: %v", path, err)
 		}
 		paths = append(paths, pathAbs)
 	}
 	c.Paths = paths
 	return nil
 }
+
+// IsConfigValid is meant to be used by logv2 to validate whether cfg is
+// a valid Log input configuration.
+// It avoids exporting [config] and [defaultConfig]
+func IsConfigValid(cfg *conf.C) error {
+	c := defaultConfig()
+	if err := cfg.Unpack(&c); err != nil {
+		return fmt.Errorf("cannot unpack config: %w", err)
+	}
+
+	return c.Validate()
+}

filebeat/input/log/input.go

@@ -44,6 +44,11 @@ import (
 	"github.com/elastic/elastic-agent-libs/monitoring"
 )
 
+// The Log input does not define an init function, nor registers itself
+// because the 'logv2' input already does it using 'log' as the input Name.
+// 'logv2' is a "proxy input" that can run either the Log input of Filestream
+// input depending on configuration and feature flag.
+
 const (
 	recursiveGlobDepth      = 8
 	harvesterErrMsg         = "Harvester could not be started on new file: %s, Err: %s"
@@ -60,13 +65,6 @@ var (
 	errDeprecated = errors.New("Log input is deprecated. Use Filestream input instead. Follow our migration guide https://www.elastic.co/guide/en/beats/filebeat/current/migrate-to-filestream.html")
 )
 
-func init() {
-	err := input.Register("log", NewInput)
-	if err != nil {
-		panic(err)
-	}
-}
-
 // Input contains the input and its config
 type Input struct {
 	cfg                 *conf.C

filebeat/input/logv2/input.go

@@ -0,0 +1,182 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package logv2
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/elastic/beats/v7/filebeat/channel"
+	v1 "github.com/elastic/beats/v7/filebeat/input"
+	"github.com/elastic/beats/v7/filebeat/input/filestream"
+	loginput "github.com/elastic/beats/v7/filebeat/input/log"
+	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
+	"github.com/elastic/beats/v7/libbeat/feature"
+	"github.com/elastic/beats/v7/libbeat/features"
+	"github.com/elastic/beats/v7/libbeat/management"
+	"github.com/elastic/beats/v7/libbeat/statestore"
+	"github.com/elastic/elastic-agent-libs/config"
+	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/go-concert/unison"
+)
+
+const pluginName = "log"
+
+func init() {
+	// Register an input V1, to replace the Log input one.
+	if err := v1.Register(pluginName, newV1Input); err != nil {
+		panic(err)
+	}
+}
+
+// runAsFilestream validates cfg as a Log input configuration, if it is
+// valid, then checks whether the configuration should be run as
+// Filestream input. On any error the boolean value must be ignore and
+// no input started. runAsFilestream also sets the input type accordingly.
+func runAsFilestream(cfg *config.C) (bool, error) {
+	// First of all, ensure the Log input configuration is valid.
+	// This ensures we return configuration errors compatible
+	// with the log input.
+	if err := loginput.IsConfigValid(cfg); err != nil {
+		return false, err
+	}
+
+	// Global feature flag that forces all Log input instances
+	// to run as Filestream.
+	if features.LogInputRunFilestream() {
+		return true, nil
+	}
+
+	// Only allow to run the Log input as Filestream if Filebeat
+	// is running under Elastic Agent.
+	if !management.UnderAgent() {
+		return false, nil
+	}
+
+	if ok := cfg.HasField("run_as_filestream"); ok {
+		runAsFilestream, err := cfg.Bool("run_as_filestream", -1)
+		if err != nil {
+			return false, fmt.Errorf("cannot parse 'run_as_filestream': %w", err)
+		}
+
+		if runAsFilestream {
+			// ID is required to run as Filestream input
+			if !cfg.HasField("id") {
+				return false, errors.New("'id' is required to run 'log' input as 'filestream'")
+			}
+
+			// This should never fail because the Log input configuration
+			// already reads 'type' and validates it is a string. Overriding
+			// the field should never fail.
+			if err := cfg.SetString("type", -1, "filestream"); err != nil {
+				return false, fmt.Errorf("cannot set 'type': %w", err)
+			}
+
+			return true, nil
+		}
+	}
+
+	return false, nil
+}
+
+// newV1Input instantiates the Log input. If 'run_as_filestream' is
+// true, then v2.ErrUnknownInput is returned so the Filestream input
+// can be instantiated.
+func newV1Input(
+	cfg *config.C,
+	outlet channel.Connector,
+	context v1.Context,
+	logger *logp.Logger,
+) (v1.Input, error) {
+	// Inputs V1 should be tried last, so if this function is run we are
+	// supposed to be running as the Log input. However do not rely on the
+	// factory implementation, also check whether to run as Log or Filestream
+	// inputs.
+	asFilestream, err := runAsFilestream(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	if asFilestream {
+		return nil, v2.ErrUnknownInput
+	}
+
+	inp, err := loginput.NewInput(cfg, outlet, context, logger)
+	if err != nil {
+		return nil, fmt.Errorf("cannot create log input: %w", err)
+	}
+
+	logger.Debug("Log input running as Log input")
+	return inp, err
+}
+
+// PluginV2 returns a v2.Plugin with a manager that checks whether
+// the config is from a Log input that should run as Filestream.
+// If that is the case the Log input configuration is converted to
+// Filestream and the Filestream input returned.
+// Otherwise v2.ErrUnknownInput is returned.
+func PluginV2(logger *logp.Logger, store statestore.States) v2.Plugin {
+	// The InputManager for Filestream input is from an internal package, so we
+	// cannot instantiate it directly here. To circumvent that, we instantiate
+	// the whole Filestream Plugin
+	filestreamPlugin := filestream.Plugin(logger, store)
+
+	m := manager{
+		next:   filestreamPlugin.Manager,
+		logger: logger,
+	}
+
+	p := v2.Plugin{
+		Name:      pluginName,
+		Stability: feature.Stable,
+		Info:      "log input running filestream",
+		Doc:       "Log input running Filestream input",
+		Manager:   m,
+	}
+	return p
+}
+
+type manager struct {
+	next   v2.InputManager
+	logger *logp.Logger
+}
+
+func (m manager) Init(grp unison.Group) error {
+	return m.next.Init(grp)
+}
+
+func (m manager) Create(cfg *config.C) (v2.Input, error) {
+	// When inputs are created, inputs V2 are tried first, so if we
+	// are supposed to run as the Log input, return v2.ErrUnknownInput
+	asFilestream, err := runAsFilestream(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	if !asFilestream {
+		return nil, v2.ErrUnknownInput
+	}
+
+	newCfg, err := convertConfig(m.logger, cfg)
+	if err != nil {
+		return nil, fmt.Errorf("cannot translate log config to filestream: %w", err)
+	}
+
+	m.logger.Debug("Log input running as Filestream input")
+	return m.next.Create(newCfg)
+}

filebeat/input/logv2/input_test.go

@@ -0,0 +1,90 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package logv2
+
+import (
+	"testing"
+
+	"github.com/elastic/beats/v7/libbeat/management"
+	"github.com/elastic/elastic-agent-libs/config"
+)
+
+func TestRunAsFilestream(t *testing.T) {
+	testCases := map[string]struct {
+		cfg        *config.C
+		expectErr  bool
+		expected   bool
+		underAgent bool
+	}{
+		"simplest log input config": {
+			underAgent: true,
+			expected:   false,
+			cfg: config.MustNewConfigFrom(map[string]any{
+				"paths": []string{"/var/log.log"},
+			}),
+		},
+		"log input invalid config": {
+			// empty config is always invalid
+			cfg:       config.NewConfig(),
+			expectErr: true,
+		},
+		"invalid 'run_as_filestream'": {
+			underAgent: true,
+			cfg: config.MustNewConfigFrom(map[string]any{
+				"paths":             []string{"/var/log.log"},
+				"run_as_filestream": 42,
+			}),
+			expectErr: true,
+		},
+		"no filestream id": {
+			underAgent: true,
+			cfg: config.MustNewConfigFrom(map[string]any{
+				"paths":             []string{"/var/log.log"},
+				"run_as_filestream": true,
+			}),
+			expectErr: true,
+		},
+		"not under Elastic Agent": {
+			underAgent: false,
+			cfg: config.MustNewConfigFrom(map[string]any{
+				"paths": []string{"/var/log.log"},
+			}),
+			expectErr: false,
+			expected:  false,
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			underAgent := management.UnderAgent()
+			t.Cleanup(func() {
+				management.SetUnderAgent(underAgent)
+			})
+			management.SetUnderAgent(tc.underAgent)
+
+			got, err := runAsFilestream(tc.cfg)
+			if err != nil && !tc.expectErr {
+				t.Errorf("did not expect an error: %s", err)
+			}
+
+			if got != tc.expected {
+				t.Errorf("expecting 'runAsFilestream' to return %t, got %t", tc.expected, got)
+			}
+		})
+	}
+}