Condition represents a status condition on any Rook-Ceph Custom Resource.
@@ -5275,6 +5287,22 @@ deletion.
ObjectHasNoDependentsReason represents when a resource object has no dependents that are
blocking deletion.
|
+
"PoolEmpty" |
+PoolEmptyReason represents when a pool does not contain images or snapshots that are blocking
+deletion.
+ |
+
"PoolNotEmpty" |
+PoolNotEmptyReason represents when a pool contains images or snapshots that are blocking
+deletion.
+ |
+
"RadosNamespaceEmpty" |
+RadosNamespaceEmptyReason represents when a rados namespace does not contain images or snapshots that are blocking
+deletion.
+ |
+
"RadosNamespaceNotEmpty" |
+RadosNamespaceNotEmptyReason represents when a rados namespace contains images or snapshots that are blocking
+deletion.
+ |
"ReconcileFailed" |
ReconcileFailed represents when a resource reconciliation failed.
|
@@ -5316,9 +5344,15 @@ blocking deletion.
"Failure" |
ConditionFailure represents Failure state of an object
|
+
"PoolDeletionIsBlocked" |
+ConditionPoolDeletionIsBlocked represents when deletion of the object is blocked.
+ |
"Progressing" |
ConditionProgressing represents Progressing state of an object
|
+
"RadosNamespaceDeletionIsBlocked" |
+ConditionRadosNSDeletionIsBlocked represents when deletion of the object is blocked.
+ |
"Ready" |
ConditionReady represents Ready state of an object
|
diff --git a/deploy/charts/rook-ceph/templates/resources.yaml b/deploy/charts/rook-ceph/templates/resources.yaml
index 93e44f04d..3c53d41cc 100644
--- a/deploy/charts/rook-ceph/templates/resources.yaml
+++ b/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
diff --git a/deploy/examples/crds.yaml b/deploy/examples/crds.yaml
index 516afd049..d59a19ca0 100644
--- a/deploy/examples/crds.yaml
+++ b/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
diff --git a/pkg/apis/ceph.rook.io/v1/namespace.go b/pkg/apis/ceph.rook.io/v1/namespace.go
new file mode 100644
index 000000000..58b4b3e45
--- /dev/null
+++ b/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 = ""
+ 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
+}
diff --git a/pkg/apis/ceph.rook.io/v1/pool.go b/pkg/apis/ceph.rook.io/v1/pool.go
index 14fa7afe3..2ebe92374 100644
--- a/pkg/apis/ceph.rook.io/v1/pool.go
+++ b/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
diff --git a/pkg/apis/ceph.rook.io/v1/types.go b/pkg/apis/ceph.rook.io/v1/types.go
index a15f4e868..0bea94326 100755
--- a/pkg/apis/ceph.rook.io/v1/types.go
+++ b/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.
diff --git a/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go b/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go
index 5ad7a281d..987aa341b 100644
--- a/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go
+++ b/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
}
diff --git a/pkg/daemon/ceph/client/mirror.go b/pkg/daemon/ceph/client/mirror.go
index d3c8e82f7..06eb6c895 100644
--- a/pkg/daemon/ceph/client/mirror.go
+++ b/pkg/daemon/ceph/client/mirror.go
@@ -53,10 +53,8 @@ type Images struct {
}
const (
- mirrorModeDisabled = "disabled"
- mirrorModeInitOnly = "init-only"
- ImplicitNamespaceKey = ""
- 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}
diff --git a/pkg/daemon/ceph/client/pool.go b/pkg/daemon/ceph/client/pool.go
index 9723056ea..c60fa5965 100644
--- a/pkg/daemon/ceph/client/pool.go
+++ b/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
diff --git a/pkg/daemon/ceph/client/radosnamespace.go b/pkg/daemon/ceph/client/radosnamespace.go
index bb6b185a8..2d5c47864 100644
--- a/pkg/daemon/ceph/client/radosnamespace.go
+++ b/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
diff --git a/pkg/operator/ceph/pool/controller.go b/pkg/operator/ceph/pool/controller.go
index be3eaba23..efeaf6580 100644
--- a/pkg/operator/ceph/pool/controller.go
+++ b/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)
diff --git a/pkg/operator/ceph/pool/controller_test.go b/pkg/operator/ceph/pool/controller_test.go
index 7891337c0..861f80afb 100644
--- a/pkg/operator/ceph/pool/controller_test.go
+++ b/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{}
diff --git a/pkg/operator/ceph/pool/dependents.go b/pkg/operator/ceph/pool/dependents.go
index a0b8fb7ae..0d20f4e6a 100644
--- a/pkg/operator/ceph/pool/dependents.go
+++ b/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)
}
diff --git a/pkg/operator/ceph/pool/radosnamespace/controller.go b/pkg/operator/ceph/pool/radosnamespace/controller.go
index 49f1be3d6..11d5dc794 100644
--- a/pkg/operator/ceph/pool/radosnamespace/controller.go
+++ b/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
}
diff --git a/pkg/operator/ceph/pool/radosnamespace/controller_test.go b/pkg/operator/ceph/pool/radosnamespace/controller_test.go
index dad2e6489..e0df465a4 100644
--- a/pkg/operator/ceph/pool/radosnamespace/controller_test.go
+++ b/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)
}
})
}
diff --git a/pkg/operator/ceph/reporting/reporting.go b/pkg/operator/ceph/reporting/reporting.go
index 8a3cf28a4..0c4c9a0e6 100644
--- a/pkg/operator/ceph/reporting/reporting.go
+++ b/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)
}
}
diff --git a/pkg/operator/ceph/reporting/status.go b/pkg/operator/ceph/reporting/status.go
index 9ba0b0447..874e129fe 100644
--- a/pkg/operator/ceph/reporting/status.go
+++ b/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
diff --git a/pkg/operator/ceph/reporting/status_test.go b/pkg/operator/ceph/reporting/status_test.go
index 522438c78..5aa8b4679 100644
--- a/pkg/operator/ceph/reporting/status_test.go
+++ b/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)
diff --git a/pkg/util/dependents/dependents.go b/pkg/util/dependents/dependents.go
index 262b364b5..860ee7987 100644
--- a/pkg/util/dependents/dependents.go
+++ b/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 {