add unused condition to persistent volume claims

2026-06-30 22:36:34 +00:00 · 2026-03-17 15:06:41 +01:00
parent 1a3fd46367
commit 6c087b2724
9 changed files with 335 additions and 11 deletions
--- a/pkg/apis/core/types.go
+++ b/pkg/apis/core/types.go
@@ -616,6 +616,11 @@ const (
 	PersistentVolumeClaimVolumeModifyVolumeError PersistentVolumeClaimConditionType = "ModifyVolumeError"
 	// Volume is being modified
 	PersistentVolumeClaimVolumeModifyingVolume PersistentVolumeClaimConditionType = "ModifyingVolume"
+
+	// PersistentVolumeClaimUnused indicates whether the PVC is currently not in use by any Pod.
+	// When status is True, the PVC is not referenced by any non-terminal Pod.
+	// The lastTransitionTime indicates when the PVC last transitioned to being unused.
+	PersistentVolumeClaimUnused PersistentVolumeClaimConditionType = "Unused"
 )

 // +enum
--- a/pkg/controller/volume/pvcprotection/pvc_protection_controller.go
+++ b/pkg/controller/volume/pvcprotection/pvc_protection_controller.go
@@ -27,6 +27,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	coreinformers "k8s.io/client-go/informers/core/v1"
 	clientset "k8s.io/client-go/kubernetes"
 	corelisters "k8s.io/client-go/listers/core/v1"
@@ -36,6 +37,7 @@ import (
 	"k8s.io/klog/v2"
 	"k8s.io/kubernetes/pkg/controller/util/protectionutil"
 	"k8s.io/kubernetes/pkg/controller/volume/common"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/util/slice"
 	volumeutil "k8s.io/kubernetes/pkg/volume/util"
 )
@@ -115,6 +117,10 @@ type Controller struct {
 	pvcProcessingStore *pvcProcessingStore
 }

