[processor/k8sattributes] Use podUID instead podName to determine which pods should be deleted from cache (#43008)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

when creating deleteQueue, rely on a pod UID as a unique identifier, as
pod name isn't always unique

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes
#42978


<!--Please delete paragraphs that you did not use before submitting.-->

---------

Signed-off-by: odubajDT <ondrej.dubaj@dynatrace.com>

Author: odubajDT

.chloggen/k8satributes-delete-queue.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: processor/k8sattributes
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Use podUID instead podName to determine which pods should be deleted from cache"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [42978]
+
+# (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:
+
+# 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: []

processor/k8sattributesprocessor/internal/kube/client.go

@@ -643,41 +643,44 @@ func (c *WatchClient) deleteLoop(interval, gracePeriod time.Duration) {
 	for {
 		select {
 		case <-time.After(interval):
-			var cutoff int
-			now := time.Now()
-			c.deleteMut.Lock()
-			for i := range c.deleteQueue {
-				d := c.deleteQueue[i]
-				if d.ts.Add(gracePeriod).After(now) {
-					break
-				}
-				cutoff = i + 1
-			}
-			toDelete := c.deleteQueue[:cutoff]
-			c.deleteQueue = c.deleteQueue[cutoff:]
-			c.deleteMut.Unlock()
-
-			c.m.Lock()
-			for i := range toDelete {
-				d := toDelete[i]
-				if p, ok := c.Pods[d.id]; ok {
-					// Sanity check: make sure we are deleting the same pod
-					// and the underlying state (ip<>pod mapping) has not changed.
-					if p.Name == d.podName {
-						delete(c.Pods, d.id)
-					}
-				}
-			}
-			podTableSize := len(c.Pods)
-			c.telemetryBuilder.OtelsvcK8sPodTableSize.Record(context.Background(), int64(podTableSize))
-			c.m.Unlock()
-
+			c.deleteLoopProcessing(gracePeriod)
 		case <-c.stopCh:
 			return
 		}
 	}
 }
 
+func (c *WatchClient) deleteLoopProcessing(gracePeriod time.Duration) {
+	var cutoff int
+	now := time.Now()
+	c.deleteMut.Lock()
+	for i := range c.deleteQueue {
+		d := c.deleteQueue[i]
+		if d.ts.Add(gracePeriod).After(now) {
+			break
+		}
+		cutoff = i + 1
+	}
+	toDelete := c.deleteQueue[:cutoff]
+	c.deleteQueue = c.deleteQueue[cutoff:]
+	c.deleteMut.Unlock()
+
+	c.m.Lock()
+	for i := range toDelete {
+		d := toDelete[i]
+		if p, ok := c.Pods[d.id]; ok {
+			// Sanity check: make sure we are deleting the same pod
+			// and the underlying state (ip<>pod mapping) has not changed.
+			if p.PodUID == d.podUID {
+				delete(c.Pods, d.id)
+			}
+		}
+	}
+	podTableSize := len(c.Pods)
+	c.telemetryBuilder.OtelsvcK8sPodTableSize.Record(context.Background(), int64(podTableSize))
+	c.m.Unlock()
+}
+
 // GetPod takes an IP address or Pod UID and returns the pod the identifier is associated with.
 func (c *WatchClient) GetPod(identifier PodIdentifier) (*Pod, bool) {
 	c.m.RLock()
@@ -1463,18 +1466,18 @@ func (c *WatchClient) forgetPod(pod *api_v1.Pod) {
 		id := identifiers[i]
 		p, ok := c.GetPod(id)
 
-		if ok && p.Name == pod.Name {
-			c.appendDeleteQueue(id, pod.Name)
+		if ok && p.PodUID == string(pod.UID) {
+			c.appendDeleteQueue(id, p.PodUID)
 		}
 	}
 }
 
-func (c *WatchClient) appendDeleteQueue(podID PodIdentifier, podName string) {
+func (c *WatchClient) appendDeleteQueue(podID PodIdentifier, podUID string) {
 	c.deleteMut.Lock()
 	c.deleteQueue = append(c.deleteQueue, deleteRequest{
-		id:      podID,
-		podName: podName,
-		ts:      time.Now(),
+		id:     podID,
+		podUID: podUID,
+		ts:     time.Now(),
 	})
 	c.deleteMut.Unlock()
 }

processor/k8sattributesprocessor/internal/kube/client_test.go

@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/selection"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/tools/cache"
@@ -455,6 +456,7 @@ func TestPodDelete(t *testing.T) {
 	c.deleteQueue = c.deleteQueue[:0]
 	pod = &api_v1.Pod{}
 	pod.Status.PodIP = "1.1.1.1"
+	pod.UID = "aaaaaaaa-bbbb-cccc-dddd"
 	c.handlePodDelete(pod)
 	got = c.Pods[newPodIdentifier("connection", "k8s.pod.ip", "1.1.1.1")]
 	assert.Len(t, c.Pods, 5)
@@ -472,7 +474,6 @@ func TestPodDelete(t *testing.T) {
 	assert.Len(t, c.deleteQueue, 3)
 	deleteRequest := c.deleteQueue[0]
 	assert.Equal(t, newPodIdentifier("connection", "k8s.pod.ip", "1.1.1.1"), deleteRequest.id)
-	assert.Equal(t, "podB", deleteRequest.podName)
 	assert.False(t, deleteRequest.ts.Before(tsBeforeDelete))
 	assert.False(t, deleteRequest.ts.After(time.Now()))
 
@@ -488,12 +489,12 @@ func TestPodDelete(t *testing.T) {
 	assert.Len(t, c.deleteQueue, 5)
 	deleteRequest = c.deleteQueue[0]
 	assert.Equal(t, newPodIdentifier("connection", "k8s.pod.ip", "2.2.2.2"), deleteRequest.id)
-	assert.Equal(t, "podC", deleteRequest.podName)
+	assert.Equal(t, "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", deleteRequest.podUID)
 	assert.False(t, deleteRequest.ts.Before(tsBeforeDelete))
 	assert.False(t, deleteRequest.ts.After(time.Now()))
 	deleteRequest = c.deleteQueue[1]
 	assert.Equal(t, newPodIdentifier("resource_attribute", "k8s.pod.uid", "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"), deleteRequest.id)
-	assert.Equal(t, "podC", deleteRequest.podName)
+	assert.Equal(t, "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", deleteRequest.podUID)
 	assert.False(t, deleteRequest.ts.Before(tsBeforeDelete))
 	assert.False(t, deleteRequest.ts.After(time.Now()))
 }
@@ -568,21 +569,6 @@ func TestNodeDelete(t *testing.T) {
 	assert.Empty(t, c.Nodes)
 }
 
-func TestDeleteQueue(t *testing.T) {
-	c, _ := newTestClient(t)
-	podAddAndUpdateTest(t, c, c.handlePodAdd)
-	assert.Len(t, c.Pods, 5)
-	assert.Equal(t, "1.1.1.1", c.Pods[newPodIdentifier("connection", "k8s.pod.ip", "1.1.1.1")].Address)
-
-	// delete pod
-	pod := &api_v1.Pod{}
-	pod.Name = "podB"
-	pod.Status.PodIP = "1.1.1.1"
-	c.handlePodDelete(pod)
-	assert.Len(t, c.Pods, 5)
-	assert.Len(t, c.deleteQueue, 3)
-}
-
 func TestDeleteLoop(t *testing.T) {
 	// go c.deleteLoop(time.Second * 1)
 	c, _ := newTestClient(t)
@@ -1392,6 +1378,130 @@ func TestNamespaceExtractionRules(t *testing.T) {
 	}
 }
 
+func TestDeleteQueue(t *testing.T) {
+	makePodIdentifiers := func(pod *api_v1.Pod) []PodIdentifier {
+		return []PodIdentifier{
+			newPodIdentifier("resource_attribute", "k8s.pod.uid", string(pod.UID)),
+			{
+				{
+					Source: AssociationSource{
+						From: "resource_attribute",
+						Name: "k8s.namespace.name",
+					},
+					Value: "ns",
+				},
+				{
+					Source: AssociationSource{
+						From: "resource_attribute",
+						Name: "k8s.pod.name",
+					},
+					Value: "abc-0",
+				},
+			},
+			newPodIdentifier("resource_attribute", "k8s.pod.ip", "10.0.0.1"),
+			newPodIdentifier("connection", "", "10.0.0.1"),
+		}
+	}
+
+	makePod := func(podUid string) *api_v1.Pod {
+		pod1 := &api_v1.Pod{
+			ObjectMeta: meta_v1.ObjectMeta{
+				Name:      "abc-0",
+				Namespace: "ns",
+				UID:       types.UID(podUid),
+			},
+			Status: api_v1.PodStatus{
+				PodIP: "10.0.0.1",
+			},
+		}
+		return pod1
+	}
+
+	c, _ := newTestClient(t)
+	doAssertions := func(pod *api_v1.Pod) {
+		podIdentifiers := makePodIdentifiers(pod)
+		for _, id := range podIdentifiers {
+			foundPod, ok := c.Pods[id]
+			assert.True(t, ok, "Pod should be present in c.Pods for identifier %v", id)
+			assert.Equal(t, pod.UID, types.UID(foundPod.PodUID))
+			assert.Equal(t, "ns", foundPod.Namespace)
+			assert.Equal(t, "abc-0", foundPod.Name)
+			assert.Equal(t, "10.0.0.1", foundPod.Address)
+		}
+	}
+
+	// Clear the pods map to start fresh...
+	c.Pods = make(map[PodIdentifier]*Pod)
+	// Set associations to match what we have configured for our OpenTelemetry Collector.
+	c.Associations = []Association{
+		{
+			Sources: []AssociationSource{
+				{
+					From: "resource_attribute",
+					Name: "k8s.pod.uid",
+				},
+			},
+		},
+		{
+			Sources: []AssociationSource{
+				{
+					From: "resource_attribute",
+					Name: "k8s.namespace.name",
+				},
+				{
+					From: "resource_attribute",
+					Name: "k8s.pod.name",
+				},
+			},
+		},
+		{
+			Sources: []AssociationSource{
+				{
+					From: "resource_attribute",
+					Name: "k8s.pod.ip",
+				},
+			},
+		},
+		{
+			Sources: []AssociationSource{
+				{
+					From: "connection",
+				},
+			},
+		},
+	}
+
+	// Add a pod and verify that we can find it by all identifiers.
+	pod1 := makePod("12345678-1234-1234-1234-123456789abc")
+	c.handlePodAdd(pod1)
+	doAssertions(pod1)
+	assert.Len(t, c.Pods, 4)
+
+	// Delete a pod and verify that we can still found it by all identifiers
+	c.handlePodDelete(pod1)
+	doAssertions(pod1)
+	assert.Len(t, c.Pods, 4)
+
+	// Add a pod with the same values as pod1 except for UID, and verify that we can find it by all identifiers.
+	pod2 := makePod("87654321-4321-4321-4321-cba987654321")
+	c.handlePodAdd(pod2)
+	doAssertions(pod2)
+	assert.Len(t, c.Pods, 5) // 4 from pod2 + 1 from pod1 (the pod UID identifier)
+
+	c.deleteLoopProcessing(0 * time.Second)
+	assert.Len(t, c.Pods, 4) // Only mappings for pod2 remain
+	doAssertions(pod2)
+
+	// Delete pod2 and verify that it gets removed after the next delete loop housekeeping.
+	c.handlePodDelete(pod2)
+	assert.Len(t, c.Pods, 4) // Only mappings for pod2 remain
+	doAssertions(pod2)
+
+	// Delete loop processing should remove mappings for pod2.
+	c.deleteLoopProcessing(0 * time.Second)
+	assert.Empty(t, c.Pods) // No more mappings
+}
+
 func TestNodeExtractionRules(t *testing.T) {
 	c, _ := newTestClientWithRulesAndFilters(t, Filters{})
 

processor/k8sattributesprocessor/internal/kube/kube.go

@@ -181,9 +181,9 @@ type Node struct {
 type deleteRequest struct {
 	// id is identifier (IP address or Pod UID) of pod to remove from pods map
 	id PodIdentifier
-	// name contains name of pod to remove from pods map
-	podName string
-	ts      time.Time
+	// contains uid of pod to remove from pods map
+	podUID string
+	ts     time.Time
 }
 
 // Filters is used to instruct the client on how to filter out k8s pods.