feat(container-insights): Add K8s events pipeline to cluster-scraper

Add a logs pipeline on the cluster-scraper that watches Kubernetes
events via the k8s_events receiver and exports them through the OTLP
exporter to CloudWatch Logs. Events are enriched with pod labels,
workload identity, node metadata via nodemetadataenricher, and
cluster-level attributes.

Changes:
- New k8s_events/cw_k8s_ci_v0 receiver (gated by events.enabled)
- Transform processors for parsing involvedObject fields, setting
  workload, scope, and cloud_resource_id (all using log_statements)
- Shared processors (k8sattributes, k8snodemetadata) re-gated to be
  available when either KSM or events is enabled
- RBAC: added events to cluster-scraper ClusterRole
- values.yaml: otelContainerInsights.events.enabled (default: true)
- Integration test assertions for k8s_events routing

Author: miconeilaws

charts/amazon-cloudwatch-observability/templates/cloudwatch-agent-clusterrole.yaml

@@ -26,7 +26,7 @@ rules:
   verbs: [ "list", "watch", "get" ]
 - apiGroups: [ "" ]
   resources: [ "nodes/stats", "events" ]
-  verbs: [ "create", "get" ]
+  verbs: [ "create", "get", "list", "watch" ]
 {{- if .Values.otelContainerInsights.enabled }}
 - apiGroups: [ "" ]
   resources: [ "nodes/metrics" ]

charts/amazon-cloudwatch-observability/templates/linux/_otel-container-insights-cluster-scraper-config.tpl

@@ -3,6 +3,11 @@ extensions:
   sigv4auth/cw_k8s_ci_v0_cwotel:
     region: {{ .Values.region }}
     service: monitoring
+{{- if .Values.otelContainerInsights.events.enabled }}
+  sigv4auth/cw_k8s_ci_v0_cwlogs:
+    region: {{ .Values.region }}
+    service: logs
+{{- end }}
   nodemetadatacache/cw_k8s_ci_v0:
     namespace: {{ .Release.Namespace }}
 
@@ -48,6 +53,12 @@ receivers:
                 - {{ include "kube-state-metrics.name" . }}.{{ .Release.Namespace }}.svc:{{ .Values.kubeStateMetrics.service.port }}
 {{- end }}
 
+{{- if .Values.otelContainerInsights.events.enabled }}
+  k8s_events/cw_k8s_ci_v0:
+    auth_type: serviceAccount
+    namespaces: []
+{{- end }}
+
 processors:
   filter/cw_k8s_ci_v0_scrape_metadata:
     error_mode: ignore
@@ -216,7 +227,9 @@ processors:
           - set(attributes["cronjob"], resource.attributes["cronjob"]) where resource.attributes["cronjob"] != nil
           - set(attributes["owner_name"], resource.attributes["owner_name"]) where resource.attributes["owner_name"] != nil
           - set(attributes["owner_kind"], resource.attributes["owner_kind"]) where resource.attributes["owner_kind"] != nil
+{{- end }}
 
+{{- if or .Values.kubeStateMetrics.enabled .Values.otelContainerInsights.events.enabled }}
   k8sattributes/cw_k8s_ci_v0_pod:
     auth_type: serviceAccount
     passthrough: false
@@ -348,17 +361,87 @@ processors:
     send_batch_max_size: 500
     timeout: 10s
 
