diff --git a/deploy/cephcsi/image/Dockerfile b/deploy/cephcsi/image/Dockerfile
index dc717cd78ab..d42361bc163 100644
--- a/deploy/cephcsi/image/Dockerfile
+++ b/deploy/cephcsi/image/Dockerfile
@@ -2,7 +2,7 @@ ARG SRC_DIR="/go/src/github.com/ceph/ceph-csi/"
 ARG GO_ARCH
 ARG BASE_IMAGE
 
-FROM ${BASE_IMAGE} as updated_base
+FROM quay.ceph.io/ceph-ci/ceph:main as updated_base
 
 # TODO: remove the following cmd, when issues
 # https://github.com/ceph/ceph-container/issues/2034
diff --git a/e2e/rbd.go b/e2e/rbd.go
index 6edbfe57607..6900c72a03f 100644
--- a/e2e/rbd.go
+++ b/e2e/rbd.go
@@ -93,7 +93,7 @@ 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      = 5 // TODO: set to 10 once issues#2327 is fixed
 
 	nbdMapOptions             = "nbd:debug-rbd=20"
 	e2eDefaultCephLogStrategy = "preserve"
@@ -4857,6 +4857,7 @@ var _ = Describe("RBD", func() {
 			})
 
 			By("test volumeGroupSnapshot", func() {
+				Skip("skip to test flatten on normal snapshots")
 				supported, err := librbdSupportsVolumeGroupSnapshot(f)
 				if err != nil {
 					framework.Failf("failed to check for VolumeGroupSnapshot support: %v", err)
diff --git a/e2e/volumegroupsnapshot_base.go b/e2e/volumegroupsnapshot_base.go
index 47bb177386d..32097dddc87 100644
--- a/e2e/volumegroupsnapshot_base.go
+++ b/e2e/volumegroupsnapshot_base.go
@@ -456,6 +456,22 @@ func (v *volumeGroupSnapshotterBase) testVolumeGroupSnapshot(vol VolumeGroupSnap
 		return fmt.Errorf("failed to create volume group snapshot: %w", err)
 	}
 
+	// Create and delete 5 additional group snapshots to test flattening on minSnapshotLimit
+	for i := range 5 {
+		newVGSName := fmt.Sprintf("%s-%d", vgsName, i)
+		_, err = v.CreateVolumeGroupSnapshot(newVGSName, vgscName, pvcLabels)
+		if err != nil {
+			return fmt.Errorf("failed to create volume group snapshot %q: %w", newVGSName, err)
+		}
+	}
+	for i := range 5 {
+		newVGSName := fmt.Sprintf("%s-%d", vgsName, i)
+		err = v.DeleteVolumeGroupSnapshot(newVGSName)
+		if err != nil {
+			return fmt.Errorf("failed to delete volume group snapshot %q: %w", newVGSName, err)
+		}
+	}
+
 	clonePVCs, err := v.CreatePVCClones(volumeGroupSnapshot)
 	if err != nil {
 		return fmt.Errorf("failed to create clones: %w", err)
diff --git a/internal/rbd/controllerserver.go b/internal/rbd/controllerserver.go
index 3ec067c3ecf..a27d7e6e4e6 100644
--- a/internal/rbd/controllerserver.go
+++ b/internal/rbd/controllerserver.go
@@ -573,7 +573,7 @@ func (cs *ControllerServer) repairExistingVolume(ctx context.Context, req *csi.C
 // are more than the `minSnapshotOnImage` Add a task to flatten all the
 // temporary cloned images.
 func flattenTemporaryClonedImages(ctx context.Context, rbdVol *rbdVolume, cr *util.Credentials) error {
-	snaps, err := rbdVol.listSnapshots()
+	snaps, children, err := rbdVol.ListSnapAndChildren()
 	if err != nil {
 		if errors.Is(err, ErrImageNotFound) {
 			return status.Error(codes.InvalidArgument, err.Error())
@@ -589,9 +589,17 @@ func flattenTemporaryClonedImages(ctx context.Context, rbdVol *rbdVolume, cr *ut
 			len(snaps),
 			rbdVol,
 			maxSnapshotsOnImage)
+
+		if len(children) == 0 {
+			// if none of the child images(are in trash) exist, we can't flatten them.
+			// return ResourceExhausted error message as we have reached the hard limit.
+			log.DebugLog(ctx, "child images of image %q cannot be flatten", rbdVol)
+
+			return status.Errorf(codes.ResourceExhausted, "rbd image %q has %d snapshots but child images cannot be flattened", rbdVol, len(snaps))
+		}
 		err = flattenClonedRbdImages(
 			ctx,
-			snaps,
+			children,
 			rbdVol.Pool,
 			rbdVol.Monitors,
 			rbdVol.RbdImageName,
@@ -610,13 +618,20 @@ func flattenTemporaryClonedImages(ctx context.Context, rbdVol *rbdVolume, cr *ut
 			len(snaps),
 			rbdVol,
 			minSnapshotsOnImageToStartFlatten)
+		if len(children) == 0 {
+			// if none of the child images(are in trash) exist, we can't flatten them.
+			// return nil since we have only reach the soft limit.
+			log.DebugLog(ctx, "child images of image %q cannot be flatten", rbdVol)
+
+			return nil
+		}
 		// If we start flattening all the snapshots at one shot the volume
 		// creation time will be affected,so we will flatten only the extra
 		// snapshots.
-		snaps = snaps[minSnapshotsOnImageToStartFlatten-1:]
+		children = children[:minSnapshotsOnImageToStartFlatten]
 		err = flattenClonedRbdImages(
 			ctx,
-			snaps,
+			children,
 			rbdVol.Pool,
 			rbdVol.Monitors,
 			rbdVol.RbdImageName,
@@ -1157,7 +1172,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 +1391,6 @@ func (cs *ControllerServer) doSnapshotClone(
 		return cloneRbd, err
 	}
 
-	err = cloneRbd.flattenRbdImage(ctx, false, rbdHardMaxCloneDepth, rbdSoftMaxCloneDepth)
-	if err != nil {
-		return cloneRbd, err
-	}
-
 	return cloneRbd, nil
 }
 
diff --git a/internal/rbd/group_controllerserver.go b/internal/rbd/group_controllerserver.go
index 4e4d3c408bc..16a7cce6d92 100644
--- a/internal/rbd/group_controllerserver.go
+++ b/internal/rbd/group_controllerserver.go
@@ -130,6 +130,21 @@ 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())
+	}
+	errList := make([]error, 0)
+	for _, volume := range volumes {
+		err = volume.PrepareVolumeForSnapshot(ctx, creds)
+		if err != nil {
+			errList = append(errList, err)
+		}
+	}
+	if err != nil {
+		return nil, status.Errorf(codes.Aborted, "failed to prepare volumes for snapshot: %v", errList)
+	}
+
 	// create a temporary VolumeGroup with a different name
 	vg, err = mgr.CreateVolumeGroup(ctx, vgName)
 	if err != nil {
diff --git a/internal/rbd/manager.go b/internal/rbd/manager.go
index 8b0d6a06448..dcbcaad1004 100644
--- a/internal/rbd/manager.go
+++ b/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
 	}
diff --git a/internal/rbd/rbd_util.go b/internal/rbd/rbd_util.go
index fbae7f4181d..73847f42464 100644
--- a/internal/rbd/rbd_util.go
+++ b/internal/rbd/rbd_util.go
@@ -787,7 +787,7 @@ type trashSnapInfo struct {
 
 func flattenClonedRbdImages(
 	ctx context.Context,
-	snaps []librbd.SnapInfo,
+	children []string,
 	pool, monitors, rbdImageName string,
 	cr *util.Credentials,
 ) error {
@@ -803,26 +803,9 @@ func flattenClonedRbdImages(
 
 		return err
 	}
-	var origNameList []trashSnapInfo
-	for _, snapInfo := range snaps {
-		// check if the snapshot belongs to trash namespace.
-		isTrash, retErr := rv.isTrashSnap(snapInfo.Id)
-		if retErr != nil {
-			return retErr
-		}
-
-		if isTrash {
-			// get original snap name for the snapshot in trash namespace
-			origSnapName, retErr := rv.getOrigSnapName(snapInfo.Id)
-			if retErr != nil {
-				return retErr
-			}
-			origNameList = append(origNameList, trashSnapInfo{origSnapName})
-		}
-	}
 
-	for _, snapName := range origNameList {
-		rv.RbdImageName = snapName.origSnapName
+	for _, childName := range children {
+		rv.RbdImageName = childName
 		err = rv.flattenRbdImage(ctx, true, rbdHardMaxCloneDepth, rbdSoftMaxCloneDepth)
 		if err != nil {
 			log.ErrorLog(ctx, "failed to flatten %s; err %v", rv, err)
@@ -2048,19 +2031,26 @@ func (ri *rbdImage) DisableDeepFlatten() error {
 	return image.UpdateFeatures(librbd.FeatureDeepFlatten, false)
 }
 
-func (ri *rbdImage) listSnapshots() ([]librbd.SnapInfo, error) {
+// ListSnapAndChildren returns list of names of snapshots and child images.
+func (ri *rbdImage) ListSnapAndChildren() ([]librbd.SnapInfo, []string, error) {
 	image, err := ri.open()
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	defer image.Close()
 
-	snapInfoList, err := image.GetSnapshotNames()
+	snaps, err := image.GetSnapshotNames()
 	if err != nil {
-		return nil, err
+		return nil, nil, err
+	}
+
+	// ListChildren() returns pools, images, err.
+	_, children, err := image.ListChildren()
+	if err != nil {
+		return nil, nil, err
 	}
 
-	return snapInfoList, nil
+	return snaps, children, nil
 }
 
 // isTrashSnap returns true if the snapshot belongs to trash namespace.
@@ -2273,3 +2263,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
+}
diff --git a/internal/rbd/types/manager.go b/internal/rbd/types/manager.go
index 273bd58109f..231d4584f09 100644
--- a/internal/rbd/types/manager.go
+++ b/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)
diff --git a/internal/rbd/types/volume.go b/internal/rbd/types/volume.go
index 1f235bf16ba..a8b390b622a 100644
--- a/internal/rbd/types/volume.go
+++ b/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)