Merge pull request #17548 from rook/mergify/bp/release-1.20/pr-17535

ceph: skip base rule creation if pools exit (backport #17535)

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{