From cbe843a70b1862a2b8cdbefdd9e05db547732d0f Mon Sep 17 00:00:00 2001
From: Jina Jain <jina@signalfx.com>
Date: Fri, 21 Feb 2025 08:55:04 -0800
Subject: [PATCH] add test for k8s events option (#1670)

* add test for k8s events option

* fix typo
---
 .github/workflows/functional_test_v2.yaml     |   1 +
 functional_tests/common.go                    |  63 +++
 functional_tests/istio_test.go                |  57 +--
 functional_tests/k8sevents_test.go            | 311 +++++++++++++
 .../expected_k8sevents.yaml                   | 408 ++++++++++++++++++
 .../k8sevents_values.yaml.tmpl                |  21 +
 .../testobjects/1-serviceaccount.yaml         |   5 +
 .../testdata_k8sevents/testobjects/2-svc.yaml |  12 +
 .../testdata_k8sevents/testobjects/3-sts.yaml |  25 ++
 9 files changed, 848 insertions(+), 55 deletions(-)
 create mode 100644 functional_tests/k8sevents_test.go
 create mode 100644 functional_tests/testdata_k8sevents/expected_k8sevents.yaml
 create mode 100644 functional_tests/testdata_k8sevents/k8sevents_values.yaml.tmpl
 create mode 100644 functional_tests/testdata_k8sevents/testobjects/1-serviceaccount.yaml
 create mode 100644 functional_tests/testdata_k8sevents/testobjects/2-svc.yaml
 create mode 100644 functional_tests/testdata_k8sevents/testobjects/3-sts.yaml

diff --git a/.github/workflows/functional_test_v2.yaml b/.github/workflows/functional_test_v2.yaml
index 4ed48c26c..4a76d4dd0 100644
--- a/.github/workflows/functional_test_v2.yaml
+++ b/.github/workflows/functional_test_v2.yaml
@@ -44,6 +44,7 @@ jobs:
           - functional
           - histogram
           - configuration_switching
+          - k8sevents
           - istio
     runs-on: ubuntu-latest
     steps:
diff --git a/functional_tests/common.go b/functional_tests/common.go
index 7941cef9f..6c57f4bb1 100644
--- a/functional_tests/common.go
+++ b/functional_tests/common.go
@@ -10,9 +10,13 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/signalfxreceiver"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.opentelemetry.io/collector/receiver/receivertest"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 	"os"
 	"path/filepath"
 	"runtime"
@@ -106,6 +110,11 @@ func writeNewExpectedMetricsResult(t *testing.T, file string, metric *pmetric.Me
 	require.NoError(t, golden.WriteMetrics(t, filepath.Join("results", filepath.Base(file)), *metric))
 }
 
+func writeNewExpectedLogsResult(t *testing.T, file string, log *plog.Logs) {
+	require.NoError(t, os.MkdirAll("results", 0755))
+	require.NoError(t, golden.WriteLogs(t, filepath.Join("results", filepath.Base(file)), *log))
+}
+
 func setupSignalfxReceiver(t *testing.T, port int) *consumertest.MetricsSink {
 	mc := new(consumertest.MetricsSink)
 	f := signalfxreceiver.NewFactory()
@@ -123,3 +132,57 @@ func setupSignalfxReceiver(t *testing.T, port int) *consumertest.MetricsSink {
 
 	return mc
 }
+
+func checkPodsReady(t *testing.T, clientset *kubernetes.Clientset, namespace, labelSelector string, timeout time.Duration) {
+	require.Eventually(t, func() bool {
+		pods, err := clientset.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{
+			LabelSelector: labelSelector,
+		})
+		require.NoError(t, err)
+		if len(pods.Items) == 0 {
+			return false
+		}
+		for _, pod := range pods.Items {
+			if pod.Status.Phase != v1.PodRunning {
+				return false
+			}
+			ready := false
+			for _, condition := range pod.Status.Conditions {
+				if condition.Type == v1.PodReady && condition.Status == v1.ConditionTrue {
+					ready = true
+					break
+				}
+			}
+			if !ready {
+				return false
+			}
+		}
+		return true
+	}, timeout, 5*time.Second, "Pods in namespace %s with label %s are not ready", namespace, labelSelector)
+}
+
+func createNamespace(t *testing.T, clientset *kubernetes.Clientset, name string) {
+	ns := &v1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+	}
+	_, err := clientset.CoreV1().Namespaces().Create(context.TODO(), ns, metav1.CreateOptions{})
+	require.NoError(t, err, "failed to create namespace %s", name)
+
+	require.Eventually(t, func() bool {
+		_, err := clientset.CoreV1().Namespaces().Get(context.TODO(), name, metav1.GetOptions{})
+		return err == nil
+	}, 1*time.Minute, 5*time.Second, "namespace %s is not available", name)
+}
+
+func labelNamespace(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.Labels == nil {
+		ns.Labels = make(map[string]string)
+	}
+	ns.Labels[key] = value
+	_, err = clientset.CoreV1().Namespaces().Update(context.TODO(), ns, metav1.UpdateOptions{})
+	require.NoError(t, err)
+}
diff --git a/functional_tests/istio_test.go b/functional_tests/istio_test.go
index dd988cb29..0ec4b2d4a 100644
--- a/functional_tests/istio_test.go
+++ b/functional_tests/istio_test.go
@@ -1,3 +1,5 @@
+// Copyright Splunk Inc.
+// SPDX-License-Identifier: Apache-2.0
 //go:build istio
 
 package functional_tests
