Repository: rook/rook
Pull request #15455 (2025-02-24 20:28:50)

docs: add comprehensive Rook Ceph cluster maintenance guide
- Introduced a detailed markdown document outlining procedures for safe cluster shutdown and restart as per https://github.com/rook/rook/issues/14995
- Document includes steps to scale down applications, set Ceph flags, and orderly shutdown of Ceph components (operator, RGW, CSI, OSD, MON, MGR).
- Provides command examples for both scaling down and scaling up deployments.
- Adds additional Q&A and best practices to help cluster administrators manage maintenance with minimal downtime.

<!-- Thank you for contributing to Rook! -->

<!-- STEPS TO FOLLOW:
  1. Add a description of the changes (frequently the same as the commit description)
  2. Enter the issue number next to "Resolves #" below (if there is no tracking issue resolved, **remove that section**)
  3. Review our Contributing documentation at https://rook.io/docs/rook/latest/Contributing/development-flow/
  4. Follow the steps in the checklist below, starting with the **Commit Message Formatting**.
-->

<!-- Uncomment this section with the issue number if an issue is being resolved
**Issue 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/Contributing/development-flow/#commit-structure).
- [ ] Reviewed the developer guide on [Submitting a Pull Request](https://rook.io/docs/rook/latest/Contributing/development-flow/#submitting-a-pull-request)
- [ ] [Pending release notes](https://github.com/rook/rook/blob/master/PendingReleaseNotes.md) updated with breaking and/or notable changes for the next minor release.
- [ ] Documentation has been updated, if necessary.
- [ ] Unit tests have been added, if necessary.
- [ ] Integration tests have been added, if necessary.


patrostkowski: @travisn I have removed the QA section, moved that as inline tips and added commands for scaling up remaining deployments of crashcollector and exporter. Also, Node Maintenance sounds like a better doc name 😄 

Patch:
diff --git a/Documentation/Helm-Charts/operator-chart.md b/Documentation/Helm-Charts/operator-chart.md
index 4307fc083..6b97e20c9 100644
--- a/Documentation/Helm-Charts/operator-chart.md
+++ b/Documentation/Helm-Charts/operator-chart.md
@@ -67,9 +67,7 @@ The following table lists the configurable parameters of the rook-operator chart
 | `csi.csiAddons.enabled` | Enable CSIAddons | `false` |
 | `csi.csiAddons.repository` | CSIAddons sidecar image repository | `"quay.io/csiaddons/k8s-sidecar"` |
 | `csi.csiAddons.tag` | CSIAddons sidecar image tag | `"v0.11.0"` |
-| `csi.csiAddonsCephFSProvisionerPort` | CSI Addons server port for the Ceph FS provisioner | `9070` |
 | `csi.csiAddonsPort` | CSI Addons server port | `9070` |
-| `csi.csiAddonsRBDProvisionerPort` | CSI Addons server port for the RBD provisioner | `9070` |
 | `csi.csiCephFSPluginResource` | CEPH CSI CephFS plugin resource requirement list | see values.yaml |
 | `csi.csiCephFSPluginVolume` | The volume of the CephCSI CephFS plugin DaemonSet | `nil` |
 | `csi.csiCephFSPluginVolumeMount` | The volume mounts of the CephCSI CephFS plugin DaemonSet | `nil` |
diff --git a/Documentation/Upgrade/node-maintenance.md b/Documentation/Upgrade/node-maintenance.md
new file mode 100644
index 000000000..5aa2d0646
--- /dev/null
+++ b/Documentation/Upgrade/node-maintenance.md
@@ -0,0 +1,162 @@
+---
+title: Node Maintenance
+---
+
+This guide outlines a comprehensive procedure for safely shutting down, maintaining, and restarting a nodes in the Rook Ceph cluster.
+
+Although updating nodes by failure domain (such as zones or racks) is preferred to minimize downtime, this guide describes the full shutdown process if downtime is acceptable.
+
+## Overview
+
+The basic maintenance procedure involves the following steps:
+
+1. Stop I/O: Scale down all applications to stop any I/O activity to Ceph.
+2. Set Ceph Flags: Optionally set the noout (and related) flags to prevent Ceph from marking OSDs as out.
+3. Shutdown Ceph Components: Scale down all Ceph deployments (with mons being the last) while ensuring the operator is scaled down first.
+4. Perform Maintenance: Conduct the necessary maintenance on your nodes.
+5. Restart Ceph Components: Scale up Ceph deployments (starting with mons) and then bring up the operator.
+6. Unset Ceph Flags: Remove any flags that were set before the shutdown.
+7. Restart Applications: Scale applications back up to resume operations.
+
+## Pre-Maintenance Steps
+
+Backup & Notifications: Always ensure you have a recent backup and inform stakeholders of the planned downtime.
+
+Update Strategy: Whenever possible, plan node updates by failure domain to avoid full cluster downtime.
+
+### Environment
+
+These instructions will work for as long the environment is parameterized correctly.
+Set the following environment variables, which will be used throughout this document.
+
+```console
+# Parameterize the environment
+export ROOK_OPERATOR_NAMESPACE=rook-ceph
+export ROOK_CLUSTER_NAMESPACE=rook-ceph
+```
+
+## Shutdown Procedure
+
+### **1. Scale Down Applications**
+
+Before any Ceph-specific actions, scale down all applications interacting with Ceph to ensure no new I/O is initiated.
+
+### **2. Set Ceph Flags**
+
+Setting the noout flag (and any additional flags) can help prevent Ceph from marking OSDs as out during the maintenance window.
+
+```console
+kubectl -n $ROOK_CLUSTER_NAMESPACE exec -it deploy/rook-ceph-tools -- bash
+ceph osd set noout
+# Optionally, set other flags as needed (e.g., norebalance, nodown)
+```
+
+!!! tip
+    The ordering of Ceph flags is not critical. You may set them in any order during shutdown and unset them in reverse order during startup.
+
+### **3. Scale Down the Operator**
+
+Scale down the rook-ceph-operator first to avoid it automatically restarting deployments you are scaling down.
+
+```console
+kubectl -n $ROOK_OPERATOR_NAMESPACE scale deployment rook-ceph-operator --replicas=0
+```
+
+### **4. Scale Down Ceph Deployments**
+
+Scale down the Ceph components in a controlled order. For example, you may choose the following order:
+
+```console
+RGW deployments > CSI CephFS plugin provisioners > CSI RBD plugin provisioners > OSD deployments > MON deployments > MGR deployments
+```
+
+A script to scale these down looks like:
+
+```console
+for _category in rook-ceph-rgw csi-cephfsplugin-provisioner csi-rbdplugin-provisioner rook-ceph-osd rook-ceph-mon rook-ceph-mgr rook-ceph-exporter rook-ceph-crashcollector; do
+    for _item in $(kubectl get deployment -n rook-ceph | awk '/^'"${_category}"'/{print $1}'); do
+        kubectl -n rook-ceph scale deployment ${_item} --replicas=0;
+        while [[ $(kubectl get deployment -n rook-ceph ${_item} -o jsonpath='{.status.readyReplicas}') != "" ]]; do
+            sleep 5;
+        done;
+    done;
+done
+```
+
+### **5. Node Maintenance**
+
+With all Ceph components safely scaled down, you can now perform your node maintenance tasks such as OS upgrades, hardware replacements, or other planned updates.
+
+## Startup Procedure
+
+### **1. Scale Up Ceph Components**
+
+Bring up the Ceph deployments in the following order to help ensure a healthy state before the operator takes over:
+
+Scale up MON deployments first:
+
+```console
+for _item in $(kubectl get deployment -n $ROOK_CLUSTER_NAMESPACE | awk '/^rook-ceph-mon/{print $1}'); do
+    kubectl -n $ROOK_CLUSTER_NAMESPACE scale deployment ${_item} --replicas=1;
+    while [[ $(kubectl get deployment -n $ROOK_CLUSTER_NAMESPACE ${_item} -o jsonpath='{.status.replicas}') != "1" ]]; do
+        sleep 5;
+    done;
+done
+```
+
+Then scale up OSD and MGR deployments:
+
+```console
+for _category in rook-ceph-mgr rook-ceph-osd; do
+    for _item in $(kubectl get deployment -n $ROOK_CLUSTER_NAMESPACE | awk '/^'"${_category}"'/{print $1}'); do
+        kubectl -n $ROOK_CLUSTER_NAMESPACE scale deployment ${_item} --replicas=1;
+        while [[ $(kubectl get deployment -n $ROOK_CLUSTER_NAMESPACE ${_item} -o jsonpath='{.status.replicas}') != "1" ]]; do
+            sleep 5;
+        done;
+    done;
+done
+```
+
+Finally, scale up other deployments:
+
+```console
+for _category in rook-ceph-exporter rook-ceph-crashcollector; do
+    for _item in $(kubectl get deployment -n $ROOK_CLUSTER_NAMESPACE | awk '/^'"${_category}"'/{print $1}'); do
+        kubectl -n $ROOK_CLUSTER_NAMESPACE scale deployment ${_item} --replicas=1;
+        while [[ $(kubectl get deployment -n $ROOK_CLUSTER_NAMESPACE ${_item} -o jsonpath='{.status.replicas}') != "1" ]]; do
+            sleep 5;
+        done;
+    done;
+done
+```
+
+### **2. Scale Up the Operator**
+
+Once the Ceph components are stable and healthy, scale the operator back up to allow it to reconcile any remaining changes.
+
+```console
+kubectl -n $ROOK_OPERATOR_NAMESPACE scale deployment rook-ceph-operator --replicas=1
+```
+
+!!! tip
+    Bringing the operator up after the core components ensures a smoother reconciliation process.
+
+### **3. Unset Ceph Flags**
+
+After the Ceph cluster is back online and healthy, unset any flags that were previously set.
+
+```console
+kubectl -n $ROOK_CLUSTER_NAMESPACE exec -it deploy/rook-ceph-tools -- bash
+ceph osd unset noout
+# Unset any additional flags as necessary
+```
+
+### **4. Scale Up Applications**
+
+Finally, scale your applications back to their original replica count to resume normal operations.
+
+### CSI Plugins Behavior
+
+Observation: Even when the csi-cephfsplugin-provisioner and csi-rbdplugin-provisioner deployments are scaled to 0 replicas, the corresponding CSI plugin pods (from DaemonSets) remain running.
+
+Explanation: Scaling down the csi-cephfsplugin-provisioner and csi-rbdplugin-provisioner deployments prevents new provisioning requests, but the CSI plugin pods persist because they are managed by DaemonSets. Since DaemonSets are stateless, they can be deleted if needed, and the Rook operator will automatically recreate them when restarted. However, deletion is not required to stop provisioning.
diff --git a/deploy/charts/rook-ceph-cluster/templates/ingress.yaml b/deploy/charts/rook-ceph-cluster/templates/ingress.yaml
index f915575f4..6d86c5f1d 100644
--- a/deploy/charts/rook-ceph-cluster/templates/ingress.yaml
+++ b/deploy/charts/rook-ceph-cluster/templates/ingress.yaml
@@ -10,7 +10,7 @@ metadata:
   {{- end }}
 spec:
   rules:
-    - host: {{ .Values.ingress.dashboard.host.name | quote }}
+    - host: {{ .Values.ingress.dashboard.host.name }}
       http:
         paths:
           - path: {{ .Values.ingress.dashboard.host.path | default "/" }}
diff --git a/deploy/charts/rook-ceph/templates/configmap.yaml b/deploy/charts/rook-ceph/templates/configmap.yaml
index 03e79640f..576764acc 100644
--- a/deploy/charts/rook-ceph/templates/configmap.yaml
+++ b/deploy/charts/rook-ceph/templates/configmap.yaml
@@ -208,12 +208,6 @@ data:
 {{- if .Values.csi.csiAddonsPort }}
   CSIADDONS_PORT: {{ .Values.csi.csiAddonsPort | quote }}
 {{- end }}
-{{- if .Values.csi.csiAddonsRBDProvisionerPort }}
-  CSIADDONS_RBD_PROVISIONER_PORT: {{ .Values.csi.csiAddonsRBDProvisionerPort | quote }}
-{{- end }}
-{{- if .Values.csi.csiAddonsCephFSProvisionerPort }}
-  CSIADDONS_CEPHFS_PROVISIONER_PORT: {{ .Values.csi.csiAddonsCephFSProvisionerPort | quote }}
-{{- end }}
 {{- if .Values.csi.forceCephFSKernelClient }}
   CSI_FORCE_CEPHFS_KERNEL_CLIENT: {{ .Values.csi.forceCephFSKernelClient | quote }}
 {{- end }}
diff --git a/deploy/charts/rook-ceph/values.yaml b/deploy/charts/rook-ceph/values.yaml
index 4ce764357..16efe55a2 100644
--- a/deploy/charts/rook-ceph/values.yaml
+++ b/deploy/charts/rook-ceph/values.yaml
@@ -443,12 +443,6 @@ csi:
   # -- CSI Addons server port
   # @default -- `9070`
   csiAddonsPort:
-  # -- CSI Addons server port for the RBD provisioner
-  # @default -- `9070`
-  csiAddonsRBDProvisionerPort:
-  # -- CSI Addons server port for the Ceph FS provisioner
-  # @default -- `9070`
-  csiAddonsCephFSProvisionerPort:
 
   # -- Enable Ceph Kernel clients on kernel < 4.17. If your kernel does not support quotas for CephFS
   # you may want to disable this setting. However, this will cause an issue during upgrades
diff --git a/deploy/examples/operator.yaml b/deploy/examples/operator.yaml
index b1fcd62b3..5af25f6da 100644
--- a/deploy/examples/operator.yaml
+++ b/deploy/examples/operator.yaml
@@ -470,12 +470,7 @@ data:
   # CSI_CEPHFS_LIVENESS_METRICS_PORT: "9081"
   # Configure CSI RBD liveness metrics port
   # CSI_RBD_LIVENESS_METRICS_PORT: "9080"
-
-  # We can override the ports for each individual component that uses the CSIADDONS sidecar
-  # This is useful if we're running in hostNetwork, where ports may conflict on the same host
   # CSIADDONS_PORT: "9070"
-  # CSIADDONS_RBD_PROVISIONER_PORT: "9070"
-  # CSIADDONS_CEPHFS_PROVISIONER_PORT: "9070"
 
   # Set CephFS Kernel mount options to use https://docs.ceph.com/en/latest/man/8/mount.ceph/#options
   # Set to "ms_mode=secure" when connections.encrypted is enabled in CephCluster CR
@@ -577,9 +572,6 @@ data:
 
   # RevisionHistoryLimit value for all deployments created by rook.
   # ROOK_REVISION_HISTORY_LIMIT: "3"
-
-  #  Custom label to identify node hostname. If not set `kubernetes.io/hostname` will be used
-  ROOK_CUSTOM_HOSTNAME_LABEL: ""
 ---
 # OLM: BEGIN OPERATOR DEPLOYMENT
 apiVersion: apps/v1
@@ -660,6 +652,10 @@ spec:
             - name: ROOK_UNREACHABLE_NODE_TOLERATION_SECONDS
               value: "5"
 
+            #  Custom label to identify node hostname. If not set `kubernetes.io/hostname` will be used
+            - name: ROOK_CUSTOM_HOSTNAME_LABEL
+              value: ""
+
             # The name of the node to pass with the downward API
             - name: NODE_NAME
               valueFrom:
diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go
index f499cccce..11eecc632 100644
--- a/pkg/operator/ceph/cluster/controller.go
+++ b/pkg/operator/ceph/cluster/controller.go
@@ -187,7 +187,10 @@ func add(opManagerContext context.Context, mgr manager.Manager, r reconcile.Reco
 
 	// Watch for changes on the hotplug config map
 	// TODO: to improve, can we run this against the operator namespace only?
-	disableVal := k8sutil.GetOperatorSetting(disableHotplugEnv, "false")
+	disableVal, err := k8sutil.GetOperatorSetting(opManagerContext, context.Clientset, opcontroller.OperatorSettingConfigMapName, disableHotplugEnv, "false")
+	if err != nil {
+		logger.Errorf("failed to get %s setting %s. %v", disableHotplugEnv, opcontroller.OperatorSettingConfigMapName, err)
+	}
 	if disableVal != "true" {
 		logger.Info("enabling hotplug orchestration")
 		s := source.Kind[client.Object](
diff --git a/pkg/operator/ceph/cluster/watcher.go b/pkg/operator/ceph/cluster/watcher.go
index c5ee33184..57363bfa2 100644
--- a/pkg/operator/ceph/cluster/watcher.go
+++ b/pkg/operator/ceph/cluster/watcher.go
@@ -159,21 +159,27 @@ func (c *clientCluster) onK8sNode(ctx context.Context, object runtime.Object, op
 }
 
 func (c *clientCluster) handleNodeFailure(ctx context.Context, cluster *cephv1.CephCluster, node *corev1.Node, opNamespace string) error {
-	watchForNodeLoss := k8sutil.GetOperatorSetting("ROOK_WATCH_FOR_NODE_FAILURE", "true")
+	watchForNodeLoss, err := k8sutil.GetOperatorSetting(ctx, c.context.Clientset, opcontroller.OperatorSettingConfigMapName, "ROOK_WATCH_FOR_NODE_FAILURE", "true")
+	if err != nil {
+		return pkgerror.Wrapf(err, "failed to get configmap value `ROOK_WATCH_FOR_NODE_FAILURE`.")
+	}
 
 	if strings.ToLower(watchForNodeLoss) != "true" {
 		logger.Debugf("not watching for node failures since `ROOK_WATCH_FOR_NODE_FAILURE` is set to %q", watchForNodeLoss)
 		return nil
 	}
 
-	disabledCSI := k8sutil.GetOperatorSetting("ROOK_CSI_DISABLE_DRIVER", "false")
+	disabledCSI, err := k8sutil.GetOperatorSetting(ctx, c.context.Clientset, opcontroller.OperatorSettingConfigMapName, "ROOK_CSI_DISABLE_DRIVER", "false")
+	if err != nil {
+		return pkgerror.Wrapf(err, "failed to get configmap value `ROOK_CSI_DISABLE_DRIVER`.")
+	}
 
 	if strings.ToLower(disabledCSI) != "false" {
 		logger.Debugf("not watching for node failures since `ROOK_CSI_DISABLE_DRIVER` is set to %q, skip creating networkFence", disabledCSI)
 		return nil
 	}
 
-	_, err := c.context.ApiExtensionsClient.ApiextensionsV1().CustomResourceDefinitions().Get(ctx, "networkfences.csiaddons.openshift.io", metav1.GetOptions{})
+	_, err = c.context.ApiExtensionsClient.ApiextensionsV1().CustomResourceDefinitions().Get(ctx, "networkfences.csiaddons.openshift.io", metav1.GetOptions{})
 	if err != nil {
 		if errors.IsNotFound(err) {
 			logger.Debug("networkfences.csiaddons.openshift.io CRD not found, skip creating networkFence")
diff --git a/pkg/operator/ceph/controller.go b/pkg/operator/ceph/controller.go
index 5aa86f336..879a51b2e 100644
--- a/pkg/operator/ceph/controller.go
+++ b/pkg/operator/ceph/controller.go
@@ -33,6 +33,7 @@ import (
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/rook/rook/pkg/util"
 	v1 "k8s.io/api/core/v1"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -102,43 +103,52 @@ func (r *ReconcileConfig) Reconcile(context context.Context, request reconcile.R
 
 func (r *ReconcileConfig) reconcile(request reconcile.Request) (reconcile.Result, error) {
 	// Fetch the operator's configmap
+	opConfig := &v1.ConfigMap{}
 	logger.Debugf("reconciling %s", request.NamespacedName)
-	if request.Name == opcontroller.OperatorSettingConfigMapName {
-		if err := k8sutil.ApplyOperatorSettingsConfigmap(r.opManagerContext, r.context.Clientset); err != nil {
-			return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to get operator setting configmap")
+	err := r.client.Get(r.opManagerContext, request.NamespacedName, opConfig)
+	if err != nil {
+		if kerrors.IsNotFound(err) {
+			logger.Debug("operator's configmap resource not found. will use default value or env var.")
+		} else {
+			// Error reading the object - requeue the request.
+			return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to get operator's configmap")
 		}
+	} else {
+		// Populate the operator's config
+		r.config.Parameters = opConfig.Data
 	}
+
 	// Reconcile Ceph CLI timeout, since the clusterd context is passed to by pointer to all CRD
 	// controllers they will receive the update
-	opcontroller.SetCephCommandsTimeout()
+	opcontroller.SetCephCommandsTimeout(r.config.Parameters)
 
 	// Reconcile Operator's logging level
-	reconcileOperatorLogLevel()
+	reconcileOperatorLogLevel(opConfig.Data)
 
 	// Reconcile discovery daemon
-	err := r.reconcileDiscoveryDaemon()
+	err = r.reconcileDiscoveryDaemon(opConfig.Data)
 	if err != nil {
 		return opcontroller.ImmediateRetryResult, err
 	}
 
-	opcontroller.SetAllowLoopDevices()
-	opcontroller.SetEnforceHostNetwork()
-	opcontroller.SetRevisionHistoryLimit()
-	opcontroller.SetObcAllowAdditionalConfigFields()
+	opcontroller.SetAllowLoopDevices(r.config.Parameters)
+	opcontroller.SetEnforceHostNetwork(r.config.Parameters)
+	opcontroller.SetRevisionHistoryLimit(r.config.Parameters)
+	opcontroller.SetObcAllowAdditionalConfigFields(r.config.Parameters)
 
 	logger.Infof("%s done reconciling", controllerName)
 	return reconcile.Result{}, nil
 }
 
-func reconcileOperatorLogLevel() {
-	rookLogLevel := k8sutil.GetOperatorSetting("ROOK_LOG_LEVEL", util.DefaultLogLevel.String())
+func reconcileOperatorLogLevel(data map[string]string) {
+	rookLogLevel := k8sutil.GetValue(data, "ROOK_LOG_LEVEL", util.DefaultLogLevel.String())
 	util.SetGlobalLogLevel(rookLogLevel, logger)
 }
 
-func (r *ReconcileConfig) reconcileDiscoveryDaemon() error {
+func (r *ReconcileConfig) reconcileDiscoveryDaemon(data map[string]string) error {
 	rookDiscover := discover.New(r.context.Clientset)
-	if opcontroller.DiscoveryDaemonEnabled() {
-		if err := rookDiscover.Start(r.opManagerContext, r.config.OperatorNamespace, r.config.Image, r.config.ServiceAccount, true); err != nil {
+	if opcontroller.DiscoveryDaemonEnabled(r.config.Parameters) {
+		if err := rookDiscover.Start(r.opManagerContext, r.config.OperatorNamespace, r.config.Image, r.config.ServiceAccount, data, true); err != nil {
 			return errors.Wrap(err, "failed to start device discovery daemonset")
 		}
 	} else {
diff --git a/pkg/operator/ceph/controller/controller_utils.go b/pkg/operator/ceph/controller/controller_utils.go
index d55d00124..8cb661d74 100644
--- a/pkg/operator/ceph/controller/controller_utils.go
+++ b/pkg/operator/ceph/controller/controller_utils.go
@@ -40,6 +40,7 @@ type OperatorConfig struct {
 	Image             string
 	ServiceAccount    string
 	NamespaceToWatch  string
+	Parameters        map[string]string
 }
 
 // ClusterHealth is passed to the various monitoring go routines to stop them when the context is cancelled
@@ -98,13 +99,13 @@ var (
 	obcAllowAdditionalConfigFields = strings.Split(obcAllowAdditionalConfigFieldsDefaultValue, ",")
 )
 
-func DiscoveryDaemonEnabled() bool {
-	return k8sutil.GetOperatorSetting("ROOK_ENABLE_DISCOVERY_DAEMON", "false") == "true"
+func DiscoveryDaemonEnabled(data map[string]string) bool {
+	return k8sutil.GetValue(data, "ROOK_ENABLE_DISCOVERY_DAEMON", "false") == "true"
 }
 
 // SetCephCommandsTimeout sets the timeout value of Ceph commands which are executed from Rook
-func SetCephCommandsTimeout() {
-	strTimeoutSeconds := k8sutil.GetOperatorSetting("ROOK_CEPH_COMMANDS_TIMEOUT_SECONDS", "15")
+func SetCephCommandsTimeout(data map[string]string) {
+	strTimeoutSeconds := k8sutil.GetValue(data, "ROOK_CEPH_COMMANDS_TIMEOUT_SECONDS", "15")
 	timeoutSeconds, err := strconv.Atoi(strTimeoutSeconds)
 	if err != nil || timeoutSeconds < 1 {
 		logger.Warningf("ROOK_CEPH_COMMANDS_TIMEOUT is %q but it should be >= 1, set the default value 15", strTimeoutSeconds)
@@ -113,8 +114,8 @@ func SetCephCommandsTimeout() {
 	exec.CephCommandsTimeout = time.Duration(timeoutSeconds) * time.Second
 }
 
-func SetAllowLoopDevices() {
-	strLoopDevicesAllowed := k8sutil.GetOperatorSetting("ROOK_CEPH_ALLOW_LOOP_DEVICES", "false")
+func SetAllowLoopDevices(data map[string]string) {
+	strLoopDevicesAllowed := k8sutil.GetValue(data, "ROOK_CEPH_ALLOW_LOOP_DEVICES", "false")
 	var err error
 	loopDevicesAllowed, err = strconv.ParseBool(strLoopDevicesAllowed)
 	if err != nil {
@@ -127,8 +128,8 @@ func LoopDevicesAllowed() bool {
 	return loopDevicesAllowed
 }
 
-func SetEnforceHostNetwork() {
-	strval := k8sutil.GetOperatorSetting(enforceHostNetworkSettingName, enforceHostNetworkDefaultValue)
+func SetEnforceHostNetwork(data map[string]string) {
+	strval := k8sutil.GetValue(data, enforceHostNetworkSettingName, enforceHostNetworkDefaultValue)
 	val, err := strconv.ParseBool(strval)
 	if err != nil {
 		logger.Warningf("failed to parse value %q for %q. assuming false value", strval, enforceHostNetworkSettingName)
@@ -142,8 +143,8 @@ func EnforceHostNetwork() bool {
 	return cephv1.EnforceHostNetwork()
 }
 
-func SetRevisionHistoryLimit() {
-	strval := k8sutil.GetOperatorSetting(revisionHistoryLimitSettingName, "")
+func SetRevisionHistoryLimit(data map[string]string) {
+	strval := k8sutil.GetValue(data, revisionHistoryLimitSettingName, "")
 	var limit int32
 	if strval == "" {
 		logger.Debugf("not parsing empty string to int for %q. assuming default value.", revisionHistoryLimitSettingName)
@@ -165,8 +166,8 @@ func RevisionHistoryLimit() *int32 {
 	return revisionHistoryLimit
 }
 
-func SetObcAllowAdditionalConfigFields() {
-	strval := k8sutil.GetOperatorSetting(obcAllowAdditionalConfigFieldsSettingName, obcAllowAdditionalConfigFieldsDefaultValue)
+func SetObcAllowAdditionalConfigFields(data map[string]string) {
+	strval := k8sutil.GetValue(data, obcAllowAdditionalConfigFieldsSettingName, obcAllowAdditionalConfigFieldsDefaultValue)
 	obcAllowAdditionalConfigFields = strings.Split(strval, ",")
 }
 
diff --git a/pkg/operator/ceph/cr_manager.go b/pkg/operator/ceph/cr_manager.go
index 21d9e4860..b3398d681 100644
--- a/pkg/operator/ceph/cr_manager.go
+++ b/pkg/operator/ceph/cr_manager.go
@@ -143,7 +143,11 @@ func (o *Operator) startCRDManager(context context.Context, mgrErrorCh chan erro
 		}
 	}
 
-	metricsBindAddress := k8sutil.GetOperatorSetting("ROOK_OPERATOR_METRICS_BIND_ADDRESS", "0")
+	metricsBindAddress, err := k8sutil.GetOperatorSetting(context, o.context.Clientset, opcontroller.OperatorSettingConfigMapName, "ROOK_OPERATOR_METRICS_BIND_ADDRESS", "0")
+	if err != nil {
+		mgrErrorCh <- errors.Wrap(err, "failed to get configmap value `ROOK_OPERATOR_METRICS_BIND_ADDRESS`.")
+		return
+	}
 	skipNameValidation := true
 	// Set up a manager
 	mgrOpts := manager.Options{
diff --git a/pkg/operator/ceph/csi/controller.go b/pkg/operator/ceph/csi/controller.go
index 809b2da4d..b9c6c5034 100644
--- a/pkg/operator/ceph/csi/controller.go
+++ b/pkg/operator/ceph/csi/controller.go
@@ -42,6 +42,7 @@ import (
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 )
 
 const (
@@ -157,17 +158,32 @@ func (r *ReconcileCSI) reconcile(request reconcile.Request) (reconcile.Result, e
 		return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed creating csi config map")
 	}
 
-	if err := k8sutil.ApplyOperatorSettingsConfigmap(r.opManagerContext, r.context.Clientset); err != nil {
-		return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to apply operator settings configmap")
+	// Fetch the operator's configmap. We force the NamespaceName to the operator since the request
+	// could be a CephCluster. If so the NamespaceName will be the one from the cluster and thus the
+	// CM won't be found
+	opNamespaceName := types.NamespacedName{Name: opcontroller.OperatorSettingConfigMapName, Namespace: r.opConfig.OperatorNamespace}
+	opConfig := &v1.ConfigMap{}
+	err = r.client.Get(r.opManagerContext, opNamespaceName, opConfig)
+	if err != nil {
+		if kerrors.IsNotFound(err) {
+			logger.Debug("operator's configmap resource not found. will use default value or env var.")
+			r.opConfig.Parameters = make(map[string]string)
+		} else {
+			// Error reading the object - requeue the request.
+			return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to get operator's configmap")
+		}
+	} else {
+		// Populate the operator's config
+		r.opConfig.Parameters = opConfig.Data
 	}
 
-	enableCSIOperator, err = strconv.ParseBool(k8sutil.GetOperatorSetting("ROOK_USE_CSI_OPERATOR", "false"))
+	enableCSIOperator, err = strconv.ParseBool(k8sutil.GetValue(r.opConfig.Parameters, "ROOK_USE_CSI_OPERATOR", "false"))
 	if err != nil {
 		return reconcileResult, errors.Wrap(err, "unable to parse value for 'ROOK_USE_CSI_OPERATOR'")
 	}
 
 	// do not recocnile if csi driver is disabled
-	disableCSI, err := strconv.ParseBool(k8sutil.GetOperatorSetting("ROOK_CSI_DISABLE_DRIVER", "false"))
+	disableCSI, err := strconv.ParseBool(k8sutil.GetValue(r.opConfig.Parameters, "ROOK_CSI_DISABLE_DRIVER", "false"))
 	if err != nil {
 		return reconcile.Result{}, errors.Wrap(err, "unable to parse value for 'ROOK_CSI_DISABLE_DRIVER")
 	} else if disableCSI {
diff --git a/pkg/operator/ceph/csi/csi.go b/pkg/operator/ceph/csi/csi.go
index 6919eaf54..9769a519f 100644
--- a/pkg/operator/ceph/csi/csi.go
+++ b/pkg/operator/ceph/csi/csi.go
@@ -56,31 +56,31 @@ func (r *ReconcileCSI) validateAndConfigureDrivers(ownerInfo *k8sutil.OwnerInfo)
 func (r *ReconcileCSI) setParams() error {
 	var err error
 
-	if EnableRBD, err = strconv.ParseBool(k8sutil.GetOperatorSetting("ROOK_CSI_ENABLE_RBD", "true")); err != nil {
+	if EnableRBD, err = strconv.ParseBool(k8sutil.GetValue(r.opConfig.Parameters, "ROOK_CSI_ENABLE_RBD", "true")); err != nil {
 		return errors.Wrap(err, "unable to parse value for 'ROOK_CSI_ENABLE_RBD'")
 	}
 
-	if EnableCephFS, err = strconv.ParseBool(k8sutil.GetOperatorSetting("ROOK_CSI_ENABLE_CEPHFS", "true")); err != nil {
+	if EnableCephFS, err = strconv.ParseBool(k8sutil.GetValue(r.opConfig.Parameters, "ROOK_CSI_ENABLE_CEPHFS", "true")); err != nil {
 		return errors.Wrap(err, "unable to parse value for 'ROOK_CSI_ENABLE_CEPHFS'")
 	}
 
-	if EnableNFS, err = strconv.ParseBool(k8sutil.GetOperatorSetting("ROOK_CSI_ENABLE_NFS", "false")); err != nil {
+	if EnableNFS, err = strconv.ParseBool(k8sutil.GetValue(r.opConfig.Parameters, "ROOK_CSI_ENABLE_NFS", "false")); err != nil {
 		return errors.Wrap(err, "unable to parse value for 'ROOK_CSI_ENABLE_NFS'")
 	}
 
-	if CSIParam.EnableCSIHostNetwork, err = strconv.ParseBool(k8sutil.GetOperatorSetting("CSI_ENABLE_HOST_NETWORK", "true")); err != nil {
+	if CSIParam.EnableCSIHostNetwork, err = strconv.ParseBool(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_HOST_NETWORK", "true")); err != nil {
 		return errors.Wrap(err, "failed to parse value for 'CSI_ENABLE_HOST_NETWORK'")
 	}
 
 	// If not set or set to anything but "false", the kernel client will be enabled
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_FORCE_CEPHFS_KERNEL_CLIENT", "true"), "false") {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_FORCE_CEPHFS_KERNEL_CLIENT", "true"), "false") {
 		CSIParam.ForceCephFSKernelClient = "false"
 	} else {
 		CSIParam.ForceCephFSKernelClient = "true"
 	}
 
 	// parse RPC timeout
-	timeout := k8sutil.GetOperatorSetting(grpcTimeout, "150")
+	timeout := k8sutil.GetValue(r.opConfig.Parameters, grpcTimeout, "150")
 	timeoutSeconds, err := strconv.Atoi(timeout)
 	if err != nil {
 		logger.Errorf("failed to parse %q. Defaulting to %d. %v", grpcTimeout, defaultGRPCTimeout, err)
@@ -93,32 +93,20 @@ func (r *ReconcileCSI) setParams() error {
 	CSIParam.GRPCTimeout = time.Duration(timeoutSeconds) * time.Second
 
 	// parse Liveness port
-	CSIParam.CephFSLivenessMetricsPort, err = getPortFromConfig("CSI_CEPHFS_LIVENESS_METRICS_PORT", DefaultCephFSLivenessMerticsPort)
+	CSIParam.CephFSLivenessMetricsPort, err = getPortFromConfig(r.opConfig.Parameters, "CSI_CEPHFS_LIVENESS_METRICS_PORT", DefaultCephFSLivenessMerticsPort)
 	if err != nil {
 		return errors.Wrap(err, "error getting CSI CephFS liveness metrics port.")
 	}
-
-	CSIParam.CSIAddonsPort, err = getPortFromConfig("CSIADDONS_PORT", DefaultCSIAddonsPort)
+	CSIParam.CSIAddonsPort, err = getPortFromConfig(r.opConfig.Parameters, "CSIADDONS_PORT", DefaultCSIAddonsPort)
 	if err != nil {
 		return errors.Wrap(err, "failed to get CSI Addons port")
 	}
-
-	CSIParam.CSIAddonsRBDProvisionerPort, err = getPortFromConfig("CSIADDONS_RBD_PROVISIONER_PORT", DefaultCSIAddonsRBDProvisionerPort)
-	if err != nil {
-		return errors.Wrap(err, "failed to get CSI Addons port for RBD provisioner")
-	}
-
-	CSIParam.CSIAddonsCephFSProvisionerPort, err = getPortFromConfig("CSIADDONS_CEPHFS_PROVISIONER_PORT", DefaultCSIAddonsCephFSProvisionerPort)
-	if err != nil {
-		return errors.Wrap(err, "failed to get CSI Addons port for Ceph FS provisioner")
-	}
-
-	CSIParam.RBDLivenessMetricsPort, err = getPortFromConfig("CSI_RBD_LIVENESS_METRICS_PORT", DefaultRBDLivenessMerticsPort)
+	CSIParam.RBDLivenessMetricsPort, err = getPortFromConfig(r.opConfig.Parameters, "CSI_RBD_LIVENESS_METRICS_PORT", DefaultRBDLivenessMerticsPort)
 	if err != nil {
 		return errors.Wrap(err, "error getting CSI RBD liveness metrics port.")
 	}
 
-	CSIParam.EnableLiveness, err = strconv.ParseBool(k8sutil.GetOperatorSetting("CSI_ENABLE_LIVENESS", "false"))
+	CSIParam.EnableLiveness, err = strconv.ParseBool(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_LIVENESS", "false"))
 	if err != nil {
 		return errors.Wrap(err, "failed to parse value for 'CSI_ENABLE_LIVENESS'")
 	}
@@ -126,31 +114,31 @@ func (r *ReconcileCSI) setParams() error {
 	CSIParam.Privileged = controller.HostPathRequiresPrivileged()
 
 	// default value `system-node-critical` is the highest available priority
-	CSIParam.PluginPriorityClassName = k8sutil.GetOperatorSetting("CSI_PLUGIN_PRIORITY_CLASSNAME", "")
+	CSIParam.PluginPriorityClassName = k8sutil.GetValue(r.opConfig.Parameters, "CSI_PLUGIN_PRIORITY_CLASSNAME", "")
 
 	// default value `system-cluster-critical` is applied for some
 	// critical pods in cluster but less priority than plugin pods
-	CSIParam.ProvisionerPriorityClassName = k8sutil.GetOperatorSetting("CSI_PROVISIONER_PRIORITY_CLASSNAME", "")
+	CSIParam.ProvisionerPriorityClassName = k8sutil.GetValue(r.opConfig.Parameters, "CSI_PROVISIONER_PRIORITY_CLASSNAME", "")
 
 	CSIParam.EnableOMAPGenerator = false
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_ENABLE_OMAP_GENERATOR", "false"), "true") {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_OMAP_GENERATOR", "false"), "true") {
 		CSIParam.EnableOMAPGenerator = true
 	}
 
 	CSIParam.EnableCSIDriverSeLinuxMount = true
 
 	CSIParam.EnableRBDSnapshotter = true
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_ENABLE_RBD_SNAPSHOTTER", "true"), "false") {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_RBD_SNAPSHOTTER", "true"), "false") {
 		CSIParam.EnableRBDSnapshotter = false
 	}
 
 	CSIParam.EnableCephFSSnapshotter = true
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_ENABLE_CEPHFS_SNAPSHOTTER", "true"), "false") {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_CEPHFS_SNAPSHOTTER", "true"), "false") {
 		CSIParam.EnableCephFSSnapshotter = false
 	}
 
 	CSIParam.EnableNFSSnapshotter = true
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_ENABLE_NFS_SNAPSHOTTER", "true"), "false") {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_NFS_SNAPSHOTTER", "true"), "false") {
 		CSIParam.EnableNFSSnapshotter = false
 	}
 
@@ -159,36 +147,36 @@ func (r *ReconcileCSI) setParams() error {
 	if err == nil {
 		CSIParam.EnableCSIAddonsSideCar = true
 	}
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_ENABLE_CSIADDONS", ""), "false") {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_CSIADDONS", ""), "false") {
 		CSIParam.EnableCSIAddonsSideCar = false
 	}
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_ENABLE_CSIADDONS", ""), "true") {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_CSIADDONS", ""), "true") {
 		CSIParam.EnableCSIAddonsSideCar = true
 	}
 
 	CSIParam.EnableCSITopology = false
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_ENABLE_TOPOLOGY", "false"), "true") {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_TOPOLOGY", "false"), "true") {
 		CSIParam.EnableCSITopology = true
 	}
 
 	CSIParam.EnableCSIEncryption = false
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_ENABLE_ENCRYPTION", "false"), "true") {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_ENCRYPTION", "false"), "true") {
 		CSIParam.EnableCSIEncryption = true
 	}
 
 	CSIParam.CSIEnableMetadata = false
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_ENABLE_METADATA", "false"), "true") {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_METADATA", "false"), "true") {
 		CSIParam.CSIEnableMetadata = true
 	}
 
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_CEPHFS_PLUGIN_UPDATE_STRATEGY", rollingUpdate), onDelete) {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_CEPHFS_PLUGIN_UPDATE_STRATEGY", rollingUpdate), onDelete) {
 		CSIParam.CephFSPluginUpdateStrategy = onDelete
 	} else {
 		CSIParam.CephFSPluginUpdateStrategy = rollingUpdate
-		CSIParam.CephFSPluginUpdateStrategyMaxUnavailable = k8sutil.GetOperatorSetting("CSI_CEPHFS_PLUGIN_UPDATE_STRATEGY_MAX_UNAVAILABLE", "1")
+		CSIParam.CephFSPluginUpdateStrategyMaxUnavailable = k8sutil.GetValue(r.opConfig.Parameters, "CSI_CEPHFS_PLUGIN_UPDATE_STRATEGY_MAX_UNAVAILABLE", "1")
 	}
 
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_NFS_PLUGIN_UPDATE_STRATEGY", rollingUpdate), onDelete) {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_NFS_PLUGIN_UPDATE_STRATEGY", rollingUpdate), onDelete) {
 		CSIParam.NFSPluginUpdateStrategy = onDelete
 	} else {
 		CSIParam.NFSPluginUpdateStrategy = rollingUpdate
@@ -196,19 +184,19 @@ func (r *ReconcileCSI) setParams() error {
 
 	// Default values are based on Kubernetes official documentation.
 	// https://kubernetes.io/docs/tasks/manage-daemon/update-daemon-set/#daemonset-update-strategy
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_RBD_PLUGIN_UPDATE_STRATEGY", rollingUpdate), onDelete) {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_RBD_PLUGIN_UPDATE_STRATEGY", rollingUpdate), onDelete) {
 		CSIParam.RBDPluginUpdateStrategy = onDelete
 	} else {
 		CSIParam.RBDPluginUpdateStrategy = rollingUpdate
-		CSIParam.RBDPluginUpdateStrategyMaxUnavailable = k8sutil.GetOperatorSetting("CSI_RBD_PLUGIN_UPDATE_STRATEGY_MAX_UNAVAILABLE", "1")
+		CSIParam.RBDPluginUpdateStrategyMaxUnavailable = k8sutil.GetValue(r.opConfig.Parameters, "CSI_RBD_PLUGIN_UPDATE_STRATEGY_MAX_UNAVAILABLE", "1")
 	}
 
 	CSIParam.EnablePluginSelinuxHostMount = false
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_PLUGIN_ENABLE_SELINUX_HOST_MOUNT", "false"), "true") {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_PLUGIN_ENABLE_SELINUX_HOST_MOUNT", "false"), "true") {
 		CSIParam.EnablePluginSelinuxHostMount = true
 	}
 
-	logLevel := k8sutil.GetOperatorSetting("CSI_LOG_LEVEL", "")
+	logLevel := k8sutil.GetValue(r.opConfig.Parameters, "CSI_LOG_LEVEL", "")
 	CSIParam.LogLevel = defaultLogLevel
 	if logLevel != "" {
 		l, err := strconv.ParseUint(logLevel, 10, 8)
@@ -219,7 +207,7 @@ func (r *ReconcileCSI) setParams() error {
 		}
 	}
 
-	sidecarLogLevel := k8sutil.GetOperatorSetting("CSI_SIDECAR_LOG_LEVEL", "")
+	sidecarLogLevel := k8sutil.GetValue(r.opConfig.Parameters, "CSI_SIDECAR_LOG_LEVEL", "")
 	CSIParam.SidecarLogLevel = defaultSidecarLogLevel
 	if sidecarLogLevel != "" {
 		l, err := strconv.ParseUint(sidecarLogLevel, 10, 8)
@@ -230,7 +218,7 @@ func (r *ReconcileCSI) setParams() error {
 		}
 	}
 
-	leaderElectionLeaseDuration := k8sutil.GetOperatorSetting("CSI_LEADER_ELECTION_LEASE_DURATION", "")
+	leaderElectionLeaseDuration := k8sutil.GetValue(r.opConfig.Parameters, "CSI_LEADER_ELECTION_LEASE_DURATION", "")
 	CSIParam.LeaderElectionLeaseDuration = defaultLeaderElectionLeaseDuration
 	if leaderElectionLeaseDuration != "" {
 		d, err := time.ParseDuration(leaderElectionLeaseDuration)
@@ -241,7 +229,7 @@ func (r *ReconcileCSI) setParams() error {
 		}
 	}
 
-	leaderElectionRenewDeadline := k8sutil.GetOperatorSetting("CSI_LEADER_ELECTION_RENEW_DEADLINE", "")
+	leaderElectionRenewDeadline := k8sutil.GetValue(r.opConfig.Parameters, "CSI_LEADER_ELECTION_RENEW_DEADLINE", "")
 	CSIParam.LeaderElectionRenewDeadline = defaultLeaderElectionRenewDeadline
 	if leaderElectionRenewDeadline != "" {
 		d, err := time.ParseDuration(leaderElectionRenewDeadline)
@@ -252,7 +240,7 @@ func (r *ReconcileCSI) setParams() error {
 		}
 	}
 
-	leaderElectionRetryPeriod := k8sutil.GetOperatorSetting("CSI_LEADER_ELECTION_RETRY_PERIOD", "")
+	leaderElectionRetryPeriod := k8sutil.GetValue(r.opConfig.Parameters, "CSI_LEADER_ELECTION_RETRY_PERIOD", "")
 	CSIParam.LeaderElectionRetryPeriod = defaultLeaderElectionRetryPeriod
 	if leaderElectionRetryPeriod != "" {
 		d, err := time.ParseDuration(leaderElectionRetryPeriod)
@@ -269,7 +257,7 @@ func (r *ReconcileCSI) setParams() error {
 		if len(nodes.Items) == 1 {
 			CSIParam.ProvisionerReplicas = 1
 		} else {
-			replicaStr := k8sutil.GetOperatorSetting("CSI_PROVISIONER_REPLICAS", "2")
+			replicaStr := k8sutil.GetValue(r.opConfig.Parameters, "CSI_PROVISIONER_REPLICAS", "2")
 			replicas, err := strconv.ParseInt(replicaStr, 10, 32)
 			if err != nil {
 				logger.Errorf("failed to parse CSI_PROVISIONER_REPLICAS. Defaulting to %d. %v", defaultProvisionerReplicas, err)
@@ -281,39 +269,39 @@ func (r *ReconcileCSI) setParams() error {
 		logger.Errorf("failed to get nodes. Defaulting the number of replicas of provisioner pods to %d. %v", CSIParam.ProvisionerReplicas, err)
 	}
 
-	CSIParam.CSIPluginImage = getImage("ROOK_CSI_CEPH_IMAGE", DefaultCSIPluginImage)
-	CSIParam.RegistrarImage = getImage("ROOK_CSI_REGISTRAR_IMAGE", DefaultRegistrarImage)
-	CSIParam.ProvisionerImage = getImage("ROOK_CSI_PROVISIONER_IMAGE", DefaultProvisionerImage)
-	CSIParam.AttacherImage = getImage("ROOK_CSI_ATTACHER_IMAGE", DefaultAttacherImage)
-	CSIParam.SnapshotterImage = getImage("ROOK_CSI_SNAPSHOTTER_IMAGE", DefaultSnapshotterImage)
-	CSIParam.ResizerImage = getImage("ROOK_CSI_RESIZER_IMAGE", DefaultResizerImage)
-	CSIParam.KubeletDirPath = k8sutil.GetOperatorSetting("ROOK_CSI_KUBELET_DIR_PATH", DefaultKubeletDirPath)
-	CSIParam.CSIAddonsImage = getImage("ROOK_CSIADDONS_IMAGE", DefaultCSIAddonsImage)
-	CSIParam.CSIDomainLabels = k8sutil.GetOperatorSetting("CSI_TOPOLOGY_DOMAIN_LABELS", "")
-	csiCephFSPodLabels := k8sutil.GetOperatorSetting("ROOK_CSI_CEPHFS_POD_LABELS", "")
+	CSIParam.CSIPluginImage = getImage(r.opConfig.Parameters, "ROOK_CSI_CEPH_IMAGE", DefaultCSIPluginImage)
+	CSIParam.RegistrarImage = getImage(r.opConfig.Parameters, "ROOK_CSI_REGISTRAR_IMAGE", DefaultRegistrarImage)
+	CSIParam.ProvisionerImage = getImage(r.opConfig.Parameters, "ROOK_CSI_PROVISIONER_IMAGE", DefaultProvisionerImage)
+	CSIParam.AttacherImage = getImage(r.opConfig.Parameters, "ROOK_CSI_ATTACHER_IMAGE", DefaultAttacherImage)
+	CSIParam.SnapshotterImage = getImage(r.opConfig.Parameters, "ROOK_CSI_SNAPSHOTTER_IMAGE", DefaultSnapshotterImage)
+	CSIParam.ResizerImage = getImage(r.opConfig.Parameters, "ROOK_CSI_RESIZER_IMAGE", DefaultResizerImage)
+	CSIParam.KubeletDirPath = k8sutil.GetValue(r.opConfig.Parameters, "ROOK_CSI_KUBELET_DIR_PATH", DefaultKubeletDirPath)
+	CSIParam.CSIAddonsImage = getImage(r.opConfig.Parameters, "ROOK_CSIADDONS_IMAGE", DefaultCSIAddonsImage)
+	CSIParam.CSIDomainLabels = k8sutil.GetValue(r.opConfig.Parameters, "CSI_TOPOLOGY_DOMAIN_LABELS", "")
+	csiCephFSPodLabels := k8sutil.GetValue(r.opConfig.Parameters, "ROOK_CSI_CEPHFS_POD_LABELS", "")
 	CSIParam.CSICephFSPodLabels = k8sutil.ParseStringToLabels(csiCephFSPodLabels)
-	csiNFSPodLabels := k8sutil.GetOperatorSetting("ROOK_CSI_NFS_POD_LABELS", "")
+	csiNFSPodLabels := k8sutil.GetValue(r.opConfig.Parameters, "ROOK_CSI_NFS_POD_LABELS", "")
 	CSIParam.CSINFSPodLabels = k8sutil.ParseStringToLabels(csiNFSPodLabels)
-	csiRBDPodLabels := k8sutil.GetOperatorSetting("ROOK_CSI_RBD_POD_LABELS", "")
+	csiRBDPodLabels := k8sutil.GetValue(r.opConfig.Parameters, "ROOK_CSI_RBD_POD_LABELS", "")
 	CSIParam.CSIRBDPodLabels = k8sutil.ParseStringToLabels(csiRBDPodLabels)
-	CSIParam.CSIClusterName = k8sutil.GetOperatorSetting("CSI_CLUSTER_NAME", "")
-	CSIParam.ImagePullPolicy = k8sutil.GetOperatorSetting("ROOK_CSI_IMAGE_PULL_POLICY", DefaultCSIImagePullPolicy)
-	CSIParam.CephFSKernelMountOptions = k8sutil.GetOperatorSetting("CSI_CEPHFS_KERNEL_MOUNT_OPTIONS", "")
+	CSIParam.CSIClusterName = k8sutil.GetValue(r.opConfig.Parameters, "CSI_CLUSTER_NAME", "")
+	CSIParam.ImagePullPolicy = k8sutil.GetValue(r.opConfig.Parameters, "ROOK_CSI_IMAGE_PULL_POLICY", DefaultCSIImagePullPolicy)
+	CSIParam.CephFSKernelMountOptions = k8sutil.GetValue(r.opConfig.Parameters, "CSI_CEPHFS_KERNEL_MOUNT_OPTIONS", "")
 
 	CSIParam.CephFSAttachRequired = true
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_CEPHFS_ATTACH_REQUIRED", "true"), "false") {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_CEPHFS_ATTACH_REQUIRED", "true"), "false") {
 		CSIParam.CephFSAttachRequired = false
 	}
 	CSIParam.RBDAttachRequired = true
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_RBD_ATTACH_REQUIRED", "true"), "false") {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_RBD_ATTACH_REQUIRED", "true"), "false") {
 		CSIParam.RBDAttachRequired = false
 	}
 	CSIParam.NFSAttachRequired = true
-	if strings.EqualFold(k8sutil.GetOperatorSetting("CSI_NFS_ATTACH_REQUIRED", "true"), "false") {
+	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_NFS_ATTACH_REQUIRED", "true"), "false") {
 		CSIParam.NFSAttachRequired = false
 	}
 
-	CSIParam.DriverNamePrefix = k8sutil.GetOperatorSetting("CSI_DRIVER_NAME_PREFIX", r.opConfig.OperatorNamespace)
+	CSIParam.DriverNamePrefix = k8sutil.GetValue(r.opConfig.Parameters, "CSI_DRIVER_NAME_PREFIX", r.opConfig.OperatorNamespace)
 
 	crd, err := r.context.ApiExtensionsClient.ApiextensionsV1().CustomResourceDefinitions().Get(context.TODO(), "volumegroupsnapshotclasses.groupsnapshot.storage.k8s.io", metav1.GetOptions{})
 	if err != nil && !kerrors.IsNotFound(err) {
@@ -324,7 +312,7 @@ func (r *ReconcileCSI) setParams() error {
 	if err == nil && len(crd.Spec.Versions) > 0 {
 		ver := crd.Spec.Versions[0]
 		// Determine if VolumeGroupSnapshot feature should be disabled
-		disableVGS := strings.EqualFold(k8sutil.GetOperatorSetting("CSI_ENABLE_VOLUME_GROUP_SNAPSHOT", "true"), "false")
+		disableVGS := strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_VOLUME_GROUP_SNAPSHOT", "true"), "false")
 		const (
 			enableVolumeGroupSnapshotFlag = "--enable-volume-group-snapshots="
 			featureGateFlag               = "--feature-gates=CSIVolumeGroupSnapshot="
@@ -346,7 +334,7 @@ func (r *ReconcileCSI) setParams() error {
 		}
 	}
 
-	kubeApiBurst := k8sutil.GetOperatorSetting("CSI_KUBE_API_BURST", "")
+	kubeApiBurst := k8sutil.GetValue(r.opConfig.Parameters, "CSI_KUBE_API_BURST", "")
 	CSIParam.KubeApiBurst = 0
 	if kubeApiBurst != "" {
 		k, err := strconv.ParseUint(kubeApiBurst, 10, 16)
@@ -357,7 +345,7 @@ func (r *ReconcileCSI) setParams() error {
 		}
 	}
 
-	kubeApiQPS := k8sutil.GetOperatorSetting("CSI_KUBE_API_QPS", "")
+	kubeApiQPS := k8sutil.GetValue(r.opConfig.Parameters, "CSI_KUBE_API_QPS", "")
 	CSIParam.KubeApiQPS = 0
 	if kubeApiQPS != "" {
 		k, err := strconv.ParseFloat(kubeApiQPS, 32)
diff --git a/pkg/operator/ceph/csi/operator_config.go b/pkg/operator/ceph/csi/operator_config.go
index 7e013245c..638e512b2 100644
--- a/pkg/operator/ceph/csi/operator_config.go
+++ b/pkg/operator/ceph/csi/operator_config.go
@@ -94,9 +94,9 @@ func (r *ReconcileCSI) generateCSIOpConfigSpec(cluster cephv1.CephCluster, opCon
 				PodCommonSpec: csiopv1a1.PodCommonSpec{
 					PrioritylClassName: &CSIParam.ProvisionerPriorityClassName,
 					Affinity: &v1.Affinity{
-						NodeAffinity: getNodeAffinity(pluginNodeAffinityEnv, &v1.NodeAffinity{}),
+						NodeAffinity: getNodeAffinity(r.opConfig.Parameters, pluginNodeAffinityEnv, &v1.NodeAffinity{}),
 					},
-					Tolerations: getToleration(pluginTolerationsEnv, []v1.Toleration{}),
+					Tolerations: getToleration(r.opConfig.Parameters, pluginTolerationsEnv, []v1.Toleration{}),
 				},
 				Resources:              csiopv1a1.NodePluginResourcesSpec{},
 				KubeletDirPath:         CSIParam.KubeletDirPath,
@@ -106,9 +106,9 @@ func (r *ReconcileCSI) generateCSIOpConfigSpec(cluster cephv1.CephCluster, opCon
 				PodCommonSpec: csiopv1a1.PodCommonSpec{
 					PrioritylClassName: &CSIParam.PluginPriorityClassName,
 					Affinity: &v1.Affinity{
-						NodeAffinity: getNodeAffinity(provisionerNodeAffinityEnv, &v1.NodeAffinity{}),
+						NodeAffinity: getNodeAffinity(r.opConfig.Parameters, provisionerNodeAffinityEnv, &v1.NodeAffinity{}),
 					},
-					Tolerations: getToleration(provisionerTolerationsEnv, []v1.Toleration{}),
+					Tolerations: getToleration(r.opConfig.Parameters, provisionerTolerationsEnv, []v1.Toleration{}),
 				},
 				Replicas:  &CSIParam.ProvisionerReplicas,
 				Resources: csiopv1a1.ControllerPluginResourcesSpec{},
diff --git a/pkg/operator/ceph/csi/operator_driver.go b/pkg/operator/ceph/csi/operator_driver.go
index 99c845a2b..f5457be45 100644
--- a/pkg/operator/ceph/csi/operator_driver.go
+++ b/pkg/operator/ceph/csi/operator_driver.go
@@ -92,11 +92,11 @@ func (r *ReconcileCSI) createOrUpdateRBDDriverResource(cluster cephv1.CephCluste
 		Spec: spec,
 	}
 
-	rbdDriver.Spec.ControllerPlugin.Resources = createDriverControllerPluginResources(rbdPluginResource)
+	rbdDriver.Spec.ControllerPlugin.Resources = createDriverControllerPluginResources(r.opConfig.Parameters, rbdPluginResource)
 	rbdDriver.Spec.Liveness = &csiopv1a1.LivenessSpec{
 		MetricsPort: int(CSIParam.RBDLivenessMetricsPort),
 	}
-	rbdDriver.Spec.NodePlugin.Resources = createDriverNodePluginResouces(rbdProvisionerResource)
+	rbdDriver.Spec.NodePlugin.Resources = createDriverNodePluginResouces(r.opConfig.Parameters, rbdProvisionerResource)
 	rbdDriver.Spec.NodePlugin.UpdateStrategy = &v1.DaemonSetUpdateStrategy{
 		Type: v1.RollingUpdateDaemonSetStrategyType,
 	}
@@ -143,12 +143,12 @@ func (r *ReconcileCSI) createOrUpdateCephFSDriverResource(cluster cephv1.CephClu
 		cephFsDriver.Spec.SnapshotPolicy = csiopv1a1.VolumeGroupSnapshotPolicy
 	}
 
-	cephFsDriver.Spec.ControllerPlugin.Resources = createDriverControllerPluginResources(cephFSPluginResource)
+	cephFsDriver.Spec.ControllerPlugin.Resources = createDriverControllerPluginResources(r.opConfig.Parameters, cephFSPluginResource)
 	cephFsDriver.Spec.Liveness = &csiopv1a1.LivenessSpec{
 		MetricsPort: int(CSIParam.CephFSLivenessMetricsPort),
 	}
 
-	cephFsDriver.Spec.NodePlugin.Resources = createDriverNodePluginResouces(cephFSProvisionerResource)
+	cephFsDriver.Spec.NodePlugin.Resources = createDriverNodePluginResouces(r.opConfig.Parameters, cephFSProvisionerResource)
 	cephFsDriver.Spec.NodePlugin.UpdateStrategy = &v1.DaemonSetUpdateStrategy{
 		Type: v1.RollingUpdateDaemonSetStrategyType,
 	}
@@ -190,9 +190,9 @@ func (r *ReconcileCSI) createOrUpdateNFSDriverResource(cluster cephv1.CephCluste
 		Spec: spec,
 	}
 
-	NFSDriver.Spec.ControllerPlugin.Resources = createDriverControllerPluginResources(nfsPluginResource)
+	NFSDriver.Spec.ControllerPlugin.Resources = createDriverControllerPluginResources(r.opConfig.Parameters, nfsPluginResource)
 
-	NFSDriver.Spec.NodePlugin.Resources = createDriverNodePluginResouces(nfsProvisionerResource)
+	NFSDriver.Spec.NodePlugin.Resources = createDriverNodePluginResouces(r.opConfig.Parameters, nfsProvisionerResource)
 	NFSDriver.Spec.NodePlugin.UpdateStrategy = &v1.DaemonSetUpdateStrategy{
 		Type: v1.RollingUpdateDaemonSetStrategyType,
 	}
@@ -270,9 +270,9 @@ func (r *ReconcileCSI) generateDriverSpec(clusterName string) (csiopv1a1.DriverS
 			PodCommonSpec: csiopv1a1.PodCommonSpec{
 				PrioritylClassName: &CSIParam.ProvisionerPriorityClassName,
 				Affinity: &corev1.Affinity{
-					NodeAffinity: getNodeAffinity(pluginNodeAffinityEnv, &corev1.NodeAffinity{}),
+					NodeAffinity: getNodeAffinity(r.opConfig.Parameters, pluginNodeAffinityEnv, &corev1.NodeAffinity{}),
 				},
-				Tolerations: getToleration(pluginTolerationsEnv, []corev1.Toleration{}),
+				Tolerations: getToleration(r.opConfig.Parameters, pluginTolerationsEnv, []corev1.Toleration{}),
 			},
 			Resources:              csiopv1a1.NodePluginResourcesSpec{},
 			KubeletDirPath:         CSIParam.KubeletDirPath,
@@ -282,9 +282,9 @@ func (r *ReconcileCSI) generateDriverSpec(clusterName string) (csiopv1a1.DriverS
 			PodCommonSpec: csiopv1a1.PodCommonSpec{
 				PrioritylClassName: &CSIParam.PluginPriorityClassName,
 				Affinity: &corev1.Affinity{
-					NodeAffinity: getNodeAffinity(provisionerNodeAffinityEnv, &corev1.NodeAffinity{}),
+					NodeAffinity: getNodeAffinity(r.opConfig.Parameters, provisionerNodeAffinityEnv, &corev1.NodeAffinity{}),
 				},
-				Tolerations: getToleration(provisionerTolerationsEnv, []corev1.Toleration{}),
+				Tolerations: getToleration(r.opConfig.Parameters, provisionerTolerationsEnv, []corev1.Toleration{}),
 			},
 			Replicas:  &CSIParam.ProvisionerReplicas,
 			Resources: csiopv1a1.ControllerPluginResourcesSpec{},
@@ -294,9 +294,9 @@ func (r *ReconcileCSI) generateDriverSpec(clusterName string) (csiopv1a1.DriverS
 	}, nil
 }
 
-func createDriverControllerPluginResources(key string) csiopv1a1.ControllerPluginResourcesSpec {
+func createDriverControllerPluginResources(opConfig map[string]string, key string) csiopv1a1.ControllerPluginResourcesSpec {
 	controllerPluginResources := csiopv1a1.ControllerPluginResourcesSpec{}
-	resource := getComputeResource(key)
+	resource := getComputeResource(opConfig, key)
 
 	for _, r := range resource {
 		if !reflect.DeepEqual(r.Resource, corev1.ResourceRequirements{}) {
@@ -347,9 +347,9 @@ func createDriverControllerPluginResources(key string) csiopv1a1.ControllerPlugi
 	return controllerPluginResources
 }
 
-func createDriverNodePluginResouces(key string) csiopv1a1.NodePluginResourcesSpec {
+func createDriverNodePluginResouces(opConfig map[string]string, key string) csiopv1a1.NodePluginResourcesSpec {
 	nodePluginResources := csiopv1a1.NodePluginResourcesSpec{}
-	resource := getComputeResource(key)
+	resource := getComputeResource(opConfig, key)
 
 	for _, r := range resource {
 		if !reflect.DeepEqual(r.Resource, corev1.ResourceRequirements{}) {
diff --git a/pkg/operator/ceph/csi/spec.go b/pkg/operator/ceph/csi/spec.go
index 5b968453f..9381619cc 100644
--- a/pkg/operator/ceph/csi/spec.go
+++ b/pkg/operator/ceph/csi/spec.go
@@ -83,8 +83,6 @@ type Param struct {
 	SidecarLogLevel                          uint8
 	CephFSLivenessMetricsPort                uint16
 	CSIAddonsPort                            uint16
-	CSIAddonsRBDProvisionerPort              uint16
-	CSIAddonsCephFSProvisionerPort           uint16
 	RBDLivenessMetricsPort                   uint16
 	KubeApiBurst                             uint16
 	KubeApiQPS                               float32
@@ -221,14 +219,12 @@ const (
 	// kubelet directory path
 	DefaultKubeletDirPath = "/var/lib/kubelet"
 
-	// gRPC metrics and liveness port for CephFS and RBD
-	DefaultCephFSGRPCMerticsPort          uint16 = 9091
-	DefaultCephFSLivenessMerticsPort      uint16 = 9081
-	DefaultRBDGRPCMerticsPort             uint16 = 9090
-	DefaultRBDLivenessMerticsPort         uint16 = 9080
-	DefaultCSIAddonsPort                  uint16 = 9070
-	DefaultCSIAddonsRBDProvisionerPort    uint16 = 9070
-	DefaultCSIAddonsCephFSProvisionerPort uint16 = 9070
+	// grpc metrics and liveness port for cephfs  and rbd
+	DefaultCephFSGRPCMerticsPort     uint16 = 9091
+	DefaultCephFSLivenessMerticsPort uint16 = 9081
+	DefaultRBDGRPCMerticsPort        uint16 = 9090
+	DefaultRBDLivenessMerticsPort    uint16 = 9080
+	DefaultCSIAddonsPort             uint16 = 9070
 
 	// default log level for csi containers
 	defaultLogLevel        uint8 = 0
@@ -426,25 +422,25 @@ func (r *ReconcileCSI) startDrivers(ownerInfo *k8sutil.OwnerInfo) error {
 	}
 
 	// get common provisioner tolerations and node affinity
-	provisionerTolerations := getToleration(provisionerTolerationsEnv, []corev1.Toleration{})
-	provisionerNodeAffinity := getNodeAffinity(provisionerNodeAffinityEnv, &corev1.NodeAffinity{})
+	provisionerTolerations := getToleration(r.opConfig.Parameters, provisionerTolerationsEnv, []corev1.Toleration{})
+	provisionerNodeAffinity := getNodeAffinity(r.opConfig.Parameters, provisionerNodeAffinityEnv, &corev1.NodeAffinity{})
 
 	// get common plugin tolerations and node affinity
-	pluginTolerations := getToleration(pluginTolerationsEnv, []corev1.Toleration{})
-	pluginNodeAffinity := getNodeAffinity(pluginNodeAffinityEnv, &corev1.NodeAffinity{})
+	pluginTolerations := getToleration(r.opConfig.Parameters, pluginTolerationsEnv, []corev1.Toleration{})
+	pluginNodeAffinity := getNodeAffinity(r.opConfig.Parameters, pluginNodeAffinityEnv, &corev1.NodeAffinity{})
 
 	if rbdPlugin != nil {
 		// get RBD plugin tolerations and node affinity, defaults to common tolerations and node affinity if not specified
-		rbdPluginTolerations := getToleration(rbdPluginTolerationsEnv, pluginTolerations)
-		rbdPluginNodeAffinity := getNodeAffinity(rbdPluginNodeAffinityEnv, pluginNodeAffinity)
+		rbdPluginTolerations := getToleration(r.opConfig.Parameters, rbdPluginTolerationsEnv, pluginTolerations)
+		rbdPluginNodeAffinity := getNodeAffinity(r.opConfig.Parameters, rbdPluginNodeAffinityEnv, pluginNodeAffinity)
 		// apply RBD plugin tolerations and node affinity
 		applyToPodSpec(&rbdPlugin.Spec.Template.Spec, rbdPluginNodeAffinity, rbdPluginTolerations)
 		// apply resource request and limit to rbdplugin containers
-		applyResourcesToContainers(rbdPluginResource, &rbdPlugin.Spec.Template.Spec)
+		applyResourcesToContainers(r.opConfig.Parameters, rbdPluginResource, &rbdPlugin.Spec.Template.Spec)
 		// apply custom mounts to volumes
-		applyVolumeToPodSpec(rbdPluginVolume, &rbdPlugin.Spec.Template.Spec)
+		applyVolumeToPodSpec(r.opConfig.Parameters, rbdPluginVolume, &rbdPlugin.Spec.Template.Spec)
 		// apply custom mounts to volume mounts
-		applyVolumeMountToContainer(rbdPluginVolumeMount, "csi-rbdplugin", &rbdPlugin.Spec.Template.Spec)
+		applyVolumeMountToContainer(r.opConfig.Parameters, rbdPluginVolumeMount, "csi-rbdplugin", &rbdPlugin.Spec.Template.Spec)
 		err = ownerInfo.SetControllerReference(rbdPlugin)
 		if err != nil {
 			return errors.Wrapf(err, "failed to set owner reference to rbd plugin daemonset %q", rbdPlugin.Name)
@@ -462,12 +458,12 @@ func (r *ReconcileCSI) startDrivers(ownerInfo *k8sutil.OwnerInfo) error {
 
 	if rbdProvisionerDeployment != nil {
 		// get RBD provisioner tolerations and node affinity, defaults to common tolerations and node affinity if not specified
-		rbdProvisionerTolerations := getToleration(rbdProvisionerTolerationsEnv, provisionerTolerations)
-		rbdProvisionerNodeAffinity := getNodeAffinity(rbdProvisionerNodeAffinityEnv, provisionerNodeAffinity)
+		rbdProvisionerTolerations := getToleration(r.opConfig.Parameters, rbdProvisionerTolerationsEnv, provisionerTolerations)
+		rbdProvisionerNodeAffinity := getNodeAffinity(r.opConfig.Parameters, rbdProvisionerNodeAffinityEnv, provisionerNodeAffinity)
 		// apply RBD provisioner tolerations and node affinity
 		applyToPodSpec(&rbdProvisionerDeployment.Spec.Template.Spec, rbdProvisionerNodeAffinity, rbdProvisionerTolerations)
 		// apply resource request and limit to rbd provisioner containers
-		applyResourcesToContainers(rbdProvisionerResource, &rbdProvisionerDeployment.Spec.Template.Spec)
+		applyResourcesToContainers(r.opConfig.Parameters, rbdProvisionerResource, &rbdProvisionerDeployment.Spec.Template.Spec)
 		err = ownerInfo.SetControllerReference(rbdProvisionerDeployment)
 		if err != nil {
 			return errors.Wrapf(err, "failed to set owner reference to rbd provisioner deployment %q", rbdProvisionerDeployment.Name)
@@ -504,16 +500,16 @@ func (r *ReconcileCSI) startDrivers(ownerInfo *k8sutil.OwnerInfo) error {
 
 	if cephfsPlugin != nil {
 		// get CephFS plugin tolerations and node affinity, defaults to common tolerations and node affinity if not specified
-		cephFSPluginTolerations := getToleration(cephFSPluginTolerationsEnv, pluginTolerations)
-		cephFSPluginNodeAffinity := getNodeAffinity(cephFSPluginNodeAffinityEnv, pluginNodeAffinity)
+		cephFSPluginTolerations := getToleration(r.opConfig.Parameters, cephFSPluginTolerationsEnv, pluginTolerations)
+		cephFSPluginNodeAffinity := getNodeAffinity(r.opConfig.Parameters, cephFSPluginNodeAffinityEnv, pluginNodeAffinity)
 		// apply CephFS plugin tolerations and node affinity
 		applyToPodSpec(&cephfsPlugin.Spec.Template.Spec, cephFSPluginNodeAffinity, cephFSPluginTolerations)
 		// apply resource request and limit to cephfs plugin containers
-		applyResourcesToContainers(cephFSPluginResource, &cephfsPlugin.Spec.Template.Spec)
+		applyResourcesToContainers(r.opConfig.Parameters, cephFSPluginResource, &cephfsPlugin.Spec.Template.Spec)
 		// apply custom mounts to volumes
-		applyVolumeToPodSpec(cephFSPluginVolume, &cephfsPlugin.Spec.Template.Spec)
+		applyVolumeToPodSpec(r.opConfig.Parameters, cephFSPluginVolume, &cephfsPlugin.Spec.Template.Spec)
 		// apply custom mounts to volume mounts
-		applyVolumeMountToContainer(cephFSPluginVolumeMount, "csi-cephfsplugin", &cephfsPlugin.Spec.Template.Spec)
+		applyVolumeMountToContainer(r.opConfig.Parameters, cephFSPluginVolumeMount, "csi-cephfsplugin", &cephfsPlugin.Spec.Template.Spec)
 		err = ownerInfo.SetControllerReference(cephfsPlugin)
 		if err != nil {
 			return errors.Wrapf(err, "failed to set owner reference to cephfs plugin daemonset %q", cephfsPlugin.Name)
@@ -532,13 +528,13 @@ func (r *ReconcileCSI) startDrivers(ownerInfo *k8sutil.OwnerInfo) error {
 
 	if cephfsProvisionerDeployment != nil {
 		// get CephFS provisioner tolerations and node affinity, defaults to common tolerations and node affinity if not specified
-		cephFSProvisionerTolerations := getToleration(cephFSProvisionerTolerationsEnv, provisionerTolerations)
-		cephFSProvisionerNodeAffinity := getNodeAffinity(cephFSProvisionerNodeAffinityEnv, provisionerNodeAffinity)
+		cephFSProvisionerTolerations := getToleration(r.opConfig.Parameters, cephFSProvisionerTolerationsEnv, provisionerTolerations)
+		cephFSProvisionerNodeAffinity := getNodeAffinity(r.opConfig.Parameters, cephFSProvisionerNodeAffinityEnv, provisionerNodeAffinity)
 		// apply CephFS provisioner tolerations and node affinity
 		applyToPodSpec(&cephfsProvisionerDeployment.Spec.Template.Spec, cephFSProvisionerNodeAffinity, cephFSProvisionerTolerations)
 		// get resource details for cephfs provisioner
 		// apply resource request and limit to cephfs provisioner containers
-		applyResourcesToContainers(cephFSProvisionerResource, &cephfsProvisionerDeployment.Spec.Template.Spec)
+		applyResourcesToContainers(r.opConfig.Parameters, cephFSProvisionerResource, &cephfsProvisionerDeployment.Spec.Template.Spec)
 		err = ownerInfo.SetControllerReference(cephfsProvisionerDeployment)
 		if err != nil {
 			return errors.Wrapf(err, "failed to set owner reference to cephfs provisioner deployment %q", cephfsProvisionerDeployment.Name)
@@ -574,16 +570,16 @@ func (r *ReconcileCSI) startDrivers(ownerInfo *k8sutil.OwnerInfo) error {
 
 	if nfsPlugin != nil {
 		// get NFS plugin tolerations and node affinity, defaults to common tolerations and node affinity if not specified
-		nfsPluginTolerations := getToleration(nfsPluginTolerationsEnv, pluginTolerations)
-		nfsPluginNodeAffinity := getNodeAffinity(nfsPluginNodeAffinityEnv, pluginNodeAffinity)
+		nfsPluginTolerations := getToleration(r.opConfig.Parameters, nfsPluginTolerationsEnv, pluginTolerations)
+		nfsPluginNodeAffinity := getNodeAffinity(r.opConfig.Parameters, nfsPluginNodeAffinityEnv, pluginNodeAffinity)
 		// apply NFS plugin tolerations and node affinity
 		applyToPodSpec(&nfsPlugin.Spec.Template.Spec, nfsPluginNodeAffinity, nfsPluginTolerations)
 		// apply resource request and limit to nfs plugin containers
-		applyResourcesToContainers(nfsPluginResource, &nfsPlugin.Spec.Template.Spec)
+		applyResourcesToContainers(r.opConfig.Parameters, nfsPluginResource, &nfsPlugin.Spec.Template.Spec)
 		// apply custom mounts to volumes
-		applyVolumeToPodSpec(nfsPluginVolume, &nfsPlugin.Spec.Template.Spec)
+		applyVolumeToPodSpec(r.opConfig.Parameters, nfsPluginVolume, &nfsPlugin.Spec.Template.Spec)
 		// apply custom mounts to volume mounts
-		applyVolumeMountToContainer(nfsPluginVolumeMount, "csi-nfsplugin", &nfsPlugin.Spec.Template.Spec)
+		applyVolumeMountToContainer(r.opConfig.Parameters, nfsPluginVolumeMount, "csi-nfsplugin", &nfsPlugin.Spec.Template.Spec)
 		err = ownerInfo.SetControllerReference(nfsPlugin)
 		if err != nil {
 			return errors.Wrapf(err, "failed to set owner reference to nfs plugin daemonset %q", nfsPlugin.Name)
@@ -602,13 +598,13 @@ func (r *ReconcileCSI) startDrivers(ownerInfo *k8sutil.OwnerInfo) error {
 
 	if nfsProvisionerDeployment != nil {
 		// get NFS provisioner tolerations and node affinity, defaults to common tolerations and node affinity if not specified
-		nfsProvisionerTolerations := getToleration(nfsProvisionerTolerationsEnv, provisionerTolerations)
-		nfsProvisionerNodeAffinity := getNodeAffinity(nfsProvisionerNodeAffinityEnv, provisionerNodeAffinity)
+		nfsProvisionerTolerations := getToleration(r.opConfig.Parameters, nfsProvisionerTolerationsEnv, provisionerTolerations)
+		nfsProvisionerNodeAffinity := getNodeAffinity(r.opConfig.Parameters, nfsProvisionerNodeAffinityEnv, provisionerNodeAffinity)
 		// apply NFS provisioner tolerations and node affinity
 		applyToPodSpec(&nfsProvisionerDeployment.Spec.Template.Spec, nfsProvisionerNodeAffinity, nfsProvisionerTolerations)
 		// get resource details for nfs provisioner
 		// apply resource request and limit to nfs provisioner containers
-		applyResourcesToContainers(nfsProvisionerResource, &nfsProvisionerDeployment.Spec.Template.Spec)
+		applyResourcesToContainers(r.opConfig.Parameters, nfsProvisionerResource, &nfsProvisionerDeployment.Spec.Template.Spec)
 		err = ownerInfo.SetControllerReference(nfsProvisionerDeployment)
 		if err != nil {
 			return errors.Wrapf(err, "failed to set owner reference to nfs provisioner deployment %q", nfsProvisionerDeployment.Name)
@@ -634,7 +630,7 @@ func (r *ReconcileCSI) startDrivers(ownerInfo *k8sutil.OwnerInfo) error {
 	if EnableRBD {
 		err = csiDriverobj.createCSIDriverInfo(
 			r.opManagerContext, r.context.Clientset,
-			RBDDriverName, k8sutil.GetOperatorSetting("CSI_RBD_FSGROUPPOLICY", string(k8scsi.FileFSGroupPolicy)),
+			RBDDriverName, k8sutil.GetValue(r.opConfig.Parameters, "CSI_RBD_FSGROUPPOLICY", string(k8scsi.FileFSGroupPolicy)),
 			tp.Param.RBDAttachRequired, CSIParam.EnableCSIDriverSeLinuxMount)
 		if err != nil {
 			return errors.Wrapf(err, "failed to create CSI driver object for %q", RBDDriverName)
@@ -643,7 +639,7 @@ func (r *ReconcileCSI) startDrivers(ownerInfo *k8sutil.OwnerInfo) error {
 	if EnableCephFS {
 		err = csiDriverobj.createCSIDriverInfo(
 			r.opManagerContext, r.context.Clientset,
-			CephFSDriverName, k8sutil.GetOperatorSetting("CSI_CEPHFS_FSGROUPPOLICY", string(k8scsi.FileFSGroupPolicy)),
+			CephFSDriverName, k8sutil.GetValue(r.opConfig.Parameters, "CSI_CEPHFS_FSGROUPPOLICY", string(k8scsi.FileFSGroupPolicy)),
 			tp.Param.CephFSAttachRequired, CSIParam.EnableCSIDriverSeLinuxMount)
 		if err != nil {
 			return errors.Wrapf(err, "failed to create CSI driver object for %q", CephFSDriverName)
@@ -651,7 +647,7 @@ func (r *ReconcileCSI) startDrivers(ownerInfo *k8sutil.OwnerInfo) error {
 	}
 	if EnableNFS {
 		err = csiDriverobj.createCSIDriverInfo(r.opManagerContext, r.context.Clientset,
-			NFSDriverName, k8sutil.GetOperatorSetting("CSI_NFS_FSGROUPPOLICY", string(k8scsi.FileFSGroupPolicy)),
+			NFSDriverName, k8sutil.GetValue(r.opConfig.Parameters, "CSI_NFS_FSGROUPPOLICY", string(k8scsi.FileFSGroupPolicy)),
 			tp.Param.NFSAttachRequired, CSIParam.EnableCSIDriverSeLinuxMount)
 		if err != nil {
 			return errors.Wrapf(err, "failed to create CSI driver object for %q", NFSDriverName)
diff --git a/pkg/operator/ceph/csi/template/cephfs/csi-cephfsplugin-provisioner-dep.yaml b/pkg/operator/ceph/csi/template/cephfs/csi-cephfsplugin-provisioner-dep.yaml
index ff9554ad7..5ed29f704 100644
--- a/pkg/operator/ceph/csi/template/cephfs/csi-cephfsplugin-provisioner-dep.yaml
+++ b/pkg/operator/ceph/csi/template/cephfs/csi-cephfsplugin-provisioner-dep.yaml
@@ -216,7 +216,7 @@ spec:
             - "--node-id=$(NODE_ID)"
             - "--v={{ .LogLevel }}"
             - "--csi-addons-address=$(CSIADDONS_ENDPOINT)"
-            - "--controller-port={{ .CSIAddonsCephFSProvisionerPort }}"
+            - "--controller-port={{ .CSIAddonsPort }}"
             - "--pod=$(POD_NAME)"
             - "--namespace=$(POD_NAMESPACE)"
             - "--pod-uid=$(POD_UID)"
@@ -231,7 +231,7 @@ spec:
             - "--log_file={{ .CsiLogRootPath }}/log/{{ .CsiComponentName }}/csi-addons.log"
             {{ end }}
           ports:
-            - containerPort: {{ .CSIAddonsCephFSProvisionerPort }}
+            - containerPort: {{ .CSIAddonsPort }}
           env:
             - name: NODE_ID
               valueFrom:
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 bacd76682..135338322 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
@@ -169,7 +169,7 @@ spec:
             - "--node-id=$(NODE_ID)"
             - "--v={{ .LogLevel }}"
             - "--csi-addons-address=$(CSIADDONS_ENDPOINT)"
-            - "--controller-port={{ .CSIAddonsRBDProvisionerPort }}"
+            - "--controller-port={{ .CSIAddonsPort }}"
             - "--pod=$(POD_NAME)"
             - "--namespace=$(POD_NAMESPACE)"
             - "--pod-uid=$(POD_UID)"
@@ -184,7 +184,7 @@ spec:
             - "--log_file={{ .CsiLogRootPath }}/log/{{ .CsiComponentName }}/csi-addons.log"
             {{ end }}
           ports:
-            - containerPort: {{ .CSIAddonsRBDProvisionerPort }}
+            - containerPort: {{ .CSIAddonsPort }}
           env:
             - name: NODE_ID
               valueFrom:
diff --git a/pkg/operator/ceph/csi/util.go b/pkg/operator/ceph/csi/util.go
index 60f549bb1..4c3e70827 100644
--- a/pkg/operator/ceph/csi/util.go
+++ b/pkg/operator/ceph/csi/util.go
@@ -98,8 +98,8 @@ func applyLogrotateSidecar(specTemplate *corev1.PodTemplateSpec, name, templateD
 	specTemplate.Spec.Containers = append(specTemplate.Spec.Containers, logrotateSidecarContainer)
 }
 
-func applyResourcesToContainers(key string, podspec *corev1.PodSpec) {
-	resource := getComputeResource(key)
+func applyResourcesToContainers(opConfig map[string]string, key string, podspec *corev1.PodSpec) {
+	resource := getComputeResource(opConfig, key)
 
 	for _, r := range resource {
 		for i, c := range podspec.Containers {
@@ -110,12 +110,12 @@ func applyResourcesToContainers(key string, podspec *corev1.PodSpec) {
 	}
 }
 
-func getComputeResource(key string) []k8sutil.ContainerResource {
+func getComputeResource(opConfig map[string]string, key string) []k8sutil.ContainerResource {
 	// Add Resource list if any
 	resource := []k8sutil.ContainerResource{}
 	var err error
 
-	if resourceRaw := k8sutil.GetOperatorSetting(key, ""); resourceRaw != "" {
+	if resourceRaw := k8sutil.GetValue(opConfig, key, ""); resourceRaw != "" {
 		resource, err = k8sutil.YamlToContainerResourceArray(resourceRaw)
 		if err != nil {
 			logger.Warningf("failed to parse %q. %v", resourceRaw, err)
@@ -124,9 +124,9 @@ func getComputeResource(key string) []k8sutil.ContainerResource {
 	return resource
 }
 
-func getToleration(tolerationsName string, defaultTolerations []corev1.Toleration) []corev1.Toleration {
+func getToleration(opConfig map[string]string, tolerationsName string, defaultTolerations []corev1.Toleration) []corev1.Toleration {
 	// Add toleration if any, otherwise return defaultTolerations
-	tolerationsRaw := k8sutil.GetOperatorSetting(tolerationsName, "")
+	tolerationsRaw := k8sutil.GetValue(opConfig, tolerationsName, "")
 	if tolerationsRaw == "" {
 		return defaultTolerations
 	}
@@ -147,9 +147,9 @@ func getToleration(tolerationsName string, defaultTolerations []corev1.Toleratio
 	return tolerations
 }
 
-func getNodeAffinity(nodeAffinityName string, defaultNodeAffinity *corev1.NodeAffinity) *corev1.NodeAffinity {
+func getNodeAffinity(opConfig map[string]string, nodeAffinityName string, defaultNodeAffinity *corev1.NodeAffinity) *corev1.NodeAffinity {
 	// Add NodeAffinity if any, otherwise return defaultNodeAffinity
-	nodeAffinity := k8sutil.GetOperatorSetting(nodeAffinityName, "")
+	nodeAffinity := k8sutil.GetValue(opConfig, nodeAffinityName, "")
 	if nodeAffinity == "" {
 		return defaultNodeAffinity
 	}
@@ -168,9 +168,9 @@ func applyToPodSpec(pod *corev1.PodSpec, n *corev1.NodeAffinity, t []corev1.Tole
 	}
 }
 
-func getPortFromConfig(env string, defaultPort uint16) (uint16, error) {
-	port := k8sutil.GetOperatorSetting(env, strconv.Itoa(int(defaultPort)))
-	if strings.TrimSpace(k8sutil.GetOperatorSetting(env, strconv.Itoa(int(defaultPort)))) == "" {
+func getPortFromConfig(data map[string]string, env string, defaultPort uint16) (uint16, error) {
+	port := k8sutil.GetValue(data, env, strconv.Itoa(int(defaultPort)))
+	if strings.TrimSpace(k8sutil.GetValue(data, env, strconv.Itoa(int(defaultPort)))) == "" {
 		return defaultPort, nil
 	}
 	p, err := strconv.ParseUint(port, 10, 64)
@@ -203,8 +203,8 @@ func GetPodAntiAffinity(key, value string) corev1.PodAntiAffinity {
 	}
 }
 
-func applyVolumeToPodSpec(configName string, podspec *corev1.PodSpec) {
-	volumesRaw := k8sutil.GetOperatorSetting(configName, "")
+func applyVolumeToPodSpec(opConfig map[string]string, configName string, podspec *corev1.PodSpec) {
+	volumesRaw := k8sutil.GetValue(opConfig, configName, "")
 	if volumesRaw == "" {
 		return
 	}
@@ -230,8 +230,8 @@ func applyVolumeToPodSpec(configName string, podspec *corev1.PodSpec) {
 	}
 }
 
-func applyVolumeMountToContainer(configName, containerName string, podspec *corev1.PodSpec) {
-	volumeMountsRaw := k8sutil.GetOperatorSetting(configName, "")
+func applyVolumeMountToContainer(opConfig map[string]string, configName, containerName string, podspec *corev1.PodSpec) {
+	volumeMountsRaw := k8sutil.GetValue(opConfig, configName, "")
 	if volumeMountsRaw == "" {
 		return
 	}
@@ -267,8 +267,8 @@ func applyVolumeMountToContainer(configName, containerName string, podspec *core
 
 // getImage returns the image for the given setting name. If the image does not contain version,
 // the default version is appended from the default image.
-func getImage(settingName, defaultImage string) string {
-	image := k8sutil.GetOperatorSetting(settingName, defaultImage)
+func getImage(data map[string]string, settingName, defaultImage string) string {
+	image := k8sutil.GetValue(data, settingName, defaultImage)
 	if !strings.Contains(image, ":") {
 		version := strings.SplitN(defaultImage, ":", 2)[1]
 		image = fmt.Sprintf("%s:%s", image, version)
diff --git a/pkg/operator/ceph/object/bucket/controller.go b/pkg/operator/ceph/object/bucket/controller.go
index b9a4007f7..eed00adc9 100644
--- a/pkg/operator/ceph/object/bucket/controller.go
+++ b/pkg/operator/ceph/object/bucket/controller.go
@@ -36,6 +36,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 )
 
 const (
@@ -139,6 +140,25 @@ func (r *ReconcileBucket) reconcile(request reconcile.Request) (reconcile.Result
 		return reconcile.Result{}, nil
 	}
 
+	// Fetch the operator's configmap. We force the NamespaceName to the operator since the request
+	// could be a CephCluster. If so the NamespaceName will be the one from the cluster and thus the
+	// CM won't be found
+	opNamespaceName := types.NamespacedName{Name: opcontroller.OperatorSettingConfigMapName, Namespace: r.opConfig.OperatorNamespace}
+	opConfig := &v1.ConfigMap{}
+	err = r.client.Get(r.opManagerContext, opNamespaceName, opConfig)
+	if err != nil {
+		if kerrors.IsNotFound(err) {
+			logger.Debug("operator's configmap resource not found. will use default value or env var.")
+			r.opConfig.Parameters = make(map[string]string)
+		} else {
+			// Error reading the object - requeue the request.
+			return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to get operator's configmap")
+		}
+	} else {
+		// Populate the operator's config
+		r.opConfig.Parameters = opConfig.Data
+	}
+
 	// Populate clusterInfo during each reconcile
 	clusterInfo, _, _, err := opcontroller.LoadClusterInfo(r.context, r.opManagerContext, cephCluster.Namespace, &cephCluster.Spec)
 	if err != nil {
@@ -157,7 +177,7 @@ func (r *ReconcileBucket) reconcile(request reconcile.Request) (reconcile.Result
 
 	// note: the error return below is ignored and is expected to be removed from the
 	//   bucket library's `NewProvisioner` function
-	bucketController, _ := NewBucketController(r.context.KubeConfig, bucketProvisioner)
+	bucketController, _ := NewBucketController(r.context.KubeConfig, bucketProvisioner, r.opConfig.Parameters)
 
 	// We must run this in a go routine since RunWithContext() blocks and waits for the context to
 	// be Done. However, since it has a context, the go routine will exit on reload with SIGHUP
diff --git a/pkg/operator/ceph/object/bucket/util.go b/pkg/operator/ceph/object/bucket/util.go
index 7e8df47ea..19e294cb9 100644
--- a/pkg/operator/ceph/object/bucket/util.go
+++ b/pkg/operator/ceph/object/bucket/util.go
@@ -43,9 +43,9 @@ const (
 	objectStoreEndpoint  = "endpoint"
 )
 
-func NewBucketController(cfg *rest.Config, p *Provisioner) (*provisioner.Provisioner, error) {
+func NewBucketController(cfg *rest.Config, p *Provisioner, data map[string]string) (*provisioner.Provisioner, error) {
 	const allNamespaces = ""
-	provName, err := cephObject.GetObjectBucketProvisioner(p.clusterInfo.Namespace)
+	provName, err := cephObject.GetObjectBucketProvisioner(data, p.clusterInfo.Namespace)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to get provisioner name")
 	}
diff --git a/pkg/operator/ceph/object/objectstore.go b/pkg/operator/ceph/object/objectstore.go
index 67a41626a..080367731 100644
--- a/pkg/operator/ceph/object/objectstore.go
+++ b/pkg/operator/ceph/object/objectstore.go
@@ -1137,10 +1137,10 @@ func poolName(poolPrefix, poolName string) string {
 }
 
 // GetObjectBucketProvisioner returns the bucket provisioner name appended with operator namespace if OBC is watching on it
-func GetObjectBucketProvisioner(namespace string) (string, error) {
+func GetObjectBucketProvisioner(data map[string]string, namespace string) (string, error) {
 	provName := bucketProvisionerName
-	obcWatchOnNamespace := k8sutil.GetOperatorSetting("ROOK_OBC_WATCH_OPERATOR_NAMESPACE", "false")
-	obcProvisionerNamePrefix := k8sutil.GetOperatorSetting("ROOK_OBC_PROVISIONER_NAME_PREFIX", "")
+	obcWatchOnNamespace := k8sutil.GetValue(data, "ROOK_OBC_WATCH_OPERATOR_NAMESPACE", "false")
+	obcProvisionerNamePrefix := k8sutil.GetValue(data, "ROOK_OBC_PROVISIONER_NAME_PREFIX", "")
 	if obcProvisionerNamePrefix != "" {
 		errList := validation.IsDNS1123Label(obcProvisionerNamePrefix)
 		if len(errList) > 0 {
diff --git a/pkg/operator/ceph/operator.go b/pkg/operator/ceph/operator.go
index 2674c41ab..e43ba8c78 100644
--- a/pkg/operator/ceph/operator.go
+++ b/pkg/operator/ceph/operator.go
@@ -129,11 +129,6 @@ func (o *Operator) runCRDManager() {
 	// Create the context and the cancellation function
 	opManagerContext, opManagerStop = context.WithCancel(context.Background())
 
-	// Initialize the operator settings from the configmap
-	if err := k8sutil.ApplyOperatorSettingsConfigmap(opManagerContext, o.context.Clientset); err != nil {
-		panic("failed to load operator settings configmap. " + err.Error())
-	}
-
 	// The operator config manager is also watching for changes here so if the operator config map
 	// content changes for ROOK_CURRENT_NAMESPACE_ONLY we must reload the operator CRD manager
 	o.namespaceToWatch()
@@ -154,7 +149,7 @@ func (o *Operator) runCRDManager() {
 }
 
 func (o *Operator) namespaceToWatch() {
-	currentNamespaceOnly := k8sutil.GetOperatorSetting("ROOK_CURRENT_NAMESPACE_ONLY", "true")
+	currentNamespaceOnly, _ := k8sutil.GetOperatorSetting(opManagerContext, o.context.Clientset, opcontroller.OperatorSettingConfigMapName, "ROOK_CURRENT_NAMESPACE_ONLY", "true")
 	if currentNamespaceOnly == "true" {
 		o.config.NamespaceToWatch = o.config.OperatorNamespace
 		logger.Infof("watching the current namespace %q for Ceph CRs", o.config.OperatorNamespace)
diff --git a/pkg/operator/discover/discover.go b/pkg/operator/discover/discover.go
index dca033d57..5b73b8cc0 100644
--- a/pkg/operator/discover/discover.go
+++ b/pkg/operator/discover/discover.go
@@ -73,16 +73,16 @@ func New(clientset kubernetes.Interface) *Discover {
 }
 
 // Start the discover
-func (d *Discover) Start(ctx context.Context, namespace, discoverImage, securityAccount string, useCephVolume bool) error {
-	err := d.createDiscoverDaemonSet(ctx, namespace, discoverImage, securityAccount, useCephVolume)
+func (d *Discover) Start(ctx context.Context, namespace, discoverImage, securityAccount string, data map[string]string, useCephVolume bool) error {
+	err := d.createDiscoverDaemonSet(ctx, namespace, discoverImage, securityAccount, data, useCephVolume)
 	if err != nil {
 		return fmt.Errorf("failed to start discover daemonset. %v", err)
 	}
 	return nil
 }
 
-func (d *Discover) createDiscoverDaemonSet(ctx context.Context, namespace, discoverImage, securityAccount string, useCephVolume bool) error {
-	discoveryInterval := k8sutil.GetOperatorSetting(discoverIntervalEnv, defaultDiscoverInterval)
+func (d *Discover) createDiscoverDaemonSet(ctx context.Context, namespace, discoverImage, securityAccount string, data map[string]string, useCephVolume bool) error {
+	discoveryInterval := k8sutil.GetValue(data, discoverIntervalEnv, defaultDiscoverInterval)
 
 	discoveryParameters := []string{"discover",
 		"--discover-interval", discoveryInterval}
@@ -90,7 +90,7 @@ func (d *Discover) createDiscoverDaemonSet(ctx context.Context, namespace, disco
 		discoveryParameters = append(discoveryParameters, "--use-ceph-volume")
 	}
 
-	discoverDaemonResourcesRaw := k8sutil.GetOperatorSetting(discoverDaemonResourcesEnv, "")
+	discoverDaemonResourcesRaw := k8sutil.GetValue(data, discoverDaemonResourcesEnv, "")
 	discoverDaemonResources, err := k8sutil.YamlToContainerResource(discoverDaemonResourcesRaw)
 	if err != nil {
 		logger.Warningf("failed to parse.%s %v", discoverDaemonResourcesRaw, err)
@@ -176,7 +176,7 @@ func (d *Discover) createDiscoverDaemonSet(ctx context.Context, namespace, disco
 						},
 					},
 					HostNetwork:       opcontroller.EnforceHostNetwork(),
-					PriorityClassName: k8sutil.GetOperatorSetting(discoverDaemonsetPriorityClassNameEnv, ""),
+					PriorityClassName: k8sutil.GetValue(data, discoverDaemonsetPriorityClassNameEnv, ""),
 					SecurityContext:   &v1.PodSecurityContext{},
 				},
 			},
@@ -191,18 +191,18 @@ func (d *Discover) createDiscoverDaemonSet(ctx context.Context, namespace, disco
 	}
 
 	// Add toleration if any
-	tolerationValue := k8sutil.GetOperatorSetting(discoverDaemonsetTolerationEnv, "")
+	tolerationValue := k8sutil.GetValue(data, discoverDaemonsetTolerationEnv, "")
 	if tolerationValue != "" {
 		ds.Spec.Template.Spec.Tolerations = []v1.Toleration{
 			{
 				Effect:   v1.TaintEffect(tolerationValue),
 				Operator: v1.TolerationOpExists,
-				Key:      k8sutil.GetOperatorSetting(discoverDaemonsetTolerationKeyEnv, ""),
+				Key:      k8sutil.GetValue(data, discoverDaemonsetTolerationKeyEnv, ""),
 			},
 		}
 	}
 
-	tolerationsRaw := k8sutil.GetOperatorSetting(discoverDaemonsetTolerationsEnv, "")
+	tolerationsRaw := k8sutil.GetValue(data, discoverDaemonsetTolerationsEnv, "")
 	tolerations, err := k8sutil.YamlToTolerations(tolerationsRaw)
 	if err != nil {
 		logger.Warningf("failed to parse %s. %+v", tolerationsRaw, err)
@@ -211,7 +211,7 @@ func (d *Discover) createDiscoverDaemonSet(ctx context.Context, namespace, disco
 	ds.Spec.Template.Spec.Tolerations = append(ds.Spec.Template.Spec.Tolerations, tolerations...)
 
 	// Add NodeAffinity if any
-	nodeAffinity := k8sutil.GetOperatorSetting(discoverDaemonSetNodeAffinityEnv, "")
+	nodeAffinity := k8sutil.GetValue(data, discoverDaemonSetNodeAffinityEnv, "")
 	if nodeAffinity != "" {
 		v1NodeAffinity, err := k8sutil.GenerateNodeAffinity(nodeAffinity)
 		if err != nil {
@@ -224,7 +224,7 @@ func (d *Discover) createDiscoverDaemonSet(ctx context.Context, namespace, disco
 		logger.Infof("nodeAffinity: %s", v1NodeAffinity)
 	}
 
-	podLabels := k8sutil.GetOperatorSetting(discoverDaemonSetPodLabelsEnv, "")
+	podLabels := k8sutil.GetValue(data, discoverDaemonSetPodLabelsEnv, "")
 	if podLabels != "" {
 		podLabels := k8sutil.ParseStringToLabels(podLabels)
 		// Override / Set the app label even if set by the user as
diff --git a/pkg/operator/k8sutil/configmap.go b/pkg/operator/k8sutil/configmap.go
index 7e57bb7d7..08b932430 100644
--- a/pkg/operator/k8sutil/configmap.go
+++ b/pkg/operator/k8sutil/configmap.go
@@ -20,17 +20,17 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"sync"
 	"time"
 
 	"github.com/pkg/errors"
 	v1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 )
 
-var loadedOperatorSettings bool
+var loggedOperatorSettings = sync.Map{}
 
 // DeleteConfigMap deletes a ConfigMap.
 func DeleteConfigMap(ctx context.Context, clientset kubernetes.Interface, cmName, namespace string, opts *DeleteOptions) error {
@@ -70,43 +70,50 @@ func CreateOrUpdateConfigMap(ctx context.Context, clientset kubernetes.Interface
 	return existingCm, nil
 }
 
-// GetOperatorSetting gets the operator setting from Env Var merged with ConfigMap
+// GetOperatorSetting gets the operator setting from ConfigMap or Env Var
 // returns defaultValue if setting is not found
-func GetOperatorSetting(settingName, defaultValue string) string {
-	if !loadedOperatorSettings {
-		logger.Warningf("WARNING: attempting to load operator setting %q before configmap is loaded", settingName)
-	}
-	if settingValue, ok := os.LookupEnv(settingName); ok {
-		return settingValue
+func GetOperatorSetting(context context.Context, clientset kubernetes.Interface, configMapName, settingName, defaultValue string) (string, error) {
+	// config must be in operator pod namespace
+	namespace := os.Getenv(PodNamespaceEnvVar)
+	cm, err := clientset.CoreV1().ConfigMaps(namespace).Get(context, configMapName, metav1.GetOptions{})
+	if err != nil {
+		if kerrors.IsNotFound(err) {
+			if settingValue, ok := os.LookupEnv(settingName); ok {
+				logger.Infof("%s=%q (env var)", settingName, settingValue)
+				return settingValue, nil
+			}
+			logger.Infof("%s=%q (default)", settingName, defaultValue)
+			return defaultValue, nil
+		}
+		return defaultValue, fmt.Errorf("error reading ConfigMap %q. %v", configMapName, err)
 	}
-	return defaultValue
+
+	return GetValue(cm.Data, settingName, defaultValue), nil
 }
 
-func ApplyOperatorSettingsConfigmap(ctx context.Context, clientset kubernetes.Interface) error {
-	namespacedName := types.NamespacedName{Namespace: os.Getenv(PodNamespaceEnvVar), Name: "rook-ceph-operator-config"}
-	logger.Debugf("loading operator settings configmap from %v", namespacedName)
-	opConfig, err := clientset.CoreV1().ConfigMaps(namespacedName.Namespace).Get(ctx, namespacedName.Name, metav1.GetOptions{})
+func GetValue(data map[string]string, settingName, defaultValue string) string {
+	settingValue := defaultValue
+	settingSource := "default"
 
-	if err != nil {
-		if kerrors.IsNotFound(err) {
-			loadedOperatorSettings = true
-			logger.Debug("operator's configmap resource not found. will use default value or env var.")
-			return nil
-		}
-		return err
+	if val, ok := data[settingName]; ok {
+		settingValue = val
+		settingSource = "configmap"
+	} else if val, ok := os.LookupEnv(settingName); ok {
+		settingValue = val
+		settingSource = "env var"
 	}
 
-	for key, value := range opConfig.Data {
-		currentValue := os.Getenv(key)
-		if currentValue == value {
-			continue
-		}
-		logger.Infof("operator setting %q = %q", key, value)
-		if err := os.Setenv(key, value); err != nil {
-			logger.Errorf("failed to set env var %q = %q. %v", key, value, err)
-		}
+	if logChangedSettings(settingName, settingValue) {
+		logger.Infof("%s=%q (%s)", settingName, settingValue, settingSource)
+	}
+
+	return settingValue
+}
+
+func logChangedSettings(settingName, settingValue string) bool {
+	if val, ok := loggedOperatorSettings.Load(settingName); !ok || val != settingValue {
+		loggedOperatorSettings.Store(settingName, settingValue)
+		return true
 	}
-	loadedOperatorSettings = true
-	logger.Debug("done loading operator settings configmap")
-	return nil
+	return false
 }


Test patch:
diff --git a/pkg/operator/ceph/controller/controller_utils_test.go b/pkg/operator/ceph/controller/controller_utils_test.go
index d1eac8297..cda45d991 100644
--- a/pkg/operator/ceph/controller/controller_utils_test.go
+++ b/pkg/operator/ceph/controller/controller_utils_test.go
@@ -18,7 +18,6 @@ package controller
 
 import (
 	ctx "context"
-	"os"
 	"testing"
 	"time"
 
@@ -81,94 +80,83 @@ func TestCanIgnoreHealthErrStatusInReconcile(t *testing.T) {
 }
 
 func TestSetCephCommandsTimeout(t *testing.T) {
-	defer os.Unsetenv("ROOK_CEPH_COMMANDS_TIMEOUT_SECONDS")
-
-	SetCephCommandsTimeout()
+	SetCephCommandsTimeout(map[string]string{})
 	assert.Equal(t, 15*time.Second, exec.CephCommandsTimeout)
 
 	exec.CephCommandsTimeout = 0
-	os.Setenv("ROOK_CEPH_COMMANDS_TIMEOUT_SECONDS", "0")
-	SetCephCommandsTimeout()
+	SetCephCommandsTimeout(map[string]string{"ROOK_CEPH_COMMANDS_TIMEOUT_SECONDS": "0"})
 	assert.Equal(t, 15*time.Second, exec.CephCommandsTimeout)
 
 	exec.CephCommandsTimeout = 0
-	os.Setenv("ROOK_CEPH_COMMANDS_TIMEOUT_SECONDS", "1")
-	SetCephCommandsTimeout()
+	SetCephCommandsTimeout(map[string]string{"ROOK_CEPH_COMMANDS_TIMEOUT_SECONDS": "1"})
 	assert.Equal(t, 1*time.Second, exec.CephCommandsTimeout)
 }
 
 func TestSetAllowLoopDevices(t *testing.T) {
-	defer os.Unsetenv("ROOK_CEPH_ALLOW_LOOP_DEVICES")
-
-	SetAllowLoopDevices()
+	SetAllowLoopDevices(map[string]string{})
 	assert.False(t, LoopDevicesAllowed())
 
-	os.Setenv("ROOK_CEPH_ALLOW_LOOP_DEVICES", "foo")
-	SetAllowLoopDevices()
+	SetAllowLoopDevices(map[string]string{"ROOK_CEPH_ALLOW_LOOP_DEVICES": "foo"})
 	assert.False(t, LoopDevicesAllowed())
 
-	os.Setenv("ROOK_CEPH_ALLOW_LOOP_DEVICES", "false")
-	SetAllowLoopDevices()
+	SetAllowLoopDevices(map[string]string{"ROOK_CEPH_ALLOW_LOOP_DEVICES": "false"})
 	assert.False(t, LoopDevicesAllowed())
 
-	os.Setenv("ROOK_CEPH_ALLOW_LOOP_DEVICES", "true")
-	SetAllowLoopDevices()
+	SetAllowLoopDevices(map[string]string{"ROOK_CEPH_ALLOW_LOOP_DEVICES": "true"})
 	assert.True(t, LoopDevicesAllowed())
 }
 
 func TestSetEnforceHostNetwork(t *testing.T) {
-	defer os.Unsetenv(enforceHostNetworkSettingName)
-
 	logger.Infof("testing default value for %v", enforceHostNetworkSettingName)
-	SetEnforceHostNetwork()
+	opConfig := map[string]string{}
+	SetEnforceHostNetwork(opConfig)
 	assert.False(t, EnforceHostNetwork())
 
 	// test invalid setting
 	var value string = "foo"
 	logger.Infof("testing invalid value'%v' for %v", value, enforceHostNetworkSettingName)
-	os.Setenv(enforceHostNetworkSettingName, value)
-	SetEnforceHostNetwork()
+	opConfig[enforceHostNetworkSettingName] = value
+	SetEnforceHostNetwork(opConfig)
 	assert.False(t, EnforceHostNetwork())
 
 	// test valid settings
 	value = "true"
 	logger.Infof("testing valid value'%v' for %v", value, enforceHostNetworkSettingName)
-	os.Setenv(enforceHostNetworkSettingName, value)
-	SetEnforceHostNetwork()
+	opConfig[enforceHostNetworkSettingName] = value
+	SetEnforceHostNetwork(opConfig)
 	assert.True(t, EnforceHostNetwork())
 
 	value = "false"
 	logger.Infof("testing valid value'%v' for %v", value, enforceHostNetworkSettingName)
-	os.Setenv(enforceHostNetworkSettingName, value)
-	SetEnforceHostNetwork()
+	opConfig[enforceHostNetworkSettingName] = value
+	SetEnforceHostNetwork(opConfig)
 	assert.False(t, EnforceHostNetwork())
 }
 
 func TestSetRevisionHistoryLimit(t *testing.T) {
-	defer os.Unsetenv(revisionHistoryLimitSettingName)
-
+	opConfig := map[string]string{}
 	t.Run("ROOK_REVISION_HISTORY_LIMIT: test default value", func(t *testing.T) {
-		SetRevisionHistoryLimit()
+		SetRevisionHistoryLimit(opConfig)
 		assert.Nil(t, RevisionHistoryLimit())
 	})
 
 	var value string = "foo"
 	t.Run("ROOK_REVISION_HISTORY_LIMIT: test invalid value 'foo'", func(t *testing.T) {
-		os.Setenv(revisionHistoryLimitSettingName, value)
-		SetRevisionHistoryLimit()
+		opConfig[revisionHistoryLimitSettingName] = value
+		SetRevisionHistoryLimit(opConfig)
 		assert.Nil(t, RevisionHistoryLimit())
 	})
 
 	t.Run("ROOK_REVISION_HISTORY_LIMIT: test empty string value", func(t *testing.T) {
 		value = ""
-		os.Setenv(revisionHistoryLimitSettingName, value)
-		SetRevisionHistoryLimit()
+		opConfig[revisionHistoryLimitSettingName] = value
+		SetRevisionHistoryLimit(opConfig)
 		assert.Nil(t, RevisionHistoryLimit())
 	})
 	t.Run("ROOK_REVISION_HISTORY_LIMIT:  test valig value '10'", func(t *testing.T) {
 		value = "10"
-		os.Setenv(revisionHistoryLimitSettingName, value)
-		SetRevisionHistoryLimit()
+		opConfig[revisionHistoryLimitSettingName] = value
+		SetRevisionHistoryLimit(opConfig)
 		assert.Equal(t, int32(10), *RevisionHistoryLimit())
 	})
 }
@@ -261,39 +249,36 @@ func TestIsReadyToReconcile(t *testing.T) {
 
 func TestObcAllowAdditionalConfigFields(t *testing.T) {
 	tests := []struct {
-		name             string
-		additionalConfig string
-		shouldAllow      []string
-		shouldDisallow   []string
+		name           string
+		data           map[string]string
+		shouldAllow    []string
+		shouldDisallow []string
 	}{
-		{"not set", "<notset>",
+		{"not set", map[string]string{},
 			[]string{"maxObjects", "maxSize"}, // default allowlist is unlikely to need changing EVER
 			[]string{"bucketMaxObjects", "bucketMaxSize", "bucketPolicy", "bucketLifecycle", "bucketOwner", "random"}},
-		{"set to empty", "",
+		{"set to empty", map[string]string{"ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS": ""},
 			[]string{}, // admin can allow no quota options if desired
 			[]string{"maxObjects", "maxSize", "bucketMaxObjects", "bucketMaxSize", "bucketPolicy", "bucketLifecycle", "bucketOwner", "random"}},
-		{"set to default", "maxObjects,maxSize",
+		{"set to default", map[string]string{"ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS": "maxObjects,maxSize"},
 			[]string{"maxObjects", "maxSize"},
 			[]string{"bucketMaxObjects", "bucketMaxSize", "bucketPolicy", "bucketLifecycle", "bucketOwner", "random"}},
-		{"all quota fields", "maxObjects,maxSize,bucketMaxObjects,bucketMaxSize",
+		{"all quota fields", map[string]string{"ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS": "maxObjects,maxSize,bucketMaxObjects,bucketMaxSize"},
 			[]string{"maxObjects", "maxSize", "bucketMaxObjects", "bucketMaxSize"},
 			[]string{"bucketPolicy", "bucketLifecycle", "bucketOwner", "random"}},
-		{"all fields", "maxObjects,maxSize,bucketMaxObjects,bucketMaxSize,bucketPolicy,bucketLifecycle,bucketOwner",
+		{"all fields", map[string]string{"ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS": "maxObjects,maxSize,bucketMaxObjects,bucketMaxSize,bucketPolicy,bucketLifecycle,bucketOwner"},
 			[]string{"maxObjects", "maxSize", "bucketMaxObjects", "bucketMaxSize", "bucketPolicy", "bucketLifecycle", "bucketOwner"},
 			[]string{"random"}},
 		// this mechanism doesn't do any field checking - that isn't it's job - it merely handles
 		// allow-listing essentially arbitrary config keys
-		{"all fields including unknown", "maxObjects,maxSize,bucketMaxObjects,bucketMaxSize,bucketPolicy,bucketLifecycle,bucketOwner,random",
+		{"all fields including unknown", map[string]string{"ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS": "maxObjects,maxSize,bucketMaxObjects,bucketMaxSize,bucketPolicy,bucketLifecycle,bucketOwner,random"},
 			[]string{"maxObjects", "maxSize", "bucketMaxObjects", "bucketMaxSize", "bucketPolicy", "bucketLifecycle", "bucketOwner", "random"},
 			[]string{"otherRandom"}},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if tt.additionalConfig != "<notset>" {
-				os.Setenv("ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS", tt.additionalConfig)
-				defer os.Unsetenv("ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS")
-			}
-			SetObcAllowAdditionalConfigFields()
+			SetObcAllowAdditionalConfigFields(tt.data)
+
 			for _, toAllow := range tt.shouldAllow {
 				assert.True(t, ObcAdditionalConfigKeyIsAllowed(toAllow))
 			}
diff --git a/pkg/operator/ceph/controller_test.go b/pkg/operator/ceph/controller_test.go
index a9f08b672..ed4485611 100644
--- a/pkg/operator/ceph/controller_test.go
+++ b/pkg/operator/ceph/controller_test.go
@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"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"
@@ -30,7 +31,9 @@ import (
 	"github.com/rook/rook/pkg/operator/test"
 	"github.com/rook/rook/pkg/util/exec"
 	"github.com/stretchr/testify/assert"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
@@ -61,6 +64,7 @@ func TestOperatorController(t *testing.T) {
 
 		// Register operator types with the runtime scheme.
 		s := scheme.Scheme
+		s.AddKnownTypes(cephv1.SchemeGroupVersion, &v1.ConfigMap{}, &v1.ConfigMapList{})
 
 		// Create a fake client to mock API calls.
 		cl := fake.NewClientBuilder().WithScheme(s).Build()
@@ -91,6 +95,7 @@ func TestOperatorController(t *testing.T) {
 
 		// Register operator types with the runtime scheme.
 		s := scheme.Scheme
+		s.AddKnownTypes(cephv1.SchemeGroupVersion, &v1.ConfigMap{}, &v1.ConfigMapList{})
 
 		// Create a fake client to mock API calls.
 		cl := fake.NewClientBuilder().WithScheme(s).Build()
@@ -123,12 +128,23 @@ func TestOperatorController(t *testing.T) {
 		// Register operator types with the runtime scheme.
 		// s := scheme.Scheme
 		s := clientgoscheme.Scheme
+		// s.AddKnownTypes(cephv1.SchemeGroupVersion, &v1.ConfigMap{})
 
-		os.Setenv("ROOK_CEPH_COMMANDS_TIMEOUT_SECONDS", "20")
-		defer os.Unsetenv("ROOK_CEPH_COMMANDS_TIMEOUT_SECONDS")
+		opConfigCM := &v1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      controller.OperatorSettingConfigMapName,
+				Namespace: namespace,
+			},
+			Data: map[string]string{
+				"ROOK_CEPH_COMMANDS_TIMEOUT_SECONDS": "20",
+			},
+		}
 
+		object := []runtime.Object{
+			opConfigCM,
+		}
 		// Create a fake client to mock API calls.
-		cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects().Build()
+		cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
 
 		// Create a ReconcileCSI object with the scheme and fake client.
 		r := &ReconcileConfig{
@@ -157,6 +173,7 @@ func TestOperatorController(t *testing.T) {
 
 		// Register operator types with the runtime scheme.
 		s := scheme.Scheme
+		s.AddKnownTypes(cephv1.SchemeGroupVersion, &v1.ConfigMap{}, &v1.ConfigMapList{})
 
 		// Create a fake client to mock API calls.
 		cl := fake.NewClientBuilder().WithScheme(s).Build()
@@ -192,12 +209,23 @@ func TestOperatorController(t *testing.T) {
 		// Register operator types with the runtime scheme.
 		// s := scheme.Scheme
 		s := clientgoscheme.Scheme
+		// s.AddKnownTypes(cephv1.SchemeGroupVersion, &v1.ConfigMap{})
 
-		os.Setenv("ROOK_CEPH_ALLOW_LOOP_DEVICES", "true")
-		defer os.Unsetenv("ROOK_CEPH_ALLOW_LOOP_DEVICES")
+		opConfigCM := &v1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      controller.OperatorSettingConfigMapName,
+				Namespace: namespace,
+			},
+			Data: map[string]string{
+				"ROOK_CEPH_ALLOW_LOOP_DEVICES": "true",
+			},
+		}
 
+		object := []runtime.Object{
+			opConfigCM,
+		}
 		// Create a fake client to mock API calls.
-		cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects().Build()
+		cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
 
 		// Create a ReconcileCSI object with the scheme and fake client.
 		r := &ReconcileConfig{
diff --git a/pkg/operator/ceph/csi/util_test.go b/pkg/operator/ceph/csi/util_test.go
index 74860696e..29f5cbced 100644
--- a/pkg/operator/ceph/csi/util_test.go
+++ b/pkg/operator/ceph/csi/util_test.go
@@ -47,27 +47,28 @@ func TestDeploymentTemplate(t *testing.T) {
 func TestGetPortFromConfig(t *testing.T) {
 	key := "TEST_CSI_PORT_ENV"
 	var defaultPort uint16 = 8000
+	data := map[string]string{}
 
 	// empty env variable
-	port, err := getPortFromConfig(key, defaultPort)
+	port, err := getPortFromConfig(data, key, defaultPort)
 	assert.Nil(t, err)
 	assert.Equal(t, port, defaultPort)
 
 	// valid port is set in env
 	t.Setenv(key, "9000")
-	port, err = getPortFromConfig(key, defaultPort)
+	port, err = getPortFromConfig(data, key, defaultPort)
 	assert.Nil(t, err)
 	assert.Equal(t, port, uint16(9000))
 
 	// higher port value is set in env
 	t.Setenv(key, "65536")
-	port, err = getPortFromConfig(key, defaultPort)
+	port, err = getPortFromConfig(data, key, defaultPort)
 	assert.Error(t, err)
 	assert.Equal(t, port, defaultPort)
 
 	// negative port is set in env
 	t.Setenv(key, "-1")
-	port, err = getPortFromConfig(key, defaultPort)
+	port, err = getPortFromConfig(data, key, defaultPort)
 	assert.Error(t, err)
 	assert.Equal(t, port, defaultPort)
 
@@ -79,6 +80,7 @@ func TestApplyingResourcesToRBDPlugin(t *testing.T) {
 	tp := templateParam{}
 	rbdPlugin, err := templateToDaemonSet("rbdplugin", RBDPluginTemplatePath, tp)
 	assert.Nil(t, err)
+	params := make(map[string]string)
 
 	// need to build using map[string]interface{} because the following resource
 	// doesn't serialise nicely
@@ -101,9 +103,8 @@ func TestApplyingResourcesToRBDPlugin(t *testing.T) {
 
 	resourceRaw, err := yaml.Marshal(resource)
 	assert.Nil(t, err)
-	os.Setenv(rbdPluginResource, string(resourceRaw))
-	defer os.Unsetenv(rbdPluginResource)
-	applyResourcesToContainers(rbdPluginResource, &rbdPlugin.Spec.Template.Spec)
+	params[rbdPluginResource] = string(resourceRaw)
+	applyResourcesToContainers(params, rbdPluginResource, &rbdPlugin.Spec.Template.Spec)
 	assert.Equal(t, rbdPlugin.Spec.Template.Spec.Containers[0].Resources.Requests.Memory().String(), "128Mi")
 	assert.Equal(t, rbdPlugin.Spec.Template.Spec.Containers[0].Resources.Limits.Memory().String(), "256Mi")
 	assert.Equal(t, rbdPlugin.Spec.Template.Spec.Containers[0].Resources.Requests.Cpu().String(), "100m")
@@ -112,6 +113,7 @@ func TestApplyingResourcesToRBDPlugin(t *testing.T) {
 
 func Test_applyVolumeToPodSpec(t *testing.T) {
 	// when no volumes specified
+	config := make(map[string]string)
 	configKey := "TEST_CSI_PLUGIN_VOLUME"
 	dsName := "test-ds"
 	tp := templateParam{
@@ -122,7 +124,7 @@ func Test_applyVolumeToPodSpec(t *testing.T) {
 	defaultVolumes := 11
 	ds, err := templateToDaemonSet(dsName, RBDPluginTemplatePath, tp)
 	assert.Nil(t, err)
-	applyVolumeToPodSpec(configKey, &ds.Spec.Template.Spec)
+	applyVolumeToPodSpec(config, configKey, &ds.Spec.Template.Spec)
 
 	assert.Len(t, ds.Spec.Template.Spec.Volumes, defaultVolumes)
 
@@ -130,7 +132,7 @@ func Test_applyVolumeToPodSpec(t *testing.T) {
 	tp.CSILogRotation = true
 	ds, err = templateToDaemonSet(dsName, RBDPluginTemplatePath, tp)
 	assert.Nil(t, err)
-	applyVolumeToPodSpec(configKey, &ds.Spec.Template.Spec)
+	applyVolumeToPodSpec(config, configKey, &ds.Spec.Template.Spec)
 	assert.Len(t, ds.Spec.Template.Spec.Volumes, defaultVolumes+2)
 	tp.CSILogRotation = false
 
@@ -143,11 +145,10 @@ func Test_applyVolumeToPodSpec(t *testing.T) {
 	}
 	volumeRaw, err := yaml.Marshal(volumes)
 	assert.Nil(t, err)
-	os.Setenv(configKey, string(volumeRaw))
-	defer os.Unsetenv(configKey)
+	config[configKey] = string(volumeRaw)
 	ds, err = templateToDaemonSet(dsName, RBDPluginTemplatePath, tp)
 	assert.Nil(t, err)
-	applyVolumeToPodSpec(configKey, &ds.Spec.Template.Spec)
+	applyVolumeToPodSpec(config, configKey, &ds.Spec.Template.Spec)
 
 	assert.Len(t, ds.Spec.Template.Spec.Volumes, defaultVolumes+1)
 	// add one more volume
@@ -157,10 +158,10 @@ func Test_applyVolumeToPodSpec(t *testing.T) {
 	})
 	volumeRaw, err = yaml.Marshal(volumes)
 	assert.Nil(t, err)
-	os.Setenv(configKey, string(volumeRaw))
+	config[configKey] = string(volumeRaw)
 	ds, err = templateToDaemonSet(dsName, RBDPluginTemplatePath, tp)
 	assert.Nil(t, err)
-	applyVolumeToPodSpec(configKey, &ds.Spec.Template.Spec)
+	applyVolumeToPodSpec(config, configKey, &ds.Spec.Template.Spec)
 	assert.Len(t, ds.Spec.Template.Spec.Volumes, defaultVolumes+2)
 	// override existing volume configuration
 	volumes[1].VolumeSource = corev1.VolumeSource{
@@ -168,10 +169,10 @@ func Test_applyVolumeToPodSpec(t *testing.T) {
 	}
 	volumeRaw, err = yaml.Marshal(volumes)
 	assert.Nil(t, err)
-	os.Setenv(configKey, string(volumeRaw))
+	config[configKey] = string(volumeRaw)
 	ds, err = templateToDaemonSet(dsName, RBDPluginTemplatePath, tp)
 	assert.Nil(t, err)
-	applyVolumeToPodSpec(configKey, &ds.Spec.Template.Spec)
+	applyVolumeToPodSpec(config, configKey, &ds.Spec.Template.Spec)
 	assert.Len(t, ds.Spec.Template.Spec.Volumes, defaultVolumes+2)
 	// remove existing volume configuration
 	volumes = []corev1.Volume{
@@ -182,15 +183,16 @@ func Test_applyVolumeToPodSpec(t *testing.T) {
 	}
 	volumeRaw, err = yaml.Marshal(volumes)
 	assert.Nil(t, err)
-	os.Setenv(configKey, string(volumeRaw))
+	config[configKey] = string(volumeRaw)
 	ds, err = templateToDaemonSet(dsName, RBDPluginTemplatePath, tp)
 	assert.Nil(t, err)
-	applyVolumeToPodSpec(configKey, &ds.Spec.Template.Spec)
+	applyVolumeToPodSpec(config, configKey, &ds.Spec.Template.Spec)
 	assert.Len(t, ds.Spec.Template.Spec.Volumes, defaultVolumes+1)
 }
 
 func Test_applyVolumeMountToContainer(t *testing.T) {
 	// when no volumes specified
+	config := make(map[string]string)
 	configKey := "TEST_CSI_PLUGIN_VOLUME_MOUNT"
 	dsName := "test-ds"
 	rbdContainerName := "csi-rbdplugin"
@@ -202,7 +204,7 @@ func Test_applyVolumeMountToContainer(t *testing.T) {
 	defaultVolumes := 10
 	ds, err := templateToDaemonSet(dsName, RBDPluginTemplatePath, tp)
 	assert.Nil(t, err)
-	applyVolumeMountToContainer(configKey, rbdContainerName, &ds.Spec.Template.Spec)
+	applyVolumeMountToContainer(config, configKey, rbdContainerName, &ds.Spec.Template.Spec)
 
 	assert.Len(t, ds.Spec.Template.Spec.Containers[1].VolumeMounts, defaultVolumes)
 
@@ -210,7 +212,7 @@ func Test_applyVolumeMountToContainer(t *testing.T) {
 	tp.CSILogRotation = true
 	ds, err = templateToDaemonSet(dsName, RBDPluginTemplatePath, tp)
 	assert.Nil(t, err)
-	applyVolumeMountToContainer(configKey, rbdContainerName, &ds.Spec.Template.Spec)
+	applyVolumeMountToContainer(config, configKey, rbdContainerName, &ds.Spec.Template.Spec)
 	assert.Len(t, ds.Spec.Template.Spec.Containers[1].VolumeMounts, defaultVolumes+1)
 	tp.CSILogRotation = false
 
@@ -223,11 +225,10 @@ func Test_applyVolumeMountToContainer(t *testing.T) {
 	}
 	volumeMountsRaw, err := yaml.Marshal(volumeMounts)
 	assert.Nil(t, err)
-	os.Setenv(configKey, string(volumeMountsRaw))
-	defer os.Unsetenv(configKey)
+	config[configKey] = string(volumeMountsRaw)
 	ds, err = templateToDaemonSet(dsName, RBDPluginTemplatePath, tp)
 	assert.Nil(t, err)
-	applyVolumeMountToContainer(configKey, rbdContainerName, &ds.Spec.Template.Spec)
+	applyVolumeMountToContainer(config, configKey, rbdContainerName, &ds.Spec.Template.Spec)
 
 	assert.Len(t, ds.Spec.Template.Spec.Containers[1].VolumeMounts, defaultVolumes+1)
 	// add one more volumemount
@@ -237,19 +238,19 @@ func Test_applyVolumeMountToContainer(t *testing.T) {
 	})
 	volumeMountsRaw, err = yaml.Marshal(volumeMounts)
 	assert.Nil(t, err)
-	os.Setenv(configKey, string(volumeMountsRaw))
+	config[configKey] = string(volumeMountsRaw)
 	ds, err = templateToDaemonSet(dsName, RBDPluginTemplatePath, tp)
 	assert.Nil(t, err)
-	applyVolumeMountToContainer(configKey, rbdContainerName, &ds.Spec.Template.Spec)
+	applyVolumeMountToContainer(config, configKey, rbdContainerName, &ds.Spec.Template.Spec)
 	assert.Len(t, ds.Spec.Template.Spec.Containers[1].VolumeMounts, defaultVolumes+2)
 	// override existing volume configuration
 	volumeMounts[1].MountPath = "/run/test/run"
 	volumeMountsRaw, err = yaml.Marshal(volumeMounts)
 	assert.Nil(t, err)
-	os.Setenv(configKey, string(volumeMountsRaw))
+	config[configKey] = string(volumeMountsRaw)
 	ds, err = templateToDaemonSet(dsName, RBDPluginTemplatePath, tp)
 	assert.Nil(t, err)
-	applyVolumeMountToContainer(configKey, rbdContainerName, &ds.Spec.Template.Spec)
+	applyVolumeMountToContainer(config, configKey, rbdContainerName, &ds.Spec.Template.Spec)
 	assert.Len(t, ds.Spec.Template.Spec.Containers[1].VolumeMounts, defaultVolumes+2)
 	// remove existing volume configuration
 	volumeMounts = []corev1.VolumeMount{
@@ -260,16 +261,17 @@ func Test_applyVolumeMountToContainer(t *testing.T) {
 	}
 	volumeMountsRaw, err = yaml.Marshal(volumeMounts)
 	assert.Nil(t, err)
-	os.Setenv(configKey, string(volumeMountsRaw))
+	config[configKey] = string(volumeMountsRaw)
 	ds, err = templateToDaemonSet(dsName, RBDPluginTemplatePath, tp)
 	assert.Nil(t, err)
-	applyVolumeMountToContainer(configKey, rbdContainerName, &ds.Spec.Template.Spec)
+	applyVolumeMountToContainer(config, configKey, rbdContainerName, &ds.Spec.Template.Spec)
 	assert.Len(t, ds.Spec.Template.Spec.Containers[1].VolumeMounts, defaultVolumes+1)
 }
 
 func Test_getImage(t *testing.T) {
 	type args struct {
-		userImage    string
+		data         map[string]string
+		settingName  string
 		defaultImage string
 	}
 	tests := []struct {
@@ -280,6 +282,8 @@ func Test_getImage(t *testing.T) {
 		{
 			name: "test with default image",
 			args: args{
+				data:         map[string]string{},
+				settingName:  "ROOK_CSI_CEPH_IMAGE",
 				defaultImage: "quay.io/cephcsi/cephcsi:v3.13.0",
 			},
 			want: DefaultCSIPluginImage,
@@ -287,7 +291,10 @@ func Test_getImage(t *testing.T) {
 		{
 			name: "test with user image",
 			args: args{
-				userImage:    "registry.io/private/cephcsi:v8",
+				data: map[string]string{
+					"ROOK_CSI_CEPH_IMAGE": "registry.io/private/cephcsi:v8",
+				},
+				settingName:  "ROOK_CSI_CEPH_IMAGE",
 				defaultImage: "quay.io/cephcsi/cephcsi:v3.11.0",
 			},
 			want: "registry.io/private/cephcsi:v8",
@@ -295,7 +302,10 @@ func Test_getImage(t *testing.T) {
 		{
 			name: "test with user image without version",
 			args: args{
-				userImage:    "registry.io/private/cephcsi",
+				data: map[string]string{
+					"ROOK_CSI_CEPH_IMAGE": "registry.io/private/cephcsi",
+				},
+				settingName:  "ROOK_CSI_CEPH_IMAGE",
 				defaultImage: "quay.io/cephcsi/cephcsi:v3.11.0",
 			},
 			want: "registry.io/private/cephcsi:v3.11.0",
@@ -303,11 +313,7 @@ func Test_getImage(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if tt.args.userImage != "" {
-				os.Setenv("ROOK_CSI_CEPH_IMAGE", tt.args.userImage)
-				defer os.Unsetenv("ROOK_CSI_CEPH_IMAGE")
-			}
-			got := getImage("ROOK_CSI_CEPH_IMAGE", tt.args.defaultImage)
+			got := getImage(tt.args.data, tt.args.settingName, tt.args.defaultImage)
 			assert.Equal(t, tt.want, got)
 		})
 	}
diff --git a/pkg/operator/ceph/object/bucket/provisioner_test.go b/pkg/operator/ceph/object/bucket/provisioner_test.go
index 0e44132a4..a098b51f9 100644
--- a/pkg/operator/ceph/object/bucket/provisioner_test.go
+++ b/pkg/operator/ceph/object/bucket/provisioner_test.go
@@ -22,7 +22,6 @@ import (
 	"fmt"
 	"io"
 	"net/http"
-	"os"
 	"strings"
 	"testing"
 
@@ -502,7 +501,7 @@ func TestProvisioner_additionalConfigSpecFromMap(t *testing.T) {
 	})
 
 	t.Run("maxObjects field should be set", func(t *testing.T) {
-		opcontroller.SetObcAllowAdditionalConfigFields()
+		opcontroller.SetObcAllowAdditionalConfigFields(map[string]string{})
 
 		spec, err := additionalConfigSpecFromMap(map[string]string{"maxObjects": "2"})
 		assert.NoError(t, err)
@@ -511,7 +510,7 @@ func TestProvisioner_additionalConfigSpecFromMap(t *testing.T) {
 	})
 
 	t.Run("maxSize field should be set", func(t *testing.T) {
-		opcontroller.SetObcAllowAdditionalConfigFields()
+		opcontroller.SetObcAllowAdditionalConfigFields(map[string]string{})
 
 		spec, err := additionalConfigSpecFromMap(map[string]string{"maxSize": "3"})
 		assert.NoError(t, err)
@@ -520,10 +519,8 @@ func TestProvisioner_additionalConfigSpecFromMap(t *testing.T) {
 	})
 
 	t.Run("bucketMaxObjects field should be set", func(t *testing.T) {
-		os.Setenv("ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS", "bucketMaxObjects")
-		defer os.Unsetenv("ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS")
-		opcontroller.SetObcAllowAdditionalConfigFields()
-		defer opcontroller.SetObcAllowAdditionalConfigFields()
+		opcontroller.SetObcAllowAdditionalConfigFields(map[string]string{"ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS": "bucketMaxObjects"})
+		defer opcontroller.SetObcAllowAdditionalConfigFields(map[string]string{})
 
 		spec, err := additionalConfigSpecFromMap(map[string]string{"bucketMaxObjects": "4"})
 		assert.NoError(t, err)
@@ -531,10 +528,8 @@ func TestProvisioner_additionalConfigSpecFromMap(t *testing.T) {
 	})
 
 	t.Run("bucketMaxSize field should be set", func(t *testing.T) {
-		os.Setenv("ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS", "bucketMaxSize")
-		defer os.Unsetenv("ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS")
-		opcontroller.SetObcAllowAdditionalConfigFields()
-		defer opcontroller.SetObcAllowAdditionalConfigFields()
+		opcontroller.SetObcAllowAdditionalConfigFields(map[string]string{"ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS": "bucketMaxSize"})
+		defer opcontroller.SetObcAllowAdditionalConfigFields(map[string]string{})
 
 		spec, err := additionalConfigSpecFromMap(map[string]string{"bucketMaxSize": "5"})
 		assert.NoError(t, err)
@@ -542,10 +537,8 @@ func TestProvisioner_additionalConfigSpecFromMap(t *testing.T) {
 	})
 
 	t.Run("bucketPolicy field should be set", func(t *testing.T) {
-		os.Setenv("ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS", "bucketPolicy")
-		defer os.Unsetenv("ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS")
-		opcontroller.SetObcAllowAdditionalConfigFields()
-		defer opcontroller.SetObcAllowAdditionalConfigFields()
+		opcontroller.SetObcAllowAdditionalConfigFields(map[string]string{"ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS": "bucketPolicy"})
+		defer opcontroller.SetObcAllowAdditionalConfigFields(map[string]string{})
 
 		spec, err := additionalConfigSpecFromMap(map[string]string{"bucketPolicy": "foo"})
 		assert.NoError(t, err)
@@ -553,10 +546,8 @@ func TestProvisioner_additionalConfigSpecFromMap(t *testing.T) {
 	})
 
 	t.Run("bucketLifecycle field should be set", func(t *testing.T) {
-		os.Setenv("ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS", "bucketLifecycle")
-		defer os.Unsetenv("ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS")
-		opcontroller.SetObcAllowAdditionalConfigFields()
-		defer opcontroller.SetObcAllowAdditionalConfigFields()
+		opcontroller.SetObcAllowAdditionalConfigFields(map[string]string{"ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS": "bucketLifecycle"})
+		defer opcontroller.SetObcAllowAdditionalConfigFields(map[string]string{})
 
 		spec, err := additionalConfigSpecFromMap(map[string]string{"bucketLifecycle": "foo"})
 		assert.NoError(t, err)
@@ -564,10 +555,8 @@ func TestProvisioner_additionalConfigSpecFromMap(t *testing.T) {
 	})
 
 	t.Run("bucketOwner field should be set", func(t *testing.T) {
-		os.Setenv("ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS", "bucketOwner")
-		defer os.Unsetenv("ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS")
-		opcontroller.SetObcAllowAdditionalConfigFields()
-		defer opcontroller.SetObcAllowAdditionalConfigFields()
+		opcontroller.SetObcAllowAdditionalConfigFields(map[string]string{"ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS": "bucketOwner"})
+		defer opcontroller.SetObcAllowAdditionalConfigFields(map[string]string{})
 
 		spec, err := additionalConfigSpecFromMap(map[string]string{"bucketOwner": "foo"})
 		assert.NoError(t, err)
@@ -575,7 +564,7 @@ func TestProvisioner_additionalConfigSpecFromMap(t *testing.T) {
 	})
 
 	t.Run("fields disallowed by default", func(t *testing.T) {
-		opcontroller.SetObcAllowAdditionalConfigFields()
+		opcontroller.SetObcAllowAdditionalConfigFields(map[string]string{})
 
 		for _, configKey := range []string{"bucketMaxObjects", "bucketMaxSize", "bucketPolicy", "bucketLifecycle", "bucketOwner"} {
 			_, err := additionalConfigSpecFromMap(map[string]string{configKey: "foo"})
@@ -584,7 +573,7 @@ func TestProvisioner_additionalConfigSpecFromMap(t *testing.T) {
 	})
 
 	t.Run("does not fail on empty map", func(t *testing.T) {
-		opcontroller.SetObcAllowAdditionalConfigFields()
+		opcontroller.SetObcAllowAdditionalConfigFields(map[string]string{})
 
 		spec, err := additionalConfigSpecFromMap(map[string]string{})
 		assert.NoError(t, err)
diff --git a/pkg/operator/ceph/object/objectstore_test.go b/pkg/operator/ceph/object/objectstore_test.go
index 8359e9897..f384d6c70 100644
--- a/pkg/operator/ceph/object/objectstore_test.go
+++ b/pkg/operator/ceph/object/objectstore_test.go
@@ -517,43 +517,36 @@ func TestGetObjectBucketProvisioner(t *testing.T) {
 	t.Setenv(k8sutil.PodNamespaceEnvVar, testNamespace)
 
 	t.Run("watch ceph cluster namespace", func(t *testing.T) {
-		os.Setenv("ROOK_OBC_WATCH_OPERATOR_NAMESPACE", "true")
-		defer os.Unsetenv("ROOK_OBC_WATCH_OPERATOR_NAMESPACE")
-		bktprovisioner, err := GetObjectBucketProvisioner(testNamespace)
+		data := map[string]string{"ROOK_OBC_WATCH_OPERATOR_NAMESPACE": "true"}
+		bktprovisioner, err := GetObjectBucketProvisioner(data, testNamespace)
 		assert.Equal(t, fmt.Sprintf("%s.%s", testNamespace, bucketProvisionerName), bktprovisioner)
 		assert.NoError(t, err)
 	})
 
 	t.Run("watch all namespaces", func(t *testing.T) {
-		os.Setenv("ROOK_OBC_WATCH_OPERATOR_NAMESPACE", "false")
-		defer os.Unsetenv("ROOK_OBC_WATCH_OPERATOR_NAMESPACE")
-		bktprovisioner, err := GetObjectBucketProvisioner(testNamespace)
+		data := map[string]string{"ROOK_OBC_WATCH_OPERATOR_NAMESPACE": "false"}
+		bktprovisioner, err := GetObjectBucketProvisioner(data, testNamespace)
 		assert.Equal(t, bucketProvisionerName, bktprovisioner)
 		assert.NoError(t, err)
 	})
 
 	t.Run("prefix object provisioner", func(t *testing.T) {
-		os.Setenv("ROOK_OBC_PROVISIONER_NAME_PREFIX", "my-prefix")
-		defer os.Unsetenv("ROOK_OBC_PROVISIONER_NAME_PREFIX")
-		bktprovisioner, err := GetObjectBucketProvisioner(testNamespace)
+		data := map[string]string{"ROOK_OBC_PROVISIONER_NAME_PREFIX": "my-prefix"}
+		bktprovisioner, err := GetObjectBucketProvisioner(data, testNamespace)
 		assert.Equal(t, "my-prefix."+bucketProvisionerName, bktprovisioner)
 		assert.NoError(t, err)
 	})
 
 	t.Run("watch ceph cluster namespace and prefix object provisioner", func(t *testing.T) {
-		os.Setenv("ROOK_OBC_WATCH_OPERATOR_NAMESPACE", "true")
-		os.Setenv("ROOK_OBC_PROVISIONER_NAME_PREFIX", "my-prefix")
-		defer os.Unsetenv("ROOK_OBC_WATCH_OPERATOR_NAMESPACE")
-		defer os.Unsetenv("ROOK_OBC_PROVISIONER_NAME_PREFIX")
-		bktprovisioner, err := GetObjectBucketProvisioner(testNamespace)
+		data := map[string]string{"ROOK_OBC_WATCH_OPERATOR_NAMESPACE": "true", "ROOK_OBC_PROVISIONER_NAME_PREFIX": "my-prefix"}
+		bktprovisioner, err := GetObjectBucketProvisioner(data, testNamespace)
 		assert.Equal(t, "my-prefix."+bucketProvisionerName, bktprovisioner)
 		assert.NoError(t, err)
 	})
 
 	t.Run("invalid prefix value for object provisioner", func(t *testing.T) {
-		os.Setenv("ROOK_OBC_PROVISIONER_NAME_PREFIX", "my-prefix.")
-		defer os.Unsetenv("ROOK_OBC_PROVISIONER_NAME_PREFIX")
-		_, err := GetObjectBucketProvisioner(testNamespace)
+		data := map[string]string{"ROOK_OBC_PROVISIONER_NAME_PREFIX": "my-prefix."}
+		_, err := GetObjectBucketProvisioner(data, testNamespace)
 		assert.Error(t, err)
 	})
 
diff --git a/pkg/operator/discover/discover_test.go b/pkg/operator/discover/discover_test.go
index ec19cc16e..667e88bdc 100644
--- a/pkg/operator/discover/discover_test.go
+++ b/pkg/operator/discover/discover_test.go
@@ -19,12 +19,12 @@ package discover
 
 import (
 	"context"
-	"os"
 	"testing"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	discoverDaemon "github.com/rook/rook/pkg/daemon/discover"
+	"github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/rook/rook/pkg/operator/test"
 
@@ -63,7 +63,7 @@ func TestStartDiscoveryDaemonset(t *testing.T) {
 	_, err := clientset.CoreV1().Pods("rook-system").Create(ctx, &pod, metav1.CreateOptions{})
 	assert.NoError(t, err)
 	// start a basic cluster
-	err = a.Start(ctx, namespace, "rook/rook:myversion", "mysa", false)
+	err = a.Start(ctx, namespace, "rook/rook:myversion", "mysa", map[string]string{}, false)
 	assert.Nil(t, err)
 
 	// check daemonset parameters
@@ -86,11 +86,21 @@ func TestStartDiscoveryDaemonset(t *testing.T) {
 	assert.Nil(t, agentDS.Spec.Template.Spec.Tolerations)
 
 	// Test with rook override configmap setting
-	os.Setenv("DISCOVER_TOLERATIONS", "- effect: NoSchedule\n  key: node-role.kubernetes.io/control-plane\n  operator: Exists\n- effect: NoExecute\n  key: node-role.kubernetes.io/etcd\n  operator: Exists")
-	defer os.Unsetenv("DISCOVER_TOLERATIONS")
+	opConfigCM := &v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      controller.OperatorSettingConfigMapName,
+			Namespace: namespace,
+		},
+		Data: map[string]string{
+			"DISCOVER_TOLERATIONS": "- effect: NoSchedule\n  key: node-role.kubernetes.io/control-plane\n  operator: Exists\n- effect: NoExecute\n  key: node-role.kubernetes.io/etcd\n  operator: Exists",
+		},
+	}
+
+	cm, err := clientset.CoreV1().ConfigMaps(namespace).Create(ctx, opConfigCM, metav1.CreateOptions{})
+	assert.NoError(t, err)
 
 	// start a basic cluster
-	err = a.Start(ctx, namespace, "rook/rook:myversion", "mysa", false)
+	err = a.Start(ctx, namespace, "rook/rook:myversion", "mysa", cm.Data, false)
 	assert.Nil(t, err)
 
 	agentDS, err = clientset.AppsV1().DaemonSets(namespace).Get(ctx, "rook-discover", metav1.GetOptions{})
diff --git a/pkg/operator/k8sutil/configmap_test.go b/pkg/operator/k8sutil/configmap_test.go
index d4577b0fc..cc03f1fcc 100644
--- a/pkg/operator/k8sutil/configmap_test.go
+++ b/pkg/operator/k8sutil/configmap_test.go
@@ -63,7 +63,7 @@ func TestGetOperatorSetting(t *testing.T) {
 	k8s := fake.NewSimpleClientset()
 	ctx := context.TODO()
 
-	operatorSettingConfigMapName := "rook-ceph-operator-config"
+	operatorSettingConfigMapName := "rook-operator-config"
 	testNamespace := "test-namespace"
 	cm := &v1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
@@ -81,41 +81,63 @@ func TestGetOperatorSetting(t *testing.T) {
 	cmSettingValue := "storage=rook, worker"
 	defaultValue := ""
 
-	// ConfigMap is not yet applied so we expect the default value
-	setting := GetOperatorSetting(nodeAffinity, defaultValue)
-	assert.Equal(t, defaultValue, setting)
+	// ConfigMap is not found
+	setting, err := GetOperatorSetting(ctx, k8s, operatorSettingConfigMapName, nodeAffinity, defaultValue)
+	assert.NoError(t, err)
 
+	// Env Var doesn't exist
+	assert.Equal(t, defaultValue, setting)
 	// Env Var exists
 	t.Setenv(nodeAffinity, envSettingValue)
-	setting = GetOperatorSetting(nodeAffinity, defaultValue)
+	setting, err = GetOperatorSetting(ctx, k8s, operatorSettingConfigMapName, nodeAffinity, defaultValue)
+	assert.NoError(t, err)
 	assert.Equal(t, envSettingValue, setting)
 
 	// ConfigMap is found
 	t.Setenv("POD_NAMESPACE", testNamespace)
-	_, err := k8s.CoreV1().ConfigMaps(testNamespace).Create(ctx, cm, metav1.CreateOptions{})
-	assert.NoError(t, err)
-	err = ApplyOperatorSettingsConfigmap(ctx, k8s)
+	_, err = k8s.CoreV1().ConfigMaps(testNamespace).Create(ctx, cm, metav1.CreateOptions{})
 	assert.NoError(t, err)
 
 	// Setting exists in ConfigMap
-	setting = GetOperatorSetting(nodeAffinity, defaultValue)
+	setting, err = GetOperatorSetting(ctx, k8s, operatorSettingConfigMapName, nodeAffinity, defaultValue)
+	assert.NoError(t, err)
+	// Env Var exists
 	assert.Equal(t, cmSettingValue, setting)
-
 	// Env Var doesn't exist
 	err = os.Unsetenv(nodeAffinity)
 	assert.NoError(t, err)
 	assert.Equal(t, cmSettingValue, setting)
 
-	// Setting doesn't exist in ConfigMap, expect the default value
-	setting = GetOperatorSetting(podAffinity, defaultValue)
+	// Setting doesn't exist in ConfigMap
+	setting, err = GetOperatorSetting(ctx, k8s, operatorSettingConfigMapName, podAffinity, defaultValue)
+	assert.NoError(t, err)
+	// Env Var doesn't exist
 	assert.Equal(t, defaultValue, setting)
 	// Env Var exists
 	t.Setenv(podAffinity, envSettingValue)
-	setting = GetOperatorSetting(podAffinity, defaultValue)
+	setting, err = GetOperatorSetting(ctx, k8s, operatorSettingConfigMapName, podAffinity, defaultValue)
 	assert.NoError(t, err)
 	assert.Equal(t, envSettingValue, setting)
 }
 
+func TestLogChangedSettings(t *testing.T) {
+	key := "Node_Affinity"
+	value := "storage=rook, worker"
+
+	// When the key is not in the map, it should be added and return true for logging
+	ok := logChangedSettings(key, value)
+	assert.True(t, true, ok)
+
+	// When the key is already present in the map and value is same, it should return false for logging
+	ok = logChangedSettings(key, value)
+	assert.False(t, false, ok)
+
+	// When the key is already present in the map and value is updated, it should return true for logging
+	value = "storage=rook, test"
+	ok = logChangedSettings(key, value)
+	assert.True(t, true, ok)
+}
+
 func TestCreateOrUpdateConfigMap(t *testing.T) {
 	k8s := fake.NewSimpleClientset()
 	ctx := context.TODO()