+{{- if .Values.otelContainerInsights.events.enabled }}
+  transform/cw_k8s_ci_v0_events_parse:
+    error_mode: ignore
+    log_statements:
+      - context: log
+        statements:
+          - set(resource.attributes["k8s.namespace.name"], attributes["k8s.namespace.name"]) where attributes["k8s.namespace.name"] != nil
+          - set(resource.attributes["k8s.pod.name"], resource.attributes["k8s.object.name"]) where resource.attributes["k8s.object.kind"] == "Pod" and resource.attributes["k8s.object.name"] != nil
+
+  transform/cw_k8s_ci_v0_events_set_cluster_name:
+    error_mode: ignore
+    log_statements:
+      - context: log
+        statements:
+          - set(resource.attributes["k8s.cluster.name"], "{{ .Values.clusterName }}")
+          - set(resource.attributes["aws.log.group.names"], "/aws/containerinsights/{{ .Values.clusterName }}/events")
+
+  transform/cw_k8s_ci_v0_events_set_workload:
+    error_mode: ignore
+    log_statements:
+      - context: log
+        statements:
+          - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.deployment.name"]) where resource.attributes["k8s.workload.name"] == nil and resource.attributes["k8s.deployment.name"] != nil
+          - set(resource.attributes["k8s.workload.type"], "Deployment") where resource.attributes["k8s.workload.type"] == nil and resource.attributes["k8s.deployment.name"] != nil
+          - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.statefulset.name"]) where resource.attributes["k8s.workload.name"] == nil and resource.attributes["k8s.statefulset.name"] != nil
+          - set(resource.attributes["k8s.workload.type"], "StatefulSet") where resource.attributes["k8s.workload.type"] == nil and resource.attributes["k8s.statefulset.name"] != nil
+          - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.daemonset.name"]) where resource.attributes["k8s.workload.name"] == nil and resource.attributes["k8s.daemonset.name"] != nil
+          - set(resource.attributes["k8s.workload.type"], "DaemonSet") where resource.attributes["k8s.workload.type"] == nil and resource.attributes["k8s.daemonset.name"] != nil
+          - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.job.name"]) where resource.attributes["k8s.workload.name"] == nil and resource.attributes["k8s.job.name"] != nil
+          - set(resource.attributes["k8s.workload.type"], "Job") where resource.attributes["k8s.workload.type"] == nil and resource.attributes["k8s.job.name"] != nil
+          - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.cronjob.name"]) where resource.attributes["k8s.workload.name"] == nil and resource.attributes["k8s.cronjob.name"] != nil
+          - set(resource.attributes["k8s.workload.type"], "CronJob") where resource.attributes["k8s.workload.type"] == nil and resource.attributes["k8s.cronjob.name"] != nil
+          - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.replicaset.name"]) where resource.attributes["k8s.workload.name"] == nil and resource.attributes["k8s.replicaset.name"] != nil
+          - set(resource.attributes["k8s.workload.type"], "ReplicaSet") where resource.attributes["k8s.workload.type"] == nil and resource.attributes["k8s.replicaset.name"] != nil
+
+  transform/cw_k8s_ci_v0_events_set_scope:
+    error_mode: ignore
+    log_statements:
+      - context: scope
+        statements:
+          - set(scope.name, "k8s.io/events")
+          - set(attributes["cloudwatch.source"], "cloudwatch-agent")
+          - set(attributes["cloudwatch.solution"], "k8s-otel-container-insights")
+          - set(attributes["cloudwatch.pipeline"], "events")
+
+  transform/cw_k8s_ci_v0_events_set_cloud_resource_id:
+    error_mode: ignore
+    log_statements:
+      - context: resource
+        statements:
+          - set(resource.attributes["cloud.resource_id"], Concat(["arn:aws:eks:", resource.attributes["cloud.region"], ":", resource.attributes["cloud.account.id"], ":cluster/", resource.attributes["k8s.cluster.name"]], ""))
+            where resource.attributes["cloud.region"] != nil and resource.attributes["cloud.account.id"] != nil and resource.attributes["k8s.cluster.name"] != nil
+{{- end }}
+
 exporters:
   otlphttp/cw_k8s_ci_v0_cwotel:
     endpoint: {{ if .Values.otelContainerInsights.cloudwatchMetricsEndpoint }}{{ .Values.otelContainerInsights.cloudwatchMetricsEndpoint | quote }}{{ else }}"https://monitoring.{{ .Values.region }}.amazonaws.com:443"{{ end }}
     tls:
       insecure: false
     auth:
       authenticator: sigv4auth/cw_k8s_ci_v0_cwotel
+{{- if .Values.otelContainerInsights.events.enabled }}
+  otlphttp/cw_k8s_ci_v0_cwlogs:
+    endpoint: "https://logs.{{ .Values.region }}.amazonaws.com:443"
+    headers:
+      x-aws-log-group: "/aws/containerinsights/{{ .Values.clusterName }}/events"
+      x-aws-log-stream: "events"
+      x-aws-log-group-create: "true"
+      x-aws-log-stream-create: "true"
+    tls:
+      insecure: false
+    auth:
+      authenticator: sigv4auth/cw_k8s_ci_v0_cwlogs
+{{- end }}
 
 service:
   extensions:
     - sigv4auth/cw_k8s_ci_v0_cwotel
+{{- if .Values.otelContainerInsights.events.enabled }}
+    - sigv4auth/cw_k8s_ci_v0_cwlogs
+{{- end }}
     - nodemetadatacache/cw_k8s_ci_v0
   pipelines:
     metrics/cw_k8s_ci_v0_apiserver:
@@ -405,5 +488,22 @@ service:
       exporters:
         - otlphttp/cw_k8s_ci_v0_cwotel
 {{- end }}
+{{- if .Values.otelContainerInsights.events.enabled }}
+    logs/cw_k8s_ci_v0_events:
+      receivers: [k8s_events/cw_k8s_ci_v0]
+      processors:
+        - transform/cw_k8s_ci_v0_events_parse
+        - transform/cw_k8s_ci_v0_events_set_cluster_name
+        - k8sattributes/cw_k8s_ci_v0_pod
+        - k8sattributes/cw_k8s_ci_v0_node
+        - transform/cw_k8s_ci_v0_events_set_workload
+        - resourcedetection/cw_k8s_ci_v0
+        - nodemetadataenricher/cw_k8s_ci_v0
+        - transform/cw_k8s_ci_v0_events_set_scope
+        - transform/cw_k8s_ci_v0_events_set_cloud_resource_id
+        - batch/cw_k8s_ci_v0_cwotel
+      exporters:
+        - otlphttp/cw_k8s_ci_v0_cwlogs
+{{- end }}
 {{- end -}}
 

