Merge pull request #14026 from sp98/resource-cleanup

core: subvolumegroup clean up

Documentation/Storage-Configuration/ceph-teardown.md

@@ -2,6 +2,11 @@
 title: Cleanup
 ---
 
+Rook provides the following clean up options:
+
+1. [Uninstall: Clean up the entire cluster and delete all data](#cleaning-up-a-cluster)
+1. [Force delete individual resources](#force-delete-resources)
+
 ## Cleaning up a Cluster
 
 To tear down the cluster, the following resources need to be cleaned up:
@@ -179,3 +184,22 @@ If the operator is not able to remove the finalizers (i.e., the operator is not
 kubectl -n rook-ceph patch configmap rook-ceph-mon-endpoints --type merge -p '{"metadata":{"finalizers": []}}'
 kubectl -n rook-ceph patch secrets rook-ceph-mon --type merge -p '{"metadata":{"finalizers": []}}'
 ```
+
+## Force Delete Resources
+
+To keep your data safe in the cluster, Rook disallows deleting critical cluster resources by default. To override this behavior and force delete a specific custom resource, add the annotation `rook.io/force-deletion="true"` to the resource and then delete it. Rook will start a cleanup job that will delete all the related ceph resources created by that custom resource.
+
+For example, run the following commands to clean the `CephFilesystemSubVolumeGroup` resource named `my-subvolumegroup`
+
+``` console
+kubectl -n rook-ceph annotate cephfilesystemsubvolumegroups.ceph.rook.io my-subvolumegroup rook.io/force-deletion="true"
+kubectl -n rook-ceph delete cephfilesystemsubvolumegroups.ceph.rook.io my-subvolumegroup
+```
+
+Once the cleanup job is completed successfully, Rook will remove the finalizers from the deleted custom resource.
+
+This cleanup is supported only for the following custom resources:
+
+| Custom Resource                                | Ceph Resources to be cleaned up |
+| --------                                       | ------- |
+| CephFilesystemSubVolumeGroup                   | CSI stored RADOS OMAP details for pvc/volumesnapshots, subvolume snapshots, subvolume clones, subvolumes |

cmd/rook/ceph/cleanup.go

@@ -17,12 +17,14 @@ limitations under the License.
 package ceph
 
 import (
+	"fmt"
 	"os"
 
 	"github.com/rook/rook/cmd/rook/rook"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	cleanup "github.com/rook/rook/pkg/daemon/ceph/cleanup"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
+	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/rook/rook/pkg/util/flags"
 	"github.com/spf13/cobra"
@@ -40,22 +42,39 @@ var (
 
 var cleanUpCmd = &cobra.Command{
 	Use:   "clean",
-	Short: "Starts the cleanup process on the disks after ceph cluster is deleted",
+	Short: "Starts the cleanup process",
+}
+
+var cleanUpHostCmd = &cobra.Command{
+	Use:   "host",
+	Short: "Starts the cleanup process on a host after the ceph cluster is deleted",
+}
+
+var cleanUpSubVolumeGroupCmd = &cobra.Command{
+	// the subcommand matches CRD kind of the custom resource to be cleaned up
+	Use:   "CephFilesystemSubVolumeGroup",
+	Short: "Starts the cleanup process of a CephFilesystemSubVolumeGroup",
 }
 
 func init() {
 	cleanUpCmd.Flags().StringVar(&dataDirHostPath, "data-dir-host-path", "", "dataDirHostPath on the node")
 	cleanUpCmd.Flags().StringVar(&namespaceDir, "namespace-dir", "", "dataDirHostPath on the node")
-	cleanUpCmd.Flags().StringVar(&monSecret, "mon-secret", "", "monitor secret from the keyring")
-	cleanUpCmd.Flags().StringVar(&clusterFSID, "cluster-fsid", "", "ceph cluster fsid")
-	cleanUpCmd.Flags().StringVar(&sanitizeMethod, "sanitize-method", string(cephv1.SanitizeMethodQuick), "sanitize method to use (metadata or data)")
-	cleanUpCmd.Flags().StringVar(&sanitizeDataSource, "sanitize-data-source", string(cephv1.SanitizeDataSourceZero), "data source to sanitize the disk (zero or random)")
-	cleanUpCmd.Flags().Int32Var(&sanitizeIteration, "sanitize-iteration", 1, "overwrite N times the disk")
-	flags.SetFlagsFromEnv(cleanUpCmd.Flags(), rook.RookEnvVarPrefix)
-	cleanUpCmd.RunE = startCleanUp
+	cleanUpHostCmd.Flags().StringVar(&monSecret, "mon-secret", "", "monitor secret from the keyring")
+	cleanUpHostCmd.Flags().StringVar(&clusterFSID, "cluster-fsid", "", "ceph cluster fsid")
+	cleanUpHostCmd.Flags().StringVar(&sanitizeMethod, "sanitize-method", string(cephv1.SanitizeMethodQuick), "sanitize method to use (metadata or data)")
+	cleanUpHostCmd.Flags().StringVar(&sanitizeDataSource, "sanitize-data-source", string(cephv1.SanitizeDataSourceZero), "data source to sanitize the disk (zero or random)")
+	cleanUpHostCmd.Flags().Int32Var(&sanitizeIteration, "sanitize-iteration", 1, "overwrite N times the disk")
+	flags.SetFlagsFromEnv(cleanUpHostCmd.Flags(), rook.RookEnvVarPrefix)
+
+	flags.SetFlagsFromEnv(cleanUpSubVolumeGroupCmd.Flags(), rook.RookEnvVarPrefix)
+
+	cleanUpCmd.AddCommand(cleanUpHostCmd, cleanUpSubVolumeGroupCmd)
+
+	cleanUpHostCmd.RunE = startHostCleanUp
+	cleanUpSubVolumeGroupCmd.RunE = startSubVolumeGroupCleanUp
 }
 
-func startCleanUp(cmd *cobra.Command, args []string) error {
+func startHostCleanUp(cmd *cobra.Command, args []string) error {
 	rook.SetLogLevel()
 	rook.LogStartupInfo(cleanUpCmd.Flags())
 
@@ -87,3 +106,37 @@ func startCleanUp(cmd *cobra.Command, args []string) error {
 
 	return nil
 }
+
+func startSubVolumeGroupCleanUp(cmd *cobra.Command, args []string) error {
+	rook.SetLogLevel()
+	rook.LogStartupInfo(cleanUpSubVolumeGroupCmd.Flags())
+
+	ctx := cmd.Context()
+	context := createContext()
+	namespace := os.Getenv(k8sutil.PodNamespaceEnvVar)
+	clusterInfo := client.AdminClusterInfo(ctx, namespace, "")
+
+	fsName := os.Getenv(opcontroller.CephFSNameEnv)
+	if fsName == "" {
+		rook.TerminateFatal(fmt.Errorf("ceph filesystem name is not available in the pod environment variables"))
+	}
+	subVolumeGroupName := os.Getenv(opcontroller.CephFSSubVolumeGroupNameEnv)
+	if subVolumeGroupName == "" {
+		rook.TerminateFatal(fmt.Errorf("cephFS SubVolumeGroup name is not available in the pod environment variables"))
+	}
+	csiNamespace := os.Getenv(opcontroller.CSICephFSRadosNamesaceEnv)
+	if csiNamespace == "" {
+		rook.TerminateFatal(fmt.Errorf("CSI rados namespace name is not available in the pod environment variables"))
+	}
+	poolName := os.Getenv(opcontroller.CephFSMetaDataPoolNameEnv)
+	if poolName == "" {
+		rook.TerminateFatal(fmt.Errorf("cephFS metadata pool name is not available in the pod environment variables"))
+	}
+
+	err := cleanup.SubVolumeGroupCleanup(context, clusterInfo, fsName, subVolumeGroupName, poolName, csiNamespace)
+	if err != nil {
+		rook.TerminateFatal(fmt.Errorf("failed to cleanup cephFS %q SubVolumeGroup %q in the namespace %q. %v", fsName, subVolumeGroupName, namespace, err))
+	}
+
+	return nil
+}

pkg/daemon/ceph/cleanup/subvolumegroups.go

@@ -0,0 +1,143 @@
+/*
+Copyright 2024 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cleanup
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/pkg/errors"
+	"github.com/rook/rook/pkg/clusterd"
+	"github.com/rook/rook/pkg/daemon/ceph/client"
+	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
+)
+
+func SubVolumeGroupCleanup(context *clusterd.Context, clusterInfo *client.ClusterInfo, fsName, svg, poolName, csiNamespace string) error {
+	logger.Infof("starting clean up cephFS subVolumeGroup resource %q", svg)
+
+	subVolumeList, err := cephclient.ListSubvolumesInGroup(context, clusterInfo, fsName, svg)
+	if err != nil {
+		return errors.Wrapf(err, "failed to list cephFS subVolumes in subVolumeGroup %q", svg)
+	}
+
+	if len(subVolumeList) == 0 {
+		logger.Infof("no subvolumes found in cephFS subVolumeGroup %q", svg)
+		return nil
+	}
+
+	var retErr error
+	for _, subVolume := range subVolumeList {
+		logger.Infof("starting clean up of subvolume %q", subVolume.Name)
+		err := CleanUpOMAPDetails(context, clusterInfo, subVolume.Name, poolName, csiNamespace)
+		if err != nil {
+			retErr = errors.Wrapf(err, "failed to clean up OMAP details for the subvolume %q.", subVolume.Name)
+			logger.Error(retErr)
+		}
+		subVolumeSnapshots, err := cephclient.ListSubVolumeSnapshots(context, clusterInfo, fsName, subVolume.Name, svg)
+		if err != nil {
+			retErr = errors.Wrapf(err, "failed to list snapshots for subvolume %q in group %q.", subVolume.Name, svg)
+			logger.Error(retErr)
+		} else {
+			err := CancelPendingClones(context, clusterInfo, subVolumeSnapshots, fsName, subVolume.Name, svg)
+			if err != nil {
+				retErr = errors.Wrapf(err, "failed to cancel pending clones for subvolume %q in group %q", subVolume.Name, svg)
+				logger.Error(retErr)
+			}
+			err = DeleteSubVolumeSnapshots(context, clusterInfo, subVolumeSnapshots, fsName, subVolume.Name, svg)
+			if err != nil {
+				retErr = errors.Wrapf(err, "failed to delete snapshots for subvolume %q in group %q", subVolume.Name, svg)
+				logger.Error(retErr)
+			}
+		}
+		err = cephclient.DeleteSubVolume(context, clusterInfo, fsName, subVolume.Name, svg)
+		if err != nil {
+			retErr = errors.Wrapf(err, "failed to delete subvolume group %q.", subVolume.Name)
+			logger.Error(retErr)
+		}
+	}
+
+	if retErr != nil {
+		return errors.Wrapf(err, "clean up for cephFS subVolumeGroup %q didn't complete successfully.", svg)
+	}
+	logger.Infof("successfully cleaned up cephFS subVolumeGroup %q", svg)
+	return nil
+}
+
+func CancelPendingClones(context *clusterd.Context, clusterInfo *client.ClusterInfo, snapshots cephclient.SubVolumeSnapshots, fsName, subvol, svg string) error {
+	for _, snapshot := range snapshots {
+		logger.Infof("deleting any pending clones of snapshot %q of subvolume %q of group %q", snapshot.Name, subvol, svg)
+		pendingClones, err := cephclient.ListSubVolumeSnapshotPendingClones(context, clusterInfo, fsName, subvol, snapshot.Name, svg)
+		if err != nil {
+			return errors.Wrapf(err, "failed to list all the pending clones for snapshot %q", snapshot.Name)
+		}
+		for _, pendingClone := range pendingClones.Clones {
+			err := cephclient.CancelSnapshotClone(context, clusterInfo, fsName, svg, pendingClone.Name)
+			if err != nil {
+				return errors.Wrapf(err, "failed to cancel the pending clone %q for snapshot %q", pendingClone.Name, snapshot.Name)
+			}
+		}
+	}
+	return nil
+}
+
+func DeleteSubVolumeSnapshots(context *clusterd.Context, clusterInfo *client.ClusterInfo, snapshots cephclient.SubVolumeSnapshots, fsName, subvol, svg string) error {
+	for _, snapshot := range snapshots {
+		logger.Infof("deleting snapshot %q for subvolume %q in group %q", snapshot.Name, subvol, svg)
+		err := cephclient.DeleteSubvolumeSnapshot(context, clusterInfo, fsName, subvol, svg, snapshot.Name)
+		if err != nil {
+			return errors.Wrapf(err, "failed to delete snapshot %q for subvolume %q in group %q. %v", snapshot.Name, subvol, svg, err)
+		}
+		logger.Infof("successfully deleted snapshot %q for subvolume %q in group %q", snapshot.Name, subvol, svg)
+	}
+	return nil
+}
+
+func CleanUpOMAPDetails(context *clusterd.Context, clusterInfo *client.ClusterInfo, objName, poolName, namespace string) error {
+	omapValue := getOMAPValue(objName)
+	if omapValue == "" {
+		return errors.New(fmt.Sprintf("failed to get OMAP value for object %q", objName))
+	}
+	logger.Infof("OMAP value for the object %q is %q", objName, omapValue)
+	omapKey, err := cephclient.GetOMAPKey(context, clusterInfo, omapValue, poolName, namespace)
+	if err != nil {
+		return errors.Wrapf(err, "failed to get OMAP key for omapObj %q. %v", omapValue, err)
+	}
+	logger.Infof("OMAP key for the OIMAP value %q is %q", omapValue, omapKey)
+
+	// delete OMAP details
+	err = cephclient.DeleteOmapValue(context, clusterInfo, omapValue, poolName, namespace)
+	if err != nil {
+		return errors.Wrapf(err, "failed to delete OMAP value %q. %v", omapValue, err)
+	}
+	if omapKey != "" {
+		err = cephclient.DeleteOmapKey(context, clusterInfo, omapKey, poolName, namespace)
+		if err != nil {
+			return errors.Wrapf(err, "failed to delete OMAP key %q. %v", omapKey, err)
+		}
+	}
+	return nil
+}
+
+func getOMAPValue(subVol string) string {
+	splitSubvol := strings.SplitAfterN(subVol, "-", 3)
+	if len(splitSubvol) < 3 {
+		return ""
+	}
+	subvol_id := splitSubvol[len(splitSubvol)-1]
+	omapval := "csi.volume." + subvol_id
+	return omapval
+}

pkg/daemon/ceph/cleanup/subvolumegroups_test.go

@@ -0,0 +1,151 @@
+/*
+Copyright 2024 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cleanup
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/rook/rook/pkg/clusterd"
+	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
+	exectest "github.com/rook/rook/pkg/util/exec/test"
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	mockSubVolumeListResp = `[{"name":"csi-vol-9d942d45-6f48-4d9e-a707-edfcd602bd89"}]`
+	mockGetOmapValResp    = `Writing to /dev/stdout
+pvc-e2907819-b005-4582-97ec-8fa1b277f46fsh`
+	mockSubvolumeSnapshotsResp    = `[{"name":"snap0"}]`
+	mockSubvolumeSnapshotInfoResp = `{"created_at":"2024-04-0808:46:36.267888","data_pool":"myfs-replicated","has_pending_clones":"yes","pending_clones":[{"name":"clone3"}]}`
+)
+
+func TestSubVolumeGroupCleanup(t *testing.T) {
+	clusterInfo := cephclient.AdminTestClusterInfo("mycluster")
+	fsName := "myfs"
+	subVolumeGroupName := "csi"
+	poolName := "myfs-metadata"
+	csiNamespace := "csi"
+
+	t.Run("no subvolumes in subvolumegroup", func(t *testing.T) {
+		executor := &exectest.MockExecutor{}
+		executor.MockExecuteCommandWithTimeout = func(timeout time.Duration, command string, args ...string) (string, error) {
+			logger.Infof("Command: %s %v", command, args)
+			if args[0] == "fs" && args[1] == "subvolume" && args[2] == "ls" {
+				assert.Equal(t, fsName, args[3])
+				assert.Equal(t, subVolumeGroupName, args[4])
+				return "[]", nil
+			}
+			return "", errors.New("unknown command")
+		}
+		context := &clusterd.Context{Executor: executor}
+		err := SubVolumeGroupCleanup(context, clusterInfo, fsName, subVolumeGroupName, poolName, csiNamespace)
+		assert.NoError(t, err)
+	})
+
+	t.Run("subvolumes with snapshots and pending clones", func(t *testing.T) {
+		executor := &exectest.MockExecutor{}
+		executor.MockExecuteCommandWithTimeout = func(timeout time.Duration, command string, args ...string) (string, error) {
+			logger.Infof("Command: %s %v", command, args)
+			// list all subvolumes in subvolumegroup
+			if args[0] == "fs" && args[1] == "subvolume" && args[2] == "ls" {
+				assert.Equal(t, fsName, args[3])
+				assert.Equal(t, subVolumeGroupName, args[4])
+				return mockSubVolumeListResp, nil
+			}
+			if args[0] == "getomapval" {
+				assert.Equal(t, "csi.volume.9d942d45-6f48-4d9e-a707-edfcd602bd89", args[1])
+				assert.Equal(t, "csi.volname", args[2])
+				assert.Equal(t, poolName, args[4])
+				assert.Equal(t, csiNamespace, args[6])
+				return mockGetOmapValResp, nil
+			}
+			// delete OMAP value
+			if args[0] == "rm" && args[1] == "csi.volume.9d942d45-6f48-4d9e-a707-edfcd602bd89" {
+				assert.Equal(t, "-p", args[2])
+				assert.Equal(t, poolName, args[3])
+				assert.Equal(t, "--namespace", args[4])
+				assert.Equal(t, csiNamespace, args[5])
+				return "", nil
+			}
+			// delete OMAP key
+			if args[0] == "rmomapkey" && args[1] == "csi.volumes.default" {
+				assert.Equal(t, "ceph.volume.pvc-e2907819-b005-4582-97ec-8fa1b277f46fsh", args[2])
+				assert.Equal(t, "-p", args[3])
+				assert.Equal(t, poolName, args[4])
+				assert.Equal(t, "--namespace", args[5])
+				assert.Equal(t, csiNamespace, args[6])
+				return "", nil
+			}
+			// list all snapshots in a subvolume
+			if args[0] == "fs" && args[1] == "subvolume" && args[2] == "snapshot" && args[3] == "ls" {
+				assert.Equal(t, fsName, args[4])
+				assert.Equal(t, "csi-vol-9d942d45-6f48-4d9e-a707-edfcd602bd89", args[5])
+				assert.Equal(t, "--group_name", args[6])
+				assert.Equal(t, subVolumeGroupName, args[7])
+				return mockSubvolumeSnapshotsResp, nil
+			}
+			// list all pending clones in a subvolume snapshot
+			if args[0] == "fs" && args[1] == "subvolume" && args[2] == "snapshot" && args[3] == "info" {
+				assert.Equal(t, fsName, args[4])
+				assert.Equal(t, "csi-vol-9d942d45-6f48-4d9e-a707-edfcd602bd89", args[5])
+				assert.Equal(t, "snap0", args[6])
+				assert.Equal(t, "--group_name", args[7])
+				assert.Equal(t, subVolumeGroupName, args[8])
+				return mockSubvolumeSnapshotInfoResp, nil
+			}
+			// cancel pending clones
+			if args[0] == "fs" && args[1] == "clone" && args[2] == "cancel" {
+				assert.Equal(t, fsName, args[3])
+				assert.Equal(t, "clone3", args[4])
+				assert.Equal(t, "--group_name", args[5])
+				assert.Equal(t, subVolumeGroupName, args[6])
+				return "", nil
+			}
+			// delete snapshots
+			if args[0] == "fs" && args[1] == "subvolume" && args[2] == "snapshot" && args[3] == "rm" {
+				assert.Equal(t, fsName, args[4])
+				assert.Equal(t, "csi-vol-9d942d45-6f48-4d9e-a707-edfcd602bd89", args[5])
+				assert.Equal(t, "snap0", args[6])
+				assert.Equal(t, "--group_name", args[7])
+				assert.Equal(t, subVolumeGroupName, args[8])
+				return "", nil
+			}
+			// delete subvolume
+			if args[0] == "fs" && args[1] == "subvolume" && args[2] == "rm" {
+				assert.Equal(t, fsName, args[3])
+				assert.Equal(t, "csi-vol-9d942d45-6f48-4d9e-a707-edfcd602bd89", args[4])
+				assert.Equal(t, subVolumeGroupName, args[5])
+				assert.Equal(t, "--force", args[6])
+				return "", nil
+			}
+			return "", errors.New("unknown command")
+		}
+		context := &clusterd.Context{Executor: executor}
+		err := SubVolumeGroupCleanup(context, clusterInfo, fsName, subVolumeGroupName, poolName, csiNamespace)
+		assert.NoError(t, err)
+	})
+}
+
+func TestGetOmapValue(t *testing.T) {
+	result := getOMAPValue("csi-vol-3a41b367-9566-4dbb-8884-39e1fa306ea7")
+	assert.Equal(t, "csi.volume.3a41b367-9566-4dbb-8884-39e1fa306ea7", result)
+
+	result = getOMAPValue("invalidSubVolume")
+	assert.Equal(t, "", result)
+}

pkg/daemon/ceph/client/filesystem.go

@@ -435,7 +435,7 @@ type SubvolumeList []Subvolume
 const NoSubvolumeGroup = ""
 
 // ListSubvolumesInGroup lists all subvolumes present in the given filesystem's subvolume group by
-// name. If groupName is empty, list subvolumes that are not in any group. Times out after 5 seconds.
+// name. If groupName is empty, list subvolumes that are not in any group.
 var ListSubvolumesInGroup = listSubvolumesInGroup
 
 // with above, allow this to be overridden for unit testing
@@ -453,10 +453,61 @@ func listSubvolumesInGroup(context *clusterd.Context, clusterInfo *ClusterInfo,
 	if err != nil {
 		return svs, errors.Wrapf(err, "failed to list subvolumes in filesystem %q subvolume group %q", fsName, groupName)
 	}
-
 	if err := json.Unmarshal(buf, &svs); err != nil {
 		return svs, errors.Wrapf(err, "failed to unmarshal subvolume list for filesystem %q subvolume group %q", fsName, groupName)
 	}
+	return svs, nil
+}
+
+// SubVolumeSnapshot represents snapshot of a cephFS subvolume
+type SubVolumeSnapshot struct {
+	Name string `json:"name"`
+}
+
+// SubVolumeSnapshots is the list of snapshots in a CephFS subvolume
+type SubVolumeSnapshots []SubVolumeSnapshot
+
+// ListSubVolumeSnaphots lists all the subvolume snapshots present in the subvolume in the given filesystem's subvolume group.
+var ListSubVolumeSnapshots = listSubVolumeSnapshots
+
+func listSubVolumeSnapshots(context *clusterd.Context, clusterInfo *ClusterInfo, fsName, subVolumeName, groupName string) (SubVolumeSnapshots, error) {
+	svs := SubVolumeSnapshots{}
+	args := []string{"fs", "subvolume", "snapshot", "ls", fsName, subVolumeName, "--group_name", groupName}
+	cmd := NewCephCommand(context, clusterInfo, args)
+	buf, err := cmd.RunWithTimeout(exec.CephCommandsTimeout)
+	if err != nil {
+		return svs, errors.Wrapf(err, "failed to list subvolumes in filesystem %q subvolume group %q", fsName, groupName)
+	}
+
+	if err := json.Unmarshal(buf, &svs); err != nil {
+		return svs, errors.Wrapf(err, "failed to unmarshal snapshots for subvolume %q for filesystem %q subvolume group %q", subVolumeName, fsName, groupName)
+	}
 
 	return svs, nil
 }
+
+// SubVolumeSnapshotPendingClones refers to all the pending clones available in a cephFS subvolume snapshot
+type SubVolumeSnapshotPendingClones struct {
+	Clones []struct {
+		Name string `json:"name"`
+	} `json:"pending_clones"`
+}
+
+var ListSubVolumeSnapshotPendingClones = listSubVolumeSnapshotPendingClones
+
+// listSubVolumeSnapshotPendingClones lists all the pending clones available in a cephFS subvolume snapshot
+func listSubVolumeSnapshotPendingClones(context *clusterd.Context, clusterInfo *ClusterInfo, fsName, subVolumeName, snap, groupName string) (SubVolumeSnapshotPendingClones, error) {
+	pendingClones := SubVolumeSnapshotPendingClones{}
+	args := []string{"fs", "subvolume", "snapshot", "info", fsName, subVolumeName, snap, "--group_name", groupName}
+	cmd := NewCephCommand(context, clusterInfo, args)
+	buf, err := cmd.RunWithTimeout(exec.CephCommandsTimeout)
+	if err != nil {
+		return pendingClones, errors.Wrapf(err, "failed to list pending clones available for snapshot %q in filesystem %q in subvolume group %q", snap, fsName, groupName)
+	}
+
+	if err := json.Unmarshal(buf, &pendingClones); err != nil {
+		return pendingClones, errors.Wrapf(err, "failed to unmarshal pending clones list for for snapshot %q in filesystem %q subvolume group %q", snap, fsName, groupName)
+	}
+
+	return pendingClones, nil
+}

pkg/daemon/ceph/client/subvolumegroup.go

@@ -19,10 +19,12 @@ package client
 import (
 	"fmt"
 	"strconv"
+	"strings"
 
 	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
+	"github.com/rook/rook/pkg/util/exec"
 	"k8s.io/apimachinery/pkg/types"
 )
 
@@ -129,3 +131,81 @@ func validatePinningValues(pinning cephv1.CephFilesystemSubVolumeGroupSpecPinnin
 	}
 	return err
 }
+
+func GetOMAPKey(context *clusterd.Context, clusterInfo *ClusterInfo, omapObj, poolName, namespace string) (string, error) {
+	args := []string{"getomapval", omapObj, "csi.volname", "-p", poolName, "--namespace", namespace, "/dev/stdout"}
+	cmd := NewRadosCommand(context, clusterInfo, args)
+	buf, err := cmd.RunWithTimeout(exec.CephCommandsTimeout)
+	if err != nil {
+		return "", errors.Wrapf(err, "failed to list omapKeys for omapObj %q", omapObj)
+	}
+
+	// Todo: is there a way to avoid this parsing?
+	respStr := string(buf)
+	var pvcName string
+	if len(respStr) != 0 {
+		resp := strings.Split(respStr, "\n")
+		if len(resp) == 2 {
+			pvcName = resp[1]
+		}
+	}
+
+	if pvcName == "" {
+		return "", nil
+	}
+
+	omapKey := fmt.Sprintf("ceph.volume.%s", pvcName)
+	return omapKey, nil
+}
+
+func DeleteOmapValue(context *clusterd.Context, clusterInfo *ClusterInfo, omapValue, poolName, namespace string) error {
+	args := []string{"rm", omapValue, "-p", poolName, "--namespace", namespace}
+	cmd := NewRadosCommand(context, clusterInfo, args)
+	_, err := cmd.RunWithTimeout(exec.CephCommandsTimeout)
+	if err != nil {
+		return errors.Wrapf(err, "failed to delete omap value %q in pool %q", omapValue, poolName)
+	}
+	logger.Infof("successfully deleted omap value %q for pool %q", omapValue, poolName)
+	return nil
+}
+
+func DeleteOmapKey(context *clusterd.Context, clusterInfo *ClusterInfo, omapKey, poolName, namespace string) error {
+	args := []string{"rmomapkey", "csi.volumes.default", omapKey, "-p", poolName, "--namespace", namespace}
+	cmd := NewRadosCommand(context, clusterInfo, args)
+	_, err := cmd.RunWithTimeout(exec.CephCommandsTimeout)
+	if err != nil {
+		return errors.Wrapf(err, "failed to delete omapKey %q in pool %q", omapKey, poolName)
+	}
+	logger.Infof("successfully deleted omap key %q for pool %q", omapKey, poolName)
+	return nil
+}
+
+func DeleteSubVolume(context *clusterd.Context, clusterInfo *ClusterInfo, fs, subvol, svg string) error {
+	args := []string{"fs", "subvolume", "rm", fs, subvol, svg, "--force"}
+	cmd := NewCephCommand(context, clusterInfo, args)
+	_, err := cmd.RunWithTimeout(exec.CephCommandsTimeout)
+	if err != nil {
+		return errors.Wrapf(err, "failed to delete subvolume %q in filesystem %q", subvol, fs)
+	}
+	return nil
+}
+
+func DeleteSubvolumeSnapshot(context *clusterd.Context, clusterInfo *ClusterInfo, fs, subvol, svg, snap string) error {
+	args := []string{"fs", "subvolume", "snapshot", "rm", fs, subvol, snap, "--group_name", svg}
+	cmd := NewCephCommand(context, clusterInfo, args)
+	_, err := cmd.RunWithTimeout(exec.CephCommandsTimeout)
+	if err != nil {
+		return errors.Wrapf(err, "failed to delete subvolume %q in filesystem %q", subvol, fs)
+	}
+	return nil
+}
+
+func CancelSnapshotClone(context *clusterd.Context, clusterInfo *ClusterInfo, fs, svg, clone string) error {
+	args := []string{"fs", "clone", "cancel", fs, clone, "--group_name", svg}
+	cmd := NewCephCommand(context, clusterInfo, args)
+	_, err := cmd.RunWithTimeout(exec.CephCommandsTimeout)
+	if err != nil {
+		return errors.Wrapf(err, "failed to cancel clone %q in filesystem %q in group %q", clone, fs, svg)
+	}
+	return nil
+}

pkg/operator/ceph/cluster/cleanup.go

@@ -138,7 +138,7 @@ func (c *ClusterController) cleanUpJobContainer(cluster *cephv1.CephCluster, mon
 		SecurityContext: securityContext,
 		VolumeMounts:    volumeMounts,
 		Env:             envVars,
-		Args:            []string{"ceph", "clean"},
+		Args:            []string{"ceph", "clean", "host"},
 		Resources:       cephv1.GetCleanupResources(cluster.Spec.Resources),
 	}
 }

pkg/operator/ceph/controller/cleanup.go

@@ -0,0 +1,144 @@
+/*
+Copyright 2024 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package controller
+
+import (
+	"context"
+	"strings"
+
+	"github.com/pkg/errors"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/operator/k8sutil"
+	batch "k8s.io/api/batch/v1"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	k8sClient "sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+const (
+	volumeName                  = "cleanup-volume"
+	dataDirHostPath             = "ROOK_DATA_DIR_HOST_PATH"
+	CleanupAppName              = "resource-cleanup"
+	RESOURCE_CLEANUP_ANNOTATION = "rook.io/force-deletion"
+
+	// CephFSSubVolumeGroup env resources
+	CephFSSubVolumeGroupNameEnv = "SUB_VOLUME_GROUP_NAME"
+	CephFSNameEnv               = "FILESYSTEM_NAME"
+	CSICephFSRadosNamesaceEnv   = "CSI_CEPHFS_RADOS_NAMESPACE"
+	CephFSMetaDataPoolNameEnv   = "METADATA_POOL_NAME"
+)
+
+// ResourceCleanup defines an rook ceph resource to be cleaned up
+type ResourceCleanup struct {
+	resource  k8sClient.Object
+	cluster   *cephv1.CephCluster
+	rookImage string
+	// config defines the attributes of the custom resource to passed in as environment variables in the clean up job
+	config map[string]string
+}
+
+func NewResourceCleanup(obj k8sClient.Object, cluster *cephv1.CephCluster, rookImage string, config map[string]string) *ResourceCleanup {
+	return &ResourceCleanup{
+		resource:  obj,
+		rookImage: rookImage,
+		cluster:   cluster,
+		config:    config,
+	}
+}
+
+// Start a new job to perform clean up of the ceph resources. It returns true if the cleanup job has succeeded
+func (c *ResourceCleanup) StartJob(ctx context.Context, clientset kubernetes.Interface, jobName string) error {
+	podSpec := c.jobTemplateSpec()
+	job := &batch.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            jobName,
+			Namespace:       c.resource.GetNamespace(),
+			OwnerReferences: c.resource.GetOwnerReferences(),
+		},
+		Spec: batch.JobSpec{
+			Template: podSpec,
+		},
+	}
+
+	if err := k8sutil.RunReplaceableJob(ctx, clientset, job, false); err != nil {
+		return errors.Wrapf(err, "failed to run clean up job for %q resource named %q in namespace %q",
+			c.resource.GetObjectKind().GroupVersionKind().Kind, c.resource.GetName(), c.resource.GetNamespace())
+	}
+
+	return nil
+}
+
+func (c *ResourceCleanup) jobContainer() v1.Container {
+	volumeMounts := []v1.VolumeMount{}
+	envVars := []v1.EnvVar{}
+	if c.cluster.Spec.DataDirHostPath != "" {
+		hostPathVolumeMount := v1.VolumeMount{Name: volumeName, MountPath: c.cluster.Spec.DataDirHostPath}
+		volumeMounts = append(volumeMounts, hostPathVolumeMount)
+		envVars = append(envVars, []v1.EnvVar{
+			{Name: dataDirHostPath, Value: c.cluster.Spec.DataDirHostPath},
+			{Name: "ROOK_LOG_LEVEL", Value: "DEBUG"},
+			{Name: k8sutil.PodNamespaceEnvVar, Value: c.resource.GetNamespace()},
+		}...)
+	}
+	// append all the resource attributes as env variables.
+	for k, v := range c.config {
+		envVars = append(envVars, v1.EnvVar{Name: k, Value: v})
+	}
+	securityContext := PrivilegedContext(true)
+	return v1.Container{
+		Name:            "resource-cleanup",
+		Image:           c.rookImage,
+		SecurityContext: securityContext,
+		VolumeMounts:    volumeMounts,
+		Env:             envVars,
+		Args:            []string{"ceph", "clean", c.resource.GetObjectKind().GroupVersionKind().Kind},
+		Resources:       cephv1.GetCleanupResources(c.cluster.Spec.Resources),
+	}
+}
+
+func (c *ResourceCleanup) jobTemplateSpec() v1.PodTemplateSpec {
+	volumes := []v1.Volume{}
+	hostPathVolume := v1.Volume{Name: volumeName, VolumeSource: v1.VolumeSource{HostPath: &v1.HostPathVolumeSource{Path: c.cluster.Spec.DataDirHostPath}}}
+	volumes = append(volumes, hostPathVolume)
+
+	podSpec := v1.PodTemplateSpec{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: CleanupAppName,
+		},
+		Spec: v1.PodSpec{
+			Containers: []v1.Container{
+				c.jobContainer(),
+			},
+			Volumes:            volumes,
+			RestartPolicy:      v1.RestartPolicyOnFailure,
+			PriorityClassName:  cephv1.GetCleanupPriorityClassName(c.cluster.Spec.PriorityClassNames),
+			ServiceAccountName: k8sutil.DefaultServiceAccount,
+		},
+	}
+
+	return podSpec
+}
+
+// ForceDeleteRequested returns true if `rook.io/force-deletion:true` annotation is available on the resource
+func ForceDeleteRequested(annotations map[string]string) bool {
+	if value, found := annotations[RESOURCE_CLEANUP_ANNOTATION]; found {
+		if strings.EqualFold(value, "true") {
+			return true
+		}
+	}
+	return false
+}

pkg/operator/ceph/controller/cleanup_test.go

@@ -0,0 +1,82 @@
+/*
+Copyright 2024 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"testing"
+
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestJobTemplateSpec(t *testing.T) {
+	expectedHostPath := "var/lib/rook"
+	expectedNamespace := "test-rook-ceph"
+	rookImage := "test"
+	cluster := &cephv1.CephCluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: expectedNamespace,
+		},
+		Spec: cephv1.ClusterSpec{
+			DataDirHostPath: expectedHostPath,
+			CleanupPolicy: cephv1.CleanupPolicySpec{
+				Confirmation: "yes-really-destroy-data",
+			},
+		},
+	}
+	svgObj := &cephv1.CephFilesystemSubVolumeGroup{
+		TypeMeta: metav1.TypeMeta{
+			Kind: "CephFSSubvolumeGroup",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-svg",
+			Namespace: expectedNamespace,
+		},
+	}
+	testConfig := map[string]string{
+		"config1": "value1",
+		"config2": "value2",
+	}
+	cleanup := NewResourceCleanup(svgObj, cluster, rookImage, testConfig)
+	podTemplateSpec := cleanup.jobTemplateSpec()
+	assert.Equal(t, "CephFSSubvolumeGroup", podTemplateSpec.Spec.Containers[0].Args[2])
+	assert.Equal(t, "config1", podTemplateSpec.Spec.Containers[0].Env[3].Name)
+	assert.Equal(t, "value1", podTemplateSpec.Spec.Containers[0].Env[3].Value)
+	assert.Equal(t, "config2", podTemplateSpec.Spec.Containers[0].Env[4].Name)
+	assert.Equal(t, "value2", podTemplateSpec.Spec.Containers[0].Env[4].Value)
+}
+
+func TestForceDeleteRequested(t *testing.T) {
+	svgObj := &cephv1.CephFilesystemSubVolumeGroup{
+		TypeMeta: metav1.TypeMeta{
+			Kind: "CephFSSubvolumeGroup",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        "test-svg",
+			Namespace:   "test",
+			Annotations: map[string]string{},
+		},
+	}
+
+	result := ForceDeleteRequested(svgObj.Annotations)
+	assert.False(t, result)
+
+	svgObj.Annotations[RESOURCE_CLEANUP_ANNOTATION] = "true"
+	result = ForceDeleteRequested(svgObj.Annotations)
+	assert.True(t, result)
+}

pkg/operator/ceph/file/filesystem.go

@@ -169,7 +169,7 @@ func newFS(name, namespace string) *Filesystem {
 func createOrUpdatePools(f *Filesystem, context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, clusterSpec *cephv1.ClusterSpec, spec cephv1.FilesystemSpec) error {
 	// generating the metadata pool's name
 	metadataPool := cephv1.NamedPoolSpec{
-		Name:     generateMetaDataPoolName(f),
+		Name:     GenerateMetaDataPoolName(f.Name),
 		PoolSpec: spec.MetadataPool,
 	}
 	metadataPool.Application = cephfsApplication
@@ -242,7 +242,7 @@ func (f *Filesystem) doFilesystemCreate(context *clusterd.Context, clusterInfo *
 
 	spec.MetadataPool.Application = cephfsApplication
 	metadataPool := cephv1.NamedPoolSpec{
-		Name:     generateMetaDataPoolName(f),
+		Name:     GenerateMetaDataPoolName(f.Name),
 		PoolSpec: spec.MetadataPool,
 	}
 	if _, poolFound := reversedPoolMap[metadataPool.Name]; !poolFound {
@@ -307,7 +307,7 @@ func generateDataPoolNames(f *Filesystem, spec cephv1.FilesystemSpec) []string {
 	return dataPoolNames
 }
 
-// generateMetaDataPoolName generates MetaDataPool name by prefixing the filesystem name to the constant metaDataPoolSuffix
-func generateMetaDataPoolName(f *Filesystem) string {
-	return fmt.Sprintf("%s-%s", f.Name, metaDataPoolSuffix)
+// GenerateMetaDataPoolName generates MetaDataPool name by prefixing the filesystem name to the constant metaDataPoolSuffix
+func GenerateMetaDataPoolName(fsName string) string {
+	return fmt.Sprintf("%s-%s", fsName, metaDataPoolSuffix)
 }

pkg/operator/ceph/file/subvolumegroup/controller.go

@@ -40,6 +40,7 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/csi"
+	"github.com/rook/rook/pkg/operator/ceph/file"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
@@ -187,7 +188,7 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) reconcile(request reconcile.Requ
 		if cephCluster.Spec.External.Enable {
 			logger.Warningf("external subvolume group %q deletion is not supported, delete it manually", namespacedName)
 		} else {
-			err := r.deleteSubVolumeGroup(cephFilesystemSubVolumeGroup)
+			err = r.deleteSubVolumeGroup(cephFilesystemSubVolumeGroup, &cephCluster)
 			if err != nil {
 				if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) {
 					logger.Info(opcontroller.OperatorNotInitializedMessage)
@@ -326,7 +327,8 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) createOrUpdateSubVolumeGroup(cep
 }
 
 // Delete the ceph filesystem subvolume group
-func (r *ReconcileCephFilesystemSubVolumeGroup) deleteSubVolumeGroup(cephFilesystemSubVolumeGroup *cephv1.CephFilesystemSubVolumeGroup) error {
+func (r *ReconcileCephFilesystemSubVolumeGroup) deleteSubVolumeGroup(cephFilesystemSubVolumeGroup *cephv1.CephFilesystemSubVolumeGroup,
+	cephCluster *cephv1.CephCluster) error {
 	namespacedName := fmt.Sprintf("%s/%s", cephFilesystemSubVolumeGroup.Namespace, cephFilesystemSubVolumeGroup.Name)
 	logger.Infof("deleting ceph filesystem subvolume group object %q", namespacedName)
 	if err := cephclient.DeleteCephFSSubVolumeGroup(r.context, r.clusterInfo, cephFilesystemSubVolumeGroup.Spec.FilesystemName, getSubvolumeGroupName(cephFilesystemSubVolumeGroup)); err != nil {
@@ -339,7 +341,17 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) deleteSubVolumeGroup(cephFilesys
 		// If the subvolume group has subvolumes the command will fail with:
 		// Error ENOTEMPTY: error in rmdir /volumes/csi
 		if ok && (code == int(syscall.ENOTEMPTY)) {
-			return errors.Wrapf(err, "failed to delete ceph filesystem subvolume group %q, remove the subvolumes first", cephFilesystemSubVolumeGroup.Name)
+			msg := fmt.Sprintf("failed to delete ceph filesystem subvolume group %q, remove the subvolumes first", cephFilesystemSubVolumeGroup.Name)
+			if opcontroller.ForceDeleteRequested(cephFilesystemSubVolumeGroup.GetAnnotations()) {
+				// cleanup cephFS subvolumes
+				cleanupErr := r.cleanup(cephFilesystemSubVolumeGroup, cephCluster)
+				if cleanupErr != nil {
+					return errors.Wrapf(cleanupErr, "failed to clean up all the ceph resources created by subVolumeGroup %q", namespacedName)
+				}
+				msg = fmt.Sprintf("failed to delete ceph filesystem subvolume group %q, started clean up job to delete the subvolumes", cephFilesystemSubVolumeGroup.Name)
+			}
+
+			return errors.Wrapf(err, msg)
 		}
 
 		return errors.Wrapf(err, "failed to delete ceph filesystem subvolume group %q", cephFilesystemSubVolumeGroup.Name)
@@ -380,3 +392,20 @@ func buildClusterID(cephFilesystemSubVolumeGroup *cephv1.CephFilesystemSubVolume
 	clusterID := fmt.Sprintf("%s-%s-file-%s", cephFilesystemSubVolumeGroup.Namespace, cephFilesystemSubVolumeGroup.Spec.FilesystemName, getSubvolumeGroupName(cephFilesystemSubVolumeGroup))
 	return k8sutil.Hash(clusterID)
 }
+
+func (r *ReconcileCephFilesystemSubVolumeGroup) cleanup(svg *cephv1.CephFilesystemSubVolumeGroup, cephCluster *cephv1.CephCluster) error {
+	logger.Infof("starting cleanup of the ceph resources for subVolumeGroup %q in namespace %q", svg.Name, svg.Namespace)
+	cleanupConfig := map[string]string{
+		opcontroller.CephFSSubVolumeGroupNameEnv: svg.Spec.Name,
+		opcontroller.CephFSNameEnv:               svg.Spec.FilesystemName,
+		opcontroller.CSICephFSRadosNamesaceEnv:   "csi",
+		opcontroller.CephFSMetaDataPoolNameEnv:   file.GenerateMetaDataPoolName(svg.Spec.FilesystemName),
+	}
+	cleanup := opcontroller.NewResourceCleanup(svg, cephCluster, r.opConfig.Image, cleanupConfig)
+	jobName := k8sutil.TruncateNodeNameForJob("cleanup-svg-%s", fmt.Sprintf("%s-%s", svg.Spec.FilesystemName, svg.Name))
+	err := cleanup.StartJob(r.clusterInfo.Context, r.context.Clientset, jobName)
+	if err != nil {
+		return errors.Wrapf(err, "failed to run clean up job to clean the ceph resources in cephFS subVolumeGroup %q", svg.Name)
+	}
+	return nil
+}