Repository: kubernetes/kubernetes (language: go)

Issue:
PVCProtectionController can be slow for larger clusters
## References
* https://github.com/kubernetes/kubernetes/issues/76703
* https://github.com/kubernetes/kubernetes/issues/75980
* https://github.com/kubernetes/kubernetes/pull/80492
* https://github.com/kubernetes/design-proposals-archive/blob/main/storage/postpone-pvc-deletion-if-used-in-a-pod.md

## Overview

As described [here](https://github.com/kubernetes/design-proposals-archive/blob/main/storage/postpone-pvc-deletion-if-used-in-a-pod.md) PVC protection controller uses logic that postpones deletion of PVC in case it is used by Pod. When the Pod is deleted and PVC is marked for deletion as well (by setting `deletionTimestamp`) the PVC protection controllers ensures that all references to the PVC has been deleted and no Pod is using it.

PVC protection controller processes PVC one by one checking if there is still a Pod that has any reference to it. If not, the finalizer is removed and PVC is deleted from the cluster.

## The problem

Some time ago the [issue](https://github.com/kubernetes/kubernetes/issues/76703) was filled that described race condition in PVC protection controller. Basically the logic used and relied only on Pod's informer cache which could be potentially outdated and PVC protection controller could deleted PVC that it's still being used (has been recently attached). There was also [MR](https://github.com/kubernetes/kubernetes/pull/80492) that fixed it.

The problem with the new logic is that it has really slow/inefficient performance which is in order of `O(m * n)` (where `m` is the number of PVCs to be processed and `n` is the number of Pods in the cluster). [Someone even pointed it out](https://github.com/kubernetes/kubernetes/pull/80492#discussion_r306743513) in the MR but it wasn't sufficiently measured/analyzed. For active larger cluster 1k+ where there is a lot of Pods/PVCs being created and deleted the PVC protection controller is too slow to process everything. This causes large number of PVCs still being in the cluster what can slow down listing and scheduling but also puts unnecessary pressure on the API server and etcd.

What happens under the hood is that when the Pod is quickly removed from the cluster and PVC is marked for deletion we always will go past the informer check and make another API LIST call: https://github.com/kubernetes/kubernetes/blob/808c8f42d55c520e45eab93200768607cefc1451/pkg/controller/volume/pvcprotection/pvc_protection_controller.go#L212-L226

This means that for each PVC we will do full-blown LIST call that returns all Pods in given namespace. It's easy to foresee that this logic will not be fast enough to handle active namespaces with 1k+ Pods/PVCs.

Here is the PVC protection controller's work queue and the PVC count over 5 hours of active creation/deletion that tests this hypothesis. What happens here is that PVC protection controller is slow to delete the finalizer so the work queue and number of PVCs in the cluster only increases.
![slow](https://user-images.githubusercontent.com/3855799/161510944-cf39c3ef-3b97-4b90-94db-765c725e8f54.jpg)

This can easily blow up the cluster if no solution is implemented as a remedy.

## The (temporary) solutions

There are couple of ways to go around this problem:
1. Reduce the size of the namespace. When we list and process Pods/PVCs we operate on single namespace. If we reduce the size of the namespace the PVC protection controller will have an easier work.
2. Implement component that will automatically delete protection finalizer for PVCs which for sure will not be used. Basically if you know that the PVCs you are creating will not be reused you can simply automatically delete the finalizer without relying on the PVC protection controller.
3. Turn off protection finalizer (see https://kubernetes.io/docs/reference/access-authn-authz/admission-controllers/#storageobjectinuseprotection).

Probably there is more (I'm happy to hear if you have suggestions).

But all of proposed solution don't address the original problem and only make it less probable.

## The (considered) solutions 

I've been thinking on addressing the logic problem but I couldn't come up with any good idea:
* I initially considered making PVC protection controller multi-threaded. But this doesn't reduce the overall performance issues and in some cases it could even make things worse. It would probably help in most of the cases but the underlying problem would still remain.
* Then I considered batching. Currently we process one PVC at a time. We could lower number of Pod LIST calls by processing N number of PVCs and executing single Pods LIST call. So instead of having "1 PVC -> 1 Pod LIST call" we could have eg. "20 PVC -> 1 Pod LIST call". This seems like a interesting approach but it would require changing the whole logic of the controller.


Discussion / hints:
/sig storage
I wonder if we can just remove the logic of `AskApiserver` ？
> I wonder if we can just remove the logic of AskApiserver ？

+1, list all Pods in a namespace from apiserver is a bad idea, it will be terrible when there are a large number of Pods and PVCs.

If we are concerning that some Pods using this PVC may have not synced into cache yet, how about multiple check from informer with a little backoff:

1. For a terminating PVC that find not in-use from `askInformer`, record into a memory cache, and then `queue.AddAfter` with a little duration (could be 100ms~1s)
2. If the PVC is still not in-use from `askInformer` in the next reconcile, then remove its finalizer. Also we can do this more times if ppl think it is not enough to check twice from informer.

WDYT @msau42 @jsafrane @matte21 

If this is an acceptable change, I would be happy to raise a Draft PR for it.
The Kubernetes project currently lacks enough contributors to adequately respond to all issues and PRs.

This bot triages issues and PRs according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue or PR as fresh with `/remove-lifecycle stale`
- Mark this issue or PR as rotten with `/lifecycle rotten`
- Close this issue or PR with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle stale

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/
/remove-lifecycle stale
Naive solution for this problem would be to change isBeingUsed to:
* use live apiserver to get the up to date resourceVersion for pods resource (e.g. by doing a LIST call with limit=1 that will return both random element (to be ignored) and up to date resource version for pods resource)
* waiting until the pod informer reaches a given resourceVersion (probably just comparing pvcInformer.Informer().LastSyncResourceVersion() with the rv received above should be enough).
* using informer without live apiserver (we have a guarantee that the informer is up to date).

This way we can remove risk that we will make operation based on the stale informer state and be able to perform check quickly as we can avoid expensive list, use podIndexer's index instead of O(n) lookup we do in askAPIServer.

@jsafrane @VirrageS, WDYT?
I might be missing something about this proposal. But assuming we do "apiserver [call] to get the up to date resourceVersion for pods resource" and "waiting until the pod informer reaches a given resourceVersion" during this waiting the resourceVersion could be already updated and we will be waiting for something that is already old and outdated.
Probably I don't understand the concern. I was thinking about waiting for **at least** given resource version (probably I should be more clear about that). If after listing we will find that the informer is already in greater resource version that we requested that's good -- we can stop waiting immediately. If we never see specific resourceVersion, but we see larger resourceVersions, we are good as well, as the resource version for specific resource type only grows.
Hm, let me understand the proposal. So whenever we are about to delete the PVC we do:
1. "a LIST call with limit=1 that will return both random element (to be ignored) and up to date resource version for pods resource)",
2. check `resourceVersion` from previous step,
3. we wait for the informer to be up to date (it must reach at least `resourceVersion` from previous step),
4. once the informer reaches the expected state we use it to list and check all the pods.

Is that the idea?

For me it looks really reasonable. I don't see any flaws with that.
Yes, this is the idea.

Potentially in "4. once the informer reaches the expected state we use it to list and check all the pods." we can simply use index instead of iterating all the pods to make it more performant.

If the concern is that "what happens if the after point "3." someone adds the pod using PVC, but this is somewhat consistent with LIST pods approach -- we check state at some point and if it changes after the list call starts we won't see it. I presume that there is some mechanism that prevents PVCs waiting for deletion to be bound to the pods that prevents creating new pods after PVC is deleted, so this should be OK.
In fact the solution proposed by me in https://github.com/kubernetes/kubernetes/issues/109282#issuecomment-1263207774 isn't correct as the resourceVersion in LIST call and in watch can differ much if there are no changes to a given resource type, e.g.:
* WATCH on pods observed all pod changes up to RV=50
* LIST returns RV=100, as there are some changes to e.g. replicasets or configmaps using the same etcd
* As long as there is no pod change between RV=50 and RV=100 or we don't get a watch bookmark (every ~1m?), the watch will be still on RV=50, so waiting for it to be on RV=100 (as described in algorithm) may take significant time.
The Kubernetes project currently lacks enough contributors to adequately respond to all issues and PRs.

This bot triages issues and PRs according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue or PR as fresh with `/remove-lifecycle stale`
- Mark this issue or PR as rotten with `/lifecycle rotten`
- Close this issue or PR with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle stale

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/
/remove-lifecycle stale
/triage accepted
Do [watch bookmarks](https://kubernetes.io/docs/reference/using-api/api-concepts/#watch-bookmarks) help with the resource version problem mentioned above? The semantics of bookmarks seem a little vague, but maybe we can depend on a bookmark being sent frequently enough to tell our watch is up-to-date.
From the docs:
> As a client, you can request BOOKMARK events by setting the allowWatchBookmarks=true query parameter to a watch request, but you shouldn't assume bookmarks are returned at any specific interval, nor can clients assume that the API server will send any BOOKMARK event even when requested.

So I don't think this is reliable source, and here we are dealing with something that MUST work all the times.
My understanding is that when a bookmark event comes, the resource version there will tell us for sure if we are up-to-date on the pod list.

This is the same as @mborsz's proposal, except that if the bookmark events come more frequently than the watch RV gets updated, then it fixes the "significant time" problem @mboarsz mentioned.

Even if there isn't a strict frequency for the bookmark events to come, if they're frequent enough in practice that would probably be fine?
@xing-yang: The label(s) `priority/long-term` cannot be applied, because the repository doesn't have them.

<details>

In response to [this](https://github.com/kubernetes/kubernetes/issues/109282#issuecomment-1404065579):

>/priority long-term


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
/priority important-long-term
@xing-yang: The label(s) `priority/important-long-term` cannot be applied, because the repository doesn't have them.

<details>

In response to [this](https://github.com/kubernetes/kubernetes/issues/109282#issuecomment-1404066670):

>/priority important-long-term


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
/priority important-longterm
hi~any update?
One possible mitigation is to use `Node.Status.VolumesInUse` as a signal instead of `Pods`. However we would need to think through possible race conditions (ie pods getting added to nodes)
I also met the slow of deletion pvc, we have 30k pvc hang in terminating status and used up our capacity.

I also consider to recored the status from nodes, but i not sure how about multiple pod use the same pvc @msau42  
And i think Node.Status.VolumesInUse is complicated and bind too much condition, a new field may be more clear? 
can we do a timestamp based caching on the live list results? basically:
- when we do live lists, we grab the current timestamp from the api server with the [time api](https://github.com/kubernetes/apimachinery/blob/master/pkg/apis/meta/v1/time.go) right before we call the live list, then cache the results for the following delete calls for a period (cache TTL, say 1 sec).
- in the following deletevolumes, we use the list cache instead of calling live list (after already checking with informer) if the TTL hasn't passed and if the volume's deletionTimestamp is a previous time compared to the time when the list cache was generated. The cache will be refreshed if either of those conditions are not true.

I am not sure if there's any guarantee within the api server of the timestamp ordering but if so I haven't figured out a scenario that would cause a race condition. it will cause increase in processing latency if a pod is immediately deleted after a cache is generated and the volume is deleted before the cache is generated so a short TTL is desired.
We recently saw a similar issue. It was associated with many pods being deleted at once. That suggests that batching requests so that we amortize the cost of the loop over all the deleted pods would work -- I'm not sure if that's easy though in the current architecture of the controller.
Discussed offline: one possible mitigation is to give up on correctness and make this best effort by listing resources from informer. But we need to evaluate the behavior of what happens under that condition.
> Discussed offline: one possible mitigation is to give up on correctness and make this best effort by listing resources from informer. But we need to evaluate the behavior of what happens under that condition.

Wouldn't that just be going around in circles? https://github.com/kubernetes/kubernetes/pull/80492

The problem is listing every pod for every PVC. There is a lot of repeated information.

I think we know that once a PVC has a deletion timestamp, no new pods will attach to it? So if we've listed all pods & used PVCs at time t, then that list will be valid for all PVCs whose deletion timestamp is < t, and it's not necessary to hit the API server. 

This doesn't change much when PVCs are deleted infrequently. But the scaling issues we saw is when 10s of thousands of PVCs and their pods are deleted at once.
When kubelet admits a pod, it does a live Get() of PVC so it will error if PVC has a deletion timestamp on it. So new Pods won't start.

Using a timestamp based caching approach (similar to what @leiyiz proposed) of Pods and comparing to PVC deletion timestamp sounds plausible. Although in the case of thousands of PVCs deleted at once, I'm not sure how effective it will be.
Several of us in GKE have been discussing this and have come up with the following.

There’s consensus that the chance of the informer being stale is real, so any solution needs a live pod listing at some point -- a pathological case pointed out to us is where the apiserver becomes partitioned from etcd, so no informer updates come through, because there is no etcd watch to supply them.  Then, a controller reconciles on the stale data, and etcd becomes available just in time for the write that actuates the reconciliation to be handled.

There’s two ideas we have: batching work, and adding a pod list cache. We think the batching idea is better so I’ll start with that. Any comments/reactions or should we go ahead and do this?

**Batching**

Algorithm sketch
* In [processNextWorkItem](https://github.com/kubernetes/kubernetes/blob/release-1.27/pkg/controller/volume/pvcprotection/pvc_protection_controller.go#L122), pull all available items from the queue.
* call processPVC, adding a lazyLivePodList to the call
* a lazyLivePodList (naming TBD) is a type that returns a list of pods. The first time it calls the api server. Subsequent calls return the first results
    - This is needed because most times processPVC is called, no live pod list is done, because PVCs are in use and we just work off the informer

Open questions
* The controller uses a rate limiting workqueue. Do we have to change that to batch enough PVCs together? Probably we should write a PoC first and see how it works. The main point of rate limiting seems to be to not busy-retry failures, so as long as we have a rate limiter in the process work item frequency it should be fine.

**Cache**

This method assumes that deletion timestamps are increasing. We assume we can’t have a pvc A with deletion timestamp X in our informer, then do a pod listing, then have a pod start using pvc B, and then have B get a deletion timestamp X-1.

Algorithm sketch
* Using the existing PVC informer, watch for PVCs with deletion timestamps and cache the most recent timestamp in PVC protection controller memory.
* When a live pod listing is performed, mark it with the most recent PVC deletion timestamp (as seen by PVC informer).
* When [checking if a PVC marked for deletion is in use](https://github.com/kubernetes/kubernetes/blob/release-1.27/pkg/controller/volume/pvcprotection/pvc_protection_controller.go#L213), 
  - check if it is in use according to the pod informer cache; if so return in use
  - check if the deletion timestamp is less than or equal to the timestamp of the cached live pod listing; if so return in use
  - only if the previous checks fails, do a live pod listing and return according to that.

With this method, a live pod listing is only done for a newly deleted PVC. If there is a backlog of deleted PVCs, once one live listing is done, no new listing will be done until the backlog is cleaned out, which will happen quickly since the pod list is in-memory.

After a brief discussion with @liggitt on the [sig-api-machinery slack](https://kubernetes.slack.com/archives/C0EG7JC6T/p1711418407770279?thread_ts=1711053796.643179&cid=C0EG7JC6T), the batching idea seems good.

The nuances are (1) to ensure we keep the batch groups per-namespace so we keep the same filtering on pods, and (2) rate limiting should happen on the queue that makes api server calls and requeues errors. So that means we won't rate-limit the pvc informer queue. On pvc notification we add it to a per-namespace map, add the namespace to a workqueue that is rate-limited.

Please respond on this thread or reach out on slack if you have any questions or comments!
Do you plan to cache every namespace that has PVCs or do you want to release caches for old namespaces (and how old)?
With the batching method there is no cache to track --- that's its primary advantage.

Instead of caching, we make just one live api server call per batch, so it gets the same scaling benefits as using a cache.
> With the batching method there is no cache to track --- that's its primary advantage.
> 
> Instead of caching, we make just one live api server call per batch, so it gets the same scaling benefits as using a cache.

Um... every informer maintains a cache internally... if the controller is already maintaining a cache of pods (which kube-controller-manager is), a first pass checking use in cached pods would be *fast* and require no API server lookups. Only if the cached pods indicated no use for some of the PVCs pending definalization in the namespace would a live pod lookup be required.
It's probably buried in the comments above, but the current implementation does look at the informer cache and only if necessary do a live api check. We wouldn't be changing that behavior, ie if all pvcs are in use according to the informer cache then no live api call will be performed.

The comment about the cache is around the second option in [this comment](https://github.com/kubernetes/kubernetes/issues/109282#issuecomment-2013443881) which it sounds like we've decided against.
> We wouldn't be changing that behavior, ie if all pvcs are in use according to the informer cache then no live api call will be performed.

Sounds good, just double-checking
One comment about the **Cache** approach.

> This method assumes that deletion timestamps are increasing. We assume we can’t have a pvc A with deletion timestamp X in our informer, then do a pod listing, then have a pod start using pvc B, and then have B get a deletion timestamp X-1.

Is there risk that clock skew between APIServers can break this assumption?

Consider the sequence of events:

1. APIServer1 leader is running at relative time X.
2. PVC A is deleted and assigned deletionTimestamp X+1.
3. PVCProtectionController admits PVC A to its workqueue and caches X+1 as the most recent timestamp.
4. PVCProtectionController performs a list Pods call, marking this cached result with X+1.
5. APIServer1 leader election fails over to APIServer2. APIServer2 has relative time Y. Y has a clock skew S, where Y=X-S
6. PVC B is deleted and assigned deletionTimestamp Y+2. PVC B resource has a deletionTimestamp of X+2-S
7. PVCProtectionController processes its workqueue. It sees that PVC A can be deleted (correctly). Depending on S, if S > 1, PVC B appears to have been deleted before X+1, despite having been deleted after X+1 (in real terms). This could cause PVC B to be incorrectly deleted if it is actually still in use at (7).

The reproduction of this depends on the latency of APIServer failover, the potential risk of actual clock skew between APIServers, and the queue processing speed of the kube-controller-manager.

Potential enhancements:
1) Compare the admission time of a PVC to the PVCProtectionController workqueue (eg: the Y+N time) This would rely on the monotonically increasing clock of the `kube-controller-manager` process, rather than relying on the distributed clocks of the APIServer
@pwschuurman I think you're right --- this is another reason to do the batching approach.
/assign @pwschuurman 

> Open questions
>
>  * The controller uses a rate limiting workqueue. Do we have to change that to batch enough PVCs together? Probably we should write a PoC first and see how it works. The main point of rate limiting seems to be to not busy-retry failures, so as long as we have a rate limiter in the process work item frequency it should be fine.

I think the following would work in processWorkQueue:

1) Pull all PVC items off the workQueue into a list.
    * This ensures that all fetched PVCs are at least as old as the next subsequent live pod list call. The list needs to be scoped the same as the lazyLivePodList function definition.
2) In a loop, process each PVC item separately.
    * If any errors are encountered for an individual PVC, add it back to the workQueue in a rate limited manner. This retains the existing behavior of ensuring that a PVC that consistently produces an error when processed is queued through the rate limiter.

> call processPVC, adding a lazyLivePodList to the call

What happens if the first call to live Pod list fails? Should we continue to return an error for subsequent lazyLivePodList calls, or should we re-try the call right away? We may want to introduce a rate limiter within the function itself, to only retry the live() call after a certain amount of time has lapsed (this may already exist within the Kubernetes client definition). That would introduce two compounding rate limiters (one for PVCs that produce errors), one rate limiter for calls to the Kubernetes API server. There are some edge cases where we might delay more than expected depending on the number of PVCs in the workQueue, and then backoff frequency state of the rate limiter that aren't ideal. It would be nicer to have a separate backoff strategies for PVCs to be reprocessed, and for API call rate limiting and backoff.
>What happens if the first call to live Pod list fails? Should we continue to return an error for subsequent lazyLivePodList 

Currently I think we requeue in a rate-limited fashion. Having some rate limiting within the lazy list seems to have the potential for complicated interactions as you point out.

Maybe we can just stall in the queue processing if a pod list error is signaled.
Looking into if https://github.com/kubernetes/enhancements/issues/2340 can solve this by making list all pods request much cheaper.
I don't know if that will help --- it still goes out over the network, so will be called for every volume, which is too much.
> Currently I think we requeue in a rate-limited fashion. Having some rate limiting within the lazy list seems to have the potential for complicated interactions as you point out.

To make the logic consistent (where failures in `askAPIServer` result in PVC being queued in a rate limited fashion), we could continue to return the same error in subsequent `lazyLivePodList` calls. This allows for a feedback loop of API Server list call failures resulting in rate limiting of PVC processing. If we don't do this (eg: we requeue without rate limiting), we may risk overloading the APIServer (we'd be dependent on client side throttling, based on the QPS that KCM has configured for the client, or compete for QPS with other clients in KCM).

I think this only causes a slight delay for PVCs to be reprocessed:
 * Previously: If we had N PVCs queued, and the 5th rate calls to list pods failed, only the 5th PVC would get requeued in a rate limited fashion. The 6th PVC would subsequently calls list pods, and it would all
 * New Behavior: If an error occurs on list pods for the 5th PVC, the 5th and 6th (up to N) PVCs will get requeued in a rate limited fashion. This introduces a small delay for the 6th PVC to get reprocessed (it needs to wait for rate limiting timeout), rather than getting reprocessed right away in the next loop iteration. If this is the first time getting requeued, this should be a short delay ([5ms, which is the default initial delay](https://github.com/kubernetes/client-go/blob/d3682da14f7208b5bb30bc2ae94d44606a77abac/util/workqueue/default_rate_limiters.go#L52)). So I think this should be acceptable.

We could potentially requeue with rate limiting based on the error rate (eg: if unavailable, or timeout, requeue right away), but I'm concerned this could lead to a situation where the client misbehaves and sends a large number of API requests against the APIServer, up to the QPS limit.
Yeah that's a good point. We definitely want to rate limit live list calls.

I think we can work through the mechanics on the PR --- the goal should remain though to be able to delete 10s of k PVCs in minutes.

Fix patch:
diff --git a/pkg/controller/volume/pvcprotection/pvc_protection_controller.go b/pkg/controller/volume/pvcprotection/pvc_protection_controller.go
index 4632c7007e885..005a15d4b5128 100644
--- a/pkg/controller/volume/pvcprotection/pvc_protection_controller.go
+++ b/pkg/controller/volume/pvcprotection/pvc_protection_controller.go
@@ -19,6 +19,7 @@ package pvcprotection
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	v1 "k8s.io/api/core/v1"
@@ -41,6 +42,65 @@ 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
 
@@ -51,7 +111,8 @@ type Controller struct {
 	podListerSynced cache.InformerSynced
 	podIndexer      cache.Indexer
 
-	queue workqueue.TypedRateLimitingInterface[string]
+	queue              workqueue.TypedRateLimitingInterface[string]
+	pvcProcessingStore *pvcProcessingStore
 }
 
 // NewPVCProtectionController returns a new instance of PVCProtectionController.
@@ -62,6 +123,7 @@ func NewPVCProtectionController(logger klog.Logger, pvcInformer coreinformers.Pe
 			workqueue.DefaultTypedControllerRateLimiter[string](),
 			workqueue.TypedRateLimitingQueueConfig[string]{Name: "pvcprotection"},
 		),
+		pvcProcessingStore: NewPVCProcessingStore(),
 	}
 
 	e.pvcLister = pvcInformer.Lister()
@@ -100,6 +162,7 @@ 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")
@@ -109,25 +172,31 @@ 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.runWorker, time.Second)
+		go wait.UntilWithContext(ctx, c.runProcessNamespaceWorker, time.Second)
 	}
 
 	<-ctx.Done()
 }
 
-func (c *Controller) runWorker(ctx context.Context) {
-	for c.processNextWorkItem(ctx) {
+// 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() {
+	}
+}
+
+// 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 {
+func (c *Controller) processNextWorkItem() bool {
 	pvcKey, quit := c.queue.Get()
 	if quit {
 		return false
 	}
-	defer c.queue.Done(pvcKey)
 
 	pvcNamespace, pvcName, err := cache.SplitMetaNamespaceKey(pvcKey)
 	if err != nil {
@@ -135,19 +204,32 @@ func (c *Controller) processNextWorkItem(ctx context.Context) bool {
 		return true
 	}
 
-	err = c.processPVC(ctx, pvcNamespace, pvcName)
-	if err == nil {
-		c.queue.Forget(pvcKey)
-		return true
-	}
+	c.pvcProcessingStore.addOrUpdate(pvcNamespace, pvcKey, pvcName)
+	return true
+}
 
-	utilruntime.HandleError(fmt.Errorf("PVC %v failed with : %w", pvcKey, err))
-	c.queue.AddRateLimited(pvcKey)
+func (c *Controller) processPVCsByNamespace(ctx context.Context) bool {
+	pvcList, namespace := c.pvcProcessingStore.flushNextPVCsByNamespace()
+	if pvcList == nil {
+		return false
+	}
 
+	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/%v failed with: %w", pvcName, namespace, err))
+		}
+		c.queue.Done(pvcKey)
+	}
 	return true
 }
 
-func (c *Controller) processPVC(ctx context.Context, pvcNamespace, pvcName string) error {
+func (c *Controller) processPVC(ctx context.Context, pvcNamespace, pvcName string, lazyLivePodList *LazyLivePodList) error {
 	logger := klog.FromContext(ctx)
 	logger.V(4).Info("Processing PVC", "PVC", klog.KRef(pvcNamespace, pvcName))
 	startTime := time.Now()
@@ -167,7 +249,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)
+		isUsed, err := c.isBeingUsed(ctx, pvc, lazyLivePodList)
 		if err != nil {
 			return err
 		}
@@ -213,7 +295,7 @@ func (c *Controller) removeFinalizer(ctx context.Context, pvc *v1.PersistentVolu
 	return nil
 }
 
-func (c *Controller) isBeingUsed(ctx context.Context, pvc *v1.PersistentVolumeClaim) (bool, error) {
+func (c *Controller) isBeingUsed(ctx context.Context, pvc *v1.PersistentVolumeClaim, lazyLivePodList *LazyLivePodList) (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.
@@ -228,8 +310,12 @@ func (c *Controller) isBeingUsed(ctx context.Context, pvc *v1.PersistentVolumeCl
 	// Even if no Pod using pvc was found in the Informer's cache it doesn't
 	// 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.
-	return c.askAPIServer(ctx, pvc)
+	// it among those returned by a "lazy" live list.
+
+	// Use a "lazy" live pod list: lazyLivePodList caches the first successful live pod list response,
+	// so for a large number of PVC deletions in a short duration, subsequent requests can use the cached pod list
+	// instead of issuing a lot of API requests. The cache is refreshed for each run of processNextWorkItem().
+	return c.askAPIServer(ctx, pvc, lazyLivePodList)
 }
 
 func (c *Controller) askInformer(logger klog.Logger, pvc *v1.PersistentVolumeClaim) (bool, error) {
@@ -258,16 +344,24 @@ func (c *Controller) askInformer(logger klog.Logger, pvc *v1.PersistentVolumeCla
 	return false, nil
 }
 
-func (c *Controller) askAPIServer(ctx context.Context, pvc *v1.PersistentVolumeClaim) (bool, error) {
+func (c *Controller) askAPIServer(ctx context.Context, pvc *v1.PersistentVolumeClaim, lazyLivePodList *LazyLivePodList) (bool, error) {
 	logger := klog.FromContext(ctx)
-	logger.V(4).Info("Looking for Pods using PVC with a live list", "PVC", klog.KObj(pvc))
+	logger.V(4).Info("Looking for Pods using PVC", "PVC", klog.KObj(pvc))
+	if lazyLivePodList.getCache() == nil {
+		logger.V(4).Info("Live listing Pods in namespace", "namespace", pvc.Namespace)
+		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())
+		}
 
-	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)
+		}
 	}
 
-	for _, pod := range podsList.Items {
+	for _, pod := range lazyLivePodList.getCache() {
 		if c.podUsesPVC(logger, &pod, pvc) {
 			return true, nil
 		}


Test patch:
diff --git a/pkg/controller/volume/pvcprotection/pvc_protection_controller_test.go b/pkg/controller/volume/pvcprotection/pvc_protection_controller_test.go
index a75c752eaa96c..3884ac94e6a17 100644
--- a/pkg/controller/volume/pvcprotection/pvc_protection_controller_test.go
+++ b/pkg/controller/volume/pvcprotection/pvc_protection_controller_test.go
@@ -47,6 +47,7 @@ type reaction struct {
 
 const (
 	defaultNS       = "default"
+	namespace2      = "namespace-2"
 	defaultPVCName  = "pvc1"
 	defaultPodName  = "pod1"
 	defaultNodeName = "node1"
@@ -69,6 +70,22 @@ func pod() *v1.Pod {
 	}
 }
 
+func podWithConfig(name string, namespace string) *v1.Pod {
+	return &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+			UID:       defaultUID,
+		},
+		Spec: v1.PodSpec{
+			NodeName: defaultNodeName,
+		},
+		Status: v1.PodStatus{
+			Phase: v1.PodPending,
+		},
+	}
+}
+
 func unscheduled(pod *v1.Pod) *v1.Pod {
 	pod.Spec.NodeName = ""
 	return pod
@@ -117,6 +134,15 @@ func pvc() *v1.PersistentVolumeClaim {
 	}
 }
 
+func pvcWithConfig(name string, namespace string) *v1.PersistentVolumeClaim {
+	return &v1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+	}
+}
+
 func withProtectionFinalizer(pvc *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
 	pvc.Finalizers = append(pvc.Finalizers, volumeutil.PVCProtectionFinalizer)
 	return pvc
@@ -146,6 +172,23 @@ func generateUpdateErrorFunc(t *testing.T, failures int) clienttesting.ReactionF
 	}
 }
 
+func generatePodListErrorFunc(t *testing.T, failures int) clienttesting.ReactionFunc {
+	i := 0
+	return func(action clienttesting.Action) (bool, runtime.Object, error) {
+		i++
+		if i <= failures {
+			// Pod List fails
+			list, ok := action.(clienttesting.ListAction)
+			if !ok {
+				t.Fatalf("Reactor got non-list action: %+v", action)
+			}
+			return true, nil, apierrors.NewForbidden(list.GetResource().GroupResource(), "mock pod", errors.New("Mock error"))
+		}
+		// List succeeds
+		return false, nil, nil
+	}
+}
+
 func TestPVCProtectionController(t *testing.T) {
 	pvcGVR := schema.GroupVersionResource{
 		Group:    v1.GroupName,
@@ -175,7 +218,7 @@ func TestPVCProtectionController(t *testing.T) {
 		reactors []reaction
 		// PVC event to simulate. This PVC will be automatically added to
 		// initialObjects.
-		updatedPVC *v1.PersistentVolumeClaim
+		updatedPVCs []*v1.PersistentVolumeClaim
 		// Pod event to simulate. This Pod will be automatically added to
 		// initialObjects.
 		updatedPod *v1.Pod
@@ -190,20 +233,21 @@ func TestPVCProtectionController(t *testing.T) {
 		// PVC events
 		//
 		{
-			name:       "PVC without finalizer -> finalizer is added",
-			updatedPVC: pvc(),
+			name:        "PVC without finalizer -> finalizer is added",
+			updatedPVCs: []*v1.PersistentVolumeClaim{pvc(), pvcWithConfig("pvc2", "namespace-2")},
 			expectedActions: []clienttesting.Action{
 				clienttesting.NewUpdateAction(pvcGVR, defaultNS, withProtectionFinalizer(pvc())),
+				clienttesting.NewUpdateAction(pvcGVR, "namespace-2", withProtectionFinalizer(pvcWithConfig("pvc2", "namespace-2"))),
 			},
 		},
 		{
 			name:            "PVC with finalizer -> no action",
-			updatedPVC:      withProtectionFinalizer(pvc()),
+			updatedPVCs:     []*v1.PersistentVolumeClaim{withProtectionFinalizer(pvc()), withProtectionFinalizer(pvcWithConfig("pvc2", "namespace-2"))},
 			expectedActions: []clienttesting.Action{},
 		},
 		{
-			name:       "saving PVC finalizer fails -> controller retries",
-			updatedPVC: pvc(),
+			name:        "saving PVC finalizer fails -> controller retries",
+			updatedPVCs: []*v1.PersistentVolumeClaim{pvc()},
 			reactors: []reaction{
 				{
 					verb:      "update",
@@ -221,16 +265,30 @@ func TestPVCProtectionController(t *testing.T) {
 			},
 		},
 		{
-			name:       "deleted PVC with finalizer -> finalizer is removed",
-			updatedPVC: deleted(withProtectionFinalizer(pvc())),
+			name: "deleted PVC with finalizers across different namespaces -> finalizer is removed",
+			updatedPVCs: []*v1.PersistentVolumeClaim{deleted(withProtectionFinalizer(pvc())),
+				deleted(withProtectionFinalizer(pvcWithConfig("pvc2", "namespace-2")))},
 			expectedActions: []clienttesting.Action{
 				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
 				clienttesting.NewUpdateAction(pvcGVR, defaultNS, deleted(pvc())),
+				clienttesting.NewListAction(podGVR, podGVK, "namespace-2", metav1.ListOptions{}),
+				clienttesting.NewUpdateAction(pvcGVR, "namespace-2", deleted(pvcWithConfig("pvc2", "namespace-2"))),
 			},
 		},
 		{
-			name:       "finalizer removal fails -> controller retries",
-			updatedPVC: deleted(withProtectionFinalizer(pvc())),
+			name: "multiple PVCs with finalizer for the same namespace; no alive pods -> finalizer is removed",
+			updatedPVCs: []*v1.PersistentVolumeClaim{deleted(withProtectionFinalizer(pvc())),
+				deleted(withProtectionFinalizer(pvcWithConfig("pvc2", defaultNS)))},
+			expectedActions: []clienttesting.Action{
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateAction(pvcGVR, defaultNS, deleted(pvc())),
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateAction(pvcGVR, defaultNS, deleted(pvcWithConfig("pvc2", defaultNS))),
+			},
+		},
+		{
+			name:        "finalizer removal fails -> controller retries",
+			updatedPVCs: []*v1.PersistentVolumeClaim{deleted(withProtectionFinalizer(pvc()))},
 			reactors: []reaction{
 				{
 					verb:      "update",
@@ -250,12 +308,33 @@ func TestPVCProtectionController(t *testing.T) {
 				clienttesting.NewUpdateAction(pvcGVR, defaultNS, deleted(pvc())),
 			},
 		},
+		{
+			name:        "delete multiple PVCs of the same namespace; pod list fails for one PVC -> add failing PVC back to queue and continue to the next PVC",
+			updatedPVCs: []*v1.PersistentVolumeClaim{deleted(withProtectionFinalizer(pvc())), deleted(withProtectionFinalizer(pvcWithConfig("pvc2", defaultNS)))},
+			reactors: []reaction{
+				{
+					verb:      "list",
+					resource:  "pods",
+					reactorfn: generatePodListErrorFunc(t, 1 /* update fails twice*/),
+				},
+			},
+			expectedActions: []clienttesting.Action{
+				// Fails
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				// Succeed with next pvc in queue
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateAction(pvcGVR, defaultNS, deleted(pvcWithConfig("pvc2", defaultNS))),
+
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateAction(pvcGVR, defaultNS, deleted(pvc())),
+			},
+		},
 		{
 			name: "deleted PVC with finalizer + pod with the PVC exists -> finalizer is not removed",
 			initialObjects: []runtime.Object{
 				withPVC(defaultPVCName, pod()),
 			},
-			updatedPVC:      deleted(withProtectionFinalizer(pvc())),
+			updatedPVCs:     []*v1.PersistentVolumeClaim{deleted(withProtectionFinalizer(pvc()))},
 			expectedActions: []clienttesting.Action{},
 		},
 		{
@@ -263,10 +342,28 @@ func TestPVCProtectionController(t *testing.T) {
 			initialObjects: []runtime.Object{
 				withEmptyDir(withPVC("unrelatedPVC", pod())),
 			},
-			updatedPVC: deleted(withProtectionFinalizer(pvc())),
+			updatedPVCs: []*v1.PersistentVolumeClaim{deleted(withProtectionFinalizer(pvc()))},
+			expectedActions: []clienttesting.Action{
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateAction(pvcGVR, defaultNS, deleted(pvc())),
+			},
+		},
+		{
+			name: "deleted multiple PVCs with finalizer (same namespace) + pod with unrelated PVC and EmptyDir exists",
+			initialObjects: []runtime.Object{
+				withEmptyDir(withPVC("unrelatedPVC", pod())),
+			},
+			updatedPVCs: []*v1.PersistentVolumeClaim{deleted(withProtectionFinalizer(pvc())),
+				deleted(withProtectionFinalizer(pvcWithConfig("pvc2", defaultNS))),
+				deleted(withProtectionFinalizer(pvcWithConfig("pvc3", defaultNS))),
+			},
 			expectedActions: []clienttesting.Action{
 				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
 				clienttesting.NewUpdateAction(pvcGVR, defaultNS, deleted(pvc())),
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateAction(pvcGVR, defaultNS, deleted(pvcWithConfig("pvc2", defaultNS))),
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateAction(pvcGVR, defaultNS, deleted(pvcWithConfig("pvc3", defaultNS))),
 			},
 		},
 		{
@@ -274,7 +371,7 @@ func TestPVCProtectionController(t *testing.T) {
 			initialObjects: []runtime.Object{
 				withStatus(v1.PodFailed, withPVC(defaultPVCName, pod())),
 			},
-			updatedPVC:      deleted(withProtectionFinalizer(pvc())),
+			updatedPVCs:     []*v1.PersistentVolumeClaim{deleted(withProtectionFinalizer(pvc()))},
 			expectedActions: []clienttesting.Action{},
 		},
 		{
@@ -283,9 +380,25 @@ func TestPVCProtectionController(t *testing.T) {
 				withPVC(defaultPVCName, pod()),
 			},
 			informersAreLate: true,
-			updatedPVC:       deleted(withProtectionFinalizer(pvc())),
+			updatedPVCs:      []*v1.PersistentVolumeClaim{deleted(withProtectionFinalizer(pvc()))},
+			expectedActions: []clienttesting.Action{
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+			},
+		},
+
+		{
+			name: "mix of deleted PVCs with some used by a pod and some unused -> finalizer is removed only for unused PVCs",
+			initialObjects: []runtime.Object{
+				withPVC(defaultPVCName, pod()),
+				withPVC("pvc3", podWithConfig("pod2", "namespace-3")),
+			},
+			informersAreLate: true,
+			updatedPVCs:      []*v1.PersistentVolumeClaim{deleted(withProtectionFinalizer(pvc())), deleted(withProtectionFinalizer(pvcWithConfig("pvc2", defaultNS))), deleted(withProtectionFinalizer(pvcWithConfig("pvc3", "namespace-3")))},
 			expectedActions: []clienttesting.Action{
 				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateAction(pvcGVR, defaultNS, deleted(pvcWithConfig("pvc2", defaultNS))),
+				clienttesting.NewListAction(podGVR, podGVK, "namespace-3", metav1.ListOptions{}),
 			},
 		},
 		//
@@ -376,9 +489,11 @@ func TestPVCProtectionController(t *testing.T) {
 			clientObjs    []runtime.Object
 			informersObjs []runtime.Object
 		)
-		if test.updatedPVC != nil {
-			clientObjs = append(clientObjs, test.updatedPVC)
-			informersObjs = append(informersObjs, test.updatedPVC)
+		if test.updatedPVCs != nil {
+			for i := 0; i < len(test.updatedPVCs); i++ {
+				clientObjs = append(clientObjs, test.updatedPVCs[i])
+				informersObjs = append(informersObjs, test.updatedPVCs[i])
+			}
 		}
 		if test.updatedPod != nil {
 			clientObjs = append(clientObjs, test.updatedPod)
@@ -388,7 +503,6 @@ func TestPVCProtectionController(t *testing.T) {
 		if !test.informersAreLate {
 			informersObjs = append(informersObjs, test.initialObjects...)
 		}
-
 		// Create client with initial data
 		client := fake.NewSimpleClientset(clientObjs...)
 
@@ -423,8 +537,10 @@ func TestPVCProtectionController(t *testing.T) {
 		}
 
 		// Start the test by simulating an event
-		if test.updatedPVC != nil {
-			ctrl.pvcAddedUpdated(logger, test.updatedPVC)
+		if test.updatedPVCs != nil {
+			for i := 0; i < len(test.updatedPVCs); i++ {
+				ctrl.pvcAddedUpdated(logger, test.updatedPVCs[i])
+			}
 		}
 		switch {
 		case test.deletedPod != nil && test.updatedPod != nil && test.deletedPod.Namespace == test.updatedPod.Namespace && test.deletedPod.Name == test.updatedPod.Name:
@@ -445,12 +561,18 @@ func TestPVCProtectionController(t *testing.T) {
 			}
 			if ctrl.queue.Len() > 0 {
 				logger.V(5).Info("Non-empty queue, processing one", "test", test.name, "queueLength", ctrl.queue.Len())
-				ctrl.processNextWorkItem(context.TODO())
+				ctx := context.TODO()
+				ctrl.processNextWorkItem()
+				for ctrl.pvcProcessingStore.namespaceQueue.Len() != 0 {
+					ctrl.processPVCsByNamespace(ctx)
+				}
 			}
+
 			if ctrl.queue.Len() > 0 {
 				// There is still some work in the queue, process it now
 				continue
 			}
+
 			currentActionCount := len(client.Actions())
 			if currentActionCount < len(test.expectedActions) {
 				// Do not log every wait, only when the action count changes.
diff --git a/test/e2e/storage/testsuites/base.go b/test/e2e/storage/testsuites/base.go
index a6efd496512c5..067e86c61cb3e 100644
--- a/test/e2e/storage/testsuites/base.go
+++ b/test/e2e/storage/testsuites/base.go
@@ -81,6 +81,7 @@ var CSISuites = append(BaseSuites,
 	InitSnapshottableTestSuite,
 	InitSnapshottableStressTestSuite,
 	InitVolumePerformanceTestSuite,
+	InitPvcDeletionPerformanceTestSuite,
 	InitReadWriteOncePodTestSuite,
 	InitVolumeModifyTestSuite,
 )
diff --git a/test/e2e/storage/testsuites/pvcdeletionperf.go b/test/e2e/storage/testsuites/pvcdeletionperf.go
new file mode 100644
index 0000000000000..018e225a2d229
--- /dev/null
+++ b/test/e2e/storage/testsuites/pvcdeletionperf.go
@@ -0,0 +1,248 @@
+/*
+Copyright 2024 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package testsuites
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/onsi/ginkgo/v2"
+
+	v1 "k8s.io/api/core/v1"
+	storagev1 "k8s.io/api/storage/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/kubernetes/test/e2e/framework"
+	e2epod "k8s.io/kubernetes/test/e2e/framework/pod"
+	e2epv "k8s.io/kubernetes/test/e2e/framework/pv"
+	e2eskipper "k8s.io/kubernetes/test/e2e/framework/skipper"
+	storageframework "k8s.io/kubernetes/test/e2e/storage/framework"
+	admissionapi "k8s.io/pod-security-admission/api"
+)
+
+type pvcDeletionPerformanceTestSuite struct {
+	tsInfo storageframework.TestSuiteInfo
+}
+
+var _ storageframework.TestSuite = &pvcDeletionPerformanceTestSuite{}
+
+const pvcDeletionTestTimeout = 30 * time.Minute
+
+// InitPvcDeletionPerformanceTestSuite returns pvcDeletionPerformanceTestSuite that implements TestSuite interface
+// This test suite brings up a number of pods and PVCS (configured upstream), deletes the pods, and then deletes the PVCs.
+// The main goal is to record the duration for the PVC/PV deletion process for each run, and so the test doesn't set explicit expectations to match against.
+func InitPvcDeletionPerformanceTestSuite() storageframework.TestSuite {
+	return &pvcDeletionPerformanceTestSuite{
+		tsInfo: storageframework.TestSuiteInfo{
+			Name: "pvc-deletion-performance",
+			TestPatterns: []storageframework.TestPattern{
+				storageframework.BlockVolModeDynamicPV,
+			},
+		},
+	}
+}
+
+func (t *pvcDeletionPerformanceTestSuite) GetTestSuiteInfo() storageframework.TestSuiteInfo {
+	return t.tsInfo
+}
+
+func (t *pvcDeletionPerformanceTestSuite) SkipUnsupportedTests(driver storageframework.TestDriver, pattern storageframework.TestPattern) {
+}
+
+func (t *pvcDeletionPerformanceTestSuite) DefineTests(driver storageframework.TestDriver, pattern storageframework.TestPattern) {
+	type local struct {
+		config  *storageframework.PerTestConfig
+		cs      clientset.Interface
+		ns      *v1.Namespace
+		scName  string
+		pvcs    []*v1.PersistentVolumeClaim
+		options *storageframework.PerformanceTestOptions
+		stopCh  chan struct{}
+		pods    []*v1.Pod
+	}
+	var (
+		dInfo *storageframework.DriverInfo
+		l     *local
+	)
+	ginkgo.BeforeEach(func() {
+		// Check preconditions
+		dDriver := driver.(storageframework.DynamicPVTestDriver)
+		if dDriver == nil {
+			e2eskipper.Skipf("Test driver does not support dynamically created volumes")
+		}
+		dInfo = dDriver.GetDriverInfo()
+		if dInfo == nil {
+			e2eskipper.Skipf("Failed to get Driver info -- skipping")
+		}
+		if dInfo.PerformanceTestOptions == nil || dInfo.PerformanceTestOptions.ProvisioningOptions == nil {
+			e2eskipper.Skipf("Driver %s doesn't specify performance test options -- skipping", dInfo.Name)
+		}
+	})
+
+	// Set high QPS for the framework to avoid client-side throttling from the test itself,
+	// which can interfere with measuring deletion time
+	frameworkOptions := framework.Options{
+		ClientQPS:   500,
+		ClientBurst: 1000,
+	}
+	f := framework.NewFramework("pvc-deletion-performance", frameworkOptions, nil)
+	f.NamespacePodSecurityLevel = admissionapi.LevelPrivileged
+
+	ginkgo.AfterEach(func(ctx context.Context) {
+		if l != nil {
+			if l.stopCh != nil {
+				ginkgo.By("Closing informer channel")
+				close(l.stopCh)
+			}
+			deletingStats := &performanceStats{
+				mutex:             &sync.Mutex{},
+				perObjectInterval: make(map[string]*interval),
+				operationMetrics:  &storageframework.Metrics{},
+			}
+			var (
+				errs []error
+				mu   sync.Mutex
+				wg   sync.WaitGroup
+			)
+
+			wg.Add(len(l.pods))
+			for _, pod := range l.pods {
+				go func(pod *v1.Pod) {
+					defer ginkgo.GinkgoRecover()
+					defer wg.Done()
+
+					framework.Logf("Deleting pod %v", pod.Name)
+					err := e2epod.DeletePodWithWait(ctx, l.cs, pod)
+					mu.Lock()
+					defer mu.Unlock()
+					errs = append(errs, err)
+				}(pod)
+			}
+			wg.Wait()
+
+			ginkgo.By("Deleting all PVCs")
+
+			startTime := time.Now()
+			wg.Add(len(l.pvcs))
+			for _, pvc := range l.pvcs {
+				go func(pvc *v1.PersistentVolumeClaim) { // Start a goroutine for each PVC
+					defer wg.Done() // Decrement the counter when the goroutine finishes
+					startDeletingPvcTime := time.Now()
+					framework.Logf("Start deleting PVC %v", pvc.GetName())
+					deletingStats.mutex.Lock()
+					deletingStats.perObjectInterval[pvc.Name] = &interval{
+						create: startDeletingPvcTime,
+					}
+					deletingStats.mutex.Unlock()
+					err := e2epv.DeletePersistentVolumeClaim(ctx, l.cs, pvc.Name, pvc.Namespace)
+					framework.ExpectNoError(err)
+					startDeletingPVTime := time.Now()
+					err = e2epv.WaitForPersistentVolumeDeleted(ctx, l.cs, pvc.Spec.VolumeName, 1*time.Second, 100*time.Minute)
+					framework.Logf("Deleted PV %v, PVC %v in %v", pvc.Spec.VolumeName, pvc.GetName(), time.Since(startDeletingPVTime))
+					framework.ExpectNoError(err)
+				}(pvc)
+			}
+			wg.Wait()
+
+			endTime := time.Now() // Capture overall end time
+			totalDuration := endTime.Sub(startTime)
+			framework.Logf("Deleted all PVC/PVs in %v", totalDuration) // Log total deletion time
+
+			ginkgo.By(fmt.Sprintf("Deleting Storage Class %s", l.scName))
+			err := l.cs.StorageV1().StorageClasses().Delete(ctx, l.scName, metav1.DeleteOptions{})
+			framework.ExpectNoError(err)
+
+		} else {
+			ginkgo.By("Local l setup is nil")
+		}
+	})
+
+	f.It("should delete volumes at scale within performance constraints", f.WithSlow(), f.WithSerial(), func(ctx context.Context) {
+		l = &local{
+			cs:      f.ClientSet,
+			ns:      f.Namespace,
+			options: dInfo.PerformanceTestOptions,
+		}
+		l.config = driver.PrepareTest(ctx, f)
+
+		sc := driver.(storageframework.DynamicPVTestDriver).GetDynamicProvisionStorageClass(ctx, l.config, pattern.FsType)
+		ginkgo.By(fmt.Sprintf("Creating Storage Class %v", sc))
+		if sc.VolumeBindingMode != nil && *sc.VolumeBindingMode == storagev1.VolumeBindingWaitForFirstConsumer {
+			e2eskipper.Skipf("WaitForFirstConsumer binding mode currently not supported for this test")
+		}
+		ginkgo.By(fmt.Sprintf("Creating Storage Class %s", sc.Name))
+		sc, err := l.cs.StorageV1().StorageClasses().Create(ctx, sc, metav1.CreateOptions{})
+		framework.ExpectNoError(err)
+		l.scName = sc.Name
+
+		// Stats for volume provisioning operation; we only need this because imported function newPVCWatch from volumeperf.go requires this as an argument
+		// (this test itself doesn't use these stats)
+		provisioningStats := &performanceStats{
+			mutex:             &sync.Mutex{},
+			perObjectInterval: make(map[string]*interval),
+			operationMetrics:  &storageframework.Metrics{},
+		}
+		// Create a controller to watch on PVCs
+		// When all PVCs provisioned by this test are in the Bound state, the controller
+		// sends a signal to the channel
+		controller := newPVCWatch(ctx, f, l.options.ProvisioningOptions.Count, provisioningStats)
+		l.stopCh = make(chan struct{})
+		go controller.Run(l.stopCh)
+		waitForProvisionCh = make(chan []*v1.PersistentVolumeClaim)
+
+		ginkgo.By(fmt.Sprintf("Creating %d PVCs of size %s", l.options.ProvisioningOptions.Count, l.options.ProvisioningOptions.VolumeSize))
+		for i := 0; i < l.options.ProvisioningOptions.Count; i++ {
+			pvc := e2epv.MakePersistentVolumeClaim(e2epv.PersistentVolumeClaimConfig{
+				ClaimSize:        l.options.ProvisioningOptions.VolumeSize,
+				StorageClassName: &sc.Name,
+			}, l.ns.Name)
+			pvc, err = l.cs.CoreV1().PersistentVolumeClaims(l.ns.Name).Create(ctx, pvc, metav1.CreateOptions{})
+			framework.ExpectNoError(err)
+			// Store create time for each PVC
+			provisioningStats.mutex.Lock()
+			provisioningStats.perObjectInterval[pvc.Name] = &interval{
+				create: pvc.CreationTimestamp.Time,
+			}
+			provisioningStats.mutex.Unlock()
+			// Create pods
+			podConfig := e2epod.Config{
+				NS:           l.ns.Name,
+				SeLinuxLabel: e2epv.SELinuxLabel,
+			}
+			pod, _ := e2epod.MakeSecPod(&podConfig)
+			_, err = l.cs.CoreV1().Pods(pod.Namespace).Create(ctx, pod, metav1.CreateOptions{})
+			if err != nil {
+				framework.Failf("Failed to create pod [%+v]. Error: %v", pod, err)
+			}
+			framework.ExpectNoError(err)
+
+			l.pods = append(l.pods, pod)
+		}
+
+		ginkgo.By("Waiting for all PVCs to be Bound...")
+
+		select {
+		case l.pvcs = <-waitForProvisionCh:
+			framework.Logf("All PVCs in Bound state")
+		case <-time.After(pvcDeletionTestTimeout):
+			ginkgo.Fail(fmt.Sprintf("expected all PVCs to be in Bound state within %v", pvcDeletionTestTimeout.Round(time.Second)))
+		}
+	})
+
+}