Merge pull request #7284 from leseb/pool-replica-failure

ceph: fail if subfailure domains are identical

pkg/daemon/ceph/client/pool.go

@@ -337,7 +337,7 @@ func CreateReplicatedPoolForApp(context *clusterd.Context, clusterInfo *ClusterI
 		// so there is no need to create a new crush rule for the pools here.
 		crushRuleName = defaultStretchCrushRuleName
 	} else {
-		if pool.Replicated.ReplicasPerFailureDomain != 0 {
+		if pool.Replicated.ReplicasPerFailureDomain > 1 {
 			// Create a two-step CRUSH rule for pools other than stretch clusters
 			err := createTwoStepCrushRule(context, clusterInfo, clusterSpec, crushRuleName, pool)
 			if err != nil {
@@ -381,6 +381,11 @@ func createTwoStepCrushRule(context *clusterd.Context, clusterInfo *ClusterInfo,
 	if pool.Replicated.SubFailureDomain == "" {
 		pool.Replicated.SubFailureDomain = cephv1.DefaultFailureDomain
 	}
+
+	if pool.FailureDomain == pool.Replicated.SubFailureDomain {
+		return errors.Errorf("failure and subfailure domains cannot be identical, current is %q", pool.FailureDomain)
+	}
+
 	// set the crush root to the default if not already specified
 	if pool.CrushRoot == "" {
 		pool.CrushRoot = GetCrushRootFromSpec(clusterSpec)

pkg/daemon/ceph/client/pool_test.go

@@ -310,7 +310,7 @@ func testCreateStretchCrushRule(t *testing.T, alreadyExists bool) {
 	}
 	clusterInfo := AdminClusterInfo("mycluster")
 	clusterSpec := &cephv1.ClusterSpec{}
-	poolSpec := cephv1.PoolSpec{}
+	poolSpec := cephv1.PoolSpec{FailureDomain: "rack"}
 	ruleName := "testrule"
 	if alreadyExists {
 		ruleName = "replicated_ruleset"

pkg/operator/ceph/pool/validate.go

@@ -46,6 +46,12 @@ func ValidatePoolSpec(context *clusterd.Context, clusterInfo *client.ClusterInfo
 		return errors.New("both replication and erasure code settings cannot be specified")
 	}
 
+	if p.FailureDomain != "" && p.Replicated.SubFailureDomain != "" {
+		if p.FailureDomain == p.Replicated.SubFailureDomain {
+			return errors.New("failure and subfailure domain cannot be identical")
+		}
+	}
+
 	// validate pools for stretch clusters
 	if clusterSpec.IsStretchCluster() {
 		if p.IsReplicated() {

pkg/operator/ceph/pool/validate_test.go

@@ -172,6 +172,16 @@ func TestValidatePool(t *testing.T) {
 		assert.EqualError(t, err, "mirroring must be enabled to configure snapshot scheduling")
 	}
 
+	// Failure and subfailure domains
+	{
+		p := cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
+		p.Spec.FailureDomain = "host"
+		p.Spec.Replicated.SubFailureDomain = "host"
+		err = ValidatePool(context, clusterInfo, clusterSpec, &p)
+		assert.Error(t, err)
+		assert.EqualError(t, err, "failure and subfailure domain cannot be identical")
+	}
+
 }
 
 func TestValidateCrushProperties(t *testing.T) {