Merge pull request kubernetes#126723 from pwschuurman/revert-125372-pvcScalabilityFix

Revert "Improve PVC protection controller's scalability by batch-processing PVCs by namespace & caching live pod list results"

Author: k8s-ci-robot

pkg/controller/volume/pvcprotection/pvc_protection_controller.go

@@ -19,7 +19,6 @@ package pvcprotection
 import (
 	"context"
 	"fmt"
-	"sync"
 	"time"
 
 	v1 "k8s.io/api/core/v1"
@@ -42,65 +41,6 @@ import (
 
 // Controller is controller that removes PVCProtectionFinalizer
 // from PVCs that are used by no pods.
-
-type LazyLivePodList struct {
-	cache      []v1.Pod
-	controller *Controller
-}
-
-func (ll *LazyLivePodList) getCache() []v1.Pod {
-	return ll.cache
-}
-
-func (ll *LazyLivePodList) setCache(pods []v1.Pod) {
-	ll.cache = pods
-}
-
-type pvcData struct {
-	pvcKey  string
-	pvcName string
-}
-
-type pvcProcessingStore struct {
-	namespaceToPVCsMap map[string][]pvcData
-	namespaceQueue     workqueue.TypedInterface[string]
-	mu                 sync.Mutex
-}
-
-func NewPVCProcessingStore() *pvcProcessingStore {
-	return &pvcProcessingStore{
-		namespaceToPVCsMap: make(map[string][]pvcData),
-		namespaceQueue:     workqueue.NewTyped[string](),
-	}
-}
-
-func (m *pvcProcessingStore) addOrUpdate(namespace string, pvcKey, pvcName string) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if _, exists := m.namespaceToPVCsMap[namespace]; !exists {
-		m.namespaceToPVCsMap[namespace] = make([]pvcData, 0)
-		m.namespaceQueue.Add(namespace)
-	}
-	m.namespaceToPVCsMap[namespace] = append(m.namespaceToPVCsMap[namespace], pvcData{pvcKey: pvcKey, pvcName: pvcName})
-}
-
-// Returns a list of pvcs and the associated namespace to be processed downstream
-func (m *pvcProcessingStore) flushNextPVCsByNamespace() ([]pvcData, string) {
-
-	nextNamespace, quit := m.namespaceQueue.Get()
-	if quit {
-		return nil, nextNamespace
-	}
-
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	pvcs := m.namespaceToPVCsMap[nextNamespace]
-
-	delete(m.namespaceToPVCsMap, nextNamespace)
-	m.namespaceQueue.Done(nextNamespace)
-	return pvcs, nextNamespace
-}
-
 type Controller struct {
 	client clientset.Interface
 
@@ -111,8 +51,7 @@ type Controller struct {
 	podListerSynced cache.InformerSynced
 	podIndexer      cache.Indexer
 
-	queue              workqueue.TypedRateLimitingInterface[string]
-	pvcProcessingStore *pvcProcessingStore
+	queue workqueue.TypedRateLimitingInterface[string]
 }
 
 // NewPVCProtectionController returns a new instance of PVCProtectionController.
@@ -123,7 +62,6 @@ func NewPVCProtectionController(logger klog.Logger, pvcInformer coreinformers.Pe
 			workqueue.DefaultTypedControllerRateLimiter[string](),
 			workqueue.TypedRateLimitingQueueConfig[string]{Name: "pvcprotection"},
 		),
-		pvcProcessingStore: NewPVCProcessingStore(),
 	}
 
 	e.pvcLister = pvcInformer.Lister()
@@ -162,7 +100,6 @@ func NewPVCProtectionController(logger klog.Logger, pvcInformer coreinformers.Pe
 func (c *Controller) Run(ctx context.Context, workers int) {
 	defer utilruntime.HandleCrash()
 	defer c.queue.ShutDown()
-	defer c.pvcProcessingStore.namespaceQueue.ShutDown()
 
 	logger := klog.FromContext(ctx)
 	logger.Info("Starting PVC protection controller")
@@ -172,64 +109,45 @@ func (c *Controller) Run(ctx context.Context, workers int) {
 		return
 	}
 
-	go wait.UntilWithContext(ctx, c.runMainWorker, time.Second)
 	for i := 0; i < workers; i++ {
-		go wait.UntilWithContext(ctx, c.runProcessNamespaceWorker, time.Second)
+		go wait.UntilWithContext(ctx, c.runWorker, time.Second)
 	}
 
 	<-ctx.Done()
 }
 
-// Main worker batch-pulls PVC items off informer's work queue and populates namespace queue and namespace-PVCs map
-func (c *Controller) runMainWorker(ctx context.Context) {
-	for c.processNextWorkItem() {
+func (c *Controller) runWorker(ctx context.Context) {
+	for c.processNextWorkItem(ctx) {
 	}
 }
 
-// Consumer worker pulls items off namespace queue and processes associated PVCs
-func (c *Controller) runProcessNamespaceWorker(ctx context.Context) {
-	for c.processPVCsByNamespace(ctx) {
+// processNextWorkItem deals with one pvcKey off the queue.  It returns false when it's time to quit.
+func (c *Controller) processNextWorkItem(ctx context.Context) bool {
+	pvcKey, quit := c.queue.Get()
+	if quit {
+		return false
 	}
-}
+	defer c.queue.Done(pvcKey)
 
-func (c *Controller) processNextWorkItem() bool {
-	queueLength := c.queue.Len()
-	for i := 0; i < queueLength; i++ {
-		pvcKey, quit := c.queue.Get()
-		if quit {
-			return false
-		}
-		pvcNamespace, pvcName, err := cache.SplitMetaNamespaceKey(pvcKey)
-		if err != nil {
-			utilruntime.HandleError(fmt.Errorf("error parsing PVC key %q: %w", pvcKey, err))
-		}
-		c.pvcProcessingStore.addOrUpdate(pvcNamespace, pvcKey, pvcName)
+	pvcNamespace, pvcName, err := cache.SplitMetaNamespaceKey(pvcKey)
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("error parsing PVC key %q: %w", pvcKey, err))
+		return true
 	}
-	return !c.queue.ShuttingDown()
-}
 
-func (c *Controller) processPVCsByNamespace(ctx context.Context) bool {
-	pvcList, namespace := c.pvcProcessingStore.flushNextPVCsByNamespace()
-	if pvcList == nil {
-		return false
+	err = c.processPVC(ctx, pvcNamespace, pvcName)
+	if err == nil {
+		c.queue.Forget(pvcKey)
+		return true
 	}
 
-	lazyLivePodList := &LazyLivePodList{controller: c}
-	for _, item := range pvcList {
-		pvcKey, pvcName := item.pvcKey, item.pvcName
-		err := c.processPVC(ctx, namespace, pvcName, lazyLivePodList)
-		if err == nil {
-			c.queue.Forget(pvcKey)
-		} else {
-			c.queue.AddRateLimited(pvcKey)
-			utilruntime.HandleError(fmt.Errorf("PVC %v in namespace %v failed with: %w", pvcName, namespace, err))
-		}
-		c.queue.Done(pvcKey)
-	}
+	utilruntime.HandleError(fmt.Errorf("PVC %v failed with : %w", pvcKey, err))
+	c.queue.AddRateLimited(pvcKey)
+
 	return true
 }
 
-func (c *Controller) processPVC(ctx context.Context, pvcNamespace, pvcName string, lazyLivePodList *LazyLivePodList) error {
+func (c *Controller) processPVC(ctx context.Context, pvcNamespace, pvcName string) error {
 	logger := klog.FromContext(ctx)
 	logger.V(4).Info("Processing PVC", "PVC", klog.KRef(pvcNamespace, pvcName))
 	startTime := time.Now()
@@ -249,7 +167,7 @@ func (c *Controller) processPVC(ctx context.Context, pvcNamespace, pvcName strin
 	if protectionutil.IsDeletionCandidate(pvc, volumeutil.PVCProtectionFinalizer) {
 		// PVC should be deleted. Check if it's used and remove finalizer if
 		// it's not.
-		isUsed, err := c.isBeingUsed(ctx, pvc, lazyLivePodList)
+		isUsed, err := c.isBeingUsed(ctx, pvc)
 		if err != nil {
 			return err
 		}
@@ -291,11 +209,11 @@ func (c *Controller) removeFinalizer(ctx context.Context, pvc *v1.PersistentVolu
 		logger.Error(err, "Error removing protection finalizer from PVC", "PVC", klog.KObj(pvc))
 		return err
 	}
-	logger.Info("Removed protection finalizer from PVC", "PVC", klog.KObj(pvc))
+	logger.V(3).Info("Removed protection finalizer from PVC", "PVC", klog.KObj(pvc))
 	return nil
 }
 
-func (c *Controller) isBeingUsed(ctx context.Context, pvc *v1.PersistentVolumeClaim, lazyLivePodList *LazyLivePodList) (bool, error) {
+func (c *Controller) isBeingUsed(ctx context.Context, pvc *v1.PersistentVolumeClaim) (bool, error) {
 	// Look for a Pod using pvc in the Informer's cache. If one is found the
 	// correct decision to keep pvc is taken without doing an expensive live
 	// list.
@@ -311,9 +229,7 @@ func (c *Controller) isBeingUsed(ctx context.Context, pvc *v1.PersistentVolumeCl
 	// mean such a Pod doesn't exist: it might just not be in the cache yet. To
 	// be 100% confident that it is safe to delete pvc make sure no Pod is using
 	// it among those returned by a live list.
-
-	// Use lazy live pod list instead of directly calling API server
-	return c.askAPIServer(ctx, pvc, lazyLivePodList)
+	return c.askAPIServer(ctx, pvc)
 }
 
 func (c *Controller) askInformer(logger klog.Logger, pvc *v1.PersistentVolumeClaim) (bool, error) {
@@ -342,24 +258,16 @@ func (c *Controller) askInformer(logger klog.Logger, pvc *v1.PersistentVolumeCla
 	return false, nil
 }
 
-func (c *Controller) askAPIServer(ctx context.Context, pvc *v1.PersistentVolumeClaim, lazyLivePodList *LazyLivePodList) (bool, error) {
+func (c *Controller) askAPIServer(ctx context.Context, pvc *v1.PersistentVolumeClaim) (bool, error) {
 	logger := klog.FromContext(ctx)
 	logger.V(4).Info("Looking for Pods using PVC with a live list", "PVC", klog.KObj(pvc))
-	if lazyLivePodList.getCache() == nil {
-		podsList, err := c.client.CoreV1().Pods(pvc.Namespace).List(ctx, metav1.ListOptions{})
 
-		if err != nil {
-			return false, fmt.Errorf("live list of pods failed: %s", err.Error())
-		}
-
-		if podsList.Items == nil {
-			lazyLivePodList.setCache(make([]v1.Pod, 0))
-		} else {
-			lazyLivePodList.setCache(podsList.Items)
-		}
+	podsList, err := c.client.CoreV1().Pods(pvc.Namespace).List(ctx, metav1.ListOptions{})
+	if err != nil {
+		return false, fmt.Errorf("live list of pods failed: %s", err.Error())
 	}
 
-	for _, pod := range lazyLivePodList.getCache() {
+	for _, pod := range podsList.Items {
 		if c.podUsesPVC(logger, &pod, pvc) {
 			return true, nil
 		}