enable rbd mirror daemon in the cluster crd

Signed-off-by: travisn <tnielsen@redhat.com>

Documentation/ceph-cluster-crd.md

@@ -58,6 +58,9 @@ If this value is empty, each pod will get an ephemeral directory to store their
   - `hostNetwork`: uses network of the hosts instead of using the SDN below the containers.
 - `mon`: contains mon related options [mon settings](#mon-settings)
 For more details on the mons and when to choose a number other than `3`, see the [mon health design doc](https://github.com/rook/rook/blob/master/design/mon-health.md).
+- `rbdMirroring`: The settings for rbd mirror daemon(s). Configuring which pools or images to be mirrored must be completed in the rook toolbox by running the
+[rbd mirror](http://docs.ceph.com/docs/mimic/rbd/rbd-mirroring/) command.
+  - `workers`: The number of rbd daemons to perform the rbd mirroring between clusters.
 - `placement`: [placement configuration settings](#placement-configuration-settings)
 - `resources`: [resources configuration settings](#cluster-wide-resources-configuration-settings)
 - `storage`: Storage selection and configuration that will be used across the cluster.  Note that these settings can be overridden for specific nodes.

PendingReleaseNotes.md

@@ -17,6 +17,7 @@
 - The toolbox manifest now creates a deployment based on the `rook/ceph` image instead of creating a pod on a specialized `rook/ceph-toolbox` image.
 - The frequency of discovering devices on a node is reduced to 60 minutes by default, and is configurable with the setting `ROOK_DISCOVER_DEVICES_INTERVAL` in operator.yaml.
 - The number of mons can be changed by updating the `mon.count` in the cluster CRD.
+- RBD Mirroring is enabled by Rook. By setting the number of [rbd mirroring workers](Documentation/ceph-cluster-crd.md#cluster-settings), the daemon(s) will be started by rook. To configure the pools or images to be mirrored, use the Rook toolbox to run the [rbd mirror](http://docs.ceph.com/docs/mimic/rbd/rbd-mirroring/) configuration tool.
 - Object Store User creation via CRD for Ceph clusters.
 
 ## Breaking Changes

cluster/examples/kubernetes/ceph/cluster.yaml

@@ -83,6 +83,10 @@ spec:
   network:
     # toggle to use hostNetwork
     hostNetwork: false
+  rbdMirroring:
+    # The number of daemons that will perform the rbd mirroring.
+    # rbd mirroring must be configured with "rbd mirror" from the rook toolbox.
+    workers: 0
   # To control where various services will be scheduled by kubernetes, use the placement configuration sections below.
   # The example under 'all' would have all services scheduled on kubernetes nodes labeled with 'role=storage-node' and
   # tolerate taints with a key of 'storage-node'.

cmd/rook/ceph/ceph.go

@@ -66,6 +66,7 @@ func AddCommands(command *cobra.Command) {
 	command.AddCommand(mgrCmd)
 	command.AddCommand(rgwCmd)
 	command.AddCommand(mdsCmd)
+	command.AddCommand(configCmd)
 }
 
 func createContext() *clusterd.Context {

cmd/rook/ceph/config.go

@@ -0,0 +1,99 @@
+/*
+Copyright 2018 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 ceph
+
+import (
+	"fmt"
+
+	"github.com/rook/rook/cmd/rook/rook"
+	cephconfig "github.com/rook/rook/pkg/daemon/ceph/config"
+	mondaemon "github.com/rook/rook/pkg/daemon/ceph/mon"
+	"github.com/rook/rook/pkg/util"
+	"github.com/rook/rook/pkg/util/flags"
+	"github.com/spf13/cobra"
+)
+
+const (
+	configKeyringTemplate = `
+[%s]
+key = %s
+`
+)
+
+var configCmd = &cobra.Command{
+	Use:    "config-init",
+	Short:  "Generates basic ceph config",
+	Hidden: true,
+}
+
+var (
+	configKeyring  string
+	configUsername string
+)
+
+func init() {
+	configCmd.Flags().StringVar(&configKeyring, "keyring", "", "the daemon keyring")
+	configCmd.Flags().StringVar(&configUsername, "username", "", "the daemon username")
+	addCephFlags(configCmd)
+
+	flags.SetFlagsFromEnv(configCmd.Flags(), rook.RookEnvVarPrefix)
+
+	configCmd.RunE = initConfig
+}
+
+func initConfig(cmd *cobra.Command, args []string) error {
+	required := []string{
+		"username", "keyring", "mon-endpoints"}
+	if err := flags.VerifyRequiredFlags(configCmd, required); err != nil {
+		return err
+	}
+
+	if err := verifyRenamedFlags(configCmd); err != nil {
+		return err
+	}
+
+	rook.SetLogLevel()
+
+	rook.LogStartupInfo(configCmd.Flags())
+
+	clusterInfo.Monitors = mondaemon.ParseMonEndpoints(cfg.monEndpoints)
+	clusterInfo.Name = "ceph"
+	context := createContext()
+
+	keyringPath := "/etc/ceph/keyring"
+	_, err := cephconfig.GenerateConfigFile(context, &clusterInfo, "/etc/ceph", configUsername, keyringPath, nil, nil)
+	if err != nil {
+		return fmt.Errorf("failed to create config file: %+v", err)
+	}
+
+	util.WriteFileToLog(logger, cephconfig.DefaultConfigFilePath())
+
+	keyringEval := func(key string) string {
+		r := fmt.Sprintf(configKeyringTemplate, configUsername, key)
+		return r
+	}
+
+	err = cephconfig.WriteKeyring(keyringPath, configKeyring, keyringEval)
+	if err != nil {
+		return fmt.Errorf("failed to create keyring: %+v", err)
+	}
+	if err != nil {
+		rook.TerminateFatal(err)
+	}
+
+	return nil
+}

pkg/apis/ceph.rook.io/v1beta1/placement.go

@@ -20,9 +20,10 @@ import (
 )
 
 const (
-	PlacementKeyMgr = "mgr"
-	PlacementKeyMon = "mon"
-	PlacementKeyOSD = "osd"
+	PlacementKeyMgr       = "mgr"
+	PlacementKeyMon       = "mon"
+	PlacementKeyOSD       = "osd"
+	PlacementKeyRBDMirror = "rbdmirror"
 )
 
 // GetMgrPlacement returns the placement for the MGR service
@@ -39,3 +40,8 @@ func GetMonPlacement(p rook.PlacementSpec) rook.Placement {
 func GetOSDPlacement(p rook.PlacementSpec) rook.Placement {
 	return p.All().Merge(p[PlacementKeyOSD])
 }
+
+// GetRBDMirrorPlacement returns the placement for the RBD mirrors
+func GetRBDMirrorPlacement(p rook.PlacementSpec) rook.Placement {
+	return p.All().Merge(p[PlacementKeyRBDMirror])
+}

pkg/apis/ceph.rook.io/v1beta1/resources.go

@@ -21,9 +21,10 @@ import (
 )
 
 const (
-	ResourcesKeyMgr = "mgr"
-	ResourcesKeyMon = "mon"
-	ResourcesKeyOSD = "osd"
+	ResourcesKeyMgr       = "mgr"
+	ResourcesKeyMon       = "mon"
+	ResourcesKeyOSD       = "osd"
+	ResourcesKeyRBDMirror = "rbdmirror"
 )
 
 // GetMgrResources returns the placement for the MGR service
@@ -40,3 +41,8 @@ func GetMonResources(p rook.ResourceSpec) v1.ResourceRequirements {
 func GetOSDResources(p rook.ResourceSpec) v1.ResourceRequirements {
 	return p[ResourcesKeyOSD]
 }
+
+// GetRBDMirrorResources returns the placement for the RBD Mirrors
+func GetRBDMirrorResources(p rook.ResourceSpec) v1.ResourceRequirements {
+	return p[ResourcesKeyRBDMirror]
+}

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

@@ -72,6 +72,9 @@ type ClusterSpec struct {
 	// A spec for mon related options
 	Mon MonSpec `json:"mon"`
 
+	// A spec for rbd mirroring
+	RBDMirroring RBDMirroringSpec `json:"rbdMirroring"`
+
 	// Dashboard settings
 	Dashboard DashboardSpec `json:"dashboard,omitempty"`
 }
@@ -115,6 +118,10 @@ type MonSpec struct {
 	AllowMultiplePerNode bool `json:"allowMultiplePerNode"`
 }
 
+type RBDMirroringSpec struct {
+	Workers int `json:"workers"`
+}
+
 // +genclient
 // +genclient:noStatus
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/apis/ceph.rook.io/v1beta1/zz_generated.deepcopy.go

@@ -123,6 +123,7 @@ func (in *ClusterSpec) DeepCopyInto(out *ClusterSpec) {
 		}
 	}
 	out.Mon = in.Mon
+	out.RBDMirroring = in.RBDMirroring
 	out.Dashboard = in.Dashboard
 	return
 }
@@ -555,6 +556,22 @@ func (in *PoolSpec) DeepCopy() *PoolSpec {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *RBDMirroringSpec) DeepCopyInto(out *RBDMirroringSpec) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RBDMirroringSpec.
+func (in *RBDMirroringSpec) DeepCopy() *RBDMirroringSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(RBDMirroringSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ReplicatedSpec) DeepCopyInto(out *ReplicatedSpec) {
 	*out = *in

pkg/operator/ceph/cluster/cluster.go

@@ -31,6 +31,7 @@ import (
 	"github.com/rook/rook/pkg/operator/ceph/cluster/mgr"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd"
+	"github.com/rook/rook/pkg/operator/ceph/cluster/rbd"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	batch "k8s.io/api/batch/v1"
 	"k8s.io/api/core/v1"
@@ -54,8 +55,6 @@ type cluster struct {
 	Namespace string
 	Spec      *cephv1beta1.ClusterSpec
 	mons      *mon.Cluster
-	mgrs      *mgr.Cluster
-	osds      *osd.Cluster
 	stopCh    chan struct{}
 	ownerRef  metav1.OwnerReference
 }
@@ -172,21 +171,29 @@ func (c *cluster) createInstance(rookImage string) error {
 		return fmt.Errorf("failed to create initial crushmap: %+v", err)
 	}
 
-	c.mgrs = mgr.New(c.context, c.Namespace, rookImage, c.Spec.CephVersion, cephv1beta1.GetMgrPlacement(c.Spec.Placement),
+	mgrs := mgr.New(c.context, c.Namespace, rookImage, c.Spec.CephVersion, cephv1beta1.GetMgrPlacement(c.Spec.Placement),
 		c.Spec.Network.HostNetwork, c.Spec.Dashboard, cephv1beta1.GetMgrResources(c.Spec.Resources), c.ownerRef)
-	err = c.mgrs.Start()
+	err = mgrs.Start()
 	if err != nil {
 		return fmt.Errorf("failed to start the ceph mgr. %+v", err)
 	}
 
 	// Start the OSDs
-	c.osds = osd.New(c.context, c.Namespace, rookImage, c.Spec.CephVersion, c.Spec.ServiceAccount, c.Spec.Storage, c.Spec.DataDirHostPath,
+	osds := osd.New(c.context, c.Namespace, rookImage, c.Spec.CephVersion, c.Spec.ServiceAccount, c.Spec.Storage, c.Spec.DataDirHostPath,
 		cephv1beta1.GetOSDPlacement(c.Spec.Placement), c.Spec.Network.HostNetwork, cephv1beta1.GetOSDResources(c.Spec.Resources), c.ownerRef)
-	err = c.osds.Start()
+	err = osds.Start()
 	if err != nil {
 		return fmt.Errorf("failed to start the osds. %+v", err)
 	}
 
+	// Start the rbd mirroring daemon(s)
+	rbdmirror := rbd.New(c.context, c.Namespace, rookImage, c.Spec.CephVersion, cephv1beta1.GetRBDMirrorPlacement(c.Spec.Placement),
+		c.Spec.Network.HostNetwork, c.Spec.RBDMirroring, cephv1beta1.GetRBDMirrorResources(c.Spec.Resources), c.ownerRef)
+	err = rbdmirror.Start()
+	if err != nil {
+		return fmt.Errorf("failed to start the rbd mirrors. %+v", err)
+	}
+
 	logger.Infof("Done creating rook instance in namespace %s", c.Namespace)
 	return nil
 }
@@ -286,6 +293,11 @@ func clusterChanged(oldCluster, newCluster cephv1beta1.ClusterSpec, clusterRef *
 		clusterRef.mons.MonCountMutex.Unlock()
 	}
 
+	if oldCluster.RBDMirroring.Workers != newCluster.RBDMirroring.Workers {
+		logger.Infof("rbd mirrors changed from %d to %d", oldCluster.RBDMirroring.Workers, newCluster.RBDMirroring.Workers)
+		changeFound = true
+	}
+
 	if oldCluster.CephVersion.AllowUnsupported != newCluster.CephVersion.AllowUnsupported {
 		logger.Infof("ceph version allowUnsupported has changed from %t to %t", oldCluster.CephVersion.AllowUnsupported, newCluster.CephVersion.AllowUnsupported)
 		changeFound = true

pkg/operator/ceph/cluster/mgr/mgr.go

@@ -25,6 +25,7 @@ import (
 	rookalpha "github.com/rook/rook/pkg/apis/rook.io/v1alpha2"
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
+	opspec "github.com/rook/rook/pkg/operator/ceph/spec"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -35,7 +36,6 @@ var logger = capnslog.NewPackageLogger("github.com/rook/rook", "op-mgr")
 
 const (
 	appName              = "rook-ceph-mgr"
-	keyringSecretKeyName = "keyring"
 	prometheusModuleName = "prometheus"
 	metricsPort          = 9283
 )
@@ -98,7 +98,10 @@ func (c *Cluster) Start() error {
 
 		daemonName := mgrNames[i]
 		resourceName := fmt.Sprintf("%s-%s", appName, daemonName)
-		if err := c.createKeyring(c.Namespace, resourceName, daemonName); err != nil {
+		username := fmt.Sprintf("mgr.%s", daemonName)
+		access := []string{"mon", "allow profile mgr", "mds", "allow *", "osd", "allow *"}
+		cfg := opspec.KeyringConfig{Namespace: c.Namespace, ResourceName: resourceName, DaemonName: daemonName, OwnerRef: c.ownerRef, Username: username, Access: access}
+		if err := opspec.CreateKeyring(c.context, cfg); err != nil {
 			return fmt.Errorf("failed to create %s keyring. %+v", resourceName, err)
 		}
 
@@ -145,44 +148,6 @@ func (c *Cluster) Start() error {
 	return nil
 }
 
-func (c *Cluster) createKeyring(clusterName, name, daemonName string) error {
-	_, err := c.context.Clientset.CoreV1().Secrets(c.Namespace).Get(name, metav1.GetOptions{})
-	if err == nil {
-		logger.Infof("the mgr keyring was already generated")
-		return nil
-	}
-	if !errors.IsNotFound(err) {
-		return fmt.Errorf("failed to get mgr secrets. %+v", err)
-	}
-
-	// get-or-create-key for the user account
-	keyring, err := createKeyring(c.context, clusterName, daemonName)
-	if err != nil {
-		return fmt.Errorf("failed to create mgr keyring. %+v", err)
-	}
-
-	// Store the keyring in a secret
-	secrets := map[string]string{
-		keyringSecretKeyName: keyring,
-	}
-	secret := &v1.Secret{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      name,
-			Namespace: c.Namespace,
-		},
-		StringData: secrets,
-		Type:       k8sutil.RookType,
-	}
-	k8sutil.SetOwnerRef(c.context.Clientset, c.Namespace, &secret.ObjectMeta, &c.ownerRef)
-
-	_, err = c.context.Clientset.CoreV1().Secrets(c.Namespace).Create(secret)
-	if err != nil {
-		return fmt.Errorf("failed to save mgr secrets. %+v", err)
-	}
-
-	return nil
-}
-
 // Ceph docs about the prometheus module: http://docs.ceph.com/docs/master/mgr/prometheus/
 func (c *Cluster) enablePrometheusModule(clusterName string) error {
 	if err := client.MgrEnableModule(c.context, clusterName, prometheusModuleName, true); err != nil {
@@ -190,21 +155,3 @@ func (c *Cluster) enablePrometheusModule(clusterName string) error {
 	}
 	return nil
 }
-
-func getKeyringProperties(name string) (string, []string) {
-	username := fmt.Sprintf("mgr.%s", name)
-	access := []string{"mon", "allow profile mgr", "mds", "allow *", "osd", "allow *"}
-	return username, access
-}
-
-// create a keyring for the mds client with a limited set of privileges
-func createKeyring(context *clusterd.Context, clusterName, name string) (string, error) {
-	// get-or-create-key for the user account
-	username, access := getKeyringProperties(name)
-	keyring, err := client.AuthGetOrCreateKey(context, clusterName, username, access)
-	if err != nil {
-		return "", fmt.Errorf("failed to get or create auth key for %s. %+v", username, err)
-	}
-
-	return keyring, nil
-}

pkg/operator/ceph/cluster/mgr/spec.go

@@ -87,7 +87,7 @@ func (c *Cluster) makeConfigInitContainer(mgrConfig *mgrConfig) v1.Container {
 				ValueFrom: &v1.EnvVarSource{
 					SecretKeyRef: &v1.SecretKeySelector{
 						LocalObjectReference: v1.LocalObjectReference{Name: mgrConfig.ResourceName},
-						Key:                  keyringSecretKeyName,
+						Key:                  opspec.KeyringSecretKeyName,
 					}}},
 			k8sutil.PodIPEnvVar(k8sutil.PrivateIPEnvVar),
 			k8sutil.PodIPEnvVar(k8sutil.PublicIPEnvVar),

pkg/operator/ceph/cluster/rbd/mirror.go

@@ -0,0 +1,145 @@
+/*
+Copyright 2016 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 rbd for mirroring
+package rbd
+
+import (
+	"fmt"
+
+	"github.com/coreos/pkg/capnslog"
+	cephv1beta1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1beta1"
+	rookalpha "github.com/rook/rook/pkg/apis/rook.io/v1alpha2"
+	"github.com/rook/rook/pkg/clusterd"
+	"github.com/rook/rook/pkg/operator/ceph/spec"
+	"github.com/rook/rook/pkg/operator/k8sutil"
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+var logger = capnslog.NewPackageLogger("github.com/rook/rook", "rbd-mirror")
+
+const (
+	appName = "rook-ceph-rbd-mirror"
+)
+
+// Cluster represents the Rook and environment configuration settings needed to set up rbd mirroring.
+type Mirroring struct {
+	Namespace   string
+	placement   rookalpha.Placement
+	context     *clusterd.Context
+	resources   v1.ResourceRequirements
+	ownerRef    metav1.OwnerReference
+	spec        cephv1beta1.RBDMirroringSpec
+	cephVersion cephv1beta1.CephVersionSpec
+	rookVersion string
+	hostNetwork bool
+}
+
+// New creates an instance of the rbd mirroring
+func New(context *clusterd.Context, namespace, rookVersion string, cephVersion cephv1beta1.CephVersionSpec, placement rookalpha.Placement, hostNetwork bool,
+	spec cephv1beta1.RBDMirroringSpec, resources v1.ResourceRequirements, ownerRef metav1.OwnerReference) *Mirroring {
+	return &Mirroring{
+		context:     context,
+		Namespace:   namespace,
+		placement:   placement,
+		rookVersion: rookVersion,
+		cephVersion: cephVersion,
+		spec:        spec,
+		hostNetwork: hostNetwork,
+		resources:   resources,
+		ownerRef:    ownerRef,
+	}
+}
+
+// Start begins the process of running rbd mirroring daemons.
+func (m *Mirroring) Start() error {
+	logger.Infof("configure rbd mirroring with %d workers", m.spec.Workers)
+
+	access := []string{"mon", "profile rbd-mirror", "osd", "profile rbd"}
+
+	for i := 0; i < m.spec.Workers; i++ {
+		daemonName := k8sutil.IndexToName(i)
+		username := fullDaemonName(daemonName)
+		resourceName := fmt.Sprintf("%s-%s", appName, daemonName)
+		cfg := spec.KeyringConfig{Namespace: m.Namespace, ResourceName: resourceName, DaemonName: daemonName, OwnerRef: m.ownerRef, Username: username, Access: access}
+		if err := spec.CreateKeyring(m.context, cfg); err != nil {
+			return fmt.Errorf("failed to create %s keyring. %+v", resourceName, err)
+		}
+
+		// Start the deployment
+		deployment := m.makeDeployment(resourceName, daemonName)
+		if _, err := m.context.Clientset.ExtensionsV1beta1().Deployments(m.Namespace).Create(deployment); err != nil {
+			if !errors.IsAlreadyExists(err) {
+				return fmt.Errorf("failed to create %s deployment. %+v", resourceName, err)
+			}
+			logger.Infof("%s deployment already exists", resourceName)
+		} else {
+			logger.Infof("%s deployment started", resourceName)
+		}
+	}
+
+	// Remove extra rbd mirror deployments if necessary
+	err := m.removeExtraMirrors()
+	if err != nil {
+		logger.Errorf("failed to remove extra mirrors. %+v", err)
+	}
+
+	return nil
+}
+
+func (m *Mirroring) removeExtraMirrors() error {
+	opts := metav1.ListOptions{LabelSelector: fmt.Sprintf("app=%s", appName)}
+	d, err := m.context.Clientset.ExtensionsV1beta1().Deployments(m.Namespace).List(opts)
+	if err != nil {
+		return fmt.Errorf("failed to get mirrors. %+v", err)
+	}
+
+	if len(d.Items) <= m.spec.Workers {
+		logger.Infof("no extra daemons to remove")
+		return nil
+	}
+
+	for _, deploy := range d.Items {
+		daemonName, ok := deploy.Labels["rbdmirror"]
+		if !ok {
+			logger.Warningf("unrecognized rbdmirror %s", deploy.Name)
+			continue
+		}
+		index, err := k8sutil.NameToIndex(daemonName)
+		if err != nil {
+			logger.Warningf("unrecognized rbdmirror %s with label %s", deploy.Name, daemonName)
+			continue
+		}
+		if index >= m.spec.Workers {
+			logger.Infof("removing extra rbd mirror %s", daemonName)
+			var gracePeriod int64
+			propagation := metav1.DeletePropagationForeground
+			deleteOpts := metav1.DeleteOptions{GracePeriodSeconds: &gracePeriod, PropagationPolicy: &propagation}
+			if err = m.context.Clientset.ExtensionsV1beta1().Deployments(m.Namespace).Delete(deploy.Name, &deleteOpts); err != nil {
+				logger.Warningf("failed to delete rbd mirror %s. %+v", daemonName, err)
+			}
+
+			logger.Infof("removed rbd mirror %s", daemonName)
+		}
+	}
+	return nil
+}
+
+func fullDaemonName(daemonName string) string {
+	return fmt.Sprintf("client.rbd-mirror.%s", daemonName)
+}

pkg/operator/ceph/cluster/rbd/mirror_test.go

@@ -0,0 +1,70 @@
+/*
+Copyright 2016 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 rbd
+
+import (
+	"testing"
+
+	cephv1beta1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1beta1"
+	rookalpha "github.com/rook/rook/pkg/apis/rook.io/v1alpha2"
+	"github.com/rook/rook/pkg/clusterd"
+	testop "github.com/rook/rook/pkg/operator/test"
+	exectest "github.com/rook/rook/pkg/util/exec/test"
+	"github.com/stretchr/testify/assert"
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestRBDMirror(t *testing.T) {
+	clientset := testop.New(1)
+	keysCreated := map[string]bool{}
+	executor := &exectest.MockExecutor{}
+	executor.MockExecuteCommandWithOutputFile = func(debug bool, actionName string, command string, outFileArg string, args ...string) (string, error) {
+		logger.Infof("%s %+v", command, args)
+		if args[0] == "auth" && args[1] == "get-or-create-key" {
+			daemonName := args[2]
+			keysCreated[daemonName] = true
+			return "{\"key\":\"mysecurekey\"}", nil
+		}
+		return "", nil
+	}
+
+	c := New(&clusterd.Context{Clientset: clientset, Executor: executor},
+		"ns",
+		"rook/rook:myversion",
+		cephv1beta1.CephVersionSpec{Image: "ceph/ceph:myceph"},
+		rookalpha.Placement{},
+		false,
+		cephv1beta1.RBDMirroringSpec{Workers: 2},
+		v1.ResourceRequirements{},
+		metav1.OwnerReference{},
+	)
+
+	err := c.Start()
+	assert.Nil(t, err)
+	assert.True(t, keysCreated[fullDaemonName("a")])
+	assert.True(t, keysCreated[fullDaemonName("b")])
+	assert.False(t, keysCreated[fullDaemonName("c")])
+
+	opts := metav1.ListOptions{}
+	d, err := clientset.ExtensionsV1beta1().Deployments(c.Namespace).List(opts)
+	assert.Equal(t, 2, len(d.Items))
+	for _, de := range d.Items {
+		daemonName := de.Name[len(de.Name)-1:]
+		assert.True(t, keysCreated[fullDaemonName(daemonName)])
+	}
+}

pkg/operator/ceph/cluster/rbd/spec.go

@@ -0,0 +1,108 @@
+/*
+Copyright 2016 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 rbd for mirroring
+package rbd
+
+import (
+	opmon "github.com/rook/rook/pkg/operator/ceph/cluster/mon"
+	opspec "github.com/rook/rook/pkg/operator/ceph/spec"
+	"github.com/rook/rook/pkg/operator/k8sutil"
+	"k8s.io/api/core/v1"
+	extensions "k8s.io/api/extensions/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func (m *Mirroring) makeDeployment(resourceName, daemonName string) *extensions.Deployment {
+	podSpec := v1.PodTemplateSpec{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   resourceName,
+			Labels: opspec.PodLabels(appName, m.Namespace, "rbdmirror", daemonName),
+		},
+		Spec: v1.PodSpec{
+			InitContainers: []v1.Container{
+				m.makeConfigInitContainer(resourceName, daemonName),
+			},
+			Containers: []v1.Container{
+				m.makeMirroringDaemonContainer(daemonName),
+			},
+			RestartPolicy: v1.RestartPolicyAlways,
+			Volumes:       opspec.PodVolumes(""),
+			HostNetwork:   m.hostNetwork,
+		},
+	}
+	if m.hostNetwork {
+		podSpec.Spec.DNSPolicy = v1.DNSClusterFirstWithHostNet
+	}
+	m.placement.ApplyToPodSpec(&podSpec.Spec)
+
+	replicas := int32(1)
+	d := &extensions.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      resourceName,
+			Namespace: m.Namespace,
+		},
+		Spec: extensions.DeploymentSpec{Template: podSpec, Replicas: &replicas},
+	}
+	k8sutil.SetOwnerRef(m.context.Clientset, m.Namespace, &d.ObjectMeta, &m.ownerRef)
+	return d
+}
+
+func (m *Mirroring) makeConfigInitContainer(resourceName, daemonName string) v1.Container {
+	return v1.Container{
+		Name: opspec.ConfigInitContainerName,
+		Args: []string{
+			"ceph",
+			"config-init",
+		},
+		Image: k8sutil.MakeRookImage(m.rookVersion),
+		Env: []v1.EnvVar{
+			{Name: "ROOK_USERNAME", Value: fullDaemonName(daemonName)},
+			{Name: "ROOK_KEYRING",
+				ValueFrom: &v1.EnvVarSource{
+					SecretKeyRef: &v1.SecretKeySelector{
+						LocalObjectReference: v1.LocalObjectReference{Name: resourceName},
+						Key:                  opspec.KeyringSecretKeyName,
+					}}},
+			k8sutil.PodIPEnvVar(k8sutil.PrivateIPEnvVar),
+			k8sutil.PodIPEnvVar(k8sutil.PublicIPEnvVar),
+			opmon.EndpointEnvVar(),
+			k8sutil.ConfigOverrideEnvVar(),
+		},
+		VolumeMounts: opspec.RookVolumeMounts(),
+		Resources:    m.resources,
+	}
+}
+
+func (m *Mirroring) makeMirroringDaemonContainer(daemonName string) v1.Container {
+	container := v1.Container{
+		Name: "rbdmirror",
+		Command: []string{
+			"rbd-mirror",
+		},
+		Args: []string{
+			"--foreground",
+			"-n", fullDaemonName(daemonName),
+			"--conf", "/etc/ceph/ceph.conf",
+			"--keyring", "/etc/ceph/keyring",
+		},
+		Image:        m.cephVersion.Image,
+		VolumeMounts: opspec.CephVolumeMounts(),
+		Env:          k8sutil.ClusterDaemonEnvVars(),
+		Resources:    m.resources,
+	}
+	return container
+}

pkg/operator/ceph/cluster/rbd/spec_test.go

@@ -0,0 +1,55 @@
+/*
+Copyright 2016 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 rbd
+
+import (
+	"testing"
+
+	cephv1beta1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1beta1"
+	rookalpha "github.com/rook/rook/pkg/apis/rook.io/v1alpha2"
+	"github.com/rook/rook/pkg/clusterd"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestPodSpec(t *testing.T) {
+	c := New(&clusterd.Context{},
+		"ns",
+		"rook/rook:myversion",
+		cephv1beta1.CephVersionSpec{Image: "ceph/ceph:myceph"},
+		rookalpha.Placement{},
+		false,
+		cephv1beta1.RBDMirroringSpec{Workers: 2},
+		v1.ResourceRequirements{},
+		metav1.OwnerReference{},
+	)
+
+	d := c.makeDeployment("rname", "dname")
+	assert.Equal(t, "rname", d.Name)
+	spec := d.Spec.Template.Spec
+	require.Equal(t, 1, len(spec.Containers))
+	assert.Equal(t, 1, len(spec.InitContainers))
+	assert.Equal(t, 3, len(spec.Volumes))
+	cont := spec.Containers[0]
+	assert.Equal(t, "rbd-mirror", cont.Command[0])
+	assert.Equal(t, 7, len(cont.Args))
+	assert.Equal(t, "--foreground", cont.Args[0])
+	assert.Equal(t, "-n", cont.Args[1])
+	assert.Equal(t, "client.rbd-mirror.dname", cont.Args[2])
+}

pkg/operator/ceph/spec/keyring.go

@@ -0,0 +1,80 @@
+/*
+Copyright 2018 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 spec provides Kubernetes controller/pod/container spec items used for many Ceph daemons
+package spec
+
+import (
+	"fmt"
+
+	"github.com/rook/rook/pkg/clusterd"
+	"github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/operator/k8sutil"
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+	KeyringSecretKeyName = "keyring"
+)
+
+type KeyringConfig struct {
+	Namespace    string
+	ResourceName string
+	DaemonName   string
+	OwnerRef     metav1.OwnerReference
+	Username     string
+	Access       []string
+}
+
+func CreateKeyring(context *clusterd.Context, config KeyringConfig) error {
+	_, err := context.Clientset.CoreV1().Secrets(config.Namespace).Get(config.ResourceName, metav1.GetOptions{})
+	if err == nil {
+		logger.Infof("the keyring %s was already generated", config.ResourceName)
+		return nil
+	}
+	if !errors.IsNotFound(err) {
+		return fmt.Errorf("failed to get secret %s. %+v", config.ResourceName, err)
+	}
+
+	// get-or-create-key for the user account
+	keyring, err := client.AuthGetOrCreateKey(context, config.Namespace, config.Username, config.Access)
+	if err != nil {
+		return fmt.Errorf("failed to get or create auth key for %s. %+v", config.Username, err)
+	}
+
+	// Store the keyring in a secret
+	secrets := map[string]string{
+		KeyringSecretKeyName: keyring,
+	}
+	secret := &v1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      config.ResourceName,
+			Namespace: config.Namespace,
+		},
+		StringData: secrets,
+		Type:       k8sutil.RookType,
+	}
+	k8sutil.SetOwnerRef(context.Clientset, config.Namespace, &secret.ObjectMeta, &config.OwnerRef)
+
+	_, err = context.Clientset.CoreV1().Secrets(config.Namespace).Create(secret)
+	if err != nil {
+		return fmt.Errorf("failed to save mirroring secret. %+v", err)
+	}
+
+	return nil
+}

pkg/operator/ceph/spec/spec.go

@@ -18,6 +18,7 @@ limitations under the License.
 package spec
 
 import (
+	"github.com/coreos/pkg/capnslog"
 	cephconfig "github.com/rook/rook/pkg/daemon/ceph/config"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	"k8s.io/api/core/v1"
@@ -29,6 +30,8 @@ const (
 	ConfigInitContainerName = "config-init"
 )
 
+var logger = capnslog.NewPackageLogger("github.com/rook/rook", "ceph-spec")
+
 // PodVolumes fills in the volumes parameter with the common list of Kubernetes volumes for use in Ceph pods.
 func PodVolumes(dataDirHostPath string) []v1.Volume {
 	dataDirSource := v1.VolumeSource{EmptyDir: &v1.EmptyDirVolumeSource{}}

tests/framework/installer/ceph_installer.go

@@ -188,12 +188,13 @@ func (h *CephInstaller) CreateK8sRookToolbox(namespace string) (err error) {
 func (h *CephInstaller) CreateK8sRookCluster(namespace, systemNamespace string, storeType string) (err error) {
 	return h.CreateK8sRookClusterWithHostPathAndDevices(namespace, systemNamespace, storeType, false,
 		cephv1beta1.MonSpec{Count: 3, AllowMultiplePerNode: true}, true, /* startWithAllNodes */
+		1, /* rbd workers */
 		LuminousVersion)
 }
 
 //CreateK8sRookCluster creates rook cluster via kubectl
 func (h *CephInstaller) CreateK8sRookClusterWithHostPathAndDevices(namespace, systemNamespace, storeType string,
-	useAllDevices bool, mon cephv1beta1.MonSpec, startWithAllNodes bool, cephVersion cephv1beta1.CephVersionSpec) error {
+	useAllDevices bool, mon cephv1beta1.MonSpec, startWithAllNodes bool, rbdMirrorWorkers int, cephVersion cephv1beta1.CephVersionSpec) error {
 
 	dataDirHostPath, err := h.initTestDir(namespace)
 	if err != nil {
@@ -216,7 +217,7 @@ func (h *CephInstaller) CreateK8sRookClusterWithHostPathAndDevices(namespace, sy
 	}
 
 	logger.Infof("Starting Rook Cluster with yaml")
-	settings := &ClusterSettings{namespace, storeType, dataDirHostPath, useAllDevices, mon.Count, cephVersion}
+	settings := &ClusterSettings{namespace, storeType, dataDirHostPath, useAllDevices, mon.Count, rbdMirrorWorkers, cephVersion}
 	rookCluster := h.Manifests.GetRookCluster(settings)
 	if _, err := h.k8shelper.KubectlWithStdin(rookCluster, createFromStdinArgs...); err != nil {
 		return fmt.Errorf("Failed to create rook cluster : %v ", err)
@@ -230,6 +231,12 @@ func (h *CephInstaller) CreateK8sRookClusterWithHostPathAndDevices(namespace, sy
 		return err
 	}
 
+	if rbdMirrorWorkers > 0 {
+		if err := h.k8shelper.WaitForPodCount("app=rook-ceph-rbd-mirror", namespace, rbdMirrorWorkers); err != nil {
+			return err
+		}
+	}
+
 	logger.Infof("Rook Cluster started")
 	err = h.k8shelper.WaitForLabeledPodsToRun("app=rook-ceph-osd", namespace)
 	return err
@@ -272,7 +279,7 @@ func (h *CephInstaller) GetNodeHostnames() ([]string, error) {
 
 //InstallRookOnK8sWithHostPathAndDevices installs rook on k8s
 func (h *CephInstaller) InstallRookOnK8sWithHostPathAndDevices(namespace, storeType string,
-	helmInstalled, useDevices bool, mon cephv1beta1.MonSpec, startWithAllNodes bool) (bool, error) {
+	helmInstalled, useDevices bool, mon cephv1beta1.MonSpec, startWithAllNodes bool, rbdMirrorWorkers int) (bool, error) {
 
 	var err error
 	//flag used for local debuggin purpose, when rook is pre-installed
@@ -321,6 +328,7 @@ func (h *CephInstaller) InstallRookOnK8sWithHostPathAndDevices(namespace, storeT
 	//Create rook cluster
 	err = h.CreateK8sRookClusterWithHostPathAndDevices(namespace, onamespace, storeType,
 		useDevices, cephv1beta1.MonSpec{Count: mon.Count, AllowMultiplePerNode: mon.AllowMultiplePerNode}, startWithAllNodes,
+		rbdMirrorWorkers,
 		h.cephVersion)
 	if err != nil {
 		logger.Errorf("Rook cluster %s not installed, error -> %v", namespace, err)

tests/framework/installer/ceph_manifests.go

@@ -42,12 +42,13 @@ type CephManifests interface {
 }
 
 type ClusterSettings struct {
-	Namespace       string
-	StoreType       string
-	DataDirHostPath string
-	UseAllDevices   bool
-	Mons            int
-	CephVersion     cephv1beta1.CephVersionSpec
+	Namespace        string
+	StoreType        string
+	DataDirHostPath  string
+	UseAllDevices    bool
+	Mons             int
+	RBDMirrorWorkers int
+	CephVersion      cephv1beta1.CephVersionSpec
 }
 
 //CephManifestsMaster wraps rook yaml definitions
@@ -451,6 +452,8 @@ spec:
     allowMultiplePerNode: true
   dashboard:
     enabled: true
+  rbdMirroring:
+    workers: ` + strconv.Itoa(settings.RBDMirrorWorkers) + `
   metadataDevice:
   storage:
     useAllNodes: true

tests/integration/base_deploy_test.go

@@ -88,25 +88,26 @@ func HandlePanics(r interface{}, op installer.TestSuite, t func() *testing.T) {
 
 //TestCluster struct for handling panic and test suite tear down
 type TestCluster struct {
-	installer     *installer.CephInstaller
-	kh            *utils.K8sHelper
-	helper        *clients.TestClient
-	T             func() *testing.T
-	namespace     string
-	storeType     string
-	helmInstalled bool
-	useDevices    bool
-	mons          int
+	installer        *installer.CephInstaller
+	kh               *utils.K8sHelper
+	helper           *clients.TestClient
+	T                func() *testing.T
+	namespace        string
+	storeType        string
+	helmInstalled    bool
+	useDevices       bool
+	mons             int
+	rbdMirrorWorkers int
 }
 
 // StartTestCluster creates new instance of TestCluster struct
-func StartTestCluster(t func() *testing.T, namespace, storeType string, helmInstalled, useDevices bool, mons int, rookVersion string, cephVersion cephv1beta1.CephVersionSpec) (*TestCluster, *utils.K8sHelper) {
+func StartTestCluster(t func() *testing.T, namespace, storeType string, helmInstalled, useDevices bool, mons, rbdMirrorWorkers int, rookVersion string, cephVersion cephv1beta1.CephVersionSpec) (*TestCluster, *utils.K8sHelper) {
 	kh, err := utils.CreateK8sHelper(t)
 	require.NoError(t(), err)
 
 	i := installer.NewCephInstaller(t, kh.Clientset, rookVersion, cephVersion)
 
-	op := &TestCluster{i, kh, nil, t, namespace, storeType, helmInstalled, useDevices, mons}
+	op := &TestCluster{i, kh, nil, t, namespace, storeType, helmInstalled, useDevices, mons, rbdMirrorWorkers}
 
 	if rookVersion != installer.VersionMaster {
 		// make sure we have the images from a previous release locally so the test doesn't hit a timeout
@@ -122,7 +123,7 @@ func StartTestCluster(t func() *testing.T, namespace, storeType string, helmInst
 //SetUpRook is a wrapper for setting up rook
 func (op *TestCluster) Setup() {
 	isRookInstalled, err := op.installer.InstallRookOnK8sWithHostPathAndDevices(op.namespace, op.storeType,
-		op.helmInstalled, op.useDevices, cephv1beta1.MonSpec{Count: op.mons, AllowMultiplePerNode: true}, false /* startWithAllNodes */)
+		op.helmInstalled, op.useDevices, cephv1beta1.MonSpec{Count: op.mons, AllowMultiplePerNode: true}, false /* startWithAllNodes */, op.rbdMirrorWorkers)
 
 	if !isRookInstalled || err != nil {
 		logger.Errorf("Rook was not installed successfully: %v", err)

tests/integration/block_create_test.go

@@ -54,7 +54,9 @@ func (s *BlockCreateSuite) SetupSuite() {
 
 	var err error
 	s.namespace = "block-k8s-ns"
-	s.op, s.kh = StartTestCluster(s.T, s.namespace, "bluestore", false, false, 1, installer.VersionMaster, installer.MimicVersion)
+	mons := 1
+	rbdMirrorWorkers := 1
+	s.op, s.kh = StartTestCluster(s.T, s.namespace, "bluestore", false, false, mons, rbdMirrorWorkers, installer.VersionMaster, installer.MimicVersion)
 	s.testClient = clients.CreateTestClient(s.kh, s.op.installer.Manifests)
 	initialBlocks, err := s.testClient.BlockClient.List(s.namespace)
 	assert.Nil(s.T(), err)

tests/integration/helm_test.go

@@ -63,7 +63,9 @@ type HelmSuite struct {
 
 func (hs *HelmSuite) SetupSuite() {
 	hs.namespace = "helm-ns"
-	hs.op, hs.kh = StartTestCluster(hs.T, hs.namespace, "bluestore", true, false, 1, installer.VersionMaster, installer.MimicVersion)
+	mons := 1
+	rbdMirrorWorkers := 1
+	hs.op, hs.kh = StartTestCluster(hs.T, hs.namespace, "bluestore", true, false, mons, rbdMirrorWorkers, installer.VersionMaster, installer.MimicVersion)
 	hs.helper = clients.CreateTestClient(hs.kh, hs.op.installer.Manifests)
 }
 

tests/integration/smoke_test.go

@@ -76,7 +76,9 @@ type SmokeSuite struct {
 func (suite *SmokeSuite) SetupSuite() {
 	suite.namespace = "smoke-ns"
 	useDevices := true
-	suite.op, suite.k8sh = StartTestCluster(suite.T, suite.namespace, "bluestore", false, useDevices, 3, installer.VersionMaster, installer.MimicVersion)
+	mons := 3
+	rbdMirrorWorkers := 1
+	suite.op, suite.k8sh = StartTestCluster(suite.T, suite.namespace, "bluestore", false, useDevices, mons, rbdMirrorWorkers, installer.VersionMaster, installer.MimicVersion)
 	suite.helper = clients.CreateTestClient(suite.k8sh, suite.op.installer.Manifests)
 }
 

tests/integration/upgrade_test.go

@@ -61,8 +61,9 @@ type UpgradeSuite struct {
 func (s *UpgradeSuite) SetupSuite() {
 	s.namespace = "upgrade-ns"
 	useDevices := true
-
-	s.op, s.k8sh = StartTestCluster(s.T, s.namespace, "bluestore", false, useDevices, 3, installer.Version0_8, installer.LuminousVersion)
+	mons := 3
+	rbdMirrorWorkers := 0
+	s.op, s.k8sh = StartTestCluster(s.T, s.namespace, "bluestore", false, useDevices, mons, rbdMirrorWorkers, installer.Version0_8, installer.LuminousVersion)
 	s.helper = clients.CreateTestClient(s.k8sh, s.op.installer.Manifests)
 }
 

tests/integration/zblock_mount_unmount_test.go

@@ -86,7 +86,9 @@ func (s *BlockMountUnMountSuite) SetupSuite() {
 	s.pvcNameRWX = "block-persistent-rwx"
 	useHelm := false
 	useDevices := true
-	s.op, s.kh = StartTestCluster(s.T, s.namespace, "filestore", useHelm, useDevices, 1, installer.VersionMaster, installer.LuminousVersion)
+	mons := 1
+	rbdMirrorWorkers := 1
+	s.op, s.kh = StartTestCluster(s.T, s.namespace, "filestore", useHelm, useDevices, mons, rbdMirrorWorkers, installer.VersionMaster, installer.LuminousVersion)
 	s.testClient = clients.CreateTestClient(s.kh, s.op.installer.Manifests)
 	s.bc = s.testClient.BlockClient
 }

tests/longhaul/base_block_test.go

@@ -160,7 +160,8 @@ func (o LoadTestCluster) Setup() {
 	if !o.kh.IsRookInstalled(o.namespace) {
 		isRookInstalled, err := o.installer.InstallRookOnK8sWithHostPathAndDevices(o.namespace, "bluestore",
 			false, true, cephv1beta1.MonSpec{Count: 3, AllowMultiplePerNode: true},
-			true /* startWithAllNodes */)
+			true, /* startWithAllNodes */
+			1 /*rbd mirror workers*/)
 		require.NoError(o.T(), err)
 		require.True(o.T(), isRookInstalled)
 	}