Merge pull request #632 from klueska/single-informer-cd-ds-pods

Use single informer for all ComputeDomain daemonset pods

Author: jgehrcke

cmd/compute-domain-controller/daemonset.go

@@ -66,9 +66,9 @@ type DaemonSetManager struct {
 	informer      cache.SharedIndexInformer
 	mutationCache cache.MutationCache
 
+	daemonsetPodManager          *DaemonSetPodManager
 	resourceClaimTemplateManager *DaemonSetResourceClaimTemplateManager
 	cleanupManager               *CleanupManager[*appsv1.DaemonSet]
-	podManagers                  map[string]*DaemonSetPodManager
 }
 
 func NewDaemonSetManager(config *ManagerConfig, getComputeDomain GetComputeDomainFunc) *DaemonSetManager {
@@ -84,10 +84,10 @@ func NewDaemonSetManager(config *ManagerConfig, getComputeDomain GetComputeDomai
 	factory := informers.NewSharedInformerFactoryWithOptions(
 		config.clientsets.Core,
 		informerResyncPeriod,
+		informers.WithNamespace(config.driverNamespace),
 		informers.WithTweakListOptions(func(opts *metav1.ListOptions) {
 			opts.LabelSelector = metav1.FormatLabelSelector(labelSelector)
 		}),
-		informers.WithNamespace(config.driverNamespace),
 	)
 
 	informer := factory.Apps().V1().DaemonSets().Informer()
@@ -97,8 +97,8 @@ func NewDaemonSetManager(config *ManagerConfig, getComputeDomain GetComputeDomai
 		getComputeDomain: getComputeDomain,
 		factory:          factory,
 		informer:         informer,
-		podManagers:      make(map[string]*DaemonSetPodManager),
 	}
+	m.daemonsetPodManager = NewDaemonSetPodManager(config, getComputeDomain)
 	m.resourceClaimTemplateManager = NewDaemonSetResourceClaimTemplateManager(config, getComputeDomain)
 	m.cleanupManager = NewCleanupManager[*appsv1.DaemonSet](informer, getComputeDomain, m.cleanup)
 
@@ -151,6 +151,10 @@ func (m *DaemonSetManager) Start(ctx context.Context) (rerr error) {
 		return fmt.Errorf("informer cache sync for DaemonSet failed")
 	}
 
+	if err := m.daemonsetPodManager.Start(ctx); err != nil {
+		return fmt.Errorf("error starting ResourceClaimTemplate manager: %w", err)
+	}
+
 	if err := m.resourceClaimTemplateManager.Start(ctx); err != nil {
 		return fmt.Errorf("error starting ResourceClaimTemplate manager: %w", err)
 	}
@@ -163,8 +167,8 @@ func (m *DaemonSetManager) Start(ctx context.Context) (rerr error) {
 }
 
 func (m *DaemonSetManager) Stop() error {
-	if err := m.removeAllPodManagers(); err != nil {
-		return fmt.Errorf("error removing all Pod managers: %w", err)
+	if err := m.daemonsetPodManager.Stop(); err != nil {
+		return fmt.Errorf("error removing daemonset Pod manager: %w", err)
 	}
 	if err := m.resourceClaimTemplateManager.Stop(); err != nil {
 		return fmt.Errorf("error stopping ResourceClaimTemplate manager: %w", err)
@@ -266,16 +270,11 @@ func (m *DaemonSetManager) Delete(ctx context.Context, cdUID string) error {
 	}
 
 	d := ds[0]
-	key := d.Spec.Selector.MatchLabels[computeDomainLabelKey]
 
 	if err := m.resourceClaimTemplateManager.Delete(ctx, cdUID); err != nil {
 		return fmt.Errorf("error deleting ResourceClaimTemplate: %w", err)
 	}
 
-	if err := m.removePodManager(key); err != nil {
-		return fmt.Errorf("error removing Pod manager: %w", err)
-	}
-
 	if d.GetDeletionTimestamp() != nil {
 		return nil
 	}
@@ -374,10 +373,6 @@ func (m *DaemonSetManager) onAddOrUpdate(ctx context.Context, obj any) error {
 		return nil
 	}
 
-	if err := m.addPodManager(ctx, d.Spec.Selector, string(cd.UID)); err != nil {
-		return fmt.Errorf("error adding Pod manager '%s/%s': %w", d.Namespace, d.Name, err)
-	}
-
 	if int(d.Status.NumberReady) != cd.Spec.NumNodes {
 		return nil
 	}
@@ -400,57 +395,3 @@ func (m *DaemonSetManager) cleanup(ctx context.Context, cdUID string) error {
 	}
 	return nil
 }
-
-func (m *DaemonSetManager) addPodManager(ctx context.Context, labelSelector *metav1.LabelSelector, computeDomainUID string) error {
-	key := labelSelector.MatchLabels[computeDomainLabelKey]
-
-	if _, exists := m.podManagers[key]; exists {
-		return nil
-	}
-
-	podManager := NewDaemonSetPodManager(m.config, labelSelector, m.getComputeDomain, computeDomainUID)
-
-	if err := podManager.Start(ctx); err != nil {
-		return fmt.Errorf("error creating Pod manager: %w", err)
-	}
-
-	m.Lock()
-	m.podManagers[key] = podManager
-	m.Unlock()
-
-	return nil
-}
-
-func (m *DaemonSetManager) removePodManager(key string) error {
-	if _, exists := m.podManagers[key]; !exists {
-		return nil
-	}
-
-	m.Lock()
-	podManager := m.podManagers[key]
-	m.Unlock()
-
-	if err := podManager.Stop(); err != nil {
-		return fmt.Errorf("error stopping Pod manager: %w", err)
-	}
-
-	m.Lock()
-	delete(m.podManagers, key)
-	m.Unlock()
-
-	return nil
-}
-
-func (m *DaemonSetManager) removeAllPodManagers() error {
-	m.Lock()
-	for key, pm := range m.podManagers {
-		m.Unlock()
-		if err := pm.Stop(); err != nil {
-			return fmt.Errorf("error stopping Pod manager: %w", err)
-		}
-		m.Lock()
-		delete(m.podManagers, key)
-	}
-	m.Unlock()
-	return nil
-}

cmd/compute-domain-controller/daemonsetpods.go

@@ -41,12 +41,18 @@ type DaemonSetPodManager struct {
 	lister   corev1listers.PodLister
 
 	getComputeDomain GetComputeDomainFunc
-	computeDomainUID string
-
-	cleanupManager *CleanupManager[*corev1.Pod]
 }
 
-func NewDaemonSetPodManager(config *ManagerConfig, labelSelector *metav1.LabelSelector, getComputeDomain GetComputeDomainFunc, computeDomainUID string) *DaemonSetPodManager {
+func NewDaemonSetPodManager(config *ManagerConfig, getComputeDomain GetComputeDomainFunc) *DaemonSetPodManager {
+	labelSelector := &metav1.LabelSelector{
+		MatchExpressions: []metav1.LabelSelectorRequirement{
+			{
+				Key:      computeDomainLabelKey,
+				Operator: metav1.LabelSelectorOpExists,
+			},
+		},
+	}
+
 	factory := informers.NewSharedInformerFactoryWithOptions(
 		config.clientsets.Core,
 		informerResyncPeriod,
@@ -65,11 +71,8 @@ func NewDaemonSetPodManager(config *ManagerConfig, labelSelector *metav1.LabelSe
 		informer:         informer,
 		lister:           lister,
 		getComputeDomain: getComputeDomain,
-		computeDomainUID: computeDomainUID,
 	}
 
-	m.cleanupManager = NewCleanupManager[*corev1.Pod](informer, getComputeDomain, m.cleanup)
-
 	return m
 }
 
@@ -124,71 +127,24 @@ func (m *DaemonSetPodManager) Start(ctx context.Context) (rerr error) {
 		return fmt.Errorf("error syncing pod informer: %w", err)
 	}
 
-	if err := m.cleanupManager.Start(ctx); err != nil {
-		return fmt.Errorf("error starting cleanup manager: %w", err)
-	}
-
 	return nil
 }
 
 func (m *DaemonSetPodManager) Stop() error {
-	if err := m.cleanupManager.Stop(); err != nil {
-		return fmt.Errorf("error stopping cleanup manager: %w", err)
+	if m.cancelContext != nil {
+		m.cancelContext()
 	}
-	m.cancelContext()
 	m.waitGroup.Wait()
 	return nil
 }
 
-// cleanup removes nodes from the ComputeDomain status that no longer have backing pods.
-func (m *DaemonSetPodManager) cleanup(ctx context.Context, cdUID string) error {
-	cd, err := m.getComputeDomain(cdUID)
-	if err != nil {
-		return fmt.Errorf("error getting ComputeDomain: %w", err)
-	}
-	if cd == nil {
-		return nil
-	}
-
-	pods, err := m.lister.List(nil)
-	if err != nil {
-		return fmt.Errorf("error listing pods: %w", err)
-	}
-
-	newCD := cd.DeepCopy()
-
-	// Create a set of pod IPs for efficient lookup
-	podIPs := make(map[string]struct{})
-	for _, pod := range pods {
-		if pod.Status.PodIP != "" {
-			podIPs[pod.Status.PodIP] = struct{}{}
-		}
-	}
-
-	// Check if any nodes in the ComputeDomain status don't have backing pods
-	var updatedNodes []*nvapi.ComputeDomainNode
-	for _, node := range newCD.Status.Nodes {
-		if _, exists := podIPs[node.IPAddress]; exists {
-			updatedNodes = append(updatedNodes, node)
-		}
-	}
-
-	// Update the ComputeDomain status
-	if err := m.updateComputeDomainNodes(ctx, newCD, updatedNodes); err != nil {
-		return fmt.Errorf("error updating ComputeDomain status after removing stale nodes: %w", err)
-	}
-
-	klog.Infof("Successfully cleaned up stale nodes from ComputeDomain %s/%s", cd.Namespace, cd.Name)
-	return nil
-}
-
 func (m *DaemonSetPodManager) onPodDelete(ctx context.Context, obj any) error {
 	p, ok := obj.(*corev1.Pod)
 	if !ok {
 		return fmt.Errorf("failed to cast to Pod")
 	}
 
-	cd, err := m.getComputeDomain(m.computeDomainUID)
+	cd, err := m.getComputeDomain(p.Labels[computeDomainLabelKey])
 	if err != nil {
 		return fmt.Errorf("error getting ComputeDomain: %w", err)
 	}