rbd: retain intermediate RBD snapshot on temp image

Currently, Ceph-CSI deletes intermediate RBD snapshot on
temporary cloned images (`csi-vol-xxxx-temp@csi-vol-xxxx`)
which is the parent of the final clone image.

The parent-child mirroring requires both the parent and child
images to be present (i.e, not in trash).

This commit makes enhancement to `createRBDClone` function by
introducing `deleteSnap` parameter. If `deleteSnap` is true,
the snapshot is deleted after the clone is created.

This is required to support mirroring of child image with its
parent image.

Signed-off-by: Praveen M <m.praveen@ibm.com>

internal/rbd/clone.go

@@ -33,13 +33,11 @@ import (
 // checkCloneImage check the cloned image exists, if the cloned image is not
 // found it will check the temporary cloned snapshot exists, and again it will
 // check the snapshot exists on the temporary cloned image, if yes it will
-// create a new cloned and delete the temporary snapshot and adds a task to
+// create a new cloned and adds a task to flatten the temp cloned image and return success.
-// flatten the temp cloned image and return success.
 //
 // if the temporary snapshot does not exists it creates a temporary snapshot on
 // temporary cloned image and creates  a new cloned with user-provided image
-// features and delete the temporary snapshot and adds a task to flatten the
+// features and adds a task to flatten the temp cloned image and return success
-// temp cloned image and return success
 //
 // if the temporary clone does not exist and if there is a temporary snapshot
 // present on the parent image it will delete the temporary snapshot and
@@ -58,9 +56,9 @@ func (rv *rbdVolume) checkCloneImage(ctx context.Context, parentVol *rbdVolume)
 	if err != nil {
 		switch {
 		case errors.Is(err, ErrSnapNotFound):
-			// as the snapshot is not present, create new snapshot,clone and
+			// as the snapshot is not present, create new snapshot, clone and
-			// delete the temporary snapshot
+			// don't delete the temporary snapshot
-			err = createRBDClone(ctx, tempClone, rv, snap)
+			err = createRBDClone(ctx, tempClone, rv, snap, false)
 			if err != nil {
 				return false, err
 			}
@@ -99,12 +97,6 @@ func (rv *rbdVolume) checkCloneImage(ctx context.Context, parentVol *rbdVolume)
 
 		return false, err
 	}
-	err = tempClone.deleteSnapshot(ctx, snap)
-	if err != nil {
-		log.ErrorLog(ctx, "failed to delete snapshot: %v", err)
-
-		return false, err
-	}
 
 	return true, nil
 }
@@ -206,7 +198,7 @@ func (rv *rbdVolume) doSnapClone(ctx context.Context, parentVol *rbdVolume) erro
 	cloneSnap.Pool = rv.Pool
 
 	// create snapshot and temporary clone and delete snapshot
-	err := createRBDClone(ctx, parentVol, tempClone, tempSnap)
+	err := createRBDClone(ctx, parentVol, tempClone, tempSnap, true)
 	if err != nil {
 		return err
 	}
@@ -237,8 +229,8 @@ func (rv *rbdVolume) doSnapClone(ctx context.Context, parentVol *rbdVolume) erro
 
 	// create snap of temp clone from temporary cloned image
 	// create final clone
-	// delete snap of temp clone
+	// don't delete snap of temp clone (needed for mirroring of PVC-PVC clone).
-	errClone = createRBDClone(ctx, tempClone, rv, cloneSnap)
+	errClone = createRBDClone(ctx, tempClone, rv, cloneSnap, false)
 	if errClone != nil {
 		return errClone
 	}

internal/rbd/controllerserver.go

@@ -1377,7 +1377,8 @@ func (cs *ControllerServer) doSnapshotClone(
 		return cloneRbd, err
 	}
 
-	err = createRBDClone(ctx, parentVol, cloneRbd, rbdSnap)
+	// create snapshot, clone and delete snapshot
+	err = createRBDClone(ctx, parentVol, cloneRbd, rbdSnap, true)
 	if err != nil {
 		log.ErrorLog(ctx, "failed to create snapshot: %v", err)
 

internal/rbd/rbd_util.go

@@ -743,7 +743,23 @@ func (ri *rbdImage) trashRemoveImage(ctx context.Context) error {
 // DeleteTempImage deletes the temporary image created for volume datasource.
 func (rv *rbdVolume) DeleteTempImage(ctx context.Context) error {
 	tempClone := rv.generateTempClone()
-	err := tempClone.Delete(ctx)
+
+	snap := &rbdSnapshot{}
+	defer snap.Destroy(ctx)
+
+	snap.RbdImageName = tempClone.RbdImageName
+	snap.RbdSnapName = rv.RbdImageName
+	snap.Pool = rv.Pool
+	err := tempClone.deleteSnapshot(ctx, snap)
+	if err != nil {
+		if !errors.Is(err, ErrSnapNotFound) {
+			log.ErrorLog(ctx, "failed to delete snapshot: %v", err)
+
+			return err
+		}
+	}
+
+	err = tempClone.Delete(ctx)
 	if err != nil {
 		if errors.Is(err, util.ErrImageNotFound) {
 			return tempClone.ensureImageCleanup(ctx)

internal/rbd/snapshot.go

@@ -29,10 +29,14 @@ import (
 	"github.com/ceph/ceph-csi/internal/util/log"
 )
 
+// createRBDClone creates a clone of the parentVol by creating a
+// snapshot of the parentVol and cloning the snapshot to the cloneRbdVol.
+// If deleteSnap is true, the snapshot is deleted after the clone is created.
 func createRBDClone(
 	ctx context.Context,
 	parentVol, cloneRbdVol *rbdVolume,
 	snap *rbdSnapshot,
+	deleteSnap bool,
 ) error {
 	// create snapshot
 	err := parentVol.createSnapshot(ctx, snap)
@@ -58,6 +62,10 @@ func createRBDClone(
 			snap.RbdSnapName,
 			err)
 	}
+	if !deleteSnap {
+		return nil
+	}
+
 	errSnap := parentVol.deleteSnapshot(ctx, snap)
 	if errSnap != nil {
 		log.ErrorLog(ctx, "failed to delete snapshot: %v", errSnap)