pool: fix cephblockpool remove when related pool is already removed from cluster

When removeing CephBlockPool check first that related ceph pool is exist in cluster,
since it may be manually removed by mistake.

Related-Issue: #16328
Signed-off-by: Denis Egorenko <degorenko@mirantis.com>
(cherry picked from commit c9e0f89a8b)

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)