use trash to remove RBD images asynchronously

We should RBD images asynchronously to finish each reconiliation
iteration quickly. This commit supports asynchronous removal of the RBD
images by the following two steps:

1. Run `rbd trash mv` to move the RBD image to trash.
2. Run `ceph rbd task add trash remove` to add a task to remove an image
   from trash.

cf. https://github.com/ceph/ceph-csi/blob/d8a040719e0ad491ad144fe361b67e6bd2109710/docs/design/proposals/rbd-snap-clone.md#rbd-cli-commands-to-delete-a-snapshot

Signed-off-by: Ryotaro Banno <[email protected]>

internal/ceph/ceph.go

@@ -43,6 +43,8 @@ type CephCmd interface {
 	RBDInfo(pool, image string) (*RBDImageInfo, error)
 	RBDLs(pool string) ([]string, error)
 	RBDRm(pool, image string) error
+	RBDTrashMv(pool, image string) error
+	CephRBDTaskAddTrashRemove(pool, image string) error
 	RBDSnapCreate(pool, image, snap string) error
 	RBDSnapLs(pool, image string) ([]RBDSnapshot, error)
 	RBDSnapRm(pool, image, snap string) error

internal/ceph/rbd.go

@@ -62,6 +62,24 @@ func (c *cephCmdImpl) RBDRm(pool, image string) error {
 	return nil
 }
 
+// RBDTrashMv removes an RBD image asynchronously.
+func (c *cephCmdImpl) RBDTrashMv(pool, image string) error {
+	_, err := c.command.execute("rbd", "trash", "mv", fmt.Sprintf("%s/%s", pool, image))
+	if err != nil {
+		return fmt.Errorf("failed to move RBD image to trash: %w", err)
+	}
+	return nil
+}
+
+// CephRBDTaskTrashRemove adds a task to remove the image from trash.
+func (c *cephCmdImpl) CephRBDTaskAddTrashRemove(pool, imageID string) error {
+	_, err := c.command.execute("ceph", "rbd", "task", "add", "trash", "remove", fmt.Sprintf("%s/%s", pool, imageID))
+	if err != nil {
+		return fmt.Errorf("failed to add task to remove the image from trash: %w", err)
+	}
+	return nil
+}
+
 // RBDSnapCreate creates an RBD snapshot.
 func (c *cephCmdImpl) RBDSnapCreate(pool, image, snap string) error {
 	_, err := c.command.execute("rbd", "snap", "create", fmt.Sprintf("%s/%s@%s", pool, image, snap))

internal/controller/internal/testutil/fake_rbd.go

@@ -38,6 +38,14 @@ func (f *fakeRBD) RBDRm(pool, image string) error {
 	return nil
 }
 
+func (f *fakeRBD) RBDTrashMv(pool, image string) error {
+	return nil
+}
+
+func (f *fakeRBD) CephRBDTaskAddTrashRemove(pool, image string) error {
+	return nil
+}
+
 func (f *fakeRBD) RBDSnapCreate(pool, image, snap string) error {
 	key := pool + "/" + image
 

internal/controller/persistentvolume_controller.go

@@ -132,12 +132,25 @@ func (r *PersistentVolumeReconciler) removeRBDImage(ctx context.Context, pv *cor
 
 	images, err := r.ceph.RBDLs(pool)
 	if err != nil {
-		return fmt.Errorf("failed to list RBD images: %v", err)
+		return fmt.Errorf("failed to list RBD images: %w", err)
 	}
 
 	if !slices.Contains(images, image) {
 		return nil
 	}
 
-	return r.ceph.RBDRm(pool, image)
+	imageInfo, err := r.ceph.RBDInfo(pool, image)
+	if err != nil {
+		return fmt.Errorf("failed to get info about the RBD image: %s/%s: %w", pool, image, err)
+	}
+
+	if err := r.ceph.RBDTrashMv(pool, image); err != nil {
+		return fmt.Errorf("failed to move the RBD image to trash: %s/%s: %w", pool, image, err)
+	}
+
+	if err := r.ceph.CephRBDTaskAddTrashRemove(pool, imageInfo.ID); err != nil {
+		return fmt.Errorf("failed to add task to remove the RBD image from trash: %s/%s: %w", pool, image, err)
+	}
+
+	return nil
 }