[processor/k8sattributes] Add support for exposing k8s.pod.ip metadata (

#33440)

This PR adds support for exporting the `k8s.pod.ip` as part of the
metadata added by the processor.

Fixes #32960

#### Testing notes

See
#33440 (comment)

---------

Signed-off-by: ChrsMark <[email protected]>
Co-authored-by: Curtis Robert <[email protected]>

.chloggen/add_k8s_pod_ip.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: enhancement
+
+# 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: "Add support for exposing `k8s.pod.ip` as a resource attribute"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [32960]
+
+# (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: [user]

processor/k8sattributesprocessor/config.go

@@ -84,7 +84,7 @@ func (cfg *Config) Validate() error {
 	for _, field := range cfg.Extract.Metadata {
 		switch field {
 		case conventions.AttributeK8SNamespaceName, conventions.AttributeK8SPodName, conventions.AttributeK8SPodUID,
-			specPodHostName, metadataPodStartTime,
+			specPodHostName, metadataPodStartTime, metadataPodIP,
 			conventions.AttributeK8SDeploymentName, conventions.AttributeK8SDeploymentUID,
 			conventions.AttributeK8SReplicaSetName, conventions.AttributeK8SReplicaSetUID,
 			conventions.AttributeK8SDaemonSetName, conventions.AttributeK8SDaemonSetUID,

processor/k8sattributesprocessor/config_test.go

@@ -40,7 +40,7 @@ func TestLoadConfig(t *testing.T) {
 				APIConfig:   k8sconfig.APIConfig{AuthType: k8sconfig.AuthTypeKubeConfig},
 				Passthrough: false,
 				Extract: ExtractConfig{
-					Metadata: []string{"k8s.pod.name", "k8s.pod.uid", "k8s.deployment.name", "k8s.namespace.name", "k8s.node.name", "k8s.pod.start_time", "k8s.cluster.uid"},
+					Metadata: []string{"k8s.pod.name", "k8s.pod.uid", "k8s.pod.ip", "k8s.deployment.name", "k8s.namespace.name", "k8s.node.name", "k8s.pod.start_time", "k8s.cluster.uid"},
 					Annotations: []FieldExtractConfig{
 						{TagName: "a1", Key: "annotation-one", From: "pod"},
 						{TagName: "a2", Key: "annotation-two", Regex: "field=(?P<value>.+)", From: kube.MetadataFromPod},

processor/k8sattributesprocessor/documentation.md

@@ -22,6 +22,7 @@
 | k8s.node.name | The name of the Node. | Any Str | true |
 | k8s.node.uid | The UID of the Node. | Any Str | false |
 | k8s.pod.hostname | The hostname of the Pod. | Any Str | false |
+| k8s.pod.ip | The IP address of the Pod. | Any Str | false |
 | k8s.pod.name | The name of the Pod. | Any Str | true |
 | k8s.pod.start_time | The start time of the Pod. | Any Str | true |
 | k8s.pod.uid | The UID of the Pod. | Any Str | true |

processor/k8sattributesprocessor/e2e_test.go

@@ -32,6 +32,7 @@ const (
 	equal = iota
 	regex
 	exist
+	shouldnotexist
 	testKubeConfig = "/tmp/kube-config-otelcol-e2e-testing"
 	uidRe          = "[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{12}"
 	startTimeRe    = "^\\\\d{4}-\\\\d{2}-\\\\d{2}T\\\\d{2}%3A\\\\d{2}%3A\\\\d{2}(?:%2E\\\\d+)?[A-Z]?(?:[+.-](?:08%3A\\\\d{2}|\\\\d{2}[A-Z]))?$"
@@ -722,6 +723,146 @@ func TestE2E_MixRBAC(t *testing.T) {
 	}
 }
 
+// Test with `filter::namespace` set and only role binding to collector's SA. We can't get node and namespace labels/annotations.
+// While `k8s.pod.ip` is not set in `k8sattributes:extract:metadata` and the `pod_association` is not `connection`
+// we expect that the `k8s.pod.ip` metadata is not added.
+func TestE2E_NamespacedRBACNoPodIP(t *testing.T) {
+	testDir := filepath.Join("testdata", "e2e", "namespaced_rbac_no_pod_ip")
+
+	k8sClient, err := k8stest.NewK8sClient(testKubeConfig)
+	require.NoError(t, err)
+
+	nsFile := filepath.Join(testDir, "namespace.yaml")
+	buf, err := os.ReadFile(nsFile)
+	require.NoErrorf(t, err, "failed to read namespace object file %s", nsFile)
+	nsObj, err := k8stest.CreateObject(k8sClient, buf)
+	require.NoErrorf(t, err, "failed to create k8s namespace from file %s", nsFile)
+	nsName := nsObj.GetName()
+	defer func() {
+		require.NoErrorf(t, k8stest.DeleteObject(k8sClient, nsObj), "failed to delete namespace %s", nsName)
+	}()
+
+	metricsConsumer := new(consumertest.MetricsSink)
+	tracesConsumer := new(consumertest.TracesSink)
+	logsConsumer := new(consumertest.LogsSink)
+	shutdownSinks := startUpSinks(t, metricsConsumer, tracesConsumer, logsConsumer)
+	defer shutdownSinks()
+
+	testID := uuid.NewString()[:8]
+	collectorObjs := k8stest.CreateCollectorObjects(t, k8sClient, testID, filepath.Join(testDir, "collector"))
+	createTeleOpts := &k8stest.TelemetrygenCreateOpts{
+		ManifestsDir: filepath.Join(testDir, "telemetrygen"),
+		TestID:       testID,
+		OtlpEndpoint: fmt.Sprintf("otelcol-%s.%s:4317", testID, nsName),
+		DataTypes:    []string{"metrics", "logs", "traces"},
+	}
+	telemetryGenObjs, telemetryGenObjInfos := k8stest.CreateTelemetryGenObjects(t, k8sClient, createTeleOpts)
+	defer func() {
+		for _, obj := range append(collectorObjs, telemetryGenObjs...) {
+			require.NoErrorf(t, k8stest.DeleteObject(k8sClient, obj), "failed to delete object %s", obj.GetName())
+		}
+	}()
+
+	for _, info := range telemetryGenObjInfos {
+		k8stest.WaitForTelemetryGenToStart(t, k8sClient, info.Namespace, info.PodLabelSelectors, info.Workload, info.DataType)
+	}
+
+	wantEntries := 20 // Minimal number of metrics/traces/logs to wait for.
+	waitForData(t, wantEntries, metricsConsumer, tracesConsumer, logsConsumer)
+
+	tcs := []struct {
+		name     string
+		dataType component.DataType
+		service  string
+		attrs    map[string]*expectedValue
+	}{
+		{
+			name:     "traces-deployment",
+			dataType: component.DataTypeTraces,
+			service:  "test-traces-deployment",
+			attrs: map[string]*expectedValue{
+				"k8s.pod.name":             newExpectedValue(regex, "telemetrygen-"+testID+"-traces-deployment-[a-z0-9]*-[a-z0-9]*"),
+				"k8s.pod.ip":               newExpectedValue(shouldnotexist, ""),
+				"k8s.pod.uid":              newExpectedValue(regex, uidRe),
+				"k8s.pod.start_time":       newExpectedValue(exist, startTimeRe),
+				"k8s.node.name":            newExpectedValue(exist, ""),
+				"k8s.namespace.name":       newExpectedValue(equal, nsName),
+				"k8s.deployment.name":      newExpectedValue(equal, "telemetrygen-"+testID+"-traces-deployment"),
+				"k8s.deployment.uid":       newExpectedValue(regex, uidRe),
+				"k8s.replicaset.name":      newExpectedValue(regex, "telemetrygen-"+testID+"-traces-deployment-[a-z0-9]*"),
+				"k8s.replicaset.uid":       newExpectedValue(regex, uidRe),
+				"k8s.annotations.workload": newExpectedValue(equal, "deployment"),
+				"k8s.labels.app":           newExpectedValue(equal, "telemetrygen-"+testID+"-traces-deployment"),
+				"k8s.container.name":       newExpectedValue(equal, "telemetrygen"),
+				"container.image.name":     newExpectedValue(equal, "ghcr.io/open-telemetry/opentelemetry-collector-contrib/telemetrygen"),
+				"container.image.tag":      newExpectedValue(equal, "latest"),
+				"container.id":             newExpectedValue(exist, ""),
+			},
+		},
+		{
+			name:     "metrics-deployment",
+			dataType: component.DataTypeMetrics,
+			service:  "test-metrics-deployment",
+			attrs: map[string]*expectedValue{
+				"k8s.pod.name":             newExpectedValue(regex, "telemetrygen-"+testID+"-metrics-deployment-[a-z0-9]*-[a-z0-9]*"),
+				"k8s.pod.ip":               newExpectedValue(shouldnotexist, ""),
+				"k8s.pod.uid":              newExpectedValue(regex, uidRe),
+				"k8s.pod.start_time":       newExpectedValue(exist, startTimeRe),
+				"k8s.node.name":            newExpectedValue(exist, ""),
+				"k8s.namespace.name":       newExpectedValue(equal, nsName),
+				"k8s.deployment.name":      newExpectedValue(equal, "telemetrygen-"+testID+"-metrics-deployment"),
+				"k8s.deployment.uid":       newExpectedValue(regex, uidRe),
+				"k8s.replicaset.name":      newExpectedValue(regex, "telemetrygen-"+testID+"-metrics-deployment-[a-z0-9]*"),
+				"k8s.replicaset.uid":       newExpectedValue(regex, uidRe),
+				"k8s.annotations.workload": newExpectedValue(equal, "deployment"),
+				"k8s.labels.app":           newExpectedValue(equal, "telemetrygen-"+testID+"-metrics-deployment"),
+				"k8s.container.name":       newExpectedValue(equal, "telemetrygen"),
+				"container.image.name":     newExpectedValue(equal, "ghcr.io/open-telemetry/opentelemetry-collector-contrib/telemetrygen"),
+				"container.image.tag":      newExpectedValue(equal, "latest"),
+				"container.id":             newExpectedValue(exist, ""),
+			},
+		},
+		{
+			name:     "logs-deployment",
+			dataType: component.DataTypeLogs,
+			service:  "test-logs-deployment",
+			attrs: map[string]*expectedValue{
+				"k8s.pod.name":             newExpectedValue(regex, "telemetrygen-"+testID+"-logs-deployment-[a-z0-9]*-[a-z0-9]*"),
+				"k8s.pod.ip":               newExpectedValue(shouldnotexist, ""),
+				"k8s.pod.uid":              newExpectedValue(regex, uidRe),
+				"k8s.pod.start_time":       newExpectedValue(exist, startTimeRe),
+				"k8s.node.name":            newExpectedValue(exist, ""),
+				"k8s.namespace.name":       newExpectedValue(equal, nsName),
+				"k8s.deployment.name":      newExpectedValue(equal, "telemetrygen-"+testID+"-logs-deployment"),
+				"k8s.deployment.uid":       newExpectedValue(regex, uidRe),
+				"k8s.replicaset.name":      newExpectedValue(regex, "telemetrygen-"+testID+"-logs-deployment-[a-z0-9]*"),
+				"k8s.replicaset.uid":       newExpectedValue(regex, uidRe),
+				"k8s.annotations.workload": newExpectedValue(equal, "deployment"),
+				"k8s.labels.app":           newExpectedValue(equal, "telemetrygen-"+testID+"-logs-deployment"),
+				"k8s.container.name":       newExpectedValue(equal, "telemetrygen"),
+				"container.image.name":     newExpectedValue(equal, "ghcr.io/open-telemetry/opentelemetry-collector-contrib/telemetrygen"),
+				"container.image.tag":      newExpectedValue(equal, "latest"),
+				"container.id":             newExpectedValue(exist, ""),
+			},
+		},
+	}
+
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			switch tc.dataType {
+			case component.DataTypeTraces:
+				scanTracesForAttributes(t, tracesConsumer, tc.service, tc.attrs)
+			case component.DataTypeMetrics:
+				scanMetricsForAttributes(t, metricsConsumer, tc.service, tc.attrs)
+			case component.DataTypeLogs:
+				scanLogsForAttributes(t, logsConsumer, tc.service, tc.attrs)
+			default:
+				t.Fatalf("unknown data type %s", tc.dataType)
+			}
+		})
+	}
+}
+
 func scanTracesForAttributes(t *testing.T, ts *consumertest.TracesSink, expectedService string,
 	kvs map[string]*expectedValue) {
 	// Iterate over the received set of traces starting from the most recent entries due to a bug in the processor:
@@ -787,8 +928,13 @@ func scanLogsForAttributes(t *testing.T, ls *consumertest.LogsSink, expectedServ
 
 func resourceHasAttributes(resource pcommon.Resource, kvs map[string]*expectedValue) error {
 	foundAttrs := make(map[string]bool)
-	for k := range kvs {
-		foundAttrs[k] = false
+	shouldNotFoundAttrs := make(map[string]bool)
+	for k, v := range kvs {
+		if v.mode != shouldnotexist {
+			foundAttrs[k] = false
+			continue
+		}
+		shouldNotFoundAttrs[k] = false
 	}
 
 	resource.Attributes().Range(
@@ -806,6 +952,8 @@ func resourceHasAttributes(resource pcommon.Resource, kvs map[string]*expectedVa
 					}
 				case exist:
 					foundAttrs[k] = true
+				case shouldnotexist:
+					shouldNotFoundAttrs[k] = true
 				}
 
 			}
@@ -814,6 +962,11 @@ func resourceHasAttributes(resource pcommon.Resource, kvs map[string]*expectedVa
 	)
 
 	var err error
+	for k, v := range shouldNotFoundAttrs {
+		if v {
+			err = multierr.Append(err, fmt.Errorf("%v attribute should not be added", k))
+		}
+	}
 	for k, v := range foundAttrs {
 		if !v {
 			err = multierr.Append(err, fmt.Errorf("%v attribute not found", k))

processor/k8sattributesprocessor/internal/kube/client.go

@@ -425,6 +425,10 @@ func (c *WatchClient) extractPodAttributes(pod *api_v1.Pod) map[string]string {
 		tags[tagHostName] = pod.Spec.Hostname
 	}
 
+	if c.Rules.PodIP {
+		tags[K8sIPLabelName] = pod.Status.PodIP
+	}
+
 	if c.Rules.Namespace {
 		tags[conventions.AttributeK8SNamespaceName] = pod.GetNamespace()
 	}

processor/k8sattributesprocessor/internal/kube/client_test.go

@@ -814,6 +814,7 @@ func TestExtractionRules(t *testing.T) {
 			PodName:        true,
 			PodUID:         true,
 			PodHostName:    true,
+			PodIP:          true,
 			Node:           true,
 			StartTime:      true,
 		},
@@ -825,6 +826,7 @@ func TestExtractionRules(t *testing.T) {
 			"k8s.pod.name":        "auth-service-abc12-xyz3",
 			"k8s.pod.hostname":    "host1",
 			"k8s.pod.uid":         "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
+			"k8s.pod.ip":          "1.1.1.1",
 			"k8s.pod.start_time": func() string {
 				b, err := pod.GetCreationTimestamp().MarshalText()
 				require.NoError(t, err)

processor/k8sattributesprocessor/internal/kube/kube.go

@@ -205,6 +205,7 @@ type ExtractionRules struct {
 	PodName            bool
 	PodUID             bool
 	PodHostName        bool
+	PodIP              bool
 	ReplicaSetID       bool
 	ReplicaSetName     bool
 	StatefulSetUID     bool

processor/k8sattributesprocessor/internal/metadata/testdata/config.yaml

@@ -33,6 +33,8 @@ all_set:
       enabled: true
     k8s.pod.hostname:
       enabled: true
+    k8s.pod.ip:
+      enabled: true
     k8s.pod.name:
       enabled: true
     k8s.pod.start_time:
@@ -81,6 +83,8 @@ none_set:
       enabled: false
     k8s.pod.hostname:
       enabled: false
+    k8s.pod.ip:
+      enabled: false
     k8s.pod.name:
       enabled: false
     k8s.pod.start_time:

processor/k8sattributesprocessor/metadata.yaml

@@ -34,6 +34,10 @@ resource_attributes:
     description: The start time of the Pod.
     type: string
     enabled: true
+  k8s.pod.ip:
+    description: The IP address of the Pod.
+    type: string
+    enabled: false
   k8s.deployment.name:
     description: The name of the Deployment.
     type: string