pool: skip base rule creation of pools exit

createREplicatedPoolForApp creates a base Crush rule (named after the
pool) on every reconcile, even if the pool already exists and uses a
suffixed rule ( eg pool_host_ssd). This causes orphaned rules to
reappear after every operator restart, immediately after
CleanupUnusedCrushRules deletes them.

This PR ensures that the base rule is created only if the pool does not
exist yet.

Signed-off-by: Santosh <sapillai@redhat.com>

pkg/daemon/ceph/client/pool.go

@@ -457,6 +457,14 @@ func createReplicatedPoolForApp(context *clusterd.Context, clusterInfo *ClusterI
 
 	// The crush rule name is the same as the pool unless we have a stretch cluster.
 	crushRuleName := pool.Name
+
+	// Check if the pool already exists before creating crush rules
+	poolExists := false
+	poolDetails, err := GetPoolDetails(context, clusterInfo, pool.Name)
+	if err == nil {
+		poolExists = true
+	}
+
 	if clusterSpec.IsStretchCluster() {
 		// A stretch cluster enforces using the same crush rule for all pools.
 		// The stretch cluster rule is created initially by the operator when the stretch cluster is configured
@@ -478,14 +486,17 @@ func createReplicatedPoolForApp(context *clusterd.Context, clusterInfo *ClusterI
 		} else {
 			// create a crush rule for a replicated pool, if a failure domain is specified
 			checkFailureDomain = true
-			if err := createReplicationCrushRule(context, clusterInfo, clusterSpec, crushRuleName, pool); err != nil {
-				return errors.Wrapf(err, "failed to create replicated crush rule %q", crushRuleName)
+			// Only create the initial crush rule if the pool doesn't exist yet.
+			// Any updates to the crush rule will be applied later in the reconcile.
+			if !poolExists {
+				if err := createReplicationCrushRule(context, clusterInfo, clusterSpec, crushRuleName, pool); err != nil {
+					return errors.Wrapf(err, "failed to create replicated crush rule %q", crushRuleName)
+				}
 			}
 		}
 	}
 
-	poolDetails, err := GetPoolDetails(context, clusterInfo, pool.Name)
-	if err != nil {
+	if !poolExists {
 		// Create the pool since it doesn't exist yet
 		// If there was some error other than ENOENT (not exists), go ahead and ensure the pool is created anyway
 		args := []string{"osd", "pool", "create", pool.Name, pgCount, "replicated", crushRuleName, "--size", strconv.FormatUint(uint64(pool.Replicated.Size), 10)}

pkg/daemon/ceph/client/pool_test.go

@@ -251,7 +251,8 @@ func testCreateReplicaPool(t *testing.T, failureDomain, crushRoot, deviceClass,
 	clusterSpec := &cephv1.ClusterSpec{Storage: cephv1.StorageScopeSpec{Config: map[string]string{CrushRootConfigKey: "cluster-crush-root"}}}
 	err := createReplicatedPoolForApp(context, AdminTestClusterInfo("mycluster"), clusterSpec, p, DefaultPGCount)
 	assert.Nil(t, err)
-	assert.True(t, crushRuleCreated)
+	// The base crush rule should not be recreated for an existing pool.
+	assert.False(t, crushRuleCreated)
 	if compressionMode != "" {
 		assert.True(t, compressionModeCreated)
 	} else {
@@ -438,6 +439,47 @@ func TestCleanupUnusedCrushRulesNoPools(t *testing.T) {
 	assert.Empty(t, removedRules)
 }
 
+func TestExistingPoolDoesNotRecreateBaseCrushRule(t *testing.T) {
+	crushRuleCreated := false
+	executor := &exectest.MockExecutor{}
+	context := &clusterd.Context{Executor: executor}
+	executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
+		logger.Infof("Command: %s %v", command, args)
+		if args[1] == "pool" && args[2] == "get" {
+			// Pool already exists and uses a suffixed crush rule
+			return `{"pool":"mypool","pool_id":1,"size":3,"crush_rule":"mypool_host_ssd"}`, nil
+		}
+		if args[1] == "crush" && args[2] == "rule" {
+			if args[3] == "create-replicated" {
+				crushRuleCreated = true
+				return "", nil
+			}
+			if args[3] == "dump" {
+				return `{"steps": [{"item_name":"default~ssd"},{"type":"host"}]}`, nil
+			}
+		}
+		if args[1] == "pool" && args[2] == "set" {
+			return "", nil
+		}
+		return "", errors.Errorf("unexpected ceph command %q", args)
+	}
+
+	enableCrushUpdates := true
+	p := cephv1.NamedPoolSpec{
+		Name: "mypool",
+		PoolSpec: cephv1.PoolSpec{
+			FailureDomain:      "host",
+			DeviceClass:        "ssd",
+			Replicated:         cephv1.ReplicatedSpec{Size: 3},
+			EnableCrushUpdates: &enableCrushUpdates,
+		},
+	}
+	clusterSpec := &cephv1.ClusterSpec{Storage: cephv1.StorageScopeSpec{}}
+	err := createReplicatedPoolForApp(context, AdminTestClusterInfo("mycluster"), clusterSpec, p, DefaultPGCount)
+	assert.NoError(t, err)
+	assert.False(t, crushRuleCreated, "base crush rule should not be created when the pool already exists")
+}
+
 func TestExtractPoolDetails(t *testing.T) {
 	t.Run("complex crush rule skipped", func(t *testing.T) {
 		rule := ruleSpec{Steps: []stepSpec{