[Filebeat/Journald] Add support for watching folders

[belimawr]

Proposed commit message

This commit implements the use of `--directory` argument when calling `journalctl`, thus allowing it to watch for new files in the folder.

This is implemented by calling `os.Stat` on the path and checking if it is a folder, if so, then `--directory` is used instead of `--file`.

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation
• I have made corresponding change to the default configuration files
• I have added tests that prove my fix is effective or that my feature works
• I have added an entry in CHANGELOG.next.asciidoc or CHANGELOG-developer.next.asciidoc.

## Disruptive User Impact
## Author's Checklist

How to test this PR locally

Testing in Kubernetes to ingest node logs

1. Start a Kind cluster

2. Deploy the Elastic Stack, make sure it will be accessible from inside the Kind cluster
   I use my own fork of elastic-package to bind on all interfaces.

3. Build Filebeat docker image

   DEV=true SNAPSHOT=true EXTERNAL=true PACKAGES="docker" PLATFORMS=linux/amd64 mage -v package
   

4. Load the image into your Kind cluster

   kind load docker-image docker.elastic.co/beats/filebeat-oss:9.2.0-SNAPSHOT
   

5. Deploy Filebeat reading from /run/log/journal
   Note that the journald configuration points to the folder /run/log/journal

   - type: journald
     id: node-logs
     paths:
       - /run/log/journal

   filebeat-kubernetes.yaml

   apiVersion: v1
   kind: ServiceAccount
   metadata:
     name: filebeat
     namespace: kube-system
     labels:
       k8s-app: filebeat
   ---
   apiVersion: rbac.authorization.k8s.io/v1
   kind: ClusterRole
   metadata:
     name: filebeat
     labels:
       k8s-app: filebeat
   rules:
   - apiGroups: [""] # "" indicates the core API group
     resources:
     - namespaces
     - pods
     - nodes
     verbs:
     - get
     - watch
     - list
   - apiGroups: ["apps"]
     resources:
       - replicasets
     verbs: ["get", "list", "watch"]
   - apiGroups: ["batch"]
     resources:
       - jobs
     verbs: ["get", "list", "watch"]
   ---
   apiVersion: rbac.authorization.k8s.io/v1
   kind: Role
   metadata:
     name: filebeat
     # should be the namespace where filebeat is running
     namespace: kube-system
     labels:
       k8s-app: filebeat
   rules:
     - apiGroups:
         - coordination.k8s.io
       resources:
         - leases
       verbs: ["get", "create", "update"]
   ---
   apiVersion: rbac.authorization.k8s.io/v1
   kind: Role
   metadata:
     name: filebeat-kubeadm-config
     namespace: kube-system
     labels:
       k8s-app: filebeat
   rules:
     - apiGroups: [""]
       resources:
         - configmaps
       resourceNames:
         - kubeadm-config
       verbs: ["get"]
   ---
   apiVersion: rbac.authorization.k8s.io/v1
   kind: ClusterRoleBinding
   metadata:
     name: filebeat
   subjects:
   - kind: ServiceAccount
     name: filebeat
     namespace: kube-system
   roleRef:
     kind: ClusterRole
     name: filebeat
     apiGroup: rbac.authorization.k8s.io
   ---
   apiVersion: rbac.authorization.k8s.io/v1
   kind: RoleBinding
   metadata:
     name: filebeat
     namespace: kube-system
   subjects:
     - kind: ServiceAccount
       name: filebeat
       namespace: kube-system
   roleRef:
     kind: Role
     name: filebeat
     apiGroup: rbac.authorization.k8s.io
   ---
   apiVersion: rbac.authorization.k8s.io/v1
   kind: RoleBinding
   metadata:
     name: filebeat-kubeadm-config
     namespace: kube-system
   subjects:
     - kind: ServiceAccount
       name: filebeat
       namespace: kube-system
   roleRef:
     kind: Role
     name: filebeat-kubeadm-config
     apiGroup: rbac.authorization.k8s.io
   ---
   apiVersion: v1
   kind: ConfigMap
   metadata:
     name: filebeat-config
     namespace: kube-system
     labels:
       k8s-app: filebeat
   data:
     filebeat.yml: |-
       filebeat.inputs:
       - type: journald
         id: node-logs
         paths:
           - /run/log/journal
   
       processors:
         - add_host_metadata:
   
       output.elasticsearch:
         hosts: ['https://172.17.0.1:9200'] # host machine
         username: elastic
         password: changeme
         ssl.verification_mode: none
   ---
   apiVersion: apps/v1
   kind: DaemonSet
   metadata:
     name: filebeat
     namespace: kube-system
     labels:
       k8s-app: filebeat
   spec:
     selector:
       matchLabels:
         k8s-app: filebeat
     template:
       metadata:
         labels:
           k8s-app: filebeat
       spec:
         serviceAccountName: filebeat
         terminationGracePeriodSeconds: 30
         hostNetwork: true
         dnsPolicy: ClusterFirstWithHostNet
         containers:
         - name: filebeat
           image: docker.elastic.co/beats/filebeat-oss:9.2.0-SNAPSHOT
           args: [
             "-c", "/etc/filebeat.yml",
             "-e",
           ]
           env:
           - name: NODE_NAME
             valueFrom:
               fieldRef:
                 fieldPath: spec.nodeName
           securityContext:
             runAsUser: 0
             # If using Red Hat OpenShift uncomment this:
             #privileged: true
           resources:
             limits:
               memory: 200Mi
             requests:
               cpu: 100m
               memory: 100Mi
           volumeMounts:
           - name: config
             mountPath: /etc/filebeat.yml
             readOnly: true
             subPath: filebeat.yml
           - name: data
             mountPath: /usr/share/filebeat/data
           - name: varlibdockercontainers
             mountPath: /var/lib/docker/containers
             readOnly: true
           - name: varlog
             mountPath: /var/log
             readOnly: true
           - name: runlogjournal
             mountPath: /run/log/journal
             readOnly: true
         volumes:
         - name: config
           configMap:
             defaultMode: 0640
             name: filebeat-config
         - name: varlibdockercontainers
           hostPath:
             path: /var/lib/docker/containers
         - name: varlog
           hostPath:
             path: /var/log
         # data folder stores a registry of read status for all files, so we don't send everything again on a Filebeat pod restart
         - name: data
           hostPath:
             # When filebeat runs as non-root user, this directory needs to be writable by group (g+w).
             path: /var/lib/filebeat-data
             type: DirectoryOrCreate
         - name: runlogjournal
           hostPath:
             path: /run/log/journal
   ---
   

   kubectl apply -f filebeat-kubernetes.yaml
   

