diff --git a/Documentation/Helm-Charts/operator-chart.md b/Documentation/Helm-Charts/operator-chart.md index 0b751fdc5..5a6354c10 100644 --- a/Documentation/Helm-Charts/operator-chart.md +++ b/Documentation/Helm-Charts/operator-chart.md @@ -141,6 +141,7 @@ The following table lists the configurable parameters of the rook-operator chart | `csi.topology.enabled` | Enable topology based provisioning | `false` | | `currentNamespaceOnly` | Whether the operator should watch cluster CRD in its own namespace or not | `false` | | `customHostnameLabel` | Custom label to identify node hostname. If not set `kubernetes.io/hostname` will be used | `nil` | +| `deleteUnusedCrushRules` | If true, delete unused generated CRUSH rules after the mgr starts | `true` | | `disableDeviceHotplug` | Disable automatic orchestration when new devices are discovered. | `false` | | `discover.nodeAffinity` | The node labels for affinity of `discover-agent` [^1] | `nil` | | `discover.podLabels` | Labels to add to the discover pods | `nil` | diff --git a/PendingReleaseNotes.md b/PendingReleaseNotes.md index eeba44069..92e7c2b99 100644 --- a/PendingReleaseNotes.md +++ b/PendingReleaseNotes.md @@ -9,3 +9,4 @@ - RGW Account support: Added `CephObjectStoreAccount` CRD for managing RGW accounts, and `accountRef` field in `CephObjectStoreUser` to associate users with accounts. This feature is experimental and currently only supported with the Ceph main branch image (`quay.ceph.io/ceph-ci/ceph:main`). See the [Object Store Accounts](Documentation/Storage-Configuration/Object-Storage-RGW/ceph-object-accounts.md) documentation for more details. - SSE-S3 with Vault Agent: Added support for server-side encryption with SSE-S3 using HashiCorp Vault Agent authentication. See the [CephObjectStore Security Settings](Documentation/CRDs/Object-Storage/ceph-object-store-crd.md#sse-s3-with-vault-agent) for more details. +- Unused CRUSH rule cleanup: Rook now deletes unused CRUSH rules by default after the Ceph mgr starts. Set `ROOK_DELETE_UNUSED_CRUSH_RULES` to `false` in the operator config to disable this cleanup. diff --git a/deploy/charts/rook-ceph/templates/configmap.yaml b/deploy/charts/rook-ceph/templates/configmap.yaml index 4b7a1f64a..fcf9f1583 100644 --- a/deploy/charts/rook-ceph/templates/configmap.yaml +++ b/deploy/charts/rook-ceph/templates/configmap.yaml @@ -22,6 +22,7 @@ data: ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS: {{ . | quote }} {{- end }} ROOK_CEPH_ALLOW_LOOP_DEVICES: {{ .Values.allowLoopDevices | quote }} + ROOK_DELETE_UNUSED_CRUSH_RULES: {{ .Values.deleteUnusedCrushRules | quote }} ROOK_ENABLE_DISCOVERY_DAEMON: {{ .Values.enableDiscoveryDaemon | quote }} {{- with .Values.discoverDaemonUdev }} DISCOVER_DAEMON_UDEV_BLACKLIST: {{ . | quote }} diff --git a/deploy/charts/rook-ceph/values.yaml b/deploy/charts/rook-ceph/values.yaml index ccf8a95df..4cdddd18a 100644 --- a/deploy/charts/rook-ceph/values.yaml +++ b/deploy/charts/rook-ceph/values.yaml @@ -77,6 +77,9 @@ containerSecurityContext: # -- If true, loop devices are allowed to be used for osds in test clusters allowLoopDevices: false +# -- If true, delete unused generated CRUSH rules after the mgr starts +deleteUnusedCrushRules: true + ceph-csi-operator: nameOverride: ceph-csi fullnameOverride: ceph-csi diff --git a/deploy/examples/operator.yaml b/deploy/examples/operator.yaml index 404f306b8..747d9b083 100644 --- a/deploy/examples/operator.yaml +++ b/deploy/examples/operator.yaml @@ -31,6 +31,10 @@ data: # Allow using loop devices for osds in test clusters. ROOK_CEPH_ALLOW_LOOP_DEVICES: "false" + # Delete unused generated CRUSH rules after the mgr starts. Disable this if + # you need Rook to leave custom CRUSH rules in place. + ROOK_DELETE_UNUSED_CRUSH_RULES: "true" + # Enable CSI Operator ROOK_USE_CSI_OPERATOR: "true" # Enable the CSI driver. diff --git a/pkg/daemon/ceph/client/filesystem_test.go b/pkg/daemon/ceph/client/filesystem_test.go index db302a8a3..f55e019da 100644 --- a/pkg/daemon/ceph/client/filesystem_test.go +++ b/pkg/daemon/ceph/client/filesystem_test.go @@ -94,7 +94,7 @@ func TestFilesystemGetMarshal(t *testing.T) { func TestFilesystemRemove(t *testing.T) { dataDeleted := false metadataDeleted := false - crushDeleted := false + deletedPools := map[string]bool{} executor := &exectest.MockExecutor{} context := &clusterd.Context{Executor: executor} fs := CephFilesystemDetails{ @@ -119,9 +119,12 @@ func TestFilesystemRemove(t *testing.T) { } if args[0] == "osd" { if args[1] == "lspools" { - pools := []*CephStoragePoolSummary{ - {Name: "mydata", Number: 1}, - {Name: "mymetadata", Number: 2}, + pools := []*CephStoragePoolSummary{} + if !deletedPools["mydata"] { + pools = append(pools, &CephStoragePoolSummary{Name: "mydata", Number: 1}) + } + if !deletedPools["mymetadata"] { + pools = append(pools, &CephStoragePoolSummary{Name: "mymetadata", Number: 2}) } output, err := json.Marshal(pools) assert.Nil(t, err) @@ -129,25 +132,21 @@ func TestFilesystemRemove(t *testing.T) { } if args[1] == "pool" { if args[2] == "get" { - return `{"pool_id":1}`, nil + return `{"pool_id":1,"crush_rule":"myrule"}`, nil } if args[2] == "delete" { if args[3] == "mydata" { + deletedPools["mydata"] = true dataDeleted = true return "", nil } if args[3] == "mymetadata" { + deletedPools["mymetadata"] = true metadataDeleted = true return "", nil } } } - if args[1] == "crush" { - assert.Equal(t, "rule", args[2]) - assert.Equal(t, "rm", args[3]) - crushDeleted = true - return "", nil - } } emptyPool := "{\"images\":{\"count\":0,\"provisioned_bytes\":0,\"snap_count\":0},\"trash\":{\"count\":1,\"provisioned_bytes\":2048,\"snap_count\":0}}" if args[0] == "pool" { @@ -162,7 +161,6 @@ func TestFilesystemRemove(t *testing.T) { assert.Nil(t, err) assert.True(t, metadataDeleted) assert.True(t, dataDeleted) - assert.True(t, crushDeleted) } func TestFailAllStandbyReplayMDS(t *testing.T) { diff --git a/pkg/daemon/ceph/client/pool.go b/pkg/daemon/ceph/client/pool.go index 76e375100..1c732e8f4 100644 --- a/pkg/daemon/ceph/client/pool.go +++ b/pkg/daemon/ceph/client/pool.go @@ -23,10 +23,12 @@ import ( "path/filepath" "strconv" "strings" + "sync" "github.com/pkg/errors" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" "github.com/rook/rook/pkg/clusterd" + "github.com/rook/rook/pkg/util/log" "k8s.io/apimachinery/pkg/api/resource" ) @@ -39,6 +41,12 @@ const ( PgAutoscaleModeOn = "on" ) +// crushRuleMutex coordinates crush rule cleanup with pool reconciles. Pool +// reconciles take a read lock while they create or update a crush rule and +// attach it to a pool. Cleanup takes the write lock so it cannot delete a rule +// until all in-flight pool reconciles have finished referencing it. +var crushRuleMutex sync.RWMutex + type CephStoragePoolSummary struct { Name string `json:"poolname"` Number int `json:"poolnum"` @@ -305,11 +313,10 @@ func DeletePool(context *clusterd.Context, clusterInfo *ClusterInfo, name string return errors.Wrapf(err, "failed to delete pool %q. %s", name, string(output)) } - // remove the crush rule for this pool and ignore the error in case the rule is still in use or not found - args = []string{"osd", "crush", "rule", "rm", name} - output, err = NewCephCommand(context, clusterInfo, args).Run() - if err != nil { - logger.Errorf("failed to delete crush rule %q. %v. %s", name, err, string(output)) + if pool.CrushRule != "" { + if err := deleteCrushRule(context, clusterInfo, pool.CrushRule); err != nil { + logger.Warningf("failed to delete crush rule %q after deleting pool %q. %v", pool.CrushRule, name, err) + } } logger.Infof("purge completed for pool %q", name) @@ -443,6 +450,11 @@ func createReplicatedPoolForApp(context *clusterd.Context, clusterInfo *ClusterI // If it's a replicated pool, ensure the failure domain is desired checkFailureDomain := false + // Hold a read lock until the pool references any crush rule changes made + // during this reconcile so cleanup cannot remove the rule prematurely. + crushRuleMutex.RLock() + defer crushRuleMutex.RUnlock() + // The crush rule name is the same as the pool unless we have a stretch cluster. crushRuleName := pool.Name if clusterSpec.IsStretchCluster() { @@ -502,7 +514,7 @@ func createReplicatedPoolForApp(context *clusterd.Context, clusterInfo *ClusterI logger.Infof("reconciling replicated pool %s succeeded", pool.Name) if checkFailureDomain || pool.PoolSpec.DeviceClass != "" { - if err = updatePoolCrushRule(context, clusterInfo, clusterSpec, pool); err != nil { + if err = updatePoolCrushRuleLocked(context, clusterInfo, clusterSpec, pool); err != nil { return errors.Wrapf(err, "failed to update crush rule for pool %q", pool.Name) } } @@ -510,6 +522,13 @@ func createReplicatedPoolForApp(context *clusterd.Context, clusterInfo *ClusterI } func updatePoolCrushRule(context *clusterd.Context, clusterInfo *ClusterInfo, clusterSpec *cephv1.ClusterSpec, pool cephv1.NamedPoolSpec) error { + crushRuleMutex.RLock() + defer crushRuleMutex.RUnlock() + + return updatePoolCrushRuleLocked(context, clusterInfo, clusterSpec, pool) +} + +func updatePoolCrushRuleLocked(context *clusterd.Context, clusterInfo *ClusterInfo, clusterSpec *cephv1.ClusterSpec, pool cephv1.NamedPoolSpec) error { if pool.EnableCrushUpdates == nil || !*pool.EnableCrushUpdates { logger.Debugf("Skipping crush rule update for pool %q: EnableCrushUpdates is disabled", pool.Name) return nil @@ -561,7 +580,9 @@ func updatePoolCrushRule(context *clusterd.Context, clusterInfo *ClusterInfo, cl logger.Infof("updating pool %q failure domain from %q to %q with new crush rule %q", pool.Name, currentFailureDomain, pool.FailureDomain, crushRuleName) logger.Infof("crush rule %q will no longer be used by pool %q", details.CrushRule, pool.Name) - // Create a new crush rule for the expected failure domain + // The caller holds the read lock until the pool references the newly + // created crush rule so cleanup cannot remove it in between the create and + // set operations. if err := createReplicationCrushRule(context, clusterInfo, clusterSpec, crushRuleName, pool); err != nil { return errors.Wrapf(err, "failed to create replicated crush rule %q", crushRuleName) } @@ -575,6 +596,70 @@ func updatePoolCrushRule(context *clusterd.Context, clusterInfo *ClusterInfo, cl return nil } +func CleanupUnusedCrushRules(context *clusterd.Context, clusterInfo *ClusterInfo) error { + log.NamespacedInfo(clusterInfo.Namespace, logger, "attempting to acquire crush cleanup lock") + crushRuleMutex.Lock() + defer crushRuleMutex.Unlock() + log.NamespacedInfo(clusterInfo.Namespace, logger, "acquired crush cleanup lock") + + usedRules, err := getUsedCrushRules(context, clusterInfo) + if err != nil { + return err + } + + crushMap, err := GetCrushMap(context, clusterInfo) + if err != nil { + return err + } + + var lastErr error + for _, rule := range crushMap.Rules { + if _, inUse := usedRules[rule.Name]; inUse { + logger.Debugf("crush rule %q is still in use", rule.Name) + continue + } + logger.Infof("crush rule %q is unused and will be deleted", rule.Name) + if err := deleteCrushRule(context, clusterInfo, rule.Name); err != nil { + logger.Warningf("failed to delete unused crush rule %q, continuing with remaining rules. %v", rule.Name, err) + lastErr = err + } + } + + return lastErr +} + +func getUsedCrushRules(context *clusterd.Context, clusterInfo *ClusterInfo) (map[string]struct{}, error) { + usedRules := map[string]struct{}{} + + pools, err := ListPoolSummaries(context, clusterInfo) + if err != nil { + return nil, err + } + + for _, pool := range pools { + details, err := GetPoolDetails(context, clusterInfo, pool.Name) + if err != nil { + return nil, errors.Wrapf(err, "failed to get pool %q details while checking crush rule usage", pool.Name) + } + if details.CrushRule != "" { + usedRules[details.CrushRule] = struct{}{} + } + } + + return usedRules, nil +} + +func deleteCrushRule(context *clusterd.Context, clusterInfo *ClusterInfo, crushRuleName string) error { + args := []string{"osd", "crush", "rule", "rm", crushRuleName} + output, err := NewCephCommand(context, clusterInfo, args).Run() + if err != nil { + return errors.Wrapf(err, "failed to delete crush rule %q. %s", crushRuleName, string(output)) + } + + logger.Infof("deleted unused crush rule %q", crushRuleName) + return nil +} + func extractPoolDetails(rule ruleSpec) (string, string) { // find the failure domain in the crush rule, which is the first step where the // "type" property is set diff --git a/pkg/daemon/ceph/client/pool_test.go b/pkg/daemon/ceph/client/pool_test.go index f68872190..b17ed113e 100644 --- a/pkg/daemon/ceph/client/pool_test.go +++ b/pkg/daemon/ceph/client/pool_test.go @@ -203,7 +203,13 @@ func testCreateReplicaPool(t *testing.T, failureDomain, crushRoot, deviceClass, return "", nil } } + if args[1] == "lspools" { + return `[{"poolnum":2,"poolname":"mypool"}]`, nil + } if args[1] == "crush" { + if args[2] == "dump" { + return `{"rules":[{"rule_name":"replicapool_osd"}]}`, nil + } crushRuleCreated = true assert.Equal(t, "rule", args[2]) if args[3] == "dump" { @@ -368,6 +374,47 @@ func TestUpdateFailureDomain(t *testing.T) { }) } +func TestCleanupUnusedCrushRules(t *testing.T) { + removedRules := []string{} + 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 command == "ceph" && args[1] == "lspools" { + return `[{"poolnum":1,"poolname":"mypool"},{"poolnum":2,"poolname":"otherpool"}]`, nil + } + if command == "ceph" && args[1] == "pool" && args[2] == "get" { + switch args[3] { + case "mypool": + return `{"pool":"mypool","pool_id":1,"crush_rule":"mypool_zone_ssd"}`, nil + case "otherpool": + return `{"pool":"otherpool","pool_id":2,"crush_rule":"shared_rule"}`, nil + } + } + if command == "ceph" && args[1] == "crush" && args[2] == "dump" { + return `{"rules":[ + {"rule_name":"mypool"}, + {"rule_name":"mypool_zone","steps":[{},{"type":"zone"}]}, + {"rule_name":"mypool_zone_ssd","steps":[{},{"type":"zone","item_name":"default~ssd"}]}, + {"rule_name":"shared_rule","steps":[{"item_name":"default"},{"type":"host"}]}, + {"rule_name":"custom_rule","steps":[{"item_name":"default"},{"type":"zone"}]} + ]}`, nil + } + if command == "ceph" && args[1] == "crush" && args[2] == "rule" && args[3] == "rm" { + removedRules = append(removedRules, args[4]) + return "", nil + } + + return "", errors.Errorf("unexpected ceph command %q", args) + } + + err := CleanupUnusedCrushRules(context, AdminTestClusterInfo("mycluster")) + assert.NoError(t, err) + slices.Sort(removedRules) + expected := []string{"custom_rule", "mypool", "mypool_zone"} + assert.Equal(t, expected, removedRules) +} + func TestExtractPoolDetails(t *testing.T) { t.Run("complex crush rule skipped", func(t *testing.T) { rule := ruleSpec{Steps: []stepSpec{ diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go index 613b6e8fa..0c2480ff7 100755 --- a/pkg/operator/ceph/cluster/cluster.go +++ b/pkg/operator/ceph/cluster/cluster.go @@ -55,7 +55,8 @@ import ( ) const ( - detectVersionName = "rook-ceph-detect-version" + detectVersionName = "rook-ceph-detect-version" + deleteUnusedCrushRulesSetting = "ROOK_DELETE_UNUSED_CRUSH_RULES" ) var telemetryMutex sync.Mutex @@ -571,9 +572,27 @@ func (c *cluster) postMgrStartupActions() error { if err := c.updateConfigStoreFromCRD(); err != nil { return errors.Wrap(err, "failed to set config store options") } + if !cleanupUnusedCrushRulesEnabled(c.Namespace) { + log.NamespacedInfo(c.Namespace, logger, "skipping cleanup of unused crush rules because %s is disabled", deleteUnusedCrushRulesSetting) + return nil + } + if err := client.CleanupUnusedCrushRules(c.context, c.ClusterInfo); err != nil { + log.NamespacedError(c.Namespace, logger, "failed to clean up unused crush rules. %v", err) + } return nil } +func cleanupUnusedCrushRulesEnabled(namespace string) bool { + setting := k8sutil.GetOperatorSetting(deleteUnusedCrushRulesSetting, "true") + enabled, err := strconv.ParseBool(setting) + if err != nil { + log.NamespacedWarning(namespace, logger, "invalid value %q for %s, defaulting to true. %v", setting, deleteUnusedCrushRulesSetting, err) + return true + } + + return enabled +} + func (c *cluster) configureStorageSettings() error { if !c.shouldSetClusterFullSettings() { return nil diff --git a/pkg/operator/ceph/cluster/cluster_test.go b/pkg/operator/ceph/cluster/cluster_test.go index ed6fce0ab..561000ec0 100644 --- a/pkg/operator/ceph/cluster/cluster_test.go +++ b/pkg/operator/ceph/cluster/cluster_test.go @@ -232,6 +232,66 @@ func TestConfigureMsgr2(t *testing.T) { } } +func TestPostMgrStartupActionsCleansUnusedCrushRules(t *testing.T) { + removedRules := []string{} + executor := &exectest.MockExecutor{ + MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) { + logger.Infof("Command: %s %v", command, args) + if command == "ceph" && args[1] == "lspools" { + return `[{"poolnum":1,"poolname":"mypool"}]`, nil + } + if command == "ceph" && args[1] == "pool" && args[2] == "get" { + return `{"pool":"mypool","pool_id":1,"crush_rule":"mypool_zone"}`, nil + } + if command == "ceph" && args[1] == "crush" && args[2] == "dump" { + return `{"rules":[{"rule_name":"mypool"},{"rule_name":"mypool_zone","steps":[{},{"type":"zone"}]}]}`, nil + } + if command == "ceph" && args[1] == "crush" && args[2] == "rule" && args[3] == "rm" { + removedRules = append(removedRules, args[4]) + return "", nil + } + return "", errors.Errorf("unexpected ceph command %q", args) + }, + } + + c := &cluster{ + context: &clusterd.Context{ + Clientset: testop.New(t, 1), + Executor: executor, + }, + ClusterInfo: cephclient.AdminTestClusterInfo("rook-ceph"), + Spec: &cephv1.ClusterSpec{}, + } + + err := c.postMgrStartupActions() + assert.NoError(t, err) + assert.Equal(t, []string{"mypool"}, removedRules) +} + +func TestPostMgrStartupActionsSkipsUnusedCrushRuleCleanupWhenDisabled(t *testing.T) { + t.Setenv(deleteUnusedCrushRulesSetting, "false") + + executor := &exectest.MockExecutor{ + MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) { + t.Fatalf("unexpected ceph command %s %v", command, args) + return "", nil + }, + } + + c := &cluster{ + context: &clusterd.Context{ + Clientset: testop.New(t, 1), + Executor: executor, + }, + ClusterInfo: cephclient.AdminTestClusterInfo("rook-ceph"), + Spec: &cephv1.ClusterSpec{}, + Namespace: "rook-ceph", + } + + err := c.postMgrStartupActions() + assert.NoError(t, err) +} + func TestTelemetry(t *testing.T) { var expectedSettings map[string]string clientset := testop.New(t, 3) diff --git a/pkg/operator/ceph/file/filesystem_test.go b/pkg/operator/ceph/file/filesystem_test.go index 83535b70b..dc7f21776 100644 --- a/pkg/operator/ceph/file/filesystem_test.go +++ b/pkg/operator/ceph/file/filesystem_test.go @@ -221,6 +221,10 @@ func fsExecutor(t *testing.T, fsName, configDir string, multiFS bool, createData return "", nil } else if slices.Contains(args, "flag") && slices.Contains(args, "enable_multiple") { return "", nil + } else if len(args) >= 3 && args[0] == "osd" && args[1] == "crush" && args[2] == "dump" { + return `{"rules":[]}`, nil + } else if len(args) >= 4 && args[0] == "osd" && args[1] == "crush" && args[2] == "rule" && args[3] == "rm" { + return "", nil } else if reflect.DeepEqual(args[0:5], []string{"osd", "crush", "rule", "create-replicated", fsName + "-data1"}) { return "", nil } else if reflect.DeepEqual(args[0:4], []string{"osd", "pool", "create", fsName + "-data1"}) { @@ -306,6 +310,10 @@ func fsExecutor(t *testing.T, fsName, configDir string, multiFS bool, createData return "", nil } else if slices.Contains(args, "config") && slices.Contains(args, "get") { return "{}", nil + } else if len(args) >= 3 && args[0] == "osd" && args[1] == "crush" && args[2] == "dump" { + return `{"rules":[]}`, nil + } else if len(args) >= 4 && args[0] == "osd" && args[1] == "crush" && args[2] == "rule" && args[3] == "rm" { + return "", nil } else if reflect.DeepEqual(args[0:5], []string{"osd", "crush", "rule", "create-replicated", fsName + "-data1"}) { return "", nil } else if reflect.DeepEqual(args[0:3], []string{"osd", "pool", "application"}) { diff --git a/pkg/operator/ceph/pool/controller_test.go b/pkg/operator/ceph/pool/controller_test.go index 6540fcde5..03db6e6bf 100644 --- a/pkg/operator/ceph/pool/controller_test.go +++ b/pkg/operator/ceph/pool/controller_test.go @@ -159,6 +159,8 @@ func TestCephPoolName(t *testing.T) { func TestDeletePool(t *testing.T) { failOnDelete := false + deletedPools := map[string]bool{} + deletedCrushRules := map[string]bool{} clusterInfo := cephclient.AdminTestClusterInfo("mycluster") executor := &exectest.MockExecutor{ MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) { @@ -166,10 +168,17 @@ func TestDeletePool(t *testing.T) { p := "{\"images\":{\"count\":1,\"provisioned_bytes\":1024,\"snap_count\":0},\"trash\":{\"count\":1,\"provisioned_bytes\":2048,\"snap_count\":0}}" logger.Infof("Command: %s %v", command, args) if command == "ceph" && args[1] == "lspools" { + if deletedPools["mypool"] { + return `[]`, nil + } return `[{"poolnum":1,"poolname":"mypool"}]`, nil } else if command == "ceph" && args[1] == "pool" && args[2] == "get" { - return `{"pool": "mypool","pool_id": 1,"size":1}`, nil + return `{"pool": "mypool","pool_id": 1,"size":1,"crush_rule":"mypool_zone"}`, nil } else if command == "ceph" && args[1] == "pool" && args[2] == "delete" { + deletedPools[args[3]] = true + return "", nil + } else if command == "ceph" && args[1] == "crush" && args[2] == "rule" && args[3] == "rm" { + deletedCrushRules[args[4]] = true return "", nil } else if args[0] == "pool" { if args[1] == "stats" { @@ -189,29 +198,39 @@ func TestDeletePool(t *testing.T) { p := &cephv1.NamedPoolSpec{Name: "mypool"} err := deletePool(context, clusterInfo, p) assert.NoError(t, err) + assert.True(t, deletedPools["mypool"]) + assert.True(t, deletedCrushRules["mypool_zone"]) // succeed even if the pool doesn't exist p = &cephv1.NamedPoolSpec{Name: "otherpool"} err = deletePool(context, clusterInfo, p) assert.Nil(t, err) + assert.False(t, deletedPools["otherpool"]) // fail if images/snapshosts exist in the pool failOnDelete = true + deletedPools["mypool"] = false p = &cephv1.NamedPoolSpec{Name: "mypool"} err = deletePool(context, clusterInfo, p) assert.Error(t, err) + assert.False(t, deletedPools["mypool"]) // delete an erasure coded pool should also delete the EC profile ecProfileDeleted := false failOnDelete = false + ecDeletedPools := map[string]bool{} ecExecutor := &exectest.MockExecutor{ MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) { emptyPool := "{\"images\":{\"count\":0,\"provisioned_bytes\":0,\"snap_count\":0},\"trash\":{\"count\":1,\"provisioned_bytes\":2048,\"snap_count\":0}}" if command == "ceph" && args[1] == "lspools" { + if ecDeletedPools["ecpool"] { + return `[]`, nil + } return `[{"poolnum":1,"poolname":"ecpool"}]`, nil } else if command == "ceph" && args[1] == "pool" && args[2] == "get" { return `{"pool": "ecpool","pool_id": 1,"size":1}`, nil } else if command == "ceph" && args[1] == "pool" && args[2] == "delete" { + ecDeletedPools[args[3]] = true return "", nil } else if command == "ceph" && args[1] == "erasure-code-profile" && args[2] == "rm" { assert.Equal(t, "ecpool_ecprofile", args[3]) @@ -239,6 +258,7 @@ func TestDeletePool(t *testing.T) { err = deletePool(ecContext, clusterInfo, ecPool) assert.NoError(t, err) assert.True(t, ecProfileDeleted) + assert.True(t, ecDeletedPools["ecpool"]) } // TestCephBlockPoolController runs ReconcileCephBlockPool.Reconcile() against a