diff --git a/pkg/daemon/ceph/client/pool.go b/pkg/daemon/ceph/client/pool.go index 6e2958a8b..fae652bf0 100644 --- a/pkg/daemon/ceph/client/pool.go +++ b/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)} diff --git a/pkg/daemon/ceph/client/pool_test.go b/pkg/daemon/ceph/client/pool_test.go index 12301e10b..37bee7415 100644 --- a/pkg/daemon/ceph/client/pool_test.go +++ b/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{