[k8sattributesprocessor] Refactor informer initialization

processor/k8sattributesprocessor/client_test.go

@@ -9,10 +9,8 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/tools/cache"
 
-	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/k8sattributesprocessor/internal/kube"
 )
 
@@ -37,19 +35,43 @@ func selectors() (labels.Selector, fields.Selector) {
 }
 
 // newFakeClient instantiates a new FakeClient object and satisfies the ClientProvider type
-func newFakeClient(_ component.TelemetrySettings, _ k8sconfig.APIConfig, rules kube.ExtractionRules, filters kube.Filters, associations []kube.Association, _ kube.Excludes, _ kube.APIClientsetProvider, _ kube.InformerProvider, _ kube.InformerProviderNamespace, _ kube.InformerProviderReplicaSet, _ bool, _ time.Duration) (kube.Client, error) {
-	cs := fake.NewSimpleClientset()
-
+func newFakeClient(
+	_ component.TelemetrySettings,
+	rules kube.ExtractionRules,
+	filters kube.Filters,
+	associations []kube.Association,
+	_ kube.Excludes,
+	_ *kube.InformerProviders,
+	_ bool,
+	_ time.Duration,
+) (kube.Client, error) {
 	ls, fs := selectors()
+	closeCh := make(chan struct{})
+	informer, err := kube.NewFakeInformer("", ls, fs, nil, closeCh)
+	if err != nil {
+		return nil, err
+	}
+	nsInformer, err := kube.NewFakeInformer("", ls, fs, nil, closeCh)
+	if err != nil {
+		return nil, err
+	}
+	nodeInformer, err := kube.NewFakeInformer("", ls, fs, nil, closeCh)
+	if err != nil {
+		return nil, err
+	}
+	rsInformer, err := kube.NewFakeInformer("", ls, fs, nil, closeCh)
+	if err != nil {
+		return nil, err
+	}
 	return &fakeClient{
 		Pods:               map[kube.PodIdentifier]*kube.Pod{},
 		Rules:              rules,
 		Filters:            filters,
 		Associations:       associations,
-		Informer:           kube.NewFakeInformer(cs, "", ls, fs),
-		NamespaceInformer:  kube.NewFakeInformer(cs, "", ls, fs),
-		NodeInformer:       kube.NewFakeInformer(cs, "", ls, fs),
-		ReplicaSetInformer: kube.NewFakeInformer(cs, "", ls, fs),
+		Informer:           informer,
+		NamespaceInformer:  nsInformer,
+		NodeInformer:       nodeInformer,
+		ReplicaSetInformer: rsInformer,
 		StopCh:             make(chan struct{}),
 	}, nil
 }

processor/k8sattributesprocessor/factory.go

@@ -16,12 +16,10 @@ import (
 	"go.opentelemetry.io/collector/processor/xprocessor"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/k8sattributesprocessor/internal/kube"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/k8sattributesprocessor/internal/metadata"
 )
 
 var (
-	kubeClientProvider   = kube.ClientProvider(nil)
 	consumerCapabilities = consumer.Capabilities{MutatesData: true}
 	defaultExcludes      = ExcludeConfig{Pods: []ExcludePodConfig{{Name: "jaeger-agent"}, {Name: "jaeger-collector"}}}
 )

processor/k8sattributesprocessor/factory_test.go

@@ -24,9 +24,6 @@ func TestCreateDefaultConfig(t *testing.T) {
 func TestCreateProcessor(t *testing.T) {
 	factory := NewFactory()
 
-	realClient := kubeClientProvider
-	kubeClientProvider = newFakeClient
-
 	cfg := factory.CreateDefaultConfig()
 	params := processortest.NewNopSettings()
 
@@ -64,7 +61,4 @@ func TestCreateProcessor(t *testing.T) {
 	pp, err = factory.(xprocessor.Factory).CreateProfiles(context.Background(), params, cfg, consumertest.NewNop())
 	assert.NotNil(t, pp)
 	assert.NoError(t, err)
-
-	// Switch it back so other tests run afterwards will not fail on unexpected state
-	kubeClientProvider = realClient
 }