Merge pull request #13791 from rook/mergify/bp/release-1.13/pr-13772

pool: Skip crush rule update when not needed (backport #13772)

pkg/daemon/ceph/client/pool.go

@@ -502,18 +502,23 @@ func updatePoolCrushRule(context *clusterd.Context, clusterInfo *ClusterInfo, cl
 		return nil
 	}
 
-	if currentFailureDomain != pool.FailureDomain {
-		logger.Infof("creating a new crush rule for changed failure domain on crush rule %q", details.CrushRule)
-	}
-	if currentDeviceClass != pool.DeviceClass {
-		logger.Infof("creating a new crush rule for changed deviceClass on crush rule %q", details.CrushRule)
-	}
-
 	// Use a crush rule name that is unique to the desired failure domain
 	crushRuleName := fmt.Sprintf("%s_%s", pool.Name, pool.FailureDomain)
 	if pool.DeviceClass != "" {
 		crushRuleName = fmt.Sprintf("%s_%s_%s", pool.Name, pool.FailureDomain, pool.DeviceClass)
 	}
+	if crushRuleName == details.CrushRule {
+		logger.Debugf("crush rule already set to %q for pool %q (failureDomain=%s, deviceClass=%s)", crushRuleName, pool.Name, currentFailureDomain, currentDeviceClass)
+		return nil
+	}
+
+	if currentFailureDomain != pool.FailureDomain {
+		logger.Infof("creating a new crush rule for changed failure domain (%q-->%q) on crush rule %q", currentFailureDomain, pool.FailureDomain, details.CrushRule)
+	}
+	if currentDeviceClass != pool.DeviceClass {
+		logger.Infof("creating a new crush rule for changed deviceClass (%q-->%q) on crush rule %q", currentDeviceClass, pool.DeviceClass, details.CrushRule)
+	}
+
 	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)
 

pkg/daemon/ceph/client/pool_test.go

@@ -249,6 +249,8 @@ func testCreateReplicaPool(t *testing.T, failureDomain, crushRoot, deviceClass,
 func TestUpdateFailureDomain(t *testing.T) {
 	var newCrushRule string
 	currentFailureDomain := "rack"
+	currentDeviceClass := "default"
+	testCrushRuleName := "test_rule"
 	executor := &exectest.MockExecutor{}
 	context := &clusterd.Context{Executor: executor}
 	executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
@@ -256,7 +258,7 @@ func TestUpdateFailureDomain(t *testing.T) {
 		if args[1] == "pool" {
 			if args[2] == "get" {
 				assert.Equal(t, "mypool", args[3])
-				return `{"crush_rule": "test_rule"}`, nil
+				return fmt.Sprintf(`{"crush_rule": "%s"}`, testCrushRuleName), nil
 			}
 			if args[2] == "set" {
 				assert.Equal(t, "mypool", args[3])
@@ -267,7 +269,7 @@ func TestUpdateFailureDomain(t *testing.T) {
 		}
 		if args[1] == "crush" {
 			if args[2] == "rule" && args[3] == "dump" {
-				return fmt.Sprintf(`{"steps": [{"foo":"bar"},{"type":"%s"}]}`, currentFailureDomain), nil
+				return fmt.Sprintf(`{"steps": [{"item_name":"%s"},{"type":"%s"}]}`, currentDeviceClass, currentFailureDomain), nil
 			}
 			newCrushRule = "foo"
 			return "", nil
@@ -296,6 +298,7 @@ func TestUpdateFailureDomain(t *testing.T) {
 				Replicated:    cephv1.ReplicatedSpec{Size: 3},
 			},
 		}
+		testCrushRuleName = "mypool_rack"
 		clusterSpec := &cephv1.ClusterSpec{Storage: cephv1.StorageScopeSpec{}}
 		err := updatePoolCrushRule(context, AdminTestClusterInfo("mycluster"), clusterSpec, p)
 		assert.NoError(t, err)