Repository: rook/rook
Pull request #8379 (2021-07-26 10:06:26)

ceph: add an rbd-mirror bootstrap token on cluster creation 
Signed-off-by: Sébastien Han <seb@redhat.com>

<!-- Please take a look at our [Contributing](https://rook.io/docs/rook/master/development-flow.html)
documentation before submitting a Pull Request!
Thank you for contributing to Rook! -->

**Description of your changes:**

**Which issue is resolved by this Pull Request:**
Resolves #

**Checklist:**

- [ ] **Commit Message Formatting**: Commit titles and messages follow guidelines in the [developer guide](https://rook.io/docs/rook/master/development-flow.html#commit-structure).
- [ ] **Skip Tests for Docs**: Add the flag for skipping the build if this is only a documentation change. See [here](https://github.com/rook/rook/blob/master/INSTALL.md#skip-ci) for the flag.
- [ ] **Skip Unrelated Tests**: Add a flag to run tests for a specific storage provider. See [test options](https://github.com/rook/rook/blob/master/INSTALL.md#test-storage-provider).
- [ ] Reviewed the developer guide on [Submitting a Pull Request](https://rook.io/docs/rook/master/development-flow.html#submitting-a-pull-request)
- [ ] Documentation has been updated, if necessary.
- [ ] Unit tests have been added, if necessary.
- [ ] Integration tests have been added, if necessary.
- [ ] Pending release notes updated with breaking and/or notable changes, if necessary.
- [ ] Upgrade from previous release is tested and upgrade user guide is updated, if necessary.
- [ ] Code generation (`make codegen`) has been run to update object specifications, if necessary.


leseb: This depends on https://github.com/rook/rook/pull/8380 since the Secret does not contain the pool name and the "mirror pool peer bootstrap import" requires it. Once merged, the pool will be part of CephBlockPool.


Patch:
diff --git a/Documentation/ceph-teardown.md b/Documentation/ceph-teardown.md
index 58e0ac02f..db3f8dc0a 100644
--- a/Documentation/ceph-teardown.md
+++ b/Documentation/ceph-teardown.md
@@ -153,7 +153,7 @@ If for some reason the operator is not able to remove the finalizer (ie. the ope
 ```console
 for CRD in $(kubectl get crd -n rook-ceph | awk '/ceph.rook.io/ {print $1}'); do
     kubectl get -n rook-ceph "$CRD" -o name | \
-    xargs -I {} kubectl patch -n rook-ceph {} --type merge -p '{"metadata":{"finalizers": [null]}}'
+    xargs -I {} kubectl patch {} --type merge -p '{"metadata":{"finalizers": [null]}}'
 done
 ```
 
diff --git a/Documentation/ceph-upgrade.md b/Documentation/ceph-upgrade.md
index 6437bf6e7..6c87c0ac3 100644
--- a/Documentation/ceph-upgrade.md
+++ b/Documentation/ceph-upgrade.md
@@ -265,6 +265,10 @@ if applicable.
 
 Let's get started!
 
+> **IMPORTANT** If your CephCluster has specified `driveGroups` in the spec, you must follow the
+> instructions to [migrate the Drive Group spec](#migrate-the-drive-group-spec) before performing
+> any of the upgrade steps below.
+
 ### **1. Update common resources and CRDs**
 
 > Automatically updated if you are upgrading via the helm chart
@@ -298,11 +302,21 @@ Then apply the latest changes.
 kubectl apply -f common.yaml -f crds.yaml
 ```
 
-#### **Updates for optional resources**
+> **NOTE:** If your Rook-Ceph cluster was initially installed with rook v1.4 or lower, the above
+> command will return errors due to updates from Kubernetes' v1beta1 Custom Resource Definitions.
+> The error will contain text similar to `... spec.preserveUnknownFields: Invalid value...`.
+
+If you experience this error applying the latest changes to CRDs, use `kubectl`'s `replace` command
+to replace the resources followed by `apply` to verify that the resources are updated without other
+errors.
+```sh
+kubectl replace -f crds.yaml
+kubectl apply -f crds.yaml
+```
 
+#### **Updates for optional resources**
 If you have [Prometheus monitoring](ceph-monitoring.md) enabled, follow the
 step to upgrade the Prometheus RBAC resources as well.
-
 ```sh
 kubectl apply -f cluster/examples/kubernetes/ceph/monitoring/rbac.yaml
 ```
@@ -397,9 +411,7 @@ until all the daemons have been updated.
 
 ### **Ceph images**
 
-Official Ceph container images can be found on [Quay](https://quay.io/repository/ceph/ceph?tab=tags).
-Prior to August 2021, official images were on docker.io. While those images will remain on Docker Hub, all new images are being pushed to Quay.
-
+Official Ceph container images can be found on [Docker Hub](https://hub.docker.com/r/ceph/ceph/tags/).
 These images are tagged in a few ways:
 
 * The most explicit form of tags are full-ceph-version-and-build tags (e.g., `v16.2.5-20210708`).
@@ -410,7 +422,7 @@ These images are tagged in a few ways:
 
 **Ceph containers other than the official images from the registry above will not be supported.**
 
-### **Example upgrade to Ceph Pacific**
+### **Example upgrade to Ceph Octopus**
 
 #### **1. Update the main Ceph daemons**
 
@@ -465,7 +477,7 @@ kubectl -n $ROOK_OPERATOR_NAMESPACE edit configmap rook-ceph-operator-config
 The default upstream images are included below, which you can change to your desired images.
 
 ```yaml
-ROOK_CSI_CEPH_IMAGE: "quay.io/cephcsi/cephcsi:v3.4.0"
+ROOK_CSI_CEPH_IMAGE: "quay.io/cephcsi/cephcsi:v3.3.1"
 ROOK_CSI_REGISTRAR_IMAGE: "k8s.gcr.io/sig-storage/csi-node-driver-registrar:v2.2.0"
 ROOK_CSI_PROVISIONER_IMAGE: "k8s.gcr.io/sig-storage/csi-provisioner:v2.2.2"
 ROOK_CSI_ATTACHER_IMAGE: "k8s.gcr.io/sig-storage/csi-attacher:v3.2.1"
@@ -495,6 +507,6 @@ k8s.gcr.io/sig-storage/csi-node-driver-registrar:v2.2.0
 k8s.gcr.io/sig-storage/csi-provisioner:v2.2.2
 k8s.gcr.io/sig-storage/csi-resizer:v1.2.0
 k8s.gcr.io/sig-storage/csi-snapshotter:v4.1.1
-quay.io/cephcsi/cephcsi:v3.4.0
+quay.io/cephcsi/cephcsi:v3.3.1
 quay.io/csiaddons/volumereplication-operator:v0.1.0
 ```
diff --git a/Documentation/helm-operator.md b/Documentation/helm-operator.md
index b512430d0..2812a1b63 100644
--- a/Documentation/helm-operator.md
+++ b/Documentation/helm-operator.md
@@ -132,7 +132,7 @@ The following tables lists the configurable parameters of the rook-operator char
 | `csi.rbdLivenessMetricsPort`        | Ceph CSI RBD driver metrics port.                                                                                           | `8080`                                                    |
 | `csi.forceCephFSKernelClient`       | Enable Ceph Kernel clients on kernel < 4.17 which support quotas for Cephfs.                                                | `true`                                                    |
 | `csi.kubeletDirPath`                | Kubelet root directory path (if the Kubelet uses a different path for the `--root-dir` flag)                                | `/var/lib/kubelet`                                        |
-| `csi.cephcsi.image`                 | Ceph CSI image.                                                                                                             | `quay.io/cephcsi/cephcsi:v3.4.0`                          |
+| `csi.cephcsi.image`                 | Ceph CSI image.                                                                                                             | `quay.io/cephcsi/cephcsi:v3.3.1`                          |
 | `csi.rbdPluginUpdateStrategy`       | CSI Rbd plugin daemonset update strategy, supported values are OnDelete and RollingUpdate.                                  | `OnDelete`                                                |
 | `csi.cephFSPluginUpdateStrategy`    | CSI CephFS plugin daemonset update strategy, supported values are OnDelete and RollingUpdate.                               | `OnDelete`                                                |
 | `csi.registrar.image`               | Kubernetes CSI registrar image.                                                                                             | `k8s.gcr.io/sig-storage/csi-node-driver-registrar:v2.2.0` |
diff --git a/PendingReleaseNotes.md b/PendingReleaseNotes.md
index ec2aabaeb..0069361af 100644
--- a/PendingReleaseNotes.md
+++ b/PendingReleaseNotes.md
@@ -41,7 +41,6 @@ So the CephCLuster spec field `image` must be updated to point to quay, like `im
 - Stretch clusters are considered stable
   - Ceph v16.2.5 or greater is required for stretch clusters
 - The use of peer secret names in CephRBDMirror is deprecated. Please use CephBlockPool CR to configure peer secret names and import peers. Checkout the `mirroring` section in the CephBlockPool [spec](Documentation/ceph-pool-crd.md#spec) for more details.
-- Update Ceph CSI to `v3.4.0` for more details read the [official release note](https://github.com/ceph/ceph-csi/releases/tag/v3.4.0)
 
 ### Cassandra
 
diff --git a/cluster/charts/rook-ceph/templates/clusterrole.yaml b/cluster/charts/rook-ceph/templates/clusterrole.yaml
index 2218ec33d..dc62d8fea 100644
--- a/cluster/charts/rook-ceph/templates/clusterrole.yaml
+++ b/cluster/charts/rook-ceph/templates/clusterrole.yaml
@@ -378,9 +378,6 @@ rules:
   - apiGroups: [""]
     resources: ["configmaps"]
     verbs: ["get", "list"]
-  - apiGroups: [""]
-    resources: ["serviceaccounts"]
-    verbs: ["get"]
 ---
 kind: ClusterRole
 apiVersion: rbac.authorization.k8s.io/v1
@@ -447,9 +444,6 @@ rules:
   - apiGroups: ["replication.storage.openshift.io"]
     resources: ["volumereplicationclasses/status"]
     verbs: ["get"]
-  - apiGroups: [""]
-    resources: ["serviceaccounts"]
-    verbs: ["get"]
 {{- end }}
 {{- if .Values.pspEnable }}
 ---
diff --git a/cluster/charts/rook-ceph/values.yaml b/cluster/charts/rook-ceph/values.yaml
index c7356f4f2..ef4221726 100644
--- a/cluster/charts/rook-ceph/values.yaml
+++ b/cluster/charts/rook-ceph/values.yaml
@@ -270,7 +270,7 @@ csi:
   #rbdLivenessMetricsPort: 9080
   #kubeletDirPath: /var/lib/kubelet
   #cephcsi:
-    #image: quay.io/cephcsi/cephcsi:v3.4.0
+    #image: quay.io/cephcsi/cephcsi:v3.3.1
   #registrar:
     #image: k8s.gcr.io/sig-storage/csi-node-driver-registrar:v2.2.0
   #provisioner:
diff --git a/cluster/examples/kubernetes/ceph/common.yaml b/cluster/examples/kubernetes/ceph/common.yaml
index d174d3f0d..8a3ad35d3 100644
--- a/cluster/examples/kubernetes/ceph/common.yaml
+++ b/cluster/examples/kubernetes/ceph/common.yaml
@@ -1072,9 +1072,6 @@ rules:
   - apiGroups: [""]
     resources: ["configmaps"]
     verbs: ["get", "list"]
-  - apiGroups: [""]
-    resources: ["serviceaccounts"]
-    verbs: ["get"]
 ---
 kind: ClusterRole
 apiVersion: rbac.authorization.k8s.io/v1
@@ -1141,9 +1138,6 @@ rules:
   - apiGroups: ["replication.storage.openshift.io"]
     resources: ["volumereplicationclasses/status"]
     verbs: ["get"]
-  - apiGroups: [""]
-    resources: ["serviceaccounts"]
-    verbs: ["get"]
 # OLM: END CSI RBD CLUSTER ROLE
 # OLM: BEGIN CSI RBD CLUSTER ROLEBINDING
 ---
diff --git a/cluster/examples/kubernetes/ceph/csi/rbd/storageclass-ec.yaml b/cluster/examples/kubernetes/ceph/csi/rbd/storageclass-ec.yaml
index c62507ffc..f49f2fa89 100644
--- a/cluster/examples/kubernetes/ceph/csi/rbd/storageclass-ec.yaml
+++ b/cluster/examples/kubernetes/ceph/csi/rbd/storageclass-ec.yaml
@@ -77,9 +77,9 @@ parameters:
   # will set default as `ext4`.
   csi.storage.k8s.io/fstype: ext4
 # uncomment the following to use rbd-nbd as mounter on supported nodes
-# **IMPORTANT**: CephCSI v3.4.0 onwards a volume healer functionality is added to reattach
-# the PVC to application pod if nodeplugin pod restart.
-# Its still in Alpha support. Therefore, this option is not recommended for production use.
+# **IMPORTANT**: If you are using rbd-nbd as the mounter, during upgrade you will be hit a ceph-csi
+# issue that causes the mount to be disconnected. You will need to follow special upgrade steps
+# to restart your application pods. Therefore, this option is not recommended.
 #mounter: rbd-nbd
 allowVolumeExpansion: true
 reclaimPolicy: Delete
diff --git a/cluster/examples/kubernetes/ceph/csi/rbd/storageclass.yaml b/cluster/examples/kubernetes/ceph/csi/rbd/storageclass.yaml
index 98ef451f7..8077139fc 100644
--- a/cluster/examples/kubernetes/ceph/csi/rbd/storageclass.yaml
+++ b/cluster/examples/kubernetes/ceph/csi/rbd/storageclass.yaml
@@ -66,9 +66,9 @@ parameters:
   # in hyperconverged settings where the volume is mounted on the same node as the osds.
   csi.storage.k8s.io/fstype: ext4
 # uncomment the following to use rbd-nbd as mounter on supported nodes
-# **IMPORTANT**: CephCSI v3.4.0 onwards a volume healer functionality is added to reattach
-# the PVC to application pod if nodeplugin pod restart.
-# Its still in Alpha support. Therefore, this option is not recommended for production use.
+# **IMPORTANT**: If you are using rbd-nbd as the mounter, during upgrade you will be hit a ceph-csi
+# issue that causes the mount to be disconnected. You will need to follow special upgrade steps
+# to restart your application pods. Therefore, this option is not recommended.
 #mounter: rbd-nbd
 allowVolumeExpansion: true
 reclaimPolicy: Delete
diff --git a/cluster/examples/kubernetes/ceph/csi/template/rbd/csi-rbdplugin.yaml b/cluster/examples/kubernetes/ceph/csi/template/rbd/csi-rbdplugin.yaml
index 5e83ac894..1d1c9bfe1 100644
--- a/cluster/examples/kubernetes/ceph/csi/template/rbd/csi-rbdplugin.yaml
+++ b/cluster/examples/kubernetes/ceph/csi/template/rbd/csi-rbdplugin.yaml
@@ -67,7 +67,6 @@ spec:
             - "--metricsport={{ .RBDGRPCMetricsPort }}"
             - "--metricspath=/metrics"
             - "--enablegrpcmetrics={{ .EnableCSIGRPCMetrics }}"
-            - "--stagingpath={{ .KubeletDirPath }}/plugins/kubernetes.io/csi/pv/"
           env:
             - name: POD_IP
               valueFrom:
diff --git a/cluster/examples/kubernetes/ceph/operator-openshift.yaml b/cluster/examples/kubernetes/ceph/operator-openshift.yaml
index 1171f7c95..4b06c3873 100644
--- a/cluster/examples/kubernetes/ceph/operator-openshift.yaml
+++ b/cluster/examples/kubernetes/ceph/operator-openshift.yaml
@@ -148,7 +148,7 @@ data:
   # The default version of CSI supported by Rook will be started. To change the version
   # of the CSI driver to something other than what is officially supported, change
   # these images to the desired release of the CSI driver.
-  # ROOK_CSI_CEPH_IMAGE: "quay.io/cephcsi/cephcsi:v3.4.0"
+  # ROOK_CSI_CEPH_IMAGE: "quay.io/cephcsi/cephcsi:v3.3.1"
   # ROOK_CSI_REGISTRAR_IMAGE: "k8s.gcr.io/sig-storage/csi-node-driver-registrar:v2.2.0"
   # ROOK_CSI_RESIZER_IMAGE: "k8s.gcr.io/sig-storage/csi-resizer:v1.2.0"
   # ROOK_CSI_PROVISIONER_IMAGE: "k8s.gcr.io/sig-storage/csi-provisioner:v2.2.2"
diff --git a/cluster/examples/kubernetes/ceph/operator.yaml b/cluster/examples/kubernetes/ceph/operator.yaml
index 02cbd7b07..9ddc8ffec 100644
--- a/cluster/examples/kubernetes/ceph/operator.yaml
+++ b/cluster/examples/kubernetes/ceph/operator.yaml
@@ -72,7 +72,7 @@ data:
   # The default version of CSI supported by Rook will be started. To change the version
   # of the CSI driver to something other than what is officially supported, change
   # these images to the desired release of the CSI driver.
-  # ROOK_CSI_CEPH_IMAGE: "quay.io/cephcsi/cephcsi:v3.4.0"
+  # ROOK_CSI_CEPH_IMAGE: "quay.io/cephcsi/cephcsi:v3.3.1"
   # ROOK_CSI_REGISTRAR_IMAGE: "k8s.gcr.io/sig-storage/csi-node-driver-registrar:v2.2.0"
   # ROOK_CSI_RESIZER_IMAGE: "k8s.gcr.io/sig-storage/csi-resizer:v1.2.0"
   # ROOK_CSI_PROVISIONER_IMAGE: "k8s.gcr.io/sig-storage/csi-provisioner:v2.2.2"
diff --git a/pkg/daemon/ceph/client/mirror.go b/pkg/daemon/ceph/client/mirror.go
index 290f875a6..bdc6ea01a 100644
--- a/pkg/daemon/ceph/client/mirror.go
+++ b/pkg/daemon/ceph/client/mirror.go
@@ -17,14 +17,33 @@ limitations under the License.
 package client
 
 import (
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"os"
+	"strings"
 
 	"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/util"
+)
+
+// PeerToken is the content of the peer token
+type PeerToken struct {
+	ClusterFSID string `json:"fsid"`
+	ClientID    string `json:"client_id"`
+	Key         string `json:"key"`
+	MonHost     string `json:"mon_host"`
+	// These fields are added by Rook and NOT part of the output of client.CreateRBDMirrorBootstrapPeer()
+	PoolID    int    `json:"pool_id"`
+	Namespace string `json:"namespace"`
+}
+
+var (
+	rbdMirrorPeerCaps      = []string{"mon", "profile rbd-mirror-peer", "osd", "profile rbd"}
+	rbdMirrorPeerKeyringID = "rbd-mirror-peer"
 )
 
 // ImportRBDMirrorBootstrapPeer add a mirror peer in the rbd-mirror configuration
@@ -32,6 +51,7 @@ func ImportRBDMirrorBootstrapPeer(context *clusterd.Context, clusterInfo *Cluste
 	logger.Infof("add rbd-mirror bootstrap peer token for pool %q", poolName)
 
 	// Token file
+	// TODO: use mktemp?
 	tokenFilePath := fmt.Sprintf("/tmp/rbd-mirror-token-%s", poolName)
 
 	// Write token into a file
@@ -307,3 +327,58 @@ func ListSnapshotSchedulesRecursively(context *clusterd.Context, clusterInfo *Cl
 	logger.Debugf("successfully recursively listed snapshot schedules for pool %q", poolName)
 	return snapshotSchedulesRecursive, nil
 }
+
+/* CreateRBDMirrorBootstrapPeerWithoutPool creates a bootstrap peer for the current cluster
+It creates the cephx user for the remote cluster to use with all the necessary details
+This function is handy on scenarios where no pools have been created yet but replication communication is required (connecting peers)
+It essentially sits above CreateRBDMirrorBootstrapPeer()
+and is a cluster-wide option in the scenario where all the pools will be mirrored to the same remote cluster
+
+So the scenario looks like:
+
+	1) Create the cephx ID on the source cluster
+
+	2) Enable a source pool for mirroring - at any time, we just don't know when
+	rbd --cluster site-a mirror pool enable image-pool image
+
+	3) Copy the key details over to the other cluster (non-ceph workflow)
+
+	4) Enable destination pool for mirroring
+	rbd --cluster site-b mirror pool enable image-pool image
+
+	5) Add the peer details to the destination pool
+
+	6) Repeat the steps flipping source and destination to enable
+	bi-directional mirroring
+*/
+func CreateRBDMirrorBootstrapPeerWithoutPool(context *clusterd.Context, clusterInfo *ClusterInfo) ([]byte, error) {
+	fullClientName := getQualifiedUser(rbdMirrorPeerKeyringID)
+	logger.Infof("create rbd-mirror bootstrap peer token %q", fullClientName)
+	key, err := AuthGetOrCreateKey(context, clusterInfo, fullClientName, rbdMirrorPeerCaps)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to create rbd-mirror peer key %q", fullClientName)
+	}
+	logger.Infof("successfully created rbd-mirror bootstrap peer token for cluster %q", clusterInfo.NamespacedName().Name)
+
+	mons := util.NewSet()
+	for _, mon := range clusterInfo.Monitors {
+		mons.Add(mon.Endpoint)
+	}
+
+	peerToken := PeerToken{
+		ClusterFSID: clusterInfo.FSID,
+		ClientID:    rbdMirrorPeerKeyringID,
+		Key:         key,
+		MonHost:     strings.Join(mons.ToSlice(), ","),
+		Namespace:   clusterInfo.Namespace,
+	}
+
+	// Marshal the Go type back to JSON
+	decodedTokenBackToJSON, err := json.Marshal(peerToken)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to encode peer token to json")
+	}
+
+	// Return the base64 encoded token
+	return []byte(base64.StdEncoding.EncodeToString(decodedTokenBackToJSON)), nil
+}
diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go
index 0f14b52ac..c6a72069f 100755
--- a/pkg/operator/ceph/cluster/cluster.go
+++ b/pkg/operator/ceph/cluster/cluster.go
@@ -214,6 +214,7 @@ func (c *ClusterController) initializeCluster(cluster *cluster) error {
 
 	// Populate ClusterInfo with the last value
 	cluster.mons.ClusterInfo = cluster.ClusterInfo
+	cluster.mons.ClusterInfo.SetName(c.namespacedName.Name)
 
 	// Start the monitoring if not already started
 	c.configureCephMonitoring(cluster, cluster.ClusterInfo)
@@ -572,5 +573,11 @@ func (c *cluster) postMonStartupActions() error {
 		}
 	}
 
+	// Create cluster-wide RBD bootstrap peer token
+	_, err = controller.CreateBootstrapPeerSecret(c.context, c.ClusterInfo, &cephv1.CephCluster{ObjectMeta: metav1.ObjectMeta{Name: c.namespacedName.Name, Namespace: c.Namespace}}, c.ownerInfo)
+	if err != nil {
+		return errors.Wrap(err, "failed to create cluster rbd bootstrap peer token")
+	}
+
 	return nil
 }
diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go
index 528ccd102..bf41cc1f8 100644
--- a/pkg/operator/ceph/cluster/controller.go
+++ b/pkg/operator/ceph/cluster/controller.go
@@ -367,6 +367,7 @@ func (c *ClusterController) reconcileCephCluster(clusterObj *cephv1.CephCluster,
 		// It's a new cluster so let's populate the struct
 		cluster = newCluster(clusterObj, c.context, c.csiConfigMutex, ownerInfo)
 	}
+	cluster.namespacedName = c.namespacedName
 
 	// Pass down the client to interact with Kubernetes objects
 	// This will be used later down by spec code to create objects like deployment, services etc
diff --git a/pkg/operator/ceph/cluster/mon/health.go b/pkg/operator/ceph/cluster/mon/health.go
index c817ecb4a..3cc774da2 100644
--- a/pkg/operator/ceph/cluster/mon/health.go
+++ b/pkg/operator/ceph/cluster/mon/health.go
@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	cephutil "github.com/rook/rook/pkg/daemon/ceph/util"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
@@ -497,6 +498,12 @@ func (c *Cluster) removeMon(daemonName string) error {
 		return errors.Wrapf(err, "failed to save mon config after failing over mon %s", daemonName)
 	}
 
+	// Update cluster-wide RBD bootstrap peer token since Monitors have changed
+	_, err := controller.CreateBootstrapPeerSecret(c.context, c.ClusterInfo, &cephv1.CephCluster{ObjectMeta: metav1.ObjectMeta{Name: c.ClusterInfo.NamespacedName().Name, Namespace: c.Namespace}}, c.ownerInfo)
+	if err != nil {
+		return errors.Wrap(err, "failed to update cluster rbd bootstrap peer token")
+	}
+
 	return nil
 }
 
diff --git a/pkg/operator/ceph/controller/mirror_peer.go b/pkg/operator/ceph/controller/mirror_peer.go
index 163bab20b..c33e7b798 100644
--- a/pkg/operator/ceph/controller/mirror_peer.go
+++ b/pkg/operator/ceph/controller/mirror_peer.go
@@ -18,7 +18,8 @@ limitations under the License.
 package controller
 
 import (
-	"context"
+	"encoding/base64"
+	"encoding/json"
 	"fmt"
 
 	"github.com/pkg/errors"
@@ -29,10 +30,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
@@ -41,25 +39,15 @@ const (
 	poolMirrorBoostrapPeerSecretName = "pool-peer-token"
 	// #nosec G101 since this is not leaking any hardcoded credentials, it's just the prefix of the secret name
 	fsMirrorBoostrapPeerSecretName = "fs-peer-token"
+	// #nosec G101 since this is not leaking any hardcoded credentials, it's just the prefix of the secret name
+	clusterMirrorBoostrapPeerSecretName = "cluster-peer-token"
 	// RBDMirrorBootstrapPeerSecretName #nosec G101 since this is not leaking any hardcoded credentials, it's just the prefix of the secret name
 	RBDMirrorBootstrapPeerSecretName = "rbdMirrorBootstrapPeerSecretName"
 	// FSMirrorBootstrapPeerSecretName #nosec G101 since this is not leaking any hardcoded credentials, it's just the prefix of the secret name
 	FSMirrorBootstrapPeerSecretName = "fsMirrorBootstrapPeerSecretName"
 )
 
-// PeerToken is the content of the peer token
-type PeerToken struct {
-	ClusterFSID string `json:"fsid"`
-	ClientID    string `json:"client_id"`
-	Key         string `json:"key"`
-	MonHost     string `json:"mon_host"`
-	// These fields are added by Rook and NOT part of the output of client.CreateRBDMirrorBootstrapPeer()
-	PoolID    int    `json:"pool_id"`
-	Namespace string `json:"namespace"`
-}
-
-func CreateBootstrapPeerSecret(ctx *clusterd.Context, clusterInfo *cephclient.ClusterInfo, object client.Object, namespacedName types.NamespacedName, scheme *runtime.Scheme) (reconcile.Result, error) {
-	context := context.TODO()
+func CreateBootstrapPeerSecret(ctx *clusterd.Context, clusterInfo *cephclient.ClusterInfo, object client.Object, ownerInfo *k8sutil.OwnerInfo) (reconcile.Result, error) {
 	var err error
 	var ns, name, daemonType string
 	var boostrapToken []byte
@@ -73,6 +61,29 @@ func CreateBootstrapPeerSecret(ctx *clusterd.Context, clusterInfo *cephclient.Cl
 		if err != nil {
 			return ImmediateRetryResult, errors.Wrapf(err, "failed to create %s-mirror bootstrap peer", daemonType)
 		}
+
+		// Add additional information to the peer token
+		boostrapToken, err = expandBootstrapPeerToken(ctx, clusterInfo, name, boostrapToken)
+		if err != nil {
+			return ImmediateRetryResult, errors.Wrap(err, "failed to add extra information to rbd-mirror bootstrap peer")
+		}
+
+	case *cephv1.CephCluster:
+		ns = objectType.Namespace
+		name = "" // We pass an empty name because this is not a pool
+		daemonType = "cluster-rbd"
+		// Create rbd mirror bootstrap peer token
+		boostrapToken, err = cephclient.CreateRBDMirrorBootstrapPeerWithoutPool(ctx, clusterInfo)
+		if err != nil {
+			return ImmediateRetryResult, errors.Wrapf(err, "failed to create %s-mirror bootstrap peer", daemonType)
+		}
+
+		// Add additional information to the peer token
+		boostrapToken, err = expandBootstrapPeerToken(ctx, clusterInfo, name, boostrapToken)
+		if err != nil {
+			return ImmediateRetryResult, errors.Wrap(err, "failed to add extra information to rbd-mirror bootstrap peer")
+		}
+
 	case *cephv1.CephFilesystem:
 		ns = objectType.Namespace
 		name = objectType.Name
@@ -81,6 +92,7 @@ func CreateBootstrapPeerSecret(ctx *clusterd.Context, clusterInfo *cephclient.Cl
 		if err != nil {
 			return ImmediateRetryResult, errors.Wrapf(err, "failed to create %s-mirror bootstrap peer", daemonType)
 		}
+
 	default:
 		return ImmediateRetryResult, errors.Wrap(err, "failed to create bootstrap peer unknown daemon type")
 	}
@@ -89,14 +101,14 @@ func CreateBootstrapPeerSecret(ctx *clusterd.Context, clusterInfo *cephclient.Cl
 	s := GenerateBootstrapPeerSecret(object, boostrapToken)
 
 	// set ownerref to the Secret
-	err = controllerutil.SetControllerReference(object, s, scheme)
+	err = ownerInfo.SetControllerReference(s)
 	if err != nil {
 		return ImmediateRetryResult, errors.Wrapf(err, "failed to set owner reference for %s-mirror bootstrap peer secret %q", daemonType, s.Name)
 	}
 
 	// Create Secret
-	logger.Debugf("store %s-mirror bootstrap token in a Kubernetes Secret %q", daemonType, s.Name)
-	_, err = ctx.Clientset.CoreV1().Secrets(ns).Create(context, s, metav1.CreateOptions{})
+	logger.Debugf("store %s-mirror bootstrap token in a Kubernetes Secret %q in namespace %q", daemonType, s.Name, ns)
+	_, err = k8sutil.CreateOrUpdateSecret(ctx.Clientset, s)
 	if err != nil && !kerrors.IsAlreadyExists(err) {
 		return ImmediateRetryResult, errors.Wrapf(err, "failed to create %s-mirror bootstrap peer %q secret", daemonType, s.Name)
 	}
@@ -117,6 +129,10 @@ func GenerateBootstrapPeerSecret(object client.Object, token []byte) *v1.Secret
 		entityType = "pool"
 		entityName = objectType.Name
 		entityNamespace = objectType.Namespace
+	case *cephv1.CephCluster:
+		entityType = "cluster"
+		entityName = objectType.Name
+		entityNamespace = objectType.Namespace
 	}
 
 	s := &v1.Secret{
@@ -140,6 +156,8 @@ func buildBoostrapPeerSecretName(object client.Object) string {
 		return fmt.Sprintf("%s-%s", fsMirrorBoostrapPeerSecretName, objectType.Name)
 	case *cephv1.CephBlockPool:
 		return fmt.Sprintf("%s-%s", poolMirrorBoostrapPeerSecretName, objectType.Name)
+	case *cephv1.CephCluster:
+		return fmt.Sprintf("%s-%s", clusterMirrorBoostrapPeerSecretName, objectType.Name)
 	}
 
 	return ""
@@ -166,7 +184,7 @@ func ValidatePeerToken(object client.Object, data map[string][]byte) error {
 	// Lookup Secret keys and content
 	keysToTest := []string{"token"}
 	switch object.(type) {
-	case *cephv1.CephBlockPool:
+	case *cephv1.CephRBDMirror:
 		keysToTest = append(keysToTest, "pool")
 	}
 
@@ -179,3 +197,40 @@ func ValidatePeerToken(object client.Object, data map[string][]byte) error {
 
 	return nil
 }
+
+func expandBootstrapPeerToken(ctx *clusterd.Context, clusterInfo *cephclient.ClusterInfo, poolName string, token []byte) ([]byte, error) {
+	// First decode the token, it's base64 encoded
+	decodedToken, err := base64.StdEncoding.DecodeString(string(token))
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to decode bootstrap peer token")
+	}
+
+	// Unmarshal the decoded value to a Go type
+	var decodedTokenToGo cephclient.PeerToken
+	err = json.Unmarshal(decodedToken, &decodedTokenToGo)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to unmarshal decoded token")
+	}
+
+	// Fetch the pool ID
+	if poolName != "" {
+		poolDetails, err := cephclient.GetPoolDetails(ctx, clusterInfo, poolName)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to get pool %q details", poolName)
+		}
+
+		// Add extra details to the token
+		decodedTokenToGo.PoolID = poolDetails.Number
+	}
+
+	decodedTokenToGo.Namespace = clusterInfo.Namespace
+
+	// Marshal the Go type back to JSON
+	decodedTokenBackToJSON, err := json.Marshal(decodedTokenToGo)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to encode go type back to json")
+	}
+
+	// Return the base64 encoded token
+	return []byte(base64.StdEncoding.EncodeToString(decodedTokenBackToJSON)), nil
+}
diff --git a/pkg/operator/ceph/csi/spec.go b/pkg/operator/ceph/csi/spec.go
index 9374380a7..573aa9d38 100644
--- a/pkg/operator/ceph/csi/spec.go
+++ b/pkg/operator/ceph/csi/spec.go
@@ -108,7 +108,7 @@ var (
 // manually challenging.
 var (
 	// image names
-	DefaultCSIPluginImage         = "quay.io/cephcsi/cephcsi:v3.4.0"
+	DefaultCSIPluginImage         = "quay.io/cephcsi/cephcsi:v3.3.1"
 	DefaultRegistrarImage         = "k8s.gcr.io/sig-storage/csi-node-driver-registrar:v2.2.0"
 	DefaultProvisionerImage       = "k8s.gcr.io/sig-storage/csi-provisioner:v2.2.2"
 	DefaultAttacherImage          = "k8s.gcr.io/sig-storage/csi-attacher:v3.2.1"
diff --git a/pkg/operator/ceph/csi/version.go b/pkg/operator/ceph/csi/version.go
index 685a488f2..8f978b02c 100644
--- a/pkg/operator/ceph/csi/version.go
+++ b/pkg/operator/ceph/csi/version.go
@@ -25,20 +25,15 @@ import (
 )
 
 var (
-	//minimum supported version is 3.0.0
-	minimum = CephCSIVersion{3, 0, 0}
+	//minimum supported version is 2.0.0
+	minimum = CephCSIVersion{2, 0, 0}
 	//supportedCSIVersions are versions that rook supports
+	releaseV210          = CephCSIVersion{2, 1, 0}
+	releasev300          = CephCSIVersion{3, 0, 0}
 	releasev310          = CephCSIVersion{3, 1, 0}
 	releasev320          = CephCSIVersion{3, 2, 0}
 	releasev330          = CephCSIVersion{3, 3, 0}
-	releasev340          = CephCSIVersion{3, 4, 0}
-	supportedCSIVersions = []CephCSIVersion{
-		minimum,
-		releasev310,
-		releasev320,
-		releasev330,
-		releasev340,
-	}
+	supportedCSIVersions = []CephCSIVersion{minimum, releaseV210, releasev300, releasev310, releasev320, releasev330}
 	// omap generator is supported in v3.2.0+
 	omapSupportedVersions = releasev320
 	// for parsing the output of `cephcsi`
diff --git a/pkg/operator/ceph/file/controller.go b/pkg/operator/ceph/file/controller.go
index 406de8527..ed6db4e48 100644
--- a/pkg/operator/ceph/file/controller.go
+++ b/pkg/operator/ceph/file/controller.go
@@ -308,7 +308,7 @@ func (r *ReconcileCephFilesystem) reconcile(request reconcile.Request) (reconcil
 
 				// Always create a bootstrap peer token in case another cluster wants to add us as a peer
 				logger.Info("reconciling create cephfs-mirror peer configuration")
-				reconcileResponse, err = opcontroller.CreateBootstrapPeerSecret(r.context, r.clusterInfo, cephFilesystem, request.NamespacedName, r.scheme)
+				reconcileResponse, err = opcontroller.CreateBootstrapPeerSecret(r.context, r.clusterInfo, cephFilesystem, k8sutil.NewOwnerInfo(cephFilesystem, r.scheme))
 				if err != nil {
 					updateStatus(r.client, request.NamespacedName, cephv1.ConditionFailure, nil)
 					return reconcileResponse, errors.Wrapf(err, "failed to create cephfs-mirror bootstrap peer for filesystem %q.", cephFilesystem.Name)
diff --git a/pkg/operator/ceph/pool/controller.go b/pkg/operator/ceph/pool/controller.go
index 98aaf6117..7a238b7f5 100644
--- a/pkg/operator/ceph/pool/controller.go
+++ b/pkg/operator/ceph/pool/controller.go
@@ -32,6 +32,7 @@ import (
 	"github.com/rook/rook/pkg/operator/ceph/cluster/mgr"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
+	"github.com/rook/rook/pkg/operator/k8sutil"
 	corev1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -282,7 +283,7 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
 	logger.Debug("reconciling create rbd mirror peer configuration")
 	if cephBlockPool.Spec.Mirroring.Enabled {
 		// Always create a bootstrap peer token in case another cluster wants to add us as a peer
-		reconcileResponse, err = opcontroller.CreateBootstrapPeerSecret(r.context, clusterInfo, cephBlockPool, request.NamespacedName, r.scheme)
+		reconcileResponse, err = opcontroller.CreateBootstrapPeerSecret(r.context, clusterInfo, cephBlockPool, k8sutil.NewOwnerInfo(cephBlockPool, r.scheme))
 		if err != nil {
 			updateStatus(r.client, request.NamespacedName, cephv1.ConditionFailure, nil)
 			return reconcileResponse, errors.Wrapf(err, "failed to create rbd-mirror bootstrap peer for pool %q.", cephBlockPool.GetName())
diff --git a/pkg/operator/ceph/pool/peers.go b/pkg/operator/ceph/pool/peers.go
index 79c34d268..496a4bcca 100644
--- a/pkg/operator/ceph/pool/peers.go
+++ b/pkg/operator/ceph/pool/peers.go
@@ -52,7 +52,7 @@ func (r *ReconcileCephBlockPool) reconcileAddBoostrapPeer(pool *cephv1.CephBlock
 		}
 
 		// Import bootstrap peer
-		err = client.ImportRBDMirrorBootstrapPeer(r.context, r.clusterInfo, string(s.Data["pool"]), string(s.Data["direction"]), s.Data["token"])
+		err = client.ImportRBDMirrorBootstrapPeer(r.context, r.clusterInfo, pool.Name, string(s.Data["direction"]), s.Data["token"])
 		if err != nil {
 			return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to import bootstrap peer token")
 		}


Test patch:
diff --git a/.github/workflows/canary-integration-test.yml b/.github/workflows/canary-integration-test.yml
index d2d89fcad..a72c33638 100644
--- a/.github/workflows/canary-integration-test.yml
+++ b/.github/workflows/canary-integration-test.yml
@@ -755,41 +755,77 @@ jobs:
         yq w -i pool-test.yaml spec.mirroring.enabled true
         yq w -i pool-test.yaml spec.mirroring.mode image
         kubectl create -f pool-test.yaml
-        timeout 60 sh -c 'until [ "$(kubectl -n rook-ceph get cephblockpool replicapool -o jsonpath='{.status.phase}'|grep -c "Ready")" -eq 1 ]; do echo "waiting for pool to created" && sleep 1; done'
+        timeout 60 sh -c 'until [ "$(kubectl -n rook-ceph get cephblockpool replicapool -o jsonpath='{.status.phase}'|grep -c "Ready")" -eq 1 ]; do echo "waiting for pool replicapool to created on cluster 1" && sleep 1; done'
+
+    - name: create replicated mirrored pool 2 on cluster 1
+      run: |
+        cd cluster/examples/kubernetes/ceph/
+        yq w -i pool-test.yaml metadata.name replicapool2
+        kubectl create -f pool-test.yaml
+        timeout 60 sh -c 'until [ "$(kubectl -n rook-ceph get cephblockpool replicapool2 -o jsonpath='{.status.phase}'|grep -c "Ready")" -eq 1 ]; do echo "waiting for pool replicapool2 to created on cluster 2" && sleep 1; done'
+        yq w -i pool-test.yaml metadata.name replicapool
 
     - name: create replicated mirrored pool on cluster 2
       run: |
         cd cluster/examples/kubernetes/ceph/
         yq w -i pool-test.yaml metadata.namespace rook-ceph-secondary
         kubectl create -f pool-test.yaml
-        timeout 60 sh -c 'until [ "$(kubectl -n rook-ceph-secondary get cephblockpool replicapool -o jsonpath='{.status.phase}'|grep -c "Ready")" -eq 1 ]; do echo "waiting for pool to created" && sleep 1; done'
+        timeout 60 sh -c 'until [ "$(kubectl -n rook-ceph-secondary get cephblockpool replicapool -o jsonpath='{.status.phase}'|grep -c "Ready")" -eq 1 ]; do echo "waiting for pool replicapool to created on cluster 1" && sleep 1; done'
+
+    - name: create replicated mirrored pool 2 on cluster 2
+      run: |
+        cd cluster/examples/kubernetes/ceph/
+        yq w -i pool-test.yaml metadata.name replicapool2
+        kubectl create -f pool-test.yaml
+        timeout 60 sh -c 'until [ "$(kubectl -n rook-ceph-secondary get cephblockpool replicapool -o jsonpath='{.status.phase}'|grep -c "Ready")" -eq 1 ]; do echo "waiting for pool replicapool2 to created on cluster 2" && sleep 1; done'
 
-    - name: create image in the pool
+    - name: create images in the pools
       run: |
         kubectl exec -n rook-ceph deploy/rook-ceph-tools -ti -- rbd -p replicapool create test -s 1G
         kubectl exec -n rook-ceph deploy/rook-ceph-tools -t -- rbd mirror image enable replicapool/test snapshot
         kubectl exec -n rook-ceph deploy/rook-ceph-tools -t -- rbd -p replicapool info test
+        kubectl exec -n rook-ceph deploy/rook-ceph-tools -ti -- rbd -p replicapool2 create test -s 1G
+        kubectl exec -n rook-ceph deploy/rook-ceph-tools -t -- rbd mirror image enable replicapool2/test snapshot
+        kubectl exec -n rook-ceph deploy/rook-ceph-tools -t -- rbd -p replicapool2 info test
 
-    - name: copy block mirror peer secret into the other cluster
+    - name: copy block mirror peer secret into the other cluster for replicapool
       run: |
         kubectl -n rook-ceph get secret pool-peer-token-replicapool -o yaml |\
         sed 's/namespace: rook-ceph/namespace: rook-ceph-secondary/g; s/name: pool-peer-token-replicapool/name: pool-peer-token-replicapool-config/g' |\
         kubectl create --namespace=rook-ceph-secondary -f -
 
-    - name: add block mirror peer secret to the other cluster
+    - name: copy block mirror peer secret into the other cluster for replicapool2 (using cluster global peer)
+      run: |
+        kubectl -n rook-ceph get secret cluster-peer-token-my-cluster -o yaml |\
+        sed 's/namespace: rook-ceph/namespace: rook-ceph-secondary/g; s/name: cluster-peer-token-my-cluster/name: cluster-peer-token-my-cluster-config/g' |\
+        kubectl create --namespace=rook-ceph-secondary -f -
+
+    - name: add block mirror peer secret to the other cluster for replicapool
       run: |
         kubectl -n rook-ceph-secondary patch cephblockpool replicapool --type merge -p '{"spec":{"mirroring":{"peers": {"secretNames": ["pool-peer-token-replicapool-config"]}}}}'
 
-    - name: verify image has been mirrored
+    - name: add block mirror peer secret to the other cluster for replicapool2 (using cluster global peer)
+      run: |
+        kubectl -n rook-ceph-secondary patch cephblockpool replicapool2 --type merge -p '{"spec":{"mirroring":{"peers": {"secretNames": ["cluster-peer-token-my-cluster-config"]}}}}'
+
+    - name: verify image has been mirrored for replicapool
+      run: |
+        # let's wait a bit for the image to be present
+        timeout 120 sh -c 'until [ "$(kubectl exec -n rook-ceph-secondary deploy/rook-ceph-tools -t -- rbd -p replicapool ls|grep -c test)" -eq 1 ]; do echo "waiting for image to be mirrored in pool replicapool" && sleep 1; done'
+
+    - name: verify image has been mirrored for replicapool2
       run: |
         # let's wait a bit for the image to be present
-        timeout 120 sh -c 'until [ "$(kubectl exec -n rook-ceph-secondary deploy/rook-ceph-tools -t -- rbd -p replicapool ls|grep -c test)" -eq 1 ]; do echo "waiting for image to be mirrored" && sleep 1; done'
+        timeout 120 sh -c 'until [ "$(kubectl exec -n rook-ceph-secondary deploy/rook-ceph-tools -t -- rbd -p replicapool2 ls|grep -c test)" -eq 1 ]; do echo "waiting for image to be mirrored in pool replicapool2" && sleep 1; done'
 
     - name: display cephblockpool and image status
       run: |
-        timeout 80 sh -c 'until [ "$(kubectl -n rook-ceph-secondary get cephblockpool replicapool -o jsonpath='{.status.mirroringStatus.summary.daemon_health}'|grep -c OK)" -eq 1 ]; do echo "waiting for mirroring status to be updated" && sleep 1; done'
-        kubectl -n rook-ceph-secondary get cephblockpool -o yaml
+        timeout 80 sh -c 'until [ "$(kubectl -n rook-ceph-secondary get cephblockpool replicapool -o jsonpath='{.status.mirroringStatus.summary.daemon_health}'|grep -c OK)" -eq 1 ]; do echo "waiting for mirroring status to be updated in replicapool" && sleep 1; done'
+        timeout 80 sh -c 'until [ "$(kubectl -n rook-ceph-secondary get cephblockpool replicapool2 -o jsonpath='{.status.mirroringStatus.summary.daemon_health}'|grep -c OK)" -eq 1 ]; do echo "waiting for mirroring status to be updated in replicapool2" && sleep 1; done'
+        kubectl -n rook-ceph-secondary get cephblockpool replicapool -o yaml
+        kubectl -n rook-ceph-secondary get cephblockpool replicapool2 -o yaml
         kubectl exec -n rook-ceph deploy/rook-ceph-tools -t -- rbd -p replicapool info test
+        kubectl exec -n rook-ceph deploy/rook-ceph-tools -t -- rbd -p replicapool2 info test
 
     - name: create replicated mirrored filesystem on cluster 1
       run: |
diff --git a/pkg/daemon/ceph/client/test/info.go b/pkg/daemon/ceph/client/test/info.go
index efb03c956..6efd21917 100644
--- a/pkg/daemon/ceph/client/test/info.go
+++ b/pkg/daemon/ceph/client/test/info.go
@@ -62,5 +62,6 @@ func CreateTestClusterInfo(monCount int) *client.ClusterInfo {
 			Endpoint: fmt.Sprintf("1.2.3.%d:6789", (i + 1)),
 		}
 	}
+	c.SetName(c.Namespace)
 	return c
 }
diff --git a/pkg/operator/ceph/cluster/mon/health_test.go b/pkg/operator/ceph/cluster/mon/health_test.go
index 77b4a1e21..1ed45b170 100644
--- a/pkg/operator/ceph/cluster/mon/health_test.go
+++ b/pkg/operator/ceph/cluster/mon/health_test.go
@@ -49,6 +49,10 @@ func TestCheckHealth(t *testing.T) {
 
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			logger.Infof("executing command: %s %+v", command, args)
+			if args[0] == "auth" && args[1] == "get-or-create-key" {
+				return "{\"key\":\"mysecurekey\"}", nil
+			}
 			return clienttest.MonInQuorumResponse(), nil
 		},
 	}
@@ -154,7 +158,13 @@ func TestEvictMonOnSameNode(t *testing.T) {
 	clientset := test.New(t, 1)
 	configDir, _ := ioutil.TempDir("", "")
 	defer os.RemoveAll(configDir)
-	context := &clusterd.Context{Clientset: clientset, ConfigDir: configDir, Executor: &exectest.MockExecutor{}, RequestCancelOrchestration: abool.New()}
+	executor := &exectest.MockExecutor{
+		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			logger.Infof("executing command: %s %+v", command, args)
+			return "{\"key\":\"mysecurekey\"}", nil
+		},
+	}
+	context := &clusterd.Context{Clientset: clientset, ConfigDir: configDir, Executor: executor, RequestCancelOrchestration: abool.New()}
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
 	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{})
 	setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 0}, "myversion")
@@ -245,6 +255,10 @@ func TestCheckHealthNotFound(t *testing.T) {
 
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			logger.Infof("executing command: %s %+v", command, args)
+			if args[0] == "auth" && args[1] == "get-or-create-key" {
+				return "{\"key\":\"mysecurekey\"}", nil
+			}
 			return clienttest.MonInQuorumResponse(), nil
 		},
 	}
@@ -304,6 +318,10 @@ func TestAddRemoveMons(t *testing.T) {
 	monQuorumResponse := clienttest.MonInQuorumResponse()
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			logger.Infof("executing command: %s %+v", command, args)
+			if args[0] == "auth" && args[1] == "get-or-create-key" {
+				return "{\"key\":\"mysecurekey\"}", nil
+			}
 			return monQuorumResponse, nil
 		},
 	}
diff --git a/pkg/operator/ceph/controller/mirror_peer_test.go b/pkg/operator/ceph/controller/mirror_peer_test.go
index fe4bb47fc..b12f70497 100644
--- a/pkg/operator/ceph/controller/mirror_peer_test.go
+++ b/pkg/operator/ceph/controller/mirror_peer_test.go
@@ -18,17 +18,22 @@ limitations under the License.
 package controller
 
 import (
+	"encoding/base64"
 	"reflect"
 	"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"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 func TestValidatePeerToken(t *testing.T) {
 	// Error: map is empty
-	b := &cephv1.CephBlockPool{}
+	b := &cephv1.CephRBDMirror{}
 	data := map[string][]byte{}
 	err := ValidatePeerToken(b, data)
 	assert.Error(t, err)
@@ -43,13 +48,18 @@ func TestValidatePeerToken(t *testing.T) {
 	err = ValidatePeerToken(b, data)
 	assert.Error(t, err)
 
-	// Success CephBlockPool
+	// Success CephRBDMirror
 	data["pool"] = []byte("foo")
 	err = ValidatePeerToken(b, data)
 	assert.NoError(t, err)
 
 	// Success CephFilesystem
-	data["pool"] = []byte("foo")
+	// "pool" is not required here
+	delete(data, "pool")
+	err = ValidatePeerToken(&cephv1.CephFilesystemMirror{}, data)
+	assert.NoError(t, err)
+
+	// Success CephFilesystem
 	err = ValidatePeerToken(&cephv1.CephFilesystemMirror{}, data)
 	assert.NoError(t, err)
 }
@@ -73,3 +83,25 @@ func TestGenerateStatusInfo(t *testing.T) {
 		})
 	}
 }
+
+func TestExpandBootstrapPeerToken(t *testing.T) {
+	executor := &exectest.MockExecutor{
+		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			if reflect.DeepEqual(args[0:5], []string{"osd", "pool", "get", "pool", "all"}) {
+				return `{"pool_id":13}`, nil
+			}
+
+			return "", errors.Errorf("unknown command args: %s", args[0:5])
+		},
+	}
+	c := &clusterd.Context{
+		Executor: executor,
+	}
+
+	newToken, err := expandBootstrapPeerToken(c, cephclient.AdminClusterInfo("mu-cluster"), "pool", []byte(`eyJmc2lkIjoiYzZiMDg3ZjItNzgyOS00ZGJiLWJjZmMtNTNkYzM0ZTBiMzVkIiwiY2xpZW50X2lkIjoicmJkLW1pcnJvci1wZWVyIiwia2V5IjoiQVFBV1lsWmZVQ1Q2RGhBQVBtVnAwbGtubDA5YVZWS3lyRVV1NEE9PSIsIm1vbl9ob3N0IjoiW3YyOjE5Mi4xNjguMTExLjEwOjMzMDAsdjE6MTkyLjE2OC4xMTEuMTA6Njc4OV0sW3YyOjE5Mi4xNjguMTExLjEyOjMzMDAsdjE6MTkyLjE2OC4xMTEuMTI6Njc4OV0sW3YyOjE5Mi4xNjguMTExLjExOjMzMDAsdjE6MTkyLjE2OC4xMTEuMTE6Njc4OV0ifQ==`))
+	assert.NoError(t, err)
+	newTokenDecoded, err := base64.StdEncoding.DecodeString(string(newToken))
+	assert.NoError(t, err)
+	assert.Contains(t, string(newTokenDecoded), "pool_id")
+	assert.Contains(t, string(newTokenDecoded), "namespace")
+}
diff --git a/pkg/operator/ceph/csi/version_test.go b/pkg/operator/ceph/csi/version_test.go
index f09dacf1a..fb7d05b99 100644
--- a/pkg/operator/ceph/csi/version_test.go
+++ b/pkg/operator/ceph/csi/version_test.go
@@ -29,7 +29,6 @@ var (
 	testReleaseV320        = CephCSIVersion{3, 2, 0}
 	testReleaseV321        = CephCSIVersion{3, 2, 1}
 	testReleaseV330        = CephCSIVersion{3, 3, 0}
-	testReleaseV340        = CephCSIVersion{3, 4, 0}
 	testVersionUnsupported = CephCSIVersion{4, 0, 0}
 )
 
@@ -83,11 +82,6 @@ func TestIsAtLeast(t *testing.T) {
 	ret = testReleaseV330.isAtLeast(&testReleaseV300)
 	assert.Equal(t, true, ret)
 
-	// Test for 3.4.0
-	// Test version which is lesser
-	ret = testReleaseV340.isAtLeast(&testReleaseV330)
-	assert.Equal(t, true, ret)
-
 	// Test version which is greater (minor)
 	version = CephCSIVersion{3, 1, 1}
 	ret = testReleaseV300.isAtLeast(&version)
@@ -102,13 +96,13 @@ func TestIsAtLeast(t *testing.T) {
 func TestSupported(t *testing.T) {
 	AllowUnsupported = false
 	ret := testMinVersion.Supported()
-	assert.Equal(t, false, ret)
+	assert.Equal(t, true, ret)
+
+	ret = testMinVersion.Supported()
+	assert.Equal(t, true, ret)
 
 	ret = testVersionUnsupported.Supported()
 	assert.Equal(t, false, ret)
-
-	ret = testReleaseV340.Supported()
-	assert.Equal(t, true, ret)
 }
 
 func TestSupportOMAPController(t *testing.T) {