Add Delete Volume Finalizer

This PR adds Finalizer on the snapshot source PVC to prevent it
from being deleted when a snapshot is being created from it.

pkg/controller/framework_test.go

@@ -177,6 +177,11 @@ func withContentFinalizer(content *crdv1.VolumeSnapshotContent) *crdv1.VolumeSna
 	return content
 }
 
+func withPVCFinalizer(pvc *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
+	pvc.ObjectMeta.Finalizers = append(pvc.ObjectMeta.Finalizers, PVCFinalizer)
+	return pvc
+}
+
 // React is a callback called by fake kubeClient from the controller.
 // In other words, every snapshot/content change performed by the controller ends
 // here.
@@ -877,7 +882,7 @@ func newClaim(name, claimUID, capacity, boundToVolume string, phase v1.Persisten
 		claim.Status.Capacity = claim.Spec.Resources.Requests
 	}
 
-	return &claim
+	return withPVCFinalizer(&claim)
 }
 
 // newClaimArray returns array with a single claim that would be returned by

pkg/controller/snapshot_controller.go

@@ -85,6 +85,8 @@ const controllerUpdateFailMsg = "snapshot controller failed to update"
 
 const IsDefaultSnapshotClassAnnotation = "snapshot.storage.kubernetes.io/is-default-class"
 
+const SnapshotsInCreationCountAnnotation = "snapshot.storage.kubernetes.io/snapshots-in-creation-count"
+
 // syncContent deals with one key off the queue.  It returns false when it's time to quit.
 func (ctrl *csiSnapshotController) syncContent(content *crdv1.VolumeSnapshotContent) error {
 	klog.V(5).Infof("synchronizing VolumeSnapshotContent[%s]", content.Name)
@@ -192,11 +194,23 @@ func (ctrl *csiSnapshotController) syncSnapshot(snapshot *crdv1.VolumeSnapshot)
 		return ctrl.addSnapshotFinalizer(snapshot)
 	}
 
+	var err error = nil
 	if !snapshot.Status.ReadyToUse {
-		return ctrl.syncUnreadySnapshot(snapshot)
+		err = ctrl.syncUnreadySnapshot(snapshot)
+	} else {
+		err = ctrl.syncReadySnapshot(snapshot)
+	}
+
+	klog.V(5).Infof("syncSnapshot[%s]: check if we should remove finalizer on snapshot source and remove it if we can", snapshotKey(snapshot))
+	// Check if we should remove finalizer on snapshot source and remove it if we can.
+	errFinalizer := ctrl.checkandRemoveSnapshotSourceFinalizer(snapshot)
+	if errFinalizer != nil {
+		klog.Errorf("error check and remove snapshot source finalizer for snapshot [%s]: %v", snapshot.Name, errFinalizer)
+		// Log an event and keep the original error from syncUnready/ReadySnapshot
+		ctrl.eventRecorder.Event(snapshot, v1.EventTypeWarning, "ErrorSnapshotSourceFinalizer", "Error check and remove PVC Finalizer for VolumeSnapshot")
 	}
-	return ctrl.syncReadySnapshot(snapshot)
 
+	return err
 }
 
 // syncReadySnapshot checks the snapshot which has been bound to snapshot content successfully before.
@@ -367,7 +381,6 @@ func (ctrl *csiSnapshotController) createSnapshot(snapshot *crdv1.VolumeSnapshot
 			// We will get an "snapshot update" event soon, this is not a big error
 			klog.V(4).Infof("createSnapshot [%s]: cannot update internal cache: %v", snapshotKey(snapshotObj), updateErr)
 		}
-
 		return nil
 	})
 	return nil
@@ -612,6 +625,23 @@ func (ctrl *csiSnapshotController) createSnapshotOperation(snapshot *crdv1.Volum
 		return snapshot, nil
 	}
 