+var unusedSinceNowFunc = metav1.Now
+
+type podUsageCheckFunc func(logger klog.Logger, pod *v1.Pod, pvc *v1.PersistentVolumeClaim) bool
+
 // NewPVCProtectionController returns a new instance of PVCProtectionController.
 func NewPVCProtectionController(logger klog.Logger, pvcInformer coreinformers.PersistentVolumeClaimInformer, podInformer coreinformers.PodInformer, cl clientset.Interface) (*Controller, error) {
 	e := &Controller{
@@ -254,10 +260,20 @@ func (c *Controller) processPVC(ctx context.Context, pvcNamespace, pvcName strin
 		return err
 	}

+	if utilfeature.DefaultFeatureGate.Enabled(features.PersistentVolumeClaimUnusedSinceTime) && pvc.DeletionTimestamp == nil {
+		isUsed, err := c.isBeingUsedWith(ctx, pvc, lazyLivePodList, c.podUsesPVCForUnusedSince)
+		if err != nil {
+			return err
+		}
+		if err := c.updateUnusedCondition(ctx, pvc, isUsed); err != nil {
+			return err
+		}
+	}
+
 	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.isBeingUsedWith(ctx, pvc, lazyLivePodList, c.podUsesPVCForDeletion)
 		if err != nil {
 			return err
 		}
@@ -303,12 +319,63 @@ func (c *Controller) removeFinalizer(ctx context.Context, pvc *v1.PersistentVolu
 	return nil
 }

-func (c *Controller) isBeingUsed(ctx context.Context, pvc *v1.PersistentVolumeClaim, lazyLivePodList *LazyLivePodList) (bool, error) {
+func (c *Controller) updateUnusedCondition(ctx context.Context, pvc *v1.PersistentVolumeClaim, isUsed bool) error {
+	existingCondition := findCondition(pvc.Status.Conditions, v1.PersistentVolumeClaimUnused)
+
+	switch {
+	case isUsed && (existingCondition == nil || existingCondition.Status == v1.ConditionTrue):
+		// PVC was unused but is now in use -> set Unused=False
+		return c.setUnusedCondition(ctx, pvc, v1.ConditionFalse, "PodUsingPVC", "A pod is currently referencing this PVC")
+	case !isUsed && (existingCondition == nil || existingCondition.Status != v1.ConditionTrue):
+		// PVC is not in use and condition doesn't reflect that -> set Unused=True
+		return c.setUnusedCondition(ctx, pvc, v1.ConditionTrue, "NoPodsUsingPVC", "No pods are currently referencing this PVC")
+	default:
+		return nil
+	}
+}
+
+func (c *Controller) setUnusedCondition(ctx context.Context, pvc *v1.PersistentVolumeClaim, status v1.ConditionStatus, reason, message string) error {
+	claimClone := pvc.DeepCopy()
+	now := unusedSinceNowFunc()
+	newCondition := v1.PersistentVolumeClaimCondition{
+		Type:               v1.PersistentVolumeClaimUnused,
+		Status:             status,
+		LastTransitionTime: now,
+		Reason:             reason,
+		Message:            message,
+	}
+
+	if existing := findCondition(claimClone.Status.Conditions, v1.PersistentVolumeClaimUnused); existing != nil {
+		*existing = newCondition
+	} else {
+		claimClone.Status.Conditions = append(claimClone.Status.Conditions, newCondition)
+	}
+
+	_, err := c.client.CoreV1().PersistentVolumeClaims(claimClone.Namespace).UpdateStatus(ctx, claimClone, metav1.UpdateOptions{})
+	logger := klog.FromContext(ctx)
+	if err != nil {
+		logger.Error(err, "Error updating Unused condition in PVC status", "PVC", klog.KObj(pvc))
+		return err
+	}
+	logger.V(3).Info("Updated Unused condition in PVC status", "PVC", klog.KObj(pvc), "status", status)
+	return nil
+}
+
+func findCondition(conditions []v1.PersistentVolumeClaimCondition, condType v1.PersistentVolumeClaimConditionType) *v1.PersistentVolumeClaimCondition {
+	for i := range conditions {
+		if conditions[i].Type == condType {
+			return &conditions[i]
+		}
+	}
+	return nil
+}
+
+func (c *Controller) isBeingUsedWith(ctx context.Context, pvc *v1.PersistentVolumeClaim, lazyLivePodList *LazyLivePodList, podUsage podUsageCheckFunc) (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.
 	logger := klog.FromContext(ctx)
-	if inUse, err := c.askInformer(logger, pvc); err != nil {
+	if inUse, err := c.askInformer(logger, pvc, podUsage); err != nil {
 		// No need to return because a live list will follow.
 		logger.Error(err, "")
 	} else if inUse {
@@ -323,10 +390,10 @@ func (c *Controller) isBeingUsed(ctx context.Context, pvc *v1.PersistentVolumeCl
 	// 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)
+	return c.askAPIServer(ctx, pvc, lazyLivePodList, podUsage)
 }

-func (c *Controller) askInformer(logger klog.Logger, pvc *v1.PersistentVolumeClaim) (bool, error) {
+func (c *Controller) askInformer(logger klog.Logger, pvc *v1.PersistentVolumeClaim, podUsage podUsageCheckFunc) (bool, error) {
 	logger.V(4).Info("Looking for Pods using PVC in the Informer's cache", "PVC", klog.KObj(pvc))

 	// The indexer is used to find pods which might use the PVC.
@@ -343,7 +410,7 @@ func (c *Controller) askInformer(logger klog.Logger, pvc *v1.PersistentVolumeCla
 		// We still need to look at each volume: that's redundant for volume.PersistentVolumeClaim,
 		// but for volume.Ephemeral we need to be sure that this particular PVC is the one
 		// created for the ephemeral volume.
-		if c.podUsesPVC(logger, pod, pvc) {
+		if podUsage(logger, pod, pvc) {
 			return true, nil
 		}
 	}
@@ -352,7 +419,7 @@ 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, lazyLivePodList *LazyLivePodList, podUsage podUsageCheckFunc) (bool, error) {
 	logger := klog.FromContext(ctx)
 	logger.V(4).Info("Looking for Pods using PVC", "PVC", klog.KObj(pvc))
 	if lazyLivePodList.getCache() == nil {
@@ -370,7 +437,7 @@ func (c *Controller) askAPIServer(ctx context.Context, pvc *v1.PersistentVolumeC
 	}

 	for _, pod := range lazyLivePodList.getCache() {
-		if c.podUsesPVC(logger, &pod, pvc) {
+		if podUsage(logger, &pod, pvc) {
 			return true, nil
 		}
 	}
@@ -379,7 +446,7 @@ func (c *Controller) askAPIServer(ctx context.Context, pvc *v1.PersistentVolumeC
 	return false, nil
 }

-func (c *Controller) podUsesPVC(logger klog.Logger, pod *v1.Pod, pvc *v1.PersistentVolumeClaim) bool {
+func (c *Controller) podUsesPVCForDeletion(logger klog.Logger, pod *v1.Pod, pvc *v1.PersistentVolumeClaim) bool {
 	// Check whether pvc is used by pod only if pod is scheduled, because
 	// kubelet sees pods after they have been scheduled and it won't allow
 	// starting a pod referencing a PVC with a non-nil deletionTimestamp.
@@ -395,6 +462,21 @@ func (c *Controller) podUsesPVC(logger klog.Logger, pod *v1.Pod, pvc *v1.Persist
 	return false
 }

+func (c *Controller) podUsesPVCForUnusedSince(logger klog.Logger, pod *v1.Pod, pvc *v1.PersistentVolumeClaim) bool {
+	if volumeutil.IsPodTerminated(pod, pod.Status) {
+		return false
+	}
+
+	for _, volume := range pod.Spec.Volumes {
+		if volume.PersistentVolumeClaim != nil && volume.PersistentVolumeClaim.ClaimName == pvc.Name ||
+			volume.Ephemeral != nil && ephemeral.VolumeClaimName(pod, &volume) == pvc.Name && ephemeral.VolumeIsForPod(pod, pvc) == nil {
+			logger.V(4).Info("Pod references PVC for unused tracking", "pod", klog.KObj(pod), "PVC", klog.KObj(pvc))
+			return true
+		}
+	}
+	return false
+}
+
 // podIsShutDown returns true if kubelet is done with the pod or
 // it was force-deleted.
 func podIsShutDown(pod *v1.Pod) bool {
@@ -439,6 +521,10 @@ func (c *Controller) pvcAddedUpdated(logger klog.Logger, obj interface{}) {
 	}
 	logger.V(4).Info("Got event on PVC", "pvc", klog.KObj(pvc))

+	if utilfeature.DefaultFeatureGate.Enabled(features.PersistentVolumeClaimUnusedSinceTime) && findCondition(pvc.Status.Conditions, v1.PersistentVolumeClaimUnused) == nil {
+		c.queue.Add(key)
+	}
+
 	if protectionutil.NeedToAddFinalizer(pvc, volumeutil.PVCProtectionFinalizer) || protectionutil.IsDeletionCandidate(pvc, volumeutil.PVCProtectionFinalizer) {
 		c.queue.Add(key)
 	}
@@ -482,7 +568,8 @@ func (*Controller) parsePod(obj interface{}) *v1.Pod {

 func (c *Controller) enqueuePVCs(logger klog.Logger, pod *v1.Pod, deleted bool) {
 	// Filter out pods that can't help us to remove a finalizer on PVC
-	if !deleted && !volumeutil.IsPodTerminated(pod, pod.Status) && pod.Spec.NodeName != "" {
+	if !utilfeature.DefaultFeatureGate.Enabled(features.PersistentVolumeClaimUnusedSinceTime) &&
+		!deleted && !volumeutil.IsPodTerminated(pod, pod.Status) && pod.Spec.NodeName != "" {
 		return
 	}

--- a/pkg/controller/volume/pvcprotection/pvc_protection_controller_test.go
+++ b/pkg/controller/volume/pvcprotection/pvc_protection_controller_test.go
@@ -30,11 +30,15 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/version"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes/fake"
 	clienttesting "k8s.io/client-go/testing"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	"k8s.io/klog/v2/ktesting"
 	"k8s.io/kubernetes/pkg/controller"
+	"k8s.io/kubernetes/pkg/features"
 	volumeutil "k8s.io/kubernetes/pkg/volume/util"
 	"k8s.io/utils/dump"
 )
@@ -148,6 +152,21 @@ func withProtectionFinalizer(pvc *v1.PersistentVolumeClaim) *v1.PersistentVolume
 	return pvc
 }

+func withUnusedCondition(status v1.ConditionStatus, ts metav1.Time, pvc *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
+	pvc.Status.Conditions = append(pvc.Status.Conditions, v1.PersistentVolumeClaimCondition{
+		Type:               v1.PersistentVolumeClaimUnused,
+		Status:             status,
+		LastTransitionTime: ts,
+		Reason:             "NoPodsUsingPVC",
+		Message:            "No pods are currently referencing this PVC",
+	})
+	if status == v1.ConditionFalse {
+		pvc.Status.Conditions[len(pvc.Status.Conditions)-1].Reason = "PodUsingPVC"
+		pvc.Status.Conditions[len(pvc.Status.Conditions)-1].Message = "A pod is currently referencing this PVC"
+	}
+	return pvc
+}
+
 func deleted(pvc *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
 	pvc.DeletionTimestamp = &metav1.Time{}
 	return pvc
@@ -207,7 +226,8 @@ func TestPVCProtectionController(t *testing.T) {
 	}

 	tests := []struct {
-		name string
+		name                     string
+		enablePVCUnusedSinceTime bool
 		// Object to insert into fake kubeclient before the test starts.
 		initialObjects []runtime.Object
 		// Whether not to insert the content of initialObjects into the
@@ -402,6 +422,178 @@ func TestPVCProtectionController(t *testing.T) {
 			},
 		},
 		//
+		// PVC UnusedSince condition — when feature gate is enabled
+		//
+		{
+			name:                     "feature enabled: unused PVC with no pods -> Unused=True condition set",
+			enablePVCUnusedSinceTime: true,
+			updatedPVCs:              []*v1.PersistentVolumeClaim{withProtectionFinalizer(pvc())},
+			expectedActions: []clienttesting.Action{
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateSubresourceAction(pvcGVR, "status", defaultNS, withUnusedCondition(v1.ConditionTrue, metav1.Unix(123, 0), withProtectionFinalizer(pvc()))),
+			},
+		},
+		{
+			name:                     "feature enabled: PVC already has Unused=True and is still unused -> no update",
+			enablePVCUnusedSinceTime: true,
+			updatedPVCs:              []*v1.PersistentVolumeClaim{withUnusedCondition(v1.ConditionTrue, metav1.Unix(100, 0), withProtectionFinalizer(pvc()))},
+			expectedActions: []clienttesting.Action{
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+			},
+		},
+		{
+			name:                     "feature enabled: running pod references PVC with Unused=True -> condition set to False",
+			enablePVCUnusedSinceTime: true,
+			initialObjects: []runtime.Object{
+				withUnusedCondition(v1.ConditionTrue, metav1.Unix(100, 0), withProtectionFinalizer(pvc())),
+			},
+			updatedPod: withStatus(v1.PodRunning, withPVC(defaultPVCName, pod())),
+			expectedActions: []clienttesting.Action{
+				clienttesting.NewUpdateSubresourceAction(pvcGVR, "status", defaultNS, withUnusedCondition(v1.ConditionFalse, metav1.Unix(123, 0), withProtectionFinalizer(pvc()))),
+			},
+		},
+		{
+			name:                     "feature enabled: PVC in use by running pod without condition -> no update",
+			enablePVCUnusedSinceTime: true,
+			initialObjects: []runtime.Object{
+				withStatus(v1.PodRunning, withPVC(defaultPVCName, pod())),
+			},
+			updatedPVCs:     []*v1.PersistentVolumeClaim{withProtectionFinalizer(pvc())},
+			expectedActions: []clienttesting.Action{},
+		},
+		{
+			name:                     "feature enabled: deleting PVC -> condition not updated, finalizer removed",
+			enablePVCUnusedSinceTime: true,
+			updatedPVCs:              []*v1.PersistentVolumeClaim{deleted(withProtectionFinalizer(pvc()))},
+			expectedActions: []clienttesting.Action{
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateAction(pvcGVR, defaultNS, deleted(pvc())),
+			},
+		},
+		{
+			name:            "feature disabled: unused PVC with no pods -> no condition action",
+			updatedPVCs:     []*v1.PersistentVolumeClaim{withProtectionFinalizer(pvc())},
+			expectedActions: []clienttesting.Action{},
+		},
+		{
+			name:                     "feature enabled: unscheduled pending pod references PVC -> PVC considered in use (no condition set)",
+			enablePVCUnusedSinceTime: true,
+			initialObjects: []runtime.Object{
+				unscheduled(withPVC(defaultPVCName, pod())),
+			},
+			updatedPVCs:     []*v1.PersistentVolumeClaim{withProtectionFinalizer(pvc())},
+			expectedActions: []clienttesting.Action{},
+		},
+		{
+			name:                     "feature enabled: terminated (Succeeded) pod does not count as using PVC -> Unused=True set",
+			enablePVCUnusedSinceTime: true,
+			initialObjects: []runtime.Object{
+				withStatus(v1.PodSucceeded, withPVC(defaultPVCName, pod())),
+			},
+			updatedPVCs: []*v1.PersistentVolumeClaim{withProtectionFinalizer(pvc())},
+			expectedActions: []clienttesting.Action{
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateSubresourceAction(pvcGVR, "status", defaultNS, withUnusedCondition(v1.ConditionTrue, metav1.Unix(123, 0), withProtectionFinalizer(pvc()))),
+			},
+		},
+		{
+			name:                     "feature enabled: terminated (Failed) pod does not count as using PVC -> Unused=True set",
+			enablePVCUnusedSinceTime: true,
+			initialObjects: []runtime.Object{
+				withStatus(v1.PodFailed, withPVC(defaultPVCName, pod())),
+			},
+			updatedPVCs: []*v1.PersistentVolumeClaim{withProtectionFinalizer(pvc())},
+			expectedActions: []clienttesting.Action{
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateSubresourceAction(pvcGVR, "status", defaultNS, withUnusedCondition(v1.ConditionTrue, metav1.Unix(123, 0), withProtectionFinalizer(pvc()))),
+			},
+		},
+		{
+			name:                     "feature enabled: status update fails -> controller retries",
+			enablePVCUnusedSinceTime: true,
+			updatedPVCs:              []*v1.PersistentVolumeClaim{withProtectionFinalizer(pvc())},
+			reactors: []reaction{
+				{
+					verb:      "update",
+					resource:  "persistentvolumeclaims",
+					reactorfn: generateUpdateErrorFunc(t, 2),
+				},
+			},
+			expectedActions: []clienttesting.Action{
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateSubresourceAction(pvcGVR, "status", defaultNS, withUnusedCondition(v1.ConditionTrue, metav1.Unix(123, 0), withProtectionFinalizer(pvc()))),
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateSubresourceAction(pvcGVR, "status", defaultNS, withUnusedCondition(v1.ConditionTrue, metav1.Unix(123, 0), withProtectionFinalizer(pvc()))),
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateSubresourceAction(pvcGVR, "status", defaultNS, withUnusedCondition(v1.ConditionTrue, metav1.Unix(123, 0), withProtectionFinalizer(pvc()))),
+			},
+		},
+		{
+			name:                     "feature enabled: pod not in informer cache but exists in API -> PVC correctly identified as in use",
+			enablePVCUnusedSinceTime: true,
+			initialObjects: []runtime.Object{
+				withPVC(defaultPVCName, pod()),
+			},
+			informersAreLate: true,
+			updatedPVCs:      []*v1.PersistentVolumeClaim{withProtectionFinalizer(pvc())},
+			expectedActions: []clienttesting.Action{
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+			},
+		},
+		{
+			name:                     "feature enabled: one of two pods deleted, remaining pod keeps PVC in use",
+			enablePVCUnusedSinceTime: true,
+			initialObjects: []runtime.Object{
+				withPVC(defaultPVCName, pod()),
+				withProtectionFinalizer(pvc()),
+			},
+			deletedPod:      withPVC(defaultPVCName, withUID("uid2", podWithConfig("pod2", defaultNS))),
+			expectedActions: []clienttesting.Action{},
+		},
+		{
+			name:                     "feature enabled: deleting PVC that already has Unused=True condition -> condition preserved when removing finalizer",
+			enablePVCUnusedSinceTime: true,
+			updatedPVCs: []*v1.PersistentVolumeClaim{
+				deleted(withUnusedCondition(v1.ConditionTrue, metav1.Unix(100, 0), withProtectionFinalizer(pvc()))),
+			},
+			expectedActions: []clienttesting.Action{
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateAction(pvcGVR, defaultNS, deleted(withUnusedCondition(v1.ConditionTrue, metav1.Unix(100, 0), pvc()))),
+			},
+		},
+		{
+			name:                     "feature enabled: pod deleted, PVC becomes unused -> Unused=True condition set",
+			enablePVCUnusedSinceTime: true,
+			initialObjects: []runtime.Object{
+				withProtectionFinalizer(pvc()),
+			},
+			deletedPod: withPVC(defaultPVCName, pod()),
+			expectedActions: []clienttesting.Action{
+				clienttesting.NewListAction(podGVR, podGVK, defaultNS, metav1.ListOptions{}),
+				clienttesting.NewUpdateSubresourceAction(pvcGVR, "status", defaultNS, withUnusedCondition(v1.ConditionTrue, metav1.Unix(123, 0), withProtectionFinalizer(pvc()))),
+			},
+		},
+		{
+			name:                     "feature enabled: used PVC without finalizer -> finalizer is added (existing behavior)",
+			enablePVCUnusedSinceTime: true,
+			initialObjects: []runtime.Object{
+				withPVC(defaultPVCName, pod()),
+			},
+			updatedPVCs: []*v1.PersistentVolumeClaim{pvc()},
+			expectedActions: []clienttesting.Action{
+				clienttesting.NewUpdateAction(pvcGVR, defaultNS, withProtectionFinalizer(pvc())),
+			},
+		},
+		{
+			name:                     "feature enabled: deleted PVC with finalizer and pod using it -> finalizer kept (existing behavior)",
+			enablePVCUnusedSinceTime: true,
+			initialObjects: []runtime.Object{
+				withPVC(defaultPVCName, pod()),
+			},
+			updatedPVCs:     []*v1.PersistentVolumeClaim{deleted(withProtectionFinalizer(pvc()))},
+			expectedActions: []clienttesting.Action{},
+		},
+		//
 		// Pod events
 		//
 		{
@@ -484,6 +676,15 @@ func TestPVCProtectionController(t *testing.T) {
 	}

 	for _, test := range tests {
+		featuregatetesting.SetFeatureGateEmulationVersionDuringTest(t, utilfeature.DefaultFeatureGate, version.MustParse("1.36"))
+		featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.PersistentVolumeClaimUnusedSinceTime, test.enablePVCUnusedSinceTime)
+
+		origNowFunc := unusedSinceNowFunc
+		unusedSinceNowFunc = func() metav1.Time { return metav1.Unix(123, 0) }
+		t.Cleanup(func() {
+			unusedSinceNowFunc = origNowFunc
+		})
+
 		// Create initial data for client and informers.
 		var (
 			clientObjs    []runtime.Object
--- a/pkg/features/kube_features.go
+++ b/pkg/features/kube_features.go
@@ -681,6 +681,12 @@ const (
 	// Enables relisting individual pods on-demand.
 	PLEGOnDemandRelist featuregate.Feature = "PLEGOnDemandRelist"

+	// owner: @ArvindParekh
+	// kep: https://kep.k8s.io/5541
+	//
+	// Adds an Unused condition to PersistentVolumeClaim status that indicates when the PVC was last used by a pod.
+	PersistentVolumeClaimUnusedSinceTime featuregate.Feature = "PersistentVolumeClaimUnusedSinceTime"
+
 	// owner: @haircommander
 	// kep: https://kep.k8s.io/2364
 	//
@@ -1655,6 +1661,10 @@ var defaultVersionedKubernetesFeatureGates = map[featuregate.Feature]featuregate
 		{Version: version.MustParse("1.36"), Default: true, PreRelease: featuregate.Beta},
 	},

+	PersistentVolumeClaimUnusedSinceTime: {
+		{Version: version.MustParse("1.36"), Default: false, PreRelease: featuregate.Alpha},
+	},
+
 	PodAndContainerStatsFromCRI: {
 		{Version: version.MustParse("1.23"), Default: false, PreRelease: featuregate.Alpha},
 	},
@@ -2462,6 +2472,8 @@ var defaultKubernetesFeatureGateDependencies = map[featuregate.Feature][]feature

 	PLEGOnDemandRelist: {},

+	PersistentVolumeClaimUnusedSinceTime: {},
+
 	PodAndContainerStatsFromCRI: {},

 	PodCertificateRequest: {AuthorizeNodeWithSelectors},
--- a/plugin/pkg/auth/authorizer/rbac/bootstrappolicy/controller_policy.go
+++ b/plugin/pkg/auth/authorizer/rbac/bootstrappolicy/controller_policy.go
@@ -443,6 +443,7 @@ func buildControllerRoles() ([]rbacv1.ClusterRole, []rbacv1.ClusterRoleBinding)
 		ObjectMeta: metav1.ObjectMeta{Name: saRolePrefix + "pvc-protection-controller"},
 		Rules: []rbacv1.PolicyRule{
 			rbacv1helpers.NewRule("get", "list", "watch", "update").Groups(legacyGroup).Resources("persistentvolumeclaims").RuleOrDie(),
+			rbacv1helpers.NewRule("update").Groups(legacyGroup).Resources("persistentvolumeclaims/status").RuleOrDie(),
 			rbacv1helpers.NewRule("list", "watch", "get").Groups(legacyGroup).Resources("pods").RuleOrDie(),
 			eventsRule(),
 		},
--- a/plugin/pkg/auth/authorizer/rbac/bootstrappolicy/testdata/controller-roles.yaml
+++ b/plugin/pkg/auth/authorizer/rbac/bootstrappolicy/testdata/controller-roles.yaml
@@ -1128,6 +1128,12 @@ items:
    - list
    - update
    - watch
+  - apiGroups:
+    - ""
+    resources:
+    - persistentvolumeclaims/status
+    verbs:
+    - update
  - apiGroups:
    - ""
    resources:
--- a/staging/src/k8s.io/api/core/v1/types.go
+++ b/staging/src/k8s.io/api/core/v1/types.go
@@ -669,6 +669,11 @@ const (
 	PersistentVolumeClaimVolumeModifyVolumeError PersistentVolumeClaimConditionType = "ModifyVolumeError"
 	// Volume is being modified
 	PersistentVolumeClaimVolumeModifyingVolume PersistentVolumeClaimConditionType = "ModifyingVolume"
+
+	// PersistentVolumeClaimUnused indicates whether the PVC is currently not in use by any Pod.
+	// When status is True, the PVC is not referenced by any non-terminal Pod.
+	// The lastTransitionTime indicates when the PVC last transitioned to being unused.
+	PersistentVolumeClaimUnused PersistentVolumeClaimConditionType = "Unused"
 )

 // +enum
--- a/test/compatibility_lifecycle/reference/feature_list.md
+++ b/test/compatibility_lifecycle/reference/feature_list.md
@@ -137,6 +137,7 @@
 | OpportunisticBatching | :ballot_box_with_check:&nbsp;1.35+ |  |  | 1.35– |  |  |  | [code](https://cs.k8s.io/?q=%5CbOpportunisticBatching%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbOpportunisticBatching%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
 | OrderedNamespaceDeletion | :ballot_box_with_check:&nbsp;1.33+ | :closed_lock_with_key:&nbsp;1.34+ |  | 1.30–1.33 | 1.34– |  |  | [code](https://cs.k8s.io/?q=%5CbOrderedNamespaceDeletion%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbOrderedNamespaceDeletion%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
 | PLEGOnDemandRelist | :ballot_box_with_check:&nbsp;1.36+ |  |  | 1.36– |  |  |  | [code](https://cs.k8s.io/?q=%5CbPLEGOnDemandRelist%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbPLEGOnDemandRelist%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
+| PersistentVolumeClaimUnusedSinceTime |  |  | 1.36– |  |  |  |  | [code](https://cs.k8s.io/?q=%5CbPersistentVolumeClaimUnusedSinceTime%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbPersistentVolumeClaimUnusedSinceTime%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
 | PodAndContainerStatsFromCRI |  |  | 1.23– |  |  |  |  | [code](https://cs.k8s.io/?q=%5CbPodAndContainerStatsFromCRI%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbPodAndContainerStatsFromCRI%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
 | PodCertificateRequest |  |  | 1.34 | 1.35– |  |  | AuthorizeNodeWithSelectors | [code](https://cs.k8s.io/?q=%5CbPodCertificateRequest%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbPodCertificateRequest%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
 | PodDeletionCost | :ballot_box_with_check:&nbsp;1.22+ |  | 1.21 | 1.22– |  |  |  | [code](https://cs.k8s.io/?q=%5CbPodDeletionCost%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbPodDeletionCost%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
--- a/test/compatibility_lifecycle/reference/versioned_feature_list.yaml
+++ b/test/compatibility_lifecycle/reference/versioned_feature_list.yaml
@@ -1335,6 +1335,12 @@
    lockToDefault: true
    preRelease: GA
    version: "1.34"
+- name: PersistentVolumeClaimUnusedSinceTime
+  versionedSpecs:
+  - default: false
+    lockToDefault: false
+    preRelease: Alpha
+    version: "1.36"
 - name: PLEGOnDemandRelist
  versionedSpecs:
  - default: true