[poc][processor/k8sattributeprocessor]: modify test to use different addresses

Signed-off-by: RealAnna <[email protected]>

Author: RealAnna

internal/k8sconfig/config.go

@@ -21,6 +21,7 @@ import (
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/dynamic"
 	k8s "k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/metadata"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/clientcmd"
@@ -168,6 +169,25 @@ func MakeDynamicClient(apiConf APIConfig) (dynamic.Interface, error) {
 	return client, nil
 }
 
+// MakeMetadataClient constructs a client-go metadata.Interface using the same APIConfig flow.
+// It mirrors MakeDynamicClient, but uses metadata.NewForConfig.
+func MakeMetadataClient(apiConf APIConfig) (metadata.Interface, error) {
+	if err := apiConf.Validate(); err != nil {
+		return nil, err
+	}
+
+	authConf, err := CreateRestConfig(apiConf)
+	if err != nil {
+		return nil, err
+	}
+
+	mc, err := metadata.NewForConfig(authConf)
+	if err != nil {
+		return nil, err
+	}
+	return mc, nil
+}
+
 // MakeOpenShiftQuotaClient can take configuration if needed for other types of auth
 // and return an OpenShift quota API client
 func MakeOpenShiftQuotaClient(apiConf APIConfig) (quotaclientset.Interface, error) {

processor/k8sattributesprocessor/internal/kube/client.go

@@ -249,22 +249,9 @@ func New(
 
 	if rules.DeploymentName || rules.DeploymentUID {
 		if informersFactory.newReplicaSetInformer == nil {
-			informersFactory.newReplicaSetInformer = newReplicaSetSharedInformer
+			informersFactory.newReplicaSetInformer = NewReplicaSetMetaInformerProvider(apiCfg)
 		}
 		c.replicasetInformer = informersFactory.newReplicaSetInformer(c.kc, c.Filters.Namespace)
-		err = c.replicasetInformer.SetTransform(
-			func(object any) (any, error) {
-				originalReplicaset, success := object.(*apps_v1.ReplicaSet)
-				if !success { // means this is a cache.DeletedFinalStateUnknown, in which case we do nothing
-					return object, nil
-				}
-
-				return removeUnnecessaryReplicaSetData(originalReplicaset), nil
-			},
-		)
-		if err != nil {
-			return nil, err
-		}
 	}
 
 	if c.extractNodeLabelsAnnotations() || c.extractNodeUID() {
@@ -1813,19 +1800,6 @@ func (c *WatchClient) addOrUpdateReplicaSet(replicaset *apps_v1.ReplicaSet) {
 	c.m.Unlock()
 }
 
-// This function removes all data from the ReplicaSet except what is required by extraction rules
-func removeUnnecessaryReplicaSetData(replicaset *apps_v1.ReplicaSet) *apps_v1.ReplicaSet {
-	transformedReplicaset := apps_v1.ReplicaSet{
-		ObjectMeta: meta_v1.ObjectMeta{
-			Name:      replicaset.GetName(),
-			Namespace: replicaset.GetNamespace(),
-			UID:       replicaset.GetUID(),
-		},
-	}
-	transformedReplicaset.SetOwnerReferences(replicaset.GetOwnerReferences())
-	return &transformedReplicaset
-}
-
 // runInformerWithDependencies starts the given informer. The second argument is a list of other informers that should complete
 // before the informer is started. This is necessary e.g. for the pod informer which requires the replica set informer
 // to be finished to correctly establish the connection to the replicaset/deployment it belongs to.

processor/k8sattributesprocessor/internal/kube/client_test.go

@@ -1081,8 +1081,7 @@ func TestExtractionRules(t *testing.T) {
 			maps.Copy(podCopy.Annotations, tc.additionalAnnotations)
 			maps.Copy(podCopy.Labels, tc.additionalLabels)
 			transformedPod := removeUnnecessaryPodData(podCopy, c.Rules)
-			transformedReplicaset := removeUnnecessaryReplicaSetData(replicaset)
-			c.handleReplicaSetAdd(transformedReplicaset)
+			c.handleReplicaSetAdd(replicaset)
 			c.handlePodAdd(transformedPod)
 			p, ok := c.GetPod(newPodIdentifier("connection", "", podCopy.Status.PodIP))
 			require.True(t, ok)
@@ -1237,8 +1236,7 @@ func TestReplicaSetExtractionRules(t *testing.T) {
 			// manually call the data removal functions here
 			// normally the informer does this, but fully emulating the informer in this test is annoying
 			transformedPod := removeUnnecessaryPodData(pod, c.Rules)
-			transformedReplicaset := removeUnnecessaryReplicaSetData(replicaset)
-			c.handleReplicaSetAdd(transformedReplicaset)
+			c.handleReplicaSetAdd(replicaset)
 			c.handlePodAdd(transformedPod)
 			p, ok := c.GetPod(newPodIdentifier("connection", "", pod.Status.PodIP))
 			require.True(t, ok)

processor/k8sattributesprocessor/internal/kube/informer.go

@@ -6,15 +6,18 @@ package kube // import "github.com/open-telemetry/opentelemetry-collector-contri
 import (
 	"context"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig"
 	apps_v1 "k8s.io/api/apps/v1"
 	batch_v1 "k8s.io/api/batch/v1"
 	api_v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/metadata"
 	"k8s.io/client-go/tools/cache"
 )
 
@@ -88,11 +91,11 @@ func newKubeSystemSharedInformer(
 ) cache.SharedInformer {
 	informer := cache.NewSharedInformer(
 		&cache.ListWatch{
-			ListFunc: func(opts metav1.ListOptions) (runtime.Object, error) {
+			ListWithContextFunc: func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
 				opts.FieldSelector = fields.OneTermEqualSelector("metadata.name", kubeSystemNamespace).String()
 				return client.CoreV1().Namespaces().List(context.Background(), opts)
 			},
-			WatchFunc: func(opts metav1.ListOptions) (watch.Interface, error) {
+			WatchFuncWithContext: func(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error) {
 				opts.FieldSelector = fields.OneTermEqualSelector("metadata.name", kubeSystemNamespace).String()
 				return client.CoreV1().Namespaces().Watch(context.Background(), opts)
 			},
@@ -129,21 +132,6 @@ func namespaceInformerWatchFunc(client kubernetes.Interface) cache.WatchFuncWith
 	}
 }
 
-func newReplicaSetSharedInformer(
-	client kubernetes.Interface,
-	namespace string,
-) cache.SharedInformer {
-	informer := cache.NewSharedInformer(
-		&cache.ListWatch{
-			ListWithContextFunc:  replicasetListFuncWithSelectors(client, namespace),
-			WatchFuncWithContext: replicasetWatchFuncWithSelectors(client, namespace),
-		},
-		&apps_v1.ReplicaSet{},
-		watchSyncPeriod,
-	)
-	return informer
-}
-
 func replicasetListFuncWithSelectors(client kubernetes.Interface, namespace string) cache.ListWithContextFunc {
 	return func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
 		return client.AppsV1().ReplicaSets(namespace).List(ctx, opts)
@@ -210,6 +198,77 @@ func statefulsetWatchFuncWithSelectors(client kubernetes.Interface, namespace st
 	}
 }
 
+// NewReplicaSetMetaInformerProvider returns a provider with the SAME signature
+// as your existing helpers, but emits metav1.PartialObjectMetadata.
+// It uses ListWithContextFunc and WatchFuncWithContext.
+func NewReplicaSetMetaInformerProvider(apiCfg k8sconfig.APIConfig) func(client kubernetes.Interface, namespace string) cache.SharedInformer {
+	return func(client kubernetes.Interface, namespace string) cache.SharedInformer {
+		// Build metadata client from APIConfig (mirrors your MakeDynamicClient flow)
+		if err := apiCfg.Validate(); err != nil {
+			return cache.NewSharedIndexInformer(
+				&cache.ListWatch{
+					ListWithContextFunc:  func(ctx context.Context, _ metav1.ListOptions) (runtime.Object, error) { return nil, err },
+					WatchFuncWithContext: func(ctx context.Context, _ metav1.ListOptions) (watch.Interface, error) { return nil, err },
+				},
+				&metav1.PartialObjectMetadata{},
+				watchSyncPeriod,
+				cache.Indexers{},
+			)
+		}
+
+		restCfg, err := k8sconfig.CreateRestConfig(apiCfg)
+		if err != nil {
+			return cache.NewSharedIndexInformer(
+				&cache.ListWatch{
+					ListWithContextFunc:  func(ctx context.Context, _ metav1.ListOptions) (runtime.Object, error) { return nil, err },
+					WatchFuncWithContext: func(ctx context.Context, _ metav1.ListOptions) (watch.Interface, error) { return nil, err },
+				},
+				&metav1.PartialObjectMetadata{},
+				watchSyncPeriod,
+				cache.Indexers{},
+			)
+		}
+
+		mc, err := metadata.NewForConfig(restCfg)
+		if err != nil {
+			return cache.NewSharedIndexInformer(
+				&cache.ListWatch{
+					ListWithContextFunc:  func(ctx context.Context, _ metav1.ListOptions) (runtime.Object, error) { return nil, err },
+					WatchFuncWithContext: func(ctx context.Context, _ metav1.ListOptions) (watch.Interface, error) { return nil, err },
+				},
+				&metav1.PartialObjectMetadata{},
+				watchSyncPeriod,
+				cache.Indexers{},
+			)
+		}
+
+		gvr := schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "replicasets"}
+
+		lw := &cache.ListWatch{
+			ListWithContextFunc: func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
+				// Populate selectors here if you mirror your existing helpers
+				if namespace == "" {
+					return mc.Resource(gvr).List(ctx, opts)
+				}
+				return mc.Resource(gvr).Namespace(namespace).List(ctx, opts)
+			},
+			WatchFuncWithContext: func(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error) {
+				if namespace == "" {
+					return mc.Resource(gvr).Watch(ctx, opts)
+				}
+				return mc.Resource(gvr).Namespace(namespace).Watch(ctx, opts)
+			},
+		}
+
+		return cache.NewSharedIndexInformer(
+			lw,
+			&metav1.PartialObjectMetadata{},
+			watchSyncPeriod,
+			cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
+		)
+	}
+}
+
 func newDaemonSetSharedInformer(
 	client kubernetes.Interface,
 	namespace string,