Merge pull request #6379 from leseb/fix-5591

ceph: add support for stretched cluster crush rule

Documentation/ceph-pool-crd.md

@@ -106,6 +106,44 @@ external-cluster-console# rbd mirror pool peer bootstrap import <token file path
 
 See the official rbd mirror documentation on [how to add a bootstrap peer](https://docs.ceph.com/docs/master/rbd/rbd-mirroring/#bootstrap-peers).
 
+### Data spread across subdomains
+
+Imagine the following topology with datacenters containing racks and then hosts:
+
+```text
+.
+├── datacenter-1
+│   ├── rack-1
+│   │   ├── host-1
+│   │   ├── host-2
+│   └── rack-2
+│       ├── host-3
+│       ├── host-4
+└── datacenter-2
+    ├── rack-3
+    │   ├── host-5
+    │   ├── host-6
+    └── rack-4
+        ├── host-7
+        └── host-8
+```
+
+As an administrator I would like to place 4 copies across both datacenter where each copy inside a datacenter is across a rack.
+This can be achieved by the following:
+
+```yaml
+apiVersion: ceph.rook.io/v1
+kind: CephBlockPool
+metadata:
+  name: replicapool
+  namespace: rook-ceph
+spec:
+  replicated:
+    size: 4
+    replicasPerFailureDomain: 2
+    subFailureDomain: rack
+```
+
 ## Pool Settings
 
 ### Metadata
@@ -118,6 +156,9 @@ See the official rbd mirror documentation on [how to add a bootstrap peer](https
 * `replicated`: Settings for a replicated pool. If specified, `erasureCoded` settings must not be specified.
   * `size`: The desired number of copies to make of the data in the pool.
   * `requireSafeReplicaSize`: set to false if you want to create a pool with size 1, setting pool size 1 could lead to data loss without recovery. Make sure you are *ABSOLUTELY CERTAIN* that is what you want.
+  * `replicasPerFailureDomain`: Sets up the number of replicas to place in a given failure domain. For instance, if the failure domain is a datacenter (cluster is
+stretched) then you will have 2 replicas per datacenter where each replica ends up on a different host. This gives you a total of 4 replicas and for this, the `size` must be set to 4. The default is 1.
+  * `subFailureDomain`: Name of the CRUSH bucket representing a sub-failure domain. In a stretched configuration this option represent the "last" bucket where replicas will end up being written. Imagine the cluster is stretched across two datacenters, you can then have 2 copies per datacenter and each copy on a different CRUSH bucket. The default is "host".
 * `erasureCoded`: Settings for an erasure-coded pool. If specified, `replicated` settings must not be specified. See below for more details on [erasure coding](#erasure-coding).
   * `dataChunks`: Number of chunks to divide the original object into
   * `codingChunks`: Number of coding chunks to generate

PendingReleaseNotes.md

@@ -17,3 +17,4 @@ v1.5...
 ### Ceph
 
 * Ceph Block Pool: add mirroring support
+* Ceph Block Pool: add `replicasPerFailureDomain` to set the number of replica in a failure domain ([#5591](https://github.com/rook/rook/issues/5591))

cluster/charts/rook-ceph/templates/resources.yaml

@@ -250,6 +250,10 @@ spec:
                       type: integer
                     requireSafeReplicaSize:
                       type: boolean
+                    replicasPerFailureDomain:
+                      type: integer
+                    subFailureDomain:
+                      type: string
                 erasureCoded:
                   properties:
                     dataChunks:
@@ -286,6 +290,10 @@ spec:
                         type: integer
                       requireSafeReplicaSize:
                         type: boolean
+                    replicasPerFailureDomain:
+                      type: integer
+                    subFailureDomain:
+                      type: string
                   erasureCoded:
                     properties:
                       dataChunks:
@@ -405,6 +413,10 @@ spec:
                       type: integer
                     requireSafeReplicaSize:
                       type: boolean
+                    replicasPerFailureDomain:
+                      type: integer
+                    subFailureDomain:
+                      type: string
                 erasureCoded:
                   properties:
                     dataChunks:
@@ -433,6 +445,10 @@ spec:
                       type: integer
                     requireSafeReplicaSize:
                       type: boolean
+                    replicasPerFailureDomain:
+                      type: integer
+                    subFailureDomain:
+                      type: string
                 erasureCoded:
                   properties:
                     dataChunks:
@@ -564,6 +580,10 @@ spec:
                   type: number
                 requireSafeReplicaSize:
                   type: boolean
+                replicasPerFailureDomain:
+                  type: integer
+                subFailureDomain:
+                  type: string
             erasureCoded:
               properties:
                 dataChunks:

cluster/examples/kubernetes/ceph/common.yaml

@@ -334,6 +334,10 @@ spec:
                       type: integer
                     requireSafeReplicaSize:
                       type: boolean
+                    replicasPerFailureDomain:
+                      type: integer
+                    subFailureDomain:
+                      type: string
                 erasureCoded:
                   properties:
                     dataChunks:
@@ -368,6 +372,10 @@ spec:
                         type: integer
                       requireSafeReplicaSize:
                         type: boolean
+                      replicasPerFailureDomain:
+                        type: integer
+                      subFailureDomain:
+                        type: string
                   erasureCoded:
                     properties:
                       dataChunks:
@@ -495,6 +503,10 @@ spec:
                       type: integer
                     requireSafeReplicaSize:
                       type: boolean
+                    replicasPerFailureDomain:
+                      type: integer
+                    subFailureDomain:
+                      type: string
                 erasureCoded:
                   properties:
                     dataChunks:
@@ -523,6 +535,10 @@ spec:
                       type: integer
                     requireSafeReplicaSize:
                       type: boolean
+                    replicasPerFailureDomain:
+                      type: integer
+                    subFailureDomain:
+                      type: string
                 erasureCoded:
                   properties:
                     dataChunks:
@@ -661,6 +677,10 @@ spec:
                   type: number
                 requireSafeReplicaSize:
                   type: boolean
+                replicasPerFailureDomain:
+                  type: integer
+                subFailureDomain:
+                  type: string
             erasureCoded:
               properties:
                 dataChunks:

cluster/examples/kubernetes/ceph/pool.yaml

@@ -18,6 +18,10 @@ spec:
     # Disallow setting pool with replica 1, this could lead to data loss without recovery.
     # Make sure you're *ABSOLUTELY CERTAIN* that is what you want
     requireSafeReplicaSize: true
+    # The number for replicas per failure domain, the value must be a divisor of the replica count. If specified, the most common value is 2 for stretch clusters, where the replica count would be 4.
+    # replicasPerFailureDomain: 2
+    # The name of the failure domain to place further down replicas
+    # subFailureDomain: host
   # Ceph CRUSH root location of the rule
   # For reference: https://docs.ceph.com/docs/nautilus/rados/operations/crush-map/#types-and-buckets
   #crushRoot: my-root

pkg/apis/ceph.rook.io/v1/types.go

@@ -242,8 +242,6 @@ const (
 	ConditionFailure     ConditionType = "Failure"
 	ConditionUpgrading   ConditionType = "Upgrading"
 	ConditionDeleting    ConditionType = "Deleting"
-	// DefaultFailureDomain for PoolSpec
-	DefaultFailureDomain = "host"
 )
 
 type ClusterState string
@@ -302,6 +300,13 @@ type CephBlockPoolList struct {
 	Items           []CephBlockPool `json:"items"`
 }
 
+const (
+	// DefaultFailureDomain for PoolSpec
+	DefaultFailureDomain = "host"
+	// DefaultCRUSHRoot is the default name of the CRUSH root bucket
+	DefaultCRUSHRoot = "default"
+)
+
 // PoolSpec represents the spec of ceph pool
 type PoolSpec struct {
 	// The failure domain: osd/host/(region or zone if available) - technically also any type in the crush map
@@ -379,6 +384,12 @@ type ReplicatedSpec struct {
 
 	// RequireSafeReplicaSize if false allows you to set replica 1
 	RequireSafeReplicaSize bool `json:"requireSafeReplicaSize"`
+
+	// ReplicasPerFailureDomain the number of replica in the specified failure domain
+	ReplicasPerFailureDomain uint `json:"replicasPerFailureDomain,omitempty"`
+
+	// SubFailureDomain the name of the sub-failure domain
+	SubFailureDomain string `json:"subFailureDomain,omitempty"`
 }
 
 // MirroredSpec represents the setting for a mirrored pool

pkg/daemon/ceph/client/crush.go

@@ -19,6 +19,7 @@ package client
 import (
 	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"strconv"
 	"strings"
 
@@ -51,26 +52,30 @@ type CrushMap struct {
 			Pos    int `json:"pos"`
 		} `json:"items"`
 	} `json:"buckets"`
-	Rules []struct {
-		ID      int    `json:"rule_id"`
-		Name    string `json:"rule_name"`
-		Ruleset int    `json:"ruleset"`
-		Type    int    `json:"type"`
-		MinSize int    `json:"min_size"`
-		MaxSize int    `json:"max_size"`
-		Steps   []struct {
-			Operation string `json:"op"`
-			Number    int    `json:"num"`
-			Item      int    `json:"item"`
-			ItemName  string `json:"item_name"`
-			Type      string `json:"type"`
-		} `json:"steps"`
-	} `json:"rules"`
+	Rules    []ruleSpec `json:"rules"`
 	Tunables struct {
 		// Add if necessary
 	} `json:"tunables"`
 }
 
+type ruleSpec struct {
+	ID      int        `json:"rule_id"`
+	Name    string     `json:"rule_name"`
+	Ruleset int        `json:"ruleset"`
+	Type    int        `json:"type"`
+	MinSize int        `json:"min_size"`
+	MaxSize int        `json:"max_size"`
+	Steps   []stepSpec `json:"steps"`
+}
+
+type stepSpec struct {
+	Operation string `json:"op"`
+	Number    uint   `json:"num"`
+	Item      int    `json:"item"`
+	ItemName  string `json:"item_name"`
+	Type      string `json:"type"`
+}
+
 // CrushFindResult is go representation of the Ceph osd find command output
 type CrushFindResult struct {
 	ID       int               `json:"osd"`
@@ -96,6 +101,25 @@ func GetCrushMap(context *clusterd.Context, clusterInfo *ClusterInfo) (CrushMap,
 	return c, nil
 }
 
+// GetCompiledCrushMap fetches the Ceph compiled version of the CRUSH map
+func GetCompiledCrushMap(context *clusterd.Context, clusterInfo *ClusterInfo) (string, error) {
+	compiledCrushMapFile, err := ioutil.TempFile("", "")
+	if err != nil {
+		return "", errors.Wrap(err, "failed to generate temporarily file")
+	}
+
+	args := []string{"osd", "getcrushmap", "--out-file", compiledCrushMapFile.Name()}
+	exec := NewCephCommand(context, clusterInfo, args)
+	exec.OutputFile = false
+	exec.JsonOutput = false
+	buf, err := exec.Run()
+	if err != nil {
+		return "", errors.Wrapf(err, "failed to get compiled crush map. %s", string(buf))
+	}
+
+	return compiledCrushMapFile.Name(), nil
+}
+
 // FindOSDInCrushMap finds an OSD in the CRUSH map
 func FindOSDInCrushMap(context *clusterd.Context, clusterInfo *ClusterInfo, osdID int) (*CrushFindResult, error) {
 	args := []string{"osd", "find", strconv.Itoa(osdID)}
@@ -167,3 +191,59 @@ func GetOSDOnHost(context *clusterd.Context, clusterInfo *ClusterInfo, node stri
 
 	return string(buf), nil
 }
+
+func compileCRUSHMap(context *clusterd.Context, crushMapPath string) error {
+	mapFile := buildCompileCRUSHFileName(crushMapPath)
+	args := []string{"--compile", crushMapPath, "--outfn", mapFile}
+	output, err := context.Executor.ExecuteCommandWithOutput("crushtool", args...)
+	if err != nil {
+		return errors.Wrapf(err, "failed to compile crush map %q. %s", mapFile, output)
+	}
+
+	return nil
+}
+
+func decompileCRUSHMap(context *clusterd.Context, crushMapPath string) error {
+	mapFile := buildDecompileCRUSHFileName(crushMapPath)
+	args := []string{"--decompile", crushMapPath, "--outfn", mapFile}
+	output, err := context.Executor.ExecuteCommandWithOutput("crushtool", args...)
+	if err != nil {
+		return errors.Wrapf(err, "failed to decompile crush map %q. %s", mapFile, output)
+	}
+
+	return nil
+}
+
+func injectCRUSHMap(context *clusterd.Context, clusterInfo *ClusterInfo, crushMapPath string) error {
+	args := []string{"osd", "setcrushmap", "--in-file", crushMapPath}
+	exec := NewCephCommand(context, clusterInfo, args)
+	exec.OutputFile = false
+	exec.JsonOutput = false
+	buf, err := exec.Run()
+	if err != nil {
+		return errors.Wrapf(err, "failed to inject crush map %q. %s", crushMapPath, string(buf))
+	}
+
+	return nil
+}
+
+func setCRUSHMap(context *clusterd.Context, clusterInfo *ClusterInfo, crushMapPath string) error {
+	args := []string{"osd", "crush", "set", crushMapPath}
+	exec := NewCephCommand(context, clusterInfo, args)
+	exec.OutputFile = false
+	exec.JsonOutput = false
+	buf, err := exec.Run()
+	if err != nil {
+		return errors.Wrapf(err, "failed to set crush map %q. %s", crushMapPath, string(buf))
+	}
+
+	return nil
+}
+
+func buildDecompileCRUSHFileName(crushMapPath string) string {
+	return fmt.Sprintf("%s.decompiled", crushMapPath)
+}
+
+func buildCompileCRUSHFileName(crushMapPath string) string {
+	return fmt.Sprintf("%s.compiled", crushMapPath)
+}

pkg/daemon/ceph/client/crush_rule.go

@@ -0,0 +1,144 @@
+/*
+Copyright 2020 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package client
+
+import (
+	"fmt"
+
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+)
+
+const (
+	crushReplicatedType        = 1
+	ruleMinSizeDefault         = 1
+	ruleMaxSizeDefault         = 10
+	stretchedCRUSHRuleTemplate = `
+rule %s {
+        id %d
+        type replicated
+        min_size %d
+        max_size %d
+        step take %s
+        step choose firstn 0 type %s
+        step chooseleaf firstn 2 type %s
+        step emit
+}
+`
+)
+
+var (
+	stepEmit = &stepSpec{Operation: "emit"}
+)
+
+func buildStretchClusterPlainCrushRule(crushMap CrushMap, ruleName string, pool cephv1.PoolSpec) string {
+	return fmt.Sprintf(
+		stretchedCRUSHRuleTemplate,
+		ruleName,
+		generateRuleID(crushMap.Rules),
+		ruleMinSizeDefault,
+		ruleMaxSizeDefault,
+		pool.CrushRoot,
+		pool.FailureDomain,
+		pool.Replicated.SubFailureDomain,
+	)
+}
+
+func buildStretchClusterCrushRule(crushMap CrushMap, ruleName string, pool cephv1.PoolSpec) *ruleSpec {
+	/*
+		The complete CRUSH rule looks like this:
+
+		   rule two_rep_per_dc {
+		           id 1
+		           type replicated
+		           min_size 1
+		           max_size 10
+		           step take root
+		           step choose firstn 0 type datacenter
+		           step chooseleaf firstn 2 type host
+		           step emit
+		   }
+
+	*/
+
+	ruleID := generateRuleID(crushMap.Rules)
+	return &ruleSpec{
+		ID:      ruleID,
+		Name:    ruleName,
+		Ruleset: ruleID,
+		Type:    crushReplicatedType,
+		MinSize: ruleMinSizeDefault,
+		MaxSize: ruleMaxSizeDefault,
+		Steps:   buildStretchClusterCrushSteps(pool),
+	}
+}
+
+func buildStretchClusterCrushSteps(pool cephv1.PoolSpec) []stepSpec {
+	// Create CRUSH rule steps
+	steps := []stepSpec{}
+
+	// Create the default step, which is essentially the entrypoint, the "root" of all requests
+	stepTakeDefault := &stepSpec{
+		Operation: "take",
+		Item:      -1,
+		ItemName:  pool.CrushRoot,
+	}
+	steps = append(steps, *stepTakeDefault)
+
+	// Steps two
+	stepTakeFailureDomain := &stepSpec{
+		Operation: "chooseleaf_firstn",
+		Number:    0,
+		Type:      pool.FailureDomain,
+	}
+	steps = append(steps, *stepTakeFailureDomain)
+
+	// Step three
+	stepTakeSubFailureDomain := &stepSpec{
+		Operation: "chooseleaf_firstn",
+		Number:    pool.Replicated.ReplicasPerFailureDomain,
+		Type:      pool.Replicated.SubFailureDomain,
+	}
+	steps = append(steps, *stepTakeSubFailureDomain)
+	steps = append(steps, *stepEmit)
+
+	return steps
+}
+
+func generateRuleID(rules []ruleSpec) int {
+	newRulesID := rules[len(rules)-1].ID + 1
+
+	for {
+		ruleIDExists := checkIfRuleIDExists(rules, newRulesID)
+		if !ruleIDExists {
+			break
+		} else {
+			newRulesID++
+		}
+	}
+
+	return newRulesID
+}
+
+func checkIfRuleIDExists(rules []ruleSpec, ID int) bool {
+	for _, rule := range rules {
+		if rule.ID == ID {
+			return true
+		}
+	}
+
+	return false
+}

pkg/daemon/ceph/client/crush_rule_test.go

@@ -0,0 +1,137 @@
+/*
+Copyright 2020 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package client
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/pkg/errors"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/clusterd"
+	exectest "github.com/rook/rook/pkg/util/exec/test"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBuildStretchClusterCrushRule(t *testing.T) {
+	var crushMap CrushMap
+	err := json.Unmarshal([]byte(testCrushMap), &crushMap)
+	assert.NoError(t, err)
+
+	pool := &cephv1.PoolSpec{
+		FailureDomain: "datacenter",
+		CrushRoot:     cephv1.DefaultCRUSHRoot,
+		Replicated: cephv1.ReplicatedSpec{
+			ReplicasPerFailureDomain: 2,
+		},
+	}
+
+	rule := buildStretchClusterCrushRule(crushMap, "stretched", *pool)
+	assert.Equal(t, 2, rule.ID)
+}
+
+func TestBuildCrushSteps(t *testing.T) {
+	pool := &cephv1.PoolSpec{
+		FailureDomain: "datacenter",
+		CrushRoot:     cephv1.DefaultCRUSHRoot,
+		Replicated: cephv1.ReplicatedSpec{
+			ReplicasPerFailureDomain: 2,
+		},
+	}
+	steps := buildStretchClusterCrushSteps(*pool)
+	assert.Equal(t, 4, len(steps))
+	assert.Equal(t, cephv1.DefaultCRUSHRoot, steps[0].ItemName)
+	assert.Equal(t, "datacenter", steps[1].Type)
+	assert.Equal(t, uint(2), steps[2].Number)
+}
+
+func TestCompileCRUSHMap(t *testing.T) {
+	executor := &exectest.MockExecutor{}
+	executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
+		logger.Infof("Command: %s %v", command, args)
+		if command == "crushtool" && args[0] == "--compile" && args[1] == "/tmp/063990228" && args[2] == "--outfn" && args[3] == "/tmp/063990228.compiled" {
+			return "3", nil
+		}
+		return "", errors.Errorf("unexpected ceph command '%v'", args)
+	}
+
+	err := compileCRUSHMap(&clusterd.Context{Executor: executor}, "/tmp/063990228")
+	assert.Nil(t, err)
+}
+
+func TestDecompileCRUSHMap(t *testing.T) {
+	executor := &exectest.MockExecutor{}
+	executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
+		logger.Infof("Command: %s %v", command, args)
+		if command == "crushtool" && args[0] == "--decompile" && args[1] == "/tmp/063990228" && args[2] == "--outfn" && args[3] == "/tmp/063990228.decompiled" {
+			return "3", nil
+		}
+		return "", errors.Errorf("unexpected ceph command '%v'", args)
+	}
+
+	err := decompileCRUSHMap(&clusterd.Context{Executor: executor}, "/tmp/063990228")
+	assert.Nil(t, err)
+}
+
+func TestInjectCRUSHMapMap(t *testing.T) {
+	executor := &exectest.MockExecutor{}
+	executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
+		logger.Infof("Command: %s %v", command, args)
+		if args[0] == "osd" && args[1] == "setcrushmap" && args[2] == "--in-file" && args[3] == "/tmp/063990228.compiled" {
+			return "3", nil
+		}
+		return "", errors.Errorf("unexpected ceph command '%v'", args)
+	}
+
+	err := injectCRUSHMap(&clusterd.Context{Executor: executor}, AdminClusterInfo("mycluster"), "/tmp/063990228.compiled")
+	assert.Nil(t, err)
+}
+
+func TestSetCRUSHMapMap(t *testing.T) {
+	executor := &exectest.MockExecutor{}
+	executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
+		logger.Infof("Command: %s %v", command, args)
+		if args[0] == "osd" && args[1] == "crush" && args[2] == "set" && args[3] == "/tmp/063990228.compiled" {
+			return "3", nil
+		}
+		return "", errors.Errorf("unexpected ceph command '%v'", args)
+	}
+
+	err := setCRUSHMap(&clusterd.Context{Executor: executor}, AdminClusterInfo("mycluster"), "/tmp/063990228.compiled")
+	assert.Nil(t, err)
+}
+
+func Test_generateRuleID(t *testing.T) {
+	tests := []struct {
+		name string
+		args []ruleSpec
+		want int
+	}{
+		{"ordered rules", []ruleSpec{{ID: 1}, {ID: 2}, {ID: 3}}, 4},
+		{"unordered rules", []ruleSpec{{ID: 1}, {ID: 3}, {ID: 2}, {ID: 5}}, 6},
+		{"unordered rules", []ruleSpec{{ID: 1}, {ID: 3}, {ID: 2}}, 4},
+		{"ordered rules", []ruleSpec{{ID: 1}, {ID: 3}}, 4},
+		{"ordered rules", []ruleSpec{{ID: 1}}, 2},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := generateRuleID(tt.args); got != tt.want {
+				t.Errorf("generateRuleID() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

pkg/daemon/ceph/client/crush_test.go

@@ -294,3 +294,8 @@ func TestCrushName(t *testing.T) {
 		}
 	}
 }
+
+func TestBuildCompiledDecompileCRUSHFileName(t *testing.T) {
+	assert.Equal(t, "/tmp/06399022.decompiled", buildDecompileCRUSHFileName("/tmp/06399022"))
+	assert.Equal(t, "/tmp/06399022.compiled", buildCompileCRUSHFileName("/tmp/06399022"))
+}

pkg/daemon/ceph/client/pool.go

@@ -19,6 +19,8 @@ package client
 import (
 	"encoding/json"
 	"fmt"
+	"os"
+	"path/filepath"
 	"strconv"
 	"strings"
 
@@ -302,9 +304,17 @@ func CreateECPoolForApp(context *clusterd.Context, clusterInfo *ClusterInfo, poo
 }
 
 func CreateReplicatedPoolForApp(context *clusterd.Context, clusterInfo *ClusterInfo, poolName string, pool cephv1.PoolSpec, pgCount, appName string) error {
-	// create a crush rule for a replicated pool, if a failure domain is specified
-	if err := createReplicationCrushRule(context, clusterInfo, poolName, pool); err != nil {
-		return err
+	// Create a CRUSH rule for stretched clusters
+	if pool.Replicated.ReplicasPerFailureDomain != 0 {
+		err := createStretchedReplicationCrushRule(context, clusterInfo, poolName, pool)
+		if err != nil {
+			return errors.Wrap(err, "failed to create stretched replicated crush rule")
+		}
+	} else {
+		// create a crush rule for a replicated pool, if a failure domain is specified
+		if err := createReplicationCrushRule(context, clusterInfo, poolName, pool); err != nil {
+			return errors.Wrap(err, "failed to create replicated crush rule")
+		}
 	}
 
 	args := []string{"osd", "pool", "create", poolName, pgCount, "replicated", poolName, "--size", strconv.FormatUint(uint64(pool.Replicated.Size), 10)}
@@ -326,17 +336,104 @@ func CreateReplicatedPoolForApp(context *clusterd.Context, clusterInfo *ClusterI
 	return nil
 }
 
+func createStretchedReplicationCrushRule(context *clusterd.Context, clusterInfo *ClusterInfo, ruleName string, pool cephv1.PoolSpec) error {
+	// set the crush failure domain to the "host" if not already specified
+	if pool.FailureDomain == "" {
+		pool.FailureDomain = cephv1.DefaultFailureDomain
+	}
+	// set the crush failure sub domain to the "host" if not already specified
+	if pool.Replicated.SubFailureDomain == "" {
+		pool.Replicated.SubFailureDomain = cephv1.DefaultFailureDomain
+	}
+	// set the crush root to the default if not already specified
+	if pool.CrushRoot == "" {
+		pool.CrushRoot = cephv1.DefaultCRUSHRoot
+	}
+
+	// Get the current CRUSH map
+	var crushMap CrushMap
+	crushMap, err := GetCrushMap(context, clusterInfo)
+	if err != nil {
+		return errors.Wrap(err, "failed to get crush map")
+	}
+
+	// Fetch the compiled crush map
+	compiledCRUSHMapFilePath, err := GetCompiledCrushMap(context, clusterInfo)
+	if err != nil {
+		return errors.Wrap(err, "failed to get crush map")
+	}
+	defer func() {
+		err := os.Remove(compiledCRUSHMapFilePath)
+		if err != nil {
+			logger.Errorf("failed to remove file %q. %v", compiledCRUSHMapFilePath, err)
+		}
+	}()
+
+	// Decompile the plain text to CRUSH binary format
+	err = decompileCRUSHMap(context, compiledCRUSHMapFilePath)
+	if err != nil {
+		return errors.Wrap(err, "failed to compile crush map")
+	}
+	decompiledCRUSHMapFilePath := buildDecompileCRUSHFileName(compiledCRUSHMapFilePath)
+	defer func() {
+		err := os.Remove(decompiledCRUSHMapFilePath)
+		if err != nil {
+			logger.Errorf("failed to remove file %q. %v", decompiledCRUSHMapFilePath, err)
+		}
+	}()
+
+	// Build plain text rule
+	plainRule := buildStretchClusterPlainCrushRule(crushMap, ruleName, pool)
+
+	// Append plain rule to the decompiled crush map
+	f, err := os.OpenFile(filepath.Clean(decompiledCRUSHMapFilePath), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0400)
+	if err != nil {
+		return errors.Wrapf(err, "failed to open decompiled crush map %q", decompiledCRUSHMapFilePath)
+	}
+	defer func() {
+		err := f.Close()
+		if err != nil {
+			logger.Errorf("failed to close file %q. %v", f.Name(), err)
+		}
+	}()
+
+	// Append the new crush rule into the crush map
+	if _, err := f.WriteString(plainRule); err != nil {
+		return errors.Wrapf(err, "failed to append replicated plain crush rule to decompiled crush map %q", decompiledCRUSHMapFilePath)
+	}
+
+	// Compile the plain text to CRUSH binary format
+	err = compileCRUSHMap(context, decompiledCRUSHMapFilePath)
+	if err != nil {
+		return errors.Wrap(err, "failed to compile crush map")
+	}
+	defer func() {
+		err := os.Remove(buildCompileCRUSHFileName(decompiledCRUSHMapFilePath))
+		if err != nil {
+			logger.Errorf("failed to remove file %q. %v", buildCompileCRUSHFileName(decompiledCRUSHMapFilePath), err)
+		}
+	}()
+
+	// Inject the new CRUSH Map
+	err = injectCRUSHMap(context, clusterInfo, buildCompileCRUSHFileName(decompiledCRUSHMapFilePath))
+	if err != nil {
+		return errors.Wrap(err, "failed to inject crush map")
+	}
+
+	return nil
+}
+
 func createReplicationCrushRule(context *clusterd.Context, clusterInfo *ClusterInfo, ruleName string, pool cephv1.PoolSpec) error {
 	failureDomain := pool.FailureDomain
 	if failureDomain == "" {
 		failureDomain = cephv1.DefaultFailureDomain
 	}
-
 	// set the crush root to the default if not already specified
-	crushRoot := "default"
-	if pool.CrushRoot != "" {
-		crushRoot = pool.CrushRoot
+	crushRoot := pool.CrushRoot
+	if pool.CrushRoot == "" {
+		crushRoot = cephv1.DefaultCRUSHRoot
 	}
+
 	args := []string{"osd", "crush", "rule", "create-replicated", ruleName, crushRoot, failureDomain}
 
 	var deviceClass string

pkg/daemon/ceph/client/pool_test.go

@@ -93,9 +93,6 @@ func testCreateECPool(t *testing.T, overwrite bool, compressionMode string) {
 	}
 }
 
-func TestCreateReplicaPool(t *testing.T) {
-	testCreateReplicaPool(t, "", "", "", "")
-}
 func TestCreateReplicaPoolWithFailureDomain(t *testing.T) {
 	testCreateReplicaPool(t, "osd", "mycrushroot", "", "")
 }

pkg/operator/ceph/object/spec_test.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"testing"
 
+	"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/daemon/ceph/client"
@@ -135,7 +136,16 @@ func TestSSLPodSpec(t *testing.T) {
 }
 
 func TestValidateSpec(t *testing.T) {
-	context := &clusterd.Context{Executor: &exectest.MockExecutor{}}
+	executor := &exectest.MockExecutor{}
+	executor.MockExecuteCommandWithOutputFile = func(command, outputFile string, args ...string) (string, error) {
+		logger.Infof("Command: %s %v", command, args)
+		if args[1] == "crush" && args[2] == "dump" {
+			return `{"types":[{"type_id": 0,"name": "osd"}, {"type_id": 1,"name": "host"}],"buckets":[{"id": -1,"name":"default"},{"id": -2,"name":"good"}, {"id": -3,"name":"host"}]}`, nil
+		}
+		return "", errors.Errorf("unexpected ceph command %q", args)
+	}
+	context := &clusterd.Context{Executor: executor}
+
 	r := &ReconcileCephObjectStore{
 		context: context,
 		cephClusterSpec: &cephv1.ClusterSpec{

pkg/operator/ceph/pool/controller_test.go

@@ -40,129 +40,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
-func TestValidatePool(t *testing.T) {
-	context := &clusterd.Context{Executor: &exectest.MockExecutor{}}
-	clusterInfo := &cephclient.ClusterInfo{Namespace: "myns"}
-
-	// not specifying some replication or EC settings is fine
-	p := cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
-	err := ValidatePool(context, clusterInfo, &p)
-	assert.Nil(t, err)
-
-	// must specify name
-	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Namespace: clusterInfo.Namespace}}
-	err = ValidatePool(context, clusterInfo, &p)
-	assert.NotNil(t, err)
-
-	// must specify namespace
-	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool"}}
-	err = ValidatePool(context, clusterInfo, &p)
-	assert.NotNil(t, err)
-
-	// must not specify both replication and EC settings
-	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
-	p.Spec.Replicated.Size = 1
-	p.Spec.Replicated.RequireSafeReplicaSize = false
-	p.Spec.ErasureCoded.CodingChunks = 2
-	p.Spec.ErasureCoded.DataChunks = 3
-	err = ValidatePool(context, clusterInfo, &p)
-	assert.NotNil(t, err)
-
-	// succeed with replication settings
-	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
-	p.Spec.Replicated.Size = 1
-	p.Spec.Replicated.RequireSafeReplicaSize = false
-	err = ValidatePool(context, clusterInfo, &p)
-	assert.Nil(t, err)
-
-	// size is 1 and RequireSafeReplicaSize is true
-	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
-	p.Spec.Replicated.Size = 1
-	p.Spec.Replicated.RequireSafeReplicaSize = true
-	err = ValidatePool(context, clusterInfo, &p)
-	assert.Error(t, err)
-
-	// succeed with ec settings
-	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
-	p.Spec.ErasureCoded.CodingChunks = 1
-	p.Spec.ErasureCoded.DataChunks = 2
-	err = ValidatePool(context, clusterInfo, &p)
-	assert.Nil(t, err)
-
-	// Tests with various compression modes
-	// succeed with compression mode "none"
-	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
-	p.Spec.Replicated.Size = 1
-	p.Spec.Replicated.RequireSafeReplicaSize = false
-	p.Spec.CompressionMode = "none"
-	err = ValidatePool(context, clusterInfo, &p)
-	assert.Nil(t, err)
-
-	// succeed with compression mode "aggressive"
-	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
-	p.Spec.Replicated.Size = 1
-	p.Spec.Replicated.RequireSafeReplicaSize = false
-	p.Spec.CompressionMode = "aggressive"
-	err = ValidatePool(context, clusterInfo, &p)
-	assert.Nil(t, err)
-
-	// fail with compression mode "unsupported"
-	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
-	p.Spec.Replicated.Size = 1
-	p.Spec.Replicated.RequireSafeReplicaSize = false
-	p.Spec.CompressionMode = "unsupported"
-	err = ValidatePool(context, clusterInfo, &p)
-	assert.Error(t, err)
-
-	// succeed with ec pool and valid compression mode
-	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
-	p.Spec.ErasureCoded.CodingChunks = 1
-	p.Spec.ErasureCoded.DataChunks = 2
-	p.Spec.CompressionMode = "passive"
-	err = ValidatePool(context, clusterInfo, &p)
-	assert.Nil(t, err)
-}
-
-func TestValidateCrushProperties(t *testing.T) {
-	executor := &exectest.MockExecutor{}
-	context := &clusterd.Context{Executor: executor}
-	clusterInfo := &cephclient.ClusterInfo{Namespace: "myns"}
-	executor.MockExecuteCommandWithOutputFile = func(command, outputFile string, args ...string) (string, error) {
-		logger.Infof("Command: %s %v", command, args)
-		if args[1] == "crush" && args[2] == "dump" {
-			return `{"types":[{"type_id": 0,"name": "osd"}],"buckets":[{"id": -1,"name":"default"},{"id": -2,"name":"good"}]}`, nil
-		}
-		return "", errors.Errorf("unexpected ceph command %q", args)
-	}
-
-	// succeed with a failure domain that exists
-	p := &cephv1.CephBlockPool{
-		ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace},
-		Spec: cephv1.PoolSpec{
-			Replicated:    cephv1.ReplicatedSpec{Size: 1, RequireSafeReplicaSize: false},
-			FailureDomain: "osd",
-		},
-	}
-	err := ValidatePool(context, clusterInfo, p)
-	assert.Nil(t, err)
-
-	// fail with a failure domain that doesn't exist
-	p.Spec.FailureDomain = "doesntexist"
-	err = ValidatePool(context, clusterInfo, p)
-	assert.NotNil(t, err)
-
-	// fail with a crush root that doesn't exist
-	p.Spec.FailureDomain = "osd"
-	p.Spec.CrushRoot = "bad"
-	err = ValidatePool(context, clusterInfo, p)
-	assert.NotNil(t, err)
-
-	// fail with a crush root that does exist
-	p.Spec.CrushRoot = "good"
-	err = ValidatePool(context, clusterInfo, p)
-	assert.Nil(t, err)
-}
-
 func TestCreatePool(t *testing.T) {
 	clusterInfo := &cephclient.ClusterInfo{Namespace: "myns"}
 	executor := &exectest.MockExecutor{

pkg/operator/ceph/pool/validate.go

@@ -83,9 +83,33 @@ func ValidatePoolSpec(context *clusterd.Context, clusterInfo *client.ClusterInfo
 		}
 	}
 
+	// validate the crush subdomain if specified
+	if p.Replicated.SubFailureDomain != "" {
+		found := false
+		for _, t := range crush.Buckets {
+			if t.Name == p.Replicated.SubFailureDomain {
+				found = true
+				break
+			}
+		}
+		if !found {
+			return errors.Errorf("unrecognized crush sub domain %s", p.Replicated.SubFailureDomain)
+		}
+	}
+
 	// validate pool replica size
-	if p.Replicated.Size == 1 && p.Replicated.RequireSafeReplicaSize {
-		return errors.Errorf("error pool size is %d and requireSafeReplicaSize is %t, must be false", p.Replicated.Size, p.Replicated.RequireSafeReplicaSize)
+	if p.IsReplicated() {
+		if p.Replicated.Size == 1 && p.Replicated.RequireSafeReplicaSize {
+			return errors.Errorf("error pool size is %d and requireSafeReplicaSize is %t, must be false", p.Replicated.Size, p.Replicated.RequireSafeReplicaSize)
+		}
+
+		if p.Replicated.Size <= p.Replicated.ReplicasPerFailureDomain {
+			return errors.Errorf("error pool size is %d and replicasPerFailureDomain is %d, size must be greater", p.Replicated.Size, p.Replicated.ReplicasPerFailureDomain)
+		}
+
+		if p.Replicated.ReplicasPerFailureDomain != 0 && p.Replicated.Size%p.Replicated.ReplicasPerFailureDomain != 0 {
+			return errors.Errorf("error replicasPerFailureDomain is %d must be a factor of the replica count %d", p.Replicated.ReplicasPerFailureDomain, p.Replicated.Size)
+		}
 	}
 
 	// validate pool compression mode if specified

pkg/operator/ceph/pool/validate_test.go

@@ -0,0 +1,186 @@
+/*
+Copyright 2020 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package pool to manage a rook pool.
+package pool
+
+import (
+	"testing"
+
+	"github.com/pkg/errors"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/clusterd"
+	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
+	exectest "github.com/rook/rook/pkg/util/exec/test"
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestValidatePool(t *testing.T) {
+	context := &clusterd.Context{Executor: &exectest.MockExecutor{}}
+	clusterInfo := &cephclient.ClusterInfo{Namespace: "myns"}
+
+	// not specifying some replication or EC settings is fine
+	p := cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
+	err := ValidatePool(context, clusterInfo, &p)
+	assert.Nil(t, err)
+
+	// must specify name
+	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Namespace: clusterInfo.Namespace}}
+	err = ValidatePool(context, clusterInfo, &p)
+	assert.NotNil(t, err)
+
+	// must specify namespace
+	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool"}}
+	err = ValidatePool(context, clusterInfo, &p)
+	assert.NotNil(t, err)
+
+	// must not specify both replication and EC settings
+	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
+	p.Spec.Replicated.Size = 1
+	p.Spec.Replicated.RequireSafeReplicaSize = false
+	p.Spec.ErasureCoded.CodingChunks = 2
+	p.Spec.ErasureCoded.DataChunks = 3
+	err = ValidatePool(context, clusterInfo, &p)
+	assert.NotNil(t, err)
+
+	// succeed with replication settings
+	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
+	p.Spec.Replicated.Size = 1
+	p.Spec.Replicated.RequireSafeReplicaSize = false
+	err = ValidatePool(context, clusterInfo, &p)
+	assert.Nil(t, err)
+
+	// size is 1 and RequireSafeReplicaSize is true
+	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
+	p.Spec.Replicated.Size = 1
+	p.Spec.Replicated.RequireSafeReplicaSize = true
+	err = ValidatePool(context, clusterInfo, &p)
+	assert.Error(t, err)
+
+	// succeed with ec settings
+	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
+	p.Spec.ErasureCoded.CodingChunks = 1
+	p.Spec.ErasureCoded.DataChunks = 2
+	err = ValidatePool(context, clusterInfo, &p)
+	assert.Nil(t, err)
+
+	// Tests with various compression modes
+	// succeed with compression mode "none"
+	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
+	p.Spec.Replicated.Size = 1
+	p.Spec.Replicated.RequireSafeReplicaSize = false
+	p.Spec.CompressionMode = "none"
+	err = ValidatePool(context, clusterInfo, &p)
+	assert.Nil(t, err)
+
+	// succeed with compression mode "aggressive"
+	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
+	p.Spec.Replicated.Size = 1
+	p.Spec.Replicated.RequireSafeReplicaSize = false
+	p.Spec.CompressionMode = "aggressive"
+	err = ValidatePool(context, clusterInfo, &p)
+	assert.Nil(t, err)
+
+	// fail with compression mode "unsupported"
+	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
+	p.Spec.Replicated.Size = 1
+	p.Spec.Replicated.RequireSafeReplicaSize = false
+	p.Spec.CompressionMode = "unsupported"
+	err = ValidatePool(context, clusterInfo, &p)
+	assert.Error(t, err)
+
+	// fail since replica size is lower than ReplicasPerFailureDomain
+	p.Spec.Replicated.ReplicasPerFailureDomain = 2
+	err = ValidatePool(context, clusterInfo, &p)
+	assert.Error(t, err)
+
+	// fail since replica size is equal than ReplicasPerFailureDomain
+	p.Spec.Replicated.Size = 2
+	p.Spec.Replicated.ReplicasPerFailureDomain = 2
+	err = ValidatePool(context, clusterInfo, &p)
+	assert.Error(t, err)
+
+	// fail since ReplicasPerFailureDomain is not a power of 2
+	p.Spec.Replicated.Size = 4
+	p.Spec.Replicated.ReplicasPerFailureDomain = 3
+	err = ValidatePool(context, clusterInfo, &p)
+	assert.Error(t, err)
+
+	// fail since ReplicasPerFailureDomain is not a power of 2
+	p.Spec.Replicated.Size = 4
+	p.Spec.Replicated.ReplicasPerFailureDomain = 5
+	err = ValidatePool(context, clusterInfo, &p)
+	assert.Error(t, err)
+
+	// Failure the sub domain does not exist
+	p.Spec.Replicated.SubFailureDomain = "dummy"
+	err = ValidatePool(context, clusterInfo, &p)
+	assert.Error(t, err)
+
+	// succeed with ec pool and valid compression mode
+	p = cephv1.CephBlockPool{ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace}}
+	p.Spec.ErasureCoded.CodingChunks = 1
+	p.Spec.ErasureCoded.DataChunks = 2
+	p.Spec.CompressionMode = "passive"
+	err = ValidatePool(context, clusterInfo, &p)
+	assert.Nil(t, err)
+}
+
+func TestValidateCrushProperties(t *testing.T) {
+	executor := &exectest.MockExecutor{}
+	context := &clusterd.Context{Executor: executor}
+	clusterInfo := &cephclient.ClusterInfo{Namespace: "myns"}
+	executor.MockExecuteCommandWithOutputFile = func(command, outputFile string, args ...string) (string, error) {
+		logger.Infof("Command: %s %v", command, args)
+		if args[1] == "crush" && args[2] == "dump" {
+			return `{"types":[{"type_id": 0,"name": "osd"}],"buckets":[{"id": -1,"name":"default"},{"id": -2,"name":"good"}, {"id": -3,"name":"host"}]}`, nil
+		}
+		return "", errors.Errorf("unexpected ceph command %q", args)
+	}
+
+	// succeed with a failure domain that exists
+	p := &cephv1.CephBlockPool{
+		ObjectMeta: metav1.ObjectMeta{Name: "mypool", Namespace: clusterInfo.Namespace},
+		Spec: cephv1.PoolSpec{
+			Replicated: cephv1.ReplicatedSpec{Size: 1, RequireSafeReplicaSize: false},
+		},
+	}
+	err := ValidatePool(context, clusterInfo, p)
+	assert.Nil(t, err)
+
+	// fail with a failure domain that doesn't exist
+	p.Spec.FailureDomain = "doesntexist"
+	err = ValidatePool(context, clusterInfo, p)
+	assert.NotNil(t, err)
+
+	// fail with a crush root that doesn't exist
+	p.Spec.FailureDomain = "osd"
+	p.Spec.CrushRoot = "bad"
+	err = ValidatePool(context, clusterInfo, p)
+	assert.NotNil(t, err)
+
+	// fail with a crush root that does exist
+	p.Spec.CrushRoot = "good"
+	err = ValidatePool(context, clusterInfo, p)
+	assert.Nil(t, err)
+
+	// Success replica size is 4 and replicasPerFailureDomain is 2
+	p.Spec.Replicated.Size = 4
+	p.Spec.Replicated.ReplicasPerFailureDomain = 2
+	err = ValidatePool(context, clusterInfo, p)
+	assert.NoError(t, err)
+}

tests/framework/installer/ceph_manifests.go

@@ -340,6 +340,10 @@ spec:
                       type: integer
                     requireSafeReplicaSize:
                       type: boolean
+                    replicasPerFailureDomain:
+                      type: integer
+                    subFailureDomain:
+                      type: string
                 erasureCoded:
                   properties:
                     dataChunks:
@@ -374,6 +378,10 @@ spec:
                         type: integer
                       requireSafeReplicaSize:
                         type: boolean
+                    replicasPerFailureDomain:
+                      type: integer
+                    subFailureDomain:
+                      type: string
                   erasureCoded:
                     properties:
                       dataChunks:
@@ -491,6 +499,10 @@ spec:
                       type: integer
                     requireSafeReplicaSize:
                       type: boolean
+                    replicasPerFailureDomain:
+                      type: integer
+                    subFailureDomain:
+                      type: string
                 erasureCoded:
                   properties:
                     dataChunks:
@@ -519,6 +531,10 @@ spec:
                       type: integer
                     requireSafeReplicaSize:
                       type: boolean
+                    replicasPerFailureDomain:
+                      type: integer
+                    subFailureDomain:
+                      type: string
                 erasureCoded:
                   properties:
                     dataChunks:
@@ -651,6 +667,10 @@ spec:
                   type: number
                 requireSafeReplicaSize:
                   type: boolean
+                replicasPerFailureDomain:
+                  type: integer
+                subFailureDomain:
+                  type: string
             erasureCoded:
               properties:
                 dataChunks: