Use journald for system module on Debian 12 (#41061)

This commit adds Debian 12 support to our system module, to support
Debian 12 we need to use the journald input to collect the system
logs.

To support it, a new, internal, input  `system-logs`is introduced, it is responsible
for deciding whether the log input or journald must be used. If `var.paths` is defined
in the module configuration, `system-logs` looks at the files, if any of the globs resolves
to one or more files the `log` input is used, otherwise the `jouranld` input is used.

This behaviour can be overridden by setting `var.use_journald` or `var.use_files`,
which will force the use of journald or files.

Other changes:
 - Journald input now support filtering by facilities
 - System tests for modules now support handling journal files
 - The `TESTING_FILEBEAT_FILEPATTERN` environment variable now is a
 comma separated list of globs, it defaults to `.log,*.journal`
 - Multiple lint warnings are fixed
 - The documentation has been updated where needed.

(cherry picked from commit cfd1f1c)

CHANGELOG.next.asciidoc

@@ -45,6 +45,8 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Redis: Added replication role as a field to submitted slowlogs
 - Added `container.image.name` to `journald` Filebeat input's Docker-specific translated fields. {pull}40450[40450]
 - Change log.file.path field in awscloudwatch input to nested object. {pull}41099[41099]
+- Remove deprecated awscloudwatch field from Filebeat. {pull}41089[41089]
+- System module events now contain `input.type: systemlogs` instead of `input.type: log` when harvesting log files. {pull}41061[41061]
 
 
 *Heartbeat*
@@ -325,6 +327,8 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Add CSV decoding capacity to azureblobstorage input {pull}40978[40978]
 - Add CSV decoding capacity to gcs input {pull}40979[40979]
 - Add CSV decoding capacity to azureblobstorage input {pull}40978[40978]
+- Jounrald input now supports filtering by facilities {pull}41061[41061]
+- System module now supports reading from jounrald. {pull}41061[41061]
 
 *Auditbeat*
 

filebeat/docs/include/use-journald.asciidoc

@@ -0,0 +1,12 @@
+*`var.use_journald`*::
+
+A boolean that when set to `true` will read logs from Journald. When
+Journald is used all events contain the tag `journald`
+
+*`var.use_files`*::
+
+A boolean that when set to `true` will read logs from the log files
+defined by `vars.paths`.
+
+If neither `var.use_journald` nor `var.use_files` are set (or both are
+`false`) {beatname_uc} will auto-detect the source for the logs.

filebeat/docs/include/var-paths.asciidoc

@@ -6,4 +6,4 @@ are also supported here. For example, you can use wildcards to fetch all files
 from a predefined level of subdirectories: `/path/to/log/*/*.log`. This
 fetches all `.log` files from the subfolders of `/path/to/log`. It does not
 fetch log files from the `/path/to/log` folder itself. If this setting is left
-empty, {beatname_uc} will choose log paths based on your operating system.
+empty, {beatname_uc} will choose log paths based on your operating system.

filebeat/docs/inputs/input-journald.asciidoc

@@ -169,6 +169,13 @@ Valid transports:
 * stdout: messages from a service's standard output or error output
 * kernel: messages from the kernel
 
+[float]
+[id="{beatname_lc}-input-{type}-facilities"]
+==== `facilities`
+
+Filter entries by facilities, facilities must be specified using their
+numeric code.
+
 [float]
 [id="{beatname_lc}-input-{type}-include-matches"]
 ==== `include_matches`

filebeat/docs/modules/system.asciidoc

@@ -23,7 +23,7 @@ include::../include/gs-link.asciidoc[]
 === Compatibility
 
 This module was tested with logs from OSes like Ubuntu 12.04, Centos 7, and
-macOS Sierra.
+macOS Sierra. For Debian 12 Journald is used to read the system logs.
 
 This module is not available for Windows.
 
@@ -65,11 +65,15 @@ include::../include/config-option-intro.asciidoc[]
 
 include::../include/var-paths.asciidoc[]
 
+include::../include/use-journald.asciidoc[]
+
 [float]
 ==== `auth` fileset settings
 
 include::../include/var-paths.asciidoc[]
 
+include::../include/use-journald.asciidoc[]
+
 *`var.tags`*::
 
 A list of tags to include in events. Including `forwarded` indicates that the

filebeat/filebeat.reference.yml

@@ -21,7 +21,18 @@ filebeat.modules:
     # Filebeat will choose the paths depending on your OS.
     #var.paths:
 
-    # Input configuration (advanced). Any input configuration option
+    # Force using journald to collect system logs
+    #var.use_journald: true|false
+
+    # Force using log files to collect system logs
+    #var.use_files: true|false
+
+    # If use_journald and use_files are false, then
+    # Filebeat will autodetect whether use to journald
+    # to collect system logs.
+
+    # Input configuration (advanced).
+    # Any input configuration option
     # can be added under this section.
     #input:
 
@@ -33,6 +44,23 @@ filebeat.modules:
     # Filebeat will choose the paths depending on your OS.
     #var.paths:
 
+    # Force using journald to collect system logs
+    #var.use_journald: true|false
+
+    # Force using log files to collect system logs
+    #var.use_files: true|false
+
+    # If use_journald and use_files are false, then
+    # Filebeat will autodetect whether use to journald
+    # to collect system logs.
+
+    # A list of tags to include in events. Including 'forwarded'
+    # indicates that the events did not originate on this host and
+    # causes host.name to not be added to events. Include
+    # 'preserve_orginal_event' causes the pipeline to retain the raw log
+    # in event.original. Defaults to [].
+    #var.tags: []
+
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:

filebeat/fileset/fileset.go

@@ -24,7 +24,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io/ioutil"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -143,11 +142,11 @@ type ProcessorRequirement struct {
 func (fs *Fileset) readManifest() (*manifest, error) {
 	cfg, err := common.LoadFile(filepath.Join(fs.modulePath, fs.name, "manifest.yml"))
 	if err != nil {
-		return nil, fmt.Errorf("Error reading manifest file: %v", err)
+		return nil, fmt.Errorf("Error reading manifest file: %w", err)
 	}
 	manifest, err := newManifest(cfg)
 	if err != nil {
-		return nil, fmt.Errorf("Error unpacking manifest: %v", err)
+		return nil, fmt.Errorf("Error unpacking manifest: %w", err)
 	}
 	return manifest, nil
 }
@@ -183,7 +182,7 @@ func (fs *Fileset) evaluateVars(info beat.Info) (map[string]interface{}, error)
 
 		vars[name], err = resolveVariable(vars, value)
 		if err != nil {
-			return nil, fmt.Errorf("Error resolving variables on %s: %v", name, err)
+			return nil, fmt.Errorf("Error resolving variables on %s: %w", name, err)
 		}
 	}
 
