rbd: use ListChildrenAttributes() instead of ListChildren()

This commit modifies listSnapAndChildren() to make use of
ListChildrenAttributes() instead of ListChildren() which
allows us to filter out images in trash.
This commit also order the alive images so that temp clone
images are followed by images backing volume snapshots so
that temp clone images are flattened first.

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

internal/rbd/clone.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 
 	"github.com/ceph/ceph-csi/internal/util"
 	"github.com/ceph/ceph-csi/internal/util/k8s"
@@ -122,7 +123,7 @@ func (rv *rbdVolume) generateTempClone() *rbdVolume {
 	// The temp cloned image name will be always (rbd image name + "-temp")
 	// this name will be always unique, as cephcsi never creates an image with
 	// this format for new rbd images
-	tempClone.RbdImageName = rv.RbdImageName + "-temp"
+	tempClone.RbdImageName = rv.RbdImageName + tempImageSuffix
 
 	return &tempClone
 }
@@ -250,3 +251,9 @@ func (rv *rbdVolume) doSnapClone(ctx context.Context, parentVol *rbdVolume) erro
 
 	return nil
 }
+
+// isTempClonedImage checks whether the image is a temporary cloned image
+// by checking the suffix of the image name.
+func isTempClonedImage(imageName string) bool {
+	return strings.HasSuffix(imageName, tempImageSuffix)
+}

internal/rbd/controllerserver.go

@@ -590,7 +590,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, children, err := rbdVol.listSnapAndChildren()
+	snapAndChildrenInfo, err := rbdVol.listSnapAndChildren()
 	if err != nil {
 		if errors.Is(err, util.ErrImageNotFound) {
 			return status.Error(codes.InvalidArgument, err.Error())
@@ -598,6 +598,17 @@ func flattenTemporaryClonedImages(ctx context.Context, rbdVol *rbdVolume, cr *ut
 
 		return status.Error(codes.Internal, err.Error())
 	}
+	children := make([]string, 0)
+	// order the temp clone images first followed by the volume snapshot
+	// images so that the temp clone images are flattened first.
+	// This is done in order to:
+	// - increase number of snapshots which can be supported for
+	// changed block tracking(rbd snap diff).
+	// - favor scalability since multiple PVCs can be restored from same
+	// volumesnapshot versus just one PVC-PVC clone.
+	children = append(children, snapAndChildrenInfo.TempCloneChildren...)
+	children = append(children, snapAndChildrenInfo.VolumeSnapshotChildren...)
+	snaps := snapAndChildrenInfo.SnapInfoList
 
 	if len(snaps) > int(maxSnapshotsOnImage) {
 		log.DebugLog(

internal/rbd/rbd_util.go

@@ -84,6 +84,10 @@ const (
 
 	// clusterNameKey cluster Key, set on RBD image.
 	clusterNameKey = "csi.ceph.com/cluster/name"
+
+	// Suffix added to the temp cloned image name.
+	// This will always be (rbd image name + "-temp").
+	tempImageSuffix = "-temp"
 )
 
 // rbdImage contains common attributes and methods for the rbdVolume and
@@ -2025,26 +2029,49 @@ func (ri *rbdImage) DisableDeepFlatten() error {
 	return image.UpdateFeatures(librbd.FeatureDeepFlatten, false)
 }
 
-// listSnapAndChildren returns list of names of snapshots and child images.
-func (ri *rbdImage) listSnapAndChildren() ([]librbd.SnapInfo, []string, error) {
+type snapAndChildrenInfo struct {
+	SnapInfoList           []librbd.SnapInfo
+	TempCloneChildren      []string
+	VolumeSnapshotChildren []string
+}
+
+// listSnapAndChildren returns list of snapshot names, volume snapshot images and
+// child temp clone images. Only child images which are not in trash are returned.
+func (ri *rbdImage) listSnapAndChildren() (*snapAndChildrenInfo, error) {
 	image, err := ri.open()
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 	defer image.Close()
 
 	snaps, err := image.GetSnapshotNames()
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
-	// ListChildren() returns pools, images, err.
-	_, children, err := image.ListChildren()
+	childImageSpecList, err := image.ListChildrenAttributes()
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
-	return snaps, children, nil
+	tempCloneChildren := make([]string, 0)
+	volSnapChildren := make([]string, 0)
+	for _, child := range childImageSpecList {
+		if child.Trash {
+			continue
+		}
+		if isTempClonedImage(child.ImageName) {
+			tempCloneChildren = append(tempCloneChildren, child.ImageName)
+		} else {
+			volSnapChildren = append(volSnapChildren, child.ImageName)
+		}
+	}
+
+	return &snapAndChildrenInfo{
+		SnapInfoList:           snaps,
+		TempCloneChildren:      tempCloneChildren,
+		VolumeSnapshotChildren: volSnapChildren,
+	}, nil
 }
 
 func (ri *rbdImage) isCompatibleEncryption(dst *rbdImage) error {