Merge pull request #15832 from rook/mergify/bp/release-1.17/pr-15817

block: Add more deletion conditions to blockpool and radosnamespace status (backport #15817)

Documentation/CRDs/specification.md

@@ -3388,6 +3388,18 @@ SnapshotScheduleStatusSpec
 <em>(Optional)</em>
 </td>
 </tr>
+<tr>
+<td>
+<code>conditions</code><br/>
+<em>
+<a href="#ceph.rook.io/v1.Condition">
+[]Condition
+</a>
+</em>
+</td>
+<td>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="ceph.rook.io/v1.CephBlockPoolStatus">CephBlockPoolStatus
@@ -5149,7 +5161,7 @@ The default is not set.</p>
 <h3 id="ceph.rook.io/v1.Condition">Condition
 </h3>
 <p>
-(<em>Appears on:</em><a href="#ceph.rook.io/v1.CephBlockPoolStatus">CephBlockPoolStatus</a>, <a href="#ceph.rook.io/v1.CephFilesystemStatus">CephFilesystemStatus</a>, <a href="#ceph.rook.io/v1.ClusterStatus">ClusterStatus</a>, <a href="#ceph.rook.io/v1.ObjectStoreStatus">ObjectStoreStatus</a>, <a href="#ceph.rook.io/v1.Status">Status</a>)
+(<em>Appears on:</em><a href="#ceph.rook.io/v1.CephBlockPoolRadosNamespaceStatus">CephBlockPoolRadosNamespaceStatus</a>, <a href="#ceph.rook.io/v1.CephBlockPoolStatus">CephBlockPoolStatus</a>, <a href="#ceph.rook.io/v1.CephFilesystemStatus">CephFilesystemStatus</a>, <a href="#ceph.rook.io/v1.ClusterStatus">ClusterStatus</a>, <a href="#ceph.rook.io/v1.ObjectStoreStatus">ObjectStoreStatus</a>, <a href="#ceph.rook.io/v1.Status">Status</a>)
 </p>
 <div>
 <p>Condition represents a status condition on any Rook-Ceph Custom Resource.</p>
@@ -5275,6 +5287,22 @@ deletion.</p>
 <td><p>ObjectHasNoDependentsReason represents when a resource object has no dependents that are
 blocking deletion.</p>
 </td>
+</tr><tr><td><p>&#34;PoolEmpty&#34;</p></td>
+<td><p>PoolEmptyReason represents when a pool does not contain images or snapshots that are blocking
+deletion.</p>
+</td>
+</tr><tr><td><p>&#34;PoolNotEmpty&#34;</p></td>
+<td><p>PoolNotEmptyReason represents when a pool contains images or snapshots that are blocking
+deletion.</p>
+</td>
+</tr><tr><td><p>&#34;RadosNamespaceEmpty&#34;</p></td>
+<td><p>RadosNamespaceEmptyReason represents when a rados namespace does not contain images or snapshots that are blocking
+deletion.</p>
+</td>
+</tr><tr><td><p>&#34;RadosNamespaceNotEmpty&#34;</p></td>
+<td><p>RadosNamespaceNotEmptyReason represents when a rados namespace contains images or snapshots that are blocking
+deletion.</p>
+</td>
 </tr><tr><td><p>&#34;ReconcileFailed&#34;</p></td>
 <td><p>ReconcileFailed represents when a resource reconciliation failed.</p>
 </td>
@@ -5316,9 +5344,15 @@ blocking deletion.</p>
 </tr><tr><td><p>&#34;Failure&#34;</p></td>
 <td><p>ConditionFailure represents Failure state of an object</p>
 </td>
+</tr><tr><td><p>&#34;PoolDeletionIsBlocked&#34;</p></td>
+<td><p>ConditionPoolDeletionIsBlocked represents when deletion of the object is blocked.</p>
+</td>
 </tr><tr><td><p>&#34;Progressing&#34;</p></td>
 <td><p>ConditionProgressing represents Progressing state of an object</p>
 </td>
+</tr><tr><td><p>&#34;RadosNamespaceDeletionIsBlocked&#34;</p></td>
+<td><p>ConditionRadosNSDeletionIsBlocked represents when deletion of the object is blocked.</p>
+</td>
 </tr><tr><td><p>&#34;Ready&#34;</p></td>
 <td><p>ConditionReady represents Ready state of an object</p>
 </td>

deploy/charts/rook-ceph/templates/resources.yaml

@@ -104,6 +104,28 @@ spec:
             status:
               description: Status represents the status of a CephBlockPool Rados Namespace
               properties:
+                conditions:
+                  items:
+                    description: Condition represents a status condition on any Rook-Ceph Custom Resource.
+                    properties:
+                      lastHeartbeatTime:
+                        format: date-time
+                        type: string
+                      lastTransitionTime:
+                        format: date-time
+                        type: string
+                      message:
+                        type: string
+                      reason:
+                        description: ConditionReason is a reason for a condition
+                        type: string
+                      status:
+                        type: string
+                      type:
+                        description: ConditionType represent a resource's status
+                        type: string
+                    type: object
+                  type: array
                 info:
                   additionalProperties:
                     type: string

deploy/examples/crds.yaml

@@ -107,6 +107,28 @@ spec:
             status:
               description: Status represents the status of a CephBlockPool Rados Namespace
               properties:
+                conditions:
+                  items:
+                    description: Condition represents a status condition on any Rook-Ceph Custom Resource.
+                    properties:
+                      lastHeartbeatTime:
+                        format: date-time
+                        type: string
+                      lastTransitionTime:
+                        format: date-time
+                        type: string
+                      message:
+                        type: string
+                      reason:
+                        description: ConditionReason is a reason for a condition
+                        type: string
+                      status:
+                        type: string
+                      type:
+                        description: ConditionType represent a resource's status
+                        type: string
+                    type: object
+                  type: array
                 info:
                   additionalProperties:
                     type: string

pkg/apis/ceph.rook.io/v1/namespace.go

@@ -0,0 +1,31 @@
+/*
+Copyright 2025 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 v1
+
+const (
+	ImplicitNamespaceKey = "<implicit>"
+	ImplicitNamespaceVal = ""
+)
+
+func GetRadosNamespaceName(cephBlockPoolRadosNamespace *CephBlockPoolRadosNamespace) string {
+	if cephBlockPoolRadosNamespace.Spec.Name == ImplicitNamespaceKey {
+		return ImplicitNamespaceVal
+	} else if cephBlockPoolRadosNamespace.Spec.Name != "" {
+		return cephBlockPoolRadosNamespace.Spec.Name
+	}
+	return cephBlockPoolRadosNamespace.Name
+}

pkg/apis/ceph.rook.io/v1/pool.go

@@ -95,6 +95,10 @@ func (p *CephBlockPool) GetStatusConditions() *[]Condition {
 	return &p.Status.Conditions
 }
 
+func (p *CephBlockPoolRadosNamespace) GetStatusConditions() *[]Condition {
+	return &p.Status.Conditions
+}
+
 // SnapshotSchedulesEnabled returns whether snapshot schedules are desired
 func (p *MirroringSpec) SnapshotSchedulesEnabled() bool {
 	return len(p.SnapshotSchedules) > 0

pkg/apis/ceph.rook.io/v1/types.go

@@ -568,6 +568,18 @@ const (
 	// ObjectHasNoDependentsReason represents when a resource object has no dependents that are
 	// blocking deletion.
 	ObjectHasNoDependentsReason ConditionReason = "ObjectHasNoDependents"
+	// PoolNotEmptyReason represents when a pool contains images or snapshots that are blocking
+	// deletion.
+	PoolNotEmptyReason ConditionReason = "PoolNotEmpty"
+	// PoolEmptyReason represents when a pool does not contain images or snapshots that are blocking
+	// deletion.
+	PoolEmptyReason ConditionReason = "PoolEmpty"
+	// RadosNamespaceNotEmptyReason represents when a rados namespace contains images or snapshots that are blocking
+	// deletion.
+	RadosNamespaceNotEmptyReason ConditionReason = "RadosNamespaceNotEmpty"
+	// RadosNamespaceEmptyReason represents when a rados namespace does not contain images or snapshots that are blocking
+	// deletion.
+	RadosNamespaceEmptyReason ConditionReason = "RadosNamespaceEmpty"
 )
 
 // ConditionType represent a resource's status
@@ -589,6 +601,10 @@ const (
 
 	// ConditionDeletionIsBlocked represents when deletion of the object is blocked.
 	ConditionDeletionIsBlocked ConditionType = "DeletionIsBlocked"
+	// ConditionPoolDeletionIsBlocked represents when deletion of the object is blocked.
+	ConditionPoolDeletionIsBlocked ConditionType = "PoolDeletionIsBlocked"
+	// ConditionRadosNSDeletionIsBlocked represents when deletion of the object is blocked.
+	ConditionRadosNSDeletionIsBlocked ConditionType = "RadosNamespaceDeletionIsBlocked"
 )
 
 // ClusterState represents the state of a Ceph Cluster
@@ -3534,6 +3550,7 @@ type CephBlockPoolRadosNamespaceStatus struct {
 	MirroringInfo *MirroringInfoSpec `json:"mirroringInfo,omitempty"`
 	// +optional
 	SnapshotScheduleStatus *SnapshotScheduleStatusSpec `json:"snapshotScheduleStatus,omitempty"`
+	Conditions             []Condition                 `json:"conditions,omitempty"`
 }
 
 // Represents the source of a volume to mount.

pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go

@@ -502,6 +502,13 @@ func (in *CephBlockPoolRadosNamespaceStatus) DeepCopyInto(out *CephBlockPoolRado
 		*out = new(SnapshotScheduleStatusSpec)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.Conditions != nil {
+		in, out := &in.Conditions, &out.Conditions
+		*out = make([]Condition, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
 	return
 }
 

pkg/daemon/ceph/client/mirror.go

@@ -53,10 +53,8 @@ type Images struct {
 }
 
 const (
-	mirrorModeDisabled   = "disabled"
-	mirrorModeInitOnly   = "init-only"
-	ImplicitNamespaceKey = "<implicit>"
-	ImplicitNamespaceVal = ""
+	mirrorModeDisabled = "disabled"
+	mirrorModeInitOnly = "init-only"
 )
 
 var (
@@ -398,8 +396,8 @@ func EnableRBDRadosNamespaceMirroring(context *clusterd.Context, clusterInfo *Cl
 		return errors.Errorf("ceph version %q does not support mirroring in rados namespace %q with --remote-namespace flag, supported version are v20 and above.", clusterInfo.CephVersion.String(), poolAndRadosNamespaceName)
 	}
 
-	if remoteNamespace != nil && *remoteNamespace == ImplicitNamespaceKey {
-		*remoteNamespace = ImplicitNamespaceVal
+	if remoteNamespace != nil && *remoteNamespace == cephv1.ImplicitNamespaceKey {
+		*remoteNamespace = cephv1.ImplicitNamespaceVal
 	}
 
 	args := []string{"mirror", "pool", "enable", poolAndRadosNamespaceName, mode}

pkg/daemon/ceph/client/pool.go

@@ -220,22 +220,54 @@ func CreatePoolWithPGs(context *clusterd.Context, clusterInfo *ClusterInfo, clus
 		true /* enableECOverwrite */)
 }
 
