rbd: add support for flattenMode option for replication

This commit adds support for flattenMode option
for replication.
If the flattenMode is set to "force" in
volumereplicationclass parameters, cephcsi will
add a task to flatten the image if it has parent.
This enable cephcsi to then mirror such images after
flattening them.
The error message when the image's parent is
in trash or unmirrored is improved as well.

Signed-off-by: Rakshith R <rar@redhat.com>

internal/rbd/controllerserver.go

@@ -1020,21 +1020,11 @@ func cleanupRBDImage(ctx context.Context,
 
 	// delete the temporary rbd image created as part of volume clone during
 	// create volume
-	tempClone := rbdVol.generateTempClone()
-	err = tempClone.deleteImage(ctx)
+	err = rbdVol.DeleteTempImage(ctx)
 	if err != nil {
-		if errors.Is(err, ErrImageNotFound) {
-			err = tempClone.ensureImageCleanup(ctx)
-			if err != nil {
-				return nil, status.Error(codes.Internal, err.Error())
-			}
-		} else {
-			// return error if it is not ErrImageNotFound
-			log.ErrorLog(ctx, "failed to delete rbd image: %s with error: %v",
-				tempClone, err)
+		log.ErrorLog(ctx, "failed to delete temporary rbd image: %v", err)
 
-			return nil, status.Error(codes.Internal, err.Error())
-		}
+		return nil, status.Error(codes.Internal, err.Error())
 	}
 
 	// Deleting rbd image

internal/rbd/mirror.go

@@ -25,6 +25,69 @@ import (
 	librbd "github.com/ceph/go-ceph/rbd"
 )
 
+// FlattenMode is used to indicate the flatten mode for an RBD image.
+type FlattenMode string
+
+const (
+	// FlattenModeNever indicates that the image should never be flattened.
+	FlattenModeNever FlattenMode = "never"
+	// FlattenModeForce indicates that the image with the parent must be flattened.
+	FlattenModeForce FlattenMode = "force"
+)
+
+// HandleParentImageExistence checks the image's parent.
+// if the parent image does not exist and is not in trash, it returns nil.
+// if the flattenMode is FlattenModeForce, it flattens the image itself.
+// if the parent image is in trash, it returns an error.
+// if the parent image exists and is not enabled for mirroring, it returns an error.
+func (rv *rbdVolume) HandleParentImageExistence(
+	ctx context.Context,
+	flattenMode FlattenMode,
+) error {
+	if rv.ParentName == "" && !rv.ParentInTrash {
+		return nil
+	}
+
+	if flattenMode == FlattenModeForce {
+		// Delete temp image that exists for volume datasource since
+		// it is no longer required when the live image is flattened.
+		err := rv.DeleteTempImage(ctx)
+		if err != nil {
+			return fmt.Errorf("failed to delete temporary rbd image: %w", err)
+		}
+
+		err = rv.flattenRbdImage(ctx, true, 0, 0)
+		if err != nil {
+			return err
+		}
+	}
+
+	if rv.ParentInTrash {
+		return fmt.Errorf("%w: failed to enable mirroring on image %q:"+
+			" parent is in trash",
+			ErrFailedPrecondition, rv)
+	}
+
+	parent, err := rv.getParent()
+	if err != nil {
+		return err
+	}
+	parentMirroringInfo, err := parent.GetImageMirroringInfo()
+	if err != nil {
+		return fmt.Errorf(
+			"failed to get mirroring info of parent %q of image %q: %w",
+			parent, rv, err)
+	}
+
+	if parentMirroringInfo.State != librbd.MirrorImageEnabled {
+		return fmt.Errorf("%w: failed to enable mirroring on image %q: "+
+			"parent image %q is not enabled for mirroring",
+			ErrFailedPrecondition, rv, parent)
+	}
+
+	return nil
+}
+
 // EnableImageMirroring enables mirroring on an image.
 func (ri *rbdImage) EnableImageMirroring(mode librbd.ImageMirrorMode) error {
 	image, err := ri.open()

internal/rbd/rbd_util.go

@@ -703,6 +703,22 @@ func (ri *rbdImage) trashRemoveImage(ctx context.Context) error {
 	return nil
 }
 
+// DeleteTempImage deletes the temporary image created for volume datasource.
+func (rv *rbdVolume) DeleteTempImage(ctx context.Context) error {
+	tempClone := rv.generateTempClone()
+	err := tempClone.deleteImage(ctx)
+	if err != nil {
+		if errors.Is(err, ErrImageNotFound) {
+			return tempClone.ensureImageCleanup(ctx)
+		} else {
+			// return error if it is not ErrImageNotFound
+			return err
+		}
+	}
+
+	return nil
+}
+
 func (ri *rbdImage) getCloneDepth(ctx context.Context) (uint, error) {
 	var depth uint
 	vol := rbdVolume{}