charts/amazon-cloudwatch-observability/values.yaml

@@ -1135,3 +1135,6 @@ otelContainerInsights:
   metricResolution: "30s"
   # Override the CloudWatch Metrics OTLP endpoint. If empty, defaults to monitoring.<region>.amazonaws.com.
   cloudwatchMetricsEndpoint: ""
+  ## Kubernetes Events collection. Events are collected on the cluster-scraper agent and exported as logs.
+  events:
+    enabled: true

integration-tests/amazon-cloudwatch-observability/validations/minikube/scenarios/feature_targeted_default_test.go

@@ -122,6 +122,8 @@ func validateOTELConfigRouting(t *testing.T, agentMap map[string]unstructured.Un
 			"cloudwatch-agent otelConfig should NOT contain apiserver receiver (cluster-level)")
 		assert.False(t, strings.Contains(otelConfig, "cw_k8s_ci_v0_kube_state_metrics"),
 			"cloudwatch-agent otelConfig should NOT contain kube_state_metrics receiver (cluster-level)")
+		assert.False(t, strings.Contains(otelConfig, "k8s_events"),
+			"cloudwatch-agent otelConfig should NOT contain k8s_events receiver (cluster-level)")
 	})
 
 	// Validate cluster-scraper has cluster-level OTEL config
@@ -147,6 +149,8 @@ func validateOTELConfigRouting(t *testing.T, agentMap map[string]unstructured.Un
 			"cluster-scraper otelConfig should contain apiserver receiver (cluster-level)")
 		assert.True(t, strings.Contains(otelConfig, "cw_k8s_ci_v0_kube_state_metrics"),
 			"cluster-scraper otelConfig should contain kube_state_metrics receiver (cluster-level)")
+		assert.True(t, strings.Contains(otelConfig, "k8s_events"),
+			"cluster-scraper otelConfig should contain k8s_events receiver")
 
 		// Cluster-level config should NOT contain node-level receivers
 		assert.False(t, strings.Contains(otelConfig, "kubeletstats"),

integration-tests/amazon-cloudwatch-observability/validations/minikube/scenarios/feature_targeted_multi_agent_test.go

@@ -134,6 +134,8 @@ func validateCloudWatchAgentFullConfig(t *testing.T, agentMap map[string]unstruc
 		"cloudwatch-agent otelConfig should contain kubeletstats receiver (node-level)")
 	assert.False(t, strings.Contains(otelConfig, "cw_k8s_ci_v0_apiserver"),
 		"cloudwatch-agent otelConfig should NOT contain apiserver receiver (cluster-level)")
+	assert.False(t, strings.Contains(otelConfig, "k8s_events"),
+		"cloudwatch-agent otelConfig should NOT contain k8s_events receiver (cluster-level)")
 }
 
 // validatePrometheusAgentMinimalConfig verifies prometheus-agent gets minimal config:
@@ -221,6 +223,8 @@ func validateClusterScraperConfig(t *testing.T, agentMap map[string]unstructured
 		"cluster-scraper otelConfig should contain apiserver receiver (cluster-level)")
 	assert.True(t, strings.Contains(otelConfig, "cw_k8s_ci_v0_kube_state_metrics"),
 		"cluster-scraper otelConfig should contain kube_state_metrics receiver (cluster-level)")
+	assert.True(t, strings.Contains(otelConfig, "k8s_events"),
+		"cluster-scraper otelConfig should contain k8s_events receiver")
 	assert.False(t, strings.Contains(otelConfig, "kubeletstats"),
 		"cluster-scraper otelConfig should NOT contain kubeletstats receiver (node-level)")
 }

integration-tests/amazon-cloudwatch-observability/validations/minikube/scenarios/feature_targeted_split_features_test.go

@@ -238,6 +238,8 @@ func validateSplitFeaturesClusterScraperConfig(t *testing.T, agentMap map[string
 		"cluster-scraper otelConfig should contain apiserver receiver (cluster-level)")
 	assert.True(t, strings.Contains(otelConfig, "cw_k8s_ci_v0_kube_state_metrics"),
 		"cluster-scraper otelConfig should contain kube_state_metrics receiver (cluster-level)")
+	assert.True(t, strings.Contains(otelConfig, "k8s_events"),
+		"cluster-scraper otelConfig should contain k8s_events receiver")
 	assert.False(t, strings.Contains(otelConfig, "kubeletstats"),
 		"cluster-scraper otelConfig should NOT contain kubeletstats receiver (node-level)")
 }