support filtering on a specific unit

CHANGELOG.md

@@ -12,6 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - ARM64 Container Image: [PR 381](https://github.com/observIQ/stanza/pull/381)
 - TCP Input: Minimum TLS version is now configurable: [PR 400](https://github.com/observIQ/stanza/pull/400)
 - Systemd service: Set `TimeoutSec` [PR 402](https://github.com/observIQ/stanza/pull/402)
+- Journald input: Opetioanl `unit` parameter for filtering [PR 403](https://github.com/observIQ/stanza/pull/403)
 - Updated dependencies:
   - go.uber.org/multierr [PR 387](https://github.com/observIQ/stanza/pull/387)
   - go.etcd.io/bbolt [PR 385](https://github.com/observIQ/stanza/pull/385)

docs/operators/journald_input.md

@@ -15,6 +15,7 @@ The `journald_input` operator will use the `__REALTIME_TIMESTAMP` field of the j
 | `poll_interval`   | 200ms            | The duration between journal polls                                                               |
 | `directory`       |                  | A directory containing journal files to read entries from                                        |
 | `files`           |                  | A list of journal files to read entries from                                                     |
+| `unit`            |                  | Filter on a specific unit                                                                        |
 | `write_to`        | $                | The record [field](/docs/types/field.md) written to when creating a new log entry                |
 | `start_at`        | `end`            | At startup, where to start reading logs from the file. Options are `beginning` or `end`          |
 | `labels`          | {}               | A map of `key: value` labels to add to the entry's labels                                        |
@@ -71,3 +72,45 @@ Output entry sample:
   }
 }
 ```
+
+#### Filter on a Unit
+
+Configuration:
+```yaml
+pipeline:
+- type: journald_input
+  unit: docker.service
+```
+
+Output entry sample:
+```json
+{
+  "timestamp": "2021-08-20T20:44:33.72269-04:00",
+  "severity": 0,
+  "record": {
+    "MESSAGE": "time=\"2021-08-20T20:44:33.722649189-04:00\" level=warning msg=\"cleanup warnings time=\\\"2021-08-20T20:44:33-04:00\\\" level=info msg=\\\"starting signal loop\\\" namespace=moby pid=1221814\\n\"",
+    "PRIORITY": "6",
+    "SYSLOG_FACILITY": "3",
+    "SYSLOG_IDENTIFIER": "dockerd",
+    "_BOOT_ID": "da9e7908ac5748e4b1452e4f18355fec",
+    "_CAP_EFFECTIVE": "1ffffffffff",
+    "_CMDLINE": "containerd --config /var/run/docker/containerd/containerd.toml --log-level info",
+    "_COMM": "containerd",
+    "_EXE": "/usr/bin/containerd",
+    "_GID": "0",
+    "_HOSTNAME": "control-plane.minikube.internal",
+    "_MACHINE_ID": "3d7a7d5c419d468e81ff7c9a59b2deec",
+    "_PID": "3875",
+    "_SELINUX_CONTEXT": "kernel",
+    "_STREAM_ID": "4b96fbe8fcff41ee8663b7cf125dc0fb",
+    "_SYSTEMD_CGROUP": "/system.slice/docker.service",
+    "_SYSTEMD_INVOCATION_ID": "6dbcbf51f10e471a9616319be9b575cb",
+    "_SYSTEMD_SLICE": "system.slice",
+    "_SYSTEMD_UNIT": "docker.service",
+    "_TRANSPORT": "stdout",
+    "_UID": "0",
+    "__CURSOR": "s=b7ef2bc7c5d3441ebfa90e7b94ab92ee;i=5d93f9;b=da9e7908ac5748e4b1452e4f18355fec;m=c2a7fc009;t=5ca0716724542;x=f85983418a47c115",
+    "__MONOTONIC_TIMESTAMP": "52252622857"
+  }
+}
+```

operator/builtin/input/journald/journald.go

@@ -40,6 +40,7 @@ type JournaldInputConfig struct {
 	Directory    *string         `json:"directory,omitempty"     yaml:"directory,omitempty"`
 	Files        []string        `json:"files,omitempty"         yaml:"files,omitempty"`
 	StartAt      string          `json:"start_at,omitempty"      yaml:"start_at,omitempty"`
+	Unit         string          `json:"unit,omitempty"          yaml:"unit,omitempty"`
 	PollInterval helper.Duration `json:"poll_interval,omitempty" yaml:"poll_interval,omitempty"`
 }
 
@@ -81,6 +82,10 @@ func (c JournaldInputConfig) Build(buildContext operator.BuildContext) ([]operat
 		}
 	}
 
+	if c.Unit != "" {
+		args = append(args, fmt.Sprintf("--unit=%s", c.Unit))
+	}
+
 	journaldInput := &JournaldInput{
 		InputOperator: inputOperator,
 		persist:       helper.NewScopedDBPersister(buildContext.Database, c.ID()),