[processor/k8s_attributes] Share processor across signal types (#47686)

#### Description

Allow sharing of k8sattributes processor instances between pipelines.
The implementation uses sharedcomponent and is hidden behind a feature
gate.

#### Link to tracking issue
Partially fixes
#36234.

#### Testing

Added a e2e test that:

- Enabled the feature gate
- Creates two processors with different configuration, one of which is
shared between two pipelines
- Verifies that data emitted by both is correct
- Verifies that the shared processor is actually shared by looking at
logs

#### Documentation

Autogenerated feature gate documentation.

Author: swiatekm

.chloggen/k8sattributesprocessor-sharedcomponent.yaml

@@ -0,0 +1,30 @@
+# 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. 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: Allow k8sattributes processors to be shared between pipelines
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [36234]
+
+# (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: |
+  When the processor.k8sattributes.ShareProcessorBetweenPipelines feature flag is enabled, k8sattributes processors
+  using the same configuration are shared between pipelines. This reduces the local cache size and the number of 
+  connections to the K8s API Server.
+
+# 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: []

internal/datadog/e2e/go.mod

@@ -336,6 +336,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig v0.150.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/metadataproviders v0.150.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/pdatautil v0.150.0 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent v0.150.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/datadog v0.150.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/experimentalmetricmetadata v0.150.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl v0.150.0 // indirect
@@ -536,6 +537,8 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/k8sa
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/pdatautil => ../../pdatautil
 
+replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent => ../../sharedcomponent
+
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/core/xidutils => ../../../pkg/core/xidutils
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor => ../../../processor/deltatocumulativeprocessor

pkg/xk8stest/k8s_collector.go

@@ -232,3 +232,24 @@ func fetchContainerLogs(ctx context.Context, coreClient corev1client.CoreV1Inter
 	}
 	return strings.TrimRight(string(logs), "\n")
 }
+
+// FetchPodLogs returns the full log output for the first running pod matching
+// the given labels in the specified namespace. It is intended for e2e test
+// assertions that inspect collector behavior via its log output.
+func FetchPodLogs(t *testing.T, client *K8sClient, namespace string, podLabels map[string]any) string {
+	t.Helper()
+	coreClient, err := corev1client.NewForConfig(client.restConfig)
+	require.NoError(t, err, "failed to create core client for pod logs")
+
+	podGVR := schema.GroupVersionResource{Version: "v1", Resource: "pods"}
+	listOptions := metav1.ListOptions{LabelSelector: SelectorFromMap(podLabels).String()}
+	list, err := client.DynamicClient.Resource(podGVR).Namespace(namespace).List(t.Context(), listOptions)
+	require.NoError(t, err, "failed to list pods")
+	require.NotEmpty(t, list.Items, "no pods found matching labels")
+
+	podName := list.Items[0].GetName()
+	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+	defer cancel()
+
+	return fetchContainerLogs(ctx, coreClient, namespace, podName, "", false, nil)
+}

processor/k8sattributesprocessor/documentation.md

@@ -320,6 +320,7 @@ This component has the following feature gates:
 | `k8sattr.labelsAnnotationsSingular.allow` | deprecated | When enabled, default k8s label and annotation resource attribute keys will be singular, instead of plural | v0.125.0 | v0.145.0 | [Link](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/39774) |
 | `processor.k8sattributes.DontEmitV0K8sConventions` | alpha | When enabled, semconv legacy attributes are disabled. | v0.145.0 | N/A | [Link](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/44589) |
 | `processor.k8sattributes.EmitV1K8sConventions` | alpha | When enabled, semconv stable attributes are enabled. | v0.145.0 | N/A | [Link](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/44589) |
+| `processor.k8sattributes.ShareProcessorBetweenPipelines` | alpha | When enabled, processor instances with identical configuration are shared across different signal type pipelines, reducing duplicate Kubernetes API watchers. | v0.150.0 | N/A | [Link](https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/2450) |
 | `processor.k8sattributes.telemetry.disableOldFormatMetrics` | alpha | When enabled, old formatted internal telemetry metrics are disabled. | v0.146.0 | N/A | [Link](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/45871) |
 | `processor.k8sattributes.telemetry.enableNewFormatMetrics` | alpha | When enabled, new formatted internal telemetry metrics are enabled. | v0.146.0 | N/A | [Link](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/45871) |
 

