Fix system module with both filesets enabled (#41381)

The system module did not define an ID at the root of the config, that made the V2 input loader only start the first journald input it saw because they both ended up with the same identifier (type, ID and path). This is fixed by defining an ID at the root of the configuration templates.

The journald input now also adds the input_id key to its loggers and a non-fatal error is now logged at debug level.

The system-logs input is now marked as experimental instead of stable.

Fix lint warnings by moving toJournalConfig to input_linux.go

Move TestSystemLogsCanUseLogInput to a file without the
linux build constraint so it can run on all OSes supported by the
system integration.

.buildkite/filebeat/filebeat-pipeline.yml

@@ -111,6 +111,9 @@ steps:
         artifact_paths:
           - "filebeat/build/*.xml"
           - "filebeat/build/*.json"
+          - "filebeat/build/integration-tests/*"
+          - "filebeat/build/integration-tests/Test*/*"
+          - "filebeat/build/integration-tests/Test*/data/**/*"
         plugins:
           - test-collector#v1.10.2:
               files: "filebeat/build/TEST-*.xml"

filebeat/input/journald/config.go

@@ -38,6 +38,9 @@ var includeMatchesWarnOnce sync.Once
 
 // Config stores the options of a journald input.
 type config struct {
+	// ID is the input ID, each instance must have a unique ID
+	ID string `config:"id"`
+
 	// Paths stores the paths to the journal files to be read.
 	Paths []string `config:"paths"`
 

filebeat/input/journald/input.go

@@ -42,6 +42,7 @@ type journalReader interface {
 }
 
 type journald struct {
+	ID                 string
 	Backoff            time.Duration
 	MaxBackoff         time.Duration
 	Since              time.Duration
@@ -108,6 +109,7 @@ func Configure(cfg *conf.C) ([]cursor.Source, cursor.Input, error) {
 	}
 
 	return sources, &journald{
+		ID:                 config.ID,
 		Since:              config.Since,
 		Seek:               config.Seek,
 		Matches:            journalfield.IncludeMatches(config.Matches),
@@ -124,7 +126,7 @@ func (inp *journald) Name() string { return pluginName }
 
 func (inp *journald) Test(src cursor.Source, ctx input.TestContext) error {
 	reader, err := journalctl.New(
-		ctx.Logger,
+		ctx.Logger.With("input_id", inp.ID),
 		ctx.Cancelation,
 		inp.Units,
 		inp.Identifiers,
@@ -149,7 +151,9 @@ func (inp *journald) Run(
 	cursor cursor.Cursor,
 	publisher cursor.Publisher,
 ) error {
-	logger := ctx.Logger.With("path", src.Name())
+	logger := ctx.Logger.
+		With("path", src.Name()).
+		With("input_id", inp.ID)
 	currentCheckpoint := initCheckpoint(logger, cursor)
 
 	mode := inp.Seek

filebeat/input/journald/pkg/journalctl/journalctl.go

@@ -15,13 +15,16 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//go:build linux
+
 package journalctl
 
 import (
 	"bufio"
 	"errors"
 	"fmt"
 	"io"
+	"io/fs"
 	"os/exec"
 	"strings"
 	"sync"
@@ -97,7 +100,31 @@ func Factory(canceller input.Canceler, logger *logp.Logger, binary string, args
 			data, err := reader.ReadBytes('\n')
 			if err != nil {
 				if !errors.Is(err, io.EOF) {
-					logger.Errorf("cannot read from journalctl stdout: '%s'", err)
+					var logError = false
+					var pathError *fs.PathError
+					if errors.As(err, &pathError) {
+						// Because we're reading from the stdout from a process that will
+						// eventually exit, it can happen that when reading we get the
+						// fs.PathError below instead of an io.EOF. This is expected,
+						// it only means the process has exited, its stdout has been
+						// closed and there is nothing else for us to read.
+						// This is expected and does not cause any data loss.
+						// So we log at level debug to have it in our logs if ever needed
+						// while avoiding adding error level logs on user's deployments
+						// for situations that are well handled.
+						if pathError.Op == "read" &&
+							pathError.Path == "|0" &&
+							pathError.Err.Error() == "file already closed" {
+							logger.Debugf("cannot read from journalctl stdout: '%s'", err)
+						} else {
+							logError = true
+						}
+					} else {
+						logError = true
+					}
+					if logError {
+						logger.Errorf("cannot read from journalctl stdout: '%s'", err)
+					}
 				}
 				return
 			}

filebeat/input/journald/pkg/journalctl/mode.go

@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//go:build linux
+
 package journalctl
 
 import "fmt"

filebeat/input/journald/pkg/journalctl/mode_test.go

@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//go:build linux
+
 package journalctl
 
 import (

filebeat/input/journald/pkg/journalctl/reader.go

@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//go:build linux
+
 package journalctl
 
 import (
@@ -113,9 +115,8 @@ type Reader struct {
 	jctl        Jctl
 	jctlFactory JctlFactory
 
-	backoff  backoff.Backoff
-	seekMode SeekMode
-	state    readerState
+	backoff backoff.Backoff
+	state   readerState
 }
 
 // handleSeekAndCursor returns the correct arguments for seek and cursor.
@@ -311,7 +312,7 @@ func (r *Reader) Next(cancel input.Canceler) (JournalEntry, error) {
 			// We recreate the backoff so r.backoff.Last().IsZero()
 			// will return true next time it's called making us to
 			// wait in case jouranlctl crashes in less than 5s.
-			if !r.backoff.Last().IsZero() && time.Now().Sub(r.backoff.Last()) > 5*time.Second {
+			if !r.backoff.Last().IsZero() && time.Since(r.backoff.Last()) > 5*time.Second {
 				r.backoff = backoff.NewExpBackoff(cancel.Done(), 100*time.Millisecond, 2*time.Second)
 			} else {
 				r.backoff.Wait()

filebeat/input/journald/pkg/journalctl/reader_test.go

@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//go:build linux
+
 package journalctl
 
 import (

filebeat/input/journald/pkg/journalfield/conv.go

@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//go:build linux
+
 package journalfield
 
 import (

filebeat/input/journald/pkg/journalfield/matcher.go

@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//go:build linux
+
 package journalfield
 
 import (

filebeat/input/systemlogs/input.go

@@ -93,7 +93,7 @@ func PluginV2(logger *logp.Logger, store cursor.StateStore) v2.Plugin {
 
 	return v2.Plugin{
 		Name:       pluginName,
-		Stability:  feature.Stable,
+		Stability:  feature.Experimental,
 		Deprecated: false,
 		Info:       "system-logs input",
 		Doc:        "The system-logs input collects system logs on Linux by reading them from journald or traditional log files",
@@ -172,47 +172,6 @@ func useJournald(c *conf.C) (bool, error) {
 	return true, nil
 }
 
-func toJournaldConfig(cfg *conf.C) (*conf.C, error) { //nolint:unused // It's used on Linux
-	newCfg, err := cfg.Child("journald", -1)
-	if err != nil {
-		return nil, fmt.Errorf("cannot extract 'journald' block: %w", err)
-	}
-
-	if _, err := cfg.Remove("journald", -1); err != nil {
-		return nil, err
-	}
-
-	if _, err := cfg.Remove("type", -1); err != nil {
-		return nil, err
-	}
-
-	if _, err := cfg.Remove("files", -1); err != nil {
-		return nil, err
-	}
-
-	if _, err := cfg.Remove("use_journald", -1); err != nil {
-		return nil, err
-	}
-
-	if _, err := cfg.Remove("use_files", -1); err != nil {
-		return nil, err
-	}
-
-	if err := newCfg.Merge(cfg); err != nil {
-		return nil, err
-	}
-
-	if err := newCfg.SetString("type", -1, "journald"); err != nil {
-		return nil, fmt.Errorf("cannot set 'type': %w", err)
-	}
-
-	if err := cfg.SetString("type", -1, pluginName); err != nil {
-		return nil, fmt.Errorf("cannot set type back to '%s': %w", pluginName, err)
-	}
-
-	return newCfg, nil
-}
-
 func toFilesConfig(cfg *conf.C) (*conf.C, error) {
 	newCfg, err := cfg.Child("files", -1)
 	if err != nil {

filebeat/input/systemlogs/input_linux.go

@@ -47,3 +47,44 @@ func configure(cfg *conf.C) ([]cursor.Source, cursor.Input, error) {
 
 	return journald.Configure(journaldCfg)
 }
+
+func toJournaldConfig(cfg *conf.C) (*conf.C, error) {
+	newCfg, err := cfg.Child("journald", -1)
+	if err != nil {
+		return nil, fmt.Errorf("cannot extract 'journald' block: %w", err)
+	}
+
+	if _, err := cfg.Remove("journald", -1); err != nil {
+		return nil, err
+	}
+
+	if _, err := cfg.Remove("type", -1); err != nil {
+		return nil, err
+	}
+
+	if _, err := cfg.Remove("files", -1); err != nil {
+		return nil, err
+	}
+
+	if _, err := cfg.Remove("use_journald", -1); err != nil {
+		return nil, err
+	}
+
+	if _, err := cfg.Remove("use_files", -1); err != nil {
+		return nil, err
+	}
+
+	if err := newCfg.Merge(cfg); err != nil {
+		return nil, err
+	}
+
+	if err := newCfg.SetString("type", -1, "journald"); err != nil {
+		return nil, fmt.Errorf("cannot set 'type': %w", err)
+	}
+
+	if err := cfg.SetString("type", -1, pluginName); err != nil {
+		return nil, fmt.Errorf("cannot set type back to '%s': %w", pluginName, err)
+	}
+
+	return newCfg, nil
+}

filebeat/module/system/auth/config/auth.yml

@@ -1,4 +1,6 @@
 type: system-logs
+id: system-auth
+
 {{ if .use_journald }}
 use_journald: true
 {{ end }}

filebeat/module/system/syslog/config/syslog.yml

@@ -1,4 +1,5 @@
 type: system-logs
+id: system-syslog
 
 {{ if .use_journald }}
 use_journald: true