feat: support extra pipeline processors in events otel collector

observability-events-otel-collector/README.md

@@ -26,9 +26,10 @@ Enriched keys land on the event's resource attributes:
 | Annotations | `k8s.object.annotation.<key>`      |
 | Owner ref   | `k8s.object.owner.{kind,name,uid}` |
 
-The pipeline is `k8s_events → k8seventenrich → batch → exporter`. See
-[`k8seventenrichprocessor/README.md`](k8seventenrichprocessor/README.md) for the
-full processor reference.
+The default pipeline is `k8s_events → k8seventenrich → batch → exporter`; extra
+processors can be inserted into it — see
+[Customizing the pipeline](#customizing-the-pipeline). For the full event enrichment processor
+reference, see [`k8seventenrichprocessor/README.md`](k8seventenrichprocessor/README.md).
 
 ## Prerequisites
 
@@ -47,7 +48,7 @@ helm upgrade --install observability-events-otel-collector \
   oci://ghcr.io/openchoreo/helm-charts/observability-events-otel-collector \
   --create-namespace \
   --namespace openchoreo-observability-plane \
-  --version 0.1.1
+  --version 0.2.0
 ```
 
 Out of the box the collector enriches events and prints them to its pod log via
@@ -72,7 +73,7 @@ Compatible with `observability-logs-opensearch` community module (>= version 0.6
 ```bash
 helm upgrade --install observability-events-otel-collector \
   oci://ghcr.io/openchoreo/helm-charts/observability-events-otel-collector \
-  --namespace openchoreo-observability-plane --version 0.1.1 \
+  --namespace openchoreo-observability-plane --version 0.2.0 \
   -f - <<'EOF'
 collector:
   extraEnv:
@@ -113,7 +114,7 @@ Compatible with `observability-logs-openobserve` community module (>= version 0.
 ```bash
 helm upgrade --install observability-events-otel-collector \
   oci://ghcr.io/openchoreo/helm-charts/observability-events-otel-collector \
-  --namespace openchoreo-observability-plane --create-namespace --version 0.1.1 \
+  --namespace openchoreo-observability-plane --create-namespace --version 0.2.0 \
   -f - <<'EOF'
 collector:
   extraEnv:
@@ -166,6 +167,101 @@ pipelineExporters:
 > If you need a pipeline the structured values don't cover, set `configOverride` to a
 > raw collector config and it replaces the rendered one entirely.
 
+## Customizing the pipeline
+
+Extra processors go in `extraProcessors` (the definitions) and are ordered by
+`pipelineProcessors` (the chain), mirroring `exporters` / `pipelineExporters`.
+
+The distribution is a curated build with the following upstream processor types.
+They can be used alongside the built-in `k8seventenrich` and `batch`:
+
+| Type       | Use                                                                                                                       |
+| ---------- | ------------------------------------------------------------------------------------------------------------------------- |
+| `resource` | Add/rewrite **resource** attributes — stamp origin metadata on every event.                                               |
+| `filter`   | Drop events matching an [OTTL](https://opentelemetry.io/docs/collector/transforming-telemetry/) condition, before export. |
+
+Order matters. **`batch` must be last** (if included) so batching happens
+after all other processing — the chart fails the render if it isn't.
+**`k8seventenrich` is required** — the render fails without it — and must come
+**before** anything that reads its output (`k8s.object.label.*`,
+`k8s.object.annotation.*`, `k8s.object.owner.*`).
+
+> `pipelineProcessors` is a full **replacement** list, not a merge. To add one
+> processor you must re-list `k8seventenrich` (mandatory) and `batch` (if needed).
+
+### Stamping origin attributes (e.g. multi-cluster / multi-plane)
+
+When several collectors fan into one backend, events need an origin attribute to
+stay distinguishable at query time. Env values come from `collector.extraEnv` —
+no separate mechanism is needed for `${env:...}`:
+
+```bash
+helm upgrade --install observability-events-otel-collector \
+  oci://ghcr.io/openchoreo/helm-charts/observability-events-otel-collector \
+  --namespace openchoreo-observability-plane --version 0.2.0 \
+  -f - <<'EOF'
+collector:
+  extraEnv:
+    - name: REGION
+      value: us-east-1
+    - name: PLANE_KIND
+      value: dataplane
+    - name: PLANE_NAME
+      value: prod
+
+extraProcessors:
+  resource/cluster_identity:
+    attributes:
+      - { key: cloud.region, value: "${env:REGION}", action: upsert }
+      - { key: openchoreo.plane_kind, value: "${env:PLANE_KIND}", action: upsert }
+      - { key: openchoreo.plane_name, value: "${env:PLANE_NAME}", action: upsert }
+
+pipelineProcessors:
+  - k8seventenrich
+  - resource/cluster_identity
+  - batch
+EOF
+```
+
+### Dropping noisy events
+
+Events are high-volume; `filter` trims them before they cost anything downstream.
+Conditions are OTTL and drop a record when they evaluate **true**:
+
+```yaml
+extraProcessors:
+  filter/drop_kube_system:
+    error_mode: ignore
+    logs:
+      log_record:
+        - 'attributes["k8s.namespace.name"] == "kube-system"'
+
+pipelineProcessors:
+  - k8seventenrich
+  - filter/drop_kube_system
+  - batch
+```
+
+Filter on the event's own fields with `k8s.event.reason` /
+`k8s.event.reporting_controller`, or on the enriched resource attributes that
+`k8seventenrich` adds. Note the receiver emits no event-type _attribute_ — the
+event's `Normal` / `Warning` type is carried as the log record's severity. To keep
+only warnings and above:
+
+```yaml
+extraProcessors:
+  filter/warnings_only:
+    error_mode: ignore
+    logs:
+      log_record:
+        - 'severity_text == "Normal"'
+
+pipelineProcessors:
+  - k8seventenrich
+  - filter/warnings_only
+  - batch
+```
+
 ## Tuning enrichment
 
 The `enrichment` value maps 1:1 to the `k8seventenrich` processor config.
@@ -213,7 +309,7 @@ persistence:
 ```bash
 helm upgrade --install observability-events-otel-collector \
   oci://ghcr.io/openchoreo/helm-charts/observability-events-otel-collector \
-  --namespace openchoreo-observability-plane --version 0.1.1 --reuse-values \
+  --namespace openchoreo-observability-plane --version 0.2.0 --reuse-values \
   --set persistence.enabled=true \
   --set persistence.storageClassName=<your-storage-class>
 ```

observability-events-otel-collector/VERSION

@@ -1 +1 @@
-0.1.2
+0.2.0

observability-events-otel-collector/builder-config.yaml

@@ -16,6 +16,11 @@ receivers:
 
 processors:
   - gomod: go.opentelemetry.io/collector/processor/batchprocessor v0.153.0
+  # Resource-attribute edits (insert/update/upsert/delete). Lets operators stamp origin
+  # attributes (region, plane, cluster) onto every event via `extraProcessors`.
+  - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/resourceprocessor v0.153.0
+  # Drops noisy events (by namespace, reason, severity) before they reach the exporter.
+  - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor v0.153.0
   # The custom processor that adds Kubernetes metadata to events emitted by the k8seventsreceiver.
   - gomod: github.com/openchoreo/community-modules/observability-events-otel-collector/k8seventenrichprocessor v0.1.0
     path: ./k8seventenrichprocessor

observability-events-otel-collector/helm/templates/NOTES.txt

@@ -3,6 +3,12 @@
 The collector is now watching Kubernetes events cluster-wide and enriching them
 with the involved object's labels, annotations, and owner reference.
 
+{{ if .Values.configOverride -}}
+Pipeline: defined by `configOverride` — the structured values are not in effect.
+{{- else -}}
+Pipeline: k8s_events -> {{ join " -> " .Values.pipelineProcessors }} -> {{ join ", " .Values.pipelineExporters }}
+{{- end }}
+
 Watch it work:
   kubectl -n {{ .Release.Namespace }} rollout status deploy/{{ include "events-collector.fullname" . }}
   kubectl -n {{ .Release.Namespace }} logs -f deploy/{{ include "events-collector.fullname" . }}

observability-events-otel-collector/helm/templates/configmap.yaml

@@ -13,11 +13,43 @@ data:
     {{- if .Values.configOverride }}
     {{- toYaml .Values.configOverride | nindent 4 }}
     {{- else }}
+    {{- /*
+      Helm's coalesce drops keys the user set to null (e.g. a commented-out body), so
+      these maps/lists can arrive as an untyped nil. `hasKey`, `len` and `uniq` all
+      error on that, while `range`, `has` and `join` tolerate it — normalise up front
+      so every guard below reports its own message instead of a reflection error.
+    */}}
+    {{- $exporters := .Values.exporters | default dict }}
+    {{- $extraProcessors := .Values.extraProcessors | default dict }}
+    {{- $pipelineProcessors := .Values.pipelineProcessors | default list }}
+    {{- $builtinProcessors := list "k8seventenrich" "batch" }}
     {{- range $e := .Values.pipelineExporters }}
-    {{- if not (hasKey $.Values.exporters $e) }}
+    {{- if not (hasKey $exporters $e) }}
     {{- fail (printf "pipelineExporters references %q, which is not defined under .Values.exporters" $e) }}
     {{- end }}
     {{- end }}
+    {{- if not $pipelineProcessors }}
+    {{- fail "pipelineProcessors must not be empty: it is a full replacement list, so list the processors you want, e.g. [k8seventenrich, batch]" }}
+    {{- end }}
+    {{- range $p := $pipelineProcessors }}
+    {{- if and (not (has $p $builtinProcessors)) (not (hasKey $extraProcessors $p)) }}
+    {{- fail (printf "pipelineProcessors references %q, which is neither a built-in processor (k8seventenrich, batch) nor defined under .Values.extraProcessors" $p) }}
+    {{- end }}
+    {{- end }}
+    {{- range $name, $cfg := $extraProcessors }}
+    {{- if has $name $builtinProcessors }}
+    {{- fail (printf "extraProcessors defines %q, which is a reserved built-in processor name; the rendered config would carry the key twice and fail to parse" $name) }}
+    {{- end }}
+    {{- end }}
+    {{- if not (has "k8seventenrich" $pipelineProcessors) }}
+    {{- fail "pipelineProcessors must include \"k8seventenrich\": without it events are not enriched, which is this module's whole purpose. Use configOverride if you need a fully custom pipeline" }}
+    {{- end }}
+    {{- if ne (len $pipelineProcessors) (len (uniq $pipelineProcessors)) }}
+    {{- fail "pipelineProcessors contains duplicate entries: the collector rejects a processor referenced more than once in a pipeline" }}
+    {{- end }}
+    {{- if and (has "batch" $pipelineProcessors) (ne (last $pipelineProcessors | toString) "batch") }}
+    {{- fail "pipelineProcessors: \"batch\" must be the last processor in the chain, so batching happens after all other processing" }}
+    {{- end }}
     extensions:
       health_check:
         endpoint: 0.0.0.0:13133
@@ -41,9 +73,15 @@ data:
       k8seventenrich:
         {{- toYaml .Values.enrichment | nindent 8 }}
       batch: {}
+      {{- range $name, $cfg := $extraProcessors }}
+      {{- if has $name $pipelineProcessors }}
+      {{ $name }}:
+        {{- toYaml $cfg | nindent 8 }}
+      {{- end }}
+      {{- end }}
 
     exporters:
-      {{- range $name, $cfg := .Values.exporters }}
+      {{- range $name, $cfg := $exporters }}
       {{- if has $name $.Values.pipelineExporters }}
       {{ $name }}:
         {{- toYaml $cfg | nindent 8 }}
@@ -58,6 +96,6 @@ data:
       pipelines:
         logs:
           receivers: [k8s_events]
-          processors: [k8seventenrich, batch]
+          processors: [{{ join ", " $pipelineProcessors }}]
           exporters: [{{ join ", " .Values.pipelineExporters }}]
     {{- end }}

observability-events-otel-collector/helm/values.yaml

@@ -106,6 +106,33 @@ enrichment:
   owner_references:
     enabled: true
 
+# Extra collector processors, keyed by processor ID (`<type>` or `<type>/<name>`).
+# Only types compiled into this distribution can be used: `resource` and `filter`,
+# alongside the built-in `k8seventenrich` and `batch`. A definition here is rendered
+# only when its key also appears in `pipelineProcessors`.
+# Values may reference env vars supplied via `collector.extraEnv` (`${env:...}`).
+extraProcessors: {}
+#   resource/cluster_identity:
+#     attributes:
+#       - { key: cloud.region, value: "${env:REGION}", action: upsert }
+#       - { key: openchoreo.plane_kind, value: "${env:PLANE_KIND}", action: upsert }
+#       - { key: openchoreo.plane_name, value: "${env:PLANE_NAME}", action: upsert }
+#   filter/drop_kube_system:
+#     error_mode: ignore
+#     logs:
+#       log_record:
+#         - 'attributes["k8s.namespace.name"] == "kube-system"'
+
+# Processors wired into the logs pipeline, in order.
+# NOTE: this is a full REPLACEMENT list, not a merge. To add one processor you must
+# re-list `k8seventenrich` and `batch` as well, or you silently lose them.
+# `k8seventenrich` is required and must run before anything that reads its output
+# (`k8s.object.label.*`, `k8s.object.annotation.*`, `k8s.object.owner.*`). `batch` is
+# optional, but if listed it must be last so batching happens after all other processing.
+pipelineProcessors:
+  - k8seventenrich
+  - batch
+
 # Exporters and pipelines configurations.
 # By default the chart ships ONLY the `debug` exporter, which prints enriched
 # events to the pod log. Point at a real backend by overriding `exporters` and