@@ -29,7 +31,6 @@ import (
 	"helm.sh/helm/v3/pkg/action"
 	"helm.sh/helm/v3/pkg/chart/loader"
 	"helm.sh/helm/v3/pkg/kube"
-	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/types"
@@ -272,34 +273,6 @@ func downloadIstio(t *testing.T, version string) string {
 	return istioctlPath
 }
 
-func checkPodsReady(t *testing.T, clientset *kubernetes.Clientset, namespace, labelSelector string, timeout time.Duration) {
-	require.Eventually(t, func() bool {
-		pods, err := clientset.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{
-			LabelSelector: labelSelector,
-		})
-		require.NoError(t, err)
-		if len(pods.Items) == 0 {
-			return false
-		}
-		for _, pod := range pods.Items {
-			if pod.Status.Phase != v1.PodRunning {
-				return false
-			}
-			ready := false
-			for _, condition := range pod.Status.Conditions {
-				if condition.Type == v1.PodReady && condition.Status == v1.ConditionTrue {
-					ready = true
-					break
-				}
-			}
-			if !ready {
-				return false
-			}
-		}
-		return true
-	}, timeout, 5*time.Second, "Pods in namespace %s with label %s are not ready", namespace, labelSelector)
-}
-
 func runCommand(t *testing.T, command string) {
 	cmd := exec.Command("sh", "-c", command)
 	cmd.Stdout = os.Stdout
@@ -348,32 +321,6 @@ func createObjectFromURL(t *testing.T, config string, url string) {
 	}
 }
 
