[receiver/journald]: add support for matches configuration

.chloggen/drosiek-journald-matches.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: journaldreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: add support for `matches` configuration
+
+# One or more tracking issues related to the change
+issues: [20295]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

pkg/stanza/docs/operators/journald_input.md

@@ -15,6 +15,7 @@ The `journald_input` operator will use the `__REALTIME_TIMESTAMP` field of the j
 | `directory`       |                  | A directory containing journal files to read entries from. |
 | `files`           |                  | A list of journal files to read entries from. |
 | `units`           |                  | A list of units to read entries from. |
+| `matches`         |                  | A list of matches to read entries from. See [Matches](#matches) example. |
 | `priority`        | `info`           | Filter output by message priorities or priority ranges. |
 | `start_at`        | `end`            | At startup, where to start reading logs from the file. Options are `beginning` or `end`. |
 | `attributes`      | {}               | A map of `key: value` pairs to add to the entry's attributes. |
@@ -33,7 +34,23 @@ The `journald_input` operator will use the `__REALTIME_TIMESTAMP` field of the j
 - type: journald_input
   priority: emerg..err
 ```
-#### Simple journald input
+
+#### Matches
+
+The following configuration:
+
+```yaml
+- type: journald_input
+  matches:
+    - _SYSTEMD_UNIT: ssh
+    - _SYSTEMD_UNIT: kubelet
+      _UID: "1000"
+  priority: info
+```
+
+will be passed to `journald` as the following arguments: `journald ... _SYSTEMD_UNIT=dbus.service + _SYSTEMD_UNIT=user@1000.service _UID=1000`
+
+### Simple journald input
 
 Configuration:
 ```yaml

pkg/stanza/operator/input/journald/journald.go

@@ -24,6 +24,8 @@ import (
 	"fmt"
 	"io"
 	"os/exec"
+	"regexp"
+	"sort"
 	"strconv"
 	"sync"
 	"time"
@@ -60,20 +62,42 @@ func NewConfigWithID(operatorID string) *Config {
 type Config struct {
 	helper.InputConfig `mapstructure:",squash"`
 
-	Directory *string  `mapstructure:"directory,omitempty"`
-	Files     []string `mapstructure:"files,omitempty"`
-	StartAt   string   `mapstructure:"start_at,omitempty"`
-	Units     []string `mapstructure:"units,omitempty"`
-	Priority  string   `mapstructure:"priority,omitempty"`
+	Directory *string       `mapstructure:"directory,omitempty"`
+	Files     []string      `mapstructure:"files,omitempty"`
+	StartAt   string        `mapstructure:"start_at,omitempty"`
+	Units     []string      `mapstructure:"units,omitempty"`
+	Priority  string        `mapstructure:"priority,omitempty"`
+	Matches   []MatchConfig `mapstructure:"matches,omitempty"`
 }
 
+type MatchConfig map[string]string
+
 // Build will build a journald input operator from the supplied configuration
 func (c Config) Build(logger *zap.SugaredLogger) (operator.Operator, error) {
 	inputOperator, err := c.InputConfig.Build(logger)
 	if err != nil {
 		return nil, err
 	}
 
+	args, err := c.buildArgs()
+	if err != nil {
+		return nil, err
+	}
+
+	return &Input{
+		InputOperator: inputOperator,
+		newCmd: func(ctx context.Context, cursor []byte) cmd {
+			if cursor != nil {
+				args = append(args, "--after-cursor", string(cursor))
+			}
+			return exec.CommandContext(ctx, "journalctl", args...) // #nosec - ...
+			// journalctl is an executable that is required for this operator to function
+		},
+		json: jsoniter.ConfigFastest,
+	}, nil
+}
+
+func (c Config) buildArgs() ([]string, error) {
 	args := make([]string, 0, 10)
 
 	// Export logs in UTC time
@@ -108,17 +132,54 @@ func (c Config) Build(logger *zap.SugaredLogger) (operator.Operator, error) {
 		}
 	}
 
-	return &Input{
-		InputOperator: inputOperator,
-		newCmd: func(ctx context.Context, cursor []byte) cmd {
-			if cursor != nil {
-				args = append(args, "--after-cursor", string(cursor))
-			}
-			return exec.CommandContext(ctx, "journalctl", args...) // #nosec - ...
-			// journalctl is an executable that is required for this operator to function
-		},
-		json: jsoniter.ConfigFastest,
-	}, nil
+	if len(c.Matches) > 0 {
+		ms, err := c.buildMatchesConfig()
+		if err != nil {
+			return nil, err
+		}
+		args = append(args, ms...)
+	}
+
+	return args, nil
+}
+
+func buildMatchConfig(mc MatchConfig) ([]string, error) {
+	re := regexp.MustCompile("^[_A-Z]+$")
+
+	// Sort keys to be consistent with every run and to be predictable for tests
+	sortedKeys := make([]string, 0, len(mc))
+	for key := range mc {
+		if !re.MatchString(key) {
+			return []string{}, fmt.Errorf("'%s' is not a valid Systemd field name", key)
+		}
+		sortedKeys = append(sortedKeys, key)
+	}
+	sort.Strings(sortedKeys)
+
+	configs := []string{}
+	for _, key := range sortedKeys {
+		configs = append(configs, fmt.Sprintf("%s=%s", key, mc[key]))
+	}
+
+	return configs, nil
+}
+
+func (c Config) buildMatchesConfig() ([]string, error) {
+	matches := []string{}
+
+	for i, mc := range c.Matches {
+		if i > 0 {
+			matches = append(matches, "+")
+		}
+		ms, err := buildMatchConfig(mc)
+		if err != nil {
+			return []string{}, err
+		}
+
+		matches = append(matches, ms...)
+	}
+
+	return matches, nil
 }
 
 // Input is an operator that process logs using journald

pkg/stanza/operator/input/journald/journald_test.go

@@ -24,6 +24,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
@@ -115,3 +116,70 @@ func TestInputJournald(t *testing.T) {
 		require.FailNow(t, "Timed out waiting for entry to be read")
 	}
 }
+
+func TestBuildConfig(t *testing.T) {
+	testCases := []struct {
+		Name          string
+		Config        func(cfg *Config)
+		Expected      []string
+		ExpectedError string
+	}{
+		{
+			Name:     "empty config",
+			Config:   func(cfg *Config) {},
+			Expected: []string{"--utc", "--output=json", "--follow", "--priority", "info"},
+		},
+		{
+			Name: "units",
+			Config: func(cfg *Config) {
+				cfg.Units = []string{
+					"dbus.service",
+					"user@1000.service",
+				}
+			},
+			Expected: []string{"--utc", "--output=json", "--follow", "--unit", "dbus.service", "--unit", "user@1000.service", "--priority", "info"},
+		},
+		{
+			Name: "matches",
+			Config: func(cfg *Config) {
+				cfg.Matches = []MatchConfig{
+					{
+						"_SYSTEMD_UNIT": "dbus.service",
+					},
+					{
+						"_UID":          "1000",
+						"_SYSTEMD_UNIT": "user@1000.service",
+					},
+				}
+			},
+			Expected: []string{"--utc", "--output=json", "--follow", "--priority", "info", "_SYSTEMD_UNIT=dbus.service", "+", "_SYSTEMD_UNIT=user@1000.service", "_UID=1000"},
+		},
+		{
+			Name: "invalid match",
+			Config: func(cfg *Config) {
+				cfg.Matches = []MatchConfig{
+					{
+						"-SYSTEMD_UNIT": "dbus.service",
+					},
+				}
+			},
+			ExpectedError: "'-SYSTEMD_UNIT' is not a valid Systemd field name",
+		},
+	}
+
+	for _, tt := range testCases {
+		t.Run(tt.Name, func(t *testing.T) {
+			cfg := NewConfigWithID("my_journald_input")
+			tt.Config(cfg)
+			args, err := cfg.buildArgs()
+
+			if tt.ExpectedError != "" {
+				require.Error(t, err)
+				require.ErrorContains(t, err, tt.ExpectedError)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tt.Expected, args)
+		})
+	}
+}