rbd: add EnsureImageCleanup to ensure image cleanup from trash

After moving moving image to trash, if `trash remove` step fails,
then external-provisioner will issue subsequent requests, in which
image will be absent in pool( will be in trash) and omap cleanup will
be done with stale image left in trash with no `trash remove` step on it.

To avoid this scenario list trash images and find corresponding id for given
image name and add a task to flatten when we encounter a ErrImageNotFound.

Fixes: #1728

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

internal/rbd/controllerserver.go

@@ -776,8 +776,13 @@ func (cs *ControllerServer) checkErrAndUndoReserve(
 		return &csi.DeleteVolumeResponse{}, nil
 	}
 
-	// All errors other than ErrImageNotFound should return an error back to the caller
+	if errors.Is(err, ErrImageNotFound) {
-	if !errors.Is(err, ErrImageNotFound) {
+		err = rbdVol.ensureImageCleanup(ctx)
+		if err != nil {
+			return nil, status.Error(codes.Internal, err.Error())
+		}
+	} else {
+		// All errors other than ErrImageNotFound should return an error back to the caller
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
@@ -924,8 +929,13 @@ func cleanupRBDImage(ctx context.Context,
 	tempClone := rbdVol.generateTempClone()
 	err = deleteImage(ctx, tempClone, cr)
 	if err != nil {
-		// return error if it is not ErrImageNotFound
+		if errors.Is(err, ErrImageNotFound) {
-		if !errors.Is(err, ErrImageNotFound) {
+			err = tempClone.ensureImageCleanup(ctx)
+			if err != nil {
+				return nil, status.Error(codes.Internal, err.Error())
+			}
+		} else {
+			// return error if it is not ErrImageNotFound
 			log.ErrorLog(ctx, "failed to delete rbd image: %s with error: %v",
 				tempClone, err)
 

internal/rbd/rbd_util.go

@@ -565,6 +565,26 @@ func (rv *rbdVolume) getTrashPath() string {
 	return trashPath + "/" + rv.ImageID
 }
 
+// ensureImageCleanup finds image in trash and if found removes it
+// from trash.
+func (rv *rbdVolume) ensureImageCleanup(ctx context.Context) error {
+	trashInfoList, err := librbd.GetTrashList(rv.ioctx)
+	if err != nil {
+		log.ErrorLog(ctx, "failed to list images in trash: %v", err)
+
+		return err
+	}
+	for _, val := range trashInfoList {
+		if val.Name == rv.RbdImageName {
+			rv.ImageID = val.Id
+
+			return trashRemoveImage(ctx, rv, rv.conn.Creds)
+		}
+	}
+
+	return nil
+}
+
 // deleteImage deletes a ceph image with provision and volume options.
 func deleteImage(ctx context.Context, pOpts *rbdVolume, cr *util.Credentials) error {
 	image := pOpts.RbdImageName
@@ -597,6 +617,12 @@ func deleteImage(ctx context.Context, pOpts *rbdVolume, cr *util.Credentials) er
 		return err
 	}
 
+	return trashRemoveImage(ctx, pOpts, cr)
+}
+
+// trashRemoveImage adds a task to trash remove an image using ceph manager if supported,
+// otherwise removes the image from trash.
+func trashRemoveImage(ctx context.Context, pOpts *rbdVolume, cr *util.Credentials) error {
 	// attempt to use Ceph manager based deletion support if available
 
 	args := []string{