@@ -246,7 +245,7 @@ func resolveVariable(vars map[string]interface{}, value interface{}) (interface{
 			if ok {
 				transf, err := ApplyTemplate(vars, s, false)
 				if err != nil {
-					return nil, fmt.Errorf("array: %v", err)
+					return nil, fmt.Errorf("array: %w", err)
 				}
 				transformed = append(transformed, transf)
 			} else {
@@ -322,33 +321,35 @@ func getTemplateFunctions(vars map[string]interface{}) (template.FuncMap, error)
 // getBuiltinVars computes the supported built in variables and groups them
 // in a dictionary
 func (fs *Fileset) getBuiltinVars(info beat.Info) (map[string]interface{}, error) {
-	host, err := os.Hostname()
-	if err != nil || len(host) == 0 {
+	osHost, err := os.Hostname()
+	if err != nil || len(osHost) == 0 {
 		return nil, fmt.Errorf("Error getting the hostname: %w", err)
 	}
-	split := strings.SplitN(host, ".", 2)
+	split := strings.SplitN(osHost, ".", 2)
 	hostname := split[0]
 	domain := ""
 	if len(split) > 1 {
 		domain = split[1]
 	}
 
-	return map[string]interface{}{
+	vars := map[string]interface{}{
 		"prefix":      info.IndexPrefix,
 		"hostname":    hostname,
 		"domain":      domain,
 		"module":      fs.mname,
 		"fileset":     fs.name,
 		"beatVersion": info.Version,
-	}, nil
+	}
+
+	return vars, nil
 }
 
 func (fs *Fileset) getInputConfig() (*conf.C, error) {
 	path, err := ApplyTemplate(fs.vars, fs.manifest.Input, false)
 	if err != nil {
 		return nil, fmt.Errorf("Error expanding vars on the input path: %w", err)
 	}
-	contents, err := ioutil.ReadFile(filepath.Join(fs.modulePath, fs.name, path))
+	contents, err := os.ReadFile(filepath.Join(fs.modulePath, fs.name, path))
 	if err != nil {
 		return nil, fmt.Errorf("Error reading input file %s: %w", path, err)
 	}
@@ -434,7 +435,7 @@ func (fs *Fileset) GetPipelines(esVersion version.V) (pipelines []pipeline, err
 			return nil, fmt.Errorf("Error expanding vars on the ingest pipeline path: %w", err)
 		}
 
-		strContents, err := ioutil.ReadFile(filepath.Join(fs.modulePath, fs.name, path))
+		strContents, err := os.ReadFile(filepath.Join(fs.modulePath, fs.name, path))
 		if err != nil {
 			return nil, fmt.Errorf("Error reading pipeline file %s: %w", path, err)
 		}
@@ -458,7 +459,11 @@ func (fs *Fileset) GetPipelines(esVersion version.V) (pipelines []pipeline, err
 			if err != nil {
 				return nil, fmt.Errorf("Failed to sanitize the YAML pipeline file: %s: %w", path, err)
 			}
-			content = newContent.(map[string]interface{})
+			var ok bool
+			content, ok = newContent.(map[string]interface{})
+			if !ok {
+				return nil, errors.New("cannot convert newContent to map[string]interface{}")
+			}
 		default:
 			return nil, fmt.Errorf("Unsupported extension '%s' for pipeline file: %s", extension, path)
 		}

filebeat/input/default-inputs/inputs_linux.go

@@ -19,6 +19,7 @@ package inputs
 
 import (
 	"github.com/elastic/beats/v7/filebeat/input/journald"
+	"github.com/elastic/beats/v7/filebeat/input/systemlogs"
 	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
 	cursor "github.com/elastic/beats/v7/filebeat/input/v2/input-cursor"
 	"github.com/elastic/beats/v7/libbeat/beat"
@@ -37,6 +38,7 @@ func osInputs(info beat.Info, log *logp.Logger, components osComponents) []v2.Pl
 	zeroPlugin := v2.Plugin{}
 	if journald := journald.Plugin(log, components); journald != zeroPlugin {
 		plugins = append(plugins, journald)
+		plugins = append(plugins, systemlogs.PluginV2(log, components))
 	}
 
 	return plugins

filebeat/input/journald/README.md

@@ -0,0 +1,57 @@
+# Journald input
+
+The Journald input reads journal entries by calling `journalctl`.
+
+## Adding entries to the journal
+The easiest way to add entries to the journal is to use `systemd-cat`:
+```
+root@vagrant-debian-12:~/filebeat# echo "Hello Journal!" | systemd-cat
+root@vagrant-debian-12:~/filebeat# journalctl -n 1
+Oct 02 04:17:01 vagrant-debian-12 CRON[1912]: pam_unix(cron:session): session closed for user root
+```
+
+The syslog identifier can be specified with the `-t` parameter:
+```
+root@vagrant-debian-12:~/filebeat# echo "Hello Journal!" | systemd-cat -t my-test
+root@vagrant-debian-12:~/filebeat# journalctl -n 1
+Oct 02 04:17:50 vagrant-debian-12 my-test[1924]: Hello Journal!
+```
+
+## Crafting a journal file
+The easiest way to craft a journal file with the entries you need is
+to use
+[`systemd-journald-remote`](https://www.freedesktop.org/software/systemd/man/latest/systemd-journal-remote.service.html).
+First we need to export some entries to a file:
+```
+root@vagrant-debian-12:~/filebeat# journalctl -g "Hello" -o export >export
+```
+One good thing of the `-o export` is that you can just concatenate the
+output of any number of runs and the result will be a valid file.
+
+Then you can use `systemd-journald-remote` to generate the journal
+file:
+```
+root@vagrant-debian-12:~/filebeat# /usr/lib/systemd/systemd-journal-remote -o example.journal export
+Finishing after writing 2 entries
+``
+Or you can run as a one liner:
+```
+root@vagrant-debian-12:~/filebeat# journalctl -g "Hello" -o export | /usr/lib/systemd/systemd-journal-remote -o example.journal -
+```
+
+Then you can read the newly created file:
+```
+root@vagrant-debian-12:~/filebeat# journalctl --file ./example.journal
+Oct 02 04:16:54 vagrant-debian-12 unknown[1908]: Hello Journal!
+Oct 02 04:17:50 vagrant-debian-12 my-test[1924]: Hello Journal!
+root@vagrant-debian-12:~/filebeat# 
+```
+
+Bear in mind that `systemd-journal-remote` will **append** to the
+output file.
+
+## References
+- https://systemd.io/JOURNAL_NATIVE_PROTOCOL/
+- https://www.freedesktop.org/software/systemd/man/latest/journalctl.html
+- https://www.freedesktop.org/software/systemd/man/latest/systemd-cat.html
+- https://www.freedesktop.org/software/systemd/man/latest/systemd-journal-remote.service.html

filebeat/input/journald/config.go

@@ -63,6 +63,9 @@ type config struct {
 	// SaveRemoteHostname defines if the original source of the entry needs to be saved.
 	SaveRemoteHostname bool `config:"save_remote_hostname"`
 
+	// Facility is a list of facilities to filter journal messages
+	Facilities []int `config:"facilities"`
+
 	// Parsers configuration
 	Parsers parser.Config `config:",inline"`
 }

filebeat/input/journald/input.go

@@ -50,6 +50,7 @@ type journald struct {
 	Units              []string
 	Transports         []string
 	Identifiers        []string
+	Facilities         []int
 	SaveRemoteHostname bool
 	Parsers            parser.Config
 	Journalctl         bool
@@ -79,7 +80,7 @@ func Plugin(log *logp.Logger, store cursor.StateStore) input.Plugin {
 			Logger:     log,
 			StateStore: store,
 			Type:       pluginName,
-			Configure:  configure,
+			Configure:  Configure,
 		},
 	}
 }
@@ -90,7 +91,7 @@ var cursorVersion = 1
 
 func (p pathSource) Name() string { return string(p) }
 
-func configure(cfg *conf.C) ([]cursor.Source, cursor.Input, error) {
+func Configure(cfg *conf.C) ([]cursor.Source, cursor.Input, error) {
 	config := defaultConfig()
 	if err := cfg.Unpack(&config); err != nil {
 		return nil, nil, err
@@ -113,6 +114,7 @@ func configure(cfg *conf.C) ([]cursor.Source, cursor.Input, error) {
 		Units:              config.Units,
 		Transports:         config.Transports,
 		Identifiers:        config.Identifiers,
+		Facilities:         config.Facilities,
 		SaveRemoteHostname: config.SaveRemoteHostname,
 		Parsers:            config.Parsers,
 	}, nil
@@ -128,6 +130,7 @@ func (inp *journald) Test(src cursor.Source, ctx input.TestContext) error {
 		inp.Identifiers,
 		inp.Transports,
 		inp.Matches,
+		inp.Facilities,
 		journalctl.SeekHead,
 		"",
 		inp.Since,
@@ -158,6 +161,7 @@ func (inp *journald) Run(
 		inp.Identifiers,
 		inp.Transports,
 		inp.Matches,
+		inp.Facilities,
 		mode,
 		pos,
 		inp.Since,

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

@@ -137,6 +137,7 @@ func New(
 	syslogIdentifiers []string,
 	transports []string,
 	matchers journalfield.IncludeMatches,
+	facilities []int,
 	mode SeekMode,
 	cursor string,
 	since time.Duration,
@@ -166,6 +167,10 @@ func New(
 		args = append(args, fmt.Sprintf("_TRANSPORT=%s", m))
 	}
 
+	for _, facility := range facilities {
+		args = append(args, "--facility", fmt.Sprintf("%d", facility))
+	}
+
 	otherArgs := handleSeekAndCursor(mode, since, cursor)
 
 	jctl, err := newJctl(canceler, logger.Named("journalctl-runner"), "journalctl", append(args, otherArgs...)...)

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

@@ -97,7 +97,7 @@ func TestRestartsJournalctlOnError(t *testing.T) {
 		return &mock, nil
 	}
 
-	reader, err := New(logp.L(), ctx, nil, nil, nil, journalfield.IncludeMatches{}, SeekHead, "", 0, "", factory)
+	reader, err := New(logp.L(), ctx, nil, nil, nil, journalfield.IncludeMatches{}, []int{}, SeekHead, "", 0, "", factory)
 	if err != nil {
 		t.Fatalf("cannot instantiate journalctl reader: %s", err)
 	}