+func IsPoolEmpty(context *clusterd.Context, clusterInfo *ClusterInfo, name string, radosNamespaces []string) (bool, string, error) {
+	logger.Debugf("checking if pool %q in namespace %q is empty", name, clusterInfo.Namespace)
+
+	isEmpty, err := checkIsPoolEmpty(context, clusterInfo, name)
+	if err != nil {
+		return false, fmt.Sprintf("failed to check if pool %q is empty", name), err
+	}
+	if !isEmpty {
+		return false, fmt.Sprintf("pool %q contains images or snapshots and cannot be deleted", name), nil
+	}
+	// Check the rados namespaces for images/snapshots
+	for _, radosNamespace := range radosNamespaces {
+		containsImages, err := checkForImagesInRadosNamespace(context, clusterInfo, name, radosNamespace)
+		if err != nil || containsImages {
+			return false, fmt.Sprintf("rados namespace %q contains images or snapshots preventing pool %q deletion", radosNamespace, name), nil
+		}
+	}
+	// No images/snapshots found in the pool or rados namespaces
+	return true, fmt.Sprintf("pool %q is empty and can be deleted", name), nil
+}
+
 func checkForImagesInPool(context *clusterd.Context, clusterInfo *ClusterInfo, name string) error {
+	isEmpty, err := checkIsPoolEmpty(context, clusterInfo, name)
+	if err != nil {
+		return err
+	}
+	if isEmpty {
+		return nil
+	}
+	return errors.Errorf("pool %q contains images/snapshots", name)
+}
+
+func checkIsPoolEmpty(context *clusterd.Context, clusterInfo *ClusterInfo, name string) (bool, error) {
 	var err error
 	logger.Debugf("checking any images/snapshots present in pool %q", name)
 	stats, err := GetPoolStatistics(context, clusterInfo, name)
 	if err != nil {
 		if strings.Contains(err.Error(), "No such file or directory") {
-			return nil
+			return false, nil
 		}
-		return errors.Wrapf(err, "failed to list images/snapshots in pool %s", name)
+		return false, errors.Wrapf(err, "failed to list images/snapshots in pool %s", name)
 	}
 	if stats.Images.Count == 0 && stats.Images.SnapCount == 0 {
 		logger.Infof("no images/snapshots present in pool %q", name)
-		return nil
+		return true, nil
 	}
 
-	return errors.Errorf("pool %q contains images/snapshots", name)
+	return false, nil
 }
 
 // DeletePool purges a pool from Ceph

pkg/daemon/ceph/client/radosnamespace.go

@@ -69,25 +69,27 @@ func getRadosNamespaceStatistics(context *clusterd.Context, clusterInfo *Cluster
 	return &poolStats, nil
 }
 