-func createNamespace(t *testing.T, clientset *kubernetes.Clientset, name string) {
-	ns := &v1.Namespace{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: name,
-		},
-	}
-	_, err := clientset.CoreV1().Namespaces().Create(context.TODO(), ns, metav1.CreateOptions{})
-	require.NoError(t, err, "failed to create namespace %s", name)
-
-	require.Eventually(t, func() bool {
-		_, err := clientset.CoreV1().Namespaces().Get(context.TODO(), name, metav1.GetOptions{})
-		return err == nil
-	}, 1*time.Minute, 5*time.Second, "namespace %s is not available", name)
-}
-
-func labelNamespace(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.Labels == nil {
-		ns.Labels = make(map[string]string)
-	}
-	ns.Labels[key] = value
-	_, err = clientset.CoreV1().Namespaces().Update(context.TODO(), ns, metav1.UpdateOptions{})
-	require.NoError(t, err)
-}
-
 func sendHTTPRequest(t *testing.T, client *http.Client, url, host, header, path string) {
 	req, err := http.NewRequest("GET", url, nil)
 	require.NoError(t, err)
diff --git a/functional_tests/k8sevents_test.go b/functional_tests/k8sevents_test.go
new file mode 100644
index 000000000..923f3d1c1
--- /dev/null
+++ b/functional_tests/k8sevents_test.go
@@ -0,0 +1,311 @@
+// Copyright Splunk Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+//go:build k8events
+
+package functional_tests
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"regexp"
+	"sync"
+	"testing"
+	"text/template"
+	"time"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/plogtest"
+	k8stest "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/xk8stest"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/splunkhecreceiver"
+	"github.com/signalfx/splunk-otel-collector-chart/functional_tests/internal"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/receiver/receivertest"
+	"helm.sh/helm/v3/pkg/action"
+	"helm.sh/helm/v3/pkg/chart/loader"
+	"helm.sh/helm/v3/pkg/kube"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/util/yaml"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/clientcmd"
+)
+
+const (
+	apiPort               = 8881
+	splunkHecReceiverPort = 8089
+)
+
+var setupRun = sync.Once{}
+var eventsLogsConsumer *consumertest.LogsSink
+
+// Env vars to control the test behavior
+// TEARDOWN_BEFORE_SETUP: if set to true, the test will run teardown before setup
+// SKIP_SETUP: if set to true, the test will skip setup
+// SKIP_TEARDOWN: if set to true, the test will skip teardown
+// SKIP_TESTS: if set to true, the test will skip the test
+// UPDATE_EXPECTED_RESULTS: if set to true, the test will update the expected results
+// KUBECONFIG: the path to the kubeconfig file
+
+func Test_K8SEvents(t *testing.T) {
+	eventsLogsConsumer := setup(t)
+	if os.Getenv("SKIP_TESTS") == "true" {
+		t.Log("Skipping tests as SKIP_TESTS is set to true")
+		return
+	}
+
+	selectedLogs := selectLogs(t, "k8s.namespace.name", "k8sevents-test", eventsLogsConsumer, func(body string) string {
+		re := regexp.MustCompile(`Successfully pulled image "busybox:latest" in \d+ms \(\d+ms including waiting\)`)
+		return re.ReplaceAllString(body, `Successfully pulled image "busybox:latest" in <time> (<time> including waiting)`)
+	})
+
+	removeFlakyLogRecordAttr(selectedLogs, "container.id") // flaky - k8sattribute processor might not have received the container id when the "started container" event is ingested
+
+	expectedLogsFile := "testdata_k8sevents/expected_k8sevents.yaml"
+	expectedLogs, err := golden.ReadLogs(expectedLogsFile)
+	require.NoError(t, err, "failed to read expected logs from file")
+
+	err = plogtest.CompareLogs(expectedLogs, selectedLogs,
+		plogtest.IgnoreTimestamp(),
+		plogtest.IgnoreObservedTimestamp(),
+		plogtest.IgnoreResourceAttributeValue("host.name"),
+		plogtest.IgnoreLogRecordAttributeValue("k8s.object.uid"),
+		plogtest.IgnoreLogRecordAttributeValue("k8s.pod.uid"),
+		plogtest.IgnoreLogRecordAttributeValue("k8s.object.resource_version"),
+		plogtest.IgnoreResourceLogsOrder(),
+		plogtest.IgnoreScopeLogsOrder(),
+		plogtest.IgnoreLogRecordsOrder(),
+	)
+	if err != nil && os.Getenv("UPDATE_EXPECTED_RESULTS") == "true" {
+		writeNewExpectedLogsResult(t, expectedLogsFile, &selectedLogs)
+	}
+	require.NoError(t, err)
+}
+
+func setup(t *testing.T) *consumertest.LogsSink {
+	setupRun.Do(func() {
+		if os.Getenv("TEARDOWN_BEFORE_SETUP") == "true" {
+			t.Log("Running teardown before setup as TEARDOWN_BEFORE_SETUP is set to true")
+			testKubeConfig, setKubeConfig := os.LookupEnv("KUBECONFIG")
+			require.True(t, setKubeConfig, "the environment variable KUBECONFIG must be set")
+			k8sClient, err := k8stest.NewK8sClient(testKubeConfig)
+			require.NoError(t, err)
+			teardown(t, k8sClient)
+		}
+
+		internal.CreateApiServer(t, apiPort)
+
+		eventsLogsConsumer = setupHECLogsReceiver(t, splunkHecReceiverPort)
+
+		if os.Getenv("SKIP_SETUP") == "true" {
+			t.Log("Skipping setup as SKIP_SETUP is set to true")
+			return
+		}
+		deployWorkloadAndCollector(t)
+	})
+
+	return eventsLogsConsumer
+}
+
+func deployWorkloadAndCollector(t *testing.T) {
+	testKubeConfig, setKubeConfig := os.LookupEnv("KUBECONFIG")
+	require.True(t, setKubeConfig, "the environment variable KUBECONFIG must be set")
+	k8sClient, err := k8stest.NewK8sClient(testKubeConfig)
+	require.NoError(t, err)
+
+	chartPath := filepath.Join("..", "helm-charts", "splunk-otel-collector")
+	chart, err := loader.Load(chartPath)
+	require.NoError(t, err)
+
+	valuesBytes, err := os.ReadFile(filepath.Join("testdata_k8sevents", "k8sevents_values.yaml.tmpl"))
+	require.NoError(t, err)
+
+	hostEp := hostEndpoint(t)
+	if len(hostEp) == 0 {
+		require.Fail(t, "host endpoint not found")
+	}
+
+	// Deploy collector
+	replacements := struct {
+		ApiURL string
+		LogURL string
+	}{
+		fmt.Sprintf("http://%s:%d", hostEp, apiPort),
+		fmt.Sprintf("http://%s:%d", hostEp, splunkHecReceiverPort),
+	}
+	tmpl, err := template.New("").Parse(string(valuesBytes))
+	require.NoError(t, err)
+	var buf bytes.Buffer
+	err = tmpl.Execute(&buf, replacements)
+	require.NoError(t, err)
+	var values map[string]interface{}
+	err = yaml.Unmarshal(buf.Bytes(), &values)
+	require.NoError(t, err)
+
+	actionConfig := new(action.Configuration)
+	if err := actionConfig.Init(kube.GetConfig(testKubeConfig, "", "default"), "default", os.Getenv("HELM_DRIVER"), func(format string, v ...interface{}) {
+		t.Logf(format+"\n", v...)
+	}); err != nil {
+		require.NoError(t, err)
+	}
+	install := action.NewInstall(actionConfig)
+	install.Namespace = "default"
+	install.ReleaseName = "sock"
+	_, err = install.Run(chart, values)
+	if err != nil {
+		t.Logf("error reported during helm install: %v\n", err)
+		retryUpgrade := action.NewUpgrade(actionConfig)
+		retryUpgrade.Namespace = "default"
+		retryUpgrade.Install = true
+		_, err = retryUpgrade.Run("sock", chart, values)
+		require.NoError(t, err)
+	}
+
+	config, err := clientcmd.BuildConfigFromFlags("", testKubeConfig)
+	require.NoError(t, err)
+	clientset, err := kubernetes.NewForConfig(config)
+	require.NoError(t, err)
+
+	checkPodsReady(t, clientset, "default", "component=otel-k8s-cluster-receiver", 3*time.Minute)
+	time.Sleep(30 * time.Second)
+
+	// Deploy the workload
+	createNamespace(t, clientset, "k8sevents-test")
+	createdObjs, err := k8stest.CreateObjects(k8sClient, "testdata_k8sevents/testobjects")
+	require.NoError(t, err)
+	require.NotEmpty(t, createdObjs)
+
+	checkPodsReady(t, clientset, "k8sevents-test", "app=k8sevents-test", 2*time.Minute)
+
+	t.Cleanup(func() {
+		if os.Getenv("SKIP_TEARDOWN") == "true" {
+			t.Log("Skipping teardown as SKIP_TEARDOWN is set to true")
+			return
+		}
+		teardown(t, k8sClient)
+	})
+}
+
+func teardown(t *testing.T, k8sClient *k8stest.K8sClient) {
+	actionConfig := new(action.Configuration)
+	testKubeConfig := os.Getenv("KUBECONFIG")
+	require.NoError(t, actionConfig.Init(kube.GetConfig(testKubeConfig, "", "default"), "default", os.Getenv("HELM_DRIVER"), t.Logf))
+	uninstall := action.NewUninstall(actionConfig)
+	_, err := uninstall.Run("sock")
+	if err != nil {
+		t.Logf("error during helm uninstall: %v\n", err)
+	}
+
+	deleteObject(t, k8sClient, `
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: k8sevents-test
+`)
+}
+
+func setupHECLogsReceiver(t *testing.T, port int) *consumertest.LogsSink {
+	f := splunkhecreceiver.NewFactory()
+	cfg := f.CreateDefaultConfig().(*splunkhecreceiver.Config)
+	cfg.Endpoint = fmt.Sprintf("0.0.0.0:%d", splunkHecReceiverPort)
+
+	receiver := new(consumertest.LogsSink)
+	rcvr, err := f.CreateLogs(context.Background(), receivertest.NewNopSettings(), cfg, receiver)
+	require.NoError(t, err)
+
+	require.NoError(t, rcvr.Start(context.Background(), componenttest.NewNopHost()))
+	t.Cleanup(func() {
+		require.NoError(t, rcvr.Shutdown(context.Background()))
+	})
+
+	return receiver
+}
+
+func deleteObject(t *testing.T, k8sClient *k8stest.K8sClient, objYAML string) {
+	obj := &unstructured.Unstructured{}
+	err := yaml.Unmarshal([]byte(objYAML), obj)
+	require.NoError(t, err)
+	k8stest.DeleteObject(k8sClient, obj)
+}
+
+func selectLogs(t *testing.T, attributeName, attributeValue string, logSink *consumertest.LogsSink, modifyBodyFunc func(string) string) plog.Logs {
+	selectedLogs := plog.NewLogs()
+	// collapse logs across resource logs into a single one to reduce flakiness in test runs
+	for h := 0; h < len(logSink.AllLogs()); h++ {
+		logs := logSink.AllLogs()[h]
+		for i := 0; i < logs.ResourceLogs().Len(); i++ {
+			resourceLogs := logs.ResourceLogs().At(i)
+			var existingResLog plog.ResourceLogs
+			foundResource := false
+			for j := 0; j < selectedLogs.ResourceLogs().Len(); j++ {
+				if compareAttributes(resourceLogs.Resource().Attributes(), selectedLogs.ResourceLogs().At(j).Resource().Attributes()) {
+					existingResLog = selectedLogs.ResourceLogs().At(j)
+					foundResource = true
+					break
+				}
+			}
+			if !foundResource {
+				existingResLog = selectedLogs.ResourceLogs().AppendEmpty()
+				resourceLogs.Resource().Attributes().CopyTo(existingResLog.Resource().Attributes())
+			}
+			for j := 0; j < resourceLogs.ScopeLogs().Len(); j++ {
+				scopeLogs := resourceLogs.ScopeLogs().At(j)
+				var existingScopeLog plog.ScopeLogs
+				foundScope := false
+				for k := 0; k < existingResLog.ScopeLogs().Len(); k++ {
+					if compareAttributes(scopeLogs.Scope().Attributes(), existingResLog.ScopeLogs().At(k).Scope().Attributes()) {
+						existingScopeLog = existingResLog.ScopeLogs().At(k)
+						foundScope = true
+						break
+					}
+				}
+				if !foundScope {
+					existingScopeLog = existingResLog.ScopeLogs().AppendEmpty()
+					scopeLogs.Scope().Attributes().CopyTo(existingScopeLog.Scope().Attributes())
+				}
+				for k := 0; k < scopeLogs.LogRecords().Len(); k++ {
+					logRecord := scopeLogs.LogRecords().At(k)
+					attributes := logRecord.Attributes()
+					if attr, ok := attributes.Get(attributeName); ok && attr.Str() == attributeValue {
+						modifiedBody := modifyBodyFunc(logRecord.Body().Str())
+						logRecord.Body().SetStr(modifiedBody)
+						logRecord.CopyTo(existingScopeLog.LogRecords().AppendEmpty())
+					}
+				}
+			}
+		}
+	}
+
+	return selectedLogs
+}
+
+func compareAttributes(attr1, attr2 pcommon.Map) bool {
+	if len(attr1.AsRaw()) != len(attr2.AsRaw()) {
+		return false
+	}
+	for k1, v1 := range attr1.AsRaw() {
+		if v2, ok := attr2.AsRaw()[k1]; !ok || v1 != v2 {
+			return false
+		}
+	}
+	return true
+}
+
+func removeFlakyLogRecordAttr(logs plog.Logs, attributeName string) {
+	for i := 0; i < logs.ResourceLogs().Len(); i++ {
+		resourceLogs := logs.ResourceLogs().At(i)
+		for j := 0; j < resourceLogs.ScopeLogs().Len(); j++ {
+			scopeLogs := resourceLogs.ScopeLogs().At(j)
+			for k := 0; k < scopeLogs.LogRecords().Len(); k++ {
+				logRecord := scopeLogs.LogRecords().At(k)
+				logRecord.Attributes().Remove(attributeName)
+			}
+		}
+	}
+}
diff --git a/functional_tests/testdata_k8sevents/expected_k8sevents.yaml b/functional_tests/testdata_k8sevents/expected_k8sevents.yaml
new file mode 100644
index 000000000..f421607e4
--- /dev/null
+++ b/functional_tests/testdata_k8sevents/expected_k8sevents.yaml
@@ -0,0 +1,408 @@
+resourceLogs:
+  - resource:
+      attributes:
+        - key: host.name
+          value:
+            stringValue: sock-splunk-otel-collector-k8s-cluster-receiver-6ff779f9bbsv6fh
+        - key: com.splunk.source
+          value:
+            stringValue: kubernetes
+        - key: com.splunk.sourcetype
+          value:
+            stringValue: kube:events
+        - key: com.splunk.index
+          value:
+            stringValue: main
+    scopeLogs:
+      - logRecords:
+          - attributes:
+              - key: deployment.environment
+                value:
+                  stringValue: dev
+              - key: k8s.cluster.name
+                value:
+                  stringValue: dev-operator
+              - key: k8s.event.action
+                value:
+                  stringValue: ""
+              - key: k8s.event.count
+                value:
+                  doubleValue: 1
+              - key: k8s.event.reason
+                value:
+                  stringValue: SuccessfulCreate
+              - key: k8s.namespace.name
+                value:
+                  stringValue: k8sevents-test
+              - key: k8s.node.name
+                value:
+                  stringValue: ""
+              - key: k8s.object.api_version
+                value:
+                  stringValue: apps/v1
+              - key: k8s.object.fieldpath
+                value:
+                  stringValue: ""
+              - key: k8s.object.kind
+                value:
+                  stringValue: StatefulSet
+              - key: k8s.object.name
+                value:
+                  stringValue: k8sevents-test
+              - key: k8s.object.resource_version
+                value:
+                  stringValue: "35892"
+              - key: k8s.object.uid
+                value:
+                  stringValue: 4332c640-23ff-4768-9e78-d4f683135646
+              - key: metric_source
+                value:
+                  stringValue: kubernetes
+              - key: os.type
+                value:
+                  stringValue: linux
+              - key: otel.log.severity.number
+                value:
+                  doubleValue: 9
+              - key: otel.log.severity.text
+                value:
+                  stringValue: Normal
+            body:
+              stringValue: create Pod k8sevents-test-0 in StatefulSet k8sevents-test successful
+            spanId: ""
+            timeUnixNano: "1740082291000000000"
+            traceId: ""
+          - attributes:
+              - key: deployment.environment
+                value:
+                  stringValue: dev
+              - key: k8s.cluster.name
+                value:
+                  stringValue: dev-operator
+              - key: k8s.event.action
+                value:
+                  stringValue: ""
+              - key: k8s.event.count
+                value:
+                  doubleValue: 1
+              - key: k8s.event.reason
+                value:
+                  stringValue: Scheduled
+              - key: k8s.namespace.name
+                value:
+                  stringValue: k8sevents-test
+              - key: k8s.node.name
+                value:
+                  stringValue: ""
+              - key: k8s.object.api_version
+                value:
+                  stringValue: v1
+              - key: k8s.object.fieldpath
+                value:
+                  stringValue: ""
+              - key: k8s.object.kind
+                value:
+                  stringValue: Pod
+              - key: k8s.object.name
+                value:
+                  stringValue: k8sevents-test-0
+              - key: k8s.object.resource_version
+                value:
+                  stringValue: "35894"
+              - key: k8s.object.uid
+                value:
+                  stringValue: 725277aa-65d5-4dbc-a0ae-6e9b0ab4d319
+              - key: metric_source
+                value:
+                  stringValue: kubernetes
+              - key: os.type
+                value:
+                  stringValue: linux
+              - key: otel.log.severity.number
+                value:
+                  doubleValue: 9
+              - key: otel.log.severity.text
+                value:
+                  stringValue: Normal
+            body:
+              stringValue: Successfully assigned k8sevents-test/k8sevents-test-0 to kind-control-plane
+            spanId: ""
+            timeUnixNano: "1740082291000000000"
+            traceId: ""
+          - attributes:
+              - key: container.image.name
+                value:
+                  stringValue: busybox
+              - key: container.image.tag
+                value:
+                  stringValue: latest
+              - key: deployment.environment
+                value:
+                  stringValue: dev
+              - key: k8s.cluster.name
+                value:
+                  stringValue: dev-operator
+              - key: k8s.event.action
+                value:
+                  stringValue: ""
+              - key: k8s.event.count
+                value:
+                  doubleValue: 1
+              - key: k8s.event.reason
+                value:
+                  stringValue: Pulling
+              - key: k8s.namespace.name
+                value:
+                  stringValue: k8sevents-test
+              - key: k8s.node.name
+                value:
+                  stringValue: kind-control-plane
+              - key: k8s.object.api_version
+                value:
+                  stringValue: v1
+              - key: k8s.object.fieldpath
+                value:
+                  stringValue: spec.containers{k8sevents-test}
+              - key: k8s.object.kind
+                value:
+                  stringValue: Pod
+              - key: k8s.object.name
+                value:
+                  stringValue: k8sevents-test-0
+              - key: k8s.object.resource_version
+                value:
+                  stringValue: "35897"
+              - key: k8s.object.uid
+                value:
+                  stringValue: 725277aa-65d5-4dbc-a0ae-6e9b0ab4d319
+              - key: k8s.pod.name
+                value:
+                  stringValue: k8sevents-test-0
+              - key: k8s.pod.uid
+                value:
+                  stringValue: 725277aa-65d5-4dbc-a0ae-6e9b0ab4d319
+              - key: metric_source
+                value:
+                  stringValue: kubernetes
+              - key: os.type
+                value:
+                  stringValue: linux
+              - key: otel.log.severity.number
+                value:
+                  doubleValue: 9
+              - key: otel.log.severity.text
+                value:
+                  stringValue: Normal
+            body:
+              stringValue: Pulling image "busybox:latest"
+            spanId: ""
+            timeUnixNano: "1740082292000000000"
+            traceId: ""
+          - attributes:
+              - key: container.image.name
+                value:
+                  stringValue: busybox
+              - key: container.image.tag
+                value:
+                  stringValue: latest
+              - key: deployment.environment
+                value:
+                  stringValue: dev
+              - key: k8s.cluster.name
+                value:
+                  stringValue: dev-operator
+              - key: k8s.event.action
+                value:
+                  stringValue: ""
+              - key: k8s.event.count
+                value:
+                  doubleValue: 1
+              - key: k8s.event.reason
+                value:
+                  stringValue: Pulled
+              - key: k8s.namespace.name
+                value:
+                  stringValue: k8sevents-test
+              - key: k8s.node.name
+                value:
+                  stringValue: kind-control-plane
+              - key: k8s.object.api_version
+                value:
+                  stringValue: v1
+              - key: k8s.object.fieldpath
+                value:
+                  stringValue: spec.containers{k8sevents-test}
+              - key: k8s.object.kind
+                value:
+                  stringValue: Pod
+              - key: k8s.object.name
+                value:
+                  stringValue: k8sevents-test-0
+              - key: k8s.object.resource_version
+                value:
+                  stringValue: "35897"
+              - key: k8s.object.uid
+                value:
+                  stringValue: 725277aa-65d5-4dbc-a0ae-6e9b0ab4d319
+              - key: k8s.pod.name
+                value:
+                  stringValue: k8sevents-test-0
+              - key: k8s.pod.uid
+                value:
+                  stringValue: 725277aa-65d5-4dbc-a0ae-6e9b0ab4d319
+              - key: metric_source
+                value:
+                  stringValue: kubernetes
+              - key: os.type
+                value:
+                  stringValue: linux
+              - key: otel.log.severity.number
+                value:
+                  doubleValue: 9
+              - key: otel.log.severity.text
+                value:
+                  stringValue: Normal
+            body:
+              stringValue: Successfully pulled image "busybox:latest" in <time> (<time> including waiting)
+            spanId: ""
+            timeUnixNano: "1740082293000000000"
+            traceId: ""
+          - attributes:
+              - key: container.image.name
+                value:
+                  stringValue: busybox
+              - key: container.image.tag
+                value:
+                  stringValue: latest
+              - key: deployment.environment
+                value:
+                  stringValue: dev
+              - key: k8s.cluster.name
+                value:
+                  stringValue: dev-operator
+              - key: k8s.event.action
+                value:
+                  stringValue: ""
+              - key: k8s.event.count
+                value:
+                  doubleValue: 1
+              - key: k8s.event.reason
+                value:
+                  stringValue: Created
+              - key: k8s.namespace.name
+                value:
+                  stringValue: k8sevents-test
+              - key: k8s.node.name
+                value:
+                  stringValue: kind-control-plane
+              - key: k8s.object.api_version
+                value:
+                  stringValue: v1
+              - key: k8s.object.fieldpath
+                value:
+                  stringValue: spec.containers{k8sevents-test}
+              - key: k8s.object.kind
+                value:
+                  stringValue: Pod
+              - key: k8s.object.name
+                value:
+                  stringValue: k8sevents-test-0
+              - key: k8s.object.resource_version
+                value:
+                  stringValue: "35897"
+              - key: k8s.object.uid
+                value:
+                  stringValue: 725277aa-65d5-4dbc-a0ae-6e9b0ab4d319
+              - key: k8s.pod.name
+                value:
+                  stringValue: k8sevents-test-0
+              - key: k8s.pod.uid
+                value:
+                  stringValue: 725277aa-65d5-4dbc-a0ae-6e9b0ab4d319
+              - key: metric_source
+                value:
+                  stringValue: kubernetes
+              - key: os.type
+                value:
+                  stringValue: linux
+              - key: otel.log.severity.number
+                value:
+                  doubleValue: 9
+              - key: otel.log.severity.text
+                value:
+                  stringValue: Normal
+            body:
+              stringValue: Created container k8sevents-test
+            spanId: ""
+            timeUnixNano: "1740082293000000000"
+            traceId: ""
+          - attributes:
+              - key: container.image.name
+                value:
+                  stringValue: busybox
+              - key: container.image.tag
+                value:
+                  stringValue: latest
+              - key: deployment.environment
+                value:
+                  stringValue: dev
+              - key: k8s.cluster.name
+                value:
+                  stringValue: dev-operator
+              - key: k8s.event.action
+                value:
+                  stringValue: ""
+              - key: k8s.event.count
+                value:
+                  doubleValue: 1
+              - key: k8s.event.reason
+                value:
+                  stringValue: Started
+              - key: k8s.namespace.name
+                value:
+                  stringValue: k8sevents-test
+              - key: k8s.node.name
+                value:
+                  stringValue: kind-control-plane
+              - key: k8s.object.api_version
+                value:
+                  stringValue: v1
+              - key: k8s.object.fieldpath
+                value:
+                  stringValue: spec.containers{k8sevents-test}
+              - key: k8s.object.kind
+                value:
+                  stringValue: Pod
+              - key: k8s.object.name
+                value:
+                  stringValue: k8sevents-test-0
+              - key: k8s.object.resource_version
+                value:
+                  stringValue: "35897"
+              - key: k8s.object.uid
+                value:
+                  stringValue: 725277aa-65d5-4dbc-a0ae-6e9b0ab4d319
+              - key: k8s.pod.name
+                value:
+                  stringValue: k8sevents-test-0
+              - key: k8s.pod.uid
+                value:
+                  stringValue: 725277aa-65d5-4dbc-a0ae-6e9b0ab4d319
+              - key: metric_source
+                value:
+                  stringValue: kubernetes
+              - key: os.type
+                value:
+                  stringValue: linux
+              - key: otel.log.severity.number
+                value:
+                  doubleValue: 9
+              - key: otel.log.severity.text
+                value:
+                  stringValue: Normal
+            body:
+              stringValue: Started container k8sevents-test
+            spanId: ""
+            timeUnixNano: "1740082293000000000"
+            traceId: ""
+        scope: {}
diff --git a/functional_tests/testdata_k8sevents/k8sevents_values.yaml.tmpl b/functional_tests/testdata_k8sevents/k8sevents_values.yaml.tmpl
new file mode 100644
index 000000000..b6cbbaac8
--- /dev/null
+++ b/functional_tests/testdata_k8sevents/k8sevents_values.yaml.tmpl
@@ -0,0 +1,21 @@
+splunkObservability:
+   realm:       CHANGEME
+   accessToken: CHANGEME
+   apiUrl: {{ .ApiURL }}
+   logsEnabled: true
+   metricsEnabled: false
+   tracesEnabled: false
+
+splunkPlatform:
+  endpoint: {{ .LogURL }}
+  token: "00000000-0000-0000-0000-0000000000000"
+
+agent:
+  enabled: false
+
+clusterReceiver:
+  enabled: true
+  eventsEnabled: true
+
+environment: dev
+clusterName: dev-operator
diff --git a/functional_tests/testdata_k8sevents/testobjects/1-serviceaccount.yaml b/functional_tests/testdata_k8sevents/testobjects/1-serviceaccount.yaml
new file mode 100644
index 000000000..d5fb9c4f4
--- /dev/null
+++ b/functional_tests/testdata_k8sevents/testobjects/1-serviceaccount.yaml
@@ -0,0 +1,5 @@
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: k8sevents-test
+  namespace: k8sevents-test
diff --git a/functional_tests/testdata_k8sevents/testobjects/2-svc.yaml b/functional_tests/testdata_k8sevents/testobjects/2-svc.yaml
new file mode 100644
index 000000000..829201ee0
--- /dev/null
+++ b/functional_tests/testdata_k8sevents/testobjects/2-svc.yaml
@@ -0,0 +1,12 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: k8sevents-test
+  namespace: k8sevents-test
+spec:
+  ports:
+    - port: 80
+      name: http
+  clusterIP: None
+  selector:
+    app: k8sevents-test
diff --git a/functional_tests/testdata_k8sevents/testobjects/3-sts.yaml b/functional_tests/testdata_k8sevents/testobjects/3-sts.yaml
new file mode 100644
index 000000000..08d1cd842
--- /dev/null
+++ b/functional_tests/testdata_k8sevents/testobjects/3-sts.yaml
@@ -0,0 +1,25 @@
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: k8sevents-test
+  namespace: k8sevents-test
+spec:
+  serviceName: k8sevents-test
+  replicas: 1
+  selector:
+    matchLabels:
+      app: k8sevents-test
+  template:
+    metadata:
+      labels:
+        app: k8sevents-test
+    spec:
+      serviceAccountName: k8sevents-test
+      containers:
+        - image: busybox:latest
+          imagePullPolicy: Always
+          name: k8sevents-test
+          command: ["sleep", "infinity"]
+          ports:
+            - containerPort: 80
+              name: http