rbd: consolidate snapshot flatten logic in PrepareVolumeForSnapshot()

This commit consolidates flatten logic checks for cloneDepth
and snapshotLimit in PrepareVolumeForSnapshot. This allows
the function to be called for both CreateSnapshot and
CreateVolumeGroupSnapshot.
Clone Depth check and flattening of grand parent image
now occurs before creation of snapshot starts.
This aligns better with how PVC-PVC clone and
PVC-restore process occurs currently.
Flattening the grandparent image once prevents
flattening of every newly created snapshot.
Snapshot in above para refers to k8s VolumeSnapshot
(which is backed by a rbd image).

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

internal/rbd/controllerserver.go

@@ -1157,7 +1157,7 @@ func (cs *ControllerServer) CreateSnapshot(
 		return cloneFromSnapshot(ctx, rbdVol, rbdSnap, cr, req.GetParameters())
 	}
 
-	err = flattenTemporaryClonedImages(ctx, rbdVol, cr)
+	err = rbdVol.PrepareVolumeForSnapshot(ctx, cr)
 	if err != nil {
 		return nil, err
 	}
@@ -1376,11 +1376,6 @@ func (cs *ControllerServer) doSnapshotClone(
 		return cloneRbd, err
 	}
 
-	err = cloneRbd.flattenRbdImage(ctx, false, rbdHardMaxCloneDepth, rbdSoftMaxCloneDepth)
-	if err != nil {
-		return cloneRbd, err
-	}
-
 	return cloneRbd, nil
 }
 

internal/rbd/group_controllerserver.go

@@ -130,6 +130,17 @@ func (cs *ControllerServer) CreateVolumeGroupSnapshot(
 			"failed to get existing one with name %q: %v", vgsName, err)
 	}
 
+	creds, err := mgr.GetCredentials()
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+	for _, volume := range volumes {
+		err = volume.PrepareVolumeForSnapshot(ctx, creds)
+		if err != nil {
+			return nil, status.Error(codes.Aborted, err.Error())
+		}
+	}
+
 	// create a temporary VolumeGroup with a different name
 	vg, err = mgr.CreateVolumeGroup(ctx, vgName)
 	if err != nil {

internal/rbd/manager.go

@@ -66,8 +66,8 @@ func (mgr *rbdManager) Destroy(ctx context.Context) {
 	}
 }
 
-// getCredentials sets up credentials and connects to the journal.
-func (mgr *rbdManager) getCredentials() (*util.Credentials, error) {
+// GetCredentials sets up credentials and connects to the journal.
+func (mgr *rbdManager) GetCredentials() (*util.Credentials, error) {
 	if mgr.creds != nil {
 		return mgr.creds, nil
 	}
@@ -87,7 +87,7 @@ func (mgr *rbdManager) getVolumeGroupJournal(clusterID string) (journal.VolumeGr
 		return mgr.vgJournal, nil
 	}
 
-	creds, err := mgr.getCredentials()
+	creds, err := mgr.GetCredentials()
 	if err != nil {
 		return nil, err
 	}
@@ -166,7 +166,7 @@ func (mgr *rbdManager) getGroupUUID(
 }
 
 func (mgr *rbdManager) GetVolumeByID(ctx context.Context, id string) (types.Volume, error) {
-	creds, err := mgr.getCredentials()
+	creds, err := mgr.GetCredentials()
 	if err != nil {
 		return nil, err
 	}
@@ -191,7 +191,7 @@ func (mgr *rbdManager) GetVolumeByID(ctx context.Context, id string) (types.Volu
 }
 
 func (mgr *rbdManager) GetSnapshotByID(ctx context.Context, id string) (types.Snapshot, error) {
-	creds, err := mgr.getCredentials()
+	creds, err := mgr.GetCredentials()
 	if err != nil {
 		return nil, err
 	}
@@ -216,7 +216,7 @@ func (mgr *rbdManager) GetSnapshotByID(ctx context.Context, id string) (types.Sn
 }
 
 func (mgr *rbdManager) GetVolumeGroupByID(ctx context.Context, id string) (types.VolumeGroup, error) {
-	creds, err := mgr.getCredentials()
+	creds, err := mgr.GetCredentials()
 	if err != nil {
 		return nil, err
 	}
@@ -230,7 +230,7 @@ func (mgr *rbdManager) GetVolumeGroupByID(ctx context.Context, id string) (types
 }
 
 func (mgr *rbdManager) CreateVolumeGroup(ctx context.Context, name string) (types.VolumeGroup, error) {
-	creds, err := mgr.getCredentials()
+	creds, err := mgr.GetCredentials()
 	if err != nil {
 		return nil, err
 	}
@@ -324,7 +324,7 @@ func (mgr *rbdManager) GetVolumeGroupSnapshotByID(
 	ctx context.Context,
 	id string,
 ) (types.VolumeGroupSnapshot, error) {
-	creds, err := mgr.getCredentials()
+	creds, err := mgr.GetCredentials()
 	if err != nil {
 		return nil, err
 	}

internal/rbd/rbd_util.go

@@ -2273,3 +2273,28 @@ func (ri *rbdImage) GetClusterID(ctx context.Context) (string, error) {
 
 	return ri.ClusterID, nil
 }
+
+func (rv *rbdVolume) PrepareVolumeForSnapshot(ctx context.Context, cr *util.Credentials) error {
+	hardLimit := rbdHardMaxCloneDepth
+	softLimit := rbdSoftMaxCloneDepth
+	err := flattenTemporaryClonedImages(ctx, rv, cr)
+	if err != nil {
+		return err
+	}
+
+	// choosing 2, since snapshot adds one depth and we'll be flattening the parent.
+	const depthToAvoidFlatten = 2
+	if rbdHardMaxCloneDepth > depthToAvoidFlatten {
+		hardLimit = rbdHardMaxCloneDepth - depthToAvoidFlatten
+	}
+	if rbdSoftMaxCloneDepth > depthToAvoidFlatten {
+		softLimit = rbdSoftMaxCloneDepth - depthToAvoidFlatten
+	}
+
+	err = rv.flattenParent(ctx, hardLimit, softLimit)
+	if err != nil {
+		return getGRPCErrorForCreateVolume(err)
+	}
+
+	return nil
+}

internal/rbd/types/manager.go

@@ -18,6 +18,8 @@ package types
 
 import (
 	"context"
+
+	"github.com/ceph/ceph-csi/internal/util"
 )
 
 // VolumeResolver can be used to construct a Volume from a CSI VolumeId.
@@ -45,6 +47,9 @@ type Manager interface {
 	// Destroy frees all resources that the Manager allocated.
 	Destroy(ctx context.Context)
 
+	// Getcredentials sets up credentials and connects to the journal.
+	GetCredentials() (*util.Credentials, error)
+
 	// GetVolumeGroupByID uses the CSI-Addons VolumeGroupId to resolve the
 	// returned VolumeGroup.
 	GetVolumeGroupByID(ctx context.Context, id string) (VolumeGroup, error)

internal/rbd/types/volume.go

@@ -58,6 +58,10 @@ type Volume interface {
 	// 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.
 	HandleParentImageExistence(ctx context.Context, flattenMode FlattenMode) error
+	// PrepareVolumeForSnapshot prepares the volume for snapshot by
+	// checking snapshots limit and clone depth limit and flattening it
+	// if required.
+	PrepareVolumeForSnapshot(ctx context.Context, cr *util.Credentials) error
 
 	// ToMirror converts the Volume to a Mirror.
 	ToMirror() (Mirror, error)