-func checkForImagesInRadosNamespace(context *clusterd.Context, clusterInfo *ClusterInfo, poolName, namespaceName string) error {
+// checkForImagesInRadosNamespace checks if there are any images or snapshots in the specified rados namespace.
+// If there are images or snapshots, it returns true and an error with details.
+func checkForImagesInRadosNamespace(context *clusterd.Context, clusterInfo *ClusterInfo, poolName, namespaceName string) (bool, error) {
 	logger.Debugf("checking any images/snapshots present in pool %s/%s in k8s namespace %q", poolName, namespaceName, clusterInfo.Namespace)
 	stats, err := getRadosNamespaceStatistics(context, clusterInfo, poolName, namespaceName)
 	if err != nil {
-		return errors.Wrapf(err, "failed to list images/snapshots in pool %s/%s", poolName, namespaceName)
+		return false, errors.Wrapf(err, "failed to list images/snapshots in pool %s/%s", poolName, namespaceName)
 	}
 	if stats.Images.Count == 0 && stats.Images.SnapCount == 0 {
 		logger.Infof("no images/snapshots present in pool %s/%s in k8s namespace %q", poolName, namespaceName, clusterInfo.Namespace)
-		return nil
+		return false, nil
 	}
 
-	return errors.Errorf("pool %s/%s contains %d images and %d snapshots", poolName, namespaceName, stats.Images.Count, stats.Images.SnapCount)
+	return true, errors.Errorf("pool %s/%s contains %d images and %d snapshots", poolName, namespaceName, stats.Images.Count, stats.Images.SnapCount)
 }
 
 // DeleteRadosNamespace delete a rados namespace.
-func DeleteRadosNamespace(context *clusterd.Context, clusterInfo *ClusterInfo, poolName, namespaceName string) error {
-	err := checkForImagesInRadosNamespace(context, clusterInfo, poolName, namespaceName)
+func DeleteRadosNamespace(context *clusterd.Context, clusterInfo *ClusterInfo, poolName, namespaceName string) (bool, error) {
+	containsImages, err := checkForImagesInRadosNamespace(context, clusterInfo, poolName, namespaceName)
 	if err != nil {
-		return errors.Wrapf(err, "failed to check if pool %s/%s has rbd images", poolName, namespaceName)
+		return containsImages, errors.Wrapf(err, "failed to check if pool %s/%s has rbd images", poolName, namespaceName)
 	}
 	logger.Infof("deleting rados namespace %s/%s in k8s namespace %q", poolName, namespaceName, clusterInfo.Namespace)
 	args := []string{"namespace", "remove", "--pool", poolName, "--namespace", namespaceName}
@@ -97,12 +99,12 @@ func DeleteRadosNamespace(context *clusterd.Context, clusterInfo *ClusterInfo, p
 	if err != nil {
 		code, ok := exec.ExitStatus(err)
 		if !ok || code != int(syscall.ENOENT) {
-			return errors.Wrapf(err, "failed to delete rados namespace %s/%s. %s", poolName, namespaceName, output)
+			return false, errors.Wrapf(err, "failed to delete rados namespace %s/%s. %s", poolName, namespaceName, output)
 		}
 	}
 
 	logger.Infof("successfully deleted rados namespace %s/%s in k8s namespace %q", poolName, namespaceName, clusterInfo.Namespace)
-	return nil
+	return false, nil
 }
 
 // ListRadosNamespacesInPool lists the rados namespaces in a pool

pkg/operator/ceph/pool/controller.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/coreos/pkg/capnslog"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/util/dependents"
 	"github.com/rook/rook/pkg/util/exec"
 
 	"github.com/pkg/errors"
@@ -254,15 +255,10 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
 	poolSpec := cephBlockPool.ToNamedPoolSpec()
 	// DELETE: the CR was deleted
 	if !cephBlockPool.GetDeletionTimestamp().IsZero() {
-		deps, err := cephBlockPoolDependents(r.context, r.clusterInfo, cephBlockPool)
-		if err != nil {
-			return reconcile.Result{}, *cephBlockPool, err
-		}
-		if !deps.Empty() {
-			err := reporting.ReportDeletionBlockedDueToDependents(r.opManagerContext, logger, r.client, cephBlockPool, deps)
+		if err := r.handleDeletionBlocked(cephBlockPool, &cephCluster); err != nil {
 			return opcontroller.WaitForRequeueIfFinalizerBlocked, *cephBlockPool, err
 		}
-		reporting.ReportDeletionNotBlockedDueToDependents(r.opManagerContext, logger, r.client, r.recorder, cephBlockPool)
+
 		// If the ceph block pool is still in the map, we must remove it during CR deletion
 		// We must remove it first otherwise the checker will panic since the status/info will be nil
 		r.cancelMirrorMonitoring(cephBlockPool)
@@ -272,12 +268,6 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
 		logger.Infof("deleting pool %q", poolSpec.Name)
 		err = deletePool(r.context, clusterInfo, &poolSpec)
 		if err != nil {
-			if opcontroller.ForceDeleteRequested(cephBlockPool.GetAnnotations()) {
-				cleanupErr := r.cleanup(cephBlockPool, &cephCluster)
-				if cleanupErr != nil {
-					return reconcile.Result{}, *cephBlockPool, errors.Wrapf(cleanupErr, "failed to create clean up job for ceph blockpool %q", cephBlockPool.Name)
-				}
-			}
 			return opcontroller.ImmediateRetryResult, *cephBlockPool, errors.Wrapf(err, "failed to delete pool %q. ", cephBlockPool.Name)
 		}
 
@@ -417,6 +407,64 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
 	return reconcile.Result{}, *cephBlockPool, nil
 }
 
+// handlePoolDeletionBlocked updates the blockpool CR status with conditions about
+// whether the pool is empty or has dependents that block deletion.
+// If the pool is not empty and force deletion is specified, create a cleanup job
+// to delete the images and snapshots forcefully.
+func (r *ReconcileCephBlockPool) handleDeletionBlocked(cephBlockPool *cephv1.CephBlockPool, cephCluster *cephv1.CephCluster) error {
+	poolSpec := cephBlockPool.ToNamedPoolSpec()
+	deletionBlocked := false
+
+	deps, err := cephBlockPoolDependents(r.context, r.clusterInfo, cephBlockPool)
+	if err != nil {
+		return err
+	}
+	var depCondition cephv1.Condition
+	if deps.Empty() {
+		_, _, depCondition = reporting.GenerateConditionUnblockedDueToDependents(cephBlockPool)
+	} else {
+		deletionBlocked = true
+		_, _, depCondition = reporting.GenerateConditionBlockedDueToDependents(cephBlockPool, deps)
+	}
+	logger.Info(depCondition.Message)
+
+	radosNamespaces := deps.OfKind(radosNamespacesKeyName)
+	isEmpty, emptyMessage, err := cephclient.IsPoolEmpty(r.context, r.clusterInfo, poolSpec.Name, radosNamespaces)
+	if err != nil {
+		return err
+	}
+	var emptyCondition cephv1.Condition
+	if isEmpty {
+		emptyCondition = dependents.DeletionBlockedDueToNonEmptyPoolCondition(false, emptyMessage)
+	} else {
+		deletionBlocked = true
+		emptyCondition = dependents.DeletionBlockedDueToNonEmptyPoolCondition(true, emptyMessage)
+	}
+	logger.Info(emptyCondition.Message)
+
+	nsName := types.NamespacedName{Namespace: cephBlockPool.Namespace, Name: cephBlockPool.Name}
+	err = reporting.UpdateStatusConditionsWithRetry(
+		r.opManagerContext, r.client, cephBlockPool, nsName, cephBlockPool.Kind, emptyCondition, depCondition)
+	if err != nil {
+		logger.Warningf("failed to update %q status with deletion blocked conditions: %v", nsName.String(), err)
+	}
+
+	if !isEmpty {
+		// Force deletion if desired
+		if opcontroller.ForceDeleteRequested(cephBlockPool.GetAnnotations()) {
+			cleanupErr := r.cleanup(cephBlockPool, cephCluster)
+			if cleanupErr != nil {
+				return errors.Wrapf(cleanupErr, "failed to create clean up job for ceph blockpool %q", cephBlockPool.Name)
+			}
+		}
+	}
+
+	if deletionBlocked {
+		return errors.Errorf("pool %q cannot be deleted because it is not empty or has dependents", cephBlockPool.Name)
+	}
+	return nil
+}
+
 func (r *ReconcileCephBlockPool) reconcileCreatePool(clusterInfo *cephclient.ClusterInfo, cephCluster *cephv1.ClusterSpec, cephBlockPool *cephv1.CephBlockPool) (reconcile.Result, error) {
 	poolSpec := cephBlockPool.ToNamedPoolSpec()
 	err := createPool(r.context, clusterInfo, cephCluster, &poolSpec)

pkg/operator/ceph/pool/controller_test.go

@@ -188,7 +188,7 @@ func TestDeletePool(t *testing.T) {
 	// delete a pool that exists
 	p := &cephv1.NamedPoolSpec{Name: "mypool"}
 	err := deletePool(context, clusterInfo, p)
-	assert.Nil(t, err)
+	assert.NoError(t, err)
 
 	// succeed even if the pool doesn't exist
 	p = &cephv1.NamedPoolSpec{Name: "otherpool"}
@@ -199,7 +199,7 @@ func TestDeletePool(t *testing.T) {
 	failOnDelete = true
 	p = &cephv1.NamedPoolSpec{Name: "mypool"}
 	err = deletePool(context, clusterInfo, p)
-	assert.NotNil(t, err)
+	assert.Error(t, err)
 }
 
 // TestCephBlockPoolController runs ReconcileCephBlockPool.Reconcile() against a
@@ -565,6 +565,131 @@ func TestCephBlockPoolController(t *testing.T) {
 	})
 }
 
+func TestDeletionBlocked(t *testing.T) {
+	// A Pool resource with metadata and spec.
+	pool := &cephv1.CephBlockPool{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-pool",
+			Namespace: "ns",
+		},
+		TypeMeta: metav1.TypeMeta{
+			Kind: "CephBlockPool",
+		},
+		Status: &cephv1.CephBlockPoolStatus{},
+	}
+	object := []runtime.Object{pool}
+	// Register operator types with the runtime scheme.
+	s := scheme.Scheme
+	s.AddKnownTypes(cephv1.SchemeGroupVersion,
+		&cephv1.CephBlockPoolList{},
+		&cephv1.CephBlockPoolRadosNamespaceList{},
+		&cephv1.CephBlockPool{},
+		&cephv1.CephBlockPoolRadosNamespace{},
+	)
+	blockImageCount := 0
+	rnsImageCount := 0
+	cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
+	executor := &exectest.MockExecutor{
+		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			logger.Infof("Command: %s %v", command, args)
+			if args[0] == "pool" {
+				if args[1] == "stats" {
+					response := "{\"images\":{\"count\":%d,\"provisioned_bytes\":0,\"snap_count\":0},\"trash\":{\"count\":1,\"provisioned_bytes\":2048,\"snap_count\":0}}"
+					if args[4] == "--namespace" {
+						return fmt.Sprintf(response, rnsImageCount), nil
+					} else {
+						return fmt.Sprintf(response, blockImageCount), nil
+					}
+				}
+			}
+			return "not implemented", nil
+		},
+	}
+	c := &clusterd.Context{
+		Executor:      executor,
+		Clientset:     testop.New(t, 1),
+		RookClientset: rookclient.NewSimpleClientset(),
+	}
+	// Create a ReconcileCephBlockPool object with the scheme and fake client.
+	r := &ReconcileCephBlockPool{
+		client:            cl,
+		scheme:            s,
+		context:           c,
+		blockPoolContexts: make(map[string]*blockPoolHealth),
+		opManagerContext:  context.TODO(),
+		recorder:          record.NewFakeRecorder(5),
+		clusterInfo:       cephclient.AdminTestClusterInfo("mycluster"),
+	}
+	cephCluster := &cephv1.CephCluster{}
+	t.Run("deletion is allowed with no images or rns", func(t *testing.T) {
+		err := r.handleDeletionBlocked(pool, cephCluster)
+		assert.NoError(t, err)
+
+		result := &cephv1.CephBlockPool{}
+		err = r.client.Get(context.TODO(), types.NamespacedName{Name: pool.Name, Namespace: pool.Namespace}, result)
+		assert.NoError(t, err)
+		assert.Equal(t, pool.Name, result.Name)
+		assert.Equal(t, pool.Namespace, result.Namespace)
+		assert.Equal(t, 2, len(pool.Status.Conditions))
+		assert.Equal(t, "PoolDeletionIsBlocked", string(pool.Status.Conditions[0].Type))
+		assert.Equal(t, v1.ConditionFalse, pool.Status.Conditions[0].Status)
+		assert.Equal(t, "PoolEmpty", string(pool.Status.Conditions[0].Reason))
+		assert.Equal(t, "DeletionIsBlocked", string(pool.Status.Conditions[1].Type))
+		assert.Equal(t, v1.ConditionFalse, pool.Status.Conditions[1].Status)
+		assert.Equal(t, "ObjectHasNoDependents", string(pool.Status.Conditions[1].Reason))
+	})
+	t.Run("image prevents deletion", func(t *testing.T) {
+		blockImageCount = 1
+		err := r.handleDeletionBlocked(pool, cephCluster)
+		assert.Error(t, err)
+
+		result := &cephv1.CephBlockPool{}
+		err = r.client.Get(context.TODO(), types.NamespacedName{Name: pool.Name, Namespace: pool.Namespace}, result)
+		assert.NoError(t, err)
+		assert.Equal(t, pool.Name, result.Name)
+		assert.Equal(t, pool.Namespace, result.Namespace)
+		assert.Equal(t, 2, len(pool.Status.Conditions))
+		assert.Equal(t, "PoolDeletionIsBlocked", string(pool.Status.Conditions[0].Type))
+		assert.Equal(t, v1.ConditionTrue, pool.Status.Conditions[0].Status)
+		assert.Equal(t, "PoolNotEmpty", string(pool.Status.Conditions[0].Reason))
+		assert.Equal(t, "DeletionIsBlocked", string(pool.Status.Conditions[1].Type))
+		assert.Equal(t, v1.ConditionFalse, pool.Status.Conditions[1].Status)
+		assert.Equal(t, "ObjectHasNoDependents", string(pool.Status.Conditions[1].Reason))
+	})
+	t.Run("rados namespace prevents deletion", func(t *testing.T) {
+		radosNamespace := &cephv1.CephBlockPoolRadosNamespace{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "test-rns",
+				Namespace: pool.Namespace,
+			},
+			Spec: cephv1.CephBlockPoolRadosNamespaceSpec{
+				BlockPoolName: pool.Name,
+			},
+		}
+		_, err := c.RookClientset.CephV1().CephBlockPoolRadosNamespaces(pool.Namespace).Create(context.TODO(), radosNamespace, metav1.CreateOptions{})
+		assert.NoError(t, err)
+
+		// A radosnamespaces prevents deletion of the pool
+		blockImageCount = 0
+		rnsImageCount = 1
+		err = r.handleDeletionBlocked(pool, cephCluster)
+		assert.Error(t, err)
+
+		result := &cephv1.CephBlockPool{}
+		err = r.client.Get(context.TODO(), types.NamespacedName{Name: pool.Name, Namespace: pool.Namespace}, result)
+		assert.NoError(t, err)
+		assert.Equal(t, pool.Name, result.Name)
+		assert.Equal(t, pool.Namespace, result.Namespace)
+		assert.Equal(t, 2, len(pool.Status.Conditions))
+		assert.Equal(t, "PoolDeletionIsBlocked", string(pool.Status.Conditions[0].Type))
+		assert.Equal(t, v1.ConditionTrue, pool.Status.Conditions[0].Status)
+		assert.Equal(t, "PoolNotEmpty", string(pool.Status.Conditions[0].Reason))
+		assert.Equal(t, "DeletionIsBlocked", string(pool.Status.Conditions[1].Type))
+		assert.Equal(t, v1.ConditionTrue, pool.Status.Conditions[1].Status)
+		assert.Equal(t, "ObjectHasDependents", string(pool.Status.Conditions[1].Reason))
+	})
+}
+
 func TestIsAnyRadosNamespaceMirrored(t *testing.T) {
 	pool := "test"
 	object := []runtime.Object{}

pkg/operator/ceph/pool/dependents.go

@@ -20,16 +20,18 @@ import (
 	"fmt"
 
 	"github.com/pkg/errors"
-	v1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/pkg/util/dependents"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+const radosNamespacesKeyName = "CephBlockPoolRadosNamespaces"
+
 // cephBlockPoolDependents returns the rbd namespaces (s) which exist in the rbd pool that should block
 // deletion.
-func cephBlockPoolDependents(clusterdCtx *clusterd.Context, clusterInfo *client.ClusterInfo, blockpool *v1.CephBlockPool) (*dependents.DependentList, error) {
+func cephBlockPoolDependents(clusterdCtx *clusterd.Context, clusterInfo *client.ClusterInfo, blockpool *cephv1.CephBlockPool) (*dependents.DependentList, error) {
 	nsName := fmt.Sprintf("%s/%s", blockpool.Namespace, blockpool.Name)
 	baseErrMsg := fmt.Sprintf("failed to get dependents of CephBlockPool %q", nsName)
 
@@ -40,9 +42,9 @@ func cephBlockPoolDependents(clusterdCtx *clusterd.Context, clusterInfo *client.
 	if err != nil {
 		return deps, errors.Wrapf(err, "%s. failed to list CephBlockPoolRadosNamespace for CephBlockPool %q", baseErrMsg, nsName)
 	}
-	for _, namespace := range namespaces.Items {
+	for i, namespace := range namespaces.Items {
 		if namespace.Spec.BlockPoolName == blockpool.Name {
-			deps.Add("CephBlockPoolRadosNamespaces", namespace.Name)
+			deps.Add(radosNamespacesKeyName, cephv1.GetRadosNamespaceName(&namespaces.Items[i]))
 		}
 		logger.Debugf("found CephBlockPoolRadosNamespace %q that does not depend on CephBlockPool %q", namespace.Name, nsName)
 	}

pkg/operator/ceph/pool/radosnamespace/controller.go

@@ -32,6 +32,7 @@ import (
 	"github.com/rook/rook/pkg/operator/ceph/csi"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
 	"github.com/rook/rook/pkg/operator/k8sutil"
+	"github.com/rook/rook/pkg/util/dependents"
 
 	"github.com/coreos/pkg/capnslog"
 	"github.com/pkg/errors"
@@ -39,6 +40,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -71,6 +73,7 @@ type ReconcileCephBlockPoolRadosNamespace struct {
 	clusterInfo            *cephclient.ClusterInfo
 	radosNamespaceContexts map[string]*mirrorHealth
 	opManagerContext       context.Context
+	recorder               record.EventRecorder
 	opConfig               opcontroller.OperatorConfig
 }
 
@@ -95,6 +98,7 @@ func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerCont
 		context:                context,
 		radosNamespaceContexts: make(map[string]*mirrorHealth),
 		opManagerContext:       opManagerContext,
+		recorder:               mgr.GetEventRecorderFor("rook-" + controllerName),
 		opConfig:               opConfig,
 	}
 }
@@ -135,46 +139,46 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
 // otherwise upon completion it will remove the work from the queue.
 func (r *ReconcileCephBlockPoolRadosNamespace) Reconcile(context context.Context, request reconcile.Request) (reconcile.Result, error) {
 	// workaround because the rook logging mechanism is not compatible with the controller-runtime logging interface
-	reconcileResponse, err := r.reconcile(request)
+	reconcileResponse, radosNamespace, err := r.reconcile(request)
 	if err != nil {
-		logger.Errorf("failed to reconcile %q %v", request.NamespacedName, err)
+		logger.Errorf("failed to reconcile %q. %v", request.NamespacedName, err)
 	}
 
-	return reconcileResponse, err
+	return reporting.ReportReconcileResult(logger, r.recorder, request, radosNamespace, reconcileResponse, err)
 }
 
-func (r *ReconcileCephBlockPoolRadosNamespace) reconcile(request reconcile.Request) (reconcile.Result, error) {
+func (r *ReconcileCephBlockPoolRadosNamespace) reconcile(request reconcile.Request) (reconcile.Result, *cephv1.CephBlockPoolRadosNamespace, error) {
 	namespacedName := request.NamespacedName
 	// Fetch the CephBlockPoolRadosNamespace instance
-	cephBlockPoolRadosNamespace := &cephv1.CephBlockPoolRadosNamespace{}
-	err := r.client.Get(r.opManagerContext, request.NamespacedName, cephBlockPoolRadosNamespace)
+	radosNamespace := &cephv1.CephBlockPoolRadosNamespace{}
+	err := r.client.Get(r.opManagerContext, request.NamespacedName, radosNamespace)
 	if err != nil {
 		if kerrors.IsNotFound(err) {
 			logger.Debugf("cephBlockPoolRadosNamespace resource %q not found. Ignoring since object must be deleted.", namespacedName)
-			return reconcile.Result{}, nil
+			return reconcile.Result{}, radosNamespace, nil
 		}
 		// Error reading the object - requeue the request.
-		return reconcile.Result{}, errors.Wrap(err, "failed to get cephBlockPoolRadosNamespace")
+		return reconcile.Result{}, radosNamespace, errors.Wrap(err, "failed to get cephBlockPoolRadosNamespace")
 	}
 
 	// Set a finalizer so we can do cleanup before the object goes away
-	generationUpdated, err := opcontroller.AddFinalizerIfNotPresent(r.opManagerContext, r.client, cephBlockPoolRadosNamespace)
+	generationUpdated, err := opcontroller.AddFinalizerIfNotPresent(r.opManagerContext, r.client, radosNamespace)
 	if err != nil {
-		return reconcile.Result{}, errors.Wrap(err, "failed to add finalizer")
+		return reconcile.Result{}, radosNamespace, errors.Wrap(err, "failed to add finalizer")
 	}
 	if generationUpdated {
-		logger.Infof("reconciling the rados namespace %q after adding finalizer", cephBlockPoolRadosNamespace.Name)
-		return reconcile.Result{}, nil
+		logger.Infof("reconciling the rados namespace %q after adding finalizer", radosNamespace.Name)
+		return reconcile.Result{}, radosNamespace, nil
 	}
 
 	// The CR was just created, initializing status fields
-	if cephBlockPoolRadosNamespace.Status == nil {
+	if radosNamespace.Status == nil {
 		r.updateStatus(r.client, request.NamespacedName, cephv1.ConditionProgressing)
 	}
 
-	poolAndRadosNamespaceName := cephBlockPoolRadosNamespace.Spec.BlockPoolName
-	if rns := getRadosNamespaceName(cephBlockPoolRadosNamespace); rns != "" {
-		poolAndRadosNamespaceName = fmt.Sprintf("%s/%s", cephBlockPoolRadosNamespace.Spec.BlockPoolName, rns)
+	poolAndRadosNamespaceName := radosNamespace.Spec.BlockPoolName
+	if rns := cephv1.GetRadosNamespaceName(radosNamespace); rns != "" {
+		poolAndRadosNamespaceName = fmt.Sprintf("%s/%s", radosNamespace.Spec.BlockPoolName, rns)
 	}
 
 	// Make sure a CephCluster is present otherwise do nothing
@@ -186,94 +190,87 @@ func (r *ReconcileCephBlockPoolRadosNamespace) reconcile(request reconcile.Reque
 		// Also, only remove the finalizer if the CephCluster is gone
 		// If not, we should wait for it to be ready
 		// This handles the case where the operator is not ready to accept Ceph command but the cluster exists
-		if !cephBlockPoolRadosNamespace.GetDeletionTimestamp().IsZero() && !cephClusterExists {
+		if !radosNamespace.GetDeletionTimestamp().IsZero() && !cephClusterExists {
 			// don't leak the health checker routine if we are force-deleting
-			r.cancelMirrorMonitoring(radosNamespaceChannelKeyName(cephBlockPoolRadosNamespace.Namespace, poolAndRadosNamespaceName))
+			r.cancelMirrorMonitoring(radosNamespaceChannelKeyName(radosNamespace.Namespace, poolAndRadosNamespaceName))
 			// Remove finalizer
-			err = opcontroller.RemoveFinalizer(r.opManagerContext, r.client, cephBlockPoolRadosNamespace)
+			err = opcontroller.RemoveFinalizer(r.opManagerContext, r.client, radosNamespace)
 			if err != nil {
-				return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to remove finalizer")
+				return opcontroller.ImmediateRetryResult, radosNamespace, errors.Wrap(err, "failed to remove finalizer")
 			}
 
 			// Return and do not requeue. Successful deletion.
-			return reconcile.Result{}, nil
+			return reconcile.Result{}, radosNamespace, nil
 		}
-		return reconcileResponse, nil
+		return reconcileResponse, radosNamespace, nil
 	}
 
 	// Populate clusterInfo during each reconcile
 	r.clusterInfo, _, _, err = opcontroller.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace, &cephCluster.Spec)
 	if err != nil {
-		return reconcile.Result{}, errors.Wrap(err, "failed to populate cluster info")
+		return reconcile.Result{}, radosNamespace, errors.Wrap(err, "failed to populate cluster info")
 	}
 	r.clusterInfo.Context = r.opManagerContext
 
 	// DELETE: the CR was deleted
-	if !cephBlockPoolRadosNamespace.GetDeletionTimestamp().IsZero() {
+	if !radosNamespace.GetDeletionTimestamp().IsZero() {
 		logger.Debugf("delete cephBlockPoolRadosNamespace %q", namespacedName)
 		// On external cluster, we don't delete the rados namespace, it has to be deleted manually
 		if cephCluster.Spec.External.Enable {
 			logger.Warning("external rados namespace %q deletion is not supported, delete it manually", namespacedName)
 		} else {
-			err := r.deleteRadosNamespace(cephBlockPoolRadosNamespace)
-			if err != nil {
+			if containsImages, err := r.deleteRadosNamespace(radosNamespace, &cephCluster); err != nil {
+				if containsImages {
+					return opcontroller.WaitForRequeueIfFinalizerBlocked, radosNamespace, err
+				}
 				if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) {
 					logger.Info(opcontroller.OperatorNotInitializedMessage)
-					return opcontroller.WaitForRequeueIfOperatorNotInitialized, nil
+					return opcontroller.WaitForRequeueIfOperatorNotInitialized, radosNamespace, nil
 				}
-				if opcontroller.ForceDeleteRequested(cephBlockPoolRadosNamespace.GetAnnotations()) {
-					cleanupErr := r.cleanup(cephBlockPoolRadosNamespace, &cephCluster)
-					if cleanupErr != nil {
-						return reconcile.Result{}, errors.Wrapf(cleanupErr, "failed to create clean up job for ceph blockpool rados namespace %q", namespacedName)
-					}
-				}
-				return reconcile.Result{}, errors.Wrapf(err, "failed to delete ceph blockpool rados namespace %q", cephBlockPoolRadosNamespace.Name)
+				return reconcile.Result{}, radosNamespace, errors.Wrapf(err, "failed to delete ceph blockpool rados namespace %q", radosNamespace.Name)
 			}
 			// If the ceph block pool is still in the map, we must remove it during CR deletion
 			// We must remove it first otherwise the checker will panic since the status/info will be nil
-			r.cancelMirrorMonitoring(radosNamespaceChannelKeyName(cephBlockPoolRadosNamespace.Namespace, poolAndRadosNamespaceName))
+			r.cancelMirrorMonitoring(radosNamespaceChannelKeyName(radosNamespace.Namespace, poolAndRadosNamespaceName))
 		}
-		err = csi.SaveClusterConfig(r.context.Clientset, buildClusterID(cephBlockPoolRadosNamespace), cephCluster.Namespace, r.clusterInfo, nil)
+		err = csi.SaveClusterConfig(r.context.Clientset, buildClusterID(radosNamespace), cephCluster.Namespace, r.clusterInfo, nil)
 		if err != nil {
-			return reconcile.Result{}, errors.Wrap(err, "failed to save cluster config")
+			return reconcile.Result{}, radosNamespace, errors.Wrap(err, "failed to save cluster config")
 		}
 		// Remove finalizer
-		err = opcontroller.RemoveFinalizer(r.opManagerContext, r.client, cephBlockPoolRadosNamespace)
+		err = opcontroller.RemoveFinalizer(r.opManagerContext, r.client, radosNamespace)
 		if err != nil {
-			return reconcile.Result{}, errors.Wrap(err, "failed to remove finalizer")
+			return reconcile.Result{}, radosNamespace, errors.Wrap(err, "failed to remove finalizer")
 		}
 
 		// Return and do not requeue. Successful deletion.
-		return reconcile.Result{}, nil
+		return reconcile.Result{}, radosNamespace, nil
 	}
 
-	cephBlockPoolRadosNamespaceName := cephBlockPoolRadosNamespace.Name
-	if cephBlockPoolRadosNamespace.Spec.Name != "" {
-		cephBlockPoolRadosNamespaceName = cephBlockPoolRadosNamespace.Spec.Name
-	}
+	radosNamespaceName := cephv1.GetRadosNamespaceName(radosNamespace)
 
 	if cephCluster.Spec.External.Enable {
 		logger.Debug("skip creating external radosnamespace in external mode, create it manually, the controller will assume it's there")
-		err = r.updateClusterConfig(cephBlockPoolRadosNamespace, cephCluster)
+		err = r.updateClusterConfig(radosNamespace, cephCluster)
 		if err != nil {
-			return reconcile.Result{}, errors.Wrap(err, "failed to save cluster config")
+			return reconcile.Result{}, radosNamespace, errors.Wrap(err, "failed to save cluster config")
 		}
 		r.updateStatus(r.client, namespacedName, cephv1.ConditionReady)
 		if csi.EnableCSIOperator() {
-			err = csi.CreateUpdateClientProfileRadosNamespace(r.clusterInfo.Context, r.client, r.clusterInfo, cephBlockPoolRadosNamespaceName, buildClusterID(cephBlockPoolRadosNamespace), cephCluster.Name)
+			err = csi.CreateUpdateClientProfileRadosNamespace(r.clusterInfo.Context, r.client, r.clusterInfo, radosNamespaceName, buildClusterID(radosNamespace), cephCluster.Name)
 			if err != nil {
-				return reconcile.Result{}, errors.Wrap(err, "failed to create ceph csi-op config CR for RadosNamespace")
+				return reconcile.Result{}, radosNamespace, errors.Wrap(err, "failed to create ceph csi-op config CR for RadosNamespace")
 			}
 		}
-		return reconcile.Result{}, nil
+		return reconcile.Result{}, radosNamespace, nil
 	}
 
 	// cephversion check is only required for enabling mirroring
-	if cephBlockPoolRadosNamespace.Spec.Mirroring != nil {
+	if radosNamespace.Spec.Mirroring != nil {
 		// Get CephCluster version
 		cephVersion, err := opcontroller.GetImageVersion(cephCluster)
 		if err != nil {
-			return opcontroller.ImmediateRetryResult, errors.Wrapf(err, "failed to fetch ceph version from cephcluster %q running in namespace %q", cephCluster.Name, cephCluster.Namespace)
+			return opcontroller.ImmediateRetryResult, radosNamespace, errors.Wrapf(err, "failed to fetch ceph version from cephcluster %q running in namespace %q", cephCluster.Name, cephCluster.Namespace)
 		}
 		if cephVersion != nil {
 			r.clusterInfo.CephVersion = *cephVersion
@@ -283,65 +280,56 @@ func (r *ReconcileCephBlockPoolRadosNamespace) reconcile(request reconcile.Reque
 	// Build the NamespacedName to fetch the CephBlockPool and make sure it exists, if not we cannot
 	// create the rados namespace
 	cephBlockPool := &cephv1.CephBlockPool{}
-	pool := cephBlockPoolRadosNamespace.Spec.BlockPoolName
+	pool := radosNamespace.Spec.BlockPoolName
 	cephBlockPoolNamespacedName := types.NamespacedName{Name: pool, Namespace: request.Namespace}
 
 	err = r.client.Get(r.opManagerContext, cephBlockPoolNamespacedName, cephBlockPool)
 	if err != nil {
 		if kerrors.IsNotFound(err) {
-			return reconcile.Result{}, errors.Wrapf(err, "failed to fetch ceph blockpool %q, cannot create rados namespace %q", pool, cephBlockPoolRadosNamespace.Name)
+			return reconcile.Result{}, radosNamespace, errors.Wrapf(err, "failed to fetch ceph blockpool %q, cannot create rados namespace %q", pool, radosNamespace.Name)
 		}
 		// Error reading the object - requeue the request.
-		return reconcile.Result{}, errors.Wrap(err, "failed to get cephBlockPoolRadosNamespace")
+		return reconcile.Result{}, radosNamespace, errors.Wrap(err, "failed to get cephBlockPoolRadosNamespace")
 	}
 
 	// If the cephBlockPool is not ready to accept commands, we should wait for it to be ready
 	if cephBlockPool.Status.Phase != cephv1.ConditionReady {
 		// We know the CR is present so it should a matter of second for it to become ready
-		return reconcile.Result{Requeue: true, RequeueAfter: 10 * time.Second}, errors.Wrapf(err, "failed to fetch ceph blockpool %q, cannot create rados namespace %q", pool, cephBlockPoolRadosNamespace.Name)
+		return reconcile.Result{Requeue: true, RequeueAfter: 10 * time.Second}, radosNamespace, errors.Wrapf(err, "failed to fetch ceph blockpool %q, cannot create rados namespace %q", pool, radosNamespace.Name)
 	}
 	// Create or Update rados namespace
-	err = r.createOrUpdateRadosNamespace(cephBlockPoolRadosNamespace)
+	err = r.createOrUpdateRadosNamespace(radosNamespace)
 	if err != nil {
 		if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) {
 			logger.Info(opcontroller.OperatorNotInitializedMessage)
-			return opcontroller.WaitForRequeueIfOperatorNotInitialized, nil
+			return opcontroller.WaitForRequeueIfOperatorNotInitialized, radosNamespace, nil
 		}
 		r.updateStatus(r.client, request.NamespacedName, cephv1.ConditionFailure)
-		return reconcile.Result{}, errors.Wrapf(err, "failed to create or update ceph pool rados namespace %q", cephBlockPoolRadosNamespace.Name)
+		return reconcile.Result{}, radosNamespace, errors.Wrapf(err, "failed to create or update ceph pool rados namespace %q", radosNamespace.Name)
 	}
 
-	err = r.updateClusterConfig(cephBlockPoolRadosNamespace, cephCluster)
+	err = r.updateClusterConfig(radosNamespace, cephCluster)
 	if err != nil {
-		return reconcile.Result{}, errors.Wrap(err, "failed to save cluster config")
+		return reconcile.Result{}, radosNamespace, errors.Wrap(err, "failed to save cluster config")
 	}
 
-	err = r.reconcileMirroring(cephBlockPoolRadosNamespace, cephBlockPool)
+	err = r.reconcileMirroring(radosNamespace, cephBlockPool)
 	if err != nil {
-		return reconcile.Result{}, err
+		return reconcile.Result{}, radosNamespace, err
 	}
 
 	r.updateStatus(r.client, namespacedName, cephv1.ConditionReady)
 
 	if csi.EnableCSIOperator() {
-		err = csi.CreateUpdateClientProfileRadosNamespace(r.clusterInfo.Context, r.client, r.clusterInfo, cephBlockPoolRadosNamespaceName, buildClusterID(cephBlockPoolRadosNamespace), cephCluster.Name)
+		err = csi.CreateUpdateClientProfileRadosNamespace(r.clusterInfo.Context, r.client, r.clusterInfo, radosNamespaceName, buildClusterID(radosNamespace), cephCluster.Name)
 		if err != nil {
-			return reconcile.Result{}, errors.Wrap(err, "failed to create ceph csi-op config CR for RadosNamespace")
+			return reconcile.Result{}, radosNamespace, errors.Wrap(err, "failed to create ceph csi-op config CR for RadosNamespace")
 		}
 	}
 
 	// Return and do not requeue
 	logger.Debugf("done reconciling cephBlockPoolRadosNamespace %q", namespacedName)
-	return reconcile.Result{}, nil
-}
-
-func getRadosNamespaceName(cephBlockPoolRadosNamespace *cephv1.CephBlockPoolRadosNamespace) string {
-	if cephBlockPoolRadosNamespace.Spec.Name == cephclient.ImplicitNamespaceKey {
-		return cephclient.ImplicitNamespaceVal
-	} else if cephBlockPoolRadosNamespace.Spec.Name != "" {
-		return cephBlockPoolRadosNamespace.Spec.Name
-	}
-	return cephBlockPoolRadosNamespace.Name
+	return reconcile.Result{}, radosNamespace, nil
 }
 
 func (r *ReconcileCephBlockPoolRadosNamespace) updateClusterConfig(cephBlockPoolRadosNamespace *cephv1.CephBlockPoolRadosNamespace, cephCluster cephv1.CephCluster) error {
@@ -353,7 +341,7 @@ func (r *ReconcileCephBlockPoolRadosNamespace) updateClusterConfig(cephBlockPool
 		ClusterInfo: cephcsi.ClusterInfo{
 			Monitors: csi.MonEndpoints(r.clusterInfo.AllMonitors(), cephCluster.Spec.RequireMsgr2()),
 			RBD: cephcsi.RBD{
-				RadosNamespace: getRadosNamespaceName(cephBlockPoolRadosNamespace),
+				RadosNamespace: cephv1.GetRadosNamespaceName(cephBlockPoolRadosNamespace),
 			},
 			CephFS: cephcsi.CephFS{
 				KernelMountOptions: r.clusterInfo.CSIDriverSpec.CephFS.KernelMountOptions,
@@ -382,11 +370,11 @@ func (r *ReconcileCephBlockPoolRadosNamespace) createOrUpdateRadosNamespace(ceph
 	namespacedName := fmt.Sprintf("%s/%s", cephBlockPoolRadosNamespace.Namespace, cephBlockPoolRadosNamespace.Name)
 	logger.Infof("creating ceph blockpool rados namespace %q", namespacedName)
 
-	if getRadosNamespaceName(cephBlockPoolRadosNamespace) == "" {
+	if cephv1.GetRadosNamespaceName(cephBlockPoolRadosNamespace) == "" {
 		logger.Info("can't create empty radosnamespace %q in the namespace %q as it is already present", cephBlockPoolRadosNamespace.Name, cephBlockPoolRadosNamespace.Namespace)
 		return nil
 	}
-	err := cephclient.CreateRadosNamespace(r.context, r.clusterInfo, cephBlockPoolRadosNamespace.Spec.BlockPoolName, getRadosNamespaceName(cephBlockPoolRadosNamespace))
+	err := cephclient.CreateRadosNamespace(r.context, r.clusterInfo, cephBlockPoolRadosNamespace.Spec.BlockPoolName, cephv1.GetRadosNamespaceName(cephBlockPoolRadosNamespace))
 	if err != nil {
 		return errors.Wrapf(err, "failed to create ceph blockpool rados namespace %q", cephBlockPoolRadosNamespace.Name)
 	}
@@ -395,21 +383,53 @@ func (r *ReconcileCephBlockPoolRadosNamespace) createOrUpdateRadosNamespace(ceph
 }
 
 // Delete the ceph blockpool rados namespace
-func (r *ReconcileCephBlockPoolRadosNamespace) deleteRadosNamespace(cephBlockPoolRadosNamespace *cephv1.CephBlockPoolRadosNamespace) error {
-	namespacedName := fmt.Sprintf("%s/%s", cephBlockPoolRadosNamespace.Namespace, cephBlockPoolRadosNamespace.Name)
-	logger.Infof("deleting ceph blockpool rados namespace object %q", namespacedName)
+func (r *ReconcileCephBlockPoolRadosNamespace) deleteRadosNamespace(radosNamespace *cephv1.CephBlockPoolRadosNamespace, cephCluster *cephv1.CephCluster) (bool, error) {
+	nsName := types.NamespacedName{Namespace: radosNamespace.Namespace, Name: radosNamespace.Name}
+	logger.Infof("deleting rados namespace %q", nsName.String())
 
-	if getRadosNamespaceName(cephBlockPoolRadosNamespace) == "" {
-		logger.Info("can't delete empty radosnamepace as it present by default.")
-		return nil
+	name := cephv1.GetRadosNamespaceName(radosNamespace)
+	if name == "" {
+		logger.Info("no need to delete implicit radosnamepace")
+		return false, nil
 	}
 
-	if err := cephclient.DeleteRadosNamespace(r.context, r.clusterInfo, cephBlockPoolRadosNamespace.Spec.BlockPoolName, getRadosNamespaceName(cephBlockPoolRadosNamespace)); err != nil {
-		return errors.Wrapf(err, "failed to delete ceph blockpool rados namespace %q", cephBlockPoolRadosNamespace.Name)
+	containsImages, deleteErr := cephclient.DeleteRadosNamespace(r.context, r.clusterInfo, radosNamespace.Spec.BlockPoolName, name)
+	// If deleteErr is not nil, it means the deletion failed, but we still want to
+	// report a condition whether the rados namespace contains images
+	if containsImages {
+		var emptyCondition cephv1.Condition
+		if containsImages {
+			emptyCondition = dependents.DeletionBlockedDueToNonEmptyRadosNSCondition(
+				true,
+				fmt.Sprintf("rados namespace %q contains images or snapshots and cannot be deleted", radosNamespace.Name))
+		} else {
+			emptyCondition = dependents.DeletionBlockedDueToNonEmptyRadosNSCondition(
+				false,
+				fmt.Sprintf("rados namespace %q is empty and can be deleted", radosNamespace.Name))
+		}
+		logger.Info(emptyCondition.Message)
+
+		err := reporting.UpdateStatusConditionsWithRetry(
+			r.opManagerContext, r.client, radosNamespace, nsName, radosNamespace.Kind, emptyCondition)
+		if err != nil {
+			logger.Warningf("failed to update %q status with deletion blocked conditions: %v", nsName.String(), err)
+		}
+
+		// Force deletion if desired
+		if opcontroller.ForceDeleteRequested(radosNamespace.GetAnnotations()) {
+			cleanupErr := r.cleanup(radosNamespace, cephCluster)
+			if cleanupErr != nil {
+				return containsImages, errors.Wrapf(cleanupErr, "failed to create clean up job for rados namespace %q", radosNamespace.Name)
+			}
+		}
 	}
 
-	logger.Infof("deleted ceph blockpool rados namespace %q", namespacedName)
-	return nil
+	if deleteErr != nil {
+		return containsImages, errors.Wrapf(deleteErr, "failed to delete rados namespace %q", radosNamespace.Name)
+	}
+
+	logger.Infof("deleted rados namespace %q", nsName.String())
+	return false, nil
 }
 
 // updateStatus updates an object with a given status
@@ -437,7 +457,7 @@ func (r *ReconcileCephBlockPoolRadosNamespace) updateStatus(client client.Client
 }
 
 func buildClusterID(cephBlockPoolRadosNamespace *cephv1.CephBlockPoolRadosNamespace) string {
-	clusterID := fmt.Sprintf("%s-%s-block-%s", cephBlockPoolRadosNamespace.Namespace, cephBlockPoolRadosNamespace.Spec.BlockPoolName, getRadosNamespaceName(cephBlockPoolRadosNamespace))
+	clusterID := fmt.Sprintf("%s-%s-block-%s", cephBlockPoolRadosNamespace.Namespace, cephBlockPoolRadosNamespace.Spec.BlockPoolName, cephv1.GetRadosNamespaceName(cephBlockPoolRadosNamespace))
 	return k8sutil.Hash(clusterID)
 }
 
@@ -445,7 +465,7 @@ func (r *ReconcileCephBlockPoolRadosNamespace) cleanup(radosNamespace *cephv1.Ce
 	logger.Infof("starting cleanup of the ceph resources for radosNamespace %q in namespace %q", radosNamespace.Name, radosNamespace.Namespace)
 	cleanupConfig := map[string]string{
 		opcontroller.CephBlockPoolNameEnv:           radosNamespace.Spec.BlockPoolName,
-		opcontroller.CephBlockPoolRadosNamespaceEnv: getRadosNamespaceName(radosNamespace),
+		opcontroller.CephBlockPoolRadosNamespaceEnv: cephv1.GetRadosNamespaceName(radosNamespace),
 	}
 	cleanup := opcontroller.NewResourceCleanup(radosNamespace, cephCluster, r.opConfig.Image, cleanupConfig)
 	jobName := k8sutil.TruncateNodeNameForJob("cleanup-radosnamespace-%s", fmt.Sprintf("%s-%s", radosNamespace.Spec.BlockPoolName, radosNamespace.Name))
@@ -461,8 +481,8 @@ func checkBlockPoolMirroring(cephBlockPool *cephv1.CephBlockPool) bool {
 }
 
 func (r *ReconcileCephBlockPoolRadosNamespace) reconcileMirroring(cephBlockPoolRadosNamespace *cephv1.CephBlockPoolRadosNamespace, cephBlockPool *cephv1.CephBlockPool) error {
-	poolAndRadosNamespaceName := fmt.Sprintf("%s/%s", cephBlockPool.Name, getRadosNamespaceName(cephBlockPoolRadosNamespace))
-	if getRadosNamespaceName(cephBlockPoolRadosNamespace) == "" {
+	poolAndRadosNamespaceName := fmt.Sprintf("%s/%s", cephBlockPool.Name, cephv1.GetRadosNamespaceName(cephBlockPoolRadosNamespace))
+	if cephv1.GetRadosNamespaceName(cephBlockPoolRadosNamespace) == "" {
 		poolAndRadosNamespaceName = cephBlockPool.Name
 	}
 

pkg/operator/ceph/pool/radosnamespace/controller_test.go

@@ -27,7 +27,6 @@ import (
 	rookclient "github.com/rook/rook/pkg/client/clientset/versioned/fake"
 	"github.com/rook/rook/pkg/client/clientset/versioned/scheme"
 	"github.com/rook/rook/pkg/clusterd"
-	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/csi"
 	"github.com/rook/rook/pkg/operator/k8sutil"
@@ -38,6 +37,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
@@ -106,6 +106,7 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 		context:          c,
 		opManagerContext: ctx,
 		opConfig:         opcontroller.OperatorConfig{Image: "ceph/ceph:v14.2.9"},
+		recorder:         record.NewFakeRecorder(5),
 	}
 
 	// Mock request to simulate Reconcile() being called on an event for a
@@ -142,6 +143,9 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 	s.AddKnownTypes(cephv1.SchemeGroupVersion, &cephv1.CephCluster{}, &cephv1.CephClusterList{})
 
 	t.Run("error - no ceph cluster", func(t *testing.T) {
+		// Create pool for updating status
+		_, err := c.RookClientset.CephV1().CephBlockPoolRadosNamespaces(namespace).Create(ctx, cephBlockPoolRadosNamespace, metav1.CreateOptions{})
+		assert.NoError(t, err)
 		res, err := r.Reconcile(ctx, req)
 		assert.NoError(t, err)
 		assert.True(t, res.Requeue)
@@ -152,7 +156,11 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 		// Create a fake client to mock API calls.
 		cl = fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
 		// Create a ReconcileCephBlockPoolRadosNamespace object with the scheme and fake client.
-		r = &ReconcileCephBlockPoolRadosNamespace{client: cl, scheme: s, context: c, opManagerContext: context.TODO(), opConfig: opcontroller.OperatorConfig{Image: "ceph/ceph:v14.2.9"}}
+		r = &ReconcileCephBlockPoolRadosNamespace{
+			client: cl, scheme: s, context: c, opManagerContext: context.TODO(),
+			recorder: record.NewFakeRecorder(5),
+			opConfig: opcontroller.OperatorConfig{Image: "ceph/ceph:v14.2.9"},
+		}
 		res, err := r.Reconcile(ctx, req)
 		assert.NoError(t, err)
 		assert.True(t, res.Requeue)
@@ -227,6 +235,7 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 			opManagerContext:       context.TODO(),
 			opConfig:               opcontroller.OperatorConfig{Image: "ceph/ceph:v14.2.9"},
 			radosNamespaceContexts: make(map[string]*mirrorHealth),
+			recorder:               record.NewFakeRecorder(5),
 		}
 
 		// Enable CSI
@@ -279,6 +288,7 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 			context:          c,
 			opManagerContext: ctx,
 			opConfig:         opcontroller.OperatorConfig{Image: "ceph/ceph:v14.2.9"},
+			recorder:         record.NewFakeRecorder(5),
 		}
 
 		// Enable CSI
@@ -346,6 +356,7 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 			opManagerContext:       context.TODO(),
 			opConfig:               opcontroller.OperatorConfig{Image: "ceph/ceph:v14.2.9"},
 			radosNamespaceContexts: make(map[string]*mirrorHealth),
+			recorder:               record.NewFakeRecorder(5),
 		}
 
 		res, err := r.Reconcile(ctx, req)
@@ -392,6 +403,7 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 			opManagerContext:       context.TODO(),
 			opConfig:               opcontroller.OperatorConfig{Image: "ceph/ceph:v14.2.9"},
 			radosNamespaceContexts: make(map[string]*mirrorHealth),
+			recorder:               record.NewFakeRecorder(5),
 		}
 
 		res, err := r.Reconcile(ctx, req)
@@ -442,6 +454,7 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 			opManagerContext:       context.TODO(),
 			opConfig:               opcontroller.OperatorConfig{Image: "ceph/ceph:v14.2.9"},
 			radosNamespaceContexts: make(map[string]*mirrorHealth),
+			recorder:               record.NewFakeRecorder(5),
 		}
 
 		res, err := r.Reconcile(ctx, req)
@@ -492,6 +505,7 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 			opManagerContext:       context.TODO(),
 			opConfig:               opcontroller.OperatorConfig{Image: "ceph/ceph:v14.2.9"},
 			radosNamespaceContexts: make(map[string]*mirrorHealth),
+			recorder:               record.NewFakeRecorder(5),
 		}
 
 		res, err := r.Reconcile(ctx, req)
@@ -544,6 +558,7 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 			opManagerContext:       context.TODO(),
 			opConfig:               opcontroller.OperatorConfig{Image: "ceph/ceph:v14.2.9"},
 			radosNamespaceContexts: make(map[string]*mirrorHealth),
+			recorder:               record.NewFakeRecorder(5),
 		}
 
 		res, err := r.Reconcile(ctx, req)
@@ -600,6 +615,7 @@ func TestCephBlockPoolRadosNamespaceController(t *testing.T) {
 			opManagerContext:       context.TODO(),
 			opConfig:               opcontroller.OperatorConfig{Image: "ceph/ceph:v14.2.9"},
 			radosNamespaceContexts: make(map[string]*mirrorHealth),
+			recorder:               record.NewFakeRecorder(5),
 		}
 
 		res, err := r.Reconcile(ctx, req)
@@ -629,9 +645,9 @@ func TestGetRadosNamespaceName(t *testing.T) {
 			"implicit-namespace",
 			cephv1.CephBlockPoolRadosNamespace{
 				ObjectMeta: metav1.ObjectMeta{Name: "cr-name1"},
-				Spec:       cephv1.CephBlockPoolRadosNamespaceSpec{Name: cephclient.ImplicitNamespaceKey},
+				Spec:       cephv1.CephBlockPoolRadosNamespaceSpec{Name: cephv1.ImplicitNamespaceKey},
 			},
-			cephclient.ImplicitNamespaceVal,
+			cephv1.ImplicitNamespaceVal,
 		},
 		{
 			"valid-namespace",
@@ -653,8 +669,8 @@ func TestGetRadosNamespaceName(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if got := getRadosNamespaceName(&tt.args); got != tt.want {
-				t.Errorf("getRadosNamespaceName() = %v, want %v", got, tt.want)
+			if got := cephv1.GetRadosNamespaceName(&tt.args); got != tt.want {
+				t.Errorf("GetRadosNamespaceName() = %v, want %v", got, tt.want)
 			}
 		})
 	}

pkg/operator/ceph/reporting/reporting.go

@@ -169,6 +169,28 @@ func ReportReconcileResult(
 	return reconcileResponse, err
 }
 
+func GenerateConditionBlockedDueToDependents(obj statusConditionGetter, deps *dependents.DependentList) (string, types.NamespacedName, cephv1.Condition) {
+	kind := obj.GetObjectKind().GroupVersionKind().Kind
+	nsName := types.NamespacedName{
+		Namespace: obj.GetNamespace(),
+		Name:      obj.GetName(),
+	}
+	blockedMsg := deps.StringWithHeader("%s %q will not be deleted until all dependents are removed", kind, nsName.String())
+
+	return kind, nsName, dependents.DeletionBlockedDueToDependentsCondition(true, blockedMsg)
+}
+
+func GenerateConditionUnblockedDueToDependents(obj statusConditionGetter) (string, types.NamespacedName, cephv1.Condition) {
+	kind := obj.GetObjectKind().GroupVersionKind().Kind
+	nsName := types.NamespacedName{
+		Namespace: obj.GetNamespace(),
+		Name:      obj.GetName(),
+	}
+	safeMsg := fmt.Sprintf("%s %q has no dependent custom resources blocking deletion", kind, nsName.String())
+
+	return kind, nsName, dependents.DeletionBlockedDueToDependentsCondition(false, safeMsg)
+}
+
 // ReportDeletionBlockedDueToDependents reports that deletion of a Rook-Ceph object is blocked due
 // to the given dependents in 3 ways:
 // 1. to the given logger
@@ -177,33 +199,33 @@ func ReportReconcileResult(
 func ReportDeletionBlockedDueToDependents(
 	ctx context.Context, logger *capnslog.PackageLogger, client client.Client, obj statusConditionGetter, deps *dependents.DependentList,
 ) error {
-	kind := obj.GetObjectKind().GroupVersionKind().Kind
-	nsName := types.NamespacedName{
-		Namespace: obj.GetNamespace(),
-		Name:      obj.GetName(),
-	}
-	blockedMsg := deps.StringWithHeader("%s %q will not be deleted until all dependents are removed", kind, nsName.String())
+	kind, nsName, blockedCond := GenerateConditionBlockedDueToDependents(obj, deps)
 
 	// 1. log
-	logger.Info(blockedMsg)
+	logger.Info(blockedCond.Message)
 
 	// 2. condition
-	blockedCond := dependents.DeletionBlockedDueToDependentsCondition(true, blockedMsg)
-	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
-		if err := client.Get(ctx, nsName, obj); err != nil {
-			return errors.Wrapf(err, "failed to get latest %s %q", kind, nsName.String())
-		}
-		if err := UpdateStatusCondition(client, obj, blockedCond); err != nil {
-			return err
-		}
-		return nil
-	})
-	if err != nil {
-		return errors.Wrapf(err, "on condition %s", blockedMsg)
+	if err := UpdateStatusConditionsWithRetry(ctx, client, obj, nsName, kind, blockedCond); err != nil {
+		return err
 	}
 
 	// 3. error for later FailedReconcile message
-	return errors.New(blockedMsg)
+	return errors.New(blockedCond.Message)
+}
+
+func UpdateStatusConditionsWithRetry(
+	ctx context.Context, client client.Client, obj statusConditionGetter,
+	nsName types.NamespacedName, kind string, conditions ...cephv1.Condition,
+) error {
+	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		if err := client.Get(ctx, nsName, obj); err != nil {
+			return errors.Wrapf(err, "failed to get latest %s %q", kind, nsName.String())
+		}
+		if err := UpdateStatusCondition(client, obj, conditions...); err != nil {
+			return errors.Wrapf(err, "failed to update %s %q status conditions", kind, nsName.String())
+		}
+		return nil
+	})
 }
 
 // ReportDeletionNotBlockedDueToDependents reports that deletion of a Rook-Ceph object is proceeding
@@ -214,32 +236,17 @@ func ReportDeletionBlockedDueToDependents(
 func ReportDeletionNotBlockedDueToDependents(
 	ctx context.Context, logger *capnslog.PackageLogger, client client.Client, recorder record.EventRecorder, obj statusConditionGetter,
 ) {
-	kind := obj.GetObjectKind().GroupVersionKind().Kind
-	nsName := types.NamespacedName{
-		Namespace: obj.GetNamespace(),
-		Name:      obj.GetName(),
-	}
-	safeMsg := fmt.Sprintf("%s %q can be deleted safely", kind, nsName.String())
+	kind, nsName, unblockedCond := GenerateConditionUnblockedDueToDependents(obj)
 	deletingMsg := fmt.Sprintf("deleting %s %q", kind, nsName.String())
 
 	// 1. log
-	logger.Infof("%s. %s", safeMsg, deletingMsg)
+	logger.Infof("%s. %s", unblockedCond.Message, deletingMsg)
 
 	// 2. event
 	recorder.Event(obj, corev1.EventTypeNormal, string(cephv1.DeletingReason), deletingMsg)
 
 	// 3. condition
-	unblockedCond := dependents.DeletionBlockedDueToDependentsCondition(false, safeMsg)
-	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
-		if err := client.Get(ctx, nsName, obj); err != nil {
-			return errors.Wrapf(err, "failed to get latest %s %q", kind, nsName.String())
-		}
-		if err := UpdateStatusCondition(client, obj, unblockedCond); err != nil {
-			return err
-		}
-		return nil
-	})
-	if err != nil {
+	if err := UpdateStatusConditionsWithRetry(ctx, client, obj, nsName, kind, unblockedCond); err != nil {
 		logger.Warningf("continuing deletion of %s %q without setting the condition. %v", kind, nsName.String(), err)
 	}
 }

pkg/operator/ceph/reporting/status.go

@@ -51,14 +51,17 @@ func UpdateStatus(client client.Client, obj client.Object) error {
 // UpdateStatusCondition updates (or adds to) the status condition to the given object. The object
 // is updated with the latest version from the server on a successful update.
 func UpdateStatusCondition(
-	client client.Client, obj statusConditionGetter, newCond cephv1.Condition,
+	client client.Client, obj statusConditionGetter, newConditions ...cephv1.Condition,
 ) error {
 	kind := obj.GetObjectKind().GroupVersionKind().Kind
 	nsName := fmt.Sprintf("%s/%s", obj.GetNamespace(), obj.GetName())
 
-	cephv1.SetStatusCondition(obj.GetStatusConditions(), newCond)
+	for _, newCondition := range newConditions {
+		cephv1.SetStatusCondition(obj.GetStatusConditions(), newCondition)
+	}
+
 	if err := UpdateStatus(client, obj); err != nil {
-		return errors.Wrapf(err, "failed to update %s %q status condition %s=%s", kind, nsName, newCond.Type, newCond.Status)
+		return errors.Wrapf(err, "failed to update %s %q status conditions", kind, nsName)
 	}
 
 	return nil

pkg/operator/ceph/reporting/status_test.go

@@ -114,20 +114,38 @@ func TestUpdateStatusCondition(t *testing.T) {
 			Phase: cephv1.ConditionDeleting,
 		},
 	}
+	fakeBlock := &cephv1.CephBlockPool{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:       "mypool",
+			Namespace:  "rook-ceph",
+			Finalizers: []string{},
+		},
+		Status: &cephv1.CephBlockPoolStatus{
+			Phase: cephv1.ConditionDeleting,
+		},
+	}
 	nsName := types.NamespacedName{
 		Namespace: fakeObject.Namespace,
 		Name:      fakeObject.Name,
 	}
+	poolName := types.NamespacedName{
+		Namespace: fakeBlock.Namespace,
+		Name:      fakeBlock.Name,
+	}
 
 	s := scheme.Scheme
-	s.AddKnownTypes(cephv1.SchemeGroupVersion, fakeObject)
-	cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(fakeObject.DeepCopy()).Build()
+	s.AddKnownTypes(cephv1.SchemeGroupVersion, fakeObject, fakeBlock)
+	cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(fakeObject.DeepCopy(), fakeBlock.DeepCopy()).Build()
 
-	// get version of the object in the fake object tracker
+	// get version of the objects in the fake object tracker
 	getObj := &cephv1.CephObjectStore{}
 	err := cl.Get(context.TODO(), nsName, getObj)
 	assert.NoError(t, err)
 	assert.Zero(t, len(getObj.Status.Conditions))
+	getBlock := &cephv1.CephBlockPool{}
+	err = cl.Get(context.TODO(), poolName, getBlock)
+	assert.NoError(t, err)
+	assert.Zero(t, len(getBlock.Status.Conditions))
 
 	t.Run("add new status", func(t *testing.T) {
 		getObj := &cephv1.CephObjectStore{}
@@ -152,6 +170,40 @@ func TestUpdateStatusCondition(t *testing.T) {
 		assert.Equal(t, "start", cond.Message)
 	})
 
+	t.Run("add two statuses", func(t *testing.T) {
+		getObj := &cephv1.CephBlockPool{}
+		err := cl.Get(context.TODO(), poolName, getObj)
+		assert.NoError(t, err)
+
+		blockedCond := cephv1.Condition{
+			Type:    cephv1.ConditionDeletionIsBlocked,
+			Status:  v1.ConditionTrue,
+			Reason:  cephv1.ConditionReason("dep"),
+			Message: "there is a dependency",
+		}
+		emptyCond := cephv1.Condition{
+			Type:    cephv1.ConditionPoolDeletionIsBlocked,
+			Status:  v1.ConditionTrue,
+			Reason:  cephv1.ConditionReason("notempty"),
+			Message: "images are in the pool",
+		}
+
+		err = UpdateStatusCondition(cl, getObj, blockedCond, emptyCond)
+		assert.NoError(t, err)
+
+		err = cl.Get(context.TODO(), poolName, getObj)
+		assert.NoError(t, err)
+		cond := getObj.Status.Conditions[0]
+		assert.Equal(t, v1.ConditionTrue, cond.Status)
+		assert.Equal(t, blockedCond.Reason, cond.Reason)
+		assert.Equal(t, blockedCond.Message, cond.Message)
+
+		cond = getObj.Status.Conditions[1]
+		assert.Equal(t, v1.ConditionTrue, cond.Status)
+		assert.Equal(t, emptyCond.Reason, cond.Reason)
+		assert.Equal(t, emptyCond.Message, cond.Message)
+	})
+
 	t.Run("update status", func(t *testing.T) {
 		getObj := &cephv1.CephObjectStore{}
 		err := cl.Get(context.TODO(), nsName, getObj)

pkg/util/dependents/dependents.go

@@ -40,6 +40,36 @@ func DeletionBlockedDueToDependentsCondition(blocked bool, message string) cephv
 	}
 }
 
+func DeletionBlockedDueToNonEmptyPoolCondition(blocked bool, message string) cephv1.Condition {
+	status := corev1.ConditionFalse
+	reason := cephv1.PoolEmptyReason
+	if blocked {
+		status = corev1.ConditionTrue
+		reason = cephv1.PoolNotEmptyReason
+	}
+	return cephv1.Condition{
+		Type:    cephv1.ConditionPoolDeletionIsBlocked,
+		Status:  status,
+		Reason:  reason,
+		Message: message,
+	}
+}
+
+func DeletionBlockedDueToNonEmptyRadosNSCondition(blocked bool, message string) cephv1.Condition {
+	status := corev1.ConditionFalse
+	reason := cephv1.RadosNamespaceEmptyReason
+	if blocked {
+		status = corev1.ConditionTrue
+		reason = cephv1.RadosNamespaceNotEmptyReason
+	}
+	return cephv1.Condition{
+		Type:    cephv1.ConditionRadosNSDeletionIsBlocked,
+		Status:  status,
+		Reason:  reason,
+		Message: message,
+	}
+}
+
 // A DependentList represents a list of dependents of a resource. Each dependent has a plural Kind
 // and a list of names of dependent resources.
 type DependentList struct {