rbd: include trashed parent images while calculating the clone depth

The `getCloneDepth()` function did not account for images that are in
the trash. A trashed image can only be opened by the image-id, and not
by name anymore.

Closes: ceph#4013
Signed-off-by: Niels de Vos <ndevos@ibm.com>

internal/rbd/nodeserver.go

@@ -598,7 +598,7 @@ func flattenImageBeforeMapping(
 		if err != nil {
 			return err
 		}
-		depth, err = volOptions.getCloneDepth(ctx)
+		depth, err = volOptions.getCloneDepth()
 		if err != nil {
 			return err
 		}

internal/rbd/rbd_util.go

@@ -697,47 +697,58 @@ func (ri *rbdImage) trashRemoveImage(ctx context.Context) error {
 	return nil
 }
 
-func (ri *rbdImage) getCloneDepth(ctx context.Context) (uint, error) {
-	var depth uint
-	vol := rbdVolume{}
+// getCloneDepth walks the parents of the image and returns the number of
+// images in the chain.
+//
+// This function re-uses the ioctx of the image to open all images in the
+// chain. There is no need to open new ioctx's for every image.
+func (ri *rbdImage) getCloneDepth() (uint, error) {
+	var (
+		depth  uint
+		parent librbd.ImageSpec
+		info   *librbd.ParentInfo
+	)
 
-	vol.Pool = ri.Pool
-	vol.Monitors = ri.Monitors
-	vol.RbdImageName = ri.RbdImageName
-	vol.RadosNamespace = ri.RadosNamespace
-	vol.conn = ri.conn.Copy()
+	image, err := ri.open()
+	if err != nil {
+		return 0, fmt.Errorf("failed to open image %s: %w", ri, err)
+	}
+
+	// get the librbd.ImageSpec of the parent
+	info, err = image.GetParent()
+
+	// Close this image, it is not used anymore. Using defer to close it
+	// and replacing the image with an other image can result in resource
+	// leaks according to golangci-lint.
+	image.Close()
 
 	for {
-		if vol.RbdImageName == "" {
-			return depth, nil
-		}
-		err := vol.openIoctx()
-		if err != nil {
-			return depth, err
+		if errors.Is(err, librbd.ErrNotFound) {
+			// image does not have more parents
+			break
+		} else if err != nil {
+			return 0, fmt.Errorf("failed to get parent of image %s: %w", ri, err)
 		}
 
-		err = vol.getImageInfo()
-		// FIXME: create and destroy the vol inside the loop.
-		// see https://github.com/ceph/ceph-csi/pull/1838#discussion_r598530807
-		vol.ioctx.Destroy()
-		vol.ioctx = nil
-		if err != nil {
-			// if the parent image is moved to trash the name will be present
-			// in rbd image info but the image will be in trash, in that case
-			// return the found depth
-			if errors.Is(err, ErrImageNotFound) {
-				return depth, nil
-			}
-			log.ErrorLog(ctx, "failed to check depth on image %s: %s", &vol, err)
+		// if there is a parent, count it to the depth
+		depth++
 
-			return depth, err
-		}
-		if vol.ParentName != "" {
-			depth++
+		// open the parent image, so that the for-loop can continue
+		// with checking for the parent of the parent
+		parent = info.Image
+		image, err = librbd.OpenImageById(ri.ioctx, parent.ImageID, librbd.NoSnapshot)
+		if err != nil {
+			return 0, fmt.Errorf("failed to open parent image ID %q, at depth %d: %w", parent.ImageID, depth, err)
 		}
-		vol.RbdImageName = vol.ParentName
-		vol.Pool = vol.ParentPool
+		info, err = image.GetParent()
+		// info and err checking is done in the top of the for loop
+
+		// Using defer in a for loop seems to be problematic. Just
+		// always close the image.
+		image.Close()
 	}
+
+	return depth, nil
 }
 
 type trashSnapInfo struct {
@@ -803,7 +814,7 @@ func (ri *rbdImage) flattenRbdImage(
 
 	// skip clone depth check if request is for force flatten
 	if !forceFlatten {
-		depth, err = ri.getCloneDepth(ctx)
+		depth, err = ri.getCloneDepth()
 		if err != nil {
 			return err
 		}