e2e: add testcase to test snapshot deletion in specific case

e2e/rbd.go

@@ -86,7 +86,8 @@ var (
 	snapshotPath           = rbdExamplePath + "snapshot.yaml"
 	deployFSAppPath        = e2eTemplatesPath + "rbd-fs-deployment.yaml"
 	deployBlockAppPath     = e2eTemplatesPath + "rbd-block-deployment.yaml"
-	defaultCloneCount      = 3 // TODO: set to 10 once issues#2327 is fixed
+	defaultCloneCount      = 3
+	higherCloneCount       = 10
 
 	nbdMapOptions             = "nbd:debug-rbd=20"
 	e2eDefaultCephLogStrategy = "preserve"
@@ -452,6 +453,15 @@ var _ = Describe("RBD", func() {
 				}
 			})
 
+			By("create multiple PVC clone immediately followed by snapshot deletion", func() {
+				validatePVCSnapshotDeletion(
+					higherCloneCount,
+					pvcPath,
+					snapshotPath,
+					pvcClonePath,
+					f)
+			})
+
 			By("create a PVC and check PVC/PV metadata on RBD image", func() {
 				pvc, err := loadPVC(pvcPath)
 				if err != nil {
@@ -2412,7 +2422,7 @@ var _ = Describe("RBD", func() {
 
 			By("create a PVC clone and bind it to an app", func() {
 				validatePVCSnapshot(
-					defaultCloneCount,
+					higherCloneCount,
 					pvcPath,
 					appPath,
 					snapshotPath,
@@ -2427,7 +2437,7 @@ var _ = Describe("RBD", func() {
 
 			By("create a PVC-PVC clone and bind it to an app", func() {
 				validatePVCClone(
-					defaultCloneCount,
+					higherCloneCount,
 					pvcPath,
 					appPath,
 					pvcSmartClonePath,

e2e/utils.go

@@ -1404,6 +1404,109 @@ func validatePVCSnapshot(
 	validateRBDImageCount(f, 0, defaultRBDPool)
 }
 
+func validatePVCSnapshotDeletion(
+	totalCount int,
+	pvcPath, snapshotPath, pvcClonePath string,
+	f *framework.Framework,
+) {
+	defer func() {
+		std, stderr, err := execCommandInToolBoxPod(f, "ceph rbd task list", rookNamespace)
+		framework.Logf("ceph rbd task list output: %s : %s", std, stderr)
+		framework.Logf("ceph rbd task list error: %v", err)
+
+	}()
+	var wg sync.WaitGroup
+	wgErrs := make([]error, totalCount)
+	err := createRBDSnapshotClass(f)
+	if err != nil {
+		framework.Failf("failed to create storageclass: %v", err)
+	}
+	defer func() {
+		err = deleteRBDSnapshotClass()
+		if err != nil {
+			framework.Failf("failed to delete VolumeSnapshotClass: %v", err)
+		}
+	}()
+
+	pvc, err := loadPVC(pvcPath)
+	if err != nil {
+		framework.Failf("failed to load PVC: %v", err)
+	}
+	pvc.Namespace = f.UniqueName
+	err = createPVCAndvalidatePV(f.ClientSet, pvc, deployTimeout)
+	if err != nil {
+		framework.Failf("failed to create PVC: %v", err)
+	}
+	validateRBDImageCount(f, 1, defaultRBDPool)
+	snap := getSnapshot(snapshotPath)
+	snap.Namespace = f.UniqueName
+	snap.Spec.Source.PersistentVolumeClaimName = &pvc.Name
+
+	pvcClone, err := loadPVC(pvcClonePath)
+	if err != nil {
+		framework.Failf("failed to load PVC: %v", err)
+	}
+	pvcClone.Namespace = f.UniqueName
+
+	// create snapshot, restore to PVC, delete snapshot
+	for i := 0; i < totalCount; i++ {
+		snap.Name = fmt.Sprintf("%s%d", f.UniqueName, i)
+		err = createSnapshot(&snap, deployTimeout)
+		if err != nil {
+			framework.Failf("failed to create snapshot: %v", err)
+		}
+
+		pvcClone.Name = fmt.Sprintf("restore-%s", snap.Name)
+		pvcClone.Spec.DataSource.Name = snap.Name
+		err = createPVCAndvalidatePV(f.ClientSet, pvcClone, deployTimeout)
+		if err != nil {
+			framework.Failf("failed to create PVC: %v", err)
+		}
+
+		err = deleteSnapshot(&snap, deployTimeout)
+		if err != nil {
+			framework.Failf("failed to delete snapshot: %v", err)
+		}
+	}
+
+	// total images in cluster is 1 parent rbd image+ total clones
+	validateRBDImageCount(f, totalCount+1, defaultRBDPool)
+	wg.Add(totalCount)
+	// delete clones
+	for i := 0; i < totalCount; i++ {
+		go func(n int, p v1.PersistentVolumeClaim) {
+			name := fmt.Sprintf("%s%d", f.UniqueName, n)
+			p.Spec.DataSource.Name = name
+			wgErrs[n] = deletePVCAndValidatePV(f.ClientSet, &p, deployTimeout)
+			wg.Done()
+		}(i, *pvcClone)
+	}
+	wg.Wait()
+
+	failed := 0
+	for i, err := range wgErrs {
+		if err != nil {
+			// not using Failf() as it aborts the test and does not log other errors
+			framework.Logf("failed to delete PVC and application (%s%d): %v", f.UniqueName, i, err)
+			failed++
+		}
+	}
+	if failed != 0 {
+		framework.Failf("deleting PVCs and applications failed, %d errors were logged", failed)
+	}
+
+	// total images in cluster is 1 parent rbd image
+	validateRBDImageCount(f, totalCount+1, defaultRBDPool)
+	// delete parent pvc
+	err = deletePVCAndValidatePV(f.ClientSet, pvc, deployTimeout)
+	if err != nil {
+		framework.Failf("failed to delete PVC: %v", err)
+	}
+
+	// validate created backend rbd images
+	validateRBDImageCount(f, 0, defaultRBDPool)
+}
+
 // validateController simulates the required operations to validate the
 // controller.
 // Controller will generates the omap data when the PV is created.

internal/rbd/controllerserver.go

@@ -1469,7 +1469,13 @@ func (cs *ControllerServer) DeleteSnapshot(
 	}
 	defer rbdVol.Destroy()
 
-	rbdVol.ImageID = rbdSnap.ImageID
+	err = flattenSnapshotImage(ctx, rbdVol)
+	if err != nil {
+		log.ErrorLog(ctx, "failed to flatten snapshot: %v", err)
+
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
 	// update parent name to delete the snapshot
 	rbdSnap.RbdImageName = rbdVol.RbdImageName
 	err = cleanUpSnapshot(ctx, rbdVol, rbdSnap, rbdVol)

internal/rbd/rbd_util.go

@@ -1618,6 +1618,22 @@ func (ri *rbdImage) getImageInfo() error {
 	return nil
 }
 
+// checkChildrenExists checks if the image has any children.
+func (ri *rbdImage) checkChildrenExists() (bool, error) {
+	image, err := ri.open()
+	if err != nil {
+		return false, fmt.Errorf("failed to open image %s: %w", ri, err)
+	}
+	defer image.Close()
+
+	_, children, err := image.ListChildren()
+	if err != nil {
+		return false, fmt.Errorf("failed to list children of image %s: %w", ri, err)
+	}
+
+	return (len(children) > 0), nil
+}
+
 // getParent returns parent image if it exists.
 func (ri *rbdImage) getParent() (*rbdImage, error) {
 	err := ri.getImageInfo()

internal/rbd/snapshot.go

@@ -67,6 +67,57 @@ func createRBDClone(
 	return nil
 }
 
+// flattenSnapshotImage flattens triggers flattening of the rbd image underlying
+// the snapshot if the following conditions are met:
+// - the parent rbd image still exists
+// - the parent rbd image has more snapshots than minSnapshotsOnImageToStartFlatten
+// - the rbd image underlying snapshot has children
+// FlattenInProgress error is ignored.
+// This process ensures that further snapshot creation on the parent image will
+// not fail due to existing snapshots' image being stuck in trash and
+// minSnapshotsOnImageToStartFlatten limit being reached.
+func flattenSnapshotImage(
+	ctx context.Context,
+	rbdVol *rbdVolume,
+) error {
+	parentImage, err := rbdVol.getParent()
+	if err != nil {
+		return fmt.Errorf("failed to get parent image: %w", err)
+	}
+	if parentImage == nil {
+		return nil
+	}
+	defer parentImage.Destroy()
+
+	snapList, err := parentImage.listSnapshots()
+	if err != nil {
+		return fmt.Errorf("failed to list snapshots: %w", err)
+	}
+
+	if len(snapList) <= int(minSnapshotsOnImageToStartFlatten) { // minSnapshotsOnImageToStartFlatten) {
+		return nil
+	}
+
+	exists, err := rbdVol.checkChildrenExists()
+	if err != nil {
+		return fmt.Errorf("failed to check children exists: %w", err)
+	}
+
+	if !exists {
+		return nil
+	}
+
+	log.DebugLog(ctx, "Flattening image %s", rbdVol.RbdImageName)
+	// flatten image
+	err = rbdVol.flattenRbdImage(ctx,
+		true, rbdHardMaxCloneDepth, rbdSoftMaxCloneDepth)
+	if err != nil && !errors.Is(err, ErrFlattenInProgress) {
+		return fmt.Errorf("failed to flatten image: %w", err)
+	}
+
+	return nil
+}
+
 // cleanUpSnapshot removes the RBD-snapshot (rbdSnap) from the RBD-image
 // (parentVol) and deletes the RBD-image rbdVol.
 func cleanUpSnapshot(