Add ID to journalbeat input (#18467) (#18590)

Add an `id` configuration option to the Journalbeat input to make it possible
to start multiple inputs on the same journal.

As an administrator this provides more flexibility in how you can deploy
Journalbeat. It gives you the configuration grainularity you need to be able
to have an input for each service or log type. You can apply different
`include_matches` to each input. And you can easily add new log types at any
point and make Journalbeat read all matching logs from the "head" without
affecting existing input cursor positions.

I also fixed a test that was modifying a file under source control by copying it
into the build directory before applying changes.

(cherry picked from commit 74a5bd2)

CHANGELOG.next.asciidoc

@@ -423,6 +423,9 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 *Journalbeat*
 
+- Added an `id` config option to inputs to allow running multiple inputs on the
+  same journal. {pull}18467{18467}
+
 *Metricbeat*
 
 - Move the windows pdh implementation from perfmon to a shared location in order for future modules/metricsets to make use of. {pull}15503[15503]

Vagrantfile

@@ -135,7 +135,7 @@ def linuxDebianProvision()
 #!/usr/bin/env bash
 set -eio pipefail
 apt-get update
-apt-get install -y make gcc python3 python3-pip python3-venv git
+apt-get install -y make gcc python3 python3-pip python3-venv git libsystemd-dev
 SCRIPT
 end
 
@@ -229,6 +229,11 @@ Vagrant.configure(2) do |config|
     c.vm.box = "ubuntu/bionic64"
     c.vm.network :forwarded_port, guest: 22, host: 2228, id: "ssh", auto_correct: true
 
+    c.vm.provider :virtualbox do |vbox|
+      vbox.memory = 4096
+      vbox.cpus = 4
+    end
+
     c.vm.provision "shell", inline: $unixProvision, privileged: false
     c.vm.provision "shell", inline: linuxGvmProvision, privileged: false
     c.vm.provision "shell", inline: linuxDebianProvision

journalbeat/_meta/config/beat.reference.yml.tmpl

@@ -18,6 +18,11 @@ journalbeat.inputs:
   # When empty starts to read from local journal.
 - paths: []
 
+  # An optional unique identifier for the input. By providing a unique `id` you
+  # can operate multiple inputs on the same journal. This allows each input's
+  # cursor to be persisted independently in the registry file.
+  #id: ""
+
   # The number of seconds to wait before trying to read again from journals.
   #backoff: 1s
   # The maximum number of seconds to wait before attempting to read again from journals.

journalbeat/_meta/config/beat.yml.tmpl

@@ -18,6 +18,11 @@ journalbeat.inputs:
   # When empty starts to read from local journal.
 - paths: []
 
+  # An optional unique identifier for the input. By providing a unique `id` you
+  # can operate multiple inputs on the same journal. This allows each input's
+  # cursor to be persisted independently in the registry file.
+  #id: ""
+
   # The number of seconds to wait before trying to read again from journals.
   #backoff: 1s
   # The maximum number of seconds to wait before attempting to read again from journals.

journalbeat/docs/config-options.asciidoc

@@ -75,6 +75,27 @@ in a Docker container. However, in this example the fields are matched using the
 You can specify the following options to configure how {beatname_uc} reads the
 journal files.
 
+[float]
+[id="{beatname_lc}-id"]
+==== `id`
+
+An optional unique identifier for the input. By providing a unique `id` you can
+operate multiple inputs on the same journal. This allows each input's cursor
+to be persisted independently in the registry file.
+
+----
+{beatname_lc}.inputs:
+- id: consul.service
+  paths: []
+  include_matches:
+    - _SYSTEMD_UNIT=consul.service
+
+- id: vault.service
+  paths: []
+  include_matches:
+    - _SYSTEMD_UNIT=vault.service
+----
+
 [float]
 [id="{beatname_lc}-paths"]
 ==== `paths`
@@ -108,10 +129,10 @@ The position to start reading the journal from. Valid settings are:
 
 * `head`: Starts reading at the beginning of the journal. After a restart,
 {beatname_uc} resends all log messages in the journal.
-* `tail`: Starts reading at the end of the journal. After a restart, 
+* `tail`: Starts reading at the end of the journal. After a restart,
 {beatname_uc} resends the last message, which might result in duplicates. If
 multiple log messages are written to a journal while {beatname_uc} is down,
-only the last log message is sent on restart. 
+only the last log message is sent on restart.
 * `cursor`: On first read, starts reading at the beginning of the journal. After a
 reload or restart, continues reading at the last known position.
 
@@ -207,7 +228,7 @@ journald fields:
 
 The following translated fields for
 https://docs.docker.com/config/containers/logging/journald/[Docker] are also
-available: 
+available:
 
 [horizontal]
 `CONTAINER_ID`::              `container.id_truncated`

journalbeat/input/config.go

@@ -28,6 +28,8 @@ import (
 
 // Config stores the options of an input.
 type Config struct {
+	// Unique ID of the input for state persistence purposes.
+	ID string `config:"id"`
 	// Paths stores the paths to the journal files to be read.
 	Paths []string `config:"paths"`
 	// Backoff is the current interval to wait before

journalbeat/input/input.go

@@ -25,8 +25,6 @@ import (
 
 	"github.com/elastic/beats/v7/libbeat/common/fmtstr"
 
-	"github.com/gofrs/uuid"
-
 	"github.com/elastic/beats/v7/journalbeat/checkpoint"
 	"github.com/elastic/beats/v7/journalbeat/reader"
 	"github.com/elastic/beats/v7/libbeat/beat"
@@ -43,7 +41,6 @@ type Input struct {
 	pipeline   beat.Pipeline
 	client     beat.Client
 	states     map[string]checkpoint.JournalState
-	id         uuid.UUID
 	logger     *logp.Logger
 	eventMeta  common.EventMetadata
 	processors beat.ProcessorList
@@ -61,13 +58,11 @@ func New(
 		return nil, err
 	}
 
-	id, err := uuid.NewV4()
-	if err != nil {
-		return nil, fmt.Errorf("error while generating ID for input: %v", err)
+	logger := logp.NewLogger("input")
+	if config.ID != "" {
+		logger = logger.With("id", config.ID)
 	}
 
-	logger := logp.NewLogger("input").With("id", id)
-
 	var readers []*reader.Reader
 	if len(config.Paths) == 0 {
 		cfg := reader.Config{
@@ -78,9 +73,10 @@ func New(
 			CursorSeekFallback: config.CursorSeekFallback,
 			Matches:            config.Matches,
 			SaveRemoteHostname: config.SaveRemoteHostname,
+			CheckpointID:       checkpointID(config.ID, reader.LocalSystemJournalID),
 		}
 
-		state := states[reader.LocalSystemJournalID]
+		state := states[cfg.CheckpointID]
 		r, err := reader.NewLocal(cfg, done, state, logger)
 		if err != nil {
 			return nil, fmt.Errorf("error creating reader for local journal: %v", err)
@@ -97,8 +93,10 @@ func New(
 			CursorSeekFallback: config.CursorSeekFallback,
 			Matches:            config.Matches,
 			SaveRemoteHostname: config.SaveRemoteHostname,
+			CheckpointID:       checkpointID(config.ID, p),
 		}
-		state := states[p]
+
+		state := states[cfg.CheckpointID]
 		r, err := reader.New(cfg, done, state, logger)
 		if err != nil {
 			return nil, fmt.Errorf("error creating reader for journal: %v", err)
@@ -119,7 +117,6 @@ func New(
 		config:     config,
 		pipeline:   b.Publisher,
 		states:     states,
-		id:         id,
 		logger:     logger,
 		eventMeta:  config.EventMetadata,
 		processors: inputProcessors,
@@ -138,7 +135,7 @@ func (i *Input) Run() {
 			Processor:     i.processors,
 		},
 		ACKCount: func(n int) {
-			i.logger.Infof("journalbeat successfully published %d events", n)
+			i.logger.Debugw("journalbeat successfully published events", "event.count", n)
 		},
 	})
 	if err != nil {
@@ -233,3 +230,12 @@ func processorsForInput(beatInfo beat.Info, config Config) (*processors.Processo
 
 	return procs, nil
 }
+
+// checkpointID returns the identifier used to track persistent state for the
+// input.
+func checkpointID(id, path string) string {
+	if id == "" {
+		return path
+	}
+	return "journald::" + path + "::" + id
+}

journalbeat/journalbeat.reference.yml

@@ -18,6 +18,11 @@ journalbeat.inputs:
   # When empty starts to read from local journal.
 - paths: []
 
+  # An optional unique identifier for the input. By providing a unique `id` you
+  # can operate multiple inputs on the same journal. This allows each input's
+  # cursor to be persisted independently in the registry file.
+  #id: ""
+
   # The number of seconds to wait before trying to read again from journals.
   #backoff: 1s
   # The maximum number of seconds to wait before attempting to read again from journals.

journalbeat/journalbeat.yml

@@ -18,6 +18,11 @@ journalbeat.inputs:
   # When empty starts to read from local journal.
 - paths: []
 
+  # An optional unique identifier for the input. By providing a unique `id` you
+  # can operate multiple inputs on the same journal. This allows each input's
+  # cursor to be persisted independently in the registry file.
+  #id: ""
+
   # The number of seconds to wait before trying to read again from journals.
   #backoff: 1s
   # The maximum number of seconds to wait before attempting to read again from journals.

journalbeat/reader/config.go

@@ -41,6 +41,8 @@ type Config struct {
 	Matches []string
 	// SaveRemoteHostname defines if the original source of the entry needs to be saved.
 	SaveRemoteHostname bool
+	// CheckpointID is the identifier to use when persisting state.
+	CheckpointID string
 }
 
 const (

journalbeat/reader/journal.go

@@ -267,7 +267,7 @@ func (r *Reader) toEvent(entry *sdjournal.JournalEntry) *beat.Event {
 	}
 
 	state := checkpoint.JournalState{
-		Path:               r.config.Path,
+		Path:               r.config.CheckpointID,
 		Cursor:             entry.Cursor,
 		RealtimeTimestamp:  entry.RealtimeTimestamp,
 		MonotonicTimestamp: entry.MonotonicTimestamp,

journalbeat/tests/system/config/journalbeat.yml.j2

@@ -1,85 +1,13 @@
-################### Beat Configuration #########################
 journalbeat.inputs:
-- paths: [{{ journal_path }}]
-  seek: {{ seek_method }}
-  {% if cursor_seek_fallback %}
-  cursor_seek_fallback: {{ cursor_seek_fallback }}
-  {% endif %}
-  include_matches: [{{ matches }}]
+{% for input in inputs %}
+- {{ input | tojson }}
+{% endfor %}
 
+{% if registry_file is defined %}
 journalbeat.registry_file: {{ registry_file }}
+{% endif %}
 
-############################# Output ##########################################
-
-# Configure what outputs to use when sending the data collected by the beat.
-# You can enable one or multiple outputs by setting enabled option to true.
 output:
-
-  ### File as output
   file:
-    # Enabling file output
-    enabled: true
-
-    # Path to the directory where to save the generated files. The option is mandatory.
     path: {{ output_file_path|default(beat.working_dir + "/output") }}
-
-
-    # Name of the generated files. The default is `journalbeat` and it generates
-    # files: `journalbeat`, `journalbeat.1`, `journalbeat.2`, etc.
     filename: {{ output_file_filename|default("journalbeat") }}
-
-    # Maximum size in kilobytes of each file. When this size is reached, the files are
-    # rotated. The default value is 10 MB.
-    #rotate_every_kb: 10000
-
-    # Maximum number of files under path. When this number of files is reached, the
-    # oldest file is deleted and the rest are shifted from last to first. The default
-    # is 7 files.
-    #number_of_files: 7
-
-
-
-############################# Beat #########################################
-
-# The name of the shipper that publishes the network data. It can be used to group
-# all the transactions sent by a single shipper in the web interface.
-# If this options is not defined, the hostname is used.
-#name:
-
-# The tags of the shipper are included in their own field with each
-# transaction published. Tags make it easy to group servers by different
-# logical properties.
-#tags: ["service-X", "web-tier"]
-
-
-
-############################# Logging #########################################
-
-#logging:
-  # Send all logging output to syslog. On Windows default is false, otherwise
-  # default is true.
-  #to_syslog: true
-
-  # Write all logging output to files. Beats automatically rotate files if configurable
-  # limit is reached.
-  #to_files: false
-
-  # Enable debug output for selected components.
-  #selectors: []
-
-  # Set log level
-  #level: error
-
-  #files:
-    # The directory where the log files will written to.
-    #path: /var/log/journalbeat
-
-    # The name of the files where the logs are written to.
-    #name: journalbeat
-
-    # Configure log file size limit. If limit is reached, log file will be
-    # automatically rotated
-    #rotateeverybytes: 10485760 # = 10MB
-
-    # Number of rotated log files to keep. Oldest files will be deleted first.
-    #keepfiles: 7