rbd: replace O(n) trash list scan with O(1) ID-based trash removal

internal/rbd/controllerserver.go

@@ -989,8 +989,7 @@ func (cs *ControllerServer) checkErrAndUndoReserve(
 	}
 
 	if errors.Is(err, rbderrors.ErrImageNotFound) {
-		notFoundErr := rbdVol.ensureImageCleanup(ctx)
-		if notFoundErr != nil {
+		if notFoundErr := rbdVol.removeImageFromTrash(ctx); notFoundErr != nil {
 			return nil, status.Errorf(codes.Internal, "failed to cleanup image %q: %v", rbdVol, notFoundErr)
 		}
 	} else {
@@ -1162,7 +1161,9 @@ func cleanupRBDImage(ctx context.Context,
 	}
 
 	// delete the temporary rbd image created as part of volume clone during
-	// create volume
+	// create volume. This must run before rbdVol.Delete() so the volume
+	// image still exists and its parent info (ParentImageID) can identify
+	// a trashed temp clone without scanning the trash list.
 	err = rbdVol.DeleteTempImage(ctx)
 	if err != nil {
 		log.ErrorLog(ctx, "failed to delete temporary rbd image: %v", err)
@@ -1641,9 +1642,9 @@ func cleanUpImageAndSnapReservation(ctx context.Context, rbdSnap *rbdSnapshot, c
 	defer rbdVol.Destroy(ctx)
 
 	// cleanup the image from trash if the error is image not found.
-	err = rbdVol.ensureImageCleanup(ctx)
+	err = rbdVol.removeImageFromTrash(ctx)
 	if err != nil {
-		log.ErrorLog(ctx, "failed to delete rbd image: %q with error: %v", rbdVol.Pool, rbdVol.VolName, err)
+		log.ErrorLog(ctx, "failed to delete rbd image %q: %v", rbdVol, err)
 
 		return status.Error(codes.Internal, err.Error())
 	}

internal/rbd/rbd_util.go

@@ -165,6 +165,9 @@ type rbdImage struct {
 
 	// ParentInTrash indicates the parent image is in trash.
 	ParentInTrash bool
+	// ParentImageID is the image ID of the parent image, populated by
+	// getImageInfo(). Valid even when the parent is in trash.
+	ParentImageID string
 
 	// RBD QoS configuration
 	QosParameters map[string]string
@@ -720,31 +723,6 @@ func isCephMgrSupported(ctx context.Context, clusterID string, err error) (bool,
 	return true, nil
 }
 
-// ensureImageCleanup finds image in trash and if found removes it
-// from trash.
-func (ri *rbdImage) ensureImageCleanup(ctx context.Context) error {
-	err := ri.openIoctx()
-	if err != nil {
-		return err
-	}
-
-	trashInfoList, err := librbd.GetTrashList(ri.ioctx)
-	if err != nil {
-		log.ErrorLog(ctx, "failed to list images in trash: %v", err)
-
-		return err
-	}
-	for _, val := range trashInfoList {
-		if val.Name == ri.RbdImageName {
-			ri.ImageID = val.Id
-
-			return ri.trashRemoveImage(ctx)
-		}
-	}
-
-	return nil
-}
-
 // Delete deletes a ceph image with provision and volume options.
 func (ri *rbdImage) Delete(ctx context.Context) error {
 	image := ri.RbdImageName
@@ -803,6 +781,11 @@ func (ri *rbdImage) trashRemoveImage(ctx context.Context) error {
 	}
 
 	_, err = ta.AddTrashRemove(admin.NewImageSpec(ri.Pool, ri.RadosNamespace, ri.ImageID))
+	if err != nil && errors.Is(err, rados.ErrNotFound) {
+		log.DebugLog(ctx, "image %s (ID %s) not found in trash, already removed", ri, ri.ImageID)
+
+		return nil
+	}
 
 	rbdCephMgrSupported, knownErr := isCephMgrSupported(ctx, ri.ClusterID, err)
 	if rbdCephMgrSupported && err != nil {
@@ -814,6 +797,12 @@ func (ri *rbdImage) trashRemoveImage(ctx context.Context) error {
 	if !rbdCephMgrSupported && knownErr != nil {
 		trashRemoveError := librbd.TrashRemove(ri.ioctx, ri.ImageID, true)
 		if trashRemoveError != nil {
+			if errors.Is(trashRemoveError, librbd.ErrNotFound) {
+				log.DebugLog(ctx, "image %s not found in trash, already removed", ri)
+
+				return nil
+			}
+
 			log.ErrorLog(ctx, "failed to delete rbd image: %s, %v", ri, trashRemoveError)
 
 			return fmt.Errorf(
@@ -829,6 +818,19 @@ func (ri *rbdImage) trashRemoveImage(ctx context.Context) error {
 	return nil
 }
 
+// removeImageFromTrash removes an image from trash by its known image ID.
+func (ri *rbdImage) removeImageFromTrash(ctx context.Context) error {
+	if ri.ImageID == "" {
+		return fmt.Errorf("image ID is empty, cannot remove %s from trash", ri)
+	}
+
+	if err := ri.openIoctx(); err != nil {
+		return err
+	}
+
+	return ri.trashRemoveImage(ctx)
+}
+
 // DeleteTempImage deletes the temporary image created for volume datasource.
 func (rv *rbdVolume) DeleteTempImage(ctx context.Context) error {
 	tempClone := rv.generateTempClone()
@@ -849,11 +851,20 @@ func (rv *rbdVolume) DeleteTempImage(ctx context.Context) error {
 	err = tempClone.Delete(ctx)
 	if err != nil {
 		if errors.Is(err, rbderrors.ErrImageNotFound) {
-			return tempClone.ensureImageCleanup(ctx)
-		} else {
-			// return error if it is not ErrImageNotFound
-			return err
+			if rv.ParentInTrash &&
+				rv.ParentName == tempClone.RbdImageName &&
+				rv.ParentImageID != "" {
+				tempClone.ImageID = rv.ParentImageID
+
+				return tempClone.removeImageFromTrash(ctx)
+			}
+
+			log.DebugLog(ctx, "temp clone %s not found and not in trash, already removed", tempClone)
+
+			return nil
 		}
+
+		return err
 	}
 
 	return nil
@@ -1814,13 +1825,17 @@ func (ri *rbdImage) getImageInfo() error {
 		// the parent is an error or not.
 		if errors.Is(err, librbd.ErrNotFound) {
 			ri.ParentName = ""
+			ri.ParentPool = ""
+			ri.ParentInTrash = false
+			ri.ParentImageID = ""
 		} else {
 			return err
 		}
 	} else {
 		ri.ParentName = parentInfo.Image.ImageName
 		ri.ParentPool = parentInfo.Image.PoolName
 		ri.ParentInTrash = parentInfo.Image.Trash
+		ri.ParentImageID = parentInfo.Image.ImageID
 	}
 	// Get image creation time
 	tm, err := image.GetCreateTimestamp()