[receiver/k8scluster] add entity attributes to namespace and container

.chloggen/add-namespace-entity-metadata.yaml

@@ -0,0 +1,32 @@
+# Use this changelog template to create an entry for release notes.
+
+# 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. filelogreceiver)
+component: k8sclusterreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Adds new descriptive attributes/metadata to the k8s.namespace and the container entity emitted from k8sclusterreceiver.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37580]
+
+# (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: |
+  - Adds the following attributes to k8s.namespace entity:
+    - k8s.namespace.phase: The phase of a namespace indicates where the namespace is in its lifecycle. E.g. 'active', 'terminating'
+    - k8s.namespace.creation_timestamp: The time when the namespace object was created.
+  - Adds the following attributes to container entity:
+    - container.creation_timestamp: The time when the container was started. Only available if container is either in 'running' or 'terminated' state.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

receiver/k8sclusterreceiver/e2e_test.go

@@ -7,21 +7,25 @@ package k8sclusterreceiver
 
 import (
 	"context"
+	"fmt"
 	"path/filepath"
 	"strings"
 	"testing"
 	"time"
 
 	"github.com/google/uuid"
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
 	"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/pdata/pmetric"
 	"go.opentelemetry.io/collector/receiver/otlpreceiver"
 	"go.opentelemetry.io/collector/receiver/receivertest"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/plogtest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/pmetrictest"
 	k8stest "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/xk8stest"
 )
@@ -245,25 +249,206 @@ func containerImageShorten(value string) string {
 	return shortenNames(value[(strings.LastIndex(value, "/") + 1):])
 }
 
