Merge pull request #13197 from rook/mergify/bp/release-1.12/pr-13069

pool: allow updating deviceClass on existing pool (backport #13069)

pkg/daemon/ceph/client/pool.go

@@ -471,21 +471,21 @@ func createReplicatedPoolForApp(context *clusterd.Context, clusterInfo *ClusterI
 
 	logger.Infof("reconciling replicated pool %s succeeded", pool.Name)
 
-	if checkFailureDomain {
-		if err = ensureFailureDomain(context, clusterInfo, clusterSpec, pool); err != nil {
+	if checkFailureDomain || pool.PoolSpec.DeviceClass != "" {
+		if err = updatePoolCrushRule(context, clusterInfo, clusterSpec, pool); err != nil {
 			return nil
 		}
 	}
 	return nil
 }
 
-func ensureFailureDomain(context *clusterd.Context, clusterInfo *ClusterInfo, clusterSpec *cephv1.ClusterSpec, pool cephv1.NamedPoolSpec) error {
-	if pool.FailureDomain == "" {
-		logger.Debugf("skipping check for failure domain on pool %q as it is not specified", pool.Name)
+func updatePoolCrushRule(context *clusterd.Context, clusterInfo *ClusterInfo, clusterSpec *cephv1.ClusterSpec, pool cephv1.NamedPoolSpec) error {
+	if pool.FailureDomain == "" && pool.DeviceClass == "" {
+		logger.Debugf("skipping check for failure domain and deviceClass on pool %q as it is not specified", pool.Name)
 		return nil
 	}
 
-	logger.Debugf("checking that pool %q has the failure domain %q", pool.Name, pool.FailureDomain)
+	logger.Debugf("checking that pool %q has the failure domain %q and deviceClass %q", pool.Name, pool.FailureDomain, pool.DeviceClass)
 	details, err := GetPoolDetails(context, clusterInfo, pool.Name)
 	if err != nil {
 		return errors.Wrapf(err, "failed to get pool %q details", pool.Name)
@@ -496,17 +496,24 @@ func ensureFailureDomain(context *clusterd.Context, clusterInfo *ClusterInfo, cl
 	if err != nil {
 		return errors.Wrapf(err, "failed to get crush rule %q", details.CrushRule)
 	}
-	currentFailureDomain := extractFailureDomain(rule)
-	if currentFailureDomain == pool.FailureDomain {
-		logger.Debugf("pool %q has the expected failure domain %q", pool.Name, pool.FailureDomain)
+	currentFailureDomain, currentDeviceClass := extractPoolDetails(rule)
+	if currentFailureDomain == pool.FailureDomain && currentDeviceClass == pool.DeviceClass {
+		logger.Infof("pool %q has the expected failure domain %q and deviceClass %q", pool.Name, pool.FailureDomain, pool.DeviceClass)
 		return nil
 	}
-	if currentFailureDomain == "" {
-		logger.Warningf("failure domain not found for crush rule %q, proceeding to create a new crush rule", details.CrushRule)
+
+	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)
+	}
 	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)
 
@@ -524,12 +531,17 @@ func ensureFailureDomain(context *clusterd.Context, clusterInfo *ClusterInfo, cl
 	return nil
 }
 
-func extractFailureDomain(rule ruleSpec) string {
+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
+
+	var failureDomain, deviceClass string
 	for i, step := range rule.Steps {
 		if step.Type != "" {
-			return step.Type
+			failureDomain = step.Type
+		}
+		if step.ItemName != "" {
+			deviceClass = step.ItemName
 		}
 		// We expect the rule to be found by the second step, or else it is a more
 		// complex rule that would not be supported for updating the failure domain
@@ -537,7 +549,7 @@ func extractFailureDomain(rule ruleSpec) string {
 			break
 		}
 	}
-	return ""
+	return failureDomain, deviceClass
 }
 
 func setCrushRule(context *clusterd.Context, clusterInfo *ClusterInfo, poolName, crushRule string) error {

pkg/daemon/ceph/client/pool_test.go

@@ -281,7 +281,7 @@ func TestUpdateFailureDomain(t *testing.T) {
 			},
 		}
 		clusterSpec := &cephv1.ClusterSpec{Storage: cephv1.StorageScopeSpec{}}
-		err := ensureFailureDomain(context, AdminTestClusterInfo("mycluster"), clusterSpec, p)
+		err := updatePoolCrushRule(context, AdminTestClusterInfo("mycluster"), clusterSpec, p)
 		assert.NoError(t, err)
 		assert.Equal(t, "", newCrushRule)
 	})
@@ -295,7 +295,7 @@ func TestUpdateFailureDomain(t *testing.T) {
 			},
 		}
 		clusterSpec := &cephv1.ClusterSpec{Storage: cephv1.StorageScopeSpec{}}
-		err := ensureFailureDomain(context, AdminTestClusterInfo("mycluster"), clusterSpec, p)
+		err := updatePoolCrushRule(context, AdminTestClusterInfo("mycluster"), clusterSpec, p)
 		assert.NoError(t, err)
 		assert.Equal(t, "", newCrushRule)
 	})
@@ -309,29 +309,30 @@ func TestUpdateFailureDomain(t *testing.T) {
 			},
 		}
 		clusterSpec := &cephv1.ClusterSpec{Storage: cephv1.StorageScopeSpec{}}
-		err := ensureFailureDomain(context, AdminTestClusterInfo("mycluster"), clusterSpec, p)
+		err := updatePoolCrushRule(context, AdminTestClusterInfo("mycluster"), clusterSpec, p)
 		assert.NoError(t, err)
 		assert.Equal(t, "mypool_zone", newCrushRule)
 	})
 }
 
-func TestExtractFailureDomain(t *testing.T) {
+func TestExtractPoolDetails(t *testing.T) {
 	t.Run("complex crush rule skipped", func(t *testing.T) {
 		rule := ruleSpec{Steps: []stepSpec{
 			{Type: ""},
 			{Type: ""},
 			{Type: "zone"},
 		}}
-		failureDomain := extractFailureDomain(rule)
+		failureDomain, _ := extractPoolDetails(rule)
 		assert.Equal(t, "", failureDomain)
 	})
 	t.Run("valid crush rule", func(t *testing.T) {
 		rule := ruleSpec{Steps: []stepSpec{
 			{Type: ""},
-			{Type: "zone"},
+			{Type: "zone", ItemName: "ssd"},
 		}}
-		failureDomain := extractFailureDomain(rule)
+		failureDomain, deviceClass := extractPoolDetails(rule)
 		assert.Equal(t, "zone", failureDomain)
+		assert.Equal(t, "ssd", deviceClass)
 	})
 }