[Kubernetes] Reroute container logs based on pod annotations

packages/kubernetes/_dev/build/docs/README.md

@@ -84,6 +84,14 @@ the masters won't be visible. In these cases it won't be possible to use `schedu
 The container-logs dataset requires access to the log files in each Kubernetes node where the container logs are stored.
 This defaults to `/var/log/containers/*${kubernetes.container.id}.log`.
 
+#### Routing
+
+The container-logs data stream allows routing logs to a different *dataset* or *namespace* using pod annotations. 
+
+For example, suppose you are running Nginx on your Kubernetes cluster, and you want to drive the Nginx container logs into a dedicated dataset or namespace. By annotating the pod with `elastic.co/namespace: nginx`, the integration will send all the container logs to the `nginx` namespace.
+
+To learn more about routing container-logs, see https://docs.elastic.co/integrations/kubernetes/container-logs.
+
 ### audit-logs
 
 The audit-logs dataset requires access to the log files on each Kubernetes node where the audit logs are stored.

packages/kubernetes/_dev/build/docs/container-logs.md

@@ -6,3 +6,81 @@ It requires access to the log files in each Kubernetes node where the container
 This defaults to `/var/log/containers/*${kubernetes.container.id}.log`.
 
 By default only {{ url "filebeat-input-filestream-parsers" "container parser" }} is enabled. Additional log parsers can be added as an advanced options configuration.
+
+
+## Rerouting based on pod annotations
+
+You can customize the routing of container logs events and sending them to different datasets and namespaces using pods' annotations.
+
+Routing customization can happen at:
+
+- pod definition time, e.g., using a deployment.
+- pod runtime, annotating pods using `kubectl`.
+
+
+### Set routing at pod definition time
+
+Here is an example of an Nginx deployment where we set both `elastic.co/dataset` and `elastic.co/namespace` annotations to route the container logs to specific datasets and namespace, respectively.
+
+```yaml
+# nginx-deployment.yaml
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: nginx-deployment
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: nginx
+  template:
+    metadata:
+      annotations:
+        elastic.co/dataset: kubernetes.container_logs.nginx
+        elastic.co/namespace: nginx
+      labels:
+        app: nginx
+        app.kubernetes.io/name: myservice
+        app.kubernetes.io/version: v0.1.2
+        app.kubernetes.io/instance: myservice-abcxzy
+    spec:
+      containers:
+        - name: nginx-container
+          image: nginx:latest
+          ports:
+            - containerPort: 80
+```
+
+
+### Set routing at runtime
+
+Suppose you want to change the container logs routing on a running container. In that case, you can annotate the pod using `kubectl`, and the integration will apply it immediately sending all the following documents to the new destination:
+
+Here is an example where we route the container logs for a pod running the Elastic Agent to the `kubernetes.container_logs.agents` dataset:
+
+```shell
+kubectl annotate pods elastic-agent-managed-daemonset-6p22g elastic.co/dataset=kubernetes.container_logs.agents
+```
+
+Here's a similar example to change the namespace on a pod running Nginx:
+
+```shell
+kubectl annotate pods elastic-agent-managed-daemonset-6p22g elastic.co/namespace=nginx
+```
+
+You can restore the standard routing by removing the annotations:
+
+```shell
+kubectl annotate pods elastic-agent-managed-daemonset-6p22g elastic.co/dataset-
+kubectl annotate pods elastic-agent-managed-daemonset-6p22g elastic.co/namespace-
+```
+
+### Annotations Reference
+
+Here are the annotations available to customize routing:
+
+
+| Label                  | Description                                              |
+| ---------------------- | -------------------------------------------------------- |
+| `elastic.co/dataset`   | Defines the target data stream's dataset for this pod.   |
+| `elastic.co/namespace` | Defines the target data stream's namespace for this pod. |

packages/kubernetes/changelog.yml

@@ -1,4 +1,9 @@
 # newer versions go on top
+- version: "1.45.0"
+  changes:
+    - description: Reroute container logs based on pod annotations.
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/7118
 - version: "1.44.0"
   changes:
     - description: Introducing kubernetes.deployment.status.* metrics

packages/kubernetes/data_stream/container_logs/_dev/test/pipeline/test-common-config.yml

@@ -0,0 +1,11 @@
+fields:
+  data_stream:
+    dataset: kubernetes.container_logs
+    namespace: default
+  kubernetes:
+    annotations:
+      elastic_co/dataset: kubernetes.container_logs.nginx
+      elastic_co/namespace: nginx
+    labels:
+      app_kubernetes_io/version: "v0.1.0"
+      app_kubernetes_io/name: "myservice"

packages/kubernetes/data_stream/container_logs/_dev/test/pipeline/test-nginx.log

@@ -0,0 +1,4 @@
+2023/07/25 15:24:11 [notice] 1#1: start worker process 33
+2023/07/25 15:24:11 [notice] 1#1: start worker process 34
+2023/07/25 15:24:11 [notice] 1#1: start worker process 35
+2023/07/25 15:24:11 [notice] 1#1: using the "epoll" event method

packages/kubernetes/data_stream/container_logs/_dev/test/pipeline/test-nginx.log-expected.json

@@ -0,0 +1,92 @@
+{
+    "expected": [
+        {
+            "data_stream": {
+                "dataset": "kubernetes.container_logs.nginx",
+                "namespace": "nginx",
+                "type": "logs"
+            },
+            "kubernetes": {
+                "annotations": {
+                    "elastic_co/dataset": "kubernetes.container_logs.nginx",
+                    "elastic_co/namespace": "nginx"
+                },
+                "labels": {
+                    "app_kubernetes_io/name": "myservice",
+                    "app_kubernetes_io/version": "v0.1.0"
+                }
+            },
+            "message": "2023/07/25 15:24:11 [notice] 1#1: start worker process 33",
+            "service": {
+                "name": "myservice",
+                "version": "v0.1.0"
+            }
+        },
+        {
+            "data_stream": {
+                "dataset": "kubernetes.container_logs.nginx",
+                "namespace": "nginx",
+                "type": "logs"
+            },
+            "kubernetes": {
+                "annotations": {
+                    "elastic_co/dataset": "kubernetes.container_logs.nginx",
+                    "elastic_co/namespace": "nginx"
+                },
+                "labels": {
+                    "app_kubernetes_io/name": "myservice",
+                    "app_kubernetes_io/version": "v0.1.0"
+                }
+            },
+            "message": "2023/07/25 15:24:11 [notice] 1#1: start worker process 34",
+            "service": {
+                "name": "myservice",
+                "version": "v0.1.0"
+            }
+        },
+        {
+            "data_stream": {
+                "dataset": "kubernetes.container_logs.nginx",
+                "namespace": "nginx",
+                "type": "logs"
+            },
+            "kubernetes": {
+                "annotations": {
+                    "elastic_co/dataset": "kubernetes.container_logs.nginx",
+                    "elastic_co/namespace": "nginx"
+                },
+                "labels": {
+                    "app_kubernetes_io/name": "myservice",
+                    "app_kubernetes_io/version": "v0.1.0"
+                }
+            },
+            "message": "2023/07/25 15:24:11 [notice] 1#1: start worker process 35",
+            "service": {
+                "name": "myservice",
+                "version": "v0.1.0"
+            }
+        },
+        {
+            "data_stream": {
+                "dataset": "kubernetes.container_logs.nginx",
+                "namespace": "nginx",
+                "type": "logs"
+            },
+            "kubernetes": {
+                "annotations": {
+                    "elastic_co/dataset": "kubernetes.container_logs.nginx",
+                    "elastic_co/namespace": "nginx"
+                },
+                "labels": {
+                    "app_kubernetes_io/name": "myservice",
+                    "app_kubernetes_io/version": "v0.1.0"
+                }
+            },
+            "message": "2023/07/25 15:24:11 [notice] 1#1: using the \"epoll\" event method",
+            "service": {
+                "name": "myservice",
+                "version": "v0.1.0"
+            }
+        }
+    ]
+}

packages/kubernetes/data_stream/container_logs/agent/stream/stream.yml.hbs

@@ -15,8 +15,43 @@ parsers:
     format: {{ containerParserFormat }}
 {{ additionalParsersConfig }}
 
-{{#if processors}}
 processors:
+{{!
+  Why do we need to add the following processors?
+  -----------------------------------------------
+
+  The kubernetes provider supports[^1] pods annotations, making it possible to add
+  them to the event using the `include_annotations` configuration option.
+
+  However, adding annotations to the event is disabled by default, and it is
+  not possible to enable it on Fleet-managed agents.
+
+  The following processors are a workaround to add the annotations to the event
+  without using the `include_annotations` configuration option.
+
+
+  [^1]: https://github.com/elastic/elastic-agent/blob/37ec2bb7ee1d2cc6c0fccf2f0cd0a44eb3d61efd/internal/pkg/composable/providers/kubernetes/pod.go#L311-L315
+}}
+- add_fields:
+    target: kubernetes
+    fields:
+      annotations.elastic_co/dataset: ${kubernetes.annotations.elastic.co/dataset|""}
+      annotations.elastic_co/namespace: ${kubernetes.annotations.elastic.co/namespace|""}
+- drop_fields:
+    fields:
+      - kubernetes.annotations.elastic_co/dataset
+    when:
+      equals:
+        kubernetes.annotations.elastic_co/dataset: ""
+    ignore_missing: true
+- drop_fields:
+    fields:
+      - kubernetes.annotations.elastic_co/namespace
+    when:
+      equals:
+        kubernetes.annotations.elastic_co/namespace: ""
+    ignore_missing: true
+{{#if processors}}      
 {{processors}}
 {{/if}}
 

packages/kubernetes/data_stream/container_logs/elasticsearch/ingest_pipeline/default.yml

@@ -0,0 +1,23 @@
+---
+description: Pipeline for Kubernetes container logs
+processors:
+  - set:
+      field: service.name
+      copy_from: kubernetes.labels.app_kubernetes_io/name
+      ignore_empty_value: true
+  - set:
+      field: service.name
+      copy_from: kubernetes.container.name
+      override: false
+      ignore_empty_value: true
+  - set:
+      field: service.version
+      copy_from: kubernetes.labels.app_kubernetes_io/version
+      ignore_empty_value: true
+on_failure:
+  - set:
+      field: event.kind
+      value: pipeline_error
+  - append:
+      field: error.message
+      value: '{{{ _ingest.on_failure_message }}}'

packages/kubernetes/data_stream/container_logs/fields/ecs.yml

@@ -22,3 +22,7 @@
   name: orchestrator.cluster.name
 - external: ecs
   name: orchestrator.cluster.url
+- external: ecs
+  name: service.name
+- external: ecs
+  name: service.version

packages/kubernetes/data_stream/container_logs/manifest.yml

@@ -1,5 +1,6 @@
 title: "Kubernetes container logs"
 type: logs
+dataset: kubernetes.container_logs
 streams:
   - input: filestream
     title: Collect Kubernetes container logs

packages/kubernetes/data_stream/container_logs/routing_rules.yml

@@ -0,0 +1,11 @@
+# Route container logs events to the correct dataset and namespace
+# based on pod annotations.
+- source_dataset: kubernetes.container_logs
+  rules:
+    - target_dataset:
+        - "{{kubernetes.annotations.elastic_co/dataset}}"
+        - "{{data_stream.dataset}}"
+      namespace:
+        - "{{kubernetes.annotations.elastic_co/namespace}}"
+        - "{{data_stream.namespace}}"
+      if: "ctx.kubernetes?.annotations != null"

packages/kubernetes/docs/README.md

@@ -84,6 +84,14 @@ the masters won't be visible. In these cases it won't be possible to use `schedu
 The container-logs dataset requires access to the log files in each Kubernetes node where the container logs are stored.
 This defaults to `/var/log/containers/*${kubernetes.container.id}.log`.
 
+#### Routing
+
+The container-logs data stream allows routing logs to a different *dataset* or *namespace* using pod annotations. 
+
+For example, suppose you are running Nginx on your Kubernetes cluster, and you want to drive the Nginx container logs into a dedicated dataset or namespace. By annotating the pod with `elastic.co/namespace: nginx`, the integration will send all the container logs to the `nginx` namespace.
+
+To learn more about routing container-logs, see https://docs.elastic.co/integrations/kubernetes/container-logs.
+
 ### audit-logs
 
 The audit-logs dataset requires access to the log files on each Kubernetes node where the audit logs are stored.