Merge pull request #16333 from rook/mergify/bp/release-1.17/pr-16329

pool: fix cephblockpool remove when related pool is already removed (backport #16329)

pkg/daemon/ceph/client/pool.go

@@ -241,6 +241,20 @@ func IsPoolEmpty(context *clusterd.Context, clusterInfo *ClusterInfo, name strin
 	return true, fmt.Sprintf("pool %q is empty and can be deleted", name), nil
 }
 
+func IsPoolPresent(context *clusterd.Context, clusterInfo *ClusterInfo, poolName string) (bool, error) {
+	pools, err := ListPoolSummaries(context, clusterInfo)
+	if err != nil {
+		return false, errors.Wrap(err, "failed to list pools")
+	}
+
+	for _, pool := range pools {
+		if pool.Name == poolName {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
 func checkForImagesInPool(context *clusterd.Context, clusterInfo *ClusterInfo, name string) error {
 	isEmpty, err := checkIsPoolEmpty(context, clusterInfo, name)
 	if err != nil {

pkg/operator/ceph/pool/controller.go

@@ -430,12 +430,22 @@ func (r *ReconcileCephBlockPool) handleDeletionBlocked(cephBlockPool *cephv1.Cep
 	logger.Info(depCondition.Message)
 
 	radosNamespaces := deps.OfKind(radosNamespacesKeyName)
-	isEmpty, emptyMessage, err := cephclient.IsPoolEmpty(r.context, r.clusterInfo, poolSpec.Name, radosNamespaces)
+	poolPresent, err := cephclient.IsPoolPresent(r.context, r.clusterInfo, poolSpec.Name)
 	if err != nil {
-		return err
+		return errors.Wrap(err, "failed to check pool presence")
+	}
+	var isEmpty bool
+	var emptyMessage string
+	if poolPresent {
+		isEmpty, emptyMessage, err = cephclient.IsPoolEmpty(r.context, r.clusterInfo, poolSpec.Name, radosNamespaces)
+		if err != nil {
+			return err
+		}
+	} else {
+		emptyMessage = fmt.Sprintf("pool %q is not found in cluster", poolSpec.Name)
 	}
 	var emptyCondition cephv1.Condition
-	if isEmpty {
+	if isEmpty || !poolPresent {
 		emptyCondition = dependents.DeletionBlockedDueToNonEmptyPoolCondition(false, emptyMessage)
 	} else {
 		deletionBlocked = true
@@ -519,21 +529,16 @@ func createPool(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo,
 
 // Delete the pool
 func deletePool(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, p *cephv1.NamedPoolSpec) error {
-	pools, err := cephclient.ListPoolSummaries(context, clusterInfo)
+	poolPresent, err := cephclient.IsPoolPresent(context, clusterInfo, p.Name)
 	if err != nil {
-		return errors.Wrap(err, "failed to list pools")
+		return errors.Wrap(err, "failed to check pool presence")
 	}
-
-	// Only delete the pool if it exists...
-	for _, pool := range pools {
-		if pool.Name == p.Name {
-			err := cephclient.DeletePool(context, clusterInfo, p.Name)
-			if err != nil {
-				return errors.Wrapf(err, "failed to delete pool %q", p.Name)
-			}
+	if poolPresent {
+		err := cephclient.DeletePool(context, clusterInfo, p.Name)
+		if err != nil {
+			return errors.Wrapf(err, "failed to delete pool %q", p.Name)
 		}
 	}
-
 	return nil
 }
 

pkg/operator/ceph/pool/controller_test.go

@@ -588,6 +588,7 @@ func TestDeletionBlocked(t *testing.T) {
 	)
 	blockImageCount := 0
 	rnsImageCount := 0
+	poolPresent := false
 	cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
@@ -601,6 +602,14 @@ func TestDeletionBlocked(t *testing.T) {
 						return fmt.Sprintf(response, blockImageCount), nil
 					}
 				}
+			} else if command == "ceph" && args[0] == "osd" {
+				if args[1] == "lspools" {
+					if poolPresent {
+						return fmt.Sprintf(`[{"poolnum":1,"poolname":"%s"}]`, pool.Name), nil
+					} else {
+						return "[]", nil
+					}
+				}
 			}
 			return "not implemented", nil
 		},
@@ -621,6 +630,24 @@ func TestDeletionBlocked(t *testing.T) {
 		clusterInfo:       cephclient.AdminTestClusterInfo("mycluster"),
 	}
 	cephCluster := &cephv1.CephCluster{}
+	t.Run("deletion is allowed with no pool actually present", 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))
+	})
+	poolPresent = true
 	t.Run("deletion is allowed with no images or rns", func(t *testing.T) {
 		err := r.handleDeletionBlocked(pool, cephCluster)
 		assert.NoError(t, err)