Fix system module with both filesets enables

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.

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

@@ -22,6 +22,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"io/fs"
 	"os/exec"
 	"strings"
 
@@ -95,7 +96,20 @@ 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) {
+						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/systemlogs/input.go

@@ -92,7 +92,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",

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