From 0300111f73c68e9afb075a239a972016884ba87c Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?S=C3=A9bastien=20Han?= <seb@redhat.com>
Date: Mon, 22 Feb 2021 16:55:30 +0100
Subject: [PATCH] ceph: fail if subfailure domains are identical
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Both cannot be identical otherwise this will create placement issues.

Signed-off-by: Sébastien Han <seb@redhat.com>
---
 pkg/daemon/ceph/client/pool.go          |  7 ++++++-
 pkg/daemon/ceph/client/pool_test.go     |  2 +-
 pkg/operator/ceph/pool/validate.go      |  6 ++++++
 pkg/operator/ceph/pool/validate_test.go | 10 ++++++++++
 4 files changed, 23 insertions(+), 2 deletions(-)

diff --git a/pkg/daemon/ceph/client/pool.go b/pkg/daemon/ceph/client/pool.go
index 9767dcb3e6d0..a53a30a87ccd 100644
--- a/pkg/daemon/ceph/client/pool.go
+++ b/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)
diff --git a/pkg/daemon/ceph/client/pool_test.go b/pkg/daemon/ceph/client/pool_test.go
index 83d54d9c3a84..275d9388a984 100644
--- a/pkg/daemon/ceph/client/pool_test.go
+++ b/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"
diff --git a/pkg/operator/ceph/pool/validate.go b/pkg/operator/ceph/pool/validate.go
index ab9f7306712d..24e38731aaf5 100644
--- a/pkg/operator/ceph/pool/validate.go
+++ b/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() {
diff --git a/pkg/operator/ceph/pool/validate_test.go b/pkg/operator/ceph/pool/validate_test.go
index 0b6b8f08e367..47f88e3fda31 100644
--- a/pkg/operator/ceph/pool/validate_test.go
+++ b/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) {