From 8fc9146056b9c0194e6f8d5c13e3faf8b782727a Mon Sep 17 00:00:00 2001
From: Madhu Rajanna <madhupr007@gmail.com>
Date: Sun, 12 Jul 2020 10:06:17 +0530
Subject: [PATCH] rbd: flatten temp cloned images

If the snapshots on the parent image exceeds
maxSnapshotsOnImage count, we need to flatten
all the temporary cloned images to over come the
krbd issue of maximum number of snapshots on
an image.

Signed-off-by: Madhu Rajanna <madhupr007@gmail.com>
---
 internal/rbd/controllerserver.go | 56 +++++++++++++++++++++++++++-----
 1 file changed, 48 insertions(+), 8 deletions(-)

diff --git a/internal/rbd/controllerserver.go b/internal/rbd/controllerserver.go
index 05f028b46..c267e3b93 100644
--- a/internal/rbd/controllerserver.go
+++ b/internal/rbd/controllerserver.go
@@ -227,15 +227,11 @@ func (cs *ControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 		return buildCreateVolumeResponse(ctx, req, rbdVol)
 	}
 
-	if parentVol != nil {
-		// flatten the image or its parent before the reservation to avoid
-		// stale entries in post creation if we return ABORT error and the
-		// delete volume is not called
-		err = parentVol.flattenCloneImage(ctx)
-		if err != nil {
-			return nil, getGRPCErrorForCreateVolume(err)
-		}
+	err = flattenParentImage(ctx, parentVol, cr)
+	if err != nil {
+		return nil, err
 	}
+
 	err = reserveVol(ctx, rbdVol, rbdSnap, cr)
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
@@ -282,6 +278,50 @@ func (cs *ControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 	return &csi.CreateVolumeResponse{Volume: volume}, nil
 }
 
+func flattenParentImage(ctx context.Context, rbdVol *rbdVolume, cr *util.Credentials) error {
+	if rbdVol != nil {
+		// flatten the image or its parent before the reservation to avoid
+		// stale entries in post creation if we return ABORT error and the
+		// delete volume is not called
+		err := rbdVol.flattenCloneImage(ctx)
+		if err != nil {
+			return getGRPCErrorForCreateVolume(err)
+		}
+
+		// flatten cloned images if the snapshot count on the parent image
+		// exceeds maxSnapshotsOnImage
+		err = flattenTemporaryClonedImages(ctx, rbdVol, cr)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// check snapshots on the rbd image, as we have limit from krbd that
+// an image cannot have more than 510 snapshot at a given point of time.
+// If the snapshots are more than the `maxSnapshotsOnImage` 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(ctx, cr)
+	if err != nil {
+		var einf ErrImageNotFound
+		if errors.As(err, &einf) {
+			return status.Error(codes.InvalidArgument, err.Error())
+		}
+		return status.Error(codes.Internal, err.Error())
+	}
+
+	if len(snaps) > int(maxSnapshotsOnImage) {
+		err = flattenClonedRbdImages(ctx, snaps, rbdVol.Pool, rbdVol.Monitors, cr)
+		if err != nil {
+			return status.Error(codes.Internal, err.Error())
+		}
+		return status.Errorf(codes.ResourceExhausted, "rbd image %s has %d snapshots", rbdVol, len(snaps))
+	}
+	return nil
+}
+
 // checkFlatten ensures that that the image chain depth is not reached
 // hardlimit or softlimit. if the softlimit is reached it adds a task and
 // return success,the hardlimit is reached it starts a task to flatten the