[processor/k8sattributes] Wait for ReplicaSet informer before starting pod informer

processor/k8sattributesprocessor/e2e_test.go

@@ -1099,8 +1099,6 @@ func TestE2E_NamespacedRBACNoPodIP(t *testing.T) {
 //	make docker-otelcontribcol
 //	KUBECONFIG=/tmp/kube-config-otelcol-e2e-testing kind load docker-image otelcontribcol:latest
 func TestE2E_ClusterRBACCollectorStartAfterTelemetryGen(t *testing.T) {
-	// TODO: Re-enable this test when the issue being tested here is fully solved: https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/37056
-	t.Skip("Skipping test as https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/37056 is not fully solved yet")
 	testDir := filepath.Join("testdata", "e2e", "clusterrbac")
 
 	k8sClient, err := k8stest.NewK8sClient(testKubeConfig)

processor/k8sattributesprocessor/internal/kube/client.go

@@ -210,9 +210,11 @@ func New(
 func (c *WatchClient) Start() error {
 	synced := make([]cache.InformerSynced, 0)
 
+	waitForReplicaSets := false
 	// start the replicaSet informer first, as the replica sets need to be
 	// present at the time the pods are handled, to correctly establish the connection between pods and deployments
 	if c.Rules.DeploymentName || c.Rules.DeploymentUID {
+		waitForReplicaSets = true
 		reg, err := c.replicasetInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 			AddFunc:    c.handleReplicaSetAdd,
 			UpdateFunc: c.handleReplicaSetUpdate,
@@ -222,7 +224,7 @@ func (c *WatchClient) Start() error {
 			return err
 		}
 		synced = append(synced, reg.HasSynced)
-		go c.replicasetInformer.Run(c.stopCh)
+		go c.runInformer(c.replicasetInformer)
 	}
 
 	reg, err := c.informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
@@ -234,7 +236,12 @@ func (c *WatchClient) Start() error {
 		return err
 	}
 	synced = append(synced, reg.HasSynced)
-	go c.informer.Run(c.stopCh)
+
+	if waitForReplicaSets {
+		go c.runInformer(c.informer, c.replicasetInformer)
+	} else {
+		go c.runInformer(c.informer)
+	}
 
 	reg, err = c.namespaceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc:    c.handleNamespaceAdd,
@@ -245,7 +252,7 @@ func (c *WatchClient) Start() error {
 		return err
 	}
 	synced = append(synced, reg.HasSynced)
-	go c.namespaceInformer.Run(c.stopCh)
+	go c.runInformer(c.namespaceInformer)
 
 	if c.nodeInformer != nil {
 		reg, err = c.nodeInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
@@ -257,7 +264,7 @@ func (c *WatchClient) Start() error {
 			return err
 		}
 		synced = append(synced, reg.HasSynced)
-		go c.nodeInformer.Run(c.stopCh)
+		go c.runInformer(c.nodeInformer)
 	}
 
 	if c.waitForMetadata {
@@ -1123,6 +1130,24 @@ func (c *WatchClient) getReplicaSet(uid string) (*ReplicaSet, bool) {
 	return nil, false
 }
 
+// runInformer starts the given informer. This method optionally takes 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.
+func (c *WatchClient) runInformer(informer cache.SharedInformer, dependencies ...cache.SharedInformer) {
+	if len(dependencies) > 0 {
+		timeoutCh := make(chan struct{})
+		// TODO hard coding the timeout for now, check if we should make this configurable
+		t := time.AfterFunc(5*time.Second, func() {
+			close(timeoutCh)
+		})
+		defer t.Stop()
+		for _, dependency := range dependencies {
+			cache.WaitForCacheSync(timeoutCh, dependency.HasSynced)
+		}
+	}
+	informer.Run(c.stopCh)
+}
+
 // ignoreDeletedFinalStateUnknown returns the object wrapped in
 // DeletedFinalStateUnknown. Useful in OnDelete resource event handlers that do
 // not need the additional context.