Merge pull request #8012 from BlaineEXE/implement-dependencies

Ceph: Implement dependencies

Documentation/ceph-cluster-crd.md

@@ -1465,36 +1465,6 @@ spec:
     image: ceph/ceph:v16.2.4 # Should match external cluster version
 ```
 
-### Cleanup policy
-
-Rook has the ability to cleanup resources and data that were deployed when a CephCluster is removed.
-The policy settings indicate which data should be forcibly deleted and in what way the data should be wiped.
-The `cleanupPolicy` has several fields:
-
-* `confirmation`: Only an empty string and `yes-really-destroy-data` are valid values for this field.
-  If this setting is empty, the cleanupPolicy settings will be ignored and Rook will not cleanup any resources during cluster removal.
-  To reinstall the cluster, the admin would then be required to follow the [cleanup guide](ceph-teardown.md) to delete the data on hosts.
-  If this setting is `yes-really-destroy-data`, the operator will automatically delete the data on hosts.
-  Because this cleanup policy is destructive, after the confirmation is set to `yes-really-destroy-data`
-  Rook will stop configuring the cluster as if the cluster is about to be destroyed.
-* `sanitizeDisks`: sanitizeDisks represents advanced settings that can be used to delete data on drives.
-  * `method`: indicates if the entire disk should be sanitized or simply ceph's metadata. Possible choices are 'quick' (default) or 'complete'
-  * `dataSource`: indicate where to get random bytes from to write on the disk. Possible choices are 'zero' (default) or 'random'.
-  Using random sources will consume entropy from the system and will take much more time then the zero source
-  * `iteration`: overwrite N times instead of the default (1). Takes an integer value
-* `allowUninstallWithVolumes`: If set to true, then the cephCluster deletion doesn't wait for the PVCs to be deleted. Default is false.
-
-To automate activation of the cleanup, you can use the following command. **WARNING: DATA WILL BE PERMANENTLY DELETED**:
-
-```console
-kubectl -n rook-ceph patch cephcluster rook-ceph --type merge -p '{"spec":{"cleanupPolicy":{"confirmation":"yes-really-destroy-data"}}}'
-```
-
-Nothing will happen until the deletion of the CR is requested, so this can still be reverted.
-However, all new configuration by the operator will be blocked with this cleanup policy enabled.
-
-Rook waits for the deletion of PVs provisioned using the cephCluster before proceeding to delete the cephCluster. To force deletion of the cephCluster without waiting for the PVs to be deleted,  you can set the allowUninstallWithVolumes to true under spec.CleanupPolicy.
-
 ### Security
 
 Rook has the ability to encrypt OSDs of clusters running on PVC via the flag (`encrypted: true`) in your `storageClassDeviceSets` [template](#pvc-based-cluster).
@@ -1619,3 +1589,45 @@ data:
 
 Note: if you are using self-signed certificates (not known/approved by a proper CA) you must pass `VAULT_SKIP_VERIFY: true`.
 Communications will remain encrypted but the validity of the certificate will not be verified.
+
+### Deleting a CephCluster
+
+During deletion of a CephCluster resource, Rook protects against accidental or premature destruction
+of user data by blocking deletion if there are any other Rook-Ceph Custom Resources that reference
+the CephCluster being deleted. Rook will warn about which other resources are blocking deletion in
+three ways until all blocking resources are deleted:
+1. An event will be registered on the CephCluster resource
+1. A status condition will be added to the CephCluster resource
+1. An error will be added to the Rook-Ceph Operator log
+
+#### Cleanup policy
+
+Rook has the ability to cleanup resources and data that were deployed when a CephCluster is removed.
+The policy settings indicate which data should be forcibly deleted and in what way the data should be wiped.
+The `cleanupPolicy` has several fields:
+
+* `confirmation`: Only an empty string and `yes-really-destroy-data` are valid values for this field.
+  If this setting is empty, the cleanupPolicy settings will be ignored and Rook will not cleanup any resources during cluster removal.
+  To reinstall the cluster, the admin would then be required to follow the [cleanup guide](ceph-teardown.md) to delete the data on hosts.
+  If this setting is `yes-really-destroy-data`, the operator will automatically delete the data on hosts.
+  Because this cleanup policy is destructive, after the confirmation is set to `yes-really-destroy-data`
+  Rook will stop configuring the cluster as if the cluster is about to be destroyed.
+* `sanitizeDisks`: sanitizeDisks represents advanced settings that can be used to delete data on drives.
+  * `method`: indicates if the entire disk should be sanitized or simply ceph's metadata. Possible choices are 'quick' (default) or 'complete'
+  * `dataSource`: indicate where to get random bytes from to write on the disk. Possible choices are 'zero' (default) or 'random'.
+  Using random sources will consume entropy from the system and will take much more time then the zero source
+  * `iteration`: overwrite N times instead of the default (1). Takes an integer value
+* `allowUninstallWithVolumes`: If set to true, then the cephCluster deletion doesn't wait for the PVCs to be deleted. Default is false.
+
+To automate activation of the cleanup, you can use the following command. **WARNING: DATA WILL BE PERMANENTLY DELETED**:
+
+```console
+kubectl -n rook-ceph patch cephcluster rook-ceph --type merge -p '{"spec":{"cleanupPolicy":{"confirmation":"yes-really-destroy-data"}}}'
+```
+
+Nothing will happen until the deletion of the CR is requested, so this can still be reverted.
+However, all new configuration by the operator will be blocked with this cleanup policy enabled.
+
+Rook waits for the deletion of PVs provisioned using the cephCluster before proceeding to delete the
+cephCluster. To force deletion of the cephCluster without waiting for the PVs to be deleted, you can
+set the allowUninstallWithVolumes to true under spec.CleanupPolicy.

PendingReleaseNotes.md

@@ -9,6 +9,11 @@ v1.7...
 ## Breaking Changes
 
 ### Ceph
+- Add user data protection when deleting Rook-Ceph Custom Resources
+  - A CephCluster will not be deleted if there are any other Rook-Ceph Custom resources referencing
+    it with the assumption that they are using the underlying Ceph cluster.
+  - See [the design](https://github.com/rook/rook/blob/master/design/ceph/resource-dependencies.md)
+    for detailed information.
 
 ## Features
 

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

@@ -3682,7 +3682,7 @@ spec:
                   type: object
                 conditions:
                   items:
-                    description: Condition represents
+                    description: Condition represents a status condition on any Rook-Ceph Custom Resource.
                     properties:
                       lastHeartbeatTime:
                         format: date-time
@@ -3693,7 +3693,7 @@ spec:
                       message:
                         type: string
                       reason:
-                        description: ClusterReasonType is cluster reason
+                        description: ConditionReason is a reason for a condition
                         type: string
                       status:
                         type: string

cluster/examples/kubernetes/ceph/crds.yaml

@@ -3682,7 +3682,7 @@ spec:
                   type: object
                 conditions:
                   items:
-                    description: Condition represents
+                    description: Condition represents a status condition on any Rook-Ceph Custom Resource.
                     properties:
                       lastHeartbeatTime:
                         format: date-time
@@ -3693,7 +3693,7 @@ spec:
                       message:
                         type: string
                       reason:
-                        description: ClusterReasonType is cluster reason
+                        description: ConditionReason is a reason for a condition
                         type: string
                       status:
                         type: string

cmd/rook/rook/rook.go

@@ -35,6 +35,7 @@ import (
 	"github.com/tevino/abool"
 	apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
 	"k8s.io/apimachinery/pkg/util/uuid"
+	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
@@ -151,6 +152,11 @@ func NewContext() *clusterd.Context {
 	context.Clientset, err = kubernetes.NewForConfig(context.KubeConfig)
 	TerminateOnError(err, "failed to create k8s clientset")
 
+	// Dynamic clientset allows dealing with resources that aren't statically typed but determined
+	// at runtime.
+	context.DynamicClientset, err = dynamic.NewForConfig(context.KubeConfig)
+	TerminateOnError(err, "failed to create dynamic clientset")
+
 	context.APIExtensionClientset, err = apiextensionsclient.NewForConfig(context.KubeConfig)
 	TerminateOnError(err, "failed to create k8s API extension clientset")
 

design/ceph/resource-dependencies.md

@@ -200,7 +200,7 @@ Reported statuses will be modified as follows:
 1. Object's status.phase should be changed to "Deleting" as soon as a deletion timestamp is detected
    and never changed
 1. A status.condition should be added if the operator is blocking deletion:
-   - Type: "BlockingDeletion"
+   - Type: "DeletionIsBlocked"
    - Status: "True"
    - Reason: "ObjectHasDependents"
    - Message: "object deletion is blocked because it has dependents:" followed by a full list of
@@ -210,7 +210,7 @@ Reported statuses will be modified as follows:
 **Event:**
 Reported events will have the following content:
 - Type: "Warning"
-- Reason: "ObjectHasDependents"
+- Reason: "ReconcileFailed"
 - Message: "object deletion is blocked because it has dependents:" followed by a full list of which
   dependents exist of which Kubernetes Object Kinds (e.g., CephBlockPools or PersistentVolumes) or
   of which Ceph kinds (e.g., pools or buckets).

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

@@ -0,0 +1,76 @@
+/*
+Copyright 2021 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
+
+import (
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// Based on code from https://github.com/kubernetes/apimachinery/blob/master/pkg/api/meta/conditions.go
+
+// SetStatusCondition sets the corresponding condition in conditions to newCondition.
+// conditions must be non-nil.
+// 1. if the condition of the specified type already exists (all fields of the existing condition are updated to
+//    newCondition, LastTransitionTime is set to now if the new status differs from the old status)
+// 2. if a condition of the specified type does not exist (LastTransitionTime is set to now() if unset, and newCondition is appended)
+func SetStatusCondition(conditions *[]Condition, newCondition Condition) {
+	if conditions == nil {
+		return
+	}
+
+	now := metav1.NewTime(time.Now())
+
+	existingCondition := FindStatusCondition(*conditions, newCondition.Type)
+	if existingCondition == nil {
+		if newCondition.LastTransitionTime.IsZero() {
+			newCondition.LastTransitionTime = now
+			newCondition.LastHeartbeatTime = now
+		}
+		*conditions = append(*conditions, newCondition)
+		return
+	}
+
+	if existingCondition.Status != newCondition.Status {
+		existingCondition.Status = newCondition.Status
+		if !newCondition.LastTransitionTime.IsZero() {
+			existingCondition.LastTransitionTime = newCondition.LastTransitionTime
+		} else {
+			existingCondition.LastTransitionTime = now
+		}
+	}
+
+	existingCondition.Reason = newCondition.Reason
+	existingCondition.Message = newCondition.Message
+	if !newCondition.LastHeartbeatTime.IsZero() {
+		existingCondition.LastHeartbeatTime = newCondition.LastHeartbeatTime
+	} else {
+		existingCondition.LastHeartbeatTime = now
+	}
+}
+
+// FindStatusCondition finds the conditionType in conditions.
+func FindStatusCondition(conditions []Condition, conditionType ConditionType) *Condition {
+	for i := range conditions {
+		if conditions[i].Type == conditionType {
+			return &conditions[i]
+		}
+	}
+
+	return nil
+}

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

@@ -0,0 +1,127 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+
+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
+
+import (
+	"reflect"
+	"testing"
+	"time"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// Based on code from https://github.com/kubernetes/apimachinery/blob/master/pkg/api/meta/conditions.go
+
+func TestSetStatusCondition(t *testing.T) {
+	oneHourBefore := metav1.Time{Time: time.Now().Add(-1 * time.Hour)}
+	oneHourAfter := metav1.Time{Time: time.Now().Add(1 * time.Hour)}
+
+	tests := []struct {
+		name       string
+		conditions []Condition
+		toAdd      Condition
+		expected   []Condition
+	}{
+		{
+			name: "should-add",
+			conditions: []Condition{
+				{Type: "first"},
+				{Type: "third"},
+			},
+			toAdd: Condition{Type: "second", Status: v1.ConditionTrue, LastTransitionTime: oneHourBefore, LastHeartbeatTime: oneHourBefore, Reason: "reason", Message: "message"},
+			expected: []Condition{
+				{Type: "first"},
+				{Type: "third"},
+				{Type: "second", Status: v1.ConditionTrue, LastTransitionTime: oneHourBefore, LastHeartbeatTime: oneHourBefore, Reason: "reason", Message: "message"},
+			},
+		},
+		{
+			name: "use-supplied-transition-time",
+			conditions: []Condition{
+				{Type: "first"},
+				{Type: "second", Status: v1.ConditionFalse},
+				{Type: "third"},
+			},
+			toAdd: Condition{Type: "second", Status: v1.ConditionTrue, LastTransitionTime: oneHourBefore, LastHeartbeatTime: oneHourBefore, Reason: "reason", Message: "message"},
+			expected: []Condition{
+				{Type: "first"},
+				{Type: "second", Status: v1.ConditionTrue, LastTransitionTime: oneHourBefore, LastHeartbeatTime: oneHourBefore, Reason: "reason", Message: "message"},
+				{Type: "third"},
+			},
+		},
+		{
+			name: "update-fields",
+			conditions: []Condition{
+				{Type: "first"},
+				{Type: "second", Status: v1.ConditionTrue, LastTransitionTime: oneHourBefore, LastHeartbeatTime: oneHourBefore},
+				{Type: "third"},
+			},
+			toAdd: Condition{Type: "second", Status: v1.ConditionTrue, LastTransitionTime: oneHourAfter, LastHeartbeatTime: oneHourAfter, Reason: "reason", Message: "message"},
+			expected: []Condition{
+				{Type: "first"},
+				{Type: "second", Status: v1.ConditionTrue, LastTransitionTime: oneHourBefore, LastHeartbeatTime: oneHourAfter, Reason: "reason", Message: "message"},
+				{Type: "third"},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			SetStatusCondition(&test.conditions, test.toAdd)
+			if !reflect.DeepEqual(test.conditions, test.expected) {
+				t.Error(test.conditions)
+			}
+		})
+	}
+}
+
+func TestFindStatusCondition(t *testing.T) {
+	tests := []struct {
+		name          string
+		conditions    []Condition
+		conditionType string
+		expected      *Condition
+	}{
+		{
+			name: "not-present",
+			conditions: []Condition{
+				{Type: "first"},
+			},
+			conditionType: "second",
+			expected:      nil,
+		},
+		{
+			name: "present",
+			conditions: []Condition{
+				{Type: "first"},
+				{Type: "second"},
+			},
+			conditionType: "second",
+			expected:      &Condition{Type: "second"},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			actual := FindStatusCondition(test.conditions, ConditionType(test.conditionType))
+			if !reflect.DeepEqual(actual, test.expected) {
+				t.Error(actual)
+			}
+		})
+	}
+}

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

@@ -383,30 +383,34 @@ type CephHealthMessage struct {
 	Message  string `json:"message"`
 }
 
-// Condition represents
+// Condition represents a status condition on any Rook-Ceph Custom Resource.
 type Condition struct {
 	Type               ConditionType      `json:"type,omitempty"`
 	Status             v1.ConditionStatus `json:"status,omitempty"`
-	Reason             ClusterReasonType  `json:"reason,omitempty"`
+	Reason             ConditionReason    `json:"reason,omitempty"`
 	Message            string             `json:"message,omitempty"`
 	LastHeartbeatTime  metav1.Time        `json:"lastHeartbeatTime,omitempty"`
 	LastTransitionTime metav1.Time        `json:"lastTransitionTime,omitempty"`
 }
 
-// ClusterReasonType is cluster reason
-type ClusterReasonType string
+// ConditionReason is a reason for a condition
+type ConditionReason string
 
 const (
 	// ClusterCreatedReason is cluster created reason
-	ClusterCreatedReason ClusterReasonType = "ClusterCreated"
+	ClusterCreatedReason ConditionReason = "ClusterCreated"
 	// ClusterConnectedReason is cluster connected reason
-	ClusterConnectedReason ClusterReasonType = "ClusterConnected"
+	ClusterConnectedReason ConditionReason = "ClusterConnected"
 	// ClusterProgressingReason is cluster progressing reason
-	ClusterProgressingReason ClusterReasonType = "ClusterProgressing"
+	ClusterProgressingReason ConditionReason = "ClusterProgressing"
 	// ClusterDeletingReason is cluster deleting reason
-	ClusterDeletingReason ClusterReasonType = "ClusterDeleting"
+	ClusterDeletingReason ConditionReason = "ClusterDeleting"
 	// ClusterConnectingReason is cluster connecting reason
-	ClusterConnectingReason ClusterReasonType = "ClusterConnecting"
+	ClusterConnectingReason ConditionReason = "ClusterConnecting"
+
+	// ObjectHasDependentsReason represents when an resource object has dependents. For example,
+	// because deletion is blocked.
+	ObjectHasDependentsReason ConditionReason = "ObjectHasDependents"
 )
 
 // ConditionType represent a resource's status
@@ -425,6 +429,9 @@ const (
 	ConditionFailure ConditionType = "Failure"
 	// ConditionDeleting represents Deleting state of an object
 	ConditionDeleting ConditionType = "Deleting"
+
+	// ConditionDeletionIsBlocked represents when deletion of the object is blocked.
+	ConditionDeletionIsBlocked ConditionType = "DeletionIsBlocked"
 )
 
 // ClusterState represents the state of a Ceph Cluster

pkg/clusterd/context.go

@@ -24,6 +24,7 @@ import (
 	"github.com/rook/rook/pkg/util/sys"
 	"github.com/tevino/abool"
 	apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
+	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -38,6 +39,9 @@ type Context struct {
 	// Clientset is a connection to the core kubernetes API
 	Clientset kubernetes.Interface
 
+	// DynamicClientset is a dynamic connection to the Kubernetes API
+	DynamicClientset dynamic.Interface
+
 	// Represents the Client provided by the controller-runtime package to interact with Kubernetes objects
 	Client client.Client
 

pkg/operator/ceph/cluster/cephstatus.go

@@ -146,7 +146,7 @@ func (c *cephStatusChecker) checkStatus() {
 }
 
 // updateStatus updates an object with a given status
-func (c *cephStatusChecker) updateCephStatus(status *cephclient.CephStatus, condition cephv1.ConditionType, reason cephv1.ClusterReasonType, message string, conditionStatus v1.ConditionStatus) {
+func (c *cephStatusChecker) updateCephStatus(status *cephclient.CephStatus, condition cephv1.ConditionType, reason cephv1.ConditionReason, message string, conditionStatus v1.ConditionStatus) {
 	clusterName := c.clusterInfo.NamespacedName()
 	cephCluster, err := c.context.RookClientset.CephV1().CephClusters(clusterName.Namespace).Get(context.TODO(), clusterName.Name, metav1.GetOptions{})
 	if err != nil {

pkg/operator/ceph/cluster/controller.go

@@ -19,6 +19,7 @@ package cluster
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"sync"
 	"time"
@@ -36,6 +37,7 @@ import (
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -116,7 +118,10 @@ func newReconciler(mgr manager.Manager, ctx *clusterd.Context, clusterController
 		panic(err)
 	}
 
-	clusterController.recorder = k8sutil.NewEventReporter(mgr.GetEventRecorderFor("ClusterController"))
+	// add "rook-" prefix to the controller name to make sure it is clear to all reading the events
+	// that they are coming from Rook. The controller name already has context that it is for Ceph
+	// and from the cluster controller.
+	clusterController.recorder = k8sutil.NewEventReporter(mgr.GetEventRecorderFor("rook-" + controllerName))
 
 	return &ReconcileCephCluster{
 		client:            mgr.GetClient(),
@@ -223,6 +228,13 @@ func (r *ReconcileCephCluster) Reconcile(context context.Context, request reconc
 	if err != nil {
 		logger.Errorf("failed to reconcile. %v", err)
 		r.clusterController.recorder.ReportIfNotPresent(cephCluster, corev1.EventTypeWarning, "ReconcileFailed", err.Error())
+		if !reconcileResponse.IsZero() {
+			// The framework will requeue immediately if there is an error. If we get an error with
+			// a non-empty reconcile response, just return the response with the error now logged as
+			// an event so that the framework can pause before the next reconcile per the response's
+			// intent.
+			return reconcileResponse, nil
+		}
 	} else {
 		logger.Debug("reconcile succeeded.")
 		r.clusterController.recorder.ReportIfNotPresent(cephCluster, corev1.EventTypeNormal, "ReconcileSucceeded", "cluster has been configured successfully")
@@ -261,49 +273,7 @@ func (r *ReconcileCephCluster) reconcile(request reconcile.Request) (reconcile.R
 
 	// DELETE: the CR was deleted
 	if !cephCluster.GetDeletionTimestamp().IsZero() {
-		doCleanup := true
-		logger.Infof("deleting ceph cluster %q", cephCluster.Name)
-
-		// Start cluster clean up only if cleanupPolicy is applied to the ceph cluster
-		stopCleanupCh := make(chan struct{})
-		if cephCluster.Spec.CleanupPolicy.HasDataDirCleanPolicy() && !cephCluster.Spec.External.Enable {
-			// Set the deleting status
-			updateStatus(r.client, request.NamespacedName, cephv1.ConditionDeleting)
-
-			monSecret, clusterFSID, err := r.clusterController.getCleanUpDetails(cephCluster.Namespace)
-			if err != nil {
-				logger.Warningf("failed to get mon secret. Skip cluster cleanup and remove finalizer. %v", err)
-				doCleanup = false
-			}
-
-			if doCleanup {
-				cephHosts, err := r.clusterController.getCephHosts(cephCluster.Namespace)
-				if err != nil {
-					return reconcile.Result{}, cephCluster, errors.Wrapf(err, "failed to find valid ceph hosts in the cluster %q", cephCluster.Namespace)
-				}
-				go r.clusterController.startClusterCleanUp(stopCleanupCh, cephCluster, cephHosts, monSecret, clusterFSID)
-			}
-		}
-
-		if doCleanup {
-			// Run delete sequence
-			response, ok := r.clusterController.requestClusterDelete(cephCluster)
-			if !ok {
-				// If the cluster cannot be deleted, requeue the request for deletion to see if the conditions
-				// will eventually be satisfied such as the volumes being removed
-				close(stopCleanupCh)
-				return response, cephCluster, nil
-			}
-		}
-
-		// Remove finalizer
-		err = removeFinalizer(r.client, request.NamespacedName)
-		if err != nil {
-			return reconcile.Result{}, cephCluster, errors.Wrap(err, "failed to remove finalizer")
-		}
-
-		// Return and do not requeue. Successful deletion.
-		return reconcile.Result{}, cephCluster, nil
+		return r.reconcileDelete(cephCluster)
 	}
 
 	// Do reconcile here!
@@ -316,6 +286,88 @@ func (r *ReconcileCephCluster) reconcile(request reconcile.Request) (reconcile.R
 	return reconcile.Result{}, cephCluster, nil
 }
 
+func (r *ReconcileCephCluster) reconcileDelete(cephCluster *cephv1.CephCluster) (reconcile.Result, *cephv1.CephCluster, error) {
+	nsName := r.clusterController.namespacedName
+	var err error
+
+	// Set the deleting status
+	opcontroller.UpdateClusterCondition(r.context, cephCluster, nsName,
+		cephv1.ConditionDeleting, corev1.ConditionTrue, cephv1.ClusterDeletingReason, "Deleting the CephCluster",
+		true /* keep all other conditions to be safe */)
+
+	dependents, err := CephClusterDependents(r.context, cephCluster.Namespace)
+	if err != nil {
+		return reconcile.Result{}, cephCluster, err
+	}
+	if !dependents.Empty() {
+		// if there are dependents, we should NOT delete the resource
+		dependentMsg := dependents.StringWithHeader("CephCluster %q will not be deleted until all dependents are removed", nsName.String())
+		logger.Errorf(dependentMsg)
+		// Set the DeletionIsBlocked=True status condition
+		// Adding the DeletionIsBlocked condition doesn't translate directly to a status/phase,
+		// so do not use the "UpdateClusterCondition" function which will modify that.
+		cond := buildDeletionBlockedDueToDependentsCondition(true, dependentMsg)
+		cephv1.SetStatusCondition(&cephCluster.Status.Conditions, cond)
+		if err := opcontroller.UpdateStatus(r.client, cephCluster); err != nil {
+			return reconcile.Result{}, cephCluster, errors.Wrapf(err, "on condition %s. failed to update CephCluster %q status condition %s=%s", dependentMsg, r.clusterController.namespacedName.String(), cond.Type, cond.Status)
+		}
+		return opcontroller.WaitForRequeueIfFinalizerBlocked, cephCluster, errors.New(dependentMsg)
+	}
+
+	deletingMsg := fmt.Sprintf("deleting CephCluster %q", cephCluster.Name)
+	// Report an event that the cluster is deleting
+	r.clusterController.recorder.ReportIfNotPresent(cephCluster, corev1.EventTypeNormal, string(cephv1.ClusterDeletingReason), deletingMsg)
+	// Set the DeletionIsBlocked=False status condition
+	cond := buildDeletionBlockedDueToDependentsCondition(false, "cluster can be deleted safely")
+	cephv1.SetStatusCondition(&cephCluster.Status.Conditions, cond)
+	if err := opcontroller.UpdateStatus(r.client, cephCluster); err != nil {
+		logger.Warningf("failed to set CephCluster %q status condition %s=%s; continuing deletion without setting the condition", r.clusterController.namespacedName.String(), cond.Type, cond.Status)
+	}
+	// Log deletion to the operator log
+	logger.Info(deletingMsg)
+
+	doCleanup := true
+
+	// Start cluster clean up only if cleanupPolicy is applied to the ceph cluster
+	stopCleanupCh := make(chan struct{})
+	if cephCluster.Spec.CleanupPolicy.HasDataDirCleanPolicy() && !cephCluster.Spec.External.Enable {
+		monSecret, clusterFSID, err := r.clusterController.getCleanUpDetails(cephCluster.Namespace)
+		if err != nil {
+			logger.Warningf("failed to get mon secret. Skip cluster cleanup and remove finalizer. %v", err)
+			doCleanup = false
+		}
+
+		if doCleanup {
+			cephHosts, err := r.clusterController.getCephHosts(cephCluster.Namespace)
+			if err != nil {
+				return reconcile.Result{}, cephCluster, errors.Wrapf(err, "failed to find valid ceph hosts in the cluster %q", cephCluster.Namespace)
+			}
+			go r.clusterController.startClusterCleanUp(stopCleanupCh, cephCluster, cephHosts, monSecret, clusterFSID)
+		}
+	}
+
+	if doCleanup {
+		// Run delete sequence
+		response, err := r.clusterController.requestClusterDelete(cephCluster)
+		if err != nil {
+			// If the cluster cannot be deleted, requeue the request for deletion to see if the conditions
+			// will eventually be satisfied such as the volumes being removed
+			// TODO: is it a memory leak to not close the channel in other cases?
+			close(stopCleanupCh)
+			return response, cephCluster, errors.Wrapf(err, "failed to clean up CephCluster %q", nsName.String())
+		}
+	}
+
+	// Remove finalizer
+	err = removeFinalizer(r.client, nsName)
+	if err != nil {
+		return reconcile.Result{}, cephCluster, errors.Wrap(err, "failed to remove finalizer")
+	}
+
+	// Return and do not requeue. Successful deletion.
+	return reconcile.Result{}, cephCluster, nil
+}
+
 // NewClusterController create controller for watching cluster custom resources created
 func NewClusterController(context *clusterd.Context, rookImage string, volumeAttachment attachment.Attachment, operatorConfigCallbacks []func() error, addClusterCallbacks []func() error) *ClusterController {
 	return &ClusterController{
@@ -359,16 +411,16 @@ func (c *ClusterController) onAdd(clusterObj *cephv1.CephCluster, ownerInfo *k8s
 	return c.initializeCluster(cluster, clusterObj)
 }
 
-func (c *ClusterController) requestClusterDelete(cluster *cephv1.CephCluster) (reconcile.Result, bool) {
-	opcontroller.UpdateCondition(c.context, c.namespacedName, cephv1.ConditionDeleting, corev1.ConditionTrue, cephv1.ClusterDeletingReason, "Cluster is deleting")
+func (c *ClusterController) requestClusterDelete(cluster *cephv1.CephCluster) (reconcile.Result, error) {
+	nsName := fmt.Sprintf("%s/%s", cluster.Namespace, cluster.Name)
 
 	if existing, ok := c.clusterMap[cluster.Namespace]; ok && existing.namespacedName.Name != cluster.Name {
-		logger.Errorf("skipping deletion of cluster cr %q in namespace %q. cluster CR %q already exists in this namespace. only one cluster cr per namespace is supported.",
-			cluster.Name, cluster.Namespace, existing.namespacedName.Name)
-		return reconcile.Result{}, true
+		logger.Errorf("skipping deletion of CephCluster %q. CephCluster CR %q already exists in this namespace. only one cluster cr per namespace is supported.",
+			nsName, existing.namespacedName.Name)
+		return reconcile.Result{}, nil // do not requeue the delete
 	}
 
-	logger.Infof("delete event for cluster %q in namespace %q", cluster.Name, cluster.Namespace)
+	logger.Infof("cleaning up CephCluster %q", nsName)
 
 	if cluster, ok := c.clusterMap[cluster.Namespace]; ok {
 		// if not already stopped, stop clientcontroller and bucketController
@@ -391,25 +443,18 @@ func (c *ClusterController) requestClusterDelete(cluster *cephv1.CephCluster) (r
 	} else {
 		err := c.checkIfVolumesExist(cluster)
 		if err != nil {
-			opcontroller.UpdateCondition(c.context, c.namespacedName, cephv1.ConditionDeleting, corev1.ConditionFalse, "ClusterDeleting", err.Error())
-			logger.Errorf("failed to check if volumes exist. %v", err)
-			return opcontroller.WaitForRequeueIfFinalizerBlocked, false
+			return opcontroller.WaitForRequeueIfFinalizerBlocked, errors.Wrapf(err, "failed to check if volumes exist for CephCluster in namespace %q", cluster.Namespace)
 		}
 	}
 
-	// Only valid when the cluster is not external
 	if cluster.Spec.External.Enable {
 		purgeExternalCluster(c.context.Clientset, cluster.Namespace)
-		return reconcile.Result{}, true
-	}
-
-	// If the StorageClass retain policy of an encrypted cluster with KMS is Delete we also delete the keys
-	if cluster.Spec.Storage.IsOnPVCEncrypted() && cluster.Spec.Security.KeyManagementService.IsEnabled() {
+	} else if cluster.Spec.Storage.IsOnPVCEncrypted() && cluster.Spec.Security.KeyManagementService.IsEnabled() {
+		// If the StorageClass retain policy of an encrypted cluster with KMS is Delete we also delete the keys
 		// Delete keys from KMS
 		err := c.deleteOSDEncryptionKeyFromKMS(cluster)
 		if err != nil {
-			logger.Errorf("failed to delete osd encryption keys from kms. %v", err)
-			return reconcile.Result{}, true
+			logger.Errorf("failed to delete osd encryption keys for CephCluster %q from kms; deletion will continue. %v", nsName, err)
 		}
 	}
 
@@ -417,7 +462,7 @@ func (c *ClusterController) requestClusterDelete(cluster *cephv1.CephCluster) (r
 		delete(c.clusterMap, cluster.Namespace)
 	}
 
-	return reconcile.Result{}, true
+	return reconcile.Result{}, nil
 }
 
 func (c *ClusterController) checkIfVolumesExist(cluster *cephv1.CephCluster) error {
@@ -505,27 +550,6 @@ func (c *ClusterController) checkPVPresentInCluster(drivers []string, clusterID
 	return false, nil
 }
 
-// updateStatus updates an object with a given status
-func updateStatus(client client.Client, name types.NamespacedName, status cephv1.ConditionType) {
-	cephCluster := &cephv1.CephCluster{}
-	err := client.Get(context.TODO(), name, cephCluster)
-	if err != nil {
-		if kerrors.IsNotFound(err) {
-			logger.Debug("CephCluster resource not found. Ignoring since object must be deleted.")
-			return
-		}
-		logger.Warningf("failed to retrieve ceph cluster %q to update status to %q. %v", name, status, err)
-		return
-	}
-
-	cephCluster.Status.Phase = status
-	if err := opcontroller.UpdateStatus(client, cephCluster); err != nil {
-		logger.Errorf("failed to set ceph cluster %q status to %q. %v", cephCluster.Name, status, err)
-		return
-	}
-	logger.Debugf("ceph cluster %q status updated to %q", name, status)
-}
-
 // removeFinalizer removes a finalizer
 func removeFinalizer(client client.Client, name types.NamespacedName) error {
 	cephCluster := &cephv1.CephCluster{}
@@ -581,3 +605,16 @@ func (c *ClusterController) deleteOSDEncryptionKeyFromKMS(currentCluster *cephv1
 
 	return nil
 }
+
+func buildDeletionBlockedDueToDependentsCondition(blocked bool, message string) cephv1.Condition {
+	status := v1.ConditionFalse
+	if blocked {
+		status = v1.ConditionTrue
+	}
+	return cephv1.Condition{
+		Type:    cephv1.ConditionDeletionIsBlocked,
+		Status:  status,
+		Reason:  cephv1.ObjectHasDependentsReason,
+		Message: message,
+	}
+}

pkg/operator/ceph/cluster/controller_test.go

@@ -17,8 +17,10 @@ limitations under the License.
 package cluster
 
 import (
+	"context"
 	"os"
 	"testing"
+	"time"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	rookalpha "github.com/rook/rook/pkg/apis/rook.io/v1alpha2"
@@ -27,116 +29,242 @@ import (
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	testop "github.com/rook/rook/pkg/operator/test"
 	"github.com/stretchr/testify/assert"
+	"github.com/tevino/abool"
+	corev1 "k8s.io/api/core/v1"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	dynamicfake "k8s.io/client-go/dynamic/fake"
+	k8sfake "k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/tools/record"
+	clientfake "sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
-func TestClusterDeleteFlexEnabled(t *testing.T) {
-	nodeName := "node841"
-	clusterName := "cluster684"
-	pvName := "pvc-540"
-	rookSystemNamespace := "rook-system-6413"
-
-	os.Setenv("ROOK_ENABLE_FLEX_DRIVER", "true")
-	os.Setenv(k8sutil.PodNamespaceEnvVar, rookSystemNamespace)
-	defer os.Unsetenv("ROOK_ENABLE_FLEX_DRIVER")
-	defer os.Unsetenv(k8sutil.PodNamespaceEnvVar)
-
-	context := &clusterd.Context{
-		Clientset: testop.New(t, 3),
+func TestReconcile_DeleteCephCluster(t *testing.T) {
+	ctx := context.TODO()
+	cephNs := "rook-ceph"
+	clusterName := "my-cluster"
+	nsName := types.NamespacedName{
+		Name:      clusterName,
+		Namespace: cephNs,
 	}
-	listCount := 0
-	volumeAttachmentController := &attachment.MockAttachment{
-		MockList: func(namespace string) (*rookalpha.VolumeList, error) {
-			listCount++
-			if listCount == 1 {
-				// first listing returns an existing volume attachment, so the controller should wait
-				return &rookalpha.VolumeList{
-					Items: []rookalpha.Volume{
-						{
-							ObjectMeta: metav1.ObjectMeta{
-								Name:      pvName,
-								Namespace: rookSystemNamespace,
-							},
-							Attachments: []rookalpha.Attachment{
-								{
-									Node:        nodeName,
-									ClusterName: clusterName,
+
+	fakeCluster := &cephv1.CephCluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:              clusterName,
+			Namespace:         cephNs,
+			DeletionTimestamp: &metav1.Time{Time: time.Now()},
+		},
+	}
+
+	fakePool := &cephv1.CephBlockPool{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "CephBlockPool",
+			APIVersion: "ceph.rook.io/v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "my-block-pool",
+			Namespace: cephNs,
+		},
+	}
+
+	// create a Rook-Ceph scheme to use for our tests
+	scheme := runtime.NewScheme()
+	assert.NoError(t, cephv1.AddToScheme(scheme))
+
+	t.Run("deletion blocked while dependencies exist", func(t *testing.T) {
+		// set up clusterd.Context
+		clusterdCtx := &clusterd.Context{
+			Clientset: k8sfake.NewSimpleClientset(),
+			// reconcile looks for fake dependencies in the dynamic clientset
+			DynamicClientset:           dynamicfake.NewSimpleDynamicClient(scheme, fakePool),
+			RequestCancelOrchestration: abool.New(),
+		}
+
+		// set up ClusterController
+		volumeAttachmentController := &attachment.MockAttachment{
+			MockList: func(namespace string) (*rookalpha.VolumeList, error) {
+				t.Log("test vol attach list")
+				return &rookalpha.VolumeList{Items: []rookalpha.Volume{}}, nil
+			},
+		}
+		operatorConfigCallbacks := []func() error{
+			func() error {
+				t.Log("test op config callback")
+				return nil
+			},
+		}
+		addCallbacks := []func() error{
+			func() error {
+				t.Log("test success callback")
+				return nil
+			},
+		}
+
+		// create the cluster controller and tell it that the cluster has been deleted
+		controller := NewClusterController(clusterdCtx, "", volumeAttachmentController, operatorConfigCallbacks, addCallbacks)
+		fakeRecorder := record.NewFakeRecorder(5)
+		controller.recorder = k8sutil.NewEventReporter(fakeRecorder)
+
+		// Create a fake client to mock API calls
+		// Make sure it has the fake CephCluster that is to be deleted in it
+		client := clientfake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(fakeCluster).Build()
+
+		// Create a ReconcileCephClient object with the scheme and fake client.
+		reconcileCephCluster := &ReconcileCephCluster{
+			client:            client,
+			scheme:            scheme,
+			context:           clusterdCtx,
+			clusterController: controller,
+		}
+
+		req := reconcile.Request{NamespacedName: nsName}
+
+		resp, err := reconcileCephCluster.Reconcile(ctx, req)
+		assert.NoError(t, err)
+		assert.NotZero(t, resp.RequeueAfter)
+		event := <-fakeRecorder.Events
+		assert.Contains(t, event, "CephBlockPools")
+		assert.Contains(t, event, "my-block-pool")
+
+		blockedCluster := &cephv1.CephCluster{}
+		err = client.Get(ctx, nsName, blockedCluster)
+		assert.NoError(t, err)
+		status := blockedCluster.Status
+		assert.Equal(t, cephv1.ConditionDeleting, status.Phase)
+		assert.Equal(t, cephv1.ClusterState(cephv1.ConditionDeleting), status.State)
+		assert.Equal(t, corev1.ConditionTrue, cephv1.FindStatusCondition(status.Conditions, cephv1.ConditionDeleting).Status)
+		assert.Equal(t, corev1.ConditionTrue, cephv1.FindStatusCondition(status.Conditions, cephv1.ConditionDeletionIsBlocked).Status)
+
+		// delete blocking dependency
+		gvr := cephv1.SchemeGroupVersion.WithResource("cephblockpools")
+		err = clusterdCtx.DynamicClientset.Resource(gvr).Namespace(cephNs).Delete(ctx, "my-block-pool", metav1.DeleteOptions{})
+		assert.NoError(t, err)
+
+		resp, err = reconcileCephCluster.Reconcile(ctx, req)
+		assert.NoError(t, err)
+		assert.True(t, resp.IsZero())
+		event = <-fakeRecorder.Events
+		assert.Contains(t, event, "ClusterDeleting")
+
+		unblockedCluster := &cephv1.CephCluster{}
+		err = client.Get(ctx, nsName, unblockedCluster)
+		assert.Error(t, err)
+		assert.True(t, kerrors.IsNotFound(err))
+	})
+}
+
+func Test_checkIfVolumesExist(t *testing.T) {
+	t.Run("flexvolume enabled", func(t *testing.T) {
+		nodeName := "node841"
+		clusterName := "cluster684"
+		pvName := "pvc-540"
+		rookSystemNamespace := "rook-system-6413"
+
+		os.Setenv("ROOK_ENABLE_FLEX_DRIVER", "true")
+		os.Setenv(k8sutil.PodNamespaceEnvVar, rookSystemNamespace)
+		defer os.Unsetenv("ROOK_ENABLE_FLEX_DRIVER")
+		defer os.Unsetenv(k8sutil.PodNamespaceEnvVar)
+
+		context := &clusterd.Context{
+			Clientset: testop.New(t, 3),
+		}
+		listCount := 0
+		volumeAttachmentController := &attachment.MockAttachment{
+			MockList: func(namespace string) (*rookalpha.VolumeList, error) {
+				listCount++
+				if listCount == 1 {
+					// first listing returns an existing volume attachment, so the controller should wait
+					return &rookalpha.VolumeList{
+						Items: []rookalpha.Volume{
+							{
+								ObjectMeta: metav1.ObjectMeta{
+									Name:      pvName,
+									Namespace: rookSystemNamespace,
+								},
+								Attachments: []rookalpha.Attachment{
+									{
+										Node:        nodeName,
+										ClusterName: clusterName,
+									},
 								},
 							},
 						},
-					},
-				}, nil
-			}
+					}, nil
+				}
 
-			// subsequent listings should return no volume attachments, meaning that they have all
-			// been cleaned up and the controller can move on.
-			return &rookalpha.VolumeList{Items: []rookalpha.Volume{}}, nil
+				// subsequent listings should return no volume attachments, meaning that they have all
+				// been cleaned up and the controller can move on.
+				return &rookalpha.VolumeList{Items: []rookalpha.Volume{}}, nil
 
-		},
-	}
-	operatorConfigCallbacks := []func() error{
-		func() error {
-			logger.Infof("test success callback")
-			return nil
-		},
-	}
-	addCallbacks := []func() error{
-		func() error {
-			logger.Infof("test success callback")
-			return nil
-		},
-	}
-	// create the cluster controller and tell it that the cluster has been deleted
-	controller := NewClusterController(context, "", volumeAttachmentController, operatorConfigCallbacks, addCallbacks)
-	clusterToDelete := &cephv1.CephCluster{ObjectMeta: metav1.ObjectMeta{Namespace: clusterName}}
+			},
+		}
+		operatorConfigCallbacks := []func() error{
+			func() error {
+				logger.Infof("test success callback")
+				return nil
+			},
+		}
+		addCallbacks := []func() error{
+			func() error {
+				logger.Infof("test success callback")
+				return nil
+			},
+		}
+		// create the cluster controller and tell it that the cluster has been deleted
+		controller := NewClusterController(context, "", volumeAttachmentController, operatorConfigCallbacks, addCallbacks)
+		clusterToDelete := &cephv1.CephCluster{ObjectMeta: metav1.ObjectMeta{Namespace: clusterName}}
 
-	// The test returns a volume on the first call
-	assert.Error(t, controller.checkIfVolumesExist(clusterToDelete))
+		// The test returns a volume on the first call
+		assert.Error(t, controller.checkIfVolumesExist(clusterToDelete))
 
-	// The test does not return volumes on the second call
-	assert.NoError(t, controller.checkIfVolumesExist(clusterToDelete))
-}
+		// The test does not return volumes on the second call
+		assert.NoError(t, controller.checkIfVolumesExist(clusterToDelete))
+	})
 
-func TestClusterDeleteFlexDisabled(t *testing.T) {
-	clusterName := "cluster684"
-	rookSystemNamespace := "rook-system-6413"
+	t.Run("flexvolume disabled (CSI)", func(t *testing.T) {
+		clusterName := "cluster684"
+		rookSystemNamespace := "rook-system-6413"
 
-	os.Setenv("ROOK_ENABLE_FLEX_DRIVER", "false")
-	os.Setenv(k8sutil.PodNamespaceEnvVar, rookSystemNamespace)
-	defer os.Unsetenv("ROOK_ENABLE_FLEX_DRIVER")
-	defer os.Unsetenv(k8sutil.PodNamespaceEnvVar)
+		os.Setenv("ROOK_ENABLE_FLEX_DRIVER", "false")
+		os.Setenv(k8sutil.PodNamespaceEnvVar, rookSystemNamespace)
+		defer os.Unsetenv("ROOK_ENABLE_FLEX_DRIVER")
+		defer os.Unsetenv(k8sutil.PodNamespaceEnvVar)
 
-	context := &clusterd.Context{
-		Clientset: testop.New(t, 3),
-	}
-	listCount := 0
-	volumeAttachmentController := &attachment.MockAttachment{
-		MockList: func(namespace string) (*rookalpha.VolumeList, error) {
-			listCount++
-			return &rookalpha.VolumeList{Items: []rookalpha.Volume{}}, nil
+		context := &clusterd.Context{
+			Clientset: testop.New(t, 3),
+		}
+		listCount := 0
+		volumeAttachmentController := &attachment.MockAttachment{
+			MockList: func(namespace string) (*rookalpha.VolumeList, error) {
+				listCount++
+				return &rookalpha.VolumeList{Items: []rookalpha.Volume{}}, nil
 
-		},
-	}
-	operatorConfigCallbacks := []func() error{
-		func() error {
-			logger.Infof("test success callback")
-			return nil
-		},
-	}
-	addCallbacks := []func() error{
-		func() error {
-			logger.Infof("test success callback")
-			os.Setenv("ROOK_ENABLE_FLEX_DRIVER", "true")
-			os.Setenv(k8sutil.PodNamespaceEnvVar, rookSystemNamespace)
-			defer os.Unsetenv("ROOK_ENABLE_FLEX_DRIVER")
-			return nil
-		},
-	}
-	// create the cluster controller and tell it that the cluster has been deleted
-	controller := NewClusterController(context, "", volumeAttachmentController, operatorConfigCallbacks, addCallbacks)
-	clusterToDelete := &cephv1.CephCluster{ObjectMeta: metav1.ObjectMeta{Namespace: clusterName}}
-	assert.NoError(t, controller.checkIfVolumesExist(clusterToDelete))
+			},
+		}
+		operatorConfigCallbacks := []func() error{
+			func() error {
+				logger.Infof("test success callback")
+				return nil
+			},
+		}
+		addCallbacks := []func() error{
+			func() error {
+				logger.Infof("test success callback")
+				os.Setenv("ROOK_ENABLE_FLEX_DRIVER", "true")
+				os.Setenv(k8sutil.PodNamespaceEnvVar, rookSystemNamespace)
+				defer os.Unsetenv("ROOK_ENABLE_FLEX_DRIVER")
+				return nil
+			},
+		}
+		// create the cluster controller and tell it that the cluster has been deleted
+		controller := NewClusterController(context, "", volumeAttachmentController, operatorConfigCallbacks, addCallbacks)
+		clusterToDelete := &cephv1.CephCluster{ObjectMeta: metav1.ObjectMeta{Namespace: clusterName}}
+		assert.NoError(t, controller.checkIfVolumesExist(clusterToDelete))
 
-	// Ensure that the listing of volume attachments was never called.
-	assert.Equal(t, 0, listCount)
+		// Ensure that the listing of volume attachments was never called.
+		assert.Equal(t, 0, listCount)
+	})
 }

pkg/operator/ceph/cluster/dependents.go

@@ -0,0 +1,79 @@
+/*
+Copyright 2021 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 cluster
+
+import (
+	"context"
+	"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"
+	"github.com/rook/rook/pkg/util/dependents"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+var (
+	// must use plural kinds
+	cephClusterDependentPluralKinds []string = []string{
+		"CephBlockPools",
+		"CephRBDMirrors",
+		"CephFilesystems",
+		"CephFilesystemMirrors",
+		"CephObjectStores",
+		"CephObjectStoreUsers",
+		"CephObjectZones",
+		"CephObjectZoneGroups",
+		"CephObjectRealms",
+		"CephNFSes",
+		"CephClients",
+	}
+)
+
+// CephClusterDependents returns a DependentList of dependents of a CephCluster in the namespace.
+func CephClusterDependents(c *clusterd.Context, namespace string) (*dependents.DependentList, error) {
+	ctx := context.TODO()
+
+	dependents := dependents.NewDependentList()
+	errs := []error{}
+
+	for _, pluralKind := range cephClusterDependentPluralKinds {
+		resource := pluralKindToResource(pluralKind)
+		gvr := cephv1.SchemeGroupVersion.WithResource(resource)
+		list, err := c.DynamicClientset.Resource(gvr).Namespace(namespace).List(ctx, metav1.ListOptions{})
+		if err != nil {
+			errs = append(errs, errors.Wrapf(err, "failed to list %s", pluralKind))
+			continue
+		}
+		if len(list.Items) > 0 {
+			for _, obj := range list.Items {
+				dependents.Add(pluralKind, obj.GetName())
+			}
+		}
+	}
+	// returns a nil error if there are no errors in the list
+	outErr := util.AggregateErrors(errs, "failed to list some dependents for CephCluster in namespace %q", namespace)
+
+	return dependents, outErr
+}
+
+func pluralKindToResource(pluralKind string) string {
+	// The dynamic client wants resources which are lower-case versions of the plural Kinds for
+	// Kubernetes CRDs in almost all cases.
+	return strings.ToLower(pluralKind)
+}

pkg/operator/ceph/cluster/dependents_test.go

@@ -0,0 +1,258 @@
+/*
+Copyright 2021 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 cluster
+
+import (
+	"testing"
+
+	"github.com/pkg/errors"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/clusterd"
+	"github.com/stretchr/testify/assert"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	dynamicfake "k8s.io/client-go/dynamic/fake"
+	k8stesting "k8s.io/client-go/testing"
+)
+
+func TestCephClusterDependents(t *testing.T) {
+	scheme := runtime.NewScheme()
+	assert.NoError(t, cephv1.AddToScheme(scheme))
+
+	ns := "test-ceph-cluster-dependents"
+
+	var c *clusterd.Context
+
+	newClusterdCtx := func(objects ...runtime.Object) *clusterd.Context {
+		dynInt := dynamicfake.NewSimpleDynamicClient(scheme, objects...)
+		return &clusterd.Context{
+			DynamicClientset: dynInt,
+		}
+	}
+
+	// Create objectmeta with the given name in our test namespace
+	meta := func(name string) v1.ObjectMeta {
+		return v1.ObjectMeta{
+			Name:      name,
+			Namespace: ns,
+		}
+	}
+
+	t.Run("CephBlockPools", func(t *testing.T) {
+		c = newClusterdCtx(
+			&cephv1.CephBlockPool{ObjectMeta: meta("block-pool-1")},
+			&cephv1.CephBlockPool{ObjectMeta: meta("block-pool-2")},
+		)
+		deps, err := CephClusterDependents(c, ns)
+		assert.NoError(t, err)
+		assert.False(t, deps.Empty())
+		assert.ElementsMatch(t, []string{"CephBlockPools"}, deps.PluralKinds())
+		assert.ElementsMatch(t, []string{"block-pool-1", "block-pool-2"}, deps.OfPluralKind("CephBlockPools"))
+	})
+
+	t.Run("CephRBDMirrors", func(t *testing.T) {
+		c = newClusterdCtx(
+			&cephv1.CephRBDMirror{ObjectMeta: meta("rbdmirror")},
+		)
+		deps, err := CephClusterDependents(c, ns)
+		assert.NoError(t, err)
+		assert.False(t, deps.Empty())
+		assert.ElementsMatch(t, []string{"CephRBDMirrors"}, deps.PluralKinds())
+		assert.ElementsMatch(t, []string{"rbdmirror"}, deps.OfPluralKind("CephRBDMirrors"))
+	})
+
+	t.Run("CephFilesystems", func(t *testing.T) {
+		c = newClusterdCtx(
+			&cephv1.CephFilesystem{ObjectMeta: meta("filesystem-1")},
+			&cephv1.CephFilesystem{ObjectMeta: meta("filesystem-2")},
+			&cephv1.CephFilesystem{ObjectMeta: meta("filesystem-3")},
+		)
+		deps, err := CephClusterDependents(c, ns)
+		assert.NoError(t, err)
+		assert.False(t, deps.Empty())
+		assert.ElementsMatch(t, []string{"CephFilesystems"}, deps.PluralKinds())
+		assert.ElementsMatch(t, []string{"filesystem-1", "filesystem-2", "filesystem-3"}, deps.OfPluralKind("CephFilesystems"))
+	})
+
+	t.Run("CephFilesystemMirrors", func(t *testing.T) {
+		c = newClusterdCtx(
+			&cephv1.CephFilesystemMirror{ObjectMeta: meta("fsmirror")},
+		)
+		deps, err := CephClusterDependents(c, ns)
+		assert.NoError(t, err)
+		assert.False(t, deps.Empty())
+		assert.ElementsMatch(t, []string{"CephFilesystemMirrors"}, deps.PluralKinds())
+		assert.ElementsMatch(t, []string{"fsmirror"}, deps.OfPluralKind("CephFilesystemMirrors"))
+	})
+
+	t.Run("CephObjectStores", func(t *testing.T) {
+		c = newClusterdCtx(
+			&cephv1.CephObjectStore{ObjectMeta: meta("objectstore-1")},
+			&cephv1.CephObjectStore{ObjectMeta: meta("objectstore-2")},
+		)
+		deps, err := CephClusterDependents(c, ns)
+		assert.NoError(t, err)
+		assert.False(t, deps.Empty())
+		assert.ElementsMatch(t, []string{"CephObjectStores"}, deps.PluralKinds())
+		assert.ElementsMatch(t, []string{"objectstore-1", "objectstore-2"}, deps.OfPluralKind("CephObjectStores"))
+	})
+
+	t.Run("CephObjectStoreUsers", func(t *testing.T) {
+		c = newClusterdCtx(
+			&cephv1.CephObjectStoreUser{ObjectMeta: meta("u1")},
+			&cephv1.CephObjectStoreUser{ObjectMeta: meta("u2")},
+			&cephv1.CephObjectStoreUser{ObjectMeta: meta("u3")},
+			&cephv1.CephObjectStoreUser{ObjectMeta: meta("u4")},
+			&cephv1.CephObjectStoreUser{ObjectMeta: meta("u5")},
+		)
+		deps, err := CephClusterDependents(c, ns)
+		assert.NoError(t, err)
+		assert.False(t, deps.Empty())
+		assert.ElementsMatch(t, []string{"CephObjectStoreUsers"}, deps.PluralKinds())
+		assert.ElementsMatch(t, []string{"u1", "u2", "u3", "u4", "u5"}, deps.OfPluralKind("CephObjectStoreUsers"))
+	})
+
+	t.Run("CephObjectZones", func(t *testing.T) {
+		c = newClusterdCtx(
+			&cephv1.CephObjectZone{ObjectMeta: meta("zone-1")},
+			&cephv1.CephObjectZone{ObjectMeta: meta("zone-2")},
+		)
+		deps, err := CephClusterDependents(c, ns)
+		assert.NoError(t, err)
+		assert.False(t, deps.Empty())
+		assert.ElementsMatch(t, []string{"CephObjectZones"}, deps.PluralKinds())
+		assert.ElementsMatch(t, []string{"zone-1", "zone-2"}, deps.OfPluralKind("CephObjectZones"))
+	})
+
+	t.Run("CephObjectZoneGroups", func(t *testing.T) {
+		c = newClusterdCtx(
+			&cephv1.CephObjectZoneGroup{ObjectMeta: meta("group-1")},
+			&cephv1.CephObjectZoneGroup{ObjectMeta: meta("group-2")},
+		)
+		deps, err := CephClusterDependents(c, ns)
+		assert.NoError(t, err)
+		assert.False(t, deps.Empty())
+		assert.ElementsMatch(t, []string{"CephObjectZoneGroups"}, deps.PluralKinds())
+		assert.ElementsMatch(t, []string{"group-1", "group-2"}, deps.OfPluralKind("CephObjectZoneGroups"))
+	})
+
+	t.Run("CephObjectRealms", func(t *testing.T) {
+		c = newClusterdCtx(
+			&cephv1.CephObjectRealm{ObjectMeta: meta("realm-1")},
+			&cephv1.CephObjectRealm{ObjectMeta: meta("realm-2")},
+		)
+		deps, err := CephClusterDependents(c, ns)
+		assert.NoError(t, err)
+		assert.False(t, deps.Empty())
+		assert.ElementsMatch(t, []string{"CephObjectRealms"}, deps.PluralKinds())
+		assert.ElementsMatch(t, []string{"realm-1", "realm-2"}, deps.OfPluralKind("CephObjectRealms"))
+	})
+
+	t.Run("CephNFSes", func(t *testing.T) {
+		c = newClusterdCtx(
+			&cephv1.CephNFS{ObjectMeta: meta("nfs-1")},
+			&cephv1.CephNFS{ObjectMeta: meta("nfs-2")},
+		)
+		deps, err := CephClusterDependents(c, ns)
+		assert.NoError(t, err)
+		assert.False(t, deps.Empty())
+		assert.ElementsMatch(t, []string{"CephNFSes"}, deps.PluralKinds())
+		assert.ElementsMatch(t, []string{"nfs-1", "nfs-2"}, deps.OfPluralKind("CephNFSes"))
+	})
+
+	t.Run("CephClients", func(t *testing.T) {
+		c = newClusterdCtx(
+			&cephv1.CephClient{ObjectMeta: meta("client-1")},
+			&cephv1.CephClient{ObjectMeta: meta("client-2")},
+			&cephv1.CephClient{ObjectMeta: meta("client-3")},
+		)
+		deps, err := CephClusterDependents(c, ns)
+		assert.NoError(t, err)
+		assert.False(t, deps.Empty())
+		assert.ElementsMatch(t, []string{"CephClients"}, deps.PluralKinds())
+		assert.ElementsMatch(t, []string{"client-1", "client-2", "client-3"}, deps.OfPluralKind("CephClients"))
+	})
+
+	t.Run("All", func(t *testing.T) {
+		c = newClusterdCtx(
+			&cephv1.CephBlockPool{ObjectMeta: meta("pool-1")},
+			&cephv1.CephRBDMirror{ObjectMeta: meta("rbdmirror-1")},
+			&cephv1.CephRBDMirror{ObjectMeta: meta("rbdmirror-2")},
+			&cephv1.CephFilesystem{ObjectMeta: meta("filesystem-1")},
+			&cephv1.CephFilesystemMirror{ObjectMeta: meta("fsmirror-1")},
+			&cephv1.CephFilesystemMirror{ObjectMeta: meta("fsmirror-2")},
+			&cephv1.CephObjectStore{ObjectMeta: meta("objectstore-1")},
+			&cephv1.CephObjectStoreUser{ObjectMeta: meta("u1")},
+			&cephv1.CephObjectZone{ObjectMeta: meta("zone-1")},
+			&cephv1.CephObjectZoneGroup{ObjectMeta: meta("group-1")},
+			&cephv1.CephObjectRealm{ObjectMeta: meta("realm-1")},
+			&cephv1.CephNFS{ObjectMeta: meta("nfs-1")},
+			&cephv1.CephClient{ObjectMeta: meta("client-1")},
+		)
+		deps, err := CephClusterDependents(c, ns)
+		assert.NoError(t, err)
+		assert.False(t, deps.Empty())
+		assert.ElementsMatch(t, []string{"CephBlockPools", "CephRBDMirrors", "CephFilesystems",
+			"CephFilesystemMirrors", "CephObjectStores", "CephObjectStoreUsers", "CephObjectZones",
+			"CephObjectZoneGroups", "CephObjectRealms", "CephNFSes", "CephClients"}, deps.PluralKinds())
+		assert.ElementsMatch(t, []string{"pool-1"}, deps.OfPluralKind("CephBlockPools"))
+		assert.ElementsMatch(t, []string{"rbdmirror-1", "rbdmirror-2"}, deps.OfPluralKind("CephRBDMirrors"))
+		assert.ElementsMatch(t, []string{"filesystem-1"}, deps.OfPluralKind("CephFilesystems"))
+		assert.ElementsMatch(t, []string{"fsmirror-1", "fsmirror-2"}, deps.OfPluralKind("CephFilesystemMirrors"))
+		assert.ElementsMatch(t, []string{"objectstore-1"}, deps.OfPluralKind("CephObjectStores"))
+		assert.ElementsMatch(t, []string{"u1"}, deps.OfPluralKind("CephObjectStoreUsers"))
+		assert.ElementsMatch(t, []string{"zone-1"}, deps.OfPluralKind("CephObjectZones"))
+		assert.ElementsMatch(t, []string{"group-1"}, deps.OfPluralKind("CephObjectZoneGroups"))
+		assert.ElementsMatch(t, []string{"realm-1"}, deps.OfPluralKind("CephObjectRealms"))
+		assert.ElementsMatch(t, []string{"nfs-1"}, deps.OfPluralKind("CephNFSes"))
+		assert.ElementsMatch(t, []string{"client-1"}, deps.OfPluralKind("CephClients"))
+
+		t.Run("and no dependencies in another namespace", func(t *testing.T) {
+			deps, err := CephClusterDependents(c, "other-namespace")
+			assert.NoError(t, err)
+			assert.True(t, deps.Empty())
+		})
+	})
+
+	t.Run("With errors", func(t *testing.T) {
+		dynInt := dynamicfake.NewSimpleDynamicClient(scheme,
+			&cephv1.CephBlockPool{ObjectMeta: meta("pool-1")},
+			&cephv1.CephFilesystem{ObjectMeta: meta("filesystem-1")},
+			&cephv1.CephObjectStore{ObjectMeta: meta("objectstore-1")},
+		)
+		// add reactor to cause failures when listing block and nfs (but not object, fs, or any others)
+		var listReactor k8stesting.ReactionFunc = func(action k8stesting.Action) (handled bool, ret runtime.Object, err error) {
+			r := action.GetResource().Resource
+			if r == "cephblockpools" || r == "cephnfses" {
+				return true, nil, errors.Errorf("fake error listing %q", r)
+			}
+			return false, nil, nil
+		}
+		dynInt.PrependReactor("list", "*", listReactor)
+		c := &clusterd.Context{
+			DynamicClientset: dynInt,
+		}
+		deps, err := CephClusterDependents(c, ns)
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), "CephBlockPools")
+		assert.Contains(t, err.Error(), "CephNFSes")
+		assert.False(t, deps.Empty())
+		assert.ElementsMatch(t, []string{"CephFilesystems", "CephObjectStores"}, deps.PluralKinds())
+		assert.ElementsMatch(t, []string{"filesystem-1"}, deps.OfPluralKind("CephFilesystems"))
+		assert.ElementsMatch(t, []string{"objectstore-1"}, deps.OfPluralKind("CephObjectStores"))
+	})
+}

pkg/operator/ceph/cluster/osd/integration_test.go

@@ -109,7 +109,7 @@ func testOSDIntegration(t *testing.T) {
 	}()
 	// stub out the conditionExportFunc to do nothing. we do not have a fake Rook interface that
 	// allows us to interact with a CephCluster resource like the fake K8s clientset.
-	updateConditionFunc = func(c *clusterd.Context, namespaceName types.NamespacedName, conditionType cephv1.ConditionType, status corev1.ConditionStatus, reason cephv1.ClusterReasonType, message string) {
+	updateConditionFunc = func(c *clusterd.Context, namespaceName types.NamespacedName, conditionType cephv1.ConditionType, status corev1.ConditionStatus, reason cephv1.ConditionReason, message string) {
 		// do nothing
 	}
 

pkg/operator/ceph/cluster/osd/osd_test.go

@@ -132,7 +132,7 @@ func TestAddRemoveNode(t *testing.T) {
 	}()
 	// stub out the conditionExportFunc to do nothing. we do not have a fake Rook interface that
 	// allows us to interact with a CephCluster resource like the fake K8s clientset.
-	updateConditionFunc = func(c *clusterd.Context, namespaceName types.NamespacedName, conditionType cephv1.ConditionType, status corev1.ConditionStatus, reason cephv1.ClusterReasonType, message string) {
+	updateConditionFunc = func(c *clusterd.Context, namespaceName types.NamespacedName, conditionType cephv1.ConditionType, status corev1.ConditionStatus, reason cephv1.ConditionReason, message string) {
 		// do nothing
 	}
 

pkg/operator/ceph/cluster/osd/update_test.go

@@ -120,7 +120,7 @@ func Test_updateExistingOSDs(t *testing.T) {
 
 	// stub out the conditionExportFunc to do nothing. we do not have a fake Rook interface that
 	// allows us to interact with a CephCluster resource like the fake K8s clientset.
-	updateConditionFunc = func(c *clusterd.Context, namespaceName types.NamespacedName, conditionType cephv1.ConditionType, status corev1.ConditionStatus, reason cephv1.ClusterReasonType, message string) {
+	updateConditionFunc = func(c *clusterd.Context, namespaceName types.NamespacedName, conditionType cephv1.ConditionType, status corev1.ConditionStatus, reason cephv1.ConditionReason, message string) {
 		// do nothing
 	}
 	shouldCheckOkToStopFunc = func(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo) bool {

pkg/operator/ceph/controller/conditions.go

@@ -29,9 +29,10 @@ import (
 )
 
 // UpdateCondition function will export each condition into the cluster custom resource
-func UpdateCondition(c *clusterd.Context, namespaceName types.NamespacedName, conditionType cephv1.ConditionType, status v1.ConditionStatus, reason cephv1.ClusterReasonType, message string) {
-	cluster, err := c.RookClientset.CephV1().CephClusters(namespaceName.Namespace).Get(context.TODO(), namespaceName.Name, metav1.GetOptions{})
-	if err != nil {
+func UpdateCondition(c *clusterd.Context, namespaceName types.NamespacedName, conditionType cephv1.ConditionType, status v1.ConditionStatus, reason cephv1.ConditionReason, message string) {
+	// use client.Client unit test this more easily with updating statuses which must use the client
+	cluster := &cephv1.CephCluster{}
+	if err := c.Client.Get(context.TODO(), namespaceName, cluster); err != nil {
 		logger.Errorf("failed to get cluster %v to update the conditions. %v", namespaceName, err)
 		return
 	}
@@ -41,7 +42,7 @@ func UpdateCondition(c *clusterd.Context, namespaceName types.NamespacedName, co
 
 // UpdateClusterCondition function will export each condition into the cluster custom resource
 func UpdateClusterCondition(c *clusterd.Context, cluster *cephv1.CephCluster, namespaceName types.NamespacedName, conditionType cephv1.ConditionType, status v1.ConditionStatus,
-	reason cephv1.ClusterReasonType, message string, preserveAllConditions bool) {
+	reason cephv1.ConditionReason, message string, preserveAllConditions bool) {
 
 	// Keep the conditions that already existed if they are in the list of long-term conditions,
 	// otherwise discard the temporary conditions
@@ -52,7 +53,11 @@ func UpdateClusterCondition(c *clusterd.Context, cluster *cephv1.CephCluster, na
 		// The transient conditions are not persisted. However, if the currently requested condition is not expected to
 		// reset the transient conditions, they are retained. For example, if the operator is checking for ceph health
 		// in the middle of the reconcile, the progress condition should not be reset by the status check update.
-		if preserveAllConditions || condition.Reason == cephv1.ClusterCreatedReason || condition.Reason == cephv1.ClusterConnectedReason {
+		if preserveAllConditions ||
+			condition.Reason == cephv1.ClusterCreatedReason ||
+			condition.Reason == cephv1.ClusterConnectedReason ||
+			condition.Type == cephv1.ConditionDeleting ||
+			condition.Type == cephv1.ConditionDeletionIsBlocked {
 			if conditionType != condition.Type {
 				conditions = append(conditions, condition)
 				continue
@@ -83,12 +88,15 @@ func UpdateClusterCondition(c *clusterd.Context, cluster *cephv1.CephCluster, na
 	conditions = append(conditions, *currentCondition)
 	cluster.Status.Conditions = conditions
 
-	cluster.Status.Phase = conditionType
-	if state := translatePhasetoState(conditionType, status); state != "" {
-		cluster.Status.State = state
+	// Once the cluster begins deleting, the phase should not revert back to any other phase
+	if cluster.Status.Phase != cephv1.ConditionDeleting {
+		cluster.Status.Phase = conditionType
+		if state := translatePhasetoState(conditionType, status); state != "" {
+			cluster.Status.State = state
+		}
+		cluster.Status.Message = currentCondition.Message
+		logger.Debugf("CephCluster %q status: %q. %q", namespaceName.Namespace, cluster.Status.Phase, cluster.Status.Message)
 	}
-	cluster.Status.Message = currentCondition.Message
-	logger.Debugf("CephCluster %q status: %q. %q", namespaceName.Namespace, cluster.Status.Phase, cluster.Status.Message)
 
 	if err := UpdateStatus(c.Client, cluster); err != nil {
 		logger.Errorf("failed to update cluster condition to %+v. %v", *currentCondition, err)
@@ -115,6 +123,9 @@ func translatePhasetoState(phase cephv1.ConditionType, status v1.ConditionStatus
 		return cephv1.ClusterStateCreating
 	case cephv1.ConditionReady:
 		return cephv1.ClusterStateCreated
+	case cephv1.ConditionDeleting:
+		// "Deleting" was not a state before, so just translate the "Deleting" condition directly.
+		return cephv1.ClusterState(cephv1.ConditionDeleting)
 	default:
 		return ""
 	}

pkg/util/dependents/dependents.go

@@ -0,0 +1,90 @@
+/*
+Copyright 2021 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 dependents
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+)
+
+// 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 {
+	d map[string][]string // map from dependent Resource to a list of dependent names
+}
+
+// NewDependentList creates a new empty DependentList.
+func NewDependentList() *DependentList {
+	return &DependentList{
+		d: make(map[string][]string),
+	}
+}
+
+// Empty returns true if the DependentList is empty or false otherwise.
+func (d *DependentList) Empty() bool {
+	return len(d.d) == 0
+}
+
+// Add adds a dependent name for a plural Kind to the DependentList.
+func (d *DependentList) Add(pluralKind string, name string) {
+	names, ok := d.d[pluralKind]
+	if !ok {
+		d.d[pluralKind] = []string{name}
+		return
+	}
+	d.d[pluralKind] = append(names, name)
+}
+
+// PluralKinds returns the plural Kinds that have dependents.
+func (d *DependentList) PluralKinds() []string {
+	kinds := []string{}
+	for k := range d.d {
+		kinds = append(kinds, k)
+	}
+	return kinds
+}
+
+// OfPluralKind returns the names of dependents of the Kind (plural), or an empty list if no
+// dependents exist.
+func (d *DependentList) OfPluralKind(pluralKind string) []string {
+	names, ok := d.d[pluralKind]
+	if !ok {
+		return []string{}
+	}
+	return names
+}
+
+// StringWithHeader outputs the dependent list as a pretty-printed string headed with the given
+// formatting directive (followed by a colon). It outputs dependents in alphabetical order by the
+// plural Kind.
+// Example:
+//    StringWithHeader("dependents of my %q", "mom")  -->
+//    `dependents of my "mom": FirstResources: [name1], SecondResources: [name2 name2 name3]`
+func (d *DependentList) StringWithHeader(headerFormat string, args ...interface{}) string {
+	header := fmt.Sprintf(headerFormat, args...)
+	if len(d.d) == 0 {
+		return fmt.Sprintf("%s: none", header)
+	}
+	deps := make([]string, 0, len(d.d))
+	for pluralKind, names := range d.d {
+		deps = append(deps, fmt.Sprintf("%s: %v", pluralKind, names))
+	}
+	sort.Strings(deps) // always output a consistent ordering
+	allDeps := strings.Join(deps, ", ")
+	return fmt.Sprintf("%s: %s", header, allDeps)
+}

pkg/util/dependents/dependents_test.go

@@ -0,0 +1,97 @@
+/*
+Copyright 2021 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 dependents
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDependentList(t *testing.T) {
+	containsExactlyOne := func(s, substr string) {
+		assert.Equal(t, 1, strings.Count(s, substr))
+	}
+	isBefore := func(s, before, after string) {
+		assert.Less(t, strings.Index(s, before), strings.Index(s, after))
+	}
+
+	t.Run("empty", func(t *testing.T) {
+		d := NewDependentList()
+		assert.True(t, d.Empty())
+	})
+
+	t.Run("one resource, one dependent", func(t *testing.T) {
+		d := NewDependentList()
+		d.Add("MyResources", "my-resource-1")
+		assert.False(t, d.Empty())
+		assert.ElementsMatch(t, []string{"MyResources"}, d.PluralKinds())
+		assert.ElementsMatch(t, []string{"my-resource-1"}, d.OfPluralKind("MyResources"))
+		toString := d.StringWithHeader("header")
+		containsExactlyOne(toString, "header:")
+		containsExactlyOne(toString, "MyResources")
+		containsExactlyOne(toString, "my-resource-1")
+	})
+
+	t.Run("one resource - multiple dependents", func(t *testing.T) {
+		d := NewDependentList()
+		d.Add("MyResources", "my-resource-1")
+		d.Add("MyResources", "my-resource-2")
+		d.Add("MyResources", "my-resource-3")
+		assert.False(t, d.Empty())
+		assert.ElementsMatch(t, []string{"MyResources"}, d.PluralKinds())
+		assert.ElementsMatch(t, []string{"my-resource-1", "my-resource-2", "my-resource-3"}, d.OfPluralKind("MyResources"))
+		assert.ElementsMatch(t, []string{}, d.OfPluralKind("OtherKinds"))
+		toString := d.StringWithHeader("head with arg %d", 1)
+		containsExactlyOne(toString, "head with arg 1:")
+		containsExactlyOne(toString, "MyResources")
+		containsExactlyOne(toString, "my-resource-1")
+		containsExactlyOne(toString, "my-resource-2")
+		containsExactlyOne(toString, "my-resource-3")
+
+	})
+
+	t.Run("multiple resources - multiple dependents", func(t *testing.T) {
+		d := NewDependentList()
+		d.Add("MyResources", "my-resource-2")
+		d.Add("MyResources", "my-resource-4")
+		d.Add("YourResources", "your-resource-1")
+		d.Add("TheirResources", "their-resource-5")
+		d.Add("TheirResources", "their-resource-6")
+		assert.False(t, d.Empty())
+		assert.ElementsMatch(t, []string{"MyResources", "YourResources", "TheirResources"}, d.PluralKinds())
+		assert.ElementsMatch(t, []string{"my-resource-2", "my-resource-4"}, d.OfPluralKind("MyResources"))
+		assert.ElementsMatch(t, []string{"your-resource-1"}, d.OfPluralKind("YourResources"))
+		assert.ElementsMatch(t, []string{"their-resource-5", "their-resource-6"}, d.OfPluralKind("TheirResources"))
+		assert.ElementsMatch(t, []string{}, d.OfPluralKind("OtherKinds"))
+		toString := d.StringWithHeader("head with arg %s", "mom")
+		t.Log(toString)
+		containsExactlyOne(toString, "head with arg mom:")
+		containsExactlyOne(toString, "MyResources")
+		containsExactlyOne(toString, "my-resource-2")
+		containsExactlyOne(toString, "my-resource-4")
+		containsExactlyOne(toString, "YourResources")
+		containsExactlyOne(toString, "your-resource-1")
+		containsExactlyOne(toString, "TheirResources")
+		containsExactlyOne(toString, "their-resource-5")
+		containsExactlyOne(toString, "their-resource-6")
+		// ensure alphabetical ordering
+		isBefore(toString, "MyResources", "TheirResources")
+		isBefore(toString, "TheirResources", "YourResources")
+	})
+}

pkg/util/error.go

@@ -0,0 +1,44 @@
+/*
+Copyright 2021 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 util
+
+import (
+	"fmt"
+
+	"github.com/pkg/errors"
+)
+
+// AggregateErrors takes a list of errors formats them into a pretty, user-readable list headed by
+// the text "errors:". All errors in the list will lose any context besides their error string.
+// If the errs list is empty, AggregateErrors returns nil.
+// Example:
+//    AggregateErrors(errList, "errors for my %q", "mom")  -->
+//    `errors for my "mom":
+//        error 1
+//        error 2
+//        etc.`
+func AggregateErrors(errs []error, format string, args ...interface{}) error {
+	if len(errs) == 0 {
+		return nil
+	}
+
+	errString := fmt.Sprintf(format+":", args...)
+	for _, err := range errs {
+		errString = fmt.Sprintf("%s\n    %s", errString, err.Error())
+	}
+	return errors.Errorf(errString)
+}