Add k8sevent transform (#1691)

* add k8sevents transform processor

* update chlog

* add processor pod.uid

* resolve conflicts

* fix test

.chloggen/k8sevent_transform.yaml

@@ -0,0 +1,19 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'enhancement'
+# The name of the component, or a single word describing the area of concern, (e.g. agent, clusterReceiver, gateway, operator, chart, other)
+component: clusterReceiver
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: For the option, `clusterReceiver.eventsEnabled`, the logs pipeline for k8s_events now adds attributes of the type `k8s.<objectkind>.name` and `k8s.<objectkind>.uid`.
+# One or more tracking issues related to the change
+issues: [1691]
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  For example, if the log k8s event is about object type `StatefulSet`, the exported log to Splunk will have these 2 additional attributes:
+  ```
+    k8s.statefulset.name: value(k8s.object.name)
+    k8s.statefulset.uid: value(k8s.object.uid)
+  ```
+  The existing attributes `k8s.object.kind`, `k8s.object.name` and `k8s.object.uid` are still present.
+  In addition to these, if the event is for kind Pod, and the k8s.object.fieldPath has a specific container spec, the log will have an additional attribute `k8s.container.name` with the value of the container name.

examples/collector-cluster-receiver-only/rendered_manifests/configmap-cluster-receiver.yaml

@@ -47,6 +47,9 @@ data:
           - container.image.name
           - container.image.tag
         pod_association:
+        - sources:
+          - from: resource_attribute
+            name: k8s.pod.uid
         - sources:
           - from: resource_attribute
             name: k8s.namespace.name

examples/collector-cluster-receiver-only/rendered_manifests/deployment-cluster-receiver.yaml

@@ -31,7 +31,7 @@ spec:
         component: otel-k8s-cluster-receiver
         release: default
       annotations:
-        checksum/config: c1519af3592effec9d730342c24b83201af65992aa06d5dee3b88f7556369316
+        checksum/config: b805f026a08cbab958512e0705b339006fb1825dc843c8e1be17737805602151
     spec:
       serviceAccountName: default-splunk-otel-collector
       nodeSelector:

functional_tests/internal/common.go

@@ -174,3 +174,14 @@ func LoadCollectorChart(t *testing.T) *chart.Chart {
 	require.NoError(t, err)
 	return c
 }
+
+func AnnotateNamespace(t *testing.T, clientset *kubernetes.Clientset, name, key, value string) {
+	ns, err := clientset.CoreV1().Namespaces().Get(context.TODO(), name, metav1.GetOptions{})
+	require.NoError(t, err)
+	if ns.Annotations == nil {
+		ns.Annotations = make(map[string]string)
+	}
+	ns.Annotations[key] = value
+	_, err = clientset.CoreV1().Namespaces().Update(context.TODO(), ns, metav1.UpdateOptions{})
+	require.NoError(t, err)
+}

functional_tests/k8sevents/k8sevents_test.go

@@ -58,8 +58,8 @@ func Test_K8SEvents(t *testing.T) {
 	t.Run("CheckK8SEventsLogs", func(t *testing.T) {
 		actualLogs := selectResLogs("com.splunk.sourcetype", "kube:events", eventsLogsConsumer)
 		k8sEventsLogs := selectLogs(t, "k8s.namespace.name", "k8sevents-test", &actualLogs, func(body string) string {
-			re := regexp.MustCompile(`Successfully pulled image "busybox:latest" in .* \(.* including waiting\).*`)
-			return re.ReplaceAllString(body, `Successfully pulled image "busybox:latest" in <time> (<time> including waiting)`)
+			re := regexp.MustCompile(`Successfully pulled image "(busybox|alpine):latest" in .* \(.* including waiting\).*`)
+			return re.ReplaceAllString(body, `Successfully pulled image "$1:latest" in <time> (<time> including waiting)`)
 		})
 
 		// These container attributes may not get added by the k8sattributesprocessor on the events about container image pull/start
@@ -76,6 +76,7 @@ func Test_K8SEvents(t *testing.T) {
 			plogtest.IgnoreObservedTimestamp(),
 			plogtest.IgnoreResourceAttributeValue("host.name"),
 			plogtest.IgnoreLogRecordAttributeValue("k8s.object.uid"),
+			plogtest.IgnoreLogRecordAttributeValue("k8s.statefulset.uid"),
 			plogtest.IgnoreLogRecordAttributeValue("k8s.pod.uid"),
 			plogtest.IgnoreLogRecordAttributeValue("k8s.object.resource_version"),
 			plogtest.IgnoreResourceLogsOrder(),
@@ -205,6 +206,7 @@ func deployWorkloadAndCollector(t *testing.T) {
 
 	// Deploy the workload
 	internal.CreateNamespace(t, clientset, "k8sevents-test")
+	internal.AnnotateNamespace(t, clientset, "k8sevents-test", "com.splunk.index", "index_from_namespace")
 	createdObjs, err := k8stest.CreateObjects(k8sClient, "testdata/testobjects")
 	require.NoError(t, err)
 	require.NotEmpty(t, createdObjs)