6. Ensure the logs are coming in in Kibana

   1. Create a dataview to show logs from filebeat-9.2.0
   2. Go to Discover and check for the logs

Using journal files

1. Decompress all journal?.journal.gz files and prepare the destination folder

   cd filebeat/input/journald/testdata
   gunzip journal?.journal.gz
   mkdir /tmp/journald
   

2. Start Filebeat with the following configuration

   filebeat.yaml

   filebeat.inputs:
     - type: journald
       id: foo
       paths:
         - /tmp/journal
   
   queue.mem:
     flush.timeout: 0
   
   output.discard:
     enabled: true
   
   http.enabled: true
   http.host: localhost
   http.port: 5066
   

   You might see some erros in the logs because `/tmp/journald` is empty, ignore them.

3. Start monitoring the stats endpoint for published events

   watch -c 'curl -s http://localhost:5066/stats|jq '.filebeat.events''
   

   You will see a JSON like:

   {
     "active": 0,
     "added": 0,
     "done": 0
   }    

   Keep an eye in the done counter, that's the number of published events.

4. Move the first journal file to /tmp/journald and wait until all events are published

   mv journal1.journal /tmp/journal
   

   Wait untill the done counter is 10.

5. Repeat the process for the other two journal files
   There are 10 entries in each file, so at the end of the test, there should be 30 published files.

   mv journal2.journal /tmp/journal
   # Wait 'done' to be 20
   mv journal3.journal /tmp/journal
   # Wait 'done' to be 30
   

Related issues

• Closes [Journald] Better support for reading from directories #46657

## Use cases
## Screenshots
## Logs

[github-actions]

🤖 GitHub comments

Expand to view the GitHub comments

Just comment with:

• run docs-build : Re-trigger the docs validation. (use unformatted text in the comment!)

[elasticmachine]

Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane)

[AndersonQ]

@belimawr a suggestion, when decompressing the files, pass in the -k flag so the original file won't be deleted as the files are tracked by git :)

[github-actions]

@Mergifyio backport 8.18 8.19 9.0 9.1

[mergify]

backport 8.18 8.19 9.0 9.1

✅ Backports have been created

• #46715 [8.18](backport #46682) [Filebeat/Journald] Add support for watching folders has been created for branch 8.18
• #46716 [8.19](backport #46682) [Filebeat/Journald] Add support for watching folders has been created for branch 8.19
• #46717 [9.0](backport #46682) [Filebeat/Journald] Add support for watching folders has been created for branch 9.0
• #46718 [9.1](backport #46682) [Filebeat/Journald] Add support for watching folders has been created for branch 9.1