[receiver/k8s_cluster] Send entity delete events

.chloggen/k8scluster-receiver-delete-events.yaml

@@ -0,0 +1,22 @@
+# 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: "receiver/k8s_cluster"
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add onDelete handler to emit the experimental entity delete events
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [40278]
+
+# 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: [user]

receiver/k8sclusterreceiver/internal/metadata/entities.go

@@ -21,7 +21,8 @@ func GetEntityEvents(oldMetadata, newMetadata map[metadataPkg.ResourceID]*Kubern
 			entityEvent := out.AppendEmpty()
 			entityEvent.SetTimestamp(timestamp)
 			entityEvent.ID().PutStr(oldObj.ResourceIDKey, string(oldObj.ResourceID))
-			entityEvent.SetEntityDelete()
+			deleteEvent := entityEvent.SetEntityDelete()
+			deleteEvent.SetEntityType(oldObj.EntityType)
 		}
 	}
 

receiver/k8sclusterreceiver/receiver_test.go

@@ -280,11 +280,10 @@ func TestReceiverWithMetadata(t *testing.T) {
 	}, 10*time.Second, 100*time.Millisecond,
 		"metadata not collected")
 
-	// Must have 3 entity events: once for the add, followed by an update and
-	// then another update, which unlike metadata calls actually happens since
-	// even unchanged entities trigger an event.
+	// Must have 4 entity events: once for the add, followed by an update and
+	// then another update, and then a delete.
 	require.Eventually(t, func() bool {
-		return logsConsumer.LogRecordCount() == 3
+		return logsConsumer.LogRecordCount() == 4
 	}, 10*time.Second, 100*time.Millisecond,
 		"entity events not collected")
 

receiver/k8sclusterreceiver/watcher.go

@@ -259,6 +259,7 @@ func (rw *resourceWatcher) setupInformer(gvk schema.GroupVersionKind, informer c
 	_, err = informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc:    rw.onAdd,
 		UpdateFunc: rw.onUpdate,
+		DeleteFunc: rw.onDelete,
 	})
 	if err != nil {
 		rw.logger.Error("error adding event handler to informer", zap.Error(err))
@@ -292,6 +293,17 @@ func (rw *resourceWatcher) onUpdate(oldObj, newObj any) {
 	rw.syncMetadataUpdate(rw.objMetadata(oldObj), rw.objMetadata(newObj))
 }
 
+func (rw *resourceWatcher) onDelete(oldObj any) {
+	rw.waitForInitialInformerSync()
+
+	// Sync metadata only if there's at least one destination for it to sent.
+	if !rw.hasDestination() {
+		return
+	}
+
+	rw.syncMetadataUpdate(rw.objMetadata(oldObj), map[experimentalmetricmetadata.ResourceID]*metadata.KubernetesMetadata{})
+}
+
 // objMetadata returns the metadata for the given object.
 func (rw *resourceWatcher) objMetadata(obj any) map[experimentalmetricmetadata.ResourceID]*metadata.KubernetesMetadata {
 	switch o := obj.(type) {

receiver/k8sclusterreceiver/watcher_test.go

@@ -299,6 +299,7 @@ func TestSyncMetadataAndEmitEntityEvents(t *testing.T) {
 	lr = logsConsumer.AllLogs()[4].ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0)
 	expected = map[string]any{
 		"otel.entity.event.type": "entity_delete",
+		"otel.entity.type":       "k8s.pod",
 		"otel.entity.id":         map[string]any{"k8s.pod.uid": "pod0"},
 	}
 	assert.Equal(t, expected, lr.Attributes().AsRaw())