[processor/k8s_attributes] Fix memory leak in k8s_attributes map (open-telemetry#48987)

[processor/k8s_attributes] Fix memory leak in k8s_attributes map due to
missing IP or other dynamic attributes from Pod deletion event object.

#### Component(s)

[processor/k8s_attributes]

#### Description
In environments with high pod churn or rapid scale-down, the
`k8sattributes` processor can leak pod IP-based cache entries
(`connection: <IP>` and `resource_attribute: k8s.pod.ip`) in the
internal `c.Pods` map. This leads to unbounded memory growth and a
stable baseline of leaked keys even after the actual pods have been
scaled down to zero.

#### Root Cause
When a pod is deleted, the `WatchClient`'s
[`forgetPod`](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/processor/k8sattributesprocessor/internal/kube/client.go#L1635)
method is called to purge the cached entries:

```go
func (c *WatchClient) forgetPod(pod *api_v1.Pod) {
	podToRemove := c.podFromAPI(pod)
	identifiers := c.getIdentifiersFromAssoc(podToRemove)
	for i := range identifiers {
		id := identifiers[i]
		p, ok := c.GetPod(id)

		if ok && p.PodUID == string(pod.UID) {
			c.appendDeleteQueue(id, p.PodUID)
		}
	}
}
```

1. `forgetPod` builds `podToRemove` from the incoming delete event
payload `pod`.
2. It then calls `getIdentifiersFromAssoc(podToRemove)` to determine
which keys to add to the delete queue.
3. If the CNI has already reclaimed/cleared the Pod's IP address by the
time the final `DELETE` event is dispatched (or if the status payload is
incomplete), `pod.Status.PodIP` will be empty (`""`).
4. As a result, `getIdentifiersFromAssoc` **only** generates the
UID-based identifier (`resource_attribute: k8s.pod.uid`). The IP-based
identifiers (`connection: <IP>` and `resource_attribute: k8s.pod.ip`)
are skipped.
5. Consequently, only the UID-based key is queued for deletion, while
the IP-based keys are never cleared and are permanently leaked in the
`c.Pods` map.

#### Proposed Solution
Instead of relying on the incoming delete event status to determine the
keys to delete, `forgetPod` should look up the cached pod from `c.Pods`
using the pod's UID (which is always present in the event). The cached
pod object is guaranteed to contain the IP address and all attributes as
they were stored during the pod's lifecycle.

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

<!--Describe what testing was performed and which tests were added.-->
#### Testing

Added a unit test case that catches the issue of a Pod object from a
deletion event that is missing IP fields.

<!--Describe the documentation added.-->
#### Documentation

Added a changelog entry for a bug fix.

<!--Authorship attestation. See AGENTS.md for details. AI agents must
not check this box on behalf
of the user; the human author must check it themselves before the PR is
ready for review.-->
#### Authorship

- [ X ] I, a human, wrote this pull request description myself.

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

Author: giuliano-sider

.chloggen/fix_k8sattributes_cache_key_leak.yaml

@@ -0,0 +1,19 @@
+# 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. receiver/filelog)
+component: processor/k8s_attributes
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Fix cache key memory leak in k8sattributesprocessor when a Pod's IP is missing or cleared from the delete event"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [48986]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+subtext: |
+  When a Pod is deleted, if its IP address is missing or already cleared from the delete event status,
+  the processor now looks up the cached pod by its UID to retrieve the stored IP and correctly queue
+  all associated keys for deletion.
+
+change_logs: [user]

processor/k8sattributesprocessor/internal/kube/client.go

@@ -1655,8 +1655,27 @@ func (c *WatchClient) addOrUpdatePod(pod *api_v1.Pod) {
 }
 
 func (c *WatchClient) forgetPod(pod *api_v1.Pod) {
-	podToRemove := c.podFromAPI(pod)
-	identifiers := c.getIdentifiersFromAssoc(podToRemove)
+	// Look up the cached pod using its UID. Unlike dynamic status attributes (such as IP addresses)
+	// which may be cleared or missing in the final DELETE event payload, the Pod UID is immutable
+	// and guaranteed to be present. Using the cached pod ensures we generate and clean up all keys
+	// under which the pod was originally registered.
+	uidKey := PodIdentifier{
+		PodIdentifierAttributeFromResourceAttribute(string(conventions.K8SPodUIDKey), string(pod.UID)),
+	}
+	c.m.RLock()
+	cachedPod, ok := c.Pods[uidKey]
+	c.m.RUnlock()
+
+	var identifiers []PodIdentifier
+	if ok {
+		identifiers = c.getIdentifiersFromAssoc(cachedPod)
+	} else {
+		// Fallback: if the pod was never added to the cache (e.g. startup/informer sync edge cases),
+		// generate deletion keys from the incoming delete event payload directly.
+		podToRemove := c.podFromAPI(pod)
+		identifiers = c.getIdentifiersFromAssoc(podToRemove)
+	}
+
 	for i := range identifiers {
 		id := identifiers[i]
 		p, ok := c.GetPod(id)

processor/k8sattributesprocessor/internal/kube/client_test.go

@@ -4874,3 +4874,48 @@ func TestExtractPodAttributesClusterUIDRace(t *testing.T) {
 
 	wg.Wait()
 }
+
+func TestPodDeleteIPMissingFromDeleteEvent(t *testing.T) {
+	c, _ := newTestClient(t)
+
+	pod := &api_v1.Pod{}
+	pod.Name = "podLeak"
+	pod.Status.PodIP = "4.4.4.4"
+	pod.UID = "uid-leak-test"
+	c.handlePodAdd(pod)
+
+	// Map should have 3 keys for this pod
+	assert.Contains(t, c.Pods, newPodIdentifier("resource_attribute", "k8s.pod.uid", "uid-leak-test"))
+	assert.Contains(t, c.Pods, newPodIdentifier("connection", "k8s.pod.ip", "4.4.4.4"))
+	assert.Contains(t, c.Pods, newPodIdentifier("resource_attribute", "k8s.pod.ip", "4.4.4.4"))
+
+	// Clear the delete queue
+	c.deleteQueue = c.deleteQueue[:0]
+
+	// Simulate delete event where PodIP is missing/empty!
+	deletePod := &api_v1.Pod{}
+	deletePod.Name = "podLeak"
+	deletePod.UID = "uid-leak-test"
+	deletePod.Status.PodIP = ""
+
+	c.handlePodDelete(deletePod)
+
+	// In the bug state, only the UID-based keys are queued for deletion, leaving the IP-based keys leaked.
+	// We expect all unique key patterns (UID, connection IP, and Pod IP) to be queued for deletion.
+	var ids []PodIdentifier
+	for _, req := range c.deleteQueue {
+		assert.Equal(t, "uid-leak-test", req.podUID)
+		ids = append(ids, req.id)
+	}
+
+	// Verify that each of the three key patterns was queued at least once
+	assert.Contains(t, ids, newPodIdentifier("resource_attribute", "k8s.pod.uid", "uid-leak-test"))
+	assert.Contains(t, ids, newPodIdentifier("connection", "k8s.pod.ip", "4.4.4.4"))
+	assert.Contains(t, ids, newPodIdentifier("resource_attribute", "k8s.pod.ip", "4.4.4.4"))
+
+	// Run the sweep logic to process all queued deletions immediately
+	c.deleteLoopProcessing(0)
+
+	// In the fixed state, all keys associated with the pod should be successfully cleaned up from the cache.
+	assert.Empty(t, c.Pods)
+}