feat: Adds k8sObjects event receiver under feature flag (#4127)

By default helm chart uses raw_k8s_events receiver. The feature flag has to be set
to false in order to use k8sObjects receiver for fetching events.

Signed-off-by: Dhruv Shah <dhruv.shah@sumologic.com>

Author: dhruv-shah-sumo

.changelog/4127.added.txt

@@ -0,0 +1 @@
+feat: Adds k8sObjects event receiver under feature flag

deploy/helm/sumologic/README.md

@@ -60,6 +60,39 @@ processors:
           - set(time, Now()) where time_unix_nano == 0
           - set(attributes["timestamp"], Int(time_unix_nano / 1000000))
 
+{{- if not .Values.otelevents.useSumoK8sEventReceiver }}
+  ## Reformat body to align k8sobjects payload format with the older raw_k8s_events format
+  ## so that existing event queries continue to work
+  transform/reformat_body:
+    log_statements:
+      - context: log
+        statements:
+          # Copy 'object' from body to attributes
+          - set(attributes["object"], body["object"])
+          # Copy 'type' from body to attributes
+          - set(attributes["type"], body["type"])
+          # Lift 'message' out of 'object' and put it in 'body'
+          - set(body, attributes["object"]["message"])
+
+  ## Drop DELETED and other non-watch events (keep only ADDED and MODIFIED)
+  filter/watch_types:
+    logs:
+      log_record:
+        - attributes["type"] != "ADDED" and attributes["type"] != "MODIFIED"
+
+  ## Populate severity_text and severity_number from the event type field
+  transform/set_severity:
+    log_statements:
+      - context: log
+        statements:
+          # Warning → WARN (13)
+          - set(severity_text, "Warning") where attributes["object"]["type"] == "Warning"
+          - set(severity_number, 13) where attributes["object"]["type"] == "Warning"
+          # Normal → INFO (9)
+          - set(severity_text, "Normal") where attributes["object"]["type"] == "Normal"
+          - set(severity_number, 9) where attributes["object"]["type"] == "Normal"
+{{- end }}
+
 {{- if .Values.otelevents.reportConfigMaps }}
   ## Extract configmap name and chart label from body
   transform/extract_configmap_name:
@@ -91,7 +124,15 @@ receivers:
         mode: pull
         interval: 24h
 {{- end }}
+{{- if .Values.otelevents.useSumoK8sEventReceiver }}
   raw_k8s_events: {}
+{{- else }}
+  k8sobjects/events:
+    auth_type: serviceAccount
+    objects:
+      - name: events
+        mode: watch
+{{- end }}
 
 service:
   extensions:
@@ -132,13 +173,22 @@ service:
         - sumologic/sumologic-mock
 {{- end }}
       processors:
+{{- if not .Values.otelevents.useSumoK8sEventReceiver }}
+        - transform/reformat_body
+        - filter/watch_types
+        - transform/set_severity
+{{- end }}
         - resource/add_cluster
         - source
         - sumologic
         - transform/add_timestamp
         - batch
       receivers:
+{{- if .Values.otelevents.useSumoK8sEventReceiver }}
         - raw_k8s_events
+{{- else }}
+        - k8sobjects/events
+{{- end }}
 
 
   telemetry:

deploy/helm/sumologic/conf/events/otelcol/config.yaml

@@ -2071,6 +2071,11 @@ otelevents:
   ## Metrics Level (basic, normal, detailed)
   metricsLevel: normal
 
+  ## When true (default), use the in-house raw_k8s_events receiver for Kubernetes event collection.
+  ## When false, use the upstream k8sobjects receiver with additional transform and filter processors.
+  ## Note: k8sobjects receiver does not de-duplicate events on collector restarts.
+  useSumoK8sEventReceiver: true
+
   ## Enable reporting of ConfigMaps. Only Sumologic collection configmaps are reported for now.
   reportConfigMaps: true
 

deploy/helm/sumologic/values.yaml

@@ -0,0 +1,3 @@
+otelevents:
+  useSumoK8sEventReceiver: false
+  reportConfigMaps: false

tests/helm/testdata/goldenfile/events_otc/k8sobjects-events.input.yaml

@@ -0,0 +1,112 @@
+---
+# Source: sumologic/templates/events/otelcol/configmap.yaml
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: RELEASE-NAME-sumologic-otelcol-events
+  namespace: sumologic
+  labels:
+    app: RELEASE-NAME-sumologic-otelcol-events
+    chart: "sumologic-%CURRENT_CHART_VERSION%"
+    release: "RELEASE-NAME"
+    heritage: "Helm"
+data:
+  config.yaml: |
+    exporters:
+      sumologic:
+        client: k8s_%CURRENT_CHART_VERSION%
+        endpoint: ${SUMO_ENDPOINT_DEFAULT_OTLP_EVENTS_SOURCE}
+        log_format: otlp
+        sending_queue:
+          enabled: true
+          storage: file_storage
+    extensions:
+      file_storage:
+        directory: /var/lib/storage/events
+        timeout: 10s
+      health_check:
+        endpoint: ${env:MY_POD_IP}:13133
+      pprof: {}
+    processors:
+      batch:
+        send_batch_max_size: 2048
+        send_batch_size: 1024
+        timeout: 1s
+      filter/watch_types:
+        logs:
+          log_record:
+          - attributes["type"] != "ADDED" and attributes["type"] != "MODIFIED"
+      resource/add_cluster:
+        attributes:
+        - action: upsert
+          key: cluster
+          value: kubernetes
+      source:
+        collector: kubernetes
+        source_category: kubernetes/events
+        source_category_prefix: ""
+        source_category_replace_dash: /
+        source_name: events
+      sumologic:
+        add_cloud_namespace: false
+      transform/add_timestamp:
+        log_statements:
+        - context: log
+          statements:
+          - set(time, Now()) where time_unix_nano == 0
+          - set(attributes["timestamp"], Int(time_unix_nano / 1000000))
+      transform/reformat_body:
+        log_statements:
+        - context: log
+          statements:
+          - set(attributes["object"], body["object"])
+          - set(attributes["type"], body["type"])
+          - set(body, attributes["object"]["message"])
+      transform/set_severity:
+        log_statements:
+        - context: log
+          statements:
+          - set(severity_text, "Warning") where attributes["object"]["type"] == "Warning"
+          - set(severity_number, 13) where attributes["object"]["type"] == "Warning"
+          - set(severity_text, "Normal") where attributes["object"]["type"] == "Normal"
+          - set(severity_number, 9) where attributes["object"]["type"] == "Normal"
+    receivers:
+      k8sobjects/events:
+        auth_type: serviceAccount
+        objects:
+        - mode: watch
+          name: events
+    service:
+      extensions:
+      - health_check
+      - file_storage
+      - pprof
+      pipelines:
+        logs/events:
+          exporters:
+          - sumologic
+          processors:
+          - transform/reformat_body
+          - filter/watch_types
+          - transform/set_severity
+          - resource/add_cluster
+          - source
+          - sumologic
+          - transform/add_timestamp
+          - batch
+          receivers:
+          - k8sobjects/events
+      telemetry:
+        logs:
+          level: info
+        metrics:
+          level: normal
+          readers:
+          - pull:
+              exporter:
+                prometheus:
+                  host: ${env:MY_POD_IP}
+                  port: 8888
+                  without_scope_info: true
+                  without_type_suffix: true
+                  without_units: true