+	if ctrl.needToAddSnapshotSourceFinalizer(snapshot) {
+		// PVC is not being deleted -> it should have the finalizer.
+		klog.V(5).Infof("createSnapshotOperation: Add Finalizer for source of snapshot [%s]", snapshot.Name)
+		err := ctrl.addSnapshotSourceFinalizer(snapshot)
+		if err != nil {
+			klog.Errorf("createSnapshotOperation failed to add finalizer for source of snapshot %s", err)
+			return nil, err
+		}
+		// Increment SnapshotsInCreationCount
+		klog.V(5).Infof("createSnapshotOperation[%s]: Calling updateSnapshotsInCreationCount", snapshot.Name)
+		err = ctrl.updateSnapshotsInCreationCount(snapshot, true)
+		if err != nil {
+			klog.Errorf("createSnapshotOperation failed to update snapshots in creation count annotation %s", err)
+			return nil, err
+		}
+	}
+
 	class, volume, contentName, snapshotterCredentials, err := ctrl.getCreateSnapshotInput(snapshot)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get input parameters to create snapshot %s: %q", snapshot.Name, err)
@@ -1068,3 +1098,165 @@ func (ctrl *csiSnapshotController) removeSnapshotFinalizer(snapshot *crdv1.Volum
 	klog.V(5).Infof("Removed protection finalizer from volume snapshot %s", snapshotKey(snapshot))
 	return nil
 }