processor/k8sattributesprocessor/e2e_test.go

@@ -12,6 +12,7 @@ import (
 	"os/exec"
 	"path/filepath"
 	"regexp"
+	"strings"
 	"testing"
 	"time"
 
@@ -1968,6 +1969,146 @@ func waitForData(t *testing.T, entriesNum int, mc *consumertest.MetricsSink, tc
 		len(mc.AllMetrics()), len(tc.AllTraces()), len(lc.AllLogs()), len(pc.AllProfiles()), timeoutMinutes)
 }
 
+// TestE2E_SharedProcessor verifies that the k8s_attributes processor is shared across signal types
+// when using the same processor configuration, and that different configurations produce separate
+// processor instances. The test deploys a collector with two k8s_attributes processor configs:
+//   - k8s_attributes/full: extracts pod name, uid, namespace, deployment, node name, and pod labels.
+//     Used by both the traces and metrics pipelines.
+//   - k8s_attributes/minimal: extracts only the pod name.
+//     Used by the logs pipeline.
+//
+// The test then asserts that traces and metrics both receive the full set of attributes (proving the
+// shared processor works for both signal types), while logs receive only the minimal set (proving a
+// different config creates a separate processor instance).
+func TestE2E_SharedProcessor(t *testing.T) {
+	testDir := filepath.Join("testdata", "e2e", "sharedprocessor")
+
+	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)
+
+	testNs := nsObj.GetName()
+	defer func() {
+		require.NoErrorf(t, k8stest.DeleteObject(k8sClient, nsObj), "failed to delete namespace %s", testNs)
+	}()
+
+	metricsConsumer := new(consumertest.MetricsSink)
+	tracesConsumer := new(consumertest.TracesSink)
+	logsConsumer := new(consumertest.LogsSink)
+	profilesConsumer := new(consumertest.ProfilesSink)
+	shutdownSinks := startUpSinks(t, metricsConsumer, tracesConsumer, logsConsumer, profilesConsumer)
+	defer shutdownSinks()
+
+	testID := uuid.NewString()[:8]
+	collectorObjs := k8stest.CreateCollectorObjects(t, k8sClient, testID, filepath.Join(testDir, "collector"), map[string]string{}, "")
+	createTeleOpts := &k8stest.TelemetrygenCreateOpts{
+		ManifestsDir: filepath.Join(testDir, "telemetrygen"),
+		TestID:       testID,
+		OtlpEndpoint: fmt.Sprintf("otelcol-%s.%s:4317", testID, testNs),
+		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 := 128
+	waitForData(t, wantEntries, metricsConsumer, tracesConsumer, logsConsumer, profilesConsumer)
+
+	// Verify sharing by inspecting the collector's log output.
+	// The kube client logs "k8s filtering" once per Start() call. With sharedcomponent,
+	// Start() is called once per unique processor config. We have two configs
+	// (k8s_attributes/full and k8s_attributes/minimal), so we expect exactly 2 occurrences.
+	// Without sharing, each pipeline would get its own processor, producing 3 occurrences
+	// (traces + metrics + logs).
+	collectorPodLabels := map[string]any{
+		"app.kubernetes.io/name":     "opentelemetry-collector",
+		"app.kubernetes.io/instance": "otelcol-" + testID,
+	}
+	podLogs := k8stest.FetchPodLogs(t, k8sClient, testNs, collectorPodLabels)
+	initCount := strings.Count(podLogs, "k8s filtering")
+	assert.Equal(t, 2, initCount,
+		"expected 2 kube client initializations (one per unique processor config), got %d; "+
+			"this suggests processors are not being shared across signal types", initCount)
+
+	// Attributes that the "full" processor (traces + metrics) should add.
+	fullAttrs := map[string]*expectedValue{
+		"k8s.pod.name":        newExpectedValue(regex, "telemetrygen-"+testID+"-.*-deployment-[a-z0-9]*-[a-z0-9]*"),
+		"k8s.pod.uid":         newExpectedValue(regex, uidRe),
+		"k8s.namespace.name":  newExpectedValue(equal, testNs),
+		"k8s.deployment.name": newExpectedValue(regex, "telemetrygen-"+testID+"-.*-deployment"),
+		"k8s.node.name":       newExpectedValue(exist, ""),
+		"k8s.labels.app":      newExpectedValue(regex, "telemetrygen-"+testID+"-.*-deployment"),
+	}
+
+	// Attributes that the "minimal" processor (logs) should add.
+	// It only extracts k8s.pod.name; the other attributes must NOT be present.
+	minimalAttrs := map[string]*expectedValue{
+		"k8s.pod.name":        newExpectedValue(regex, "telemetrygen-"+testID+"-.*-deployment-[a-z0-9]*-[a-z0-9]*"),
+		"k8s.pod.uid":         newExpectedValue(shouldnotexist, ""),
+		"k8s.deployment.name": newExpectedValue(shouldnotexist, ""),
+		"k8s.node.name":       newExpectedValue(shouldnotexist, ""),
+		"k8s.labels.app":      newExpectedValue(shouldnotexist, ""),
+	}
+
+	tcs := []struct {
+		name     string
+		dataType pipeline.Signal
+		service  string
+		attrs    map[string]*expectedValue
+	}{
+		{
+			// Traces use k8s_attributes/full – expect all attributes.
+			name:     "traces-deployment-full",
+			dataType: pipeline.SignalTraces,
+			service:  "test-traces-deployment",
+			attrs:    fullAttrs,
+		},
+		{
+			// Metrics also use k8s_attributes/full – same shared processor,
+			// so they must produce the same set of attributes.
+			name:     "metrics-deployment-full",
+			dataType: pipeline.SignalMetrics,
+			service:  "test-metrics-deployment",
+			attrs:    fullAttrs,
+		},
+		{
+			// Logs use k8s_attributes/minimal – different processor instance,
+			// so only k8s.pod.name should be present.
+			name:     "logs-deployment-minimal",
+			dataType: pipeline.SignalLogs,
+			service:  "test-logs-deployment",
+			attrs:    minimalAttrs,
+		},
+	}
+
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			switch tc.dataType {
+			case pipeline.SignalTraces:
+				scanTracesForAttributes(t, tracesConsumer, tc.service, tc.attrs)
+			case pipeline.SignalMetrics:
+				scanMetricsForAttributes(t, metricsConsumer, tc.service, tc.attrs)
+			case pipeline.SignalLogs:
+				scanLogsForAttributes(t, logsConsumer, tc.service, tc.attrs)
+			default:
+				t.Fatalf("unknown data type %s", tc.dataType)
+			}
+		})
+	}
+}
+
 func TestE2E_ContainerIDAssociation(t *testing.T) {
 	testDir := filepath.Join("testdata", "e2e", "container_id_association_only")
 

processor/k8sattributesprocessor/factory.go

@@ -17,6 +17,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/k8sattributesprocessor/internal/kube"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/k8sattributesprocessor/internal/metadata"
 )
