[processor/k8sattributes]: apply attribute value from retrieved k8s object if original value is empty (open-telemetry#36466)

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

This PR extends the previous check whether an attribute value has not
been present with a check for an empty value in the original resource
attributes. This way, attributes, such as `k8s.namespace.name` will be
set based on the retrieved kubernetes resource also if the original
value has been set to e.g. an empty string

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

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

---------

Signed-off-by: Florian Bacher <[email protected]>
Co-authored-by: Christos Markou <[email protected]>

Author: 2 people

.chloggen/k8s-override-empty-attributes.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: k8sattributesprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Override extracted k8s attributes if original value has been empty
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [36373]
+
+# (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/processor.go

@@ -144,9 +144,7 @@ func (kp *kubernetesprocessor) processResource(ctx context.Context, resource pco
 
 	for i := range podIdentifierValue {
 		if podIdentifierValue[i].Source.From == kube.ConnectionSource && podIdentifierValue[i].Value != "" {
-			if _, found := resource.Attributes().Get(kube.K8sIPLabelName); !found {
-				resource.Attributes().PutStr(kube.K8sIPLabelName, podIdentifierValue[i].Value)
-			}
+			setResourceAttribute(resource.Attributes(), kube.K8sIPLabelName, podIdentifierValue[i].Value)
 			break
 		}
 	}
@@ -161,9 +159,7 @@ func (kp *kubernetesprocessor) processResource(ctx context.Context, resource pco
 			kp.logger.Debug("getting the pod", zap.Any("pod", pod))
 
 			for key, val := range pod.Attributes {
-				if _, found := resource.Attributes().Get(key); !found {
-					resource.Attributes().PutStr(key, val)
-				}
+				setResourceAttribute(resource.Attributes(), key, val)
 			}
 			kp.addContainerAttributes(resource.Attributes(), pod)
 		}
@@ -173,29 +169,30 @@ func (kp *kubernetesprocessor) processResource(ctx context.Context, resource pco
 	if namespace != "" {
 		attrsToAdd := kp.getAttributesForPodsNamespace(namespace)
 		for key, val := range attrsToAdd {
-			if _, found := resource.Attributes().Get(key); !found {
-				resource.Attributes().PutStr(key, val)
-			}
+			setResourceAttribute(resource.Attributes(), key, val)
 		}
 	}
 
 	nodeName := getNodeName(pod, resource.Attributes())
 	if nodeName != "" {
 		attrsToAdd := kp.getAttributesForPodsNode(nodeName)
 		for key, val := range attrsToAdd {
-			if _, found := resource.Attributes().Get(key); !found {
-				resource.Attributes().PutStr(key, val)
-			}
+			setResourceAttribute(resource.Attributes(), key, val)
 		}
 		nodeUID := kp.getUIDForPodsNode(nodeName)
 		if nodeUID != "" {
-			if _, found := resource.Attributes().Get(conventions.AttributeK8SNodeUID); !found {
-				resource.Attributes().PutStr(conventions.AttributeK8SNodeUID, nodeUID)
-			}
+			setResourceAttribute(resource.Attributes(), conventions.AttributeK8SNodeUID, nodeUID)
 		}
 	}
 }
 
+func setResourceAttribute(attributes pcommon.Map, key string, val string) {
+	attr, found := attributes.Get(key)
+	if !found || attr.AsString() == "" {
+		attributes.PutStr(key, val)
+	}
+}
+
 func getNamespace(pod *kube.Pod, resAttrs pcommon.Map) string {
 	if pod != nil && pod.Namespace != "" {
 		return pod.Namespace
@@ -233,19 +230,13 @@ func (kp *kubernetesprocessor) addContainerAttributes(attrs pcommon.Map, pod *ku
 		return
 	}
 	if containerSpec.Name != "" {
-		if _, found := attrs.Get(conventions.AttributeK8SContainerName); !found {
-			attrs.PutStr(conventions.AttributeK8SContainerName, containerSpec.Name)
-		}
+		setResourceAttribute(attrs, conventions.AttributeK8SContainerName, containerSpec.Name)
 	}
 	if containerSpec.ImageName != "" {
-		if _, found := attrs.Get(conventions.AttributeContainerImageName); !found {
-			attrs.PutStr(conventions.AttributeContainerImageName, containerSpec.ImageName)
-		}
+		setResourceAttribute(attrs, conventions.AttributeContainerImageName, containerSpec.ImageName)
 	}
 	if containerSpec.ImageTag != "" {
-		if _, found := attrs.Get(conventions.AttributeContainerImageTag); !found {
-			attrs.PutStr(conventions.AttributeContainerImageTag, containerSpec.ImageTag)
-		}
+		setResourceAttribute(attrs, conventions.AttributeContainerImageTag, containerSpec.ImageTag)
 	}
 	// attempt to get container ID from restart count
 	runID := -1

processor/k8sattributesprocessor/processor_test.go

@@ -1635,3 +1635,62 @@ func (nh *nopHost) GetExtensions() map[component.ID]component.Component {
 func (nh *nopHost) Report(event *componentstatus.Event) {
 	nh.reportFunc(event)
 }
+
+func Test_setResourceAttribute(t *testing.T) {
+	tests := []struct {
+		name       string
+		attributes func() pcommon.Map
+		key        string
+		val        string
+		wantAttrs  func() pcommon.Map
+	}{
+		{
+			name:       "attribute not present - add value",
+			attributes: pcommon.NewMap,
+			key:        "foo",
+			val:        "bar",
+			wantAttrs: func() pcommon.Map {
+				m := pcommon.NewMap()
+				m.PutStr("foo", "bar")
+				return m
+			},
+		},
+		{
+			name: "attribute present with non-empty value - do not overwrite value",
+			attributes: func() pcommon.Map {
+				m := pcommon.NewMap()
+				m.PutStr("foo", "bar")
+				return m
+			},
+			key: "foo",
+			val: "baz",
+			wantAttrs: func() pcommon.Map {
+				m := pcommon.NewMap()
+				m.PutStr("foo", "bar")
+				return m
+			},
+		},
+		{
+			name: "attribute present with empty value - set value",
+			attributes: func() pcommon.Map {
+				m := pcommon.NewMap()
+				m.PutStr("foo", "")
+				return m
+			},
+			key: "foo",
+			val: "bar",
+			wantAttrs: func() pcommon.Map {
+				m := pcommon.NewMap()
+				m.PutStr("foo", "bar")
+				return m
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			attrs := tt.attributes()
+			setResourceAttribute(attrs, tt.key, tt.val)
+			require.Equal(t, tt.wantAttrs(), attrs)
+		})
+	}
+}