+
+// addSnapshotSourceFinalizer adds a Finalizer for VolumeSnapshot Source PVC.
+func (ctrl *csiSnapshotController) addSnapshotSourceFinalizer(snapshot *crdv1.VolumeSnapshot) error {
+	// Get snapshot source which is a PVC
+	pvc, err := ctrl.getClaimFromVolumeSnapshot(snapshot)
+	if err != nil {
+		return err
+	}
+
+	pvcClone := pvc.DeepCopy()
+	pvcClone.ObjectMeta.Finalizers = append(pvcClone.ObjectMeta.Finalizers, PVCFinalizer)
+	_, err = ctrl.client.CoreV1().PersistentVolumeClaims(pvcClone.Namespace).Update(pvcClone)
+	if err != nil {
+		return newControllerUpdateError(pvcClone.Name, err.Error())
+	}
+
+	klog.V(5).Infof("Added protection finalizer to persistent volume claim %s", pvc.Name)
+	return nil
+}
+
+// removeSnapshotSourceFinalizer removes a Finalizer for VolumeSnapshot Source PVC.
+func (ctrl *csiSnapshotController) removeSnapshotSourceFinalizer(snapshot *crdv1.VolumeSnapshot) error {
+	// Get snapshot source which is a PVC
+	pvc, err := ctrl.getClaimFromVolumeSnapshot(snapshot)
+	if err != nil {
+		return err
+	}
+
+	pvcClone := pvc.DeepCopy()
+	pvcClone.ObjectMeta.Finalizers = slice.RemoveString(pvcClone.ObjectMeta.Finalizers, PVCFinalizer, nil)
+
+	_, err = ctrl.client.CoreV1().PersistentVolumeClaims(pvcClone.Namespace).Update(pvcClone)
+	if err != nil {
+		return newControllerUpdateError(pvcClone.Name, err.Error())
+	}
+
+	klog.V(5).Infof("Removed protection finalizer from persistent volume claim %s", pvc.Name)
+	return nil
+}
+
+// isSnapshotSourceBeingUsed checks if a PVC is being used as a source to create a snapshot.
+func (ctrl *csiSnapshotController) isSnapshotSourceBeingUsed(snapshot *crdv1.VolumeSnapshot) bool {
+	klog.V(5).Infof("isSnapshotSourceBeingUsed[%s]: started", snapshotKey(snapshot))
+	// Get snapshot source which is a PVC
+	pvc, err := ctrl.getClaimFromVolumeSnapshot(snapshot)
+	if err != nil {
+		klog.Errorf("isSnapshotSourceBeingUsed: failed to get claim from snapshot: %v", err)
+		return false
+	}
+
+	annSnapCountPtr, err := getSnapshotsInCreationCountAnnotation(pvc.ObjectMeta)
+	if err != nil {
+		klog.Errorf("isSnapshotSourceBeingUsed: failed to get snapshots in creation count annotation: %v", err)
+		return false
+	}
+	if annSnapCountPtr != nil {
+		klog.V(5).Infof("isSnapshotSourceBeingUsed[%s]: annSnapshotPtr %d", snapshotKey(snapshot), *annSnapCountPtr)
+		if *annSnapCountPtr > 0 {
+			klog.Infof("isSnapshotSourceBeingUsed[%s]: There are %d snapshots being created from source PVC %s", snapshotKey(snapshot), *annSnapCountPtr, pvc.Name)
+			return true
+		}
+	}
+	klog.V(5).Infof("isSnapshotSourceBeingUsed: no snapshot is being created from volume %s", pvc.Name)
+	return false
+}
+
+// updateSnapshotsInCreationCount increments or decrements the snapshots in creation count annotation
+func (ctrl *csiSnapshotController) updateSnapshotsInCreationCount(snapshot *crdv1.VolumeSnapshot, increment bool) error {
+	klog.V(5).Infof("updateSnapshotsInCreationCount[%s]: started. increment is [%t]", snapshot.Name, increment)
+	// Get snapshot source which is a PVC
+	pvc, err := ctrl.getClaimFromVolumeSnapshot(snapshot)
+	if err != nil {
+		return err
+	}
+	annSnapCountPtr, err := getSnapshotsInCreationCountAnnotation(pvc.ObjectMeta)
+	if err != nil {
+		klog.Errorf("updateSnapshotsInCreationCount: failed to get snapshots in creation count annotation: %v", err)
+		return err
+	}
+	var snapCount int64 = 0
+	if increment {
+		if annSnapCountPtr != nil {
+			snapCount = *annSnapCountPtr + 1
+			klog.Infof("updateSnapshotsInCreationCount[%s]: snapshots in creation count is incremented to %d", snapshot.Name, snapCount)
+		} else {
+			snapCount = 1
+			klog.Infof("updateSnapshotsInCreationCount[%s]: snapshots in creation count started at %d", snapshot.Name, snapCount)
+		}
+	} else {
+		if annSnapCountPtr != nil {
+			klog.V(5).Infof("updateSnapshotsInCreationCount[%s]: current snapshots in creation count is %d.", snapshot.Name, *annSnapCountPtr)
+			if *annSnapCountPtr > 0 {
+				snapCount = *annSnapCountPtr - 1
+				klog.Infof("updateSnapshotsInCreationCount[%s]: snapshots in creation count is %d after decrementing", snapshot.Name, snapCount)
+			}
+		}
+	}
+	metav1.SetMetaDataAnnotation(&pvc.ObjectMeta, SnapshotsInCreationCountAnnotation, fmt.Sprintf("%d", snapCount))
+	klog.Infof("updateSnapshotsInCreationCount[%s]: sets SnapshotsInCreationCountAnnotation to %d", snapshot.Name, snapCount)
+	pvcClone := pvc.DeepCopy()
+	// Update pvc object
+	_, err = ctrl.client.CoreV1().PersistentVolumeClaims(pvcClone.Namespace).Update(pvcClone)
+	if err != nil {
+		return newControllerUpdateError(pvcClone.Name, err.Error())
+	}
+
+	return nil
+}
+
+// checkandRemoveSnapshotSourceFinalizer checks if the snapshot source finalizer should be removed
+// and removed it if needed.
+func (ctrl *csiSnapshotController) checkandRemoveSnapshotSourceFinalizer(snapshot *crdv1.VolumeSnapshot) error {
+	// Get snapshot source which is a PVC
+	pvc, err := ctrl.getClaimFromVolumeSnapshot(snapshot)
+	if err != nil {
+		klog.Infof("cannot get claim from snapshot [%s]: [%v] Claim may be deleted already. No need to remove finalizer on the claim.", snapshot.Name, err)
+		return nil
+	}
+
+	klog.V(5).Infof("checkandRemoveSnapshotSourceFinalizer for snapshot [%s]: snapshot status [%#v]", snapshot.Name, snapshot.Status)
+
+	// Check if snapshot creation completed or failed. If so, decrement snapshots creation count annotation
+	if snapshot.Status.ReadyToUse == true || snapshot.Status.Error != nil {
+		klog.Infof("checkandRemoveSnapshotSourceFinalizer: snapshot %s is in Ready or Error status. Decrement the snapshots in creation count.", snapshot.Name)
+		// Decrement the snapshots in creation count
+		err = ctrl.updateSnapshotsInCreationCount(snapshot, false)
+		if err != nil {
+			klog.Errorf("checkandRemoveSnapshotSourceFinalizer [%s]: updateSnapshotsInCreationCount failed to update snapshots in creation count annotation %v", snapshot.Name, err)
+			return err
+		}
+	}
+
+	// Check if there is a Finalizer on PVC to be removed
+	if slice.ContainsString(pvc.ObjectMeta.Finalizers, PVCFinalizer, nil) {
+		// There is a Finalizer on PVC. Check if PVC is used
+		// and remove finalizer if it's not used.
+		// Check if snapshotsInCreationCount is 0.
+		isUsed := ctrl.isSnapshotSourceBeingUsed(snapshot)
+		if !isUsed {
+			klog.Infof("checkandRemoveSnapshotSourceFinalizer[%s]: Remove Finalizer for PVC %s as it is not used by snapshots in creation", snapshot.Name, pvc.Name)
+			err = ctrl.removeSnapshotSourceFinalizer(snapshot)
+			if err != nil {
+				klog.Errorf("checkandRemoveSnapshotSourceFinalizer [%s]: removeSnapshotSourceFinalizer failed to remove finalizer %v", snapshot.Name, err)
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+// needToAddSnapshotSourceFinalizer checks if a Finalizer needs to be added for the snapshot source.
+func (ctrl *csiSnapshotController) needToAddSnapshotSourceFinalizer(snapshot *crdv1.VolumeSnapshot) bool {
+	// Get snapshot source which is a PVC
+	pvc, err := ctrl.getClaimFromVolumeSnapshot(snapshot)
+	if err != nil {
+		klog.Errorf("failed to check if there is a need to add Finalizer for snapshot source: %v", err)
+		return false
+	}
+
+	return pvc.ObjectMeta.DeletionTimestamp == nil && !slice.ContainsString(pvc.ObjectMeta.Finalizers, PVCFinalizer, nil)
+}

pkg/controller/util.go

@@ -75,6 +75,9 @@ var snapshotterSecretParams = deprecatedSecretParamsMap{
 	secretNamespaceKey:           prefixedSnapshotterSecretNamespaceKey,
 }
 
+// Name of finalizer on PVCs that have been used as a source to create VolumeSnapshots
+const PVCFinalizer = "snapshot.storage.kubernetes.io/pvc-protection"
+
 func snapshotKey(vs *crdv1.VolumeSnapshot) string {
 	return fmt.Sprintf("%s/%s", vs.Namespace, vs.Name)
 }
@@ -206,6 +209,25 @@ func verifyAndGetSecretNameAndNamespaceTemplate(secret deprecatedSecretParamsMap
 	}
 }
 
+// getSnapshotsInCreationCountAnnotation returns the SnapshotsInCreationCount annotation.
+func getSnapshotsInCreationCountAnnotation(obj metav1.ObjectMeta) (*int64, error) {
+	if metav1.HasAnnotation(obj, SnapshotsInCreationCountAnnotation) {
+		snapCount := obj.Annotations[SnapshotsInCreationCountAnnotation]
+		klog.V(5).Infof("getSnapshotsInCreationCountAnnotation: %s from annotation", snapCount)
+		i, err := strconv.ParseInt(snapCount, 10, 64)
+		if err != nil {
+			klog.Errorf("getSnapshotsInCreationCountAnnotation: failed to parse SnapshotsInCreationCountAnnotation: %v", err)
+			return nil, err
+		}
+		klog.Infof("getSnapshotsInCreationCountAnnotation: There are %d snapshots being created from source PVC", i)
+
+		return &i, nil
+	}
+
+	// Annotation not set
+	return nil, nil
+}
+
 // getSecretReference returns a reference to the secret specified in the given nameTemplate
 //  and namespaceTemplate, or an error if the templates are not specified correctly.
 // No lookup of the referenced secret is performed, and the secret may or may not exist.