fix: Unify pull/watch events before observek8sattributes (#273)

The unification of events is currently defined in transform/object,
which was moved after observek8sattributes due to the Secret redaction
requirements.

However, the observek8sattributes custom processor expects the data from
pull/watch to be unified.

The solution to this is to extract the unification logic into a
different processor, which can be injected in the pipeline before
observek8sattributes.

charts/agent/Chart.yaml

@@ -2,7 +2,7 @@ apiVersion: v2
 name: agent
 description: Chart to install K8s collection stack based on Observe Agent
 type: application
-version: 0.30.5
+version: 0.30.6
 appVersion: "1.1.0"
 dependencies:
   - name: opentelemetry-collector

charts/agent/README.md

@@ -1,6 +1,6 @@
 # agent
 
-![Version: 0.30.5](https://img.shields.io/badge/Version-0.30.5-informational?style=flat-square) ![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square) ![AppVersion: 1.1.0](https://img.shields.io/badge/AppVersion-1.1.0-informational?style=flat-square)
+![Version: 0.30.6](https://img.shields.io/badge/Version-0.30.6-informational?style=flat-square) ![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square) ![AppVersion: 1.1.0](https://img.shields.io/badge/AppVersion-1.1.0-informational?style=flat-square)
 
 > [!CAUTION]
 > This chart is under active development and is not meant to be installed yet.

charts/agent/templates/_cluster-events-config.tpl

@@ -76,18 +76,24 @@ processors:
 
 {{- include "config.processors.attributes.observek8sattributes" . | nindent 2 }}
 
-  # transform for k8s objects
-  transform/object:
+  transform/unify:
     error_mode: ignore
     log_statements:
       - context: log
         statements:
-          - set(attributes["observe_filter"], "objects_pull_watch")
           # unwrapping for the object_watch stream
           - set(attributes["observe_transform"]["control"]["isDelete"], true) where body["object"] != nil and body["type"] == "DELETED"
           - set(attributes["observe_transform"]["control"]["debug_source"], "watch") where body["object"] != nil and body["type"] != nil
           - set(attributes["observe_transform"]["control"]["debug_source"], "pull") where body["object"] == nil or body["type"] == nil
           - set(body, body["object"]) where body["object"] != nil and body["type"] != nil
+
+  # transform for k8s objects
+  transform/object:
+    error_mode: ignore
+    log_statements:
+      - context: log
+        statements:
+          - set(attributes["observe_filter"], "objects_pull_watch")
           # native columns: valid_from, valid_to, kind
           - set(attributes["observe_transform"]["valid_from"], observed_time_unix_nano)
           - set(attributes["observe_transform"]["valid_to"], Int(observed_time_unix_nano) + 5400000000000)
@@ -429,7 +435,7 @@ service:
   pipelines:
       logs/objects:
         receivers: [k8sobjects/objects]
-        processors: [memory_limiter, batch, resource/observe_common, observek8sattributes, transform/object]
+        processors: [memory_limiter, batch, resource/observe_common, transform/unify, observek8sattributes, transform/object]
         exporters: [otlphttp/observe/entity, debug/override]
       logs/cluster:
         receivers: [k8sobjects/cluster]