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: #4013
Signed-off-by: Niels de Vos <[email protected]>

internal/rbd/rbd_util.go

@@ -697,47 +697,69 @@ func (ri *rbdImage) trashRemoveImage(ctx context.Context) error {
 	return nil
 }
 
+func getImageSpec(spec librbd.ImageSpec) string {
+	if spec.Trash {
+		return spec.ImageID
+	}
+
+	if spec.PoolNamespace != "" {
+		return fmt.Sprintf("%s/%s/%s", spec.PoolName, spec.PoolNamespace, spec.ImageName)
+	}
+
+	return fmt.Sprintf("%s/%s", spec.PoolName, spec.ImageName)
+}
+
+// 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(ctx context.Context) (uint, error) {
-	var depth uint
-	vol := rbdVolume{}
+	var (
+		depth  uint
+		parent librbd.ImageSpec
+	)
 
-	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)
+	}
+	defer image.Close()
 
 	for {
-		if vol.RbdImageName == "" {
-			return depth, nil
-		}
-		err := vol.openIoctx()
+		// get the librbd.ImageSpec of the parent
+		info, err := image.GetParent()
 		if err != nil {
-			return depth, err
+			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)
+		parent = info.Image
+		if parent.ImageID == "" {
+			break
+		}
 
-			return depth, err
+		// of there is a parent, count it to the depth
+		depth++
+
+		// open the parent image, so that the for-loop can continue
+		// with checking for the parent of the parent
+
+		// if the parent is in the trash, the image must be opened with
+		// OpenImageById()
+		if parent.Trash {
+			image, err = librbd.OpenImageById(ri.ioctx, parent.ImageID, librbd.NoSnapshot)
+		} else {
+			image, err = librbd.OpenImage(ri.ioctx, getImageSpec(parent), librbd.NoSnapshot)
 		}
-		if vol.ParentName != "" {
-			depth++
+
+		if err != nil {
+			return 0, fmt.Errorf("failed to open parent image at depth %d: %w", depth, err)
 		}
-		vol.RbdImageName = vol.ParentName
-		vol.Pool = vol.ParentPool
+
+		defer image.Close()
 	}
+
+	return depth, nil
 }
 
 type trashSnapInfo struct {