Repository: kubernetes-csi/external-provisioner
Pull request #422 (2020-03-11 10:43:41)

Added controller for PVC deletion
/kind feature


**What this PR does / why we need it**:

This PR addresses an attempt to split PVC handling between `external-provisioner` and `sig-storage-lib-external-provisioner`, also introducing a way to handle finalizers, related to in-progress PVC cloning.

PVC referenced in a data source is now updated with finalizer `provisioner.storage.kubernetes.io/cloning-protectionr` at the moment when a `ProvisionExt` method is called, used to dynamically provision a PV to which the content will be cloned at.

When cloning is finished - all PVC referencing the one as
a data source goes from `Pending` to `Bound` state, which allows the finalizer to be removed before deletion.
 
**Which issue(s) this PR fixes**:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes https://github.com/kubernetes-csi/external-provisioner/issues/414

**Special notes for your reviewer**:

**Does this PR introduce a user-facing change?**:

```release-note
finalizers:
- provisioner.storage.kubernetes.io/cloning-protection

is now set on PVCs which are the source ones for the cloning,
preventing their removal before the cloning finishes.
```



Patch:
diff --git a/cmd/csi-provisioner/csi-provisioner.go b/cmd/csi-provisioner/csi-provisioner.go
index 669a0a700..816b54263 100644
--- a/cmd/csi-provisioner/csi-provisioner.go
+++ b/cmd/csi-provisioner/csi-provisioner.go
@@ -172,13 +172,36 @@ func main() {
 	timeStamp := time.Now().UnixNano() / int64(time.Millisecond)
 	identity := strconv.FormatInt(timeStamp, 10) + "-" + strconv.Itoa(rand.Intn(10000)) + "-" + provisionerName
 
+	factory := informers.NewSharedInformerFactory(clientset, ctrl.ResyncPeriodOfCsiNodeInformer)
+
+	// -------------------------------
+	// Listers
+	// Create informer to prevent hit the API server for all resource request
+	scLister := factory.Storage().V1().StorageClasses().Lister()
+	claimLister := factory.Core().V1().PersistentVolumeClaims().Lister()
+
+	var csiNodeLister storagelistersv1beta1.CSINodeLister
+	var nodeLister v1.NodeLister
+	if ctrl.SupportsTopology(pluginCapabilities) {
+		csiNodeLister = factory.Storage().V1beta1().CSINodes().Lister()
+		nodeLister = factory.Core().V1().Nodes().Lister()
+	}
+
+	// -------------------------------
+	// PersistentVolumeClaims informer
+	rateLimiter := workqueue.NewItemExponentialFailureRateLimiter(*retryIntervalStart, *retryIntervalMax)
+	claimQueue := workqueue.NewNamedRateLimitingQueue(rateLimiter, "claims")
+	claimInformer := factory.Core().V1().PersistentVolumeClaims().Informer()
+
+	// Setup options
 	provisionerOptions := []func(*controller.ProvisionController) error{
 		controller.LeaderElection(false), // Always disable leader election in provisioner lib. Leader election should be done here in the CSI provisioner level instead.
 		controller.FailedProvisionThreshold(0),
 		controller.FailedDeleteThreshold(0),
-		controller.RateLimiter(workqueue.NewItemExponentialFailureRateLimiter(*retryIntervalStart, *retryIntervalMax)),
+		controller.RateLimiter(rateLimiter),
 		controller.Threadiness(int(*workerThreads)),
 		controller.CreateProvisionedPVLimiter(workqueue.DefaultControllerRateLimiter()),
+		controller.ClaimsInformer(claimInformer),
 	}
 
 	translator := csitrans.New()
@@ -193,17 +216,6 @@ func main() {
 		provisionerOptions = append(provisionerOptions, controller.AdditionalProvisionerNames([]string{supportsMigrationFromInTreePluginName}))
 	}
 
-	// Create informer to prevent hit the API server for all resource request
-	factory := informers.NewSharedInformerFactory(clientset, ctrl.ResyncPeriodOfCsiNodeInformer)
-	scLister := factory.Storage().V1().StorageClasses().Lister()
-
-	var csiNodeLister storagelistersv1beta1.CSINodeLister
-	var nodeLister v1.NodeLister
-	if ctrl.SupportsTopology(pluginCapabilities) {
-		csiNodeLister = factory.Storage().V1beta1().CSINodes().Lister()
-		nodeLister = factory.Core().V1().Nodes().Lister()
-	}
-
 	// Create the provisioner: it implements the Provisioner interface expected by
 	// the controller
 	csiProvisioner := ctrl.NewCSIProvisioner(
@@ -223,6 +235,7 @@ func main() {
 		scLister,
 		csiNodeLister,
 		nodeLister,
+		claimLister,
 		*extraCreateMetadata,
 	)
 
@@ -234,6 +247,13 @@ func main() {
 		provisionerOptions...,
 	)
 
+	csiClaimController := ctrl.NewCloningProtectionController(
+		clientset,
+		claimLister,
+		claimInformer,
+		claimQueue,
+	)
+
 	run := func(context.Context) {
 		stopCh := context.Background().Done()
 		factory.Start(stopCh)
@@ -244,6 +264,7 @@ func main() {
 			}
 		}
 
+		go csiClaimController.Run(int(*workerThreads), stopCh)
 		provisionController.Run(wait.NeverStop)
 	}
 
diff --git a/pkg/controller/clone_controller.go b/pkg/controller/clone_controller.go
new file mode 100644
index 000000000..f947495f9
--- /dev/null
+++ b/pkg/controller/clone_controller.go
@@ -0,0 +1,206 @@
+package controller
+
+import (
+	"fmt"
+	"time"
+
+	v1 "k8s.io/api/core/v1"
+	apierrs "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/labels"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/kubernetes"
+	corelisters "k8s.io/client-go/listers/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog"
+	"sigs.k8s.io/sig-storage-lib-external-provisioner/controller"
+)
+
+//
+// This package introduce a way to handle finalizers, related to in-progress PVC cloning. This is a two step approach:
+//
+// 1) PVC referenced as a data source is now updated with a finalizer `provisioner.storage.kubernetes.io/cloning-protection` during a ProvisionExt method call.
+// The detection of cloning in-progress is based on the assumption that a PVC with `spec.DataSource` pointing on a another PVC will go into `Pending` state.
+// The downside of this, is that fact that any other reason causing PVC to stay in the `Pending` state also blocks resource from deletion it from deletion
+//
+// 2) When cloning is finished for each PVC referencing the one as a data source,
+// this PVC will go from `Pending` to `Bound` state. That allows remove the finalizer.
+//
+
+// CloningProtectionController is storing all related interfaces
+// to handle cloning protection finalizer removal after CSI cloning is finished
+type CloningProtectionController struct {
+	client        kubernetes.Interface
+	claimLister   corelisters.PersistentVolumeClaimLister
+	claimInformer cache.SharedInformer
+	claimQueue    workqueue.RateLimitingInterface
+}
+
+// NewCloningProtectionController creates new controller for additional CSI claim protection capabilities
+func NewCloningProtectionController(
+	client kubernetes.Interface,
+	claimLister corelisters.PersistentVolumeClaimLister,
+	claimInformer cache.SharedInformer,
+	claimQueue workqueue.RateLimitingInterface,
+) *CloningProtectionController {
+	controller := &CloningProtectionController{
+		client:        client,
+		claimLister:   claimLister,
+		claimInformer: claimInformer,
+		claimQueue:    claimQueue,
+	}
+	return controller
+}
+
+// Run is a main CloningProtectionController handler
+func (p *CloningProtectionController) Run(threadiness int, stopCh <-chan struct{}) {
+	klog.Info("Starting CloningProtection controller")
+	defer utilruntime.HandleCrash()
+	defer p.claimQueue.ShutDown()
+
+	claimHandler := cache.ResourceEventHandlerFuncs{
+		AddFunc:    p.enqueueClaimUpadate,
+		UpdateFunc: func(_ interface{}, newObj interface{}) { p.enqueueClaimUpadate(newObj) },
+	}
+	p.claimInformer.AddEventHandlerWithResyncPeriod(claimHandler, controller.DefaultResyncPeriod)
+
+	for i := 0; i < threadiness; i++ {
+		go wait.Until(p.runClaimWorker, time.Second, stopCh)
+	}
+
+	go p.claimInformer.Run(stopCh)
+
+	klog.Infof("Started CloningProtection controller")
+	<-stopCh
+	klog.Info("Shutting down CloningProtection controller")
+}
+
+func (p *CloningProtectionController) runClaimWorker() {
+	for p.processNextClaimWorkItem() {
+	}
+}
+
+// processNextClaimWorkItem processes items from claimQueue
+func (p *CloningProtectionController) processNextClaimWorkItem() bool {
+	obj, shutdown := p.claimQueue.Get()
+	if shutdown {
+		return false
+	}
+
+	err := func(obj interface{}) error {
+		defer p.claimQueue.Done(obj)
+		var key string
+		var ok bool
+		if key, ok = obj.(string); !ok {
+			p.claimQueue.Forget(obj)
+			return fmt.Errorf("expected string in workqueue but got %#v", obj)
+		}
+
+		if err := p.syncClaimHandler(key); err != nil {
+			klog.Warningf("Retrying syncing claim %q after %v failures", key, p.claimQueue.NumRequeues(obj))
+			p.claimQueue.AddRateLimited(obj)
+		} else {
+			p.claimQueue.Forget(obj)
+		}
+
+		return nil
+	}(obj)
+
+	if err != nil {
+		utilruntime.HandleError(err)
+		return true
+	}
+
+	return true
+}
+
+// enqueueClaimUpadate takes a PVC obj and stores it into the claim work queue.
+func (p *CloningProtectionController) enqueueClaimUpadate(obj interface{}) {
+	new, ok := obj.(*v1.PersistentVolumeClaim)
+	if !ok {
+		utilruntime.HandleError(fmt.Errorf("expected claim but got %+v", new))
+		return
+	}
+
+	// Timestamp didn't appear
+	if new.DeletionTimestamp == nil {
+		return
+	}
+
+	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
+	if err != nil {
+		utilruntime.HandleError(err)
+		return
+	}
+
+	p.claimQueue.Add(key)
+}
+
+// syncClaimHandler gets the claim from informer's cache then calls syncClaim
+func (p *CloningProtectionController) syncClaimHandler(key string) error {
+	namespace, name, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
+		return nil
+	}
+
+	claim, err := p.claimLister.PersistentVolumeClaims(namespace).Get(name)
+	if err != nil {
+		if apierrs.IsNotFound(err) {
+			utilruntime.HandleError(fmt.Errorf("Item '%s' in work queue no longer exists", key))
+			return nil
+		}
+
+		return err
+	}
+
+	return p.syncClaim(claim)
+}
+
+// syncClaim removes finalizers from a PVC, when cloning is finished
+func (p *CloningProtectionController) syncClaim(claim *v1.PersistentVolumeClaim) error {
+	if !checkFinalizer(claim, pvcCloneFinalizer) {
+		return nil
+	}
+
+	// Checking for PVCs in the same namespace to have other states aside from Pending, which means that cloning is still in progress
+	pvcList, err := p.claimLister.PersistentVolumeClaims(claim.Namespace).List(labels.Everything())
+	if err != nil {
+		return err
+	}
+
+	// Check for pvc state with DataSource pointing to claim
+	for _, pvc := range pvcList {
+		if pvc.Spec.DataSource == nil {
+			continue
+		}
+
+		// Requeue when at least one PVC is still works on cloning
+		if pvc.Spec.DataSource.Kind == pvcKind &&
+			pvc.Spec.DataSource.Name == claim.Name &&
+			pvc.Status.Phase == v1.ClaimPending {
+			return fmt.Errorf("PVC '%s' is in 'Pending' state, cloning in progress", pvc.Name)
+		}
+	}
+
+	// Remove clone finalizer
+	finalizers := make([]string, 0)
+	for _, finalizer := range claim.ObjectMeta.Finalizers {
+		if finalizer != pvcCloneFinalizer {
+			finalizers = append(finalizers, finalizer)
+		}
+	}
+	claim.ObjectMeta.Finalizers = finalizers
+
+	if _, err = p.client.CoreV1().PersistentVolumeClaims(claim.Namespace).Update(claim); err != nil {
+		if !apierrs.IsNotFound(err) {
+			// Couldn't remove finalizer and the object still exists, the controller may
+			// try to remove the finalizer again on the next update
+			klog.Infof("failed to remove clone finalizer from PVC %v", claim.Name)
+			return err
+		}
+	}
+
+	return nil
+}
diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
index c6d038506..54c1a8a0e 100644
--- a/pkg/controller/controller.go
+++ b/pkg/controller/controller.go
@@ -134,6 +134,8 @@ const (
 	annStorageProvisioner = "volume.beta.kubernetes.io/storage-provisioner"
 
 	snapshotNotBound = "snapshot %s not bound"
+
+	pvcCloneFinalizer = "provisioner.storage.kubernetes.io/cloning-protection"
 )
 
 var (
@@ -220,6 +222,7 @@ type csiProvisioner struct {
 	scLister                              storagelistersv1.StorageClassLister
 	csiNodeLister                         storagelistersv1beta1.CSINodeLister
 	nodeLister                            corelisters.NodeLister
+	claimLister                           corelisters.PersistentVolumeClaimLister
 	extraCreateMetadata                   bool
 }
 
@@ -284,6 +287,7 @@ func NewCSIProvisioner(client kubernetes.Interface,
 	scLister storagelistersv1.StorageClassLister,
 	csiNodeLister storagelistersv1beta1.CSINodeLister,
 	nodeLister corelisters.NodeLister,
+	claimLister corelisters.PersistentVolumeClaimLister,
 	extraCreateMetadata bool,
 ) controller.Provisioner {
 
@@ -306,6 +310,7 @@ func NewCSIProvisioner(client kubernetes.Interface,
 		scLister:                              scLister,
 		csiNodeLister:                         csiNodeLister,
 		nodeLister:                            nodeLister,
+		claimLister:                           claimLister,
 		extraCreateMetadata:                   extraCreateMetadata,
 	}
 	return provisioner
@@ -531,6 +536,13 @@ func (p *csiProvisioner) ProvisionExt(options controller.ProvisionOptions) (*v1.
 		req.VolumeContentSource = volumeContentSource
 	}
 
+	if options.PVC.Spec.DataSource != nil && rc.clone {
+		err = p.setCloneFinalizer(options.PVC)
+		if err != nil {
+			return nil, controller.ProvisioningNoChange, err
+		}
+	}
+
 	if p.supportsTopology() {
 		requirements, err := GenerateAccessibilityRequirements(
 			p.client,
@@ -711,6 +723,21 @@ func (p *csiProvisioner) ProvisionExt(options controller.ProvisionOptions) (*v1.
 	return pv, controller.ProvisioningFinished, nil
 }
 
+func (p *csiProvisioner) setCloneFinalizer(pvc *v1.PersistentVolumeClaim) error {
+	claim, err := p.claimLister.PersistentVolumeClaims(pvc.Namespace).Get(pvc.Spec.DataSource.Name)
+	if err != nil {
+		return err
+	}
+
+	if !checkFinalizer(claim, pvcCloneFinalizer) {
+		claim.Finalizers = append(claim.Finalizers, pvcCloneFinalizer)
+		_, err := p.client.CoreV1().PersistentVolumeClaims(claim.Namespace).Update(claim)
+		return err
+	}
+
+	return nil
+}
+
 func (p *csiProvisioner) supportsTopology() bool {
 	return SupportsTopology(p.pluginCapabilities)
 }
@@ -765,7 +792,7 @@ func (p *csiProvisioner) getVolumeContentSource(options controller.ProvisionOpti
 // getPVCSource verifies DataSource.Kind of type PersistentVolumeClaim, making sure that the requested PVC is available/ready
 // returns the VolumeContentSource for the requested PVC
 func (p *csiProvisioner) getPVCSource(options controller.ProvisionOptions) (*csi.VolumeContentSource, error) {
-	sourcePVC, err := p.client.CoreV1().PersistentVolumeClaims(options.PVC.Namespace).Get(options.PVC.Spec.DataSource.Name, metav1.GetOptions{})
+	sourcePVC, err := p.claimLister.PersistentVolumeClaims(options.PVC.Namespace).Get(options.PVC.Spec.DataSource.Name)
 	if err != nil {
 		return nil, fmt.Errorf("error getting PVC %s (namespace %q) from api server: %v", options.PVC.Spec.DataSource.Name, options.PVC.Namespace, err)
 	}
@@ -1257,3 +1284,12 @@ func cleanupVolume(p *csiProvisioner, delReq *csi.DeleteVolumeRequest, provision
 	}
 	return err
 }
+
+func checkFinalizer(obj metav1.Object, finalizer string) bool {
+	for _, f := range obj.GetFinalizers() {
+		if f == finalizer {
+			return true
+		}
+	}
+	return false
+}


Test patch:
diff --git a/pkg/controller/clone_controller_test.go b/pkg/controller/clone_controller_test.go
new file mode 100644
index 000000000..aa20b3e4c
--- /dev/null
+++ b/pkg/controller/clone_controller_test.go
@@ -0,0 +1,199 @@
+package controller
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/informers"
+	fakeclientset "k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/util/workqueue"
+)
+
+var requestedBytes int64 = 1000
+var fakeSc1 string = "fake-sc-1"
+
+const (
+	srcName      = "clone-source-pvc"
+	dstName      = "destination-pvc"
+	srcNamespace = "fake-pvc-namespace"
+	pvName       = "test-testi"
+)
+
+func pvcFinalizers(pvc *v1.PersistentVolumeClaim, finalizers ...string) *v1.PersistentVolumeClaim {
+	pvc.Finalizers = append(pvc.Finalizers, finalizers...)
+	return pvc
+}
+
+func pvcDataSourceClone(srcName string, pvc *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
+	apiGr := ""
+	pvc.Spec.DataSource = &v1.TypedLocalObjectReference{
+		APIGroup: &apiGr,
+		Kind:     pvcKind,
+		Name:     srcName,
+	}
+	return pvc
+}
+
+func pvcNamed(name string, pvc *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
+	pvc.Name = name
+	return pvc
+}
+
+func pvcNamespaced(namespace string, pvc *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
+	pvc.Namespace = namespace
+	return pvc
+}
+
+func pvcPhase(phase v1.PersistentVolumeClaimPhase, pvc *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
+	pvc.Status.Phase = phase
+	return pvc
+}
+
+func baseClaim() *v1.PersistentVolumeClaim {
+	return fakeClaim(srcName, srcNamespace, "fake-claim-uid", requestedBytes, pvName, v1.ClaimBound, &fakeSc1, "").DeepCopy()
+}
+
+func pvcDeletionMarked(pvc *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
+	timeX := metav1.NewTime(time.Now())
+	pvc.DeletionTimestamp = &timeX
+	return pvc
+}
+
+// TestCloneFinalizerRemoval tests create volume clone
+func TestCloneFinalizerRemoval(t *testing.T) {
+	testcases := map[string]struct {
+		initialClaims   []runtime.Object
+		cloneSource     runtime.Object
+		expectFinalizer bool
+		expectError     error
+	}{
+		"delete source pvc with no cloning in progress": {
+			cloneSource:   pvcFinalizers(baseClaim(), pvcCloneFinalizer),
+			initialClaims: []runtime.Object{pvcDataSourceClone(srcName, pvcNamed(dstName, baseClaim()))},
+		},
+		"delete source pvc when destination pvc status is claim pending": {
+			cloneSource:     pvcFinalizers(baseClaim(), pvcCloneFinalizer),
+			initialClaims:   []runtime.Object{pvcPhase(v1.ClaimPending, pvcDataSourceClone(srcName, pvcNamed(dstName, baseClaim())))},
+			expectFinalizer: true,
+			expectError:     fmt.Errorf("PVC '%s' is in 'Pending' state, cloning in progress", dstName),
+		},
+		"delete source pvc when at least one destination pvc status is claim pending": {
+			cloneSource: pvcFinalizers(baseClaim(), pvcCloneFinalizer),
+			initialClaims: []runtime.Object{
+				pvcDataSourceClone(srcName, pvcNamed(dstName, baseClaim())),
+				pvcPhase(v1.ClaimPending, pvcDataSourceClone(srcName, pvcNamed(dstName+"1", baseClaim()))),
+			},
+			expectFinalizer: true,
+			expectError:     fmt.Errorf("PVC '%s' is in 'Pending' state, cloning in progress", dstName+"1"),
+		},
+		"delete source pvc located in another namespace should not block": {
+			cloneSource:   pvcNamespaced(srcNamespace+"1", pvcFinalizers(baseClaim(), pvcCloneFinalizer)),
+			initialClaims: []runtime.Object{pvcPhase(v1.ClaimPending, pvcDataSourceClone(srcName, pvcNamed(dstName+"1", baseClaim())))},
+		},
+		"delete source pvc which is not cloned by any other pvc": {
+			cloneSource: pvcFinalizers(baseClaim(), pvcCloneFinalizer),
+		},
+		"delete source pvc without finalizer": {
+			cloneSource: baseClaim(),
+		},
+	}
+
+	for k, tc := range testcases {
+		tc := tc
+		t.Run(k, func(t *testing.T) {
+			t.Parallel()
+
+			objects := append(tc.initialClaims, tc.cloneSource)
+			clientSet := fakeclientset.NewSimpleClientset(objects...)
+			cloningProtector := fakeCloningProtector(clientSet, objects...)
+
+			// Simulate Delete behavior
+			claim := pvcDeletionMarked(tc.cloneSource.(*v1.PersistentVolumeClaim))
+			err := cloningProtector.syncClaim(claim)
+
+			// Get updated claim after reconcile finish
+			claim, _ = clientSet.CoreV1().PersistentVolumeClaims(claim.Namespace).Get(claim.Name, metav1.GetOptions{})
+
+			// Check finalizers removal
+			if tc.expectFinalizer && !checkFinalizer(claim, pvcCloneFinalizer) {
+				t.Errorf("Claim finalizer was expected to be found on: %s", claim.Name)
+			} else if !tc.expectFinalizer && checkFinalizer(claim, pvcCloneFinalizer) {
+				t.Errorf("Claim finalizer was not expected to be found on: %s", claim.Name)
+			}
+
+			if tc.expectError == nil && err != nil {
+				t.Errorf("Caught an unexpected error during 'syncClaim' run: %s", err)
+			} else if tc.expectError != nil && err == nil {
+				t.Errorf("Expected error during 'syncClaim' run, got nil: %s", tc.expectError)
+			} else if tc.expectError != nil && err != nil && tc.expectError.Error() != err.Error() {
+				t.Errorf("Unexpected error during 'syncClaim' run:\n\t%s\nExpected:\n\t%s", err, tc.expectError)
+			}
+		})
+	}
+
+}
+
+// TestEnqueueClaimUpadate ensure that PVCs will be processed for finalizer removal only on deletionTimestamp being set on the resource
+func TestEnqueueClaimUpadate(t *testing.T) {
+	testcases := map[string]struct {
+		claim    *v1.PersistentVolumeClaim
+		queueLen int
+	}{
+		"enqueue claim with deletionTimestamp": {
+			claim:    pvcDeletionMarked(baseClaim()),
+			queueLen: 1,
+		},
+		"enqueue claim without deletionTimestamp": {
+			claim:    baseClaim(),
+			queueLen: 0,
+		},
+	}
+
+	for k, tc := range testcases {
+		tc := tc
+		t.Run(k, func(t *testing.T) {
+			t.Parallel()
+
+			objects := []runtime.Object{}
+			clientSet := fakeclientset.NewSimpleClientset(objects...)
+			cloningProtector := fakeCloningProtector(clientSet, objects...)
+
+			// Simulate queue behavior
+			cloningProtector.enqueueClaimUpadate(tc.claim)
+
+			if cloningProtector.claimQueue.Len() != tc.queueLen {
+				t.Errorf("claimQueue should contain %d items, got: %d", tc.queueLen, cloningProtector.claimQueue.Len())
+			}
+		})
+	}
+}
+
+func fakeCloningProtector(client *fakeclientset.Clientset, objects ...runtime.Object) *CloningProtectionController {
+	utilruntime.ReallyCrash = false
+
+	informerFactory := informers.NewSharedInformerFactory(client, 1*time.Second)
+	claimInformer := informerFactory.Core().V1().PersistentVolumeClaims().Informer()
+	claimLister := informerFactory.Core().V1().PersistentVolumeClaims().Lister()
+	rateLimiter := workqueue.NewItemExponentialFailureRateLimiter(time.Second, 2*time.Second)
+	claimQueue := workqueue.NewNamedRateLimitingQueue(rateLimiter, "claims")
+
+	for _, claim := range objects {
+		claimInformer.GetStore().Add(claim)
+	}
+
+	informerFactory.WaitForCacheSync(context.TODO().Done())
+	go informerFactory.Start(context.TODO().Done())
+
+	return NewCloningProtectionController(
+		client,
+		claimLister,
+		claimInformer,
+		claimQueue,
+	)
+}
diff --git a/pkg/controller/controller_test.go b/pkg/controller/controller_test.go
index f045b4a73..6fcd9edc2 100644
--- a/pkg/controller/controller_test.go
+++ b/pkg/controller/controller_test.go
@@ -404,7 +404,7 @@ func TestCreateDriverReturnsInvalidCapacityDuringProvision(t *testing.T) {
 
 	pluginCaps, controllerCaps := provisionCapabilities()
 	csiProvisioner := NewCSIProvisioner(nil, 5*time.Second, "test-provisioner", "test",
-		5, csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, false)
+		5, csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, nil, false)
 
 	// Requested PVC with requestedBytes storage
 	deletePolicy := v1.PersistentVolumeReclaimDelete
@@ -1669,7 +1669,7 @@ func runProvisionTest(t *testing.T, k string, tc provisioningTestcase, requested
 
 	pluginCaps, controllerCaps := provisionCapabilities()
 	csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5, csiConn.conn,
-		nil, provisionDriverName, pluginCaps, controllerCaps, supportsMigrationFromInTreePluginName, false, csitrans.New(), nil, nil, nil, tc.withExtraMetadata)
+		nil, provisionDriverName, pluginCaps, controllerCaps, supportsMigrationFromInTreePluginName, false, csitrans.New(), nil, nil, nil, nil, tc.withExtraMetadata)
 
 	out := &csi.CreateVolumeResponse{
 		Volume: &csi.Volume{
@@ -2405,7 +2405,7 @@ func TestProvisionFromSnapshot(t *testing.T) {
 
 		pluginCaps, controllerCaps := provisionFromSnapshotCapabilities()
 		csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5, csiConn.conn,
-			client, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, false)
+			client, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, nil, false)
 
 		out := &csi.CreateVolumeResponse{
 			Volume: &csi.Volume{
@@ -2575,11 +2575,11 @@ func TestProvisionWithTopologyEnabled(t *testing.T) {
 
 			clientSet := fakeclientset.NewSimpleClientset(nodes, csiNodes)
 
-			scLister, csiNodeLister, nodeLister, stopChan := listers(clientSet)
+			scLister, csiNodeLister, nodeLister, claimLister, stopChan := listers(clientSet)
 			defer close(stopChan)
 
 			csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5,
-				csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), scLister, csiNodeLister, nodeLister, false)
+				csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), scLister, csiNodeLister, nodeLister, claimLister, false)
 
 			pv, err := csiProvisioner.Provision(controller.ProvisionOptions{
 				StorageClass: &storagev1.StorageClass{},
@@ -2634,7 +2634,7 @@ func TestProvisionWithTopologyDisabled(t *testing.T) {
 	clientSet := fakeclientset.NewSimpleClientset()
 	pluginCaps, controllerCaps := provisionWithTopologyCapabilities()
 	csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5,
-		csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, false)
+		csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, nil, false)
 
 	out := &csi.CreateVolumeResponse{
 		Volume: &csi.Volume{
@@ -2804,9 +2804,9 @@ func runDeleteTest(t *testing.T, k string, tc deleteTestcase) {
 	}
 
 	pluginCaps, controllerCaps := provisionCapabilities()
-	scLister, _, _, _ := listers(clientSet)
+	scLister, _, _, _, _ := listers(clientSet)
 	csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5,
-		csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), scLister, nil, nil, false)
+		csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), scLister, nil, nil, nil, false)
 
 	err = csiProvisioner.Delete(tc.persistentVolume)
 	if tc.expectErr && err == nil {
@@ -2901,12 +2901,14 @@ func TestProvisionFromPVC(t *testing.T) {
 		restoredVolSizeBig   bool                     // set to request a smaller volSize than source PVC, default false
 		expectedPVSpec       *pvSpec                  // set to expected PVSpec on success, for deep comparison, default nil
 		cloneUnsupported     bool                     // set to state clone feature not supported in capabilities, default false
+		expectFinalizers     bool                     // while set, expects clone protection finalizers to be set on a PVC
 		sourcePVStatusPhase  v1.PersistentVolumePhase // set to change source PV Status.Phase, default "Bound"
 		expectErr            bool                     // set to state, test is expected to return errors, default false
 	}{
 		"provision with pvc data source": {
-			clonePVName: pvName,
-			volOpts:     generatePVCForProvisionFromPVC(srcNamespace, srcName, fakeSc1, requestedBytes, ""),
+			clonePVName:      pvName,
+			volOpts:          generatePVCForProvisionFromPVC(srcNamespace, srcName, fakeSc1, requestedBytes, ""),
+			expectFinalizers: true,
 			expectedPVSpec: &pvSpec{
 				Name:          pvName,
 				ReclaimPolicy: v1.PersistentVolumeReclaimDelete,
@@ -2938,6 +2940,7 @@ func TestProvisionFromPVC(t *testing.T) {
 		"provision with pvc data source destination too small": {
 			clonePVName:          pvName,
 			volOpts:              generatePVCForProvisionFromPVC(srcNamespace, srcName, fakeSc1, requestedBytes+1, ""),
+			expectFinalizers:     true,
 			restoredVolSizeSmall: true,
 			expectErr:            true,
 		},
@@ -3027,9 +3030,9 @@ func TestProvisionFromPVC(t *testing.T) {
 			expectErr:   true,
 		},
 		"provision block data source is block": {
-			clonePVName: blockModePVName,
-			volOpts:     generatePVCForProvisionFromPVC(srcNamespace, blockModePVName, fakeSc1, requestedBytes, "block"),
-			expectErr:   false,
+			clonePVName:      blockModePVName,
+			volOpts:          generatePVCForProvisionFromPVC(srcNamespace, blockModePVName, fakeSc1, requestedBytes, "block"),
+			expectFinalizers: true,
 		},
 		"provision block but data source is filesystem": {
 			clonePVName: filesystemPVName,
@@ -3042,14 +3045,16 @@ func TestProvisionFromPVC(t *testing.T) {
 			expectErr:   true,
 		},
 		"provision nil mode data source is nil": {
-			clonePVName: pvName,
-			volOpts:     generatePVCForProvisionFromPVC(srcNamespace, srcName, fakeSc1, requestedBytes, ""),
-			expectErr:   false,
+			clonePVName:      pvName,
+			volOpts:          generatePVCForProvisionFromPVC(srcNamespace, srcName, fakeSc1, requestedBytes, ""),
+			expectFinalizers: true,
+			expectErr:        false,
 		},
 		"provision filesystem data source is filesystem": {
-			clonePVName: filesystemPVName,
-			volOpts:     generatePVCForProvisionFromPVC(srcNamespace, filesystemPVName, fakeSc1, requestedBytes, "filesystem"),
-			expectErr:   false,
+			clonePVName:      filesystemPVName,
+			volOpts:          generatePVCForProvisionFromPVC(srcNamespace, filesystemPVName, fakeSc1, requestedBytes, "filesystem"),
+			expectFinalizers: true,
+			expectErr:        false,
 		},
 		"provision filesystem but data source is block": {
 			clonePVName: blockModePVName,
@@ -3057,9 +3062,10 @@ func TestProvisionFromPVC(t *testing.T) {
 			expectErr:   true,
 		},
 		"provision filesystem data source is nil": {
-			clonePVName: pvName,
-			volOpts:     generatePVCForProvisionFromPVC(srcNamespace, srcName, fakeSc1, requestedBytes, "filesystem"),
-			expectErr:   false,
+			clonePVName:      pvName,
+			volOpts:          generatePVCForProvisionFromPVC(srcNamespace, srcName, fakeSc1, requestedBytes, "filesystem"),
+			expectFinalizers: true,
+			expectErr:        false,
 		},
 	}
 
@@ -3217,15 +3223,29 @@ func TestProvisionFromPVC(t *testing.T) {
 				}).Return(&csi.DeleteVolumeResponse{}, nil).Times(1)
 			}
 
+			_, _, _, claimLister, _ := listers(clientSet)
+
 			// Phase: execute the test
 			csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5, csiConn.conn,
-				nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, false)
+				nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil, nil, nil, claimLister, false)
 
 			pv, err = csiProvisioner.Provision(tc.volOpts)
 			if tc.expectErr && err == nil {
 				t.Errorf("test %q: Expected error, got none", k)
 			}
 
+			if tc.volOpts.PVC.Spec.DataSource != nil {
+				claim, _ := claimLister.PersistentVolumeClaims(tc.volOpts.PVC.Namespace).Get(tc.volOpts.PVC.Spec.DataSource.Name)
+				if claim != nil {
+					set := checkFinalizer(claim, pvcCloneFinalizer)
+					if tc.expectFinalizers && !set {
+						t.Errorf("Claim %s does not have clone protection finalizer set", claim.Name)
+					} else if !tc.expectFinalizers && set {
+						t.Errorf("Claim %s should not have clone protection finalizer set", claim.Name)
+					}
+				}
+			}
+
 			// Phase: process test responses
 			// process responses
 			if !tc.expectErr && err != nil {
@@ -3325,7 +3345,7 @@ func TestProvisionWithMigration(t *testing.T) {
 			pluginCaps, controllerCaps := provisionCapabilities()
 			csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner",
 				"test", 5, csiConn.conn, nil, driverName, pluginCaps, controllerCaps,
-				inTreePluginName, false, mockTranslator, nil, nil, nil, false)
+				inTreePluginName, false, mockTranslator, nil, nil, nil, nil, false)
 
 			// Set up return values (AnyTimes to avoid overfitting on implementation)
 
@@ -3485,7 +3505,7 @@ func TestDeleteMigration(t *testing.T) {
 			pluginCaps, controllerCaps := provisionCapabilities()
 			csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner",
 				"test", 5, csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "",
-				false, mockTranslator, nil, nil, nil, false)
+				false, mockTranslator, nil, nil, nil, nil, false)
 
 			// Set mock return values (AnyTimes to avoid overfitting on implementation details)
 			mockTranslator.EXPECT().IsPVMigratable(gomock.Any()).Return(tc.expectTranslation).AnyTimes()
diff --git a/pkg/controller/topology_test.go b/pkg/controller/topology_test.go
index 39e913d71..dee0476cb 100644
--- a/pkg/controller/topology_test.go
+++ b/pkg/controller/topology_test.go
@@ -395,7 +395,7 @@ func TestStatefulSetSpreading(t *testing.T) {
 
 	kubeClient := fakeclientset.NewSimpleClientset(nodes, csiNodes)
 
-	_, csiNodeLister, nodeLister, stopChan := listers(kubeClient)
+	_, csiNodeLister, nodeLister, _, stopChan := listers(kubeClient)
 	defer close(stopChan)
 
 	for name, tc := range testcases {
@@ -1085,7 +1085,7 @@ func TestTopologyAggregation(t *testing.T) {
 
 			kubeClient := fakeclientset.NewSimpleClientset(nodes, csiNodes)
 
-			_, csiNodeLister, nodeLister, stopChan := listers(kubeClient)
+			_, csiNodeLister, nodeLister, _, stopChan := listers(kubeClient)
 			defer close(stopChan)
 
 			var selectedNode *v1.Node
@@ -1338,7 +1338,7 @@ func TestPreferredTopologies(t *testing.T) {
 			kubeClient := fakeclientset.NewSimpleClientset(nodes, csiNodes)
 			selectedNode := &nodes.Items[0]
 
-			_, csiNodeLister, nodeLister, stopChan := listers(kubeClient)
+			_, csiNodeLister, nodeLister, _, stopChan := listers(kubeClient)
 			defer close(stopChan)
 
 			requirements, err := GenerateAccessibilityRequirements(
@@ -1552,13 +1552,19 @@ func requisiteEqual(t1, t2 []*csi.Topology) bool {
 	return unchecked.Len() == 0
 }
 
-func listers(kubeClient *fakeclientset.Clientset) (storagelistersv1.StorageClassLister, storagelistersv1beta1.CSINodeLister, corelisters.NodeLister, chan struct{}) {
+func listers(kubeClient *fakeclientset.Clientset) (
+	storagelistersv1.StorageClassLister,
+	storagelistersv1beta1.CSINodeLister,
+	corelisters.NodeLister,
+	corelisters.PersistentVolumeClaimLister,
+	chan struct{}) {
 	factory := informers.NewSharedInformerFactory(kubeClient, ResyncPeriodOfCsiNodeInformer)
 	stopChan := make(chan struct{})
 	scLister := factory.Storage().V1().StorageClasses().Lister()
 	csiNodeLister := factory.Storage().V1beta1().CSINodes().Lister()
 	nodeLister := factory.Core().V1().Nodes().Lister()
+	claimLister := factory.Core().V1().PersistentVolumeClaims().Lister()
 	factory.Start(stopChan)
 	factory.WaitForCacheSync(stopChan)
-	return scLister, csiNodeLister, nodeLister, stopChan
+	return scLister, csiNodeLister, nodeLister, claimLister, stopChan
 }