-func startUpSink(t *testing.T, mc *consumertest.MetricsSink) func() {
+func startUpSink(t *testing.T, consumer interface{}) func() {
 	f := otlpreceiver.NewFactory()
 	cfg := f.CreateDefaultConfig().(*otlpreceiver.Config)
 	cfg.HTTP = nil
 	cfg.GRPC.NetAddr.Endpoint = "0.0.0.0:4317"
 
-	rcvr, err := f.CreateMetrics(context.Background(), receivertest.NewNopSettings(), cfg, mc)
+	var err error
+	var rcvr component.Component
+
+	switch c := consumer.(type) {
+	case *consumertest.MetricsSink:
+		rcvr, err = f.CreateMetrics(context.Background(), receivertest.NewNopSettings(), cfg, c)
+	case *consumertest.LogsSink:
+		rcvr, err = f.CreateLogs(context.Background(), receivertest.NewNopSettings(), cfg, c)
+	default:
+		t.Fatalf("unsupported consumer type: %T", c)
+	}
+
+	require.NoError(t, err, "failed creating receiver")
 	require.NoError(t, rcvr.Start(context.Background(), componenttest.NewNopHost()))
-	require.NoError(t, err, "failed creating metrics receiver")
+
 	return func() {
-		assert.NoError(t, rcvr.Shutdown(context.Background()))
+		require.NoError(t, rcvr.Shutdown(context.Background()))
 	}
 }
 
-func waitForData(t *testing.T, entriesNum int, mc *consumertest.MetricsSink) {
+func waitForData(t *testing.T, entriesNum int, consumer interface{}) {
 	timeoutMinutes := 3
 	require.Eventuallyf(t, func() bool {
-		return len(mc.AllMetrics()) > entriesNum
+		switch c := consumer.(type) {
+		case *consumertest.MetricsSink:
+			return len(c.AllMetrics()) > entriesNum
+		case *consumertest.LogsSink:
+			return len(c.AllLogs()) > entriesNum
+		default:
+			t.Fatalf("unsupported consumer type: %T", c)
+			return false
+		}
 	}, time.Duration(timeoutMinutes)*time.Minute, 1*time.Second,
-		"failed to receive %d entries,  received %d metrics in %d minutes", entriesNum,
-		len(mc.AllMetrics()), timeoutMinutes)
+		"failed to receive %d entries in %d minutes", entriesNum, timeoutMinutes)
+}
+
+// TestE2ENamespaceMetadata tests the k8s cluster receiver's exporting of entities in a real k8s cluster
+func TestE2ENamespaceMetadata(t *testing.T) {
+	k8sClient, err := k8stest.NewK8sClient(testKubeConfig)
+	require.NoError(t, err)
+
+	logsConsumer := new(consumertest.LogsSink)
+	shutdownSink := startUpSink(t, logsConsumer)
+	defer shutdownSink()
+
+	testID := uuid.NewString()[:8]
+	collectorObjs := k8stest.CreateCollectorObjects(t, k8sClient, testID, filepath.Join(".", "testdata", "e2e", "entities-test", "collector"), map[string]string{}, "")
+
+	t.Cleanup(func() {
+		for _, obj := range collectorObjs {
+			require.NoErrorf(t, k8stest.DeleteObject(k8sClient, obj), "failed to delete object %s", obj.GetName())
+		}
+	})
+
+	namespaceObj, err := k8stest.CreateObjects(k8sClient, filepath.Join(".", "testdata", "e2e", "entities-test", "testobjects"))
+	require.NoErrorf(t, err, "failed to create test k8s objects")
+
+	entityType := "k8s.namespace"
+	entityNameKey := "k8s.namespace.name"
+	entityName := "test-entities-ns"
+	namespaceLogs := waitForEntityLogs(t, entityType, entityNameKey, entityName, logsConsumer)
+
+	expected, err := golden.ReadLogs("./testdata/e2e/entities-test/expected-ns.yaml")
+	require.NoError(t, err)
+
+	commonReplacements := map[string]map[string]string{
+		"otel.entity.attributes": {
+			"k8s.namespace.creation_timestamp": "2025-01-01T00:00:00Z",
+		},
+		"otel.entity.id": {
+			"k8s.namespace.uid": "entity-id",
+		},
+	}
+
+	replaceLogValues(t, namespaceLogs[0], commonReplacements)
+
+	require.NoError(t, plogtest.CompareLogs(expected, namespaceLogs[0],
+		plogtest.IgnoreTimestamp(),
+		plogtest.IgnoreObservedTimestamp(),
+		plogtest.IgnoreScopeLogsOrder(),
+		plogtest.IgnoreLogRecordsOrder(),
+	))
+
+	logsConsumer.Reset()
+
+	// Delete test namespace object to trigger terminating phase and check if new event log is generated with the correct phase
+	require.NoErrorf(t, k8stest.DeleteObjects(k8sClient, namespaceObj), "failed to delete test k8s objects")
+
+	namespaceLogs = waitForEntityLogs(t, entityType, entityNameKey, entityName, logsConsumer)
+
+	replaceLogValues(t, namespaceLogs[0], commonReplacements)
+
+	// update the phase in expected log to terminating
+	replaceLogValues(t, expected, map[string]map[string]string{
+		"otel.entity.attributes": {
+			"k8s.namespace.phase": "terminating",
+		},
+	})
+
+	require.NoError(t, plogtest.CompareLogs(expected, namespaceLogs[0],
+		plogtest.IgnoreTimestamp(),
+		plogtest.IgnoreObservedTimestamp(),
+		plogtest.IgnoreScopeLogsOrder(),
+		plogtest.IgnoreLogRecordsOrder(),
+	))
+}
+
+// filterEntityLogs returns logs that contain the entity with the given entityType and entityNameKey and entityName.
+func filterEntityLogs(logs []plog.Logs, entityType, entityNameKey, entityName string) []plog.Logs {
+	var entityLogs []plog.Logs
+	for _, log := range logs {
+		for i := 0; i < log.ResourceLogs().Len(); i++ {
+			resourceLog := log.ResourceLogs().At(i)
+			for j := 0; j < resourceLog.ScopeLogs().Len(); j++ {
+				scopeLog := resourceLog.ScopeLogs().At(j)
+				if containsEntity(scopeLog, entityType, entityNameKey, entityName) {
+					entityLogs = append(entityLogs, log)
+					break
+				}
+			}
+		}
+	}
+	return entityLogs
+}
+
+// containsEntity returns true if the given scopeLog contains an entity with the given entityType, entityNameKey and entityName.
+func containsEntity(scopeLog plog.ScopeLogs, entityType, entityNameKey, entityName string) bool {
+	for k := 0; k < scopeLog.LogRecords().Len(); k++ {
+		logRecord := scopeLog.LogRecords().At(k)
+		entityTypeAttr, exists := logRecord.Attributes().Get("otel.entity.type")
+		if exists && entityTypeAttr.Type() == pcommon.ValueTypeStr && entityTypeAttr.Str() == entityType {
+			entityAttributesAttr, exists := logRecord.Attributes().Get("otel.entity.attributes")
+			if exists && entityAttributesAttr.Type() == pcommon.ValueTypeMap {
+				entityAttributes := entityAttributesAttr.Map()
+				entityNameValue, exists := entityAttributes.Get(entityNameKey)
+				if exists && entityNameValue.Type() == pcommon.ValueTypeStr && entityNameValue.Str() == entityName {
+					return true
+				}
+			}
+		}
+	}
+	return false
+}
+
+func waitForEntityLogs(t *testing.T, entityType, entityNameKey, entityName string, consumer *consumertest.LogsSink) []plog.Logs {
+	timeoutMinutes := 3
+	var entityLogs []plog.Logs
+	require.Eventuallyf(t, func() bool {
+		for _, logs := range consumer.AllLogs() {
+			filteredLogs := filterEntityLogs([]plog.Logs{logs}, entityType, entityNameKey, entityName)
+			if len(filteredLogs) > 0 {
+				entityLogs = filteredLogs
+				return true
+			}
+		}
+		return false
+	}, time.Duration(timeoutMinutes)*time.Minute, 5*time.Second,
+		"failed to receive logs for entity %s with name %s in %d minutes", entityType, entityName, timeoutMinutes)
+	return entityLogs
+}
+
+// replaceLogValue replaces the value of the key in the attribute with attributeName with the placeholder.
+func replaceLogValue(logs plog.Logs, attributeName, key, placeholder string) error {
+	rls := logs.ResourceLogs()
+	for i := 0; i < rls.Len(); i++ {
+		sls := rls.At(i).ScopeLogs()
+		for j := 0; j < sls.Len(); j++ {
+			lrs := sls.At(j).LogRecords()
+			for k := 0; k < lrs.Len(); k++ {
+				lr := lrs.At(k)
+				attr, exists := lr.Attributes().Get(attributeName)
+				if !exists {
+					return fmt.Errorf("expected attribute %s not found in log record", attributeName)
+				}
+				if attr.Type() != pcommon.ValueTypeMap {
+					return fmt.Errorf("attribute %s is not of type map in log record", attributeName)
+				}
+				attrMap := attr.Map()
+				val, exists := attrMap.Get(key)
+				if !exists {
+					return fmt.Errorf("expected key %s not found in attribute %s map", key, attributeName)
+				}
+				val.SetStr(placeholder)
+			}
+		}
+	}
+	return nil
+}
+
+func replaceLogValues(t *testing.T, logs plog.Logs, replacements map[string]map[string]string) {
+	for attributeName, keys := range replacements {
+		for key, placeholder := range keys {
+			err := replaceLogValue(logs, attributeName, key, placeholder)
+			require.NoError(t, err)
+		}
+	}
 }

receiver/k8sclusterreceiver/informer_transform_test.go

@@ -10,7 +10,6 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/internal/testutils"
 )
