snap-deletion-flatten

Signed-off-by: Rakshith R <[email protected]>

e2e/rbd.go

@@ -457,15 +457,9 @@ var _ = Describe("RBD", func() {
 				validatePVCSnapshotDeletion(
 					higherCloneCount,
 					pvcPath,
-					appPath,
 					snapshotPath,
 					pvcClonePath,
-					appClonePath,
-					noKMS, noKMS,
-					defaultSCName,
-					noDataPool,
-					f,
-					noPVCValidation)
+					f)
 			})
 
 			By("create a PVC and check PVC/PV metadata on RBD image", func() {

e2e/utils.go

@@ -1404,14 +1404,17 @@ func validatePVCSnapshot(
 	validateRBDImageCount(f, 0, defaultRBDPool)
 }
 
-//nolint:gocyclo,gocognit,nestif,cyclop // reduce complexity
 func validatePVCSnapshotDeletion(
 	totalCount int,
-	pvcPath, appPath, snapshotPath, pvcClonePath, appClonePath string,
-	kms, restoreKMS kmsConfig, restoreSCName,
-	dataPool string, f *framework.Framework,
-	isEncryptedPVC validateFunc,
+	pvcPath, snapshotPath, pvcClonePath string,
+	f *framework.Framework,
 ) {
+	defer func() {
+		std, stderr, err := execCommandInToolBoxPod(f, "ceph rbd task list", f.UniqueName)
+		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)
@@ -1444,9 +1447,6 @@ func validatePVCSnapshotDeletion(
 		framework.Failf("failed to load PVC: %v", err)
 	}
 	pvcClone.Namespace = f.UniqueName
-	if restoreSCName != "" {
-		pvcClone.Spec.StorageClassName = &restoreSCName
-	}
 
 	// create snapshot, restore to PVC, delete snapshot
 	for i := 0; i < totalCount; i++ {

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(