@@ -25,6 +26,7 @@ var (
 	kubeClientProvider   = kube.ClientProvider(nil)
 	consumerCapabilities = consumer.Capabilities{MutatesData: true}
 	defaultExcludes      = ExcludeConfig{Pods: []ExcludePodConfig{{Name: "jaeger-agent"}, {Name: "jaeger-collector"}}}
+	processors           = sharedcomponent.NewSharedComponents()
 )
 
 // NewFactory returns a new factory for the k8s processor.
@@ -57,7 +59,23 @@ func createTracesProcessor(
 	cfg component.Config,
 	next consumer.Traces,
 ) (processor.Traces, error) {
-	return createTracesProcessorWithOptions(ctx, params, cfg, next)
+	if !metadata.ProcessorK8sattributesShareProcessorBetweenPipelinesFeatureGate.IsEnabled() {
+		return createTracesProcessorWithOptions(ctx, params, cfg, next)
+	}
+	sc := processors.GetOrAdd(cfg, func() component.Component {
+		return createKubernetesProcessor(params, cfg)
+	})
+	kp := sc.Unwrap().(*kubernetesprocessor)
+
+	return processorhelper.NewTraces(
+		ctx,
+		params,
+		cfg,
+		next,
+		kp.processTraces,
+		processorhelper.WithCapabilities(consumerCapabilities),
+		processorhelper.WithStart(sc.Start),
+		processorhelper.WithShutdown(sc.Shutdown))
 }
 
 func createLogsProcessor(
@@ -66,7 +84,23 @@ func createLogsProcessor(
 	cfg component.Config,
 	nextLogsConsumer consumer.Logs,
 ) (processor.Logs, error) {
-	return createLogsProcessorWithOptions(ctx, params, cfg, nextLogsConsumer)
+	if !metadata.ProcessorK8sattributesShareProcessorBetweenPipelinesFeatureGate.IsEnabled() {
+		return createLogsProcessorWithOptions(ctx, params, cfg, nextLogsConsumer)
+	}
+	sc := processors.GetOrAdd(cfg, func() component.Component {
+		return createKubernetesProcessor(params, cfg)
+	})
+	kp := sc.Unwrap().(*kubernetesprocessor)
+
+	return processorhelper.NewLogs(
+		ctx,
+		params,
+		cfg,
+		nextLogsConsumer,
+		kp.processLogs,
+		processorhelper.WithCapabilities(consumerCapabilities),
+		processorhelper.WithStart(sc.Start),
+		processorhelper.WithShutdown(sc.Shutdown))
 }
 
 func createMetricsProcessor(
@@ -75,7 +109,23 @@ func createMetricsProcessor(
 	cfg component.Config,
 	nextMetricsConsumer consumer.Metrics,
 ) (processor.Metrics, error) {
-	return createMetricsProcessorWithOptions(ctx, params, cfg, nextMetricsConsumer)
+	if !metadata.ProcessorK8sattributesShareProcessorBetweenPipelinesFeatureGate.IsEnabled() {
+		return createMetricsProcessorWithOptions(ctx, params, cfg, nextMetricsConsumer)
+	}
+	sc := processors.GetOrAdd(cfg, func() component.Component {
+		return createKubernetesProcessor(params, cfg)
+	})
+	kp := sc.Unwrap().(*kubernetesprocessor)
+
+	return processorhelper.NewMetrics(
+		ctx,
+		params,
+		cfg,
+		nextMetricsConsumer,
+		kp.processMetrics,
+		processorhelper.WithCapabilities(consumerCapabilities),
+		processorhelper.WithStart(sc.Start),
+		processorhelper.WithShutdown(sc.Shutdown))
 }
 
 func createProfilesProcessor(
@@ -84,7 +134,24 @@ func createProfilesProcessor(
 	cfg component.Config,
 	nextProfilesConsumer xconsumer.Profiles,
 ) (xprocessor.Profiles, error) {
-	return createProfilesProcessorWithOptions(ctx, params, cfg, nextProfilesConsumer)
+	if !metadata.ProcessorK8sattributesShareProcessorBetweenPipelinesFeatureGate.IsEnabled() {
+		return createProfilesProcessorWithOptions(ctx, params, cfg, nextProfilesConsumer)
+	}
+	sc := processors.GetOrAdd(cfg, func() component.Component {
+		return createKubernetesProcessor(params, cfg)
+	})
+	kp := sc.Unwrap().(*kubernetesprocessor)
+
+	return xprocessorhelper.NewProfiles(
+		ctx,
+		params,
+		cfg,
+		nextProfilesConsumer,
+		kp.processProfiles,
+		xprocessorhelper.WithCapabilities(consumerCapabilities),
+		xprocessorhelper.WithStart(sc.Start),
+		xprocessorhelper.WithShutdown(sc.Shutdown),
+	)
 }
 
 func createTracesProcessorWithOptions(

processor/k8sattributesprocessor/go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.150.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.150.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig v0.150.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent v0.150.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/xk8stest v0.150.0
 	github.com/stretchr/testify v1.11.1
 	go.opentelemetry.io/collector/client v1.56.1-0.20260415114935-307e3abdbae9
@@ -170,3 +171,5 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest => ../../pkg/pdatatest
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden => ../../pkg/golden
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent => ../../internal/sharedcomponent

processor/k8sattributesprocessor/internal/metadata/generated_feature_gates.go

@@ -163,6 +163,13 @@ feature_gates:
       When enabled, semconv stable attributes are enabled.
     from_version: v0.145.0
     reference_url: https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/44589
+  - id: processor.k8sattributes.ShareProcessorBetweenPipelines
+    stage: alpha
+    description: >-
+      When enabled, processor instances with identical configuration are shared
+      across different signal type pipelines, reducing duplicate Kubernetes API watchers.
+    from_version: v0.150.0
+    reference_url: https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/2450
   - id: processor.k8sattributes.telemetry.disableOldFormatMetrics
     stage: alpha
     description: >-