[8.x](backport #41381) Fix system module with both filesets enabled

.buildkite/filebeat/filebeat-pipeline.yml

@@ -105,6 +105,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/**/*"
         notify:
           - github_commit_status:
               context: "filebeat: Go Integration Tests"

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

@@ -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
+}