@@ -38,7 +37,6 @@ func TestTransformObject(t *testing.T) {
 					testutils.NewPodStatusWithContainer("container-name", "container-id"),
 				)
 				pod.Spec.Containers[0].Image = ""
-				pod.Status.ContainerStatuses[0].State = corev1.ContainerState{Running: &corev1.ContainerStateRunning{StartedAt: v1.Time{}}}
 				return pod
 			}(),
 			same: false,

receiver/k8sclusterreceiver/internal/container/containers.go

@@ -4,6 +4,8 @@
 package container // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/internal/container"
 
 import (
+	"time"
+
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	conventions "go.opentelemetry.io/collector/semconv/v1.6.1"
 	"go.uber.org/zap"
@@ -17,9 +19,10 @@ import (
 )
 
 const (
-	// Keys for container metadata.
-	containerKeyStatus       = "container.status"
-	containerKeyStatusReason = "container.status.reason"
+	// Keys for container metadata used for entity attributes.
+	containerKeyStatus         = "container.status"
+	containerKeyStatusReason   = "container.status.reason"
+	containerCreationTimestamp = "container.creation_timestamp"
 
 	// Values for container metadata
 	containerStatusRunning    = "running"
@@ -98,11 +101,17 @@ func GetMetadata(cs corev1.ContainerStatus) *metadata.KubernetesMetadata {
 
 	if cs.State.Running != nil {
 		mdata[containerKeyStatus] = containerStatusRunning
+		if !cs.State.Running.StartedAt.IsZero() {
+			mdata[containerCreationTimestamp] = cs.State.Running.StartedAt.Format(time.RFC3339)
+		}
 	}
 
 	if cs.State.Terminated != nil {
 		mdata[containerKeyStatus] = containerStatusTerminated
 		mdata[containerKeyStatusReason] = cs.State.Terminated.Reason
+		if !cs.State.Terminated.StartedAt.IsZero() {
+			mdata[containerCreationTimestamp] = cs.State.Terminated.StartedAt.Format(time.RFC3339)
+		}
 	}
 
 	if cs.State.Waiting != nil {