Repository: rook/rook
Pull request #9643 (2022-01-24 23:12:08)

docs: add CSI section to Multus design doc
Add a section detailing a design to handle the volume-operations-blocked
on CSI restart issue when running with Multus.

See: https://github.com/rook/rook/issues/8085

Signed-off-by: Blaine Gardner <blaine.gardner@redhat.com>

<!-- Please take a look at our [Contributing](https://rook.io/docs/rook/latest/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/latest/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/latest/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.



Patch:
diff --git a/.github/workflows/daily-nightly-jobs.yml b/.github/workflows/daily-nightly-jobs.yml
index 042bb2cdb..7c7df6e28 100644
--- a/.github/workflows/daily-nightly-jobs.yml
+++ b/.github/workflows/daily-nightly-jobs.yml
@@ -324,16 +324,13 @@ jobs:
       with:
         fetch-depth: 0
 
-    - name: run encryption KMS IBM Key Protect
+    - name: run Encryption KMS IBM Key Protect
       uses: ./.github/workflows/encryption-pvc-kms-ibm-kp
-      if: "env.IBM_KP_SERVICE_INSTANCE_ID != '' && env.IBM_KP_SERVICE_API_KEY != ''"
-      env:
-        IBM_KP_SERVICE_INSTANCE_ID: ${{ secrets.IBM_INSTANCE_ID }}
-        IBM_KP_SERVICE_API_KEY: ${{ secrets.IBM_SERVICE_API_KEY }}
+      if: "env.IBM_KP_INSTANCE_ID != '' && env.IBM_KP_SERVICE_API_KEY != ''"
       with:
-        ibm-instance-id: ${{ secrets.IBM_INSTANCE_ID }}
-        ibm-service-api-key: ${{ secrets.IBM_SERVICE_API_KEY }}
         github-token: ${{ secrets.GITHUB_TOKEN }}
+        ibm-instance-id: ${{ secrets.IBM_KP_INSTANCE_ID }}
+        ibm-service-api-key: ${{ secrets.IBM_KP_SERVICE_API_KEY }}
 
     - name: upload test result
       uses: actions/upload-artifact@v2
diff --git a/.github/workflows/encryption-pvc-kms-ibm-kp/action.yml b/.github/workflows/encryption-pvc-kms-ibm-kp/action.yml
index ee8fdc6db..4b83110a8 100644
--- a/.github/workflows/encryption-pvc-kms-ibm-kp/action.yml
+++ b/.github/workflows/encryption-pvc-kms-ibm-kp/action.yml
@@ -2,7 +2,7 @@ name: Encryption KMS IBM Key Protect
 description: Reusable workflow to test Encryption KMS IBM Key Protect
 inputs:
   ibm-instance-id:
-    description: IBM_KP_SERVICE_INSTANCE_ID from the calling workflow
+    description: IBM_KP_INSTANCE_ID from the calling workflow
     required: true
   ibm-service-api-key:
     description: IBM_KP_SERVICE_API_KEY from the calling workflow
@@ -15,12 +15,12 @@ runs:
   using: "composite"
   steps:
     - name: fail if env no present
-      if: "env.IBM_KP_SERVICE_INSTANCE_ID == '' || env.IBM_KP_SERVICE_API_KEY == ''"
+      if: "env.IBM_KP_INSTANCE_ID == '' || env.IBM_KP_SERVICE_API_KEY == ''"
       env:
-        IBM_KP_SERVICE_INSTANCE_ID: ${{ inputs.ibm-instance-id }}
+        IBM_KP_INSTANCE_ID: ${{ inputs.ibm-instance-id }}
         IBM_KP_SERVICE_API_KEY: ${{ inputs.ibm-service-api-key }}
       shell: bash --noprofile --norc -eo pipefail -x {0}
-      run: echo "IBM_KP_SERVICE_INSTANCE_ID and IBM_KP_SERVICE_API_KEY must be set in the environment" && exit 0
+      run: echo "IBM_KP_INSTANCE_ID and IBM_KP_SERVICE_API_KEY must be set in the environment" && exit 0
 
     - name: setup cluster resources
       uses: ./.github/workflows/setup-cluster-resources
@@ -42,7 +42,7 @@ runs:
     - name: deploy cluster
       shell: bash --noprofile --norc -eo pipefail -x {0}
       env:
-        IBM_KP_SERVICE_INSTANCE_ID: ${{ inputs.ibm-instance-id }}
+        IBM_KP_INSTANCE_ID: ${{ inputs.ibm-instance-id }}
         IBM_KP_SERVICE_API_KEY: ${{ inputs.ibm-service-api-key }}
       run: |
         tests/scripts/github-action-helper.sh deploy_manifest_with_local_build deploy/examples/operator.yaml
diff --git a/Documentation/ceph-cluster-crd.md b/Documentation/ceph-cluster-crd.md
index 00dc85f01..6129ef861 100755
--- a/Documentation/ceph-cluster-crd.md
+++ b/Documentation/ceph-cluster-crd.md
@@ -396,7 +396,7 @@ spec:
   * In Openshift, to use a NetworkAttachmentDefinition (NAD) across namespaces, the NAD must be deployed in the `default` namespace. The NAD is then referenced with the namespace: `default/rook-public-nw`
 
 #### Known issues with multus
-When a CephFS/RBD volume is mounted in a Pod using cephcsi and then the CSI CephFS/RBD plugin is restarted or terminated (e.g. by restarting or deleting its DaemonSet), all operations on the volume become blocked, even after restarting the CSI pods. The only workaround is to restart the node where the cephcsi plugin pod was restarted.
+When a CephFS/RBD volume is mounted in a Pod using Ceph CSI and then the CSI CephFS/RBD plugin is restarted or terminated (e.g. by restarting or deleting its DaemonSet), all operations on the volume become blocked, even after restarting the CSI pods. The only workaround is to restart the node where the Ceph CSI plugin pod was restarted.
 This issue is tracked [here](https://github.com/rook/rook/issues/8085).
 
 #### IPFamily
diff --git a/Documentation/ceph-kms.md b/Documentation/ceph-kms.md
index 24f84d553..24f0d373b 100644
--- a/Documentation/ceph-kms.md
+++ b/Documentation/ceph-kms.md
@@ -266,8 +266,8 @@ security:
   kms:
     # name of the k8s config map containing all the kms connection details
     connectionDetails:
-      KMS_PROVIDER: ibmkeyprotect
-      IBM_KP_SERVICE_INSTANCE_ID: <instance ID that was retrieved in the first paragraph>
+      KMS_PROVIDER: ibm-kp
+      IBM_KP_INSTANCE_ID: <instance ID that was retrieved in the first paragraph>
     # name of the k8s secret containing the service API Key
     tokenSecretName: ibm-kp-svc-api-key
 ```
diff --git a/Documentation/ceph-upgrade.md b/Documentation/ceph-upgrade.md
index ec6727fb1..feb8ed786 100644
--- a/Documentation/ceph-upgrade.md
+++ b/Documentation/ceph-upgrade.md
@@ -545,14 +545,14 @@ 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.5.1"
+ROOK_CSI_CEPH_IMAGE: "quay.io/cephcsi/cephcsi:v3.5.0"
 ROOK_CSI_REGISTRAR_IMAGE: "k8s.gcr.io/sig-storage/csi-node-driver-registrar:v2.4.0"
 ROOK_CSI_PROVISIONER_IMAGE: "k8s.gcr.io/sig-storage/csi-provisioner:v3.1.0"
 ROOK_CSI_ATTACHER_IMAGE: "k8s.gcr.io/sig-storage/csi-attacher:v3.4.0"
-ROOK_CSI_RESIZER_IMAGE: "k8s.gcr.io/sig-storage/csi-resizer:v1.4.0"
-ROOK_CSI_SNAPSHOTTER_IMAGE: "k8s.gcr.io/sig-storage/csi-snapshotter:v5.0.1"
-CSI_VOLUME_REPLICATION_IMAGE: "quay.io/csiaddons/volumereplication-operator:v0.3.0"
-ROOK_CSIADDONS_IMAGE: "quay.io/csiaddons/k8s-sidecar:v0.2.1"
+ROOK_CSI_RESIZER_IMAGE: "k8s.gcr.io/sig-storage/csi-resizer:v1.3.0"
+ROOK_CSI_SNAPSHOTTER_IMAGE: "k8s.gcr.io/sig-storage/csi-snapshotter:v4.2.0"
+CSI_VOLUME_REPLICATION_IMAGE: "quay.io/csiaddons/volumereplication-operator:v0.1.0"
+ROOK_CSIADDONS_IMAGE: "quay.io/csiaddons/k8s-sidecar:v0.1.0"
 ```
 
 ### **Use default images**
@@ -574,9 +574,9 @@ kubectl --namespace rook-ceph get pod -o jsonpath='{range .items[*]}{range .spec
 k8s.gcr.io/sig-storage/csi-attacher:v3.4.0
 k8s.gcr.io/sig-storage/csi-node-driver-registrar:v2.4.0
 k8s.gcr.io/sig-storage/csi-provisioner:v3.1.0
-k8s.gcr.io/sig-storage/csi-resizer:v1.4.0
-k8s.gcr.io/sig-storage/csi-snapshotter:v5.0.1
-quay.io/cephcsi/cephcsi:v3.5.1
-quay.io/csiaddons/volumereplication-operator:v0.3.0
-quay.io/csiaddons/k8s-sidecar:v0.2.1
+k8s.gcr.io/sig-storage/csi-resizer:v1.3.0
+k8s.gcr.io/sig-storage/csi-snapshotter:v4.2.0
+quay.io/cephcsi/cephcsi:v3.5.0
+quay.io/csiaddons/volumereplication-operator:v0.1.0
+quay.io/csiaddons/k8s-sidecar:v0.1.0
 ```
diff --git a/Documentation/helm-operator.md b/Documentation/helm-operator.md
index 0ba6d395e..919c5cfc3 100644
--- a/Documentation/helm-operator.md
+++ b/Documentation/helm-operator.md
@@ -134,20 +134,20 @@ 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.5.1`                          |
+| `csi.cephcsi.image`                 | Ceph CSI image.                                                                                                             | `quay.io/cephcsi/cephcsi:v3.5.0`                          |
 | `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.4.0` |
-| `csi.resizer.image`                 | Kubernetes CSI resizer image.                                                                                               | `k8s.gcr.io/sig-storage/csi-resizer:v1.4.0`               |
+| `csi.resizer.image`                 | Kubernetes CSI resizer image.                                                                                               | `k8s.gcr.io/sig-storage/csi-resizer:v1.3.0`               |
 | `csi.provisioner.image`             | Kubernetes CSI provisioner image.                                                                                           | `k8s.gcr.io/sig-storage/csi-provisioner:v3.1.0`           |
-| `csi.snapshotter.image`             | Kubernetes CSI snapshotter image.                                                                                           | `k8s.gcr.io/sig-storage/csi-snapshotter:v5.0.1`           |
+| `csi.snapshotter.image`             | Kubernetes CSI snapshotter image.                                                                                           | `k8s.gcr.io/sig-storage/csi-snapshotter:v4.2.0`           |
 | `csi.attacher.image`                | Kubernetes CSI Attacher image.                                                                                              | `k8s.gcr.io/sig-storage/csi-attacher:v3.4.0`              |
 | `csi.cephfsPodLabels`               | Labels to add to the CSI CephFS Pods.                                                                                       | <none>                                                    |
 | `csi.rbdPodLabels`                  | Labels to add to the CSI RBD Pods.                                                                                          | <none>                                                    |
 | `csi.volumeReplication.enabled`     | Enable Volume Replication.                                                                                                  | `false`                                                   |
-| `csi.volumeReplication.image`       | Volume Replication Controller image.                                                                                        | `quay.io/csiaddons/volumereplication-operator:v0.3.0`     |
+| `csi.volumeReplication.image`       | Volume Replication Controller image.                                                                                        | `quay.io/csiaddons/volumereplication-operator:v0.1.0`     |
 | `csi.csiAddons.enabled`     | Enable CSIAddons                                                                                                  | `false`                                                   |
-| `csi.csiAddons.image`       | CSIAddons Sidecar image.                                                                                        | `quay.io/csiaddons/k8s-sidecar:v0.2.1`     |
+| `csi.csiAddons.image`       | CSIAddons Sidecar image.                                                                                        | `quay.io/csiaddons/k8s-sidecar:v0.1.0`     |
 | `admissionController.tolerations`   | Array of tolerations in YAML format which will be added to admission controller deployment.                                 | <none>                                                    |
 | `admissionController.nodeAffinity`  | The node labels for affinity of the admission controller deployment (***)                                                   | <none>                                                    |
 | `allowMultipleFilesystems`          | **(experimental in  Octopus (v15))** Allows multiple filesystems to be deployed to a Ceph cluster.                          | `false`                                                   |
diff --git a/ROADMAP.md b/ROADMAP.md
index 3c5a0733c..efd74c24c 100644
--- a/ROADMAP.md
+++ b/ROADMAP.md
@@ -8,23 +8,25 @@ We hope that the items listed below will inspire further engagement from the com
 Any dates listed below and the specific issues that will ship in a given milestone are subject to change but should give a general idea of what we are planning.
 See the [Github project boards](https://github.com/rook/rook/projects) for the most up-to-date issues and their status.
 
-## Rook Ceph 1.9
+## Rook Ceph 1.8
 
-The following high level features are targeted for Rook v1.9 (early April 2022). For more detailed project tracking see the [v1.9 board](https://github.com/rook/rook/projects/24).
+The following high level features are targeted for Rook v1.8 (December 2021). For more detailed project tracking see the [v1.8 board](https://github.com/rook/rook/projects/23).
 
-* Support for Ceph Quincy
-* Support for on-wire encryption and compression [#9054](https://github.com/rook/rook/issues/9054)
-* More complete set of commands for the [Rook Krew plugin](https://github.com/rook/kubectl-rook-ceph)
+* Configure bucket notifications with a CRD ([design doc](https://github.com/rook/rook/blob/master/design/ceph/object/ceph-bucket-notification-crd.md))
+* Disaster Recovery (DR): CSI solution for application failover in the event of cluster failure
+* OSD encryption key rotation [#7925](https://github.com/rook/rook/issues/7925)
+* Kubernetes authentication for cluster-wide encryption with Vault KMS
+* Provide a conversion tool for FlexVolumes to CSI and remove the FlexVolume driver support [#4043](https://github.com/rook/rook/issues/4043)
+* Remove support for Nautilus, focusing on support for Octopus and Pacific [#7908](https://github.com/rook/rook/issues/7908)
 
 ## Themes
 
 The general areas for improvements include the following, though may not be committed to a release.
 
-* Add alpha support for COSI (Container object storage interface) with K8s 1.24 [#7843](https://github.com/rook/rook/issues/7843)
+* Add alpha support for COSI (Container object storage interface) with K8s 1.22 [#7843](https://github.com/rook/rook/issues/7843)
 * iSCSI gateway deployment [#4334](https://github.com/rook/rook/issues/4334)
 * Enable the admission controller by default [#6242](https://github.com/rook/rook/issues/6242)
-* OSD encryption key rotation [#7925](https://github.com/rook/rook/issues/7925)
+* Dashboard-driven configuration after minimal CR install
 * Simplify metadata backup and disaster recovery
-* Strengthen approach for OSDs on PVCs for a more seamless K8s management of underlying storage
 * CSI Driver improvements tracked in the [CSI repo](https://github.com/ceph/ceph-csi)
   * Support for Windows nodes
diff --git a/deploy/charts/library/templates/_cluster-psp.tpl b/deploy/charts/library/templates/_cluster-psp.tpl
index 2d735d1ee..1918f7774 100644
--- a/deploy/charts/library/templates/_cluster-psp.tpl
+++ b/deploy/charts/library/templates/_cluster-psp.tpl
@@ -61,4 +61,4 @@ subjects:
   - kind: ServiceAccount
     name: rook-ceph-cmd-reporter
     namespace: {{ .Release.Namespace }} # namespace:cluster
-{{- end }}
+{{- end -}}
diff --git a/deploy/charts/library/templates/_cluster-rolebinding.tpl b/deploy/charts/library/templates/_cluster-rolebinding.tpl
index b9748d401..4196165b2 100644
--- a/deploy/charts/library/templates/_cluster-rolebinding.tpl
+++ b/deploy/charts/library/templates/_cluster-rolebinding.tpl
@@ -90,4 +90,4 @@ subjects:
   - kind: ServiceAccount
     name: rook-ceph-purge-osd
     namespace: {{ .Release.Namespace }} # namespace:cluster
-{{- end }}
+{{- end -}}
diff --git a/deploy/charts/library/templates/_recommended-labels.tpl b/deploy/charts/library/templates/_recommended-labels.tpl
index 546e3b55c..906755c67 100644
--- a/deploy/charts/library/templates/_recommended-labels.tpl
+++ b/deploy/charts/library/templates/_recommended-labels.tpl
@@ -6,4 +6,4 @@ app.kubernetes.io/part-of: rook-ceph-operator
 app.kubernetes.io/managed-by: helm
 app.kubernetes.io/created-by: helm
 helm.sh/chart: "{{ .Chart.Name }}-{{ .Chart.Version | replace "+" "_" }}"
-{{- end }}
+{{- end -}}
diff --git a/deploy/charts/library/templates/_suffix-cluster-namespace.tpl b/deploy/charts/library/templates/_suffix-cluster-namespace.tpl
index fdf679340..6957a910c 100644
--- a/deploy/charts/library/templates/_suffix-cluster-namespace.tpl
+++ b/deploy/charts/library/templates/_suffix-cluster-namespace.tpl
@@ -14,5 +14,5 @@ If the cluster namespace is different from the operator namespace, we want to na
 {{- $clusterNamespace := .Release.Namespace -}}
 {{- if ne $clusterNamespace $operatorNamespace -}}
 {{ printf "-%s" $clusterNamespace }}
-{{- end }}
-{{- end }}
+{{- end -}}
+{{- end -}}
diff --git a/deploy/charts/rook-ceph-cluster/templates/_helpers.tpl b/deploy/charts/rook-ceph-cluster/templates/_helpers.tpl
index a8a6fb5e7..96b5fdbfd 100644
--- a/deploy/charts/rook-ceph-cluster/templates/_helpers.tpl
+++ b/deploy/charts/rook-ceph-cluster/templates/_helpers.tpl
@@ -3,14 +3,14 @@ Define the clusterName as defaulting to the release namespace
 */}}
 {{- define "clusterName" -}}
 {{ .Values.clusterName | default .Release.Namespace }}
-{{- end }}
+{{- end -}}
 
 {{/*
 Return the target Kubernetes version.
 */}}
 {{- define "capabilities.kubeVersion" -}}
 {{- default .Capabilities.KubeVersion.Version .Values.kubeVersion -}}
-{{- end }}
+{{- end -}}
 
 {{/*
 Return the appropriate apiVersion for ingress.
@@ -22,5 +22,5 @@ Return the appropriate apiVersion for ingress.
 {{- print "networking.k8s.io/v1beta1" -}}
 {{- else -}}
 {{- print "networking.k8s.io/v1" -}}
-{{- end }}
-{{- end }}
+{{- end -}}
+{{- end -}}
diff --git a/deploy/charts/rook-ceph-cluster/templates/rbac.yaml b/deploy/charts/rook-ceph-cluster/templates/rbac.yaml
index 1e3540572..7b04b373d 100644
--- a/deploy/charts/rook-ceph-cluster/templates/rbac.yaml
+++ b/deploy/charts/rook-ceph-cluster/templates/rbac.yaml
@@ -14,7 +14,7 @@ clusterrolebindings
 {{- if .Values.pspEnable }}
 ---
 {{ include "library.cluster.psp.rolebindings" . }}
-{{- end }}
+{{- end -}}
 
 {{/*
 roles
@@ -38,4 +38,4 @@ rolebindings
 {{ include "library.cluster.monitoring.rolebindings" . }}
 {{- end }}
 
-{{- end }}
+{{- end -}}
diff --git a/deploy/charts/rook-ceph-cluster/templates/volumesnapshotclass.yaml b/deploy/charts/rook-ceph-cluster/templates/volumesnapshotclass.yaml
index 646824a76..704eb43f9 100644
--- a/deploy/charts/rook-ceph-cluster/templates/volumesnapshotclass.yaml
+++ b/deploy/charts/rook-ceph-cluster/templates/volumesnapshotclass.yaml
@@ -11,15 +11,8 @@ apiVersion: snapshot.storage.k8s.io/v1beta1
 kind: VolumeSnapshotClass
 metadata:
   name: {{ $filesystemvsc.name }}
-{{- if $filesystemvsc.labels }}
-  labels:
-{{ toYaml $filesystemvsc.labels | indent 4 }}
-{{- end }}
   annotations:
     snapshot.storage.kubernetes.io/is-default-class: "{{ if default false $filesystemvsc.isDefault }}true{{ else }}false{{ end }}"
-{{- if $filesystemvsc.annotations }}
-{{ toYaml $filesystemvsc.annotations | indent 4 }}
-{{- end }}
 driver: {{ .Values.operatorNamespace }}.cephfs.csi.ceph.com
 parameters:
   clusterID: {{ .Release.Namespace }}
@@ -41,15 +34,8 @@ apiVersion: snapshot.storage.k8s.io/v1beta1
 kind: VolumeSnapshotClass
 metadata:
   name: {{ $blockpoolvsc.name }}
-{{- if $blockpoolvsc.labels }}
-  labels:
-{{ toYaml $blockpoolvsc.labels | indent 4 }}
-{{- end }}
   annotations:
     snapshot.storage.kubernetes.io/is-default-class: "{{ if default false $blockpoolvsc.isDefault }}true{{ else }}false{{ end }}"
-{{- if $blockpoolvsc.annotations }}
-{{ toYaml $blockpoolvsc.annotations | indent 4 }}
-{{- end }}
 driver: {{ .Values.operatorNamespace }}.rbd.csi.ceph.com
 parameters:
   clusterID: {{ .Release.Namespace }}
diff --git a/deploy/charts/rook-ceph-cluster/values.yaml b/deploy/charts/rook-ceph-cluster/values.yaml
index fd3e8f270..68c27158b 100644
--- a/deploy/charts/rook-ceph-cluster/values.yaml
+++ b/deploy/charts/rook-ceph-cluster/values.yaml
@@ -403,8 +403,6 @@ cephFileSystemVolumeSnapshotClass:
   name: ceph-filesystem
   isDefault: true
   deletionPolicy: Delete
-  annotations: {}
-  labels: {}
   # see https://rook.io/docs/rook/latest/ceph-csi-snapshot.html#cephfs-snapshots for available configuration
   parameters: {}
 
@@ -413,8 +411,6 @@ cephBlockPoolsVolumeSnapshotClass:
   name: ceph-block
   isDefault: false
   deletionPolicy: Delete
-  annotations: {}
-  labels: {}
   # see https://rook.io/docs/rook/latest/ceph-csi-snapshot.html#rbd-snapshots for available configuration
   parameters: {}
 
diff --git a/deploy/charts/rook-ceph/templates/cluster-rbac.yaml b/deploy/charts/rook-ceph/templates/cluster-rbac.yaml
index 06dcca401..38a21a351 100644
--- a/deploy/charts/rook-ceph/templates/cluster-rbac.yaml
+++ b/deploy/charts/rook-ceph/templates/cluster-rbac.yaml
@@ -18,7 +18,7 @@ clusterrolebindings
 {{- if .Values.pspEnable }}
 ---
 {{ include "library.cluster.psp.rolebindings" . }}
-{{- end }}
+{{- end -}}
 
 {{/*
 roles
diff --git a/deploy/charts/rook-ceph/templates/clusterrole.yaml b/deploy/charts/rook-ceph/templates/clusterrole.yaml
index a61cd8a24..f76bbe883 100644
--- a/deploy/charts/rook-ceph/templates/clusterrole.yaml
+++ b/deploy/charts/rook-ceph/templates/clusterrole.yaml
@@ -425,19 +425,19 @@ rules:
     verbs: ["update", "patch"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshots"]
-    verbs: ["get", "list", "watch", "update", "patch"]
+    verbs: ["get", "list", "watch", "update"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshotcontents"]
-    verbs: ["create", "get", "list", "watch", "update", "delete", "patch"]
+    verbs: ["create", "get", "list", "watch", "update", "delete"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshotclasses"]
     verbs: ["get", "list", "watch"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshotcontents/status"]
-    verbs: ["update", "patch"]
+    verbs: ["update"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshots/status"]
-    verbs: ["update", "patch"]
+    verbs: ["update"]
 ---
 kind: ClusterRole
 apiVersion: rbac.authorization.k8s.io/v1
@@ -501,19 +501,19 @@ rules:
     verbs: ["list", "watch", "create", "update", "patch"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshots"]
-    verbs: ["get", "list", "watch", "update", "patch"]
+    verbs: ["get", "list", "watch", "update"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshotcontents"]
-    verbs: ["create", "get", "list", "watch", "update", "delete", "patch"]
+    verbs: ["create", "get", "list", "watch", "update", "delete"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshotclasses"]
     verbs: ["get", "list", "watch"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshotcontents/status"]
-    verbs: ["update", "patch"]
+    verbs: ["update"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshots/status"]
-    verbs: ["update", "patch"]
+    verbs: ["update"]
   - apiGroups: [""]
     resources: ["persistentvolumeclaims/status"]
     verbs: ["update", "patch"]
@@ -535,4 +535,4 @@ rules:
   - apiGroups: [""]
     resources: ["serviceaccounts"]
     verbs: ["get"]
-{{- end }}
+{{- end -}}
diff --git a/deploy/charts/rook-ceph/templates/clusterrolebinding.yaml b/deploy/charts/rook-ceph/templates/clusterrolebinding.yaml
index c99ec442c..766379795 100644
--- a/deploy/charts/rook-ceph/templates/clusterrolebinding.yaml
+++ b/deploy/charts/rook-ceph/templates/clusterrolebinding.yaml
@@ -99,4 +99,4 @@ roleRef:
   kind: ClusterRole
   name: rbd-external-provisioner-runner
   apiGroup: rbac.authorization.k8s.io
-{{- end }}
+{{- end -}}
diff --git a/deploy/charts/rook-ceph/templates/configmap.yaml b/deploy/charts/rook-ceph/templates/configmap.yaml
deleted file mode 100644
index ccc0a3a48..000000000
--- a/deploy/charts/rook-ceph/templates/configmap.yaml
+++ /dev/null
@@ -1,163 +0,0 @@
-# Operator settings that can be updated without an operator restart
-# Operator settings that require an operator restart are found in the operator env vars
-kind: ConfigMap
-apiVersion: v1
-metadata:
-  name: rook-ceph-operator-config
-data:
-  ROOK_LOG_LEVEL: {{ .Values.logLevel | quote }}
-  ROOK_CEPH_COMMANDS_TIMEOUT_SECONDS: {{ .Values.cephCommandsTimeoutSeconds | quote }}
-  ROOK_OBC_WATCH_OPERATOR_NAMESPACE: {{ .Values.enableOBCWatchOperatorNamespace | quote }}
-{{- if .Values.csi }}
-  ROOK_CSI_ENABLE_RBD: {{ .Values.csi.enableRbdDriver | quote }}
-  ROOK_CSI_ENABLE_CEPHFS: {{ .Values.csi.enableCephfsDriver | quote }}
-  CSI_ENABLE_CEPHFS_SNAPSHOTTER: {{ .Values.csi.enableCephfsSnapshotter | quote }}
-  CSI_ENABLE_RBD_SNAPSHOTTER: {{ .Values.csi.enableRBDSnapshotter | quote }}
-  CSI_PLUGIN_ENABLE_SELINUX_HOST_MOUNT: {{ .Values.csi.enablePluginSelinuxHostMount | quote }}
-  CSI_ENABLE_OMAP_GENERATOR: {{ .Values.csi.enableOMAPGenerator | quote }}
-{{- if .Values.csi.pluginPriorityClassName }}
-  CSI_PLUGIN_PRIORITY_CLASSNAME: {{ .Values.csi.pluginPriorityClassName | quote }}
-{{- end }}
-{{- if .Values.csi.provisionerPriorityClassName }}
-  CSI_PROVISIONER_PRIORITY_CLASSNAME: {{ .Values.csi.provisionerPriorityClassName | quote }}
-{{- end }}
-{{- if .Values.csi.enableCSIHostNetwork }}
-  CSI_ENABLE_HOST_NETWORK: {{ .Values.csi.enableCSIHostNetwork | quote }}
-{{- end }}
-{{- if .Values.csi.cephFSPluginUpdateStrategy }}
-  CSI_CEPHFS_PLUGIN_UPDATE_STRATEGY: {{ .Values.csi.cephFSPluginUpdateStrategy | quote }}
-{{- end }}
-{{- if .Values.csi.rbdFSGroupPolicy }}
-  CSI_RBD_FSGROUPPOLICY: {{ .Values.csi.rbdFSGroupPolicy | quote }}
-{{- end }}
-{{- if .Values.csi.cephFSFSGroupPolicy }}
-  CSI_CEPHFS_FSGROUPPOLICY: {{ .Values.csi.cephFSFSGroupPolicy | quote }}
-{{- end }}
-{{- if .Values.csi.rbdPluginUpdateStrategy }}
-  CSI_RBD_PLUGIN_UPDATE_STRATEGY: {{ .Values.csi.rbdPluginUpdateStrategy | quote }}
-{{- end }}
-{{- if .Values.csi.kubeletDirPath }}
-  ROOK_CSI_KUBELET_DIR_PATH: {{ .Values.csi.kubeletDirPath | quote }}
-{{- end }}
-  ROOK_CSI_ENABLE_GRPC_METRICS: {{ .Values.csi.enableGrpcMetrics | quote }}
-{{- if .Values.csi.cephcsi }}
-{{- if .Values.csi.cephcsi.image }}
-  ROOK_CSI_CEPH_IMAGE: {{ .Values.csi.cephcsi.image | quote }}
-{{- end }}
-{{- end }}
-{{- if .Values.csi.registrar }}
-{{- if .Values.csi.registrar.image }}
-  ROOK_CSI_REGISTRAR_IMAGE: {{ .Values.csi.registrar.image | quote }}
-{{- end }}
-{{- end }}
-{{- if .Values.csi.provisioner }}
-{{- if .Values.csi.provisioner.image }}
-  ROOK_CSI_PROVISIONER_IMAGE: {{ .Values.csi.provisioner.image | quote }}
-{{- end }}
-{{- end }}
-{{- if .Values.csi.snapshotter }}
-{{- if .Values.csi.snapshotter.image }}
-  ROOK_CSI_SNAPSHOTTER_IMAGE: {{ .Values.csi.snapshotter.image | quote }}
-{{- end }}
-{{- end }}
-{{- if .Values.csi.attacher }}
-{{- if .Values.csi.attacher.image }}
-  ROOK_CSI_ATTACHER_IMAGE: {{ .Values.csi.attacher.image | quote }}
-{{- end }}
-{{- end }}
-{{- if .Values.csi.resizer }}
-{{- if .Values.csi.resizer.image }}
-  ROOK_CSI_RESIZER_IMAGE: {{ .Values.csi.resizer.image | quote }}
-{{- end }}
-{{- end }}
-{{- if .Values.csi.volumeReplication }}
-  CSI_ENABLE_VOLUME_REPLICATION: {{ .Values.csi.volumeReplication.enabled | quote }}
-{{- if .Values.csi.volumeReplication.image }}
-  CSI_VOLUME_REPLICATION_IMAGE: {{ .Values.csi.volumeReplication.image | quote }}
-{{- end }}
-{{- end }}
-{{- if .Values.csi.csiAddons }}
-  CSI_ENABLE_CSIADDONS: {{ .Values.csi.csiAddons.enabled | quote }}
-{{- if .Values.csi.csiAddons.image }}
-  ROOK_CSIADDONS_IMAGE: {{ .Values.csi.csiAddons.image | quote }}
-{{- end }}
-{{- end }}
-{{- if .Values.csi.cephfsPodLabels }}
-  ROOK_CSI_CEPHFS_POD_LABELS: {{ .Values.csi.cephfsPodLabels | quote }}
-{{- end }}
-{{- if .Values.csi.rbdPodLabels }}
-  ROOK_CSI_RBD_POD_LABELS: {{ .Values.csi.rbdPodLabels | quote }}
-{{- end }}
-{{- if .Values.csi.provisionerTolerations }}
-  CSI_PROVISIONER_TOLERATIONS: {{ toYaml .Values.csi.provisionerTolerations | quote }}
-{{- end }}
-{{- if .Values.csi.provisionerNodeAffinity }}
-  CSI_PROVISIONER_NODE_AFFINITY: {{ .Values.csi.provisionerNodeAffinity }}
-{{- end }}
-{{- if .Values.csi.rbdProvisionerTolerations }}
-  CSI_RBD_PROVISIONER_TOLERATIONS: {{ toYaml .Values.csi.rbdProvisionerTolerations | quote }}
-{{- end }}
-{{- if .Values.csi.rbdProvisionerNodeAffinity }}
-  CSI_RBD_PROVISIONER_NODE_AFFINITY: {{ .Values.csi.rbdProvisionerNodeAffinity }}
-{{- end }}
-{{- if .Values.csi.cephFSProvisionerTolerations }}
-  CSI_CEPHFS_PROVISIONER_TOLERATIONS: {{ toYaml .Values.csi.cephFSProvisionerTolerations | quote }}
-{{- end }}
-{{- if .Values.csi.cephFSProvisionerNodeAffinity }}
-  CSI_CEPHFS_PROVISIONER_NODE_AFFINITY: {{ .Values.csi.cephFSProvisionerNodeAffinity }}
-{{- end }}
-{{- if .Values.csi.allowUnsupportedVersion }}
-  ROOK_CSI_ALLOW_UNSUPPORTED_VERSION: {{ .Values.csi.allowUnsupportedVersion | quote }}
-{{- end }}
-{{- if .Values.csi.pluginTolerations }}
-  CSI_PLUGIN_TOLERATIONS: {{ toYaml .Values.csi.pluginTolerations | quote }}
-{{- end }}
-{{- if .Values.csi.pluginNodeAffinity }}
-  CSI_PLUGIN_NODE_AFFINITY: {{ .Values.csi.pluginNodeAffinity }}
-{{- end }}
-{{- if .Values.csi.rbdPluginTolerations }}
-  CSI_RBD_PLUGIN_TOLERATIONS: {{ toYaml .Values.csi.rbdPluginTolerations | quote }}
-{{- end }}
-{{- if .Values.csi.rbdPluginNodeAffinity }}
-  CSI_RBD_PLUGIN_NODE_AFFINITY: {{ .Values.csi.rbdPluginNodeAffinity }}
-{{- end }}
-{{- if .Values.csi.cephFSPluginTolerations }}
-  CSI_CEPHFS_PLUGIN_TOLERATIONS: {{ toYaml .Values.csi.cephFSPluginTolerations | quote }}
-{{- end }}
-{{- if .Values.csi.cephFSPluginNodeAffinity }}
-  CSI_CEPHFS_PLUGIN_NODE_AFFINITY: {{ .Values.csi.cephFSPluginNodeAffinity }}
-{{- end }}
-{{- if .Values.csi.cephfsGrpcMetricsPort }}
-  CSI_CEPHFS_GRPC_METRICS_PORT: {{ .Values.csi.cephfsGrpcMetricsPort | quote }}
-{{- end }}
-{{- if .Values.csi.cephfsLivenessMetricsPort }}
-  CSI_CEPHFS_LIVENESS_METRICS_PORT: {{ .Values.csi.cephfsLivenessMetricsPort | quote }}
-{{- end }}
-{{- if .Values.csi.rbdGrpcMetricsPort }}
-  CSI_RBD_GRPC_METRICS_PORT: {{ .Values.csi.rbdGrpcMetricsPort | quote }}
-{{- end }}
-{{- if .Values.csi.rbdLivenessMetricsPort }}
-  CSI_RBD_LIVENESS_METRICS_PORT: {{ .Values.csi.rbdLivenessMetricsPort | quote }}
-{{- end }}
-{{- if .Values.csi.forceCephFSKernelClient }}
-  CSI_FORCE_CEPHFS_KERNEL_CLIENT: {{ .Values.csi.forceCephFSKernelClient | quote }}
-{{- end }}
-{{- if .Values.csi.logLevel }}
-  CSI_LOG_LEVEL: {{ .Values.csi.logLevel | quote }}
-{{- end }}
-{{- if .Values.csi.provisionerReplicas }}
-  CSI_PROVISIONER_REPLICAS: {{ .Values.csi.provisionerReplicas | quote }}
-{{- end }}
-{{- if .Values.csi.csiRBDProvisionerResource }}
-  CSI_RBD_PROVISIONER_RESOURCE: {{ .Values.csi.csiRBDProvisionerResource | quote }}
-{{- end }}
-{{- if .Values.csi.csiRBDPluginResource }}
-  CSI_RBD_PLUGIN_RESOURCE: {{ .Values.csi.csiRBDPluginResource | quote }}
-{{- end }}
-{{- if .Values.csi.csiCephFSProvisionerResource }}
-  CSI_CEPHFS_PROVISIONER_RESOURCE: {{ .Values.csi.csiCephFSProvisionerResource | quote }}
-{{- end }}
-{{- if .Values.csi.csiCephFSPluginResource }}
-  CSI_CEPHFS_PLUGIN_RESOURCE: {{ .Values.csi.csiCephFSPluginResource | quote }}
-{{- end }}
-{{- end }}
diff --git a/deploy/charts/rook-ceph/templates/deployment.yaml b/deploy/charts/rook-ceph/templates/deployment.yaml
index 9e98f86f7..446e59b24 100644
--- a/deploy/charts/rook-ceph/templates/deployment.yaml
+++ b/deploy/charts/rook-ceph/templates/deployment.yaml
@@ -68,12 +68,222 @@ spec:
 {{- end }}
         - name: ROOK_HOSTPATH_REQUIRES_PRIVILEGED
           value: "{{ .Values.hostpathRequiresPrivileged }}"
+        - name: ROOK_LOG_LEVEL
+          value: {{ .Values.logLevel }}
         - name: ROOK_ENABLE_SELINUX_RELABELING
           value: "{{ .Values.enableSelinuxRelabeling }}"
         - name: ROOK_DISABLE_DEVICE_HOTPLUG
           value: "{{ .Values.disableDeviceHotplug }}"
+{{- if .Values.csi }}
+        - name: ROOK_CSI_ENABLE_RBD
+          value: {{ .Values.csi.enableRbdDriver | quote }}
+        - name: ROOK_CSI_ENABLE_CEPHFS
+          value: {{ .Values.csi.enableCephfsDriver | quote }}
+        - name: CSI_ENABLE_CEPHFS_SNAPSHOTTER
+          value: {{ .Values.csi.enableCephfsSnapshotter | quote }}
+        - name: CSI_ENABLE_RBD_SNAPSHOTTER
+          value: {{ .Values.csi.enableRBDSnapshotter | quote }}
+        - name: CSI_PLUGIN_PRIORITY_CLASSNAME
+          value: {{ .Values.csi.pluginPriorityClassName | quote }}
+        - name: CSI_PROVISIONER_PRIORITY_CLASSNAME
+          value: {{ .Values.csi.provisionerPriorityClassName | quote }}
+        - name: CSI_ENABLE_OMAP_GENERATOR
+          value: {{ .Values.csi.enableOMAPGenerator | quote }}
+        - name: CSI_ENABLE_VOLUME_REPLICATION
+          value: {{ .Values.csi.volumeReplication.enabled | quote }}
+        - name: CSI_ENABLE_CSIADDONS
+          value: {{ .Values.csi.csiAddons.enabled | quote }}
+        - name: CSI_PLUGIN_ENABLE_SELINUX_HOST_MOUNT
+          value: {{ .Values.csi.enablePluginSelinuxHostMount | quote }}
+{{- if .Values.csi.enableCSIHostNetwork }}
+        - name: CSI_ENABLE_HOST_NETWORK
+          value: {{ .Values.csi.enableCSIHostNetwork | quote }}
+{{- end }}
+{{- if .Values.csi.cephFSPluginUpdateStrategy }}
+        - name: CSI_CEPHFS_PLUGIN_UPDATE_STRATEGY
+          value: {{ .Values.csi.cephFSPluginUpdateStrategy | quote }}
+{{- end }}
+{{- if .Values.csi.rbdFSGroupPolicy }}
+        - name: CSI_RBD_FSGROUPPOLICY
+          value: {{ .Values.csi.rbdFSGroupPolicy | quote }}
+{{- end }}
+{{- if .Values.csi.cephFSFSGroupPolicy }}
+        - name: CSI_CEPHFS_FSGROUPPOLICY
+          value: {{ .Values.csi.cephFSFSGroupPolicy | quote }}
+{{- end }}
+{{- if .Values.csi.rbdPluginUpdateStrategy }}
+        - name: CSI_RBD_PLUGIN_UPDATE_STRATEGY
+          value: {{ .Values.csi.rbdPluginUpdateStrategy | quote }}
+{{- end }}
+{{- if .Values.csi.kubeletDirPath }}
+        - name: ROOK_CSI_KUBELET_DIR_PATH
+          value: {{ .Values.csi.kubeletDirPath | quote }}
+{{- end }}
+        - name: ROOK_CSI_ENABLE_GRPC_METRICS
+          value: {{ .Values.csi.enableGrpcMetrics | quote }}
+{{- if .Values.csi.cephcsi }}
+{{- if .Values.csi.cephcsi.image }}
+        - name: ROOK_CSI_CEPH_IMAGE
+          value: {{ .Values.csi.cephcsi.image | quote }}
+{{- end }}
+{{- end }}
+{{- if .Values.csi.registrar }}
+{{- if .Values.csi.registrar.image }}
+        - name: ROOK_CSI_REGISTRAR_IMAGE
+          value: {{ .Values.csi.registrar.image | quote }}
+{{- end }}
+{{- end }}
+{{- if .Values.csi.provisioner }}
+{{- if .Values.csi.provisioner.image }}
+        - name: ROOK_CSI_PROVISIONER_IMAGE
+          value: {{ .Values.csi.provisioner.image | quote }}
+{{- end }}
+{{- end }}
+{{- if .Values.csi.snapshotter }}
+{{- if .Values.csi.snapshotter.image }}
+        - name: ROOK_CSI_SNAPSHOTTER_IMAGE
+          value: {{ .Values.csi.snapshotter.image | quote }}
+{{- end }}
+{{- end }}
+{{- if .Values.csi.attacher }}
+{{- if .Values.csi.attacher.image }}
+        - name: ROOK_CSI_ATTACHER_IMAGE
+          value: {{ .Values.csi.attacher.image | quote }}
+{{- end }}
+{{- end }}
+{{- if .Values.csi.resizer }}
+{{- if .Values.csi.resizer.image }}
+        - name: ROOK_CSI_RESIZER_IMAGE
+          value: {{ .Values.csi.resizer.image | quote }}
+{{- end }}
+{{- end }}
+{{- if .Values.csi.volumeReplication }}
+{{- if .Values.csi.volumeReplication.image }}
+        - name: CSI_VOLUME_REPLICATION_IMAGE
+          value: {{ .Values.csi.volumeReplication.image | quote }}
+{{- end }}
+{{- end }}
+{{- if .Values.csi.csiAddons }}
+{{- if .Values.csi.csiAddons.image }}
+        - name: ROOK_CSIADDONS_IMAGE
+          value: {{ .Values.csi.csiAddons.image | quote }}
+{{- end }}
+{{- if .Values.csi.csiAddonsPort }}
+        - name: CSIADDONS_PORT
+          value: {{ .Values.csi.csiAddonsPort | quote }}
+{{- end }}
+{{- end }}
+{{- if .Values.csi.cephfsPodLabels }}
+        - name: ROOK_CSI_CEPHFS_POD_LABELS
+          value: {{ .Values.csi.cephfsPodLabels | quote }}
+{{- end }}
+{{- if .Values.csi.rbdPodLabels }}
+        - name: ROOK_CSI_RBD_POD_LABELS
+          value: {{ .Values.csi.rbdPodLabels | quote }}
+{{- end }}
+{{- if .Values.csi.provisionerTolerations }}
+        - name: CSI_PROVISIONER_TOLERATIONS
+          value: {{ toYaml .Values.csi.provisionerTolerations | quote }}
+{{- end }}
+{{- if .Values.csi.provisionerNodeAffinity }}
+        - name: CSI_PROVISIONER_NODE_AFFINITY
+          value: {{ .Values.csi.provisionerNodeAffinity }}
+{{- end }}
+{{- if .Values.csi.rbdProvisionerTolerations }}
+        - name: CSI_RBD_PROVISIONER_TOLERATIONS
+          value: {{ toYaml .Values.csi.rbdProvisionerTolerations | quote }}
+{{- end }}
+{{- if .Values.csi.rbdProvisionerNodeAffinity }}
+        - name: CSI_RBD_PROVISIONER_NODE_AFFINITY
+          value: {{ .Values.csi.rbdProvisionerNodeAffinity }}
+{{- end }}
+{{- if .Values.csi.cephFSProvisionerTolerations }}
+        - name: CSI_CEPHFS_PROVISIONER_TOLERATIONS
+          value: {{ toYaml .Values.csi.cephFSProvisionerTolerations | quote }}
+{{- end }}
+{{- if .Values.csi.cephFSProvisionerNodeAffinity }}
+        - name: CSI_CEPHFS_PROVISIONER_NODE_AFFINITY
+          value: {{ .Values.csi.cephFSProvisionerNodeAffinity }}
+{{- end }}
+{{- if .Values.csi.allowUnsupportedVersion }}
+        - name: ROOK_CSI_ALLOW_UNSUPPORTED_VERSION
+          value: {{ .Values.csi.allowUnsupportedVersion | quote }}
+{{- end }}
+{{- if .Values.csi.pluginTolerations }}
+        - name: CSI_PLUGIN_TOLERATIONS
+          value: {{ toYaml .Values.csi.pluginTolerations | quote }}
+{{- end }}
+{{- if .Values.csi.pluginNodeAffinity }}
+        - name: CSI_PLUGIN_NODE_AFFINITY
+          value: {{ .Values.csi.pluginNodeAffinity }}
+{{- end }}
+{{- if .Values.csi.rbdPluginTolerations }}
+        - name: CSI_RBD_PLUGIN_TOLERATIONS
+          value: {{ toYaml .Values.csi.rbdPluginTolerations | quote }}
+{{- end }}
+{{- if .Values.csi.rbdPluginNodeAffinity }}
+        - name: CSI_RBD_PLUGIN_NODE_AFFINITY
+          value: {{ .Values.csi.rbdPluginNodeAffinity }}
+{{- end }}
+{{- if .Values.csi.cephFSPluginTolerations }}
+        - name: CSI_CEPHFS_PLUGIN_TOLERATIONS
+          value: {{ toYaml .Values.csi.cephFSPluginTolerations | quote }}
+{{- end }}
+{{- if .Values.csi.cephFSPluginNodeAffinity }}
+        - name: CSI_CEPHFS_PLUGIN_NODE_AFFINITY
+          value: {{ .Values.csi.cephFSPluginNodeAffinity }}
+{{- end }}
+{{- if .Values.csi.cephfsGrpcMetricsPort }}
+        - name: CSI_CEPHFS_GRPC_METRICS_PORT
+          value: {{ .Values.csi.cephfsGrpcMetricsPort | quote }}
+{{- end }}
+{{- if .Values.csi.cephfsLivenessMetricsPort }}
+        - name: CSI_CEPHFS_LIVENESS_METRICS_PORT
+          value: {{ .Values.csi.cephfsLivenessMetricsPort | quote }}
+{{- end }}
+{{- if .Values.csi.rbdGrpcMetricsPort }}
+        - name: CSI_RBD_GRPC_METRICS_PORT
+          value: {{ .Values.csi.rbdGrpcMetricsPort | quote }}
+{{- end }}
+{{- if .Values.csi.rbdLivenessMetricsPort }}
+        - name: CSI_RBD_LIVENESS_METRICS_PORT
+          value: {{ .Values.csi.rbdLivenessMetricsPort | quote }}
+{{- end }}
+{{- if .Values.csi.forceCephFSKernelClient }}
+        - name: CSI_FORCE_CEPHFS_KERNEL_CLIENT
+          value: {{ .Values.csi.forceCephFSKernelClient | quote }}
+{{- end }}
+{{- if .Values.csi.logLevel }}
+        - name: CSI_LOG_LEVEL
+          value: {{ .Values.csi.logLevel | quote }}
+{{- end }}
+{{- if .Values.csi.provisionerReplicas }}
+        - name: CSI_PROVISIONER_REPLICAS
+          value: {{ .Values.csi.provisionerReplicas | quote }}
+{{- end }}
+{{- if .Values.csi.csiRBDProvisionerResource }}
+        - name: CSI_RBD_PROVISIONER_RESOURCE
+          value: {{ .Values.csi.csiRBDProvisionerResource | quote }}
+{{- end }}
+{{- if .Values.csi.csiRBDPluginResource }}
+        - name: CSI_RBD_PLUGIN_RESOURCE
+          value: {{ .Values.csi.csiRBDPluginResource | quote }}
+{{- end }}
+{{- if .Values.csi.csiCephFSProvisionerResource }}
+        - name: CSI_CEPHFS_PROVISIONER_RESOURCE
+          value: {{ .Values.csi.csiCephFSProvisionerResource | quote }}
+{{- end }}
+{{- if .Values.csi.csiCephFSPluginResource }}
+        - name: CSI_CEPHFS_PLUGIN_RESOURCE
+          value: {{ .Values.csi.csiCephFSPluginResource | quote }}
+{{- end }}
+{{- end }}
         - name: ROOK_ENABLE_DISCOVERY_DAEMON
           value: "{{ .Values.enableDiscoveryDaemon }}"
+        - name: ROOK_CEPH_COMMANDS_TIMEOUT_SECONDS
+          value: "{{ .Values.cephCommandsTimeoutSeconds }}"
+        - name: ROOK_OBC_WATCH_OPERATOR_NAMESPACE
+          value: "{{ .Values.enableOBCWatchOperatorNamespace }}"
 
         - name: NODE_NAME
           valueFrom:
diff --git a/deploy/charts/rook-ceph/templates/psp.yaml b/deploy/charts/rook-ceph/templates/psp.yaml
index 920299873..23c5a415b 100644
--- a/deploy/charts/rook-ceph/templates/psp.yaml
+++ b/deploy/charts/rook-ceph/templates/psp.yaml
@@ -168,4 +168,4 @@ subjects:
     name: rook-csi-rbd-provisioner-sa
     namespace: {{ .Release.Namespace }} # namespace:operator
 {{- end }}
-{{- end }}
+{{- end -}}
diff --git a/deploy/charts/rook-ceph/templates/rolebinding.yaml b/deploy/charts/rook-ceph/templates/rolebinding.yaml
index 5e5c4e6ce..5be43f9ca 100644
--- a/deploy/charts/rook-ceph/templates/rolebinding.yaml
+++ b/deploy/charts/rook-ceph/templates/rolebinding.yaml
@@ -47,7 +47,7 @@ roleRef:
   name: rbd-csi-nodeplugin
   apiGroup: rbac.authorization.k8s.io
 ---
-{{- end }}
+{{- end -}}
 kind: RoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
 metadata:
@@ -61,4 +61,4 @@ roleRef:
   kind: Role
   name: rbd-external-provisioner-cfg
   apiGroup: rbac.authorization.k8s.io
-{{- end }}
+{{- end -}}
diff --git a/deploy/charts/rook-ceph/values.yaml b/deploy/charts/rook-ceph/values.yaml
index 19edea3d0..8a869b761 100644
--- a/deploy/charts/rook-ceph/values.yaml
+++ b/deploy/charts/rook-ceph/values.yaml
@@ -276,17 +276,17 @@ csi:
   #rbdLivenessMetricsPort: 9080
   #kubeletDirPath: /var/lib/kubelet
   #cephcsi:
-    #image: quay.io/cephcsi/cephcsi:v3.5.1
+    #image: quay.io/cephcsi/cephcsi:v3.5.0
   #registrar:
     #image: k8s.gcr.io/sig-storage/csi-node-driver-registrar:v2.4.0
   #provisioner:
     #image: k8s.gcr.io/sig-storage/csi-provisioner:v3.1.0
   #snapshotter:
-    #image: k8s.gcr.io/sig-storage/csi-snapshotter:v5.0.1
+    #image: k8s.gcr.io/sig-storage/csi-snapshotter:v4.2.0
   #attacher:
     #image: k8s.gcr.io/sig-storage/csi-attacher:v3.4.0
   #resizer:
-    #image: k8s.gcr.io/sig-storage/csi-resizer:v1.4.0
+    #image: k8s.gcr.io/sig-storage/csi-resizer:v1.3.0
   # Labels to add to the CSI CephFS Deployments and DaemonSets Pods.
   #cephfsPodLabels: "key1=value1,key2=value2"
   # Labels to add to the CSI RBD Deployments and DaemonSets Pods.
@@ -296,11 +296,11 @@ csi:
   # See https://rook.io/docs/rook/latest/ceph-csi-drivers.html#rbd-mirroring
   volumeReplication:
     enabled: false
-    #image: "quay.io/csiaddons/volumereplication-operator:v0.3.0"
+    #image: "quay.io/csiaddons/volumereplication-operator:v0.1.0"
   # Enable the CSIAddons sidecar.
   csiAddons:
     enabled: false
-    #image: "quay.io/csiaddons/k8s-sidecar:v0.2.1"
+    #image: "quay.io/csiaddons/k8s-sidecar:v0.1.0"
 enableDiscoveryDaemon: false
 cephCommandsTimeoutSeconds: "15"
 
diff --git a/deploy/examples/common.yaml b/deploy/examples/common.yaml
index b560c9adc..254a03b2e 100644
--- a/deploy/examples/common.yaml
+++ b/deploy/examples/common.yaml
@@ -66,19 +66,19 @@ rules:
     verbs: ["update", "patch"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshots"]
-    verbs: ["get", "list", "watch", "update", "patch"]
+    verbs: ["get", "list", "watch", "update"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshotcontents"]
-    verbs: ["create", "get", "list", "watch", "update", "delete", "patch"]
+    verbs: ["create", "get", "list", "watch", "update", "delete"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshotclasses"]
     verbs: ["get", "list", "watch"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshotcontents/status"]
-    verbs: ["update", "patch"]
+    verbs: ["update"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshots/status"]
-    verbs: ["update", "patch"]
+    verbs: ["update"]
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
@@ -160,19 +160,19 @@ rules:
     verbs: ["list", "watch", "create", "update", "patch"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshots"]
-    verbs: ["get", "list", "watch", "update", "patch"]
+    verbs: ["get", "list", "watch", "update"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshotcontents"]
-    verbs: ["create", "get", "list", "watch", "update", "delete", "patch"]
+    verbs: ["create", "get", "list", "watch", "update", "delete"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshotclasses"]
     verbs: ["get", "list", "watch"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshotcontents/status"]
-    verbs: ["update", "patch"]
+    verbs: ["update"]
   - apiGroups: ["snapshot.storage.k8s.io"]
     resources: ["volumesnapshots/status"]
-    verbs: ["update", "patch"]
+    verbs: ["update"]
   - apiGroups: [""]
     resources: ["persistentvolumeclaims/status"]
     verbs: ["update", "patch"]
diff --git a/deploy/examples/images.txt b/deploy/examples/images.txt
index ffb69107a..dac22620e 100644
--- a/deploy/examples/images.txt
+++ b/deploy/examples/images.txt
@@ -1,10 +1,10 @@
  k8s.gcr.io/sig-storage/csi-attacher:v3.4.0
  k8s.gcr.io/sig-storage/csi-node-driver-registrar:v2.4.0
  k8s.gcr.io/sig-storage/csi-provisioner:v3.1.0
- k8s.gcr.io/sig-storage/csi-resizer:v1.4.0
- k8s.gcr.io/sig-storage/csi-snapshotter:v5.0.1
+ k8s.gcr.io/sig-storage/csi-resizer:v1.3.0
+ k8s.gcr.io/sig-storage/csi-snapshotter:v4.2.0
  quay.io/ceph/ceph:v16.2.7
- quay.io/cephcsi/cephcsi:v3.5.1
- quay.io/csiaddons/k8s-sidecar:v0.2.1
- quay.io/csiaddons/volumereplication-operator:v0.3.0
+ quay.io/cephcsi/cephcsi:v3.5.0
+ quay.io/csiaddons/k8s-sidecar:v0.1.0
+ quay.io/csiaddons/volumereplication-operator:v0.1.0
  rook/ceph:master
diff --git a/deploy/examples/operator-openshift.yaml b/deploy/examples/operator-openshift.yaml
index 555cc9276..d1340d6b4 100644
--- a/deploy/examples/operator-openshift.yaml
+++ b/deploy/examples/operator-openshift.yaml
@@ -161,11 +161,11 @@ 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.5.1"
+  # ROOK_CSI_CEPH_IMAGE: "quay.io/cephcsi/cephcsi:v3.5.0"
   # ROOK_CSI_REGISTRAR_IMAGE: "k8s.gcr.io/sig-storage/csi-node-driver-registrar:v2.4.0"
-  # ROOK_CSI_RESIZER_IMAGE: "k8s.gcr.io/sig-storage/csi-resizer:v1.4.0"
+  # ROOK_CSI_RESIZER_IMAGE: "k8s.gcr.io/sig-storage/csi-resizer:v1.3.0"
   # ROOK_CSI_PROVISIONER_IMAGE: "k8s.gcr.io/sig-storage/csi-provisioner:v3.1.0"
-  # ROOK_CSI_SNAPSHOTTER_IMAGE: "k8s.gcr.io/sig-storage/csi-snapshotter:v5.0.1"
+  # ROOK_CSI_SNAPSHOTTER_IMAGE: "k8s.gcr.io/sig-storage/csi-snapshotter:v4.2.0"
   # ROOK_CSI_ATTACHER_IMAGE: "k8s.gcr.io/sig-storage/csi-attacher:v3.4.0"
 
   # (Optional) set user created priorityclassName for csi plugin pods.
@@ -417,10 +417,10 @@ data:
   CSI_ENABLE_VOLUME_REPLICATION: "false"
   # The timeout value (in seconds) of Ceph commands. It should be >= 1. If this variable is not set or is an invalid value, it's default to 15.
   ROOK_CEPH_COMMANDS_TIMEOUT_SECONDS: "15"
-  # CSI_VOLUME_REPLICATION_IMAGE: "quay.io/csiaddons/volumereplication-operator:v0.3.0"
+  # CSI_VOLUME_REPLICATION_IMAGE: "quay.io/csiaddons/volumereplication-operator:v0.1.0"
   # Enable the csi addons sidecar.
   CSI_ENABLE_CSIADDONS: "false"
-  # ROOK_CSIADDONS_IMAGE: "quay.io/csiaddons/k8s-sidecar:v0.2.1"
+  # ROOK_CSIADDONS_IMAGE: "quay.io/csiaddons/k8s-sidecar:v0.1.0"
 ---
 # The deployment for the rook operator
 # OLM: BEGIN OPERATOR DEPLOYMENT
diff --git a/deploy/examples/operator.yaml b/deploy/examples/operator.yaml
index e5b453b51..c81ad7a9e 100644
--- a/deploy/examples/operator.yaml
+++ b/deploy/examples/operator.yaml
@@ -79,11 +79,11 @@ 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.5.1"
+  # ROOK_CSI_CEPH_IMAGE: "quay.io/cephcsi/cephcsi:v3.5.0"
   # ROOK_CSI_REGISTRAR_IMAGE: "k8s.gcr.io/sig-storage/csi-node-driver-registrar:v2.4.0"
-  # ROOK_CSI_RESIZER_IMAGE: "k8s.gcr.io/sig-storage/csi-resizer:v1.4.0"
+  # ROOK_CSI_RESIZER_IMAGE: "k8s.gcr.io/sig-storage/csi-resizer:v1.3.0"
   # ROOK_CSI_PROVISIONER_IMAGE: "k8s.gcr.io/sig-storage/csi-provisioner:v3.1.0"
-  # ROOK_CSI_SNAPSHOTTER_IMAGE: "k8s.gcr.io/sig-storage/csi-snapshotter:v5.0.1"
+  # ROOK_CSI_SNAPSHOTTER_IMAGE: "k8s.gcr.io/sig-storage/csi-snapshotter:v4.2.0"
   # ROOK_CSI_ATTACHER_IMAGE: "k8s.gcr.io/sig-storage/csi-attacher:v3.4.0"
 
   # (Optional) set user created priorityclassName for csi plugin pods.
@@ -335,10 +335,10 @@ data:
   # Before enabling, ensure the Volume Replication CRDs are created.
   # See https://rook.io/docs/rook/latest/ceph-csi-drivers.html#rbd-mirroring
   CSI_ENABLE_VOLUME_REPLICATION: "false"
-  # CSI_VOLUME_REPLICATION_IMAGE: "quay.io/csiaddons/volumereplication-operator:v0.3.0"
+  # CSI_VOLUME_REPLICATION_IMAGE: "quay.io/csiaddons/volumereplication-operator:v0.1.0"
   # Enable the csi addons sidecar.
   CSI_ENABLE_CSIADDONS: "false"
-  # ROOK_CSIADDONS_IMAGE: "quay.io/csiaddons/k8s-sidecar:v0.2.1"
+  # ROOK_CSIADDONS_IMAGE: "quay.io/csiaddons/k8s-sidecar:v0.1.0"
 ---
 # OLM: BEGIN OPERATOR DEPLOYMENT
 apiVersion: apps/v1
diff --git a/deploy/examples/osd-env-override.yaml b/deploy/examples/osd-env-override.yaml
deleted file mode 100644
index 454ccc50a..000000000
--- a/deploy/examples/osd-env-override.yaml
+++ /dev/null
@@ -1,19 +0,0 @@
-# ###############################################################################################################
-# The `rook-ceph-osd-env-override` ConfigMap is a development feature
-# that allows to inject arbitrary environment variables to OSD-related
-# containers created by the operator.
-# ###############################################################################################################
-
-apiVersion: v1
-kind: ConfigMap
-metadata:
-  name: rook-ceph-osd-env-override
-  namespace: rook-ceph
-data:
-  # Bypass the ASan's assertion that it is the very first loaded DSO.
-  # This is necessary for crimson-osd as it's currently built with
-  # the ASan sanitizer turned on which means the `libasan.so` must
-  # the be the very first loaded dynamic library. Unfortunately, this
-  # isn't fulfilled as the containers use `ld.preload`, so ASan was
-  # aborting the entire OSD.
-  ASAN_OPTIONS: verify_asan_link_order=0
diff --git a/design/ceph/multus-network.md b/design/ceph/multus-network.md
index ed039cdd0..678751b0c 100644
--- a/design/ceph/multus-network.md
+++ b/design/ceph/multus-network.md
@@ -3,9 +3,9 @@
 We have already explored and explained the benefit of multi-homed networking, so this document will not rehearse that but simply focus on the implementation for the Ceph backend.
 If you are interested in learning more about multi-homed networking you can read the [design documentation on that matter](../core/multi-homed-cluster.md).
 
-To make the story short, [Multus](https://github.com/intel/multus-cni) should allow us to get the same performance benefit as `HostNetworking` by increasing the security.
+To make the story short, [Multus](https://github.com/intel/multus-cni) should allow us to get the same performance benefit as `HostNetworking` as well as increasing the security.
 Using `HostNetworking` results in exposing **all** the network interfaces (the entire stack) of the host inside the container where Multus allows you to pick the one you want.
-Also, this removes the need of privileged containers (required for `HostNetworking`).
+Also, this minimizes the need of privileged containers (required for `HostNetworking`).
 
 ## Proposed CRD changed
 
@@ -78,7 +78,123 @@ Nothing to do in particular since they don't use any service IPs.
 
 ### CSI pods
 
-We can add annotations to these pods and they can reach out the Ceph public network, then the driver will expose the block or the filesystem normally.
+Steps must be taken to fix a CSI-with-multus issue documented
+[here](https://github.com/rook/rook/issues/8085). To summarize the issue:
+when a CephFS/RBD volume is mounted in a pod using Ceph CSI and then the CSI CephFS/RBD plugin is
+restarted or terminated (e.g. by restarting or deleting its DaemonSet), all operations on the volume
+become blocked, even after restarting the CSI pods. The only workaround is to restart the node where
+the Ceph CSI plugin pod was restarted.
+
+When deploying a CephCluster resource configured to use multus networks, a multus-connected network
+interface will be added to the host network namespace for all nodes that will run CSI plugin pods.
+This will allow Ceph CSI pods to run using host networking and still access Ceph's public multus
+network.
+
+The design for mitigating the issue is comprised of two components: a "holder" DaemonSet and a
+"mover" Daemonset.
+
+#### Holder DaemonSet and Pods
+The Rook-Ceph Operator's CSI controller creates a DaemonSet configured to use the
+`network.selectors.public` network specified for the CephCluster. This DaemonSet runs on all the
+nodes that will have CSI plugin pods. Its pods exist to "hold" a particular network interface that
+the CSI pods can reliably connect to for communication with the Ceph cluster. The process running
+will merely be an infinite sleep.
+
+These Pods should only be stopped and restarted when a node is stopped so that volume operations do
+not become blocked. The Rook-Ceph Operator's CSI controller should set the DaemonSet's update
+strategy to `OnDelete` so that the pods do not get deleted if the DaemonSet is updated while also
+ensuring that the pods will be updated on the next node reboot (or node drain).
+
+#### Mover DaemonSet and Pods
+The Rook-Ceph Operator's CSI controller also creates a second DaemonSet configured to use host
+networking. This DaemonSet also runs on all nodes that will have CSI plugin pods (and holder pods).
+Mover pods exist to "move" the multus network interface being held by the holder pod on the node
+into the host's network namespace to provide user's volumes with uninterrupted access to the Ceph
+cluster, even when the CSI driver is restarted (or updated).
+
+The mover must:
+- be a privileged container
+- have `SYS_ADMIN` and `NET_ADMIN` capabilities
+- be on the host network namespace
+- have access to the `/var/run/netns` directory
+
+In order to not leave moved interfaces dangling in the host's network namespace, mover pods must
+move interfaces back to their original namespace when CSI is being terminated. The most
+straightforward way to accomplish this is to move interfaces back when the mover is being
+terminated. If an interface is moved without user applications also being removed, this will cause
+I/O disruption. Therefore, the DaemonSet should also use the `OnDelete` update strategy so that the
+pods can be updated on node reboots (or node drains).
+
+In order to better handle unexpected corner cases that leave moved interfaces in the host network
+namespace (e.g., a mover is killed abruptly rather than gracefully terminated), instead treat "move"
+operations as a disable-and-copy operation. To do this, disable the interface in the holder pod's
+network namespace, and create a copy of the interface in the host namespace with the same MAC
+address and IP config. From the user standpoint, the interface is still "moved" because the original
+is disabled, so we keep the "mover" terminology. This merely helps Rook ensure that it is not
+accidentally losing the original information.
+
+A previous iteration of this design specified the mover application as a sidecar to Ceph CSI plugin
+pods; however, this design would mean that the mover would need to be deleted and re-created
+whenever the CSI plugin is updated, possibly resulting in I/O hangs in user pods during the update.
+Keeping the mover independent allows CSI plugin updates to happen freely without complex
+interactions between it and the mover.
+
+#### Interactions between components
+If a copied interface is left in the host network namespace after the holder pod is removed, multus
+may later give the address to a different application, and the CSI driver may try to connect to the
+different application with Ceph requests. We should try to avoid leaving interfaces on the host as
+much as possible. Killing the mover pod abruptly will leave copied interfaces, but there is no way
+to prevent this from happening.
+
+If a holder pod is deleted, the interface hold will be lost. The mover must remove the
+interface from the host's network namespace because multus may reassign the address to a different
+application. We can document for users that they should not delete holder pods, but we cannot
+prevent users from manually stopping holder pods. However, to prevent the Kubernetes scheduler from
+terminating holder pods, the pods should be given the highest possible priority so that they are not
+un-scheduled except by users.
+
+There is a possible race condition where a mover pod is killed and where a holder is deleted before
+a new mover starts up. In this case, the copied interface for the holder pod will be left in the
+host network namespace, but the mover will not get a notification that the holder was removed. In
+order to clean up from this case, upon startup, the mover should delete all interface copies in the
+host network namespace that do not have a holder pod associated with them.
+
+If the mover container is stopped, it should delete all copied interfaces in the host's
+network namespace under the assumption that the CSI plugin is being removed, possibly by a
+CephCluster being deleted or the node going down for maintenance. It might be possible to optimize
+for the case where only the mover pod is being restarted, but it is very difficult to detect the
+case where the mover is merely being restarted versus when the holder is also being removed without
+possible race conditions between which pod might be stopped first by the Kubernetes API server
+during a drain event. Therefore, focus on the simplest working implementation (described above)
+instead of risking leaving the interface copy in host net namespace which could cause issues.
+
+If an error occurs in the mover during network migration, it will fail and re-try migration until
+the operation succeeds. If necessary, the mover will try to remove a partially-copied interface.
+
+Restarting a node will cause the multus interface in the host namespace to go away. On restart, the
+holder pod will get a new interface, and the mover will copy it into the host networking namespace again.
+
+When a new node is added, holder and mover pods are added to it by their DaemonSets, and the
+move/copy process described above occurs on the node.
+
+The holder and mover DaemonSets should be deleted when the CSI driver components are removed. Both
+the termination of the holder pods as well as the mover pods triggers the mover to remove the multus
+interfaces from the host network namespace of a given node.
+
+The initial implementation of this design will be limited to supporting a single CephCluster with
+Multus until we can be sure that the CSI plugin can support multiple migrated interfaces as well as
+interfaces that are added and removed dynamically. This limitation will be enforced by allowing only
+a single instance of the holder DaemonSet. A possible (future) partial implementation may be
+possible by restarting the CSI plugin Pods when network interfaces are added or removed.
+
+**Known issue:** the Docker container runtime does not use Linux's native `/var/run/netns`
+directory. This mitigation is known to work on cri-o runtime but not Docker. Therefore, this feature
+will be disabled by default and enabled optionally by the
+`ROOK_CSI_MULTUS_USE_HOLDER_MOVER_PATTERN=true` variable in the Rook-Ceph operator's config.
+
+A previous version of the CSI proposal had the holder Pods creating "setup" and "teardown"
+Kubernetes Jobs for migrating/un-migrating the multus networks. This design was rejected since new
+Jobs wouldn't be able to be created if the Kubernetes namespace were in "Terminating" state.
 
 ## Accepted proposal
 
diff --git a/pkg/apis/ceph.rook.io/v1/security.go b/pkg/apis/ceph.rook.io/v1/security.go
index 135c97e0b..01f678898 100644
--- a/pkg/apis/ceph.rook.io/v1/security.go
+++ b/pkg/apis/ceph.rook.io/v1/security.go
@@ -50,7 +50,7 @@ func (kms *KeyManagementServiceSpec) IsVaultKMS() bool {
 
 // IsIBMKeyProtectKMS return whether IBM Key Protect KMS is configured
 func (kms *KeyManagementServiceSpec) IsIBMKeyProtectKMS() bool {
-	return getParam(kms.ConnectionDetails, "KMS_PROVIDER") == "ibmkeyprotect"
+	return getParam(kms.ConnectionDetails, "KMS_PROVIDER") == secrets.TypeIBM
 }
 
 // IsTLSEnabled return KMS TLS details are configured
diff --git a/pkg/daemon/ceph/client/config.go b/pkg/daemon/ceph/client/config.go
index 1ace5a2f2..da2f41494 100644
--- a/pkg/daemon/ceph/client/config.go
+++ b/pkg/daemon/ceph/client/config.go
@@ -292,7 +292,7 @@ func WriteCephConfig(context *clusterd.Context, clusterInfo *ClusterInfo) error
 	}
 	dst, err := ioutil.ReadFile(DefaultConfigFilePath())
 	if err == nil {
-		logger.Debugf("config file @ %s:\n%s", DefaultConfigFilePath(), dst)
+		logger.Debugf("config file @ %s: %s", DefaultConfigFilePath(), dst)
 	} else {
 		logger.Warningf("wrote and copied config file but failed to read it back from %s for logging. %v", DefaultConfigFilePath(), err)
 	}
diff --git a/pkg/daemon/ceph/client/pool.go b/pkg/daemon/ceph/client/pool.go
index 345cebb1d..c4d024521 100644
--- a/pkg/daemon/ceph/client/pool.go
+++ b/pkg/daemon/ceph/client/pool.go
@@ -116,30 +116,6 @@ func GetPoolNamesByID(context *clusterd.Context, clusterInfo *ClusterInfo) (map[
 	return names, nil
 }
 
-func getPoolApplication(context *clusterd.Context, clusterInfo *ClusterInfo, poolName string) (string, error) {
-	args := []string{"osd", "pool", "application", "get", poolName}
-	appDetails, err := NewCephCommand(context, clusterInfo, args).Run()
-	if err != nil {
-		return "", errors.Wrapf(err, "failed to get current application for pool %s", poolName)
-	}
-
-	if len(appDetails) == 0 {
-		// no application name
-		return "", nil
-	}
-	var application map[string]interface{}
-	err = json.Unmarshal([]byte(appDetails), &application)
-	if err != nil {
-		return "", errors.Wrapf(err, "unmarshal failed raw buffer response %s", string(appDetails))
-	}
-	for name := range application {
-		// Return the first application name in the list since only one is expected
-		return name, nil
-	}
-	// No application name assigned
-	return "", nil
-}
-
 // GetPoolDetails gets all the details of a given pool
 func GetPoolDetails(context *clusterd.Context, clusterInfo *ClusterInfo, name string) (CephStoragePoolDetails, error) {
 	args := []string{"osd", "pool", "get", name, "all"}
@@ -258,19 +234,10 @@ func DeletePool(context *clusterd.Context, clusterInfo *ClusterInfo, name string
 }
 
 func givePoolAppTag(context *clusterd.Context, clusterInfo *ClusterInfo, poolName, appName string) error {
-	currentAppName, err := getPoolApplication(context, clusterInfo, poolName)
-	if err != nil {
-		return errors.Wrapf(err, "failed to get application for pool %q", poolName)
-	}
-	if currentAppName == appName {
-		logger.Infof("application %q is already set on pool %q", appName, poolName)
-		return nil
-	}
-
 	args := []string{"osd", "pool", "application", "enable", poolName, appName, confirmFlag}
-	_, err = NewCephCommand(context, clusterInfo, args).Run()
+	_, err := NewCephCommand(context, clusterInfo, args).Run()
 	if err != nil {
-		return errors.Wrapf(err, "failed to enable application %q on pool %q", appName, poolName)
+		return errors.Wrapf(err, "failed to enable application %s on pool %s", appName, poolName)
 	}
 
 	return nil
diff --git a/pkg/daemon/ceph/client/subvolumegroup.go b/pkg/daemon/ceph/client/subvolumegroup.go
index f679b94ed..a73a4863e 100644
--- a/pkg/daemon/ceph/client/subvolumegroup.go
+++ b/pkg/daemon/ceph/client/subvolumegroup.go
@@ -24,34 +24,33 @@ import (
 // CreateCephFSSubVolumeGroup create a CephFS subvolume group.
 // volName is the name of the Ceph FS volume, the same as the CephFilesystem CR name.
 func CreateCephFSSubVolumeGroup(context *clusterd.Context, clusterInfo *ClusterInfo, volName, groupName string) error {
-	logger.Infof("creating cephfs subvolume group %q", volName)
+	logger.Infof("creating cephfs sub volume group %q", volName)
 	//  [--pool_layout <data_pool_name>] [--uid <uid>] [--gid <gid>] [--mode <octal_mode>]
 	args := []string{"fs", "subvolumegroup", "create", volName, groupName}
 	cmd := NewCephCommand(context, clusterInfo, args)
 	cmd.JsonOutput = false
 	output, err := cmd.Run()
 	if err != nil {
-		return errors.Wrapf(err, "failed to create subvolume group %q. %s", volName, output)
+		return errors.Wrapf(err, "failed to create sub volume group %q. %s", volName, output)
 	}
 
-	logger.Infof("successfully created cephfs subvolume group %q", volName)
+	logger.Infof("successfully created cephfs sub volume group %q", volName)
 	return nil
 }
 
 // DeleteCephFSSubVolumeGroup delete a CephFS subvolume group.
 func DeleteCephFSSubVolumeGroup(context *clusterd.Context, clusterInfo *ClusterInfo, volName, groupName string) error {
-	logger.Infof("deleting cephfs subvolume group %q", volName)
+	logger.Infof("deleting cephfs sub volume group %q", volName)
 	//  --force?
 	args := []string{"fs", "subvolumegroup", "rm", volName, groupName}
 	cmd := NewCephCommand(context, clusterInfo, args)
 	cmd.JsonOutput = false
-	output, err := cmd.Run()
+	_, err := cmd.Run()
 	if err != nil {
-		logger.Debugf("failed to delete subvolume group %q. %s. %v", volName, output, err)
 		// Intentionally don't wrap the error so the caller can inspect the return code
 		return err
 	}
 
-	logger.Infof("successfully deleted cephfs subvolume group %q", volName)
+	logger.Infof("successfully deleted cephfs sub volume group %q", volName)
 	return nil
 }
diff --git a/pkg/daemon/ceph/client/upgrade.go b/pkg/daemon/ceph/client/upgrade.go
index 5d7534386..17f8c5d57 100644
--- a/pkg/daemon/ceph/client/upgrade.go
+++ b/pkg/daemon/ceph/client/upgrade.go
@@ -46,7 +46,7 @@ func getCephMonVersionString(context *clusterd.Context, clusterInfo *ClusterInfo
 	args := []string{"version"}
 	buf, err := NewCephCommand(context, clusterInfo, args).Run()
 	if err != nil {
-		return "", errors.Wrapf(err, "failed to run 'ceph version'. %s", string(buf))
+		return "", errors.Wrap(err, "failed to run 'ceph version'")
 	}
 	output := string(buf)
 	logger.Debug(output)
diff --git a/pkg/daemon/ceph/osd/kms/ibm_key_protect.go b/pkg/daemon/ceph/osd/kms/ibm_key_protect.go
index c673372f4..675a5380d 100644
--- a/pkg/daemon/ceph/osd/kms/ibm_key_protect.go
+++ b/pkg/daemon/ceph/osd/kms/ibm_key_protect.go
@@ -18,15 +18,15 @@ package kms
 
 import (
 	kp "github.com/IBM/keyprotect-go-client"
+	"github.com/libopenstorage/secrets"
 	"github.com/pkg/errors"
 )
 
 const (
-	TypeIBM = "ibmkeyprotect"
 	// IbmKeyProtectServiceApiKey is the IBM Key Protect service API key
 	IbmKeyProtectServiceApiKey = "IBM_KP_SERVICE_API_KEY"
 	// IbmKeyProtectInstanceIdKey is the IBM Key Protect instance id
-	IbmKeyProtectInstanceIdKey = "IBM_KP_SERVICE_INSTANCE_ID"
+	IbmKeyProtectInstanceIdKey = "IBM_KP_INSTANCE_ID"
 	// IbmKeyProtectBaseUrlKey is the IBM Key Protect base url
 	IbmKeyProtectBaseUrlKey = "IBM_KP_BASE_URL"
 	// IbmKeyProtectTokenUrlKey is the IBM Key Protect token url
@@ -39,7 +39,7 @@ var (
 	kmsIBMKeyProtectMandatoryConnectionDetails = []string{IbmKeyProtectInstanceIdKey, IbmKeyProtectServiceApiKey}
 	// ErrIbmServiceApiKeyNotSet is returned when IBM_KP_SERVICE_API_KEY is not set
 	ErrIbmServiceApiKeyNotSet = errors.Errorf("%s not set.", IbmKeyProtectServiceApiKey)
-	// ErrIbmInstanceIdKeyNotSet is returned when IBM_KP_SERVICE_INSTANCE_ID is not set
+	// ErrIbmInstanceIdKeyNotSet is returned when IBM_KP_INSTANCE_ID is not set
 	ErrIbmInstanceIdKeyNotSet = errors.Errorf("%s not set.", IbmKeyProtectInstanceIdKey)
 )
 
@@ -83,4 +83,4 @@ func InitKeyProtect(config map[string]string) (*kp.Client, error) {
 }
 
 // IsIBMKeyProtect determines whether the configured KMS is IBM Key Protect
-func (c *Config) IsIBMKeyProtect() bool { return c.Provider == TypeIBM }
+func (c *Config) IsIBMKeyProtect() bool { return c.Provider == secrets.TypeIBM }
diff --git a/pkg/daemon/ceph/osd/kms/kms.go b/pkg/daemon/ceph/osd/kms/kms.go
index cf1734284..ab005c2bf 100644
--- a/pkg/daemon/ceph/osd/kms/kms.go
+++ b/pkg/daemon/ceph/osd/kms/kms.go
@@ -65,8 +65,8 @@ func NewConfig(context *clusterd.Context, clusterSpec *cephv1.ClusterSpec, clust
 		config.Provider = secrets.TypeK8s
 	case secrets.TypeVault:
 		config.Provider = secrets.TypeVault
-	case TypeIBM:
-		config.Provider = TypeIBM
+	case secrets.TypeIBM:
+		config.Provider = secrets.TypeIBM
 	default:
 		logger.Errorf("unsupported kms type %q", Provider)
 	}
@@ -248,7 +248,7 @@ func ValidateConnectionDetails(ctx context.Context, clusterdContext *clusterd.Co
 				return errors.Wrap(err, "failed to set vault kms token to an env var")
 			}
 
-		case TypeIBM:
+		case secrets.TypeIBM:
 			for _, config := range kmsIBMKeyProtectMandatoryTokenDetails {
 				v, ok := kmsToken.Data[config]
 				if !ok || len(v) == 0 {
@@ -281,7 +281,7 @@ func ValidateConnectionDetails(ctx context.Context, clusterdContext *clusterd.Co
 			}
 		}
 
-	case TypeIBM:
+	case secrets.TypeIBM:
 		for _, config := range kmsIBMKeyProtectMandatoryConnectionDetails {
 			if GetParam(securitySpec.KeyManagementService.ConnectionDetails, config) == "" {
 				return errors.Errorf("failed to validate kms config %q. cannot be empty", config)
diff --git a/pkg/daemon/ceph/osd/volume.go b/pkg/daemon/ceph/osd/volume.go
index cf41876d8..1ed586055 100644
--- a/pkg/daemon/ceph/osd/volume.go
+++ b/pkg/daemon/ceph/osd/volume.go
@@ -865,7 +865,7 @@ func GetCephVolumeLVMOSDs(context *clusterd.Context, clusterInfo *client.Cluster
 
 	var lvPath string
 	args := []string{cvMode, "list", lv, "--format", "json"}
-	result, err := callCephVolume(context, args...)
+	result, err := callCephVolume(context, false, args...)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to retrieve ceph-volume %s list results", cvMode)
 	}
@@ -1028,7 +1028,7 @@ func GetCephVolumeRawOSDs(context *clusterd.Context, clusterInfo *client.Cluster
 		args = []string{cvMode, "list", "--format", "json"}
 	}
 
-	result, err := callCephVolume(context, args...)
+	result, err := callCephVolume(context, false, args...)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to retrieve ceph-volume %s list results", cvMode)
 	}
@@ -1145,7 +1145,7 @@ func GetCephVolumeRawOSDs(context *clusterd.Context, clusterInfo *client.Cluster
 	return osds, nil
 }
 
-func callCephVolume(context *clusterd.Context, args ...string) (string, error) {
+func callCephVolume(context *clusterd.Context, requiresCombinedOutput bool, args ...string) (string, error) {
 	// Use stdbuf to capture the python output buffer such that we can write to the pod log as the
 	// logging happens instead of using the default buffering that will log everything after
 	// ceph-volume exits
@@ -1162,6 +1162,10 @@ func callCephVolume(context *clusterd.Context, args ...string) (string, error) {
 
 	// Do not use combined output for "list" calls, otherwise we will get stderr is the output and this will break the json unmarshall
 	f := context.Executor.ExecuteCommandWithOutput
+	if requiresCombinedOutput {
+		// If the action is preparing we need the combined output
+		f = context.Executor.ExecuteCommandWithCombinedOutput
+	}
 	co, err := f(baseCommand, append(baseArgs, args...)...)
 	if err != nil {
 		// Print c-v log before exiting with failure
diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go
index c9cf926f4..3a6ffdb6b 100755
--- a/pkg/operator/ceph/cluster/cluster.go
+++ b/pkg/operator/ceph/cluster/cluster.go
@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"os/exec"
 	"path"
+	"sync"
 	"syscall"
 
 	"github.com/pkg/errors"
@@ -66,7 +67,7 @@ type clusterHealth struct {
 	internalCancel context.CancelFunc
 }
 
-func newCluster(c *cephv1.CephCluster, context *clusterd.Context, ownerInfo *k8sutil.OwnerInfo) *cluster {
+func newCluster(c *cephv1.CephCluster, context *clusterd.Context, csiMutex *sync.Mutex, ownerInfo *k8sutil.OwnerInfo) *cluster {
 	return &cluster{
 		// at this phase of the cluster creation process, the identity components of the cluster are
 		// not yet established. we reserve this struct which is filled in as soon as the cluster's
@@ -78,7 +79,7 @@ func newCluster(c *cephv1.CephCluster, context *clusterd.Context, ownerInfo *k8s
 		namespacedName:     types.NamespacedName{Namespace: c.Namespace, Name: c.Name},
 		monitoringRoutines: make(map[string]*clusterHealth),
 		ownerInfo:          ownerInfo,
-		mons:               mon.New(context, c.Namespace, c.Spec, ownerInfo),
+		mons:               mon.New(context, c.Namespace, c.Spec, ownerInfo, csiMutex),
 	}
 }
 
diff --git a/pkg/operator/ceph/cluster/cluster_external.go b/pkg/operator/ceph/cluster/cluster_external.go
index fb97a9adf..284d34a9a 100644
--- a/pkg/operator/ceph/cluster/cluster_external.go
+++ b/pkg/operator/ceph/cluster/cluster_external.go
@@ -111,7 +111,7 @@ func (c *ClusterController) configureExternalCephCluster(cluster *cluster) error
 	}
 
 	// Save CSI configmap
-	err = csi.SaveClusterConfig(c.context.Clientset, c.namespacedName.Namespace, cluster.ClusterInfo, &csi.CsiClusterConfigEntry{Monitors: csi.MonEndpoints(cluster.ClusterInfo.Monitors)})
+	err = csi.SaveClusterConfig(c.context.Clientset, c.namespacedName.Namespace, cluster.ClusterInfo, c.csiConfigMutex)
 	if err != nil {
 		return errors.Wrap(err, "failed to update csi cluster config")
 	}
diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go
index f1fa42874..d232a821f 100644
--- a/pkg/operator/ceph/cluster/controller.go
+++ b/pkg/operator/ceph/cluster/controller.go
@@ -21,6 +21,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"sync"
 
 	"github.com/coreos/pkg/capnslog"
 	"github.com/pkg/errors"
@@ -84,6 +85,7 @@ type ClusterController struct {
 	context        *clusterd.Context
 	rookImage      string
 	clusterMap     map[string]*cluster
+	csiConfigMutex *sync.Mutex
 	osdChecker     *osd.OSDHealthMonitor
 	client         client.Client
 	namespacedName types.NamespacedName
@@ -325,9 +327,10 @@ func (r *ReconcileCephCluster) reconcileDelete(cephCluster *cephv1.CephCluster)
 // NewClusterController create controller for watching cluster custom resources created
 func NewClusterController(context *clusterd.Context, rookImage string) *ClusterController {
 	return &ClusterController{
-		context:    context,
-		rookImage:  rookImage,
-		clusterMap: make(map[string]*cluster),
+		context:        context,
+		rookImage:      rookImage,
+		clusterMap:     make(map[string]*cluster),
+		csiConfigMutex: &sync.Mutex{},
 	}
 }
 
@@ -340,7 +343,7 @@ func (c *ClusterController) reconcileCephCluster(clusterObj *cephv1.CephCluster,
 	cluster, ok := c.clusterMap[clusterObj.Namespace]
 	if !ok {
 		// It's a new cluster so let's populate the struct
-		cluster = newCluster(clusterObj, c.context, ownerInfo)
+		cluster = newCluster(clusterObj, c.context, c.csiConfigMutex, ownerInfo)
 	}
 	cluster.namespacedName = c.namespacedName
 
diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go
index bf7988ba3..070ce2edc 100644
--- a/pkg/operator/ceph/cluster/mon/mon.go
+++ b/pkg/operator/ceph/cluster/mon/mon.go
@@ -120,6 +120,7 @@ type Cluster struct {
 	monTimeoutList     map[string]time.Time
 	mapping            *Mapping
 	ownerInfo          *k8sutil.OwnerInfo
+	csiConfigMutex     *sync.Mutex
 	isUpgrade          bool
 	arbiterMon         string
 }
@@ -163,7 +164,7 @@ type SchedulingResult struct {
 }
 
 // New creates an instance of a mon cluster
-func New(context *clusterd.Context, namespace string, spec cephv1.ClusterSpec, ownerInfo *k8sutil.OwnerInfo) *Cluster {
+func New(context *clusterd.Context, namespace string, spec cephv1.ClusterSpec, ownerInfo *k8sutil.OwnerInfo, csiConfigMutex *sync.Mutex) *Cluster {
 	return &Cluster{
 		context:        context,
 		spec:           spec,
@@ -174,7 +175,8 @@ func New(context *clusterd.Context, namespace string, spec cephv1.ClusterSpec, o
 		mapping: &Mapping{
 			Schedule: map[string]*MonScheduleInfo{},
 		},
-		ownerInfo: ownerInfo,
+		ownerInfo:      ownerInfo,
+		csiConfigMutex: csiConfigMutex,
 	}
 }
 
@@ -1040,7 +1042,7 @@ func (c *Cluster) saveMonConfig() error {
 		return errors.Wrap(err, "failed to write connection config for new mons")
 	}
 
-	if err := csi.SaveClusterConfig(c.context.Clientset, c.Namespace, c.ClusterInfo, &csi.CsiClusterConfigEntry{Monitors: csi.MonEndpoints(c.ClusterInfo.Monitors)}); err != nil {
+	if err := csi.SaveClusterConfig(c.context.Clientset, c.Namespace, c.ClusterInfo, c.csiConfigMutex); err != nil {
 		return errors.Wrap(err, "failed to update csi cluster config")
 	}
 
diff --git a/pkg/operator/ceph/cluster/mon/predicate.go b/pkg/operator/ceph/cluster/mon/predicate.go
index 9a71666fe..b858973ce 100644
--- a/pkg/operator/ceph/cluster/mon/predicate.go
+++ b/pkg/operator/ceph/cluster/mon/predicate.go
@@ -96,7 +96,7 @@ func wereMonEndpointsUpdated(oldCMData, newCMData map[string]string) bool {
 
 			// Iterate over the map and compare the values
 			for _, v := range oldKeys {
-				if !reflect.DeepEqual(*oldMappingToGo.Schedule[v], *newMappingToGo.Schedule[v]) {
+				if !reflect.DeepEqual(oldMappingToGo.Schedule[v], newMappingToGo.Schedule[v]) {
 					logger.Debugf("oldMappingToGo.Schedule[v] AND newMappingToGo.Schedule[v]: %v | %v", oldMappingToGo.Schedule[v], newMappingToGo.Schedule[v])
 					return true
 				}
diff --git a/pkg/operator/ceph/cluster/osd/envs.go b/pkg/operator/ceph/cluster/osd/envs.go
index 7e3cf1831..11b714d88 100644
--- a/pkg/operator/ceph/cluster/osd/envs.go
+++ b/pkg/operator/ceph/cluster/osd/envs.go
@@ -32,7 +32,6 @@ const (
 	osdWalSizeEnvVarName      = "ROOK_OSD_WAL_SIZE"
 	osdsPerDeviceEnvVarName   = "ROOK_OSDS_PER_DEVICE"
 	osdDeviceClassEnvVarName  = "ROOK_OSD_DEVICE_CLASS"
-	osdConfigMapOverrideName  = "rook-ceph-osd-env-override"
 	// EncryptedDeviceEnvVarName is used in the pod spec to indicate whether the OSD is encrypted or not
 	EncryptedDeviceEnvVarName = "ROOK_ENCRYPTED_DEVICE"
 	PVCNameEnvVarName         = "ROOK_PVC_NAME"
@@ -214,19 +213,6 @@ func osdActivateEnvVar() []v1.EnvVar {
 	return append(cephVolumeEnvVar(), monEnvVars...)
 }
 
-func getEnvFromSources() []v1.EnvFromSource {
-	optionalConfigMapRef := true
-
-	return []v1.EnvFromSource{
-		{
-			ConfigMapRef: &v1.ConfigMapEnvSource{
-				LocalObjectReference: v1.LocalObjectReference{Name: osdConfigMapOverrideName},
-				Optional:             &optionalConfigMapRef,
-			},
-		},
-	}
-}
-
 func getTcmallocMaxTotalThreadCacheBytes(tcmallocMaxTotalThreadCacheBytes string) v1.EnvVar {
 	var value string
 	// If empty we read the default value from the file coming with the package
diff --git a/pkg/operator/ceph/cluster/osd/health.go b/pkg/operator/ceph/cluster/osd/health.go
index 821c146b7..b8319bd0f 100644
--- a/pkg/operator/ceph/cluster/osd/health.go
+++ b/pkg/operator/ceph/cluster/osd/health.go
@@ -186,13 +186,13 @@ func (m *OSDHealthMonitor) removeOSDDeploymentIfSafeToDestroy(outOSDid int) erro
 
 // updateCephStorage updates the CR with deviceclass details
 func (m *OSDHealthMonitor) updateCephStatus(devices []string) {
-	cephCluster := cephv1.CephCluster{}
+	cephCluster := &cephv1.CephCluster{}
 	cephClusterStorage := cephv1.CephStorage{}
 
 	for _, device := range devices {
 		cephClusterStorage.DeviceClasses = append(cephClusterStorage.DeviceClasses, cephv1.DeviceClasses{Name: device})
 	}
-	err := m.context.Client.Get(m.clusterInfo.Context, m.clusterInfo.NamespacedName(), &cephCluster)
+	err := m.context.Client.Get(m.clusterInfo.Context, m.clusterInfo.NamespacedName(), cephCluster)
 	if err != nil {
 		if kerrors.IsNotFound(err) {
 			logger.Debug("CephCluster resource not found. Ignoring since object must be deleted.")
@@ -201,9 +201,9 @@ func (m *OSDHealthMonitor) updateCephStatus(devices []string) {
 		logger.Errorf("failed to retrieve ceph cluster %q to update ceph Storage. %v", m.clusterInfo.NamespacedName().Name, err)
 		return
 	}
-	if !reflect.DeepEqual(cephCluster.Status.CephStorage, cephClusterStorage) {
+	if !reflect.DeepEqual(cephCluster.Status.CephStorage, &cephClusterStorage) {
 		cephCluster.Status.CephStorage = &cephClusterStorage
-		if err := reporting.UpdateStatus(m.context.Client, &cephCluster); err != nil {
+		if err := reporting.UpdateStatus(m.context.Client, cephCluster); err != nil {
 			logger.Errorf("failed to update cluster %q Storage. %v", m.clusterInfo.NamespacedName().Name, err)
 			return
 		}
diff --git a/pkg/operator/ceph/cluster/osd/provision_spec.go b/pkg/operator/ceph/cluster/osd/provision_spec.go
index c4a7d32b7..e9dbec33f 100644
--- a/pkg/operator/ceph/cluster/osd/provision_spec.go
+++ b/pkg/operator/ceph/cluster/osd/provision_spec.go
@@ -302,7 +302,6 @@ func (c *Cluster) provisionOSDContainer(osdProps osdProperties, copyBinariesMoun
 		Image:        c.spec.CephVersion.Image,
 		VolumeMounts: volumeMounts,
 		Env:          envVars,
-		EnvFrom:      getEnvFromSources(),
 		SecurityContext: &v1.SecurityContext{
 			Privileged:             &privileged,
 			RunAsUser:              &runAsUser,
diff --git a/pkg/operator/ceph/cluster/osd/spec.go b/pkg/operator/ceph/cluster/osd/spec.go
index 09e195e73..2f57fa52a 100644
--- a/pkg/operator/ceph/cluster/osd/spec.go
+++ b/pkg/operator/ceph/cluster/osd/spec.go
@@ -466,7 +466,6 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 				Image:           c.rookVersion,
 				VolumeMounts:    configVolumeMounts,
 				Env:             configEnvVars,
-				EnvFrom:         getEnvFromSources(),
 				SecurityContext: securityContext,
 			})
 	}
@@ -550,7 +549,6 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 					Image:           c.spec.CephVersion.Image,
 					VolumeMounts:    volumeMounts,
 					Env:             envVars,
-					EnvFrom:         getEnvFromSources(),
 					Resources:       osdProps.resources,
 					SecurityContext: securityContext,
 					StartupProbe:    controller.GenerateStartupProbeExecDaemon(opconfig.OsdType, osdID),
@@ -769,7 +767,6 @@ func (c *Cluster) getActivateOSDInitContainer(configDir, namespace, osdID string
 		VolumeMounts:    volMounts,
 		SecurityContext: controller.PrivilegedContext(true),
 		Env:             envVars,
-		EnvFrom:         getEnvFromSources(),
 		Resources:       osdProps.resources,
 	}
 
diff --git a/pkg/operator/ceph/cluster/rbd/controller.go b/pkg/operator/ceph/cluster/rbd/controller.go
index b9b42a12e..601890eab 100644
--- a/pkg/operator/ceph/cluster/rbd/controller.go
+++ b/pkg/operator/ceph/cluster/rbd/controller.go
@@ -208,7 +208,7 @@ func (r *ReconcileCephRBDMirror) reconcile(request reconcile.Request) (reconcile
 	// If the version of the Ceph monitor differs from the CephCluster CR image version we assume
 	// the cluster is being upgraded. So the controller will just wait for the upgrade to finish and
 	// then versions should match. Obviously using the cmd reporter job adds up to the deployment time
-	if !reflect.DeepEqual(*runningCephVersion, *desiredCephVersion) {
+	if !reflect.DeepEqual(runningCephVersion, desiredCephVersion) {
 		// Upgrade is in progress, let's wait for the mons to be done
 		return opcontroller.WaitForRequeueIfCephClusterIsUpgrading,
 			opcontroller.ErrorCephUpgradingRequeue(desiredCephVersion, runningCephVersion)
diff --git a/pkg/operator/ceph/controller/spec.go b/pkg/operator/ceph/controller/spec.go
index d7e185f07..e7b9b51cb 100644
--- a/pkg/operator/ceph/controller/spec.go
+++ b/pkg/operator/ceph/controller/spec.go
@@ -743,11 +743,8 @@ func ConfigureExternalMetricsEndpoint(ctx *clusterd.Context, monitoringSpec ceph
 	}
 
 	// If endpoints are identical there is nothing to do
-	// First check for nil pointers otherwise dereferencing a nil pointer will cause a panic
-	if endpoint != nil && currentEndpoints != nil {
-		if reflect.DeepEqual(*currentEndpoints, *endpoint) {
-			return nil
-		}
+	if reflect.DeepEqual(currentEndpoints, endpoint) {
+		return nil
 	}
 	logger.Debugf("diff between current endpoint and newly generated one: %v \n", cmp.Diff(currentEndpoints, endpoint, cmp.Comparer(func(x, y resource.Quantity) bool { return x.Cmp(y) == 0 })))
 
diff --git a/pkg/operator/ceph/csi/cluster_config.go b/pkg/operator/ceph/csi/cluster_config.go
index 3f4ad3b8c..03e76934f 100644
--- a/pkg/operator/ceph/csi/cluster_config.go
+++ b/pkg/operator/ceph/csi/cluster_config.go
@@ -33,22 +33,15 @@ import (
 )
 
 var (
-	logger      = capnslog.NewPackageLogger("github.com/rook/rook", "ceph-csi")
-	configMutex sync.Mutex
+	logger = capnslog.NewPackageLogger("github.com/rook/rook", "ceph-csi")
 )
 
-type CsiClusterConfigEntry struct {
-	ClusterID      string         `json:"clusterID"`
-	Monitors       []string       `json:"monitors"`
-	CephFS         *CsiCephFSSpec `json:"cephFS,omitempty"`
-	RadosNamespace string         `json:"radosNamespace,omitempty"`
+type csiClusterConfigEntry struct {
+	ClusterID string   `json:"clusterID"`
+	Monitors  []string `json:"monitors"`
 }
 
-type CsiCephFSSpec struct {
-	SubvolumeGroup string `json:"subvolumeGroup,omitempty"`
-}
-
-type csiClusterConfig []CsiClusterConfigEntry
+type csiClusterConfig []csiClusterConfigEntry
 
 // FormatCsiClusterConfig returns a json-formatted string containing
 // the cluster-to-mon mapping required to configure ceph csi.
@@ -86,7 +79,7 @@ func formatCsiClusterConfig(cc csiClusterConfig) (string, error) {
 	return string(ccJson), nil
 }
 
-func MonEndpoints(mons map[string]*cephclient.MonInfo) []string {
+func monEndpoints(mons map[string]*cephclient.MonInfo) []string {
 	endpoints := make([]string, 0)
 	for _, m := range mons {
 		endpoints = append(endpoints, m.Endpoint)
@@ -96,59 +89,32 @@ func MonEndpoints(mons map[string]*cephclient.MonInfo) []string {
 
 // updateCsiClusterConfig returns a json-formatted string containing
 // the cluster-to-mon mapping required to configure ceph csi.
-func updateCsiClusterConfig(curr, clusterKey string, newCsiClusterConfigEntry *CsiClusterConfigEntry) (string, error) {
+func updateCsiClusterConfig(
+	curr, clusterKey string, mons map[string]*cephclient.MonInfo) (string, error) {
+
 	var (
 		cc     csiClusterConfig
-		centry CsiClusterConfigEntry
+		centry csiClusterConfigEntry
 		found  bool
 	)
-
 	cc, err := parseCsiClusterConfig(curr)
 	if err != nil {
 		return "", errors.Wrap(err, "failed to parse current csi cluster config")
 	}
 
-	// Regardless of which controllers call updateCsiClusterConfig(), the values will be preserved since
-	// a lock is acquired for the update operation. So concurrent updates (rare event) will block and
-	// wait for the other update to complete. Monitors and Subvolumegroup will be updated
-	// independently and won't collide.
 	for i, centry := range cc {
 		if centry.ClusterID == clusterKey {
-			// If the new entry is nil, this means the entry is being deleted so remove it from the list
-			if newCsiClusterConfigEntry == nil {
-				cc = append(cc[:i], cc[i+1:]...)
-				found = true
-				break
-			}
-			centry.Monitors = newCsiClusterConfigEntry.Monitors
-			if newCsiClusterConfigEntry.CephFS != nil && newCsiClusterConfigEntry.CephFS.SubvolumeGroup != "" {
-				centry.CephFS = newCsiClusterConfigEntry.CephFS
-			}
-			if newCsiClusterConfigEntry.RadosNamespace != "" {
-				centry.RadosNamespace = newCsiClusterConfigEntry.RadosNamespace
-			}
+			centry.Monitors = monEndpoints(mons)
 			found = true
 			cc[i] = centry
 			break
 		}
 	}
 	if !found {
-		// If it's the first time we create the cluster, the entry does not exist, so the removal
-		// will fail with a dangling pointer
-		if newCsiClusterConfigEntry != nil {
-			centry.ClusterID = clusterKey
-			centry.Monitors = newCsiClusterConfigEntry.Monitors
-			// Add a condition not to fill with empty values
-			if newCsiClusterConfigEntry.CephFS != nil && newCsiClusterConfigEntry.CephFS.SubvolumeGroup != "" {
-				centry.CephFS = newCsiClusterConfigEntry.CephFS
-			}
-			if newCsiClusterConfigEntry.RadosNamespace != "" {
-				centry.RadosNamespace = newCsiClusterConfigEntry.RadosNamespace
-			}
-			cc = append(cc, centry)
-		}
+		centry.ClusterID = clusterKey
+		centry.Monitors = monEndpoints(mons)
+		cc = append(cc, centry)
 	}
-
 	return formatCsiClusterConfig(cc)
 }
 
@@ -189,24 +155,24 @@ func CreateCsiConfigMap(namespace string, clientset kubernetes.Interface, ownerI
 // value that is provided to ceph-csi uses in the storage class.
 // The locker l is typically a mutex and is used to prevent the config
 // map from being updated for multiple clusters simultaneously.
-func SaveClusterConfig(clientset kubernetes.Interface, clusterNamespace string, clusterInfo *cephclient.ClusterInfo, newCsiClusterConfigEntry *CsiClusterConfigEntry) error {
+func SaveClusterConfig(
+	clientset kubernetes.Interface, clusterNamespace string,
+	clusterInfo *cephclient.ClusterInfo, l sync.Locker) error {
 	if !CSIEnabled() {
-		logger.Debug("skipping csi config update because csi is not enabled")
 		return nil
 	}
-
-	configMutex.Lock()
-	defer configMutex.Unlock()
-
+	l.Lock()
+	defer l.Unlock()
 	// csi is deployed into the same namespace as the operator
 	csiNamespace := os.Getenv(k8sutil.PodNamespaceEnvVar)
 	if csiNamespace == "" {
-		return errors.Errorf("namespace value missing for %q", k8sutil.PodNamespaceEnvVar)
+		return errors.Errorf("namespace value missing for %s", k8sutil.PodNamespaceEnvVar)
 	}
-	logger.Debugf("using %q for csi configmap namespace", csiNamespace)
+	logger.Debugf("Using %+v for CSI ConfigMap Namespace", csiNamespace)
 
 	// fetch current ConfigMap contents
-	configMap, err := clientset.CoreV1().ConfigMaps(csiNamespace).Get(clusterInfo.Context, ConfigName, metav1.GetOptions{})
+	configMap, err := clientset.CoreV1().ConfigMaps(csiNamespace).Get(clusterInfo.Context,
+		ConfigName, metav1.GetOptions{})
 	if err != nil {
 		return errors.Wrap(err, "failed to fetch current csi config map")
 	}
@@ -216,8 +182,8 @@ func SaveClusterConfig(clientset kubernetes.Interface, clusterNamespace string,
 	if currData == "" {
 		currData = "[]"
 	}
-
-	newData, err := updateCsiClusterConfig(currData, clusterNamespace, newCsiClusterConfigEntry)
+	newData, err := updateCsiClusterConfig(
+		currData, clusterNamespace, clusterInfo.Monitors)
 	if err != nil {
 		return errors.Wrap(err, "failed to update csi config map data")
 	}
@@ -225,7 +191,7 @@ func SaveClusterConfig(clientset kubernetes.Interface, clusterNamespace string,
 
 	// update ConfigMap with new contents
 	if _, err := clientset.CoreV1().ConfigMaps(csiNamespace).Update(clusterInfo.Context, configMap, metav1.UpdateOptions{}); err != nil {
-		return errors.Wrap(err, "failed to update csi config map")
+		return errors.Wrapf(err, "failed to update csi config map")
 	}
 
 	return nil
diff --git a/pkg/operator/ceph/csi/predicate.go b/pkg/operator/ceph/csi/predicate.go
index bfe4055e8..2742475e9 100644
--- a/pkg/operator/ceph/csi/predicate.go
+++ b/pkg/operator/ceph/csi/predicate.go
@@ -22,6 +22,7 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/operator/ceph/controller"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	v1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
@@ -46,7 +47,7 @@ func predicateController(ctx context.Context, c client.Client, opNamespace strin
 			// If a Ceph Cluster is created we want to reconcile the csi driver
 			if cephCluster, ok := e.Object.(*cephv1.CephCluster); ok {
 				// If there are more than one ceph cluster in the same namespace do not reconcile
-				if opcontroller.DuplicateCephClusters(ctx, c, e.Object, false) {
+				if controller.DuplicateCephClusters(ctx, c, e.Object, false) {
 					return false
 				}
 
diff --git a/pkg/operator/ceph/csi/spec.go b/pkg/operator/ceph/csi/spec.go
index 86c79da4c..cad77a8f1 100644
--- a/pkg/operator/ceph/csi/spec.go
+++ b/pkg/operator/ceph/csi/spec.go
@@ -101,14 +101,14 @@ var (
 // manually challenging.
 var (
 	// image names
-	DefaultCSIPluginImage         = "quay.io/cephcsi/cephcsi:v3.5.1"
+	DefaultCSIPluginImage         = "quay.io/cephcsi/cephcsi:v3.5.0"
 	DefaultRegistrarImage         = "k8s.gcr.io/sig-storage/csi-node-driver-registrar:v2.4.0"
 	DefaultProvisionerImage       = "k8s.gcr.io/sig-storage/csi-provisioner:v3.1.0"
 	DefaultAttacherImage          = "k8s.gcr.io/sig-storage/csi-attacher:v3.4.0"
-	DefaultSnapshotterImage       = "k8s.gcr.io/sig-storage/csi-snapshotter:v5.0.1"
-	DefaultResizerImage           = "k8s.gcr.io/sig-storage/csi-resizer:v1.4.0"
-	DefaultVolumeReplicationImage = "quay.io/csiaddons/volumereplication-operator:v0.3.0"
-	DefaultCSIAddonsImage         = "quay.io/csiaddons/k8s-sidecar:v0.2.1"
+	DefaultSnapshotterImage       = "k8s.gcr.io/sig-storage/csi-snapshotter:v4.2.0"
+	DefaultResizerImage           = "k8s.gcr.io/sig-storage/csi-resizer:v1.3.0"
+	DefaultVolumeReplicationImage = "quay.io/csiaddons/volumereplication-operator:v0.1.0"
+	DefaultCSIAddonsImage         = "quay.io/csiaddons/k8s-sidecar:v0.1.0"
 
 	// Local package template path for RBD
 	//go:embed template/rbd/csi-rbdplugin.yaml
diff --git a/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin-provisioner-dep.yaml b/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin-provisioner-dep.yaml
index 9088d158d..b586ffb94 100644
--- a/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin-provisioner-dep.yaml
+++ b/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin-provisioner-dep.yaml
@@ -139,7 +139,7 @@ spec:
             - "--pod=$(POD_NAME)"
             - "--namespace=$(POD_NAMESPACE)"
             - "--pod-uid=$(POD_UID)"
-            - "--stagingpath={{ .KubeletDirPath }}/plugins/kubernetes.io/csi/"
+            - "--stagingpath={{ .KubeletDirPath }}/plugins/kubernetes.io/csi/pv/"
           ports:
             - containerPort: {{ .CSIAddonsPort }}
           env:
diff --git a/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin.yaml b/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin.yaml
index e688e6af2..2b1ae4fcb 100644
--- a/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin.yaml
+++ b/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin.yaml
@@ -132,7 +132,7 @@ spec:
             - "--pod=$(POD_NAME)"
             - "--namespace=$(POD_NAMESPACE)"
             - "--pod-uid=$(POD_UID)"
-            - "--stagingpath={{ .KubeletDirPath }}/plugins/kubernetes.io/csi/"
+            - "--stagingpath={{ .KubeletDirPath }}/plugins/kubernetes.io/csi/pv/"
           ports:
             - containerPort: {{ .CSIAddonsPort }}
           env:
diff --git a/pkg/operator/ceph/file/controller.go b/pkg/operator/ceph/file/controller.go
index d15be7b48..afc0ec253 100644
--- a/pkg/operator/ceph/file/controller.go
+++ b/pkg/operator/ceph/file/controller.go
@@ -302,7 +302,7 @@ func (r *ReconcileCephFilesystem) reconcile(request reconcile.Request) (reconcil
 	// If the version of the Ceph monitor differs from the CephCluster CR image version we assume
 	// the cluster is being upgraded. So the controller will just wait for the upgrade to finish and
 	// then versions should match. Obviously using the cmd reporter job adds up to the deployment time
-	if !reflect.DeepEqual(*runningCephVersion, *desiredCephVersion) {
+	if !reflect.DeepEqual(runningCephVersion, desiredCephVersion) {
 		// Upgrade is in progress, let's wait for the mons to be done
 		return opcontroller.WaitForRequeueIfCephClusterIsUpgrading,
 			opcontroller.ErrorCephUpgradingRequeue(desiredCephVersion, runningCephVersion)
diff --git a/pkg/operator/ceph/file/mirror/controller.go b/pkg/operator/ceph/file/mirror/controller.go
index ee595654c..001c7c838 100644
--- a/pkg/operator/ceph/file/mirror/controller.go
+++ b/pkg/operator/ceph/file/mirror/controller.go
@@ -195,7 +195,7 @@ func (r *ReconcileFilesystemMirror) reconcile(request reconcile.Request) (reconc
 	// If the version of the Ceph monitor differs from the CephCluster CR image version we assume
 	// the cluster is being upgraded. So the controller will just wait for the upgrade to finish and
 	// then versions should match. Obviously using the cmd reporter job adds up to the deployment time
-	if !reflect.DeepEqual(*runningCephVersion, *desiredCephVersion) {
+	if !reflect.DeepEqual(runningCephVersion, desiredCephVersion) {
 		// Upgrade is in progress, let's wait for the mons to be done
 		return opcontroller.WaitForRequeueIfCephClusterIsUpgrading,
 			opcontroller.ErrorCephUpgradingRequeue(desiredCephVersion, runningCephVersion)
diff --git a/pkg/operator/ceph/file/subvolumegroup/controller.go b/pkg/operator/ceph/file/subvolumegroup/controller.go
index f5f538ef2..c65434723 100644
--- a/pkg/operator/ceph/file/subvolumegroup/controller.go
+++ b/pkg/operator/ceph/file/subvolumegroup/controller.go
@@ -40,9 +40,7 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
-	"github.com/rook/rook/pkg/operator/ceph/csi"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
-	"github.com/rook/rook/pkg/operator/k8sutil"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -144,7 +142,7 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) reconcile(request reconcile.Requ
 	}
 
 	// Make sure a CephCluster is present otherwise do nothing
-	cephCluster, isReadyToReconcile, cephClusterExists, reconcileResponse := opcontroller.IsReadyToReconcile(r.opManagerContext, r.client, request.NamespacedName, controllerName)
+	_, isReadyToReconcile, cephClusterExists, reconcileResponse := opcontroller.IsReadyToReconcile(r.opManagerContext, r.client, request.NamespacedName, controllerName)
 	if !isReadyToReconcile {
 		// This handles the case where the Ceph Cluster is gone and we want to delete that CR
 		// We skip the deleteSubVolumeGroup() function since everything is gone already
@@ -175,23 +173,9 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) reconcile(request reconcile.Requ
 	// DELETE: the CR was deleted
 	if !cephFilesystemSubVolumeGroup.GetDeletionTimestamp().IsZero() {
 		logger.Debugf("deleting subvolume group %q", cephFilesystemSubVolumeGroup.Name)
-		// On external cluster, we don't delete the subvolume group, it has to be deleted manually
-		if cephCluster.Spec.External.Enable {
-			logger.Warning("external subvolume group deletion is not supported, delete it manually")
-		} else {
-			err := r.deleteSubVolumeGroup(cephFilesystemSubVolumeGroup)
-			if err != nil {
-				if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) {
-					logger.Info(opcontroller.OperatorNotInitializedMessage)
-					return opcontroller.WaitForRequeueIfOperatorNotInitialized, nil
-				}
-				return reconcile.Result{}, errors.Wrapf(err, "failed to delete ceph ceph filesystem subvolume group %q", cephFilesystemSubVolumeGroup.Name)
-			}
-		}
-
-		err = csi.SaveClusterConfig(r.context.Clientset, buildClusterID(cephFilesystemSubVolumeGroup), r.clusterInfo, nil)
+		err := r.deleteSubVolumeGroup(cephFilesystemSubVolumeGroup)
 		if err != nil {
-			return reconcile.Result{}, errors.Wrap(err, "failed to save cluster config")
+			return reconcile.Result{}, errors.Wrapf(err, "failed to delete ceph ceph filesystem subvolume group %q", cephFilesystemSubVolumeGroup.Name)
 		}
 
 		// Remove finalizer
@@ -205,65 +189,37 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) reconcile(request reconcile.Requ
 	}
 
 	// Build the NamespacedName to fetch the Filesystem and make sure it exists, if not we cannot
-	// create the subvolume group
-	// On external mode the filesystem is created externally, so we don't need to check for CRD and
-	// assume it's there
-	if !cephCluster.Spec.External.Enable {
-		cephFilesystem := &cephv1.CephFilesystem{}
-		cephFilesystemNamespacedName := types.NamespacedName{Name: cephFilesystemSubVolumeGroup.Spec.FilesystemName, Namespace: request.Namespace}
-		err = r.client.Get(r.opManagerContext, cephFilesystemNamespacedName, cephFilesystem)
-		if err != nil {
-			if kerrors.IsNotFound(err) {
-				return reconcile.Result{}, errors.Wrapf(err, "failed to fetch ceph filesystem %q, cannot create subvolume group %q", cephFilesystemSubVolumeGroup.Spec.FilesystemName, cephFilesystemSubVolumeGroup.Name)
-			}
-			// Error reading the object - requeue the request.
-			return reconcile.Result{}, errors.Wrap(err, "failed to get cephFilesystemSubVolumeGroup")
-		}
-
-		// If the CephFilesystem is not ready to accept commands, we should wait for it to be ready
-		if cephFilesystem.Status.Phase != cephv1.ConditionReady {
-			// We know the CR is present so it should a matter of second for it to become ready
-			return reconcile.Result{Requeue: true, RequeueAfter: 10 * time.Second}, errors.Wrapf(err, "failed to fetch ceph filesystem %q, cannot create subvolume group %q", cephFilesystemSubVolumeGroup.Spec.FilesystemName, cephFilesystemSubVolumeGroup.Name)
+	// create the subvolumegroup
+	cephFilesystem := &cephv1.CephFilesystem{}
+	cephFilesystemNamespacedName := types.NamespacedName{Name: cephFilesystemSubVolumeGroup.Spec.FilesystemName, Namespace: request.Namespace}
+	err = r.client.Get(r.opManagerContext, cephFilesystemNamespacedName, cephFilesystem)
+	if err != nil {
+		if kerrors.IsNotFound(err) {
+			return reconcile.Result{}, errors.Wrapf(err, "failed to fetch ceph filesystem %q, cannot create subvolumegroup %q", cephFilesystemSubVolumeGroup.Spec.FilesystemName, cephFilesystemSubVolumeGroup.Name)
 		}
+		// Error reading the object - requeue the request.
+		return reconcile.Result{}, errors.Wrap(err, "failed to get cephFilesystemSubVolumeGroup")
 	}
 
-	// Create or Update ceph filesystem subvolume group
-	// On external mode the subvolume group is created externally, so we don't need to try to create
-	// it and assume it's there already
-	if cephCluster.Spec.External.Enable {
-		logger.Debug("external subvolume group creation is not supported, create it manually, the controller will assume it's there")
-	} else {
-		err = r.createOrUpdateSubVolumeGroup(cephFilesystemSubVolumeGroup)
-		if err != nil {
-			if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) {
-				logger.Info(opcontroller.OperatorNotInitializedMessage)
-				return opcontroller.WaitForRequeueIfOperatorNotInitialized, nil
-			}
-			r.updateStatus(r.client, request.NamespacedName, cephv1.ConditionFailure)
-			return reconcile.Result{}, errors.Wrapf(err, "failed to create or update ceph filesystem subvolume group %q", cephFilesystemSubVolumeGroup.Name)
-		}
+	// If the CephFilesystem is not ready to accept commands, we should wait for it to be ready
+	if cephFilesystem.Status.Phase != cephv1.ConditionReady {
+		// We know the CR is present so it should a matter of second for it to become ready
+		return reconcile.Result{Requeue: true, RequeueAfter: 10 * time.Second}, errors.Wrapf(err, "failed to fetch ceph filesystem %q, cannot create subvolumegroup %q", cephFilesystemSubVolumeGroup.Spec.FilesystemName, cephFilesystemSubVolumeGroup.Name)
 	}
 
-	// Update CSI config map
-	// If the mon endpoints change, the mon health check go routine will take care of updating the
-	// config map, so no special care is needed in this controller
-	csiClusterConfigEntry := csi.CsiClusterConfigEntry{
-		Monitors: csi.MonEndpoints(r.clusterInfo.Monitors),
-		CephFS: &csi.CsiCephFSSpec{
-			SubvolumeGroup: cephFilesystemSubVolumeGroup.Name,
-		},
-	}
-	err = csi.SaveClusterConfig(r.context.Clientset, buildClusterID(cephFilesystemSubVolumeGroup), r.clusterInfo, &csiClusterConfigEntry)
+	// Create or Update ceph filesystem subvolume group
+	err = r.createOrUpdateSubVolumeGroup(cephFilesystemSubVolumeGroup)
 	if err != nil {
-		return reconcile.Result{}, errors.Wrap(err, "failed to save cluster config")
+		if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) {
+			logger.Info(opcontroller.OperatorNotInitializedMessage)
+			return opcontroller.WaitForRequeueIfOperatorNotInitialized, nil
+		}
+		r.updateStatus(r.client, request.NamespacedName, cephv1.ConditionFailure)
+		return reconcile.Result{}, errors.Wrapf(err, "failed to create or update ceph filesystem subvolume group %q", cephFilesystemSubVolumeGroup.Name)
 	}
 
 	// Success! Let's update the status
-	if cephCluster.Spec.External.Enable {
-		r.updateStatus(r.client, request.NamespacedName, cephv1.ConditionConnected)
-	} else {
-		r.updateStatus(r.client, request.NamespacedName, cephv1.ConditionReady)
-	}
+	r.updateStatus(r.client, request.NamespacedName, cephv1.ConditionReady)
 
 	// Return and do not requeue
 	logger.Debug("done reconciling")
@@ -285,17 +241,12 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) createOrUpdateSubVolumeGroup(cep
 // Delete the ceph filesystem subvolume group
 func (r *ReconcileCephFilesystemSubVolumeGroup) deleteSubVolumeGroup(cephFilesystemSubVolumeGroup *cephv1.CephFilesystemSubVolumeGroup) error {
 	logger.Infof("deleting ceph filesystem subvolume group object %q", cephFilesystemSubVolumeGroup.Name)
-	if err := cephclient.DeleteCephFSSubVolumeGroup(r.context, r.clusterInfo, cephFilesystemSubVolumeGroup.Spec.FilesystemName, cephFilesystemSubVolumeGroup.Name); err != nil {
+	if err := cephclient.DeleteCephFSSubVolumeGroup(r.context, r.clusterInfo, cephFilesystemSubVolumeGroup.Name, cephFilesystemSubVolumeGroup.Spec.FilesystemName); err != nil {
 		code, ok := exec.ExitStatus(err)
-		// If the subvolume group does not exit, we should not return an error
-		if ok && code == int(syscall.ENOENT) {
-			logger.Debugf("ceph filesystem subvolume group %q do not exist", cephFilesystemSubVolumeGroup.Name)
-			return nil
-		}
-		// If the subvolume group has subvolumes the command will fail with:
+		// If the subvolumegroup has subvolumes the command will fail with:
 		// Error ENOTEMPTY: error in rmdir /volumes/csi
-		if ok && (code == int(syscall.ENOTEMPTY)) {
-			return errors.Wrapf(err, "failed to delete ceph filesystem subvolume group %q, remove the subvolumes first", cephFilesystemSubVolumeGroup.Name)
+		if ok && code != int(syscall.ENOTEMPTY) {
+			return errors.Wrapf(err, "failed to delete ceph filesystem subvolume group %q, remove the subvolumes first.", cephFilesystemSubVolumeGroup.Name)
 		}
 
 		return errors.Wrapf(err, "failed to delete ceph filesystem subvolume group %q", cephFilesystemSubVolumeGroup.Name)
@@ -313,7 +264,7 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) updateStatus(client client.Clien
 			logger.Debug("CephFilesystemSubVolumeGroup resource not found. Ignoring since object must be deleted.")
 			return
 		}
-		logger.Warningf("failed to retrieve ceph filesystem subvolume group %q to update status to %q. %v", name, status, err)
+		logger.Warningf("failed to retrieve ceph ceph filesystem subvolume group %q to update status to %q. %v", name, status, err)
 		return
 	}
 	if cephFilesystemSubVolumeGroup.Status == nil {
@@ -321,15 +272,9 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) updateStatus(client client.Clien
 	}
 
 	cephFilesystemSubVolumeGroup.Status.Phase = status
-	cephFilesystemSubVolumeGroup.Status.Info = map[string]string{"clusterID": buildClusterID(cephFilesystemSubVolumeGroup)}
 	if err := reporting.UpdateStatus(client, cephFilesystemSubVolumeGroup); err != nil {
-		logger.Errorf("failed to set ceph filesystem subvolume group %q status to %q. %v", name, status, err)
+		logger.Errorf("failed to set ceph ceph filesystem subvolume group %q status to %q. %v", name, status, err)
 		return
 	}
 	logger.Debugf("ceph ceph filesystem subvolume group %q status updated to %q", name, status)
 }
-
-func buildClusterID(cephFilesystemSubVolumeGroup *cephv1.CephFilesystemSubVolumeGroup) string {
-	clusterID := fmt.Sprintf("%s-%s-file-%s", cephFilesystemSubVolumeGroup.Namespace, cephFilesystemSubVolumeGroup.Spec.FilesystemName, cephFilesystemSubVolumeGroup.Name)
-	return k8sutil.Hash(clusterID)
-}
diff --git a/pkg/operator/ceph/nfs/controller.go b/pkg/operator/ceph/nfs/controller.go
index 1e70a3c86..96fea5f6a 100644
--- a/pkg/operator/ceph/nfs/controller.go
+++ b/pkg/operator/ceph/nfs/controller.go
@@ -245,7 +245,7 @@ func (r *ReconcileCephNFS) reconcile(request reconcile.Request) (reconcile.Resul
 	// If the version of the Ceph monitor differs from the CephCluster CR image version we assume
 	// the cluster is being upgraded. So the controller will just wait for the upgrade to finish and
 	// then versions should match. Obviously using the cmd reporter job adds up to the deployment time
-	if !reflect.DeepEqual(*runningCephVersion, *desiredCephVersion) {
+	if !reflect.DeepEqual(runningCephVersion, desiredCephVersion) {
 		// Upgrade is in progress, let's wait for the mons to be done
 		return opcontroller.WaitForRequeueIfCephClusterIsUpgrading,
 			opcontroller.ErrorCephUpgradingRequeue(desiredCephVersion, runningCephVersion)
diff --git a/pkg/operator/ceph/nfs/nfs.go b/pkg/operator/ceph/nfs/nfs.go
index 2da06a4b8..236140003 100644
--- a/pkg/operator/ceph/nfs/nfs.go
+++ b/pkg/operator/ceph/nfs/nfs.go
@@ -18,6 +18,7 @@ limitations under the License.
 package nfs
 
 import (
+	"crypto/sha256"
 	"fmt"
 
 	"github.com/banzaicloud/k8s-objectmatcher/patch"
@@ -214,7 +215,13 @@ func (r *ReconcileCephNFS) createConfigMap(n *cephv1.CephNFS, name string) (stri
 		}
 	}
 
-	return configMap.Name, k8sutil.Hash(fmt.Sprintf("%v", configMap.Data)), nil
+	h := sha256.New()
+	if _, err := h.Write([]byte(fmt.Sprintf("%v", configMap.Data))); err != nil {
+		return "", "", errors.Wrapf(err, "failed to get hash of ganesha config map")
+	}
+	configHash := fmt.Sprintf("%x", h.Sum(nil))
+
+	return configMap.Name, configHash, nil
 }
 
 // Down scale the ganesha server
diff --git a/pkg/operator/ceph/object/bucket/rgw-handlers.go b/pkg/operator/ceph/object/bucket/rgw-handlers.go
index 6fc13dd78..bcce7954e 100644
--- a/pkg/operator/ceph/object/bucket/rgw-handlers.go
+++ b/pkg/operator/ceph/object/bucket/rgw-handlers.go
@@ -101,16 +101,6 @@ func (p *Provisioner) deleteOBCResource(bucketName string) error {
 		} else if errors.Is(err, admin.ErrNoSuchBucket) {
 			// opinion: "not found" is not an error
 			logger.Infof("bucket %q does not exist", p.bucketName)
-		} else if errors.Is(err, admin.ErrNoSuchKey) {
-			// ceph might return NoSuchKey than NoSuchBucket when the target bucket does not exist.
-			// then we can use GetBucketInfo() to judge the existence of the bucket.
-			// see: https://github.com/ceph/ceph/pull/44413
-			_, err2 := p.adminOpsClient.GetBucketInfo(p.clusterInfo.Context, admin.Bucket{Bucket: bucketName, PurgeObject: &thePurge})
-			if errors.Is(err2, admin.ErrNoSuchBucket) {
-				logger.Infof("bucket info %q does not exist", p.bucketName)
-			} else {
-				return errors.Wrapf(err, "failed to delete bucket %q (could not get bucket info)", bucketName)
-			}
 		} else {
 			return errors.Wrapf(err, "failed to delete bucket %q", bucketName)
 		}
diff --git a/pkg/operator/ceph/object/controller.go b/pkg/operator/ceph/object/controller.go
index e8d527549..e155d376f 100644
--- a/pkg/operator/ceph/object/controller.go
+++ b/pkg/operator/ceph/object/controller.go
@@ -312,7 +312,7 @@ func (r *ReconcileCephObjectStore) reconcile(request reconcile.Request) (reconci
 	// If the version of the Ceph monitor differs from the CephCluster CR image version we assume
 	// the cluster is being upgraded. So the controller will just wait for the upgrade to finish and
 	// then versions should match. Obviously using the cmd reporter job adds up to the deployment time
-	if !reflect.DeepEqual(*runningCephVersion, *desiredCephVersion) {
+	if !reflect.DeepEqual(runningCephVersion, desiredCephVersion) {
 		// Upgrade is in progress, let's wait for the mons to be done
 		return opcontroller.WaitForRequeueIfCephClusterIsUpgrading,
 			cephObjectStore,
diff --git a/pkg/operator/ceph/object/notification/provisioner.go b/pkg/operator/ceph/object/notification/provisioner.go
index 0331dceae..5558085b4 100644
--- a/pkg/operator/ceph/object/notification/provisioner.go
+++ b/pkg/operator/ceph/object/notification/provisioner.go
@@ -19,6 +19,7 @@ package notification
 
 import (
 	"context"
+	"net/http"
 
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/ceph/go-ceph/rgw/admin"
@@ -43,14 +44,26 @@ type provisioner struct {
 	objectStoreName  types.NamespacedName
 }
 
-func getUserCredentials(adminOpsCtx *object.AdminOpsContext, opManagerContext context.Context, username string) (accessKey string, secretKey string, err error) {
+func getUserCredentials(opManagerContext context.Context, username string, objStore *cephv1.CephObjectStore, objContext *object.Context) (accessKey string, secretKey string, err error) {
 	if len(username) == 0 {
 		err = errors.New("no user name provided")
 		return
 	}
 
+	adminAccessKey, adminSecretKey, err := object.GetAdminOPSUserCredentials(objContext, &objStore.Spec)
+	if err != nil {
+		err = errors.Wrapf(err, "failed to get Ceph RGW admin ops user credentials when getting user %q", username)
+		return
+	}
+
+	adminOpsClient, err := admin.New(objContext.Endpoint, adminAccessKey, adminSecretKey, &http.Client{})
+	if err != nil {
+		err = errors.Wrapf(err, "failed to build admin ops API connection to get user %q", username)
+		return
+	}
+
 	var u admin.User
-	u, err = adminOpsCtx.AdminOpsClient.GetUser(opManagerContext, admin.User{ID: username})
+	u, err = adminOpsClient.GetUser(opManagerContext, admin.User{ID: username})
 	if err != nil {
 		err = errors.Wrapf(err, "failed to get ceph user %q", username)
 		return
@@ -75,24 +88,12 @@ func newS3Agent(p provisioner) (*object.S3Agent, error) {
 	// CephClusterSpec is needed for GetAdminOPSUserCredentials()
 	objContext.CephClusterSpec = *p.clusterSpec
 
-	adminOpsCtx, err := object.NewMultisiteAdminOpsContext(objContext, &objStore.Spec)
-	if err != nil {
-		return nil, errors.Wrapf(err, "failed to get admin Ops context for CephObjectStore %q", p.objectStoreName)
-
-	}
-	accessKey, secretKey, err := getUserCredentials(adminOpsCtx, p.opManagerContext, p.owner)
+	accessKey, secretKey, err := getUserCredentials(p.opManagerContext, p.owner, objStore, objContext)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to get owner credentials for %q", p.owner)
 	}
-	tlsCert := make([]byte, 0)
-	if objStore.Spec.IsTLSEnabled() {
-		tlsCert, _, err = object.GetTlsCaCert(objContext, &objStore.Spec)
-		if err != nil {
-			return nil, errors.Wrap(err, "failed to fetch TLS certificate for the object store")
-		}
-	}
 
-	return object.NewS3Agent(accessKey, secretKey, objContext.Endpoint, objContext.ZoneGroup, logger.LevelAt(capnslog.DEBUG), tlsCert)
+	return object.NewS3Agent(accessKey, secretKey, objContext.Endpoint, objContext.ZoneGroup, logger.LevelAt(capnslog.DEBUG), objContext.Context.KubeConfig.CertData)
 }
 
 // TODO: convert all rules without restrictions once the AWS SDK supports that
diff --git a/pkg/operator/ceph/object/spec.go b/pkg/operator/ceph/object/spec.go
index ea7a7babd..a654abc7b 100644
--- a/pkg/operator/ceph/object/spec.go
+++ b/pkg/operator/ceph/object/spec.go
@@ -61,19 +61,9 @@ func (c *clusterConfig) createDeployment(rgwConfig *rgwConfig) (*apps.Deployment
 		return nil, err
 	}
 	replicas := int32(1)
-	strategy := apps.DeploymentStrategy{
-		Type: apps.RecreateDeploymentStrategyType,
-	}
+	// On Pacific, we can use the same keyring and have dedicated rgw instances reflected in the service map
 	if c.clusterInfo.CephVersion.IsAtLeastPacific() {
-		// On Pacific, we can use the same keyring and have dedicated rgw instances reflected in the service map
 		replicas = c.store.Spec.Gateway.Instances
-
-		// On Pacific, rgw gateway deployments rolling update
-		strategy.Type = apps.RollingUpdateDeploymentStrategyType
-		strategy.RollingUpdate = &apps.RollingUpdateDeployment{
-			MaxUnavailable: &intstr.IntOrString{IntVal: int32(1)},
-			MaxSurge:       &intstr.IntOrString{IntVal: int32(0)},
-		}
 	}
 	d := &apps.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
@@ -87,7 +77,9 @@ func (c *clusterConfig) createDeployment(rgwConfig *rgwConfig) (*apps.Deployment
 			},
 			Template: pod,
 			Replicas: &replicas,
-			Strategy: strategy,
+			Strategy: apps.DeploymentStrategy{
+				Type: apps.RecreateDeploymentStrategyType,
+			},
 		},
 	}
 	k8sutil.AddRookVersionLabelToDeployment(d)
diff --git a/pkg/operator/ceph/object/topic/provisioner.go b/pkg/operator/ceph/object/topic/provisioner.go
index 66f7124bb..e6f9ea347 100644
--- a/pkg/operator/ceph/object/topic/provisioner.go
+++ b/pkg/operator/ceph/object/topic/provisioner.go
@@ -88,10 +88,7 @@ func createSNSClient(p provisioner, objectStoreName types.NamespacedName) (*sns.
 	}
 	tlsEnabled := objStore.Spec.IsTLSEnabled()
 	if tlsEnabled {
-		tlsCert, _, err := object.GetTlsCaCert(objContext, &objStore.Spec)
-		if err != nil {
-			return nil, errors.Wrap(err, "failed to get TLS certificate for the object store")
-		}
+		tlsCert := objContext.Context.KubeConfig.CertData
 		if len(tlsCert) > 0 {
 			client.Transport = object.BuildTransportTLS(tlsCert, false)
 		}
@@ -104,7 +101,6 @@ func createSNSClient(p provisioner, objectStoreName types.NamespacedName) (*sns.
 			WithEndpoint(objContext.Endpoint).
 			WithMaxRetries(3).
 			WithDisableSSL(!tlsEnabled).
-			WithHTTPClient(&client).
 			WithLogLevel(logLevel),
 	)
 	if err != nil {
diff --git a/pkg/operator/k8sutil/resources.go b/pkg/operator/k8sutil/resources.go
index c8a01005c..800f87a6c 100644
--- a/pkg/operator/k8sutil/resources.go
+++ b/pkg/operator/k8sutil/resources.go
@@ -148,16 +148,37 @@ func (info *OwnerInfo) GetUID() types.UID {
 }
 
 func MergeResourceRequirements(first, second v1.ResourceRequirements) v1.ResourceRequirements {
-	// if the first has no limits set, apply the second limits if any are specified
-	if len(first.Limits) == 0 {
-		if len(second.Limits) > 0 {
-			first.Limits = second.Limits
+	// if the first has a value not set check if second has and set it in first
+	if _, ok := first.Limits[v1.ResourceCPU]; !ok {
+		if _, ok = second.Limits[v1.ResourceCPU]; ok {
+			if first.Limits == nil {
+				first.Limits = v1.ResourceList{}
+			}
+			first.Limits[v1.ResourceCPU] = second.Limits[v1.ResourceCPU]
 		}
 	}
-	// if the first has no requests set, apply the second requests if any are specified
-	if len(first.Requests) == 0 {
-		if len(second.Requests) > 0 {
-			first.Requests = second.Requests
+	if _, ok := first.Limits[v1.ResourceMemory]; !ok {
+		if _, ok = second.Limits[v1.ResourceMemory]; ok {
+			if first.Limits == nil {
+				first.Limits = v1.ResourceList{}
+			}
+			first.Limits[v1.ResourceMemory] = second.Limits[v1.ResourceMemory]
+		}
+	}
+	if _, ok := first.Requests[v1.ResourceCPU]; !ok {
+		if _, ok = second.Requests[v1.ResourceCPU]; ok {
+			if first.Requests == nil {
+				first.Requests = v1.ResourceList{}
+			}
+			first.Requests[v1.ResourceCPU] = second.Requests[v1.ResourceCPU]
+		}
+	}
+	if _, ok := first.Requests[v1.ResourceMemory]; !ok {
+		if _, ok = second.Requests[v1.ResourceMemory]; ok {
+			if first.Requests == nil {
+				first.Requests = v1.ResourceList{}
+			}
+			first.Requests[v1.ResourceMemory] = second.Requests[v1.ResourceMemory]
 		}
 	}
 	return first
diff --git a/pkg/util/exec/exec.go b/pkg/util/exec/exec.go
index 4d258f7fa..67913e1f5 100644
--- a/pkg/util/exec/exec.go
+++ b/pkg/util/exec/exec.go
@@ -19,8 +19,10 @@ package exec
 import (
 	"bufio"
 	"bytes"
+	"context"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"os"
 	"os/exec"
 	"reflect"
@@ -140,6 +142,93 @@ func (*CommandExecutor) ExecuteCommandWithCombinedOutput(command string, arg ...
 	return runCommandWithOutput(cmd, true)
 }
 
+// ExecuteCommandWithOutputFileTimeout Same as ExecuteCommandWithOutputFile but with a timeout limit.
+// #nosec G307 Calling defer to close the file without checking the error return is not a risk for a simple file open and close
+func (*CommandExecutor) ExecuteCommandWithOutputFileTimeout(timeout time.Duration,
+	command, outfileArg string, arg ...string) (string, error) {
+
+	outFile, err := ioutil.TempFile("", "")
+	if err != nil {
+		return "", errors.Wrap(err, "failed to open output file")
+	}
+	defer outFile.Close()
+	defer os.Remove(outFile.Name())
+
+	arg = append(arg, outfileArg, outFile.Name())
+	logCommand(command, arg...)
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	// #nosec G204 Rook controls the input to the exec arguments
+	cmd := exec.CommandContext(ctx, command, arg...)
+	cmdOut, err := cmd.CombinedOutput()
+	if err != nil {
+		cmdOut = []byte(fmt.Sprintf("%s. %s", string(cmdOut), assertErrorType(err)))
+	}
+
+	// if there was anything that went to stdout/stderr then log it, even before
+	// we return an error
+	if string(cmdOut) != "" {
+		if !strings.Contains(err.Error(), "error calling conf_read_file") {
+			logger.Debug(string(cmdOut))
+		}
+	}
+
+	if ctx.Err() == context.DeadlineExceeded {
+		return string(cmdOut), ctx.Err()
+	}
+
+	if err != nil {
+		return string(cmdOut), &CephCLIError{err: err, output: string(cmdOut)}
+	}
+
+	fileOut, err := ioutil.ReadAll(outFile)
+	if err := outFile.Close(); err != nil {
+		return "", err
+	}
+	return string(fileOut), err
+}
+
+// ExecuteCommandWithOutputFile executes a command with output on a file
+// #nosec G307 Calling defer to close the file without checking the error return is not a risk for a simple file open and close
+func (*CommandExecutor) ExecuteCommandWithOutputFile(command, outfileArg string, arg ...string) (string, error) {
+
+	// create a temporary file to serve as the output file for the command to be run and ensure
+	// it is cleaned up after this function is done
+	outFile, err := ioutil.TempFile("", "")
+	if err != nil {
+		return "", errors.Wrap(err, "failed to open output file")
+	}
+	defer outFile.Close()
+	defer os.Remove(outFile.Name())
+
+	// append the output file argument to the list or args
+	arg = append(arg, outfileArg, outFile.Name())
+
+	logCommand(command, arg...)
+	// #nosec G204 Rook controls the input to the exec arguments
+	cmd := exec.Command(command, arg...)
+	cmdOut, err := cmd.CombinedOutput()
+	if err != nil {
+		cmdOut = []byte(fmt.Sprintf("%s. %s", string(cmdOut), assertErrorType(err)))
+	}
+	// if there was anything that went to stdout/stderr then log it, even before we return an error
+	if string(cmdOut) != "" {
+		logger.Debug(string(cmdOut))
+	}
+	if err != nil {
+		return string(cmdOut), &CephCLIError{err: err, output: string(cmdOut)}
+	}
+
+	// read the entire output file and return that to the caller
+	fileOut, err := ioutil.ReadAll(outFile)
+	if err := outFile.Close(); err != nil {
+		return "", err
+	}
+	return string(fileOut), err
+}
+
 func startCommand(env []string, command string, arg ...string) (*exec.Cmd, io.ReadCloser, io.ReadCloser, error) {
 	logCommand(command, arg...)
 


Test patch:
diff --git a/.github/workflows/canary-integration-test.yml b/.github/workflows/canary-integration-test.yml
index 9aa1c7d30..fc1ea7be8 100644
--- a/.github/workflows/canary-integration-test.yml
+++ b/.github/workflows/canary-integration-test.yml
@@ -838,12 +838,9 @@ jobs:
       with:
         fetch-depth: 0
 
-    - name: run encryption KMS IBM Key Protect
+    - name: run Encryption KMS IBM Key Protect
       uses: ./.github/workflows/encryption-pvc-kms-ibm-kp
-      if: "env.IBM_KP_SERVICE_INSTANCE_ID != '' && env.IBM_KP_SERVICE_API_KEY != ''"
-      env:
-        IBM_KP_SERVICE_INSTANCE_ID: ${{ secrets.IBM_INSTANCE_ID }}
-        IBM_KP_SERVICE_API_KEY: ${{ secrets.IBM_SERVICE_API_KEY }}
+      if: "env.IBM_KP_INSTANCE_ID != '' && env.IBM_KP_SERVICE_API_KEY != ''"
       with:
         ibm-instance-id: ${{ secrets.IBM_INSTANCE_ID }}
         ibm-service-api-key: ${{ secrets.IBM_SERVICE_API_KEY }}
diff --git a/pkg/daemon/ceph/client/config_test.go b/pkg/daemon/ceph/client/config_test.go
index 4d62fcc65..8528741ea 100644
--- a/pkg/daemon/ceph/client/config_test.go
+++ b/pkg/daemon/ceph/client/config_test.go
@@ -18,6 +18,8 @@ package client
 
 import (
 	"context"
+	"io/ioutil"
+	"os"
 	"path/filepath"
 	"strings"
 	"testing"
@@ -64,7 +66,11 @@ func TestCreateDefaultCephConfig(t *testing.T) {
 func TestGenerateConfigFile(t *testing.T) {
 	ctx := context.TODO()
 	// set up a temporary config directory that will be cleaned up after test
-	configDir := t.TempDir()
+	configDir, err := ioutil.TempDir("", "TestGenerateConfigFile")
+	if err != nil {
+		t.Fatalf("failed to create temp config dir: %+v", err)
+	}
+	defer os.RemoveAll(configDir)
 
 	// create mocked cluster context and info
 	clientset := test.New(t, 3)
@@ -84,7 +90,7 @@ func TestGenerateConfigFile(t *testing.T) {
 		},
 		Data: data,
 	}
-	_, err := clientset.CoreV1().ConfigMaps(ns).Create(ctx, cm, metav1.CreateOptions{})
+	_, err = clientset.CoreV1().ConfigMaps(ns).Create(ctx, cm, metav1.CreateOptions{})
 	assert.NoError(t, err)
 
 	clusterInfo := &ClusterInfo{
diff --git a/pkg/daemon/ceph/client/pool_test.go b/pkg/daemon/ceph/client/pool_test.go
index 3c6123ea0..696e07970 100644
--- a/pkg/daemon/ceph/client/pool_test.go
+++ b/pkg/daemon/ceph/client/pool_test.go
@@ -29,8 +29,6 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-const emptyApplicationName = `{"":{}}`
-
 func TestCreateECPoolWithOverwrites(t *testing.T) {
 	testCreateECPool(t, true, "")
 }
@@ -81,9 +79,6 @@ func testCreateECPool(t *testing.T, overwrite bool, compressionMode string) {
 				}
 			}
 			if args[2] == "application" {
-				if args[3] == "get" {
-					return emptyApplicationName, nil
-				}
 				assert.Equal(t, "enable", args[3])
 				assert.Equal(t, "mypool", args[4])
 				assert.Equal(t, "myapp", args[5])
@@ -102,53 +97,6 @@ func testCreateECPool(t *testing.T, overwrite bool, compressionMode string) {
 	}
 }
 
-func TestSetPoolApplication(t *testing.T) {
-	poolName := "testpool"
-	appName := "testapp"
-	setAppName := false
-	blankAppName := false
-	clusterInfo := AdminTestClusterInfo("mycluster")
-	executor := &exectest.MockExecutor{}
-	context := &clusterd.Context{Executor: executor}
-	executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
-		logger.Infof("Command: %s %v", command, args)
-		if args[1] == "pool" && args[2] == "application" {
-			if args[3] == "get" {
-				assert.Equal(t, poolName, args[4])
-				if blankAppName {
-					return emptyApplicationName, nil
-				} else {
-					return fmt.Sprintf(`{"%s":{}}`, appName), nil
-				}
-			}
-			if args[3] == "enable" {
-				setAppName = true
-				assert.Equal(t, poolName, args[4])
-				assert.Equal(t, appName, args[5])
-				return "", nil
-			}
-		}
-		return "", errors.Errorf("unexpected ceph command %q", args)
-	}
-
-	t.Run("set pool application", func(t *testing.T) {
-		setAppName = false
-		blankAppName = true
-		err := givePoolAppTag(context, clusterInfo, poolName, appName)
-		assert.NoError(t, err)
-		assert.True(t, setAppName)
-	})
-
-	t.Run("pool application already set", func(t *testing.T) {
-		setAppName = false
-		blankAppName = false
-		err := givePoolAppTag(context, clusterInfo, poolName, appName)
-		assert.NoError(t, err)
-		assert.False(t, setAppName)
-	})
-
-}
-
 func TestCreateReplicaPoolWithFailureDomain(t *testing.T) {
 	testCreateReplicaPool(t, "osd", "mycrushroot", "", "")
 }
@@ -189,9 +137,6 @@ func testCreateReplicaPool(t *testing.T, failureDomain, crushRoot, deviceClass,
 				return "", nil
 			}
 			if args[2] == "application" {
-				if args[3] == "get" {
-					return emptyApplicationName, nil
-				}
 				assert.Equal(t, "enable", args[3])
 				assert.Equal(t, "mypool", args[4])
 				assert.Equal(t, "myapp", args[5])
@@ -520,11 +465,7 @@ func testCreatePoolWithReplicasPerFailureDomain(t *testing.T, failureDomain, cru
 			poolRuleSet = true
 			return "", nil
 		}
-		if len(args) >= 4 && args[1] == "pool" && args[2] == "application" {
-			if args[3] == "get" {
-				return emptyApplicationName, nil
-			}
-
+		if len(args) >= 4 && args[1] == "pool" && args[2] == "application" && args[3] == "enable" {
 			crushRuleName := args[4]
 			assert.Equal(t, crushRuleName, poolSpec.Name)
 			poolAppEnable = true
diff --git a/pkg/daemon/ceph/osd/device_test.go b/pkg/daemon/ceph/osd/device_test.go
index 44e7ea1c4..55c3f6a6a 100644
--- a/pkg/daemon/ceph/osd/device_test.go
+++ b/pkg/daemon/ceph/osd/device_test.go
@@ -17,6 +17,7 @@ package osd
 
 import (
 	"io/ioutil"
+	"os"
 	"path"
 	"strings"
 	"testing"
@@ -28,7 +29,8 @@ import (
 )
 
 func TestOSDBootstrap(t *testing.T) {
-	configDir := t.TempDir()
+	configDir, _ := ioutil.TempDir("", "")
+	defer os.RemoveAll(configDir)
 
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
@@ -37,6 +39,7 @@ func TestOSDBootstrap(t *testing.T) {
 	}
 
 	context := &clusterd.Context{Executor: executor, ConfigDir: configDir}
+	defer os.RemoveAll(context.ConfigDir)
 	err := createOSDBootstrapKeyring(context, client.AdminTestClusterInfo("mycluster"), configDir)
 	assert.Nil(t, err)
 
diff --git a/pkg/daemon/ceph/osd/kms/envs_test.go b/pkg/daemon/ceph/osd/kms/envs_test.go
index a8c562f93..aafb8b312 100644
--- a/pkg/daemon/ceph/osd/kms/envs_test.go
+++ b/pkg/daemon/ceph/osd/kms/envs_test.go
@@ -63,7 +63,7 @@ func TestVaultTLSEnvVarFromSecret(t *testing.T) {
 	t.Run("ibm kp", func(t *testing.T) {
 		spec := cephv1.ClusterSpec{Security: cephv1.SecuritySpec{KeyManagementService: cephv1.KeyManagementServiceSpec{
 			TokenSecretName:   "ibm-kp-token",
-			ConnectionDetails: map[string]string{"KMS_PROVIDER": TypeIBM, "IBM_KP_SERVICE_INSTANCE_ID": "1"}}},
+			ConnectionDetails: map[string]string{"KMS_PROVIDER": "ibm-kp", "IBM_KP_INSTANCE_ID": "1"}}},
 		}
 		envVars := ConfigToEnvVar(spec)
 		areEnvVarsSorted := sort.SliceIsSorted(envVars, func(i, j int) bool {
@@ -71,8 +71,8 @@ func TestVaultTLSEnvVarFromSecret(t *testing.T) {
 		})
 		assert.True(t, areEnvVarsSorted)
 		assert.Equal(t, 3, len(envVars))
-		assert.Contains(t, envVars, v1.EnvVar{Name: "KMS_PROVIDER", Value: TypeIBM})
-		assert.Contains(t, envVars, v1.EnvVar{Name: "IBM_KP_SERVICE_INSTANCE_ID", Value: "1"})
+		assert.Contains(t, envVars, v1.EnvVar{Name: "KMS_PROVIDER", Value: "ibm-kp"})
+		assert.Contains(t, envVars, v1.EnvVar{Name: "IBM_KP_INSTANCE_ID", Value: "1"})
 		assert.Contains(t, envVars, v1.EnvVar{Name: "IBM_KP_SERVICE_API_KEY", ValueFrom: &v1.EnvVarSource{SecretKeyRef: &v1.SecretKeySelector{LocalObjectReference: v1.LocalObjectReference{Name: "ibm-kp-token"}, Key: "IBM_KP_SERVICE_API_KEY"}}})
 	})
 }
@@ -140,11 +140,11 @@ func TestVaultConfigToEnvVar(t *testing.T) {
 		},
 		{
 			"ibm kp - IBM_KP_SERVICE_API_KEY is removed from the details",
-			args{spec: cephv1.ClusterSpec{Security: cephv1.SecuritySpec{KeyManagementService: cephv1.KeyManagementServiceSpec{ConnectionDetails: map[string]string{"KMS_PROVIDER": TypeIBM, "IBM_KP_SERVICE_API_KEY": "foo", "IBM_KP_SERVICE_INSTANCE_ID": "1"}, TokenSecretName: "ibm-kp-token"}}}},
+			args{spec: cephv1.ClusterSpec{Security: cephv1.SecuritySpec{KeyManagementService: cephv1.KeyManagementServiceSpec{ConnectionDetails: map[string]string{"KMS_PROVIDER": "ibm-kp", "IBM_KP_SERVICE_API_KEY": "foo", "IBM_KP_INSTANCE_ID": "1"}, TokenSecretName: "ibm-kp-token"}}}},
 			[]v1.EnvVar{
+				{Name: "IBM_KP_INSTANCE_ID", Value: "1"},
 				{Name: "IBM_KP_SERVICE_API_KEY", ValueFrom: &v1.EnvVarSource{SecretKeyRef: &v1.SecretKeySelector{LocalObjectReference: v1.LocalObjectReference{Name: "ibm-kp-token"}, Key: "IBM_KP_SERVICE_API_KEY"}}},
-				{Name: "IBM_KP_SERVICE_INSTANCE_ID", Value: "1"},
-				{Name: "KMS_PROVIDER", Value: TypeIBM},
+				{Name: "KMS_PROVIDER", Value: "ibm-kp"},
 			},
 		},
 	}
diff --git a/pkg/daemon/ceph/osd/kms/ibm_key_protect_test.go b/pkg/daemon/ceph/osd/kms/ibm_key_protect_test.go
index aeeaeb948..5a19b164a 100644
--- a/pkg/daemon/ceph/osd/kms/ibm_key_protect_test.go
+++ b/pkg/daemon/ceph/osd/kms/ibm_key_protect_test.go
@@ -35,7 +35,7 @@ func TestInitKeyProtect(t *testing.T) {
 		config[IbmKeyProtectServiceApiKey] = "foo"
 	})
 
-	t.Run("IBM_KP_SERVICE_INSTANCE_ID not set", func(t *testing.T) {
+	t.Run("IBM_KP_INSTANCE_ID not set", func(t *testing.T) {
 		_, err := InitKeyProtect(config)
 		assert.Error(t, err)
 		assert.ErrorIs(t, err, ErrIbmInstanceIdKeyNotSet)
diff --git a/pkg/daemon/ceph/osd/kms/kms_test.go b/pkg/daemon/ceph/osd/kms/kms_test.go
index 459d8718b..180f9b32b 100644
--- a/pkg/daemon/ceph/osd/kms/kms_test.go
+++ b/pkg/daemon/ceph/osd/kms/kms_test.go
@@ -58,7 +58,7 @@ func TestValidateConnectionDetails(t *testing.T) {
 	}
 	ibmSecuritySpec := &cephv1.SecuritySpec{KeyManagementService: cephv1.KeyManagementServiceSpec{
 		ConnectionDetails: map[string]string{
-			"KMS_PROVIDER": TypeIBM,
+			"KMS_PROVIDER": "ibm-kp",
 		},
 	}}
 
@@ -183,8 +183,8 @@ func TestValidateConnectionDetails(t *testing.T) {
 		assert.NoError(t, err)
 		err = ValidateConnectionDetails(ctx, context, ibmSecuritySpec, ns)
 		assert.Error(t, err, "")
-		assert.EqualError(t, err, "failed to validate kms config \"IBM_KP_SERVICE_INSTANCE_ID\". cannot be empty")
-		ibmSecuritySpec.KeyManagementService.ConnectionDetails["IBM_KP_SERVICE_INSTANCE_ID"] = "foo"
+		assert.EqualError(t, err, "failed to validate kms config \"IBM_KP_INSTANCE_ID\". cannot be empty")
+		ibmSecuritySpec.KeyManagementService.ConnectionDetails["IBM_KP_INSTANCE_ID"] = "foo"
 	})
 
 	t.Run("ibm kp - success", func(t *testing.T) {
diff --git a/pkg/daemon/ceph/osd/volume_test.go b/pkg/daemon/ceph/osd/volume_test.go
index 34d32b4b4..d77f57a21 100644
--- a/pkg/daemon/ceph/osd/volume_test.go
+++ b/pkg/daemon/ceph/osd/volume_test.go
@@ -324,8 +324,12 @@ func TestConfigureCVDevices(t *testing.T) {
 	}()
 
 	originalCephConfigDir := cephConfigDir
-	cephConfigDir = t.TempDir()
+	cephConfigDir, err = ioutil.TempDir("", "")
+	if err != nil {
+		t.Fatal(err)
+	}
 	defer func() {
+		os.RemoveAll(cephConfigDir)
 		cephConfigDir = originalCephConfigDir
 	}()
 
diff --git a/pkg/daemon/util/copybins_test.go b/pkg/daemon/util/copybins_test.go
index eeec30362..ca85855ee 100644
--- a/pkg/daemon/util/copybins_test.go
+++ b/pkg/daemon/util/copybins_test.go
@@ -58,13 +58,15 @@ func TestCopyBinary(t *testing.T) {
 	// testRootDir/
 	//   bin/
 	//   copy-to-dir/
-	testRootDir := t.TempDir()
+	testRootDir, err := ioutil.TempDir("", "rook-cmd-reporter-copy-binaries-test")
+	assert.NoError(t, err)
+	defer os.RemoveAll(testRootDir)
 	// create a test PATH="testRootDir/bin"
 	envPath := path.Join(testRootDir, "bin")
 	mkdir(envPath)
 	oPath := os.Getenv("PATH")
 	defer os.Setenv("PATH", oPath)
-	err := os.Setenv("PATH", envPath)
+	err = os.Setenv("PATH", envPath)
 	assert.NoError(t, err)
 	// create initial copy-to-dir
 	copyToDir := path.Join(testRootDir, "copy-to-dir")
diff --git a/pkg/operator/ceph/cluster/mgr/mgr_test.go b/pkg/operator/ceph/cluster/mgr/mgr_test.go
index 3eb562427..b846c2f26 100644
--- a/pkg/operator/ceph/cluster/mgr/mgr_test.go
+++ b/pkg/operator/ceph/cluster/mgr/mgr_test.go
@@ -19,6 +19,7 @@ package mgr
 import (
 	"context"
 	"fmt"
+	"io/ioutil"
 	"os"
 	"testing"
 	"time"
@@ -64,7 +65,7 @@ func TestStartMgr(t *testing.T) {
 	}
 
 	clientset := testop.New(t, 3)
-	configDir := t.TempDir()
+	configDir, _ := ioutil.TempDir("", "")
 	scheme := scheme.Scheme
 	err := policyv1.AddToScheme(scheme)
 	assert.NoError(t, err)
@@ -72,6 +73,7 @@ func TestStartMgr(t *testing.T) {
 	assert.NoError(t, err)
 	cl := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects().Build()
 
+	defer os.RemoveAll(configDir)
 	ctx := &clusterd.Context{
 		Executor:  executor,
 		ConfigDir: configDir,
@@ -255,7 +257,8 @@ func TestMgrSidecarReconcile(t *testing.T) {
 		},
 	}
 	clientset := testop.New(t, 3)
-	configDir := t.TempDir()
+	configDir, _ := ioutil.TempDir("", "")
+	defer os.RemoveAll(configDir)
 	ctx := &clusterd.Context{
 		Executor:  executor,
 		ConfigDir: configDir,
diff --git a/pkg/operator/ceph/cluster/mon/drain_test.go b/pkg/operator/ceph/cluster/mon/drain_test.go
index 2ff565d21..f64e1f6e7 100644
--- a/pkg/operator/ceph/cluster/mon/drain_test.go
+++ b/pkg/operator/ceph/cluster/mon/drain_test.go
@@ -18,6 +18,7 @@ package mon
 
 import (
 	"context"
+	"sync"
 	"testing"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -47,7 +48,7 @@ func createFakeCluster(t *testing.T, cephClusterObj *cephv1.CephCluster, k8sVers
 
 	cl := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects().Build()
 	clientset := test.New(t, 3)
-	c := New(&clusterd.Context{Client: cl, Clientset: clientset}, mockNamespace, cephClusterObj.Spec, ownerInfo)
+	c := New(&clusterd.Context{Client: cl, Clientset: clientset}, mockNamespace, cephClusterObj.Spec, ownerInfo, &sync.Mutex{})
 	test.SetFakeKubernetesVersion(clientset, k8sVersion)
 	return c
 }
diff --git a/pkg/operator/ceph/cluster/mon/health_test.go b/pkg/operator/ceph/cluster/mon/health_test.go
index 0f16cff07..f27a1a75b 100644
--- a/pkg/operator/ceph/cluster/mon/health_test.go
+++ b/pkg/operator/ceph/cluster/mon/health_test.go
@@ -19,8 +19,10 @@ package mon
 import (
 	"context"
 	"fmt"
+	"io/ioutil"
 	"os"
 	"reflect"
+	"sync"
 	"testing"
 	"time"
 
@@ -56,14 +58,15 @@ func TestCheckHealth(t *testing.T) {
 		},
 	}
 	clientset := test.New(t, 1)
-	configDir := t.TempDir()
+	configDir, _ := ioutil.TempDir("", "")
+	defer os.RemoveAll(configDir)
 	context := &clusterd.Context{
 		Clientset: clientset,
 		ConfigDir: configDir,
 		Executor:  executor,
 	}
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
-	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo)
+	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{})
 	// clusterInfo is nil so we return err
 	err := c.checkHealth(ctx)
 	assert.NotNil(t, err)
@@ -76,6 +79,7 @@ func TestCheckHealth(t *testing.T) {
 	c.spec.Mon.Count = 3
 	logger.Infof("initial mons: %v", c.ClusterInfo.Monitors)
 	c.waitForStart = false
+	defer os.RemoveAll(c.context.ConfigDir)
 
 	c.mapping.Schedule["f"] = &MonScheduleInfo{
 		Name:    "node0",
@@ -151,7 +155,7 @@ func TestCheckHealth(t *testing.T) {
 }
 
 func TestSkipMonFailover(t *testing.T) {
-	c := New(&clusterd.Context{}, "ns", cephv1.ClusterSpec{}, nil)
+	c := New(&clusterd.Context{}, "ns", cephv1.ClusterSpec{}, nil, nil)
 	c.ClusterInfo = clienttest.CreateTestClusterInfo(1)
 	monName := "arb"
 
@@ -187,7 +191,8 @@ func TestSkipMonFailover(t *testing.T) {
 func TestEvictMonOnSameNode(t *testing.T) {
 	ctx := context.TODO()
 	clientset := test.New(t, 1)
-	configDir := t.TempDir()
+	configDir, _ := ioutil.TempDir("", "")
+	defer os.RemoveAll(configDir)
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
 			logger.Infof("executing command: %s %+v", command, args)
@@ -196,7 +201,7 @@ func TestEvictMonOnSameNode(t *testing.T) {
 	}
 	context := &clusterd.Context{Clientset: clientset, ConfigDir: configDir, Executor: executor}
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
-	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo)
+	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{})
 	setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 0}, "myversion")
 	c.maxMonID = 2
 	c.waitForStart = false
@@ -251,7 +256,7 @@ func TestScaleMonDeployment(t *testing.T) {
 	clientset := test.New(t, 1)
 	context := &clusterd.Context{Clientset: clientset}
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
-	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo)
+	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{})
 	setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 0, AllowMultiplePerNode: true}, "myversion")
 
 	name := "a"
@@ -294,16 +299,18 @@ func TestCheckHealthNotFound(t *testing.T) {
 		},
 	}
 	clientset := test.New(t, 1)
-	configDir := t.TempDir()
+	configDir, _ := ioutil.TempDir("", "")
+	defer os.RemoveAll(configDir)
 	context := &clusterd.Context{
 		Clientset: clientset,
 		ConfigDir: configDir,
 		Executor:  executor,
 	}
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
-	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo)
+	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{})
 	setCommonMonProperties(c, 2, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
 	c.waitForStart = false
+	defer os.RemoveAll(c.context.ConfigDir)
 
 	c.mapping.Schedule["a"] = &MonScheduleInfo{
 		Name: "node0",
@@ -355,17 +362,19 @@ func TestAddRemoveMons(t *testing.T) {
 		},
 	}
 	clientset := test.New(t, 1)
-	configDir := t.TempDir()
+	configDir, _ := ioutil.TempDir("", "")
+	defer os.RemoveAll(configDir)
 	context := &clusterd.Context{
 		Clientset: clientset,
 		ConfigDir: configDir,
 		Executor:  executor,
 	}
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
-	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo)
+	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{})
 	setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 5, AllowMultiplePerNode: true}, "myversion")
 	c.maxMonID = 0 // "a" is max mon id
 	c.waitForStart = false
+	defer os.RemoveAll(c.context.ConfigDir)
 
 	// checking the health will increase the mons as desired all in one go
 	err := c.checkHealth(ctx)
diff --git a/pkg/operator/ceph/cluster/mon/mon_test.go b/pkg/operator/ceph/cluster/mon/mon_test.go
index 068c77cea..14e6f37db 100644
--- a/pkg/operator/ceph/cluster/mon/mon_test.go
+++ b/pkg/operator/ceph/cluster/mon/mon_test.go
@@ -19,11 +19,13 @@ package mon
 import (
 	"context"
 	"fmt"
+	"io/ioutil"
 	"os"
 	"path"
 	"reflect"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -79,7 +81,8 @@ func newTestStartCluster(t *testing.T, namespace string) (*clusterd.Context, err
 
 func newTestStartClusterWithQuorumResponse(t *testing.T, namespace string, monResponse func() (string, error)) (*clusterd.Context, error) {
 	clientset := test.New(t, 3)
-	configDir := t.TempDir()
+	configDir, _ := ioutil.TempDir("", "")
+	defer os.RemoveAll(configDir)
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
 			if strings.Contains(command, "ceph-authtool") {
@@ -270,7 +273,7 @@ func validateStart(t *testing.T, c *Cluster) {
 func TestPersistMons(t *testing.T) {
 	clientset := test.New(t, 1)
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
-	c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{Annotations: cephv1.AnnotationsSpec{cephv1.KeyClusterMetadata: cephv1.Annotations{"key": "value"}}}, ownerInfo)
+	c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{Annotations: cephv1.AnnotationsSpec{cephv1.KeyClusterMetadata: cephv1.Annotations{"key": "value"}}}, ownerInfo, &sync.Mutex{})
 	setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
 
 	// Persist mon a
@@ -296,9 +299,10 @@ func TestPersistMons(t *testing.T) {
 func TestSaveMonEndpoints(t *testing.T) {
 	ctx := context.TODO()
 	clientset := test.New(t, 1)
-	configDir := t.TempDir()
+	configDir, _ := ioutil.TempDir("", "")
+	defer os.RemoveAll(configDir)
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
-	c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", cephv1.ClusterSpec{}, ownerInfo)
+	c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{})
 	setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
 
 	// create the initial config map
@@ -343,9 +347,10 @@ func TestSaveMonEndpoints(t *testing.T) {
 
 func TestMaxMonID(t *testing.T) {
 	clientset := test.New(t, 1)
-	configDir := t.TempDir()
+	configDir, _ := ioutil.TempDir("", "")
+	defer os.RemoveAll(configDir)
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
-	c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", cephv1.ClusterSpec{}, ownerInfo)
+	c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{})
 	c.ClusterInfo = clienttest.CreateTestClusterInfo(1)
 
 	// when the configmap is not found, the maxMonID is -1
diff --git a/pkg/operator/ceph/cluster/mon/node_test.go b/pkg/operator/ceph/cluster/mon/node_test.go
index 98bf7fbe8..8fa24cf59 100644
--- a/pkg/operator/ceph/cluster/mon/node_test.go
+++ b/pkg/operator/ceph/cluster/mon/node_test.go
@@ -19,6 +19,7 @@ package mon
 import (
 	"context"
 	"strings"
+	"sync"
 	"testing"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -36,7 +37,7 @@ import (
 func TestNodeAffinity(t *testing.T) {
 	ctx := context.TODO()
 	clientset := test.New(t, 4)
-	c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{})
+	c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{}, &sync.Mutex{})
 	setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
 
 	c.spec.Placement = map[cephv1.KeyType]cephv1.Placement{}
@@ -166,7 +167,7 @@ func TestPodMemory(t *testing.T) {
 
 func TestHostNetwork(t *testing.T) {
 	clientset := test.New(t, 3)
-	c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{})
+	c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{}, &sync.Mutex{})
 	setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
 
 	c.spec.Network.HostNetwork = true
diff --git a/pkg/operator/ceph/cluster/mon/service_test.go b/pkg/operator/ceph/cluster/mon/service_test.go
index c8cf31e36..62b5dc574 100644
--- a/pkg/operator/ceph/cluster/mon/service_test.go
+++ b/pkg/operator/ceph/cluster/mon/service_test.go
@@ -18,6 +18,7 @@ package mon
 
 import (
 	"context"
+	"sync"
 	"testing"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -32,7 +33,7 @@ import (
 func TestCreateService(t *testing.T) {
 	ctx := context.TODO()
 	clientset := test.New(t, 1)
-	c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{})
+	c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{}, &sync.Mutex{})
 	c.ClusterInfo = client.AdminTestClusterInfo("rook-ceph")
 	m := &monConfig{ResourceName: "rook-ceph-mon-b", DaemonName: "b"}
 	clusterIP, err := c.createService(m)
diff --git a/pkg/operator/ceph/cluster/mon/spec_test.go b/pkg/operator/ceph/cluster/mon/spec_test.go
index c1f58ad9f..2b213e2be 100644
--- a/pkg/operator/ceph/cluster/mon/spec_test.go
+++ b/pkg/operator/ceph/cluster/mon/spec_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package mon
 
 import (
+	"sync"
 	"testing"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -47,6 +48,7 @@ func testPodSpec(t *testing.T, monID string, pvc bool) {
 		"ns",
 		cephv1.ClusterSpec{},
 		ownerInfo,
+		&sync.Mutex{},
 	)
 	setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "rook/rook:myversion")
 	c.spec.CephVersion = cephv1.CephVersionSpec{Image: "quay.io/ceph/ceph:myceph"}
@@ -96,6 +98,7 @@ func TestDeploymentPVCSpec(t *testing.T) {
 		"ns",
 		cephv1.ClusterSpec{},
 		ownerInfo,
+		&sync.Mutex{},
 	)
 	setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "rook/rook:myversion")
 	c.spec.CephVersion = cephv1.CephVersionSpec{Image: "quay.io/ceph/ceph:myceph"}
@@ -155,6 +158,7 @@ func testRequiredDuringScheduling(t *testing.T, hostNetwork, allowMultiplePerNod
 		"ns",
 		cephv1.ClusterSpec{},
 		&k8sutil.OwnerInfo{},
+		&sync.Mutex{},
 	)
 
 	c.spec.Network.HostNetwork = hostNetwork
diff --git a/pkg/operator/ceph/csi/cluster_config_test.go b/pkg/operator/ceph/csi/cluster_config_test.go
index 08c9e1cad..afa53e074 100644
--- a/pkg/operator/ceph/csi/cluster_config_test.go
+++ b/pkg/operator/ceph/csi/cluster_config_test.go
@@ -19,142 +19,71 @@ package csi
 import (
 	"testing"
 
+	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestUpdateCsiClusterConfig(t *testing.T) {
-	csiClusterConfigEntry := CsiClusterConfigEntry{
-		Monitors: []string{"1.2.3.4:5000"},
+	// initialize an empty list & add a simple mons list
+	mons := map[string]*cephclient.MonInfo{
+		"foo": {Name: "foo", Endpoint: "1.2.3.4:5000"},
 	}
-	csiClusterConfigEntry2 := CsiClusterConfigEntry{
-		Monitors: []string{"20.1.1.1:5000", "20.1.1.2:5000", "20.1.1.3:5000"},
+	s, err := updateCsiClusterConfig("[]", "alpha", mons)
+	assert.NoError(t, err)
+	assert.Equal(t, s,
+		`[{"clusterID":"alpha","monitors":["1.2.3.4:5000"]}]`)
+
+	// add a 2nd mon to the current cluster
+	mons["bar"] = &cephclient.MonInfo{
+		Name: "bar", Endpoint: "10.11.12.13:5000"}
+	s, err = updateCsiClusterConfig(s, "alpha", mons)
+	assert.NoError(t, err)
+	cc, err := parseCsiClusterConfig(s)
+	assert.NoError(t, err)
+	assert.Equal(t, len(cc), 1)
+	assert.Equal(t, cc[0].ClusterID, "alpha")
+	assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000")
+	assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000")
+	assert.Equal(t, len(cc[0].Monitors), 2)
+
+	// add a 2nd cluster with 3 mons
+	mons2 := map[string]*cephclient.MonInfo{
+		"flim": {Name: "flim", Endpoint: "20.1.1.1:5000"},
+		"flam": {Name: "flam", Endpoint: "20.1.1.2:5000"},
+		"blam": {Name: "blam", Endpoint: "20.1.1.3:5000"},
 	}
-	csiClusterConfigEntry3 := CsiClusterConfigEntry{
-		Monitors: []string{"10.1.1.1:5000", "10.1.1.2:5000", "10.1.1.3:5000"},
-		CephFS: &CsiCephFSSpec{
-			SubvolumeGroup: "mygroup",
-		},
-	}
-	var s string
-	var err error
-
-	t.Run("add a simple mons list", func(t *testing.T) {
-		s, err = updateCsiClusterConfig("[]", "alpha", &csiClusterConfigEntry)
-		assert.NoError(t, err)
-		assert.Equal(t, s,
-			`[{"clusterID":"alpha","monitors":["1.2.3.4:5000"]}]`)
-	})
-
-	t.Run("add a 2nd mon to the current cluster", func(t *testing.T) {
-		csiClusterConfigEntry.Monitors = append(csiClusterConfigEntry.Monitors, "10.11.12.13:5000")
-		s, err = updateCsiClusterConfig(s, "alpha", &csiClusterConfigEntry)
-		assert.NoError(t, err)
-		cc, err := parseCsiClusterConfig(s)
-		assert.NoError(t, err)
-		assert.Equal(t, 1, len(cc))
-		assert.Equal(t, "alpha", cc[0].ClusterID)
-		assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000")
-		assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000")
-		assert.Equal(t, 2, len(cc[0].Monitors))
-	})
-
-	t.Run("add a 2nd cluster with 3 mons", func(t *testing.T) {
-		s, err = updateCsiClusterConfig(s, "beta", &csiClusterConfigEntry2)
-		assert.NoError(t, err)
-		cc, err := parseCsiClusterConfig(s)
-		assert.NoError(t, err)
-		assert.Equal(t, 2, len(cc))
-		assert.Equal(t, "alpha", cc[0].ClusterID)
-		assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000")
-		assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000")
-		assert.Equal(t, 2, len(cc[0].Monitors))
-		assert.Equal(t, "beta", cc[1].ClusterID)
-		assert.Contains(t, cc[1].Monitors, "20.1.1.1:5000")
-		assert.Contains(t, cc[1].Monitors, "20.1.1.2:5000")
-		assert.Contains(t, cc[1].Monitors, "20.1.1.3:5000")
-		assert.Equal(t, 3, len(cc[1].Monitors))
-	})
-
-	t.Run("remove a mon from the 2nd cluster", func(t *testing.T) {
-		i := 2
-		// Remove last element of the slice
-		csiClusterConfigEntry2.Monitors = append(csiClusterConfigEntry2.Monitors[:i], csiClusterConfigEntry2.Monitors[i+1:]...)
-		s, err = updateCsiClusterConfig(s, "beta", &csiClusterConfigEntry2)
-		assert.NoError(t, err)
-		cc, err := parseCsiClusterConfig(s)
-		assert.NoError(t, err)
-		assert.Equal(t, 2, len(cc))
-		assert.Equal(t, "alpha", cc[0].ClusterID)
-		assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000")
-		assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000")
-		assert.Equal(t, 2, len(cc[0].Monitors))
-		assert.Equal(t, "beta", cc[1].ClusterID)
-		assert.Contains(t, cc[1].Monitors, "20.1.1.1:5000")
-		assert.Contains(t, cc[1].Monitors, "20.1.1.2:5000")
-		assert.Equal(t, 2, len(cc[1].Monitors))
-	})
-
-	t.Run("add a 3rd cluster with subvolumegroup", func(t *testing.T) {
-		s, err = updateCsiClusterConfig(s, "baba", &csiClusterConfigEntry3)
-		assert.NoError(t, err)
-		cc, err := parseCsiClusterConfig(s)
-		assert.NoError(t, err)
-		assert.Equal(t, 3, len(cc))
-		assert.Equal(t, "alpha", cc[0].ClusterID)
-		assert.Equal(t, 2, len(cc[0].Monitors))
-		assert.Equal(t, "beta", cc[1].ClusterID)
-		assert.Equal(t, "baba", cc[2].ClusterID)
-		assert.Equal(t, "10.1.1.1:5000", cc[2].Monitors[0])
-		assert.Equal(t, 3, len(cc[2].Monitors))
-		assert.Equal(t, "mygroup", cc[2].CephFS.SubvolumeGroup)
-
-	})
-
-	t.Run("add a 4th mon to the 3rd cluster and subvolumegroup is preserved", func(t *testing.T) {
-		csiClusterConfigEntry3.Monitors = append(csiClusterConfigEntry3.Monitors, "10.11.12.13:5000")
-		s, err = updateCsiClusterConfig(s, "baba", &csiClusterConfigEntry3)
-		assert.NoError(t, err)
-		cc, err := parseCsiClusterConfig(s)
-		assert.NoError(t, err)
-		assert.Equal(t, 3, len(cc))
-		assert.Equal(t, 4, len(cc[2].Monitors))
-		assert.Equal(t, "mygroup", cc[2].CephFS.SubvolumeGroup)
-	})
-
-	t.Run("remove subvolumegroup", func(t *testing.T) {
-		csiClusterConfigEntry3.CephFS.SubvolumeGroup = ""
-		s, err = updateCsiClusterConfig(s, "baba", nil)
-		assert.NoError(t, err)
-		cc, err := parseCsiClusterConfig(s)
-		assert.NoError(t, err)
-		assert.Equal(t, 2, len(cc)) // we only have 2 clusters now
-	})
-
-	t.Run("add subvolumegroup and mons after", func(t *testing.T) {
-		csiClusterConfigEntry4 := CsiClusterConfigEntry{
-			CephFS: &CsiCephFSSpec{
-				SubvolumeGroup: "mygroup2",
-			},
-		}
-		s, err = updateCsiClusterConfig(s, "quatre", &csiClusterConfigEntry4)
-		assert.NoError(t, err)
-		cc, err := parseCsiClusterConfig(s)
-		assert.NoError(t, err)
-		assert.Equal(t, 3, len(cc), cc)
-		assert.Equal(t, 0, len(cc[2].Monitors))
-		assert.Equal(t, "mygroup2", cc[2].CephFS.SubvolumeGroup, cc)
-
-		csiClusterConfigEntry4.Monitors = []string{"10.1.1.1:5000", "10.1.1.2:5000", "10.1.1.3:5000"}
-		s, err = updateCsiClusterConfig(s, "quatre", &csiClusterConfigEntry4)
-		assert.NoError(t, err)
-		cc, err = parseCsiClusterConfig(s)
-		assert.NoError(t, err)
-		assert.Equal(t, "mygroup2", cc[2].CephFS.SubvolumeGroup)
-		assert.Equal(t, 3, len(cc[2].Monitors))
-	})
-
-	t.Run("does it return error on garbage input?", func(t *testing.T) {
-		_, err = updateCsiClusterConfig("qqq", "beta", &csiClusterConfigEntry2)
-		assert.Error(t, err)
-	})
+	s, err = updateCsiClusterConfig(s, "beta", mons2)
+	assert.NoError(t, err)
+	cc, err = parseCsiClusterConfig(s)
+	assert.NoError(t, err)
+	assert.Equal(t, len(cc), 2)
+	assert.Equal(t, cc[0].ClusterID, "alpha")
+	assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000")
+	assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000")
+	assert.Equal(t, len(cc[0].Monitors), 2)
+	assert.Equal(t, cc[1].ClusterID, "beta")
+	assert.Contains(t, cc[1].Monitors, "20.1.1.1:5000")
+	assert.Contains(t, cc[1].Monitors, "20.1.1.2:5000")
+	assert.Contains(t, cc[1].Monitors, "20.1.1.3:5000")
+	assert.Equal(t, len(cc[1].Monitors), 3)
+
+	// remove a mon from the 2nd cluster
+	delete(mons2, "blam")
+	s, err = updateCsiClusterConfig(s, "beta", mons2)
+	assert.NoError(t, err)
+	cc, err = parseCsiClusterConfig(s)
+	assert.NoError(t, err)
+	assert.Equal(t, len(cc), 2)
+	assert.Equal(t, cc[0].ClusterID, "alpha")
+	assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000")
+	assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000")
+	assert.Equal(t, len(cc[0].Monitors), 2)
+	assert.Equal(t, cc[1].ClusterID, "beta")
+	assert.Contains(t, cc[1].Monitors, "20.1.1.1:5000")
+	assert.Contains(t, cc[1].Monitors, "20.1.1.2:5000")
+	assert.Equal(t, len(cc[1].Monitors), 2)
+
+	// does it return error on garbage input?
+	_, err = updateCsiClusterConfig("qqq", "beta", mons2)
+	assert.Error(t, err)
 }
diff --git a/pkg/operator/ceph/file/filesystem_test.go b/pkg/operator/ceph/file/filesystem_test.go
index 64abb1715..63b6a3971 100644
--- a/pkg/operator/ceph/file/filesystem_test.go
+++ b/pkg/operator/ceph/file/filesystem_test.go
@@ -21,6 +21,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io/ioutil"
+	"os"
 	"path"
 	"reflect"
 	"strings"
@@ -317,11 +319,12 @@ func TestCreateFilesystem(t *testing.T) {
 	ctx := context.TODO()
 	var deploymentsUpdated *[]*apps.Deployment
 	mds.UpdateDeploymentAndWait, deploymentsUpdated = testopk8s.UpdateDeploymentAndWaitStub()
-	configDir := t.TempDir()
+	configDir, _ := ioutil.TempDir("", "")
 	fsName := "myfs"
 	addDataPoolCount := 0
 	createDataPoolCount := 0
 	executor := fsExecutor(t, fsName, configDir, false, &createDataPoolCount, &addDataPoolCount)
+	defer os.RemoveAll(configDir)
 	clientset := testop.New(t, 1)
 	context := &clusterd.Context{
 		Executor:  executor,
@@ -395,12 +398,13 @@ func TestUpgradeFilesystem(t *testing.T) {
 	ctx := context.TODO()
 	var deploymentsUpdated *[]*apps.Deployment
 	mds.UpdateDeploymentAndWait, deploymentsUpdated = testopk8s.UpdateDeploymentAndWaitStub()
-	configDir := t.TempDir()
+	configDir, _ := ioutil.TempDir("", "")
 
 	fsName := "myfs"
 	addDataPoolCount := 0
 	createDataPoolCount := 0
 	executor := fsExecutor(t, fsName, configDir, false, &createDataPoolCount, &addDataPoolCount)
+	defer os.RemoveAll(configDir)
 	clientset := testop.New(t, 1)
 	context := &clusterd.Context{
 		Executor:  executor,
@@ -514,7 +518,7 @@ func TestUpgradeFilesystem(t *testing.T) {
 func TestCreateNopoolFilesystem(t *testing.T) {
 	ctx := context.TODO()
 	clientset := testop.New(t, 3)
-	configDir := t.TempDir()
+	configDir, _ := ioutil.TempDir("", "")
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
 			if strings.Contains(command, "ceph-authtool") {
@@ -526,6 +530,7 @@ func TestCreateNopoolFilesystem(t *testing.T) {
 			return "", errors.New("unknown command error")
 		},
 	}
+	defer os.RemoveAll(configDir)
 	context := &clusterd.Context{
 		Executor:  executor,
 		ConfigDir: configDir,
diff --git a/pkg/operator/ceph/file/subvolumegroup/controller_test.go b/pkg/operator/ceph/file/subvolumegroup/controller_test.go
index fca3bff14..6fffbdea5 100644
--- a/pkg/operator/ceph/file/subvolumegroup/controller_test.go
+++ b/pkg/operator/ceph/file/subvolumegroup/controller_test.go
@@ -21,14 +21,11 @@ import (
 	"os"
 	"testing"
 
-	"github.com/pkg/errors"
-
 	"github.com/coreos/pkg/capnslog"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	rookclient "github.com/rook/rook/pkg/client/clientset/versioned/fake"
 	"github.com/rook/rook/pkg/client/clientset/versioned/scheme"
 	"github.com/rook/rook/pkg/clusterd"
-	"github.com/rook/rook/pkg/operator/ceph/csi"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	testop "github.com/rook/rook/pkg/operator/test"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
@@ -47,6 +44,7 @@ func TestCephClientController(t *testing.T) {
 	capnslog.SetGlobalLogLevel(capnslog.DEBUG)
 	os.Setenv("ROOK_LOG_LEVEL", "DEBUG")
 
+	logger.Info("RUN 1")
 	var (
 		name      = "group-a"
 		namespace = "rook-ceph"
@@ -193,11 +191,11 @@ func TestCephClientController(t *testing.T) {
 
 		executor = &exectest.MockExecutor{
 			MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
-				if args[0] == "fs" && args[1] == "subvolumegroup" && args[2] == "create" {
+				if args[0] == "auth" && args[1] == "fs" && args[2] == "subvolumegroup" && args[3] == "create" {
 					return "", nil
 				}
 
-				return "", errors.Errorf("unknown command. %v", args)
+				return "", nil
 			},
 		}
 		c.Executor = executor
@@ -208,86 +206,15 @@ func TestCephClientController(t *testing.T) {
 			client:           cl,
 			scheme:           s,
 			context:          c,
-			opManagerContext: ctx,
+			opManagerContext: context.TODO(),
 		}
 
-		// Enable CSI
-		csi.EnableRBD = true
-		os.Setenv("POD_NAMESPACE", namespace)
-		// Create CSI config map
-		ownerRef := &metav1.OwnerReference{}
-		ownerInfo := k8sutil.NewOwnerInfoWithOwnerRef(ownerRef, "")
-		err = csi.CreateCsiConfigMap(namespace, c.Clientset, ownerInfo)
-		assert.NoError(t, err)
-
 		res, err := r.Reconcile(ctx, req)
 		assert.NoError(t, err)
 		assert.False(t, res.Requeue)
 
-		err = r.client.Get(ctx, req.NamespacedName, cephFilesystemSubVolumeGroup)
+		err = r.client.Get(context.TODO(), req.NamespacedName, cephFilesystemSubVolumeGroup)
 		assert.NoError(t, err)
 		assert.Equal(t, cephv1.ConditionReady, cephFilesystemSubVolumeGroup.Status.Phase)
-		assert.NotEmpty(t, cephFilesystemSubVolumeGroup.Status.Info["clusterID"])
-
-		// test that csi configmap is created
-		cm, err := c.Clientset.CoreV1().ConfigMaps(namespace).Get(ctx, csi.ConfigName, metav1.GetOptions{})
-		assert.NoError(t, err)
-		assert.NotEmpty(t, cm.Data[csi.ConfigKey])
-		assert.Contains(t, cm.Data[csi.ConfigKey], "clusterID")
-		assert.Contains(t, cm.Data[csi.ConfigKey], "group-a")
-		err = c.Clientset.CoreV1().ConfigMaps(namespace).Delete(ctx, csi.ConfigName, metav1.DeleteOptions{})
-		assert.NoError(t, err)
 	})
-
-	t.Run("success - external mode csi config is updated", func(t *testing.T) {
-		cephCluster.Spec.External.Enable = true
-		objects := []runtime.Object{
-			cephFilesystemSubVolumeGroup,
-			cephCluster,
-		}
-		// Create a fake client to mock API calls.
-		cl = fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(objects...).Build()
-		c.Client = cl
-
-		s.AddKnownTypes(cephv1.SchemeGroupVersion, &cephv1.CephBlockPoolList{})
-		// Create a ReconcileCephFilesystemSubVolumeGroup object with the scheme and fake client.
-		r = &ReconcileCephFilesystemSubVolumeGroup{
-			client:           cl,
-			scheme:           s,
-			context:          c,
-			opManagerContext: ctx,
-		}
-
-		// Enable CSI
-		csi.EnableRBD = true
-		os.Setenv("POD_NAMESPACE", namespace)
-		// Create CSI config map
-		ownerRef := &metav1.OwnerReference{}
-		ownerInfo := k8sutil.NewOwnerInfoWithOwnerRef(ownerRef, "")
-		err := csi.CreateCsiConfigMap(namespace, c.Clientset, ownerInfo)
-		assert.NoError(t, err)
-
-		res, err := r.Reconcile(ctx, req)
-		assert.NoError(t, err)
-		assert.False(t, res.Requeue)
-
-		err = r.client.Get(ctx, req.NamespacedName, cephFilesystemSubVolumeGroup)
-		assert.NoError(t, err)
-		assert.Equal(t, cephv1.ConditionConnected, cephFilesystemSubVolumeGroup.Status.Phase)
-		assert.NotEmpty(t, cephFilesystemSubVolumeGroup.Status.Info["clusterID"])
-
-		// test that csi configmap is created
-		cm, err := c.Clientset.CoreV1().ConfigMaps(namespace).Get(ctx, csi.ConfigName, metav1.GetOptions{})
-		assert.NoError(t, err)
-		assert.NotEmpty(t, cm.Data[csi.ConfigKey])
-		assert.Contains(t, cm.Data[csi.ConfigKey], "clusterID")
-		assert.Contains(t, cm.Data[csi.ConfigKey], "group-a")
-	})
-}
-
-func Test_buildClusterID(t *testing.T) {
-	longName := "foooooooooooooooooooooooooooooooooooooooooooo"
-	cephFilesystemSubVolumeGroup := &cephv1.CephFilesystemSubVolumeGroup{ObjectMeta: metav1.ObjectMeta{Namespace: "rook-ceph", Name: longName}, Spec: cephv1.CephFilesystemSubVolumeGroupSpec{FilesystemName: "myfs"}}
-	clusterID := buildClusterID(cephFilesystemSubVolumeGroup)
-	assert.Equal(t, "29e92135b7e8c014079b9f9f3566777d", clusterID)
 }
diff --git a/pkg/operator/ceph/object/bucket/rgw-handlers_test.go b/pkg/operator/ceph/object/bucket/rgw-handlers_test.go
deleted file mode 100644
index ba487b8d1..000000000
--- a/pkg/operator/ceph/object/bucket/rgw-handlers_test.go
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
-Copyright 2022 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 bucket
-
-import (
-	"bytes"
-	"encoding/json"
-	"fmt"
-	"io/ioutil"
-	"net/http"
-	"testing"
-
-	"github.com/ceph/go-ceph/rgw/admin"
-	rookclient "github.com/rook/rook/pkg/client/clientset/versioned/fake"
-	"github.com/rook/rook/pkg/clusterd"
-	"github.com/rook/rook/pkg/daemon/ceph/client"
-	cephobject "github.com/rook/rook/pkg/operator/ceph/object"
-	"github.com/rook/rook/pkg/operator/test"
-	"github.com/stretchr/testify/assert"
-)
-
-type statusError struct {
-	Code      string `json:"Code,omitempty"`
-	RequestID string `json:"RequestId,omitempty"`
-	HostID    string `json:"HostId,omitempty"`
-}
-
-func TestDeleteOBCResource(t *testing.T) {
-	clusterInfo := client.AdminTestClusterInfo("ns")
-	p := NewProvisioner(&clusterd.Context{RookClientset: rookclient.NewSimpleClientset(), Clientset: test.New(t, 1)}, clusterInfo)
-	mockClient := func(errCodeRemoveBucket string, errCodeGetBucketInfo string) *cephobject.MockClient {
-		return &cephobject.MockClient{
-			MockDo: func(req *http.Request) (*http.Response, error) {
-				if req.URL.Path == "rook-ceph-rgw-my-store.mycluster.svc/admin/bucket" {
-					if req.Method == http.MethodDelete {
-						status, _ := json.Marshal(statusError{errCodeRemoveBucket, "requestid", "hostid"})
-						return &http.Response{
-							StatusCode: 404,
-							Body:       ioutil.NopCloser(bytes.NewReader([]byte(status))),
-						}, nil
-					}
-					if req.Method == http.MethodGet {
-						status, _ := json.Marshal(statusError{errCodeGetBucketInfo, "requestid", "hostid"})
-						return &http.Response{
-							StatusCode: 404,
-							Body:       ioutil.NopCloser(bytes.NewReader([]byte(status))),
-						}, nil
-					}
-				}
-				return nil, fmt.Errorf("unexpected request: %q. method %q. path %q", req.URL.RawQuery, req.Method, req.URL.Path)
-			},
-		}
-	}
-
-	t.Run("remove bucket returns NoSuchBucket", func(t *testing.T) {
-		adminClient, err := admin.New("rook-ceph-rgw-my-store.mycluster.svc", "53S6B9S809NUP19IJ2K3", "1bXPegzsGClvoGAiJdHQD1uOW2sQBLAZM9j9VtXR", mockClient("NoSuchBucket", ""))
-		assert.NoError(t, err)
-		p.adminOpsClient = adminClient
-		err = p.deleteOBCResource("bucket")
-		assert.NoError(t, err)
-	})
-
-	t.Run("remove bucket returns NoSuchKey and get bucket info returns NoSuchBucket", func(t *testing.T) {
-		adminClient, err := admin.New("rook-ceph-rgw-my-store.mycluster.svc", "53S6B9S809NUP19IJ2K3", "1bXPegzsGClvoGAiJdHQD1uOW2sQBLAZM9j9VtXR", mockClient("NoSuchKey", "NoSuchBucket"))
-		assert.NoError(t, err)
-		p.adminOpsClient = adminClient
-		err = p.deleteOBCResource("bucket")
-		assert.NoError(t, err)
-	})
-
-	t.Run("remove bucket returns NoSuchKey and get bucket info returns an error other than NoSuchBucket", func(t *testing.T) {
-		adminClient, err := admin.New("rook-ceph-rgw-my-store.mycluster.svc", "53S6B9S809NUP19IJ2K3", "1bXPegzsGClvoGAiJdHQD1uOW2sQBLAZM9j9VtXR", mockClient("NoSuchKey", "NoSuchKey"))
-		assert.NoError(t, err)
-		p.adminOpsClient = adminClient
-		err = p.deleteOBCResource("bucket")
-		assert.Error(t, err)
-	})
-}
diff --git a/pkg/operator/ceph/object/controller_test.go b/pkg/operator/ceph/object/controller_test.go
index aba42c08e..06f0ca247 100644
--- a/pkg/operator/ceph/object/controller_test.go
+++ b/pkg/operator/ceph/object/controller_test.go
@@ -21,7 +21,6 @@ import (
 	"context"
 	"net/http"
 	"os"
-	"reflect"
 	"testing"
 	"time"
 
@@ -746,48 +745,3 @@ func TestCephObjectStoreControllerMultisite(t *testing.T) {
 		assert.True(t, calledCommitConfigChanges)
 	})
 }
-
-func TestDiffVersions(t *testing.T) {
-	executor := &exectest.MockExecutor{
-		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
-			if args[0] == "versions" {
-				return `{
-    "mon": {
-        "ceph version 17.0.0-9718-g4ff72306 (4ff723061fc15c803dcf6556d02f56bdf56de5fa) quincy (dev)": 3
-    },
-    "mgr": {
-        "ceph version 17.0.0-9718-g4ff72306 (4ff723061fc15c803dcf6556d02f56bdf56de5fa) quincy (dev)": 1
-    },
-    "osd": {
-        "ceph version 17.0.0-9718-g4ff72306 (4ff723061fc15c803dcf6556d02f56bdf56de5fa) quincy (dev)": 3
-    },
-    "mds": {
-        "ceph version 17.0.0-9718-g4ff72306 (4ff723061fc15c803dcf6556d02f56bdf56de5fa) quincy (dev)": 2
-    },
-    "rgw": {
-        "ceph version 17.0.0-9718-g4ff72306 (4ff723061fc15c803dcf6556d02f56bdf56de5fa) quincy (dev)": 1
-    },
-    "overall": {
-        "ceph version 17.0.0-9718-g4ff72306 (4ff723061fc15c803dcf6556d02f56bdf56de5fa) quincy (dev)": 10
-    }
-}`, nil
-			}
-			return "", errors.Errorf("unknown command %s %v", command, args)
-		},
-	}
-	c := &clusterd.Context{Executor: executor}
-
-	// desiredCephVersion comes from DetectCephVersion() (ceph --version) which uses ExtractCephVersion()
-	desiredCephVersion, err := cephver.ExtractCephVersion("ceph version 17.0.0-9718-g4ff72306 (4ff723061fc15c803dcf6556d02f56bdf56de5fa) quincy (dev)")
-	assert.NoError(t, err)
-
-	// runningCephVersion comes from LeastUptodateDaemonVersion()
-	runningCephVersion, err := client.LeastUptodateDaemonVersion(c, &client.ClusterInfo{Context: context.TODO()}, "mon")
-	assert.NoError(t, err)
-
-	// Compares the pointer's address with the struct so it's wrong
-	assert.False(t, reflect.DeepEqual(runningCephVersion, desiredCephVersion))
-
-	// Compares the actual value of the pointer by dereferencing the pointer
-	assert.True(t, reflect.DeepEqual(runningCephVersion, *desiredCephVersion))
-}
diff --git a/pkg/operator/ceph/object/rgw_test.go b/pkg/operator/ceph/object/rgw_test.go
index e90092382..e720bcd1a 100644
--- a/pkg/operator/ceph/object/rgw_test.go
+++ b/pkg/operator/ceph/object/rgw_test.go
@@ -18,6 +18,8 @@ package object
 
 import (
 	"context"
+	"io/ioutil"
+	"os"
 	"testing"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -49,7 +51,8 @@ func TestStartRGW(t *testing.T) {
 		},
 	}
 
-	configDir := t.TempDir()
+	configDir, _ := ioutil.TempDir("", "")
+	defer os.RemoveAll(configDir)
 	info := clienttest.CreateTestClusterInfo(1)
 	context := &clusterd.Context{Clientset: clientset, Executor: executor, ConfigDir: configDir}
 	store := simpleStore()
diff --git a/pkg/operator/ceph/pool/controller_test.go b/pkg/operator/ceph/pool/controller_test.go
index 9ab58a50c..21dbec2ec 100644
--- a/pkg/operator/ceph/pool/controller_test.go
+++ b/pkg/operator/ceph/pool/controller_test.go
@@ -56,9 +56,6 @@ func TestCreatePool(t *testing.T) {
 					return `{"k":"2","m":"1","plugin":"jerasure","technique":"reed_sol_van"}`, nil
 				}
 				if args[0] == "osd" && args[1] == "pool" && args[2] == "application" {
-					if args[3] == "get" {
-						return ``, nil
-					}
 					assert.Equal(t, "enable", args[3])
 					if args[5] != "rbd" {
 						enabledMetricsApp = true
diff --git a/pkg/operator/k8sutil/resources_test.go b/pkg/operator/k8sutil/resources_test.go
index 5eaba5fd0..fc47c4fed 100644
--- a/pkg/operator/k8sutil/resources_test.go
+++ b/pkg/operator/k8sutil/resources_test.go
@@ -37,18 +37,17 @@ func TestMergeResourceRequirements(t *testing.T) {
 	first = v1.ResourceRequirements{}
 	second = v1.ResourceRequirements{
 		Limits: v1.ResourceList{
-			v1.ResourceCPU:     *resource.NewQuantity(100.0, resource.BinarySI),
-			v1.ResourceStorage: *resource.NewQuantity(50.0, resource.BinarySI),
+			v1.ResourceCPU: *resource.NewQuantity(100.0, resource.BinarySI),
 		},
 		Requests: v1.ResourceList{
-			v1.ResourceName("foo"): *resource.NewQuantity(23.0, resource.BinarySI),
+			v1.ResourceMemory: *resource.NewQuantity(1337.0, resource.BinarySI),
 		},
 	}
 	result = MergeResourceRequirements(first, second)
-	assert.Equal(t, 2, len(result.Limits))
+	assert.Equal(t, 1, len(result.Limits))
 	assert.Equal(t, 1, len(result.Requests))
 	assert.Equal(t, "100", result.Limits.Cpu().String())
-	assert.Equal(t, "50", result.Limits.Storage().String())
+	assert.Equal(t, "1337", result.Requests.Memory().String())
 
 	first = v1.ResourceRequirements{
 		Limits: v1.ResourceList{
diff --git a/tests/framework/clients/bucket.go b/tests/framework/clients/bucket.go
index 037bcdf19..cfcc61ad4 100644
--- a/tests/framework/clients/bucket.go
+++ b/tests/framework/clients/bucket.go
@@ -150,22 +150,18 @@ func (b *BucketOperation) CheckOBMaxObject(obcName, maxobject string) bool {
 }
 
 // Checks the bucket notifications set on RGW backend bucket
-func (b *BucketOperation) CheckBucketNotificationSetonRGW(namespace, storeName, obcName, bucketname, notificationName string, helper *TestClient, tlsEnabled bool) bool {
+func (b *BucketOperation) CheckBucketNotificationSetonRGW(namespace, storeName, obcName, bucketname, notificationName, s3endpoint string) bool {
 	var s3client *rgw.S3Agent
-	var err error
-	s3endpoint, _ := helper.ObjectClient.GetEndPointUrl(namespace, storeName)
-	s3AccessKey, _ := helper.BucketClient.GetAccessKey(obcName)
-	s3SecretKey, _ := helper.BucketClient.GetSecretKey(obcName)
-	if tlsEnabled {
-		s3client, err = rgw.NewInsecureS3Agent(s3AccessKey, s3SecretKey, s3endpoint, "", true)
-	} else {
-		s3client, err = rgw.NewS3Agent(s3AccessKey, s3SecretKey, s3endpoint, "", true, nil)
-	}
+	s3AccessKey, _ := b.GetAccessKey(obcName)
+	s3SecretKey, _ := b.GetSecretKey(obcName)
+
+	//TODO : add TLS check
+	s3client, err := rgw.NewS3Agent(s3AccessKey, s3SecretKey, s3endpoint, "", true, nil)
 	if err != nil {
-		logger.Infof("failed to s3client due to %v", err)
+		logger.Errorf("S3 client creation failed with error %v", err)
 		return false
 	}
-	logger.Infof("endpoint (%s) Accesskey (%s) secret (%s)", s3endpoint, s3AccessKey, s3SecretKey)
+
 	notifications, err := s3client.Client.GetBucketNotificationConfiguration(&s3.GetBucketNotificationConfigurationRequest{
 		Bucket: &bucketname,
 	})
diff --git a/tests/framework/installer/ceph_helm_installer.go b/tests/framework/installer/ceph_helm_installer.go
index fd07352d4..f61a6feed 100644
--- a/tests/framework/installer/ceph_helm_installer.go
+++ b/tests/framework/installer/ceph_helm_installer.go
@@ -62,6 +62,9 @@ func (h *CephInstaller) configureRookOperatorViaHelm(upgrade bool) error {
 	if err := h.k8shelper.CreateNamespace(h.settings.OperatorNamespace); err != nil {
 		return errors.Errorf("failed to create namespace %s. %v", h.settings.Namespace, err)
 	}
+	if err := h.k8shelper.CreateOpConfigMap(h.settings.OperatorNamespace); err != nil {
+		return errors.Errorf("failed to create operator config map. %v", err)
+	}
 	if err := h.startAdmissionController(); err != nil {
 		return errors.Errorf("failed to start admission controllers. %v", err)
 	}
diff --git a/tests/framework/utils/k8s_helper.go b/tests/framework/utils/k8s_helper.go
index f57440c11..8d4d71f57 100644
--- a/tests/framework/utils/k8s_helper.go
+++ b/tests/framework/utils/k8s_helper.go
@@ -37,6 +37,7 @@ import (
 	rookclient "github.com/rook/rook/pkg/client/clientset/versioned"
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/crash"
+	"github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/rook/rook/pkg/util/exec"
 	"github.com/stretchr/testify/require"
@@ -374,6 +375,22 @@ func (k8sh *K8sHelper) CreateNamespace(namespace string) error {
 	return nil
 }
 
+func (k8sh *K8sHelper) CreateOpConfigMap(namespace string) error {
+	ctx := context.TODO()
+	cm := &v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      controller.OperatorSettingConfigMapName,
+			Namespace: namespace,
+		},
+	}
+	_, err := k8sh.Clientset.CoreV1().ConfigMaps(namespace).Create(ctx, cm, metav1.CreateOptions{})
+	if err != nil && !kerrors.IsAlreadyExists(err) {
+		return fmt.Errorf("failed to create operator config map %s. %+v", controller.OperatorSettingConfigMapName, err)
+	}
+
+	return nil
+}
+
 func (k8sh *K8sHelper) CountPodsWithLabel(label string, namespace string) (int, error) {
 	ctx := context.TODO()
 	options := metav1.ListOptions{LabelSelector: label}
diff --git a/tests/framework/utils/snapshot.go b/tests/framework/utils/snapshot.go
index 1a20d6144..b333ec2f6 100644
--- a/tests/framework/utils/snapshot.go
+++ b/tests/framework/utils/snapshot.go
@@ -27,7 +27,7 @@ import (
 const (
 	// snapshotterVersion from which the snapshotcontroller and CRD will be
 	// installed
-	snapshotterVersion = "v5.0.1"
+	snapshotterVersion = "v4.0.0"
 	repoURL            = "https://raw.githubusercontent.com/kubernetes-csi/external-snapshotter"
 	rbacPath           = "deploy/kubernetes/snapshot-controller/rbac-snapshot-controller.yaml"
 	controllerPath     = "deploy/kubernetes/snapshot-controller/setup-snapshot-controller.yaml"
@@ -40,8 +40,6 @@ const (
 // CheckSnapshotISReadyToUse checks snapshot is ready to use
 func (k8sh *K8sHelper) CheckSnapshotISReadyToUse(name, namespace string, retries int) (bool, error) {
 	for i := 0; i < retries; i++ {
-		// sleep first and try to check snapshot is ready to cover the error cases.
-		time.Sleep(time.Duration(i) * time.Second)
 		ready, err := k8sh.executor.ExecuteCommandWithOutput("kubectl", "get", "volumesnapshot", name, "--namespace", namespace, "-o", "jsonpath={.status.readyToUse}")
 		if err != nil {
 			return false, err
@@ -54,6 +52,7 @@ func (k8sh *K8sHelper) CheckSnapshotISReadyToUse(name, namespace string, retries
 		if val {
 			return true, nil
 		}
+		time.Sleep(RetryInterval * time.Second)
 	}
 	return false, fmt.Errorf("giving up waiting for %q snapshot in namespace %q", name, namespace)
 }
@@ -84,21 +83,21 @@ func (k8sh *K8sHelper) snapshotController(action string) error {
 // WaitForSnapshotController check snapshotcontroller is ready within given
 // retries count.
 func (k8sh *K8sHelper) WaitForSnapshotController(retries int) error {
-	namespace := "kube-system"
+	namespace := "default"
 	ctx := context.TODO()
 	snapshotterName := "snapshot-controller"
 	for i := 0; i < retries; i++ {
-		ss, err := k8sh.Clientset.AppsV1().Deployments(namespace).Get(ctx, snapshotterName, metav1.GetOptions{})
+		ss, err := k8sh.Clientset.AppsV1().StatefulSets(namespace).Get(ctx, snapshotterName, metav1.GetOptions{})
 		if err != nil && !apierrors.IsNotFound(err) {
 			return err
 		}
 		if ss.Status.ReadyReplicas > 0 && ss.Status.ReadyReplicas == ss.Status.Replicas {
 			return nil
 		}
-		logger.Infof("waiting for %q deployment in namespace %q (readyreplicas %d < replicas %d)", snapshotterName, namespace, ss.Status.ReadyReplicas, ss.Status.Replicas)
+		logger.Infof("waiting for %q statufulset in namespace %q (readyreplicas %d < replicas %d)", snapshotterName, namespace, ss.Status.ReadyReplicas, ss.Status.Replicas)
 		time.Sleep(RetryInterval * time.Second)
 	}
-	return fmt.Errorf("giving up waiting for %q deployment in namespace %q", snapshotterName, namespace)
+	return fmt.Errorf("giving up waiting for %q statufulset in namespace %q", snapshotterName, namespace)
 }
 
 // CreateSnapshotController creates the snapshotcontroller and required RBAC
@@ -113,33 +112,20 @@ func (k8sh *K8sHelper) DeleteSnapshotController() error {
 
 // snapshotCRD can be used for creating or deleting the snapshot CRD's
 func (k8sh *K8sHelper) snapshotCRD(action string) error {
-	// setting validate=false to skip CRD validation during create operation to
-	// support lower Kubernetes versions.
-	args := func(crdpath string) []string {
-		a := []string{
-			action,
-			"-f",
-			crdpath,
-		}
-		if action == "create" {
-			a = append(a, "--validate=false")
-		}
-		return a
-	}
 	snapshotClassCRD := fmt.Sprintf("%s/%s/%s", repoURL, snapshotterVersion, snapshotClassCRDPath)
-	_, err := k8sh.Kubectl(args(snapshotClassCRD)...)
+	_, err := k8sh.Kubectl(action, "-f", snapshotClassCRD)
 	if err != nil {
 		return err
 	}
 
 	snapshotContentsCRD := fmt.Sprintf("%s/%s/%s", repoURL, snapshotterVersion, volumeSnapshotContentsCRDPath)
-	_, err = k8sh.Kubectl(args(snapshotContentsCRD)...)
+	_, err = k8sh.Kubectl(action, "-f", snapshotContentsCRD)
 	if err != nil {
 		return err
 	}
 
 	snapshotCRD := fmt.Sprintf("%s/%s/%s", repoURL, snapshotterVersion, volumeSnapshotCRDPath)
-	_, err = k8sh.Kubectl(args(snapshotCRD)...)
+	_, err = k8sh.Kubectl(action, "-f", snapshotCRD)
 	if err != nil {
 		return err
 	}
diff --git a/tests/integration/ceph_base_block_test.go b/tests/integration/ceph_base_block_test.go
index 1615c739b..53d668d5a 100644
--- a/tests/integration/ceph_base_block_test.go
+++ b/tests/integration/ceph_base_block_test.go
@@ -131,20 +131,11 @@ func blockCSISnapshotTest(helper *clients.TestClient, k8sh *utils.K8sHelper, s s
 	logger.Infof("install snapshot CRD")
 	err := k8sh.CreateSnapshotCRD()
 	require.NoError(s.T(), err)
+
 	logger.Infof("install snapshot controller")
 	err = k8sh.CreateSnapshotController()
 	require.NoError(s.T(), err)
-	// cleanup the CRD and controller in defer to make sure the CRD and
-	// controller are removed as filesystem test also install CRD and controller.
-	defer func() {
-		logger.Infof("delete snapshot-controller")
-		err = k8sh.DeleteSnapshotController()
-		require.NoError(s.T(), err)
-
-		logger.Infof("delete snapshot CRD")
-		err = k8sh.DeleteSnapshotCRD()
-		require.NoError(s.T(), err)
-	}()
+
 	logger.Infof("check snapshot controller is running")
 	err = k8sh.WaitForSnapshotController(15)
 	require.NoError(s.T(), err)
@@ -246,6 +237,15 @@ func blockCSISnapshotTest(helper *clients.TestClient, k8sh *utils.K8sHelper, s s
 
 	err = helper.BlockClient.DeleteSnapshotClass(snapshotClassName, snapshotDeletePolicy, namespace)
 	require.NoError(s.T(), err)
+	logger.Infof("delete snapshot-controller")
+
+	err = k8sh.DeleteSnapshotController()
+	require.NoError(s.T(), err)
+	logger.Infof("delete snapshot CRD")
+
+	// remove snapshotcontroller and delete snapshot CRD
+	err = k8sh.DeleteSnapshotCRD()
+	require.NoError(s.T(), err)
 }
 
 // Smoke Test for Block Storage - Test check the following operations on Block Storage in order
diff --git a/tests/integration/ceph_base_file_test.go b/tests/integration/ceph_base_file_test.go
index ac26d458b..263b3bc43 100644
--- a/tests/integration/ceph_base_file_test.go
+++ b/tests/integration/ceph_base_file_test.go
@@ -118,17 +118,7 @@ func fileSystemCSISnapshotTest(helper *clients.TestClient, k8sh *utils.K8sHelper
 	logger.Infof("install snapshot controller")
 	err = k8sh.CreateSnapshotController()
 	require.NoError(s.T(), err)
-	// cleanup the CRD and controller in defer to make sure the CRD and
-	// controller are removed as block test also install CRD and controller.
-	defer func() {
-		logger.Infof("delete snapshot-controller")
-		err = k8sh.DeleteSnapshotController()
-		require.NoError(s.T(), err)
-
-		logger.Infof("delete snapshot CRD")
-		err = k8sh.DeleteSnapshotCRD()
-		require.NoError(s.T(), err)
-	}()
+
 	logger.Infof("check snapshot controller is running")
 	err = k8sh.WaitForSnapshotController(15)
 	require.NoError(s.T(), err)
@@ -231,6 +221,14 @@ func fileSystemCSISnapshotTest(helper *clients.TestClient, k8sh *utils.K8sHelper
 	err = helper.FSClient.DeleteSnapshotClass(snapshotClassName, snapshotDeletePolicy, namespace)
 	require.NoError(s.T(), err)
 	logger.Infof("delete snapshot-controller")
+
+	err = k8sh.DeleteSnapshotController()
+	require.NoError(s.T(), err)
+	logger.Infof("delete snapshot CRD")
+
+	// remove snapshotcontroller and delete snapshot CRD
+	err = k8sh.DeleteSnapshotCRD()
+	require.NoError(s.T(), err)
 }
 
 // Smoke Test for File System Storage - Test check the following operations on Filesystem Storage in order
diff --git a/tests/integration/ceph_base_object_test.go b/tests/integration/ceph_base_object_test.go
index 7e650e21a..9f1c8cc86 100644
--- a/tests/integration/ceph_base_object_test.go
+++ b/tests/integration/ceph_base_object_test.go
@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
+	"os"
 	"path/filepath"
 	"strconv"
 	"testing"
@@ -254,7 +255,9 @@ func generateRgwTlsCertSecret(t *testing.T, helper *clients.TestClient, k8sh *ut
 	ctx := context.TODO()
 	root, err := utils.FindRookRoot()
 	require.NoError(t, err, "failed to get rook root")
-	tlscertdir := t.TempDir()
+	tlscertdir, err := ioutil.TempDir(root, "tlscertdir")
+	require.NoError(t, err, "failed to create directory for TLS certs")
+	defer os.RemoveAll(tlscertdir)
 	cmdArgs := utils.CommandArgs{Command: filepath.Join(root, "tests/scripts/generate-tls-config.sh"),
 		CmdArgs: []string{tlscertdir, rgwServiceName, namespace}}
 	cmdOut := utils.ExecuteCommand(cmdArgs)
diff --git a/tests/integration/ceph_bucket_notification_test.go b/tests/integration/ceph_bucket_notification_test.go
index 21297f7b2..91b285b1c 100644
--- a/tests/integration/ceph_bucket_notification_test.go
+++ b/tests/integration/ceph_bucket_notification_test.go
@@ -23,20 +23,26 @@ import (
 
 	"github.com/rook/rook/pkg/daemon/ceph/client"
 	rgw "github.com/rook/rook/pkg/operator/ceph/object"
-	"github.com/rook/rook/tests/framework/clients"
 	"github.com/rook/rook/tests/framework/utils"
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/suite"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-func testBucketNotifications(s suite.Suite, helper *clients.TestClient, k8sh *utils.K8sHelper, namespace, storeName string) {
+func (s *ObjectSuite) TestBucketNotifications() {
 	if utils.IsPlatformOpenShift() {
 		s.T().Skip("bucket notification tests skipped on openshift")
 	}
 
+	objectStoreServicePrefix = objectStoreServicePrefixUniq
 	bucketNotificationLabelPrefix := "bucket-notification-"
+	storeName := "test-store-bucket-notification"
+	tlsEnable := false
+	namespace := s.settings.Namespace
 	obcNamespace := "default"
+	helper := s.helper
+	k8sh := s.k8sh
+	logger.Infof("Running on Rook Cluster %s", namespace)
+	createCephObjectStore(s.T(), helper, k8sh, namespace, storeName, 3, tlsEnable)
 
 	ctx := context.TODO()
 	clusterInfo := client.AdminTestClusterInfo(namespace)
@@ -51,7 +57,7 @@ func testBucketNotifications(s suite.Suite, helper *clients.TestClient, k8sh *ut
 	notificationName := "my-notification"
 	topicName := "my-topic"
 	httpEndpointService := "my-notification-sink"
-	logger.Infof("Testing Bucket Notifications on %s", storeName)
+	s3endpoint, _ := helper.ObjectClient.GetEndPointUrl(namespace, storeName)
 
 	t.Run("create CephBucketTopic", func(t *testing.T) {
 		err := helper.TopicClient.CreateTopic(topicName, storeName, httpEndpointService)
@@ -105,7 +111,7 @@ func testBucketNotifications(s suite.Suite, helper *clients.TestClient, k8sh *ut
 
 	t.Run("check CephBucketNotification created for bucket", func(t *testing.T) {
 		notificationPresent := utils.Retry(12, 2*time.Second, "notification is created for bucket", func() bool {
-			return helper.BucketClient.CheckBucketNotificationSetonRGW(namespace, storeName, obcName, bucketname, notificationName, helper, objectStore.Spec.IsTLSEnabled())
+			return helper.BucketClient.CheckBucketNotificationSetonRGW(namespace, storeName, obcName, bucketname, notificationName, s3endpoint)
 		})
 		assert.True(t, notificationPresent)
 		logger.Info("CephBucketNotification created successfully on bucket")
@@ -121,7 +127,7 @@ func testBucketNotifications(s suite.Suite, helper *clients.TestClient, k8sh *ut
 		notificationPresent := utils.Retry(12, 2*time.Second, "notification is created for bucket", func() bool {
 			// TODO : add api to fetch all the notification from backend to see if it is unaffected
 			t.Skipped()
-			return helper.BucketClient.CheckBucketNotificationSetonRGW(namespace, storeName, obcName, bucketname, notificationName, helper, objectStore.Spec.IsTLSEnabled())
+			return helper.BucketClient.CheckBucketNotificationSetonRGW(namespace, storeName, obcName, bucketname, notificationName, s3endpoint)
 		})
 		assert.True(t, notificationPresent)
 	})
@@ -136,7 +142,7 @@ func testBucketNotifications(s suite.Suite, helper *clients.TestClient, k8sh *ut
 		notificationPresent := utils.Retry(12, 2*time.Second, "notification is created for bucket", func() bool {
 			// TODO : add api to fetch all the notification from backend to see if it is unaffected
 			t.Skipped()
-			return helper.BucketClient.CheckBucketNotificationSetonRGW(namespace, storeName, obcName, bucketname, notificationName, helper, objectStore.Spec.IsTLSEnabled())
+			return helper.BucketClient.CheckBucketNotificationSetonRGW(namespace, storeName, obcName, bucketname, notificationName, s3endpoint)
 		})
 		assert.True(t, notificationPresent)
 	})
@@ -152,7 +158,7 @@ func testBucketNotifications(s suite.Suite, helper *clients.TestClient, k8sh *ut
 		// check whether existing bucket notification uneffected
 		var notificationPresent bool
 		for i := 0; i < 4; i++ {
-			notificationPresent = helper.BucketClient.CheckBucketNotificationSetonRGW(namespace, storeName, obcName, bucketname, notificationName, helper, objectStore.Spec.IsTLSEnabled())
+			notificationPresent = helper.BucketClient.CheckBucketNotificationSetonRGW(namespace, storeName, obcName, bucketname, notificationName, s3endpoint)
 			if !notificationPresent {
 				break
 			}
@@ -190,7 +196,7 @@ func testBucketNotifications(s suite.Suite, helper *clients.TestClient, k8sh *ut
 		t.Run("new-notification should be configured for bucket", func(t *testing.T) {
 			// check whether bucket notification added
 			notificationPresent := utils.Retry(12, 2*time.Second, "notification is created for bucket", func() bool {
-				return helper.BucketClient.CheckBucketNotificationSetonRGW(namespace, storeName, obcName, bucketname, newNotificationName, helper, objectStore.Spec.IsTLSEnabled())
+				return helper.BucketClient.CheckBucketNotificationSetonRGW(namespace, storeName, obcName, bucketname, newNotificationName, s3endpoint)
 			})
 			assert.True(t, notificationPresent)
 		})
@@ -263,7 +269,7 @@ func testBucketNotifications(s suite.Suite, helper *clients.TestClient, k8sh *ut
 		t.Run("notification should be configured after creating the topic", func(t *testing.T) {
 			// check whether bucket notification added, should pass since topic got created
 			notificationPresent := utils.Retry(12, 2*time.Second, "notification is created for bucket", func() bool {
-				return helper.BucketClient.CheckBucketNotificationSetonRGW(namespace, storeName, reverseOBCName, reverseBucketName, reverseNotificationName, helper, objectStore.Spec.IsTLSEnabled())
+				return helper.BucketClient.CheckBucketNotificationSetonRGW(namespace, storeName, reverseOBCName, reverseBucketName, reverseNotificationName, s3endpoint)
 			})
 			assert.True(t, notificationPresent)
 		})
diff --git a/tests/integration/ceph_object_test.go b/tests/integration/ceph_object_test.go
index 268dd2729..49e0f723b 100644
--- a/tests/integration/ceph_object_test.go
+++ b/tests/integration/ceph_object_test.go
@@ -153,10 +153,6 @@ func runObjectE2ETest(helper *clients.TestClient, k8sh *utils.K8sHelper, s suite
 
 	// now test operation of the first object store
 	testObjectStoreOperations(s, helper, k8sh, namespace, storeName)
-
-	bucketNotificationTestStoreName := "bucket-notification-" + storeName
-	createCephObjectStore(s.T(), helper, k8sh, namespace, bucketNotificationTestStoreName, 1, tlsEnable)
-	testBucketNotifications(s, helper, k8sh, namespace, bucketNotificationTestStoreName)
 }
 
 func testObjectStoreOperations(s suite.Suite, helper *clients.TestClient, k8sh *utils.K8sHelper, namespace, storeName string) {
@@ -164,7 +160,6 @@ func testObjectStoreOperations(s suite.Suite, helper *clients.TestClient, k8sh *
 	clusterInfo := client.AdminTestClusterInfo(namespace)
 	t := s.T()
 
-	logger.Infof("Testing Object Operations on %s", storeName)
 	t.Run("create CephObjectStoreUser", func(t *testing.T) {
 		createCephObjectUser(s, helper, k8sh, namespace, storeName, userid, true, true)
 		i := 0
@@ -363,8 +358,7 @@ func testObjectStoreOperations(s suite.Suite, helper *clients.TestClient, k8sh *
 		assert.True(t, k8sh.CheckPodCountAndState("rook-ceph-mgr", namespace, 1, "Running"))
 	})
 
-	// tests are complete, now delete the objectstore
-	s.T().Run("CephObjectStore should delete now that dependents are gone", func(t *testing.T) {
+	t.Run("CephObjectStore should delete now that dependents are gone", func(t *testing.T) {
 		// wait initially since it will almost never detect on the first try without this.
 		time.Sleep(3 * time.Second)
 
diff --git a/tests/manifests/test-kms-ibm-kp-spec.in b/tests/manifests/test-kms-ibm-kp-spec.in
index 6bbebd77d..4070971a8 100644
--- a/tests/manifests/test-kms-ibm-kp-spec.in
+++ b/tests/manifests/test-kms-ibm-kp-spec.in
@@ -2,6 +2,6 @@ spec:
   security:
     kms:
       connectionDetails:
-        KMS_PROVIDER: ibmkeyprotect
-        IBM_KP_SERVICE_INSTANCE_ID: $IBM_KP_SERVICE_INSTANCE_ID
+        KMS_PROVIDER: ibm-kp
+        IBM_KP_INSTANCE_ID: $IBM_KP_INSTANCE_ID
       tokenSecretName: rook-ibm-kp-token
diff --git a/tests/scripts/helm.sh b/tests/scripts/helm.sh
index 94acf558a..0c70a010e 100755
--- a/tests/scripts/helm.sh
+++ b/tests/scripts/helm.sh
@@ -2,7 +2,7 @@
 
 temp="/tmp/rook-tests-scripts-helm"
 
-helm_version="${HELM_VERSION:-"v3.8.0"}"
+helm_version="${HELM_VERSION:-"v3.6.2"}"
 arch="${ARCH:-}"
 
 detectArch() {