[8.18](backport #7995) Enable Filebeat's Journald input in some docker image variants (#8482)

* Enable Filebeat's Journald input in some docker image variants (#7995)

Filebeat's Journald input requires `journalctl`, so the systemd
package is added to elastic-agent, elastic-agent-complete, elastic-otel-collector variants.

---------

Co-authored-by: Craig MacKenzie <[email protected]>
(cherry picked from commit fd33d85)

# Conflicts:
#	dev-tools/packaging/settings.go
#	dev-tools/packaging/templates/docker/Dockerfile.elastic-agent.tmpl

* Fix merge conflicts

Mostly re-doing the addition of systemd packages due to the different
base images

* Update env var usage

* Only install jounralctl in the complete variant

* remove basic variant from TestKubernetesJournaldInput

* remove TestKubernetesJournaldInputOtel as edot collector variant is not relevant to 8.18

* Update changelog/fragments/1745609163-journalctl-on-all-docker-variants.yaml

---------

Co-authored-by: Tiago Queiroz <[email protected]>
Co-authored-by: Panos Koutsovasilis <[email protected]>

Author: 3 people

changelog/fragments/1745609163-journalctl-on-all-docker-variants.yaml

@@ -0,0 +1,35 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# Change summary; a 80ish characters long description of the change.
+summary: |
+  Ship journalctl in the elastic-agent-complete, Docker image
+  to enable reading journald logs. Journalctl is not present on 
+  Wolfi images.
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+#description: 
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: elastic-agent
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+pr: https://github.com/elastic/elastic-agent/pull/7995
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+issue: https://github.com/elastic/beats/issues/44040

dev-tools/packaging/templates/docker/Dockerfile.elastic-agent.tmpl

@@ -257,7 +257,8 @@ RUN for iter in {1..10}; do \
     $NODE_PATH/node/lib/node_modules/@elastic/synthetics/node_modules/.bin/playwright install-deps chromium && \
     DEBIAN_FRONTEND=noninteractive apt-get install --no-install-recommends --yes \
     fonts-noto \
-    fonts-noto-cjk && \
+    fonts-noto-cjk \
+    systemd && \
     exit_code=0 && break || exit_code=$? && echo "apt-get error: retry $iter in 10s" && sleep 10; \
     done; \
     (exit $exit_code)

magefile.go

@@ -2083,6 +2083,16 @@ func (Integration) Kubernetes(ctx context.Context) error {
 	return integRunner(ctx, "testing/integration", false, "")
 }
 
+// KubernetesSingle runs a single Kubernetes integration test
+func (Integration) KubernetesSingle(ctx context.Context, testName string) error {
+	// invoke integration tests
+	if err := os.Setenv("TEST_GROUPS", "kubernetes"); err != nil {
+		return err
+	}
+
+	return integRunner(ctx, "testing/integration", false, testName)
+}
+
 // KubernetesMatrix runs a matrix of kubernetes integration tests
 func (Integration) KubernetesMatrix(ctx context.Context) error {
 	// invoke integration tests

testing/integration/journald_test.go

@@ -0,0 +1,150 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License 2.0;
+// you may not use this file except in compliance with the Elastic License 2.0.
+
+//go:build integration
+
+package integration
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	"sigs.k8s.io/e2e-framework/klient/k8s"
+
+	"github.com/elastic/elastic-agent-libs/testing/estools"
+	"github.com/elastic/elastic-agent/pkg/testing/define"
+	"github.com/elastic/go-elasticsearch/v8"
+)
+
+func TestKubernetesJournaldInput(t *testing.T) {
+	info := define.Require(t, define.Requirements{
+		Stack: &define.Stack{},
+		Local: false,
+		Sudo:  false,
+		OS: []define.OS{
+			{Type: define.Kubernetes, DockerVariant: "complete"},
+		},
+		Group: define.Kubernetes,
+	})
+
+	agentConfigYAML, err := os.ReadFile(filepath.Join("testdata", "journald-input.yml"))
+	require.NoError(t, err, "failed to read journald input template")
+
+	ctx := context.Background()
+	kCtx := k8sGetContext(t, info)
+
+	schedulableNodeCount, err := k8sSchedulableNodeCount(ctx, kCtx)
+	require.NoError(t, err, "error at getting schedulable node count")
+	require.NotZero(t, schedulableNodeCount, "no schedulable Kubernetes nodes found")
+
+	namespace := kCtx.getNamespace(t)
+	hostPathType := corev1.HostPathDirectory
+
+	steps := []k8sTestStep{
+		k8sStepCreateNamespace(),
+		k8sStepDeployKustomize(
+			agentK8SKustomize,
+			"elastic-agent-standalone",
+			k8sKustomizeOverrides{
+				agentContainerExtraEnv: []corev1.EnvVar{
+					{
+						Name:  "ELASTICSEARCH_USERNAME",
+						Value: os.Getenv("ELASTICSEARCH_USERNAME"),
+					},
+					{
+						Name:  "ELASTICSEARCH_PASSWORD",
+						Value: os.Getenv("ELASTICSEARCH_PASSWORD"),
+					},
+					{
+						Name:  "EA_POLICY_NAMESPACE",
+						Value: namespace,
+					},
+				},
+				agentContainerVolumeMounts: []corev1.VolumeMount{
+					{
+						Name:      "journald-mount",
+						MountPath: "/opt/journald",
+						ReadOnly:  true,
+					},
+				},
+				agentPodVolumes: []corev1.Volume{
+					{
+						Name: "journald-mount",
+						VolumeSource: corev1.VolumeSource{
+							HostPath: &corev1.HostPathVolumeSource{
+								Path: "/run/log/journal",
+								Type: &hostPathType,
+							},
+						},
+					},
+				},
+			},
+			func(obj k8s.Object) {
+				// update the configmap to use the journald input
+				switch objWithType := obj.(type) {
+				case *corev1.ConfigMap:
+					_, ok := objWithType.Data["agent.yml"]
+					if ok {
+						objWithType.Data["agent.yml"] = string(agentConfigYAML)
+					}
+				}
+
+			}),
+		k8sStepCheckAgentStatus(
+			"app=elastic-agent-standalone",
+			schedulableNodeCount,
+			"elastic-agent-standalone",
+			map[string]bool{
+				"journald": true,
+			}),
+	}
+
+	journaldTest(
+		t,
+		info.ESClient,
+		kCtx,
+		steps,
+		fmt.Sprintf("logs-generic-%s", namespace),
+		"input.type",
+		"journald")
+}
+
+func journaldTest(
+	t *testing.T,
+	esClient *elasticsearch.Client,
+	kCtx k8sContext,
+	steps []k8sTestStep,
+	index, field, value string) {
+	t.Helper()
+
+	ctx := context.Background()
+	testNamespace := kCtx.getNamespace(t)
+
+	for _, step := range steps {
+		step(t, ctx, kCtx, testNamespace)
+	}
+
+	// Check if the context was cancelled or timed out
+	if ctx.Err() != nil {
+		t.Errorf("context error: %v", ctx.Err())
+	}
+
+	// Query the index and filter by the input type
+	docs := findESDocs(t, func() (estools.Documents, error) {
+		return estools.GetLogsForIndexWithContext(
+			ctx,
+			esClient,
+			index,
+			map[string]any{
+				field: value,
+			},
+		)
+	})
+	require.NotEmpty(t, docs, "expected logs to be found in Elasticsearch")
+}

testing/integration/kubernetes_agent_standalone_test.go

@@ -897,7 +897,7 @@ func getAgentComponentState(status atesting.AgentStatusOutput, componentName str
 // k8sDumpPods creates an archive that contains logs of all pods in the given namespace and kube-system to the given target directory
 func k8sDumpPods(t *testing.T, ctx context.Context, client klient.Client, testName string, namespace string, targetDir string, testStartTime time.Time) {
 	// Create the tar file
-	archivePath := filepath.Join(targetDir, fmt.Sprintf("%s.tar.gz", namespace))
+	archivePath := filepath.Join(targetDir, fmt.Sprintf("%s.tar", namespace))
 	tarFile, err := os.Create(archivePath)
 	if err != nil {
 		t.Logf("failed to create archive at path %q", archivePath)
@@ -1311,8 +1311,14 @@ type k8sContext struct {
 	createdAt time.Time
 }
 
-// getNamespace returns a unique namespace for the current test
+// getNamespace returns a unique namespace on every call.
+// If K8S_TESTS_NAMESPACE is set, then its value is returned,
+// otherwise a unique namespace is generated.
 func (k k8sContext) getNamespace(t *testing.T) string {
+	if ns := os.Getenv("K8S_TESTS_NAMESPACE"); ns != "" {
+		return ns
+	}
+
 	nsUUID, err := uuid.NewV4()
 	if err != nil {
 		t.Fatalf("error generating namespace UUID: %v", err)
@@ -1379,8 +1385,8 @@ func k8sGetContext(t *testing.T, info *define.Info) k8sContext {
 	err = os.MkdirAll(testLogsBasePath, 0o755)
 	require.NoError(t, err, "failed to create test logs directory")
 
-	esHost := os.Getenv("ELASTICSEARCH_HOST")
-	require.NotEmpty(t, esHost, "ELASTICSEARCH_HOST must be set")
+	esHost, err := getESHost()
+	require.NoError(t, err, "cannot parse ELASTICSEARCH_HOST")
 
 	esAPIKey, err := generateESAPIKey(info.ESClient, info.Namespace)
 	require.NoError(t, err, "failed to generate ES API key")
@@ -1440,6 +1446,8 @@ type k8sKustomizeOverrides struct {
 	agentContainerExtraEnv         []corev1.EnvVar
 	agentContainerArgs             []string
 	agentContainerMemoryLimit      string
+	agentContainerVolumeMounts     []corev1.VolumeMount
+	agentPodVolumes                []corev1.Volume
 }
 
 // k8sStepDeployKustomize renders a kustomize manifest and deploys it. Also, it tries to
@@ -1466,6 +1474,8 @@ func k8sStepDeployKustomize(kustomizePath string, containerName string, override
 
 		k8sKustomizeAdjustObjects(objects, namespace, containerName,
 			func(container *corev1.Container) {
+				container.VolumeMounts = append(container.VolumeMounts, overrides.agentContainerVolumeMounts...)
+
 				// set agent image
 				container.Image = kCtx.agentImage
 				// set ImagePullPolicy to "Never" to avoid pulling the image
@@ -1509,8 +1519,7 @@ func k8sStepDeployKustomize(kustomizePath string, containerName string, override
 				}
 
 				if overrides.agentContainerArgs != nil {
-					// drop arguments overriding default config
-					container.Args = []string{}
+					container.Args = overrides.agentContainerArgs
 				}
 			},
 			func(pod *corev1.PodSpec) {
@@ -1525,6 +1534,7 @@ func k8sStepDeployKustomize(kustomizePath string, containerName string, override
 						}
 					}
 				}
+				pod.Volumes = append(pod.Volumes, overrides.agentPodVolumes...)
 			})
 
 		t.Cleanup(func() {

testing/integration/testdata/journald-input.yml

@@ -0,0 +1,22 @@
+outputs:
+  default:
+    type: elasticsearch
+    hosts:
+      - ${ES_HOST}
+    username: ${ELASTICSEARCH_USERNAME}
+    password: ${ELASTICSEARCH_PASSWORD}
+
+agent:
+  monitoring:
+    enabled: false
+
+inputs:
+  - id: journald
+    log_level: debug
+    type: journald
+    data_stream:
+      namespace: ${env.EA_POLICY_NAMESPACE}
+    streams:
+      - id: journald-input-id
+        paths:
+          - "/opt/journald/*/*"