Correctly set 'input.type' when using the system integration (#41246) (…

…#41255)

The system integration now uses the new input system-logs that is just
a proxy to instantiate either the log input or journald. That caused
the 'input.type' in the final event to contain 'system-logs' instead
of 'log' or 'journald'. This PR fixes it buy adding a set step in the
ingest pipeline.

The ingest pipeline is used because Beats sets this filed outside of
the input code.

Co-authored-by: Pierre HILBERT <pierre.hilbert@elastic.co>
(cherry picked from commit e2815a4)

Co-authored-by: Tiago Queiroz <tiago.queiroz@elastic.co>

CHANGELOG-developer.next.asciidoc

@@ -70,6 +70,7 @@ The list below covers the major changes between 7.0.0-rc2 and main only.
 - Debug log entries from the acker (`stateful ack ...` or `stateless ack ...`) removed. {pull}39672[39672]
 - Rename x-pack/filebeat websocket input to streaming. {issue}40264[40264] {pull}40421[40421]
 - Journald input now calls `journalctl` instead of using `github.com/coreos/go-systemd/v22@v22.5.0/sdjournal`, the CGO dependency has been removed from Filebeat {pull}40061[40061]
+- System module events now contain `input.type: systemlogs` instead of `input.type: log` when harvesting log files, however the ingest pipeline sets it back to the original input (log or journald). {pull}41246[41246]
 
 ==== Bugfixes
 

filebeat/input/systemlogs/input.go

@@ -115,16 +115,20 @@ func PluginV2(logger *logp.Logger, store cursor.StateStore) v2.Plugin {
 //     return false
 //   - Otherwise return true
 func useJournald(c *conf.C) (bool, error) {
+	logger := logp.L().Named("input.system-logs")
+
 	cfg := config{}
 	if err := c.Unpack(&cfg); err != nil {
 		return false, nil
 	}
 
 	if cfg.UseJournald {
+		logger.Info("using journald input because 'use_journald' is set")
 		return true, nil
 	}
 
 	if cfg.UseFiles {
+		logger.Info("using log input because 'use_files' is set")
 		return false, nil
 	}
 
@@ -144,6 +148,9 @@ func useJournald(c *conf.C) (bool, error) {
 		if len(paths) != 0 {
 			// We found at least one system log file,
 			// journald will not be used, return early
+			logger.Info(
+				"using log input because file(s) was(were) found when testing glob '%s'",
+				g)
 			return false, nil
 		}
 	}
@@ -230,5 +237,6 @@ func toFilesConfig(cfg *conf.C) (*conf.C, error) {
 	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/ingest/files.yml

@@ -54,6 +54,9 @@ processors:
             value: '{{{ _ingest.on_failure_message }}}'
   - remove:
       field: system.auth.timestamp
+  - set:
+      field: input.type
+      value: log
 on_failure:
   - set:
       field: error.message

filebeat/module/system/auth/ingest/journald.yml

@@ -24,6 +24,9 @@ processors:
         - syslog
         - systemd
         - message_id
+  - set:
+      field: input.type
+      value: journald
 on_failure:
   - set:
       field: error.message