Repository: rook/rook
Pull request #6653 (2020-11-17 17:42:02)

ceph: support ceph cluster and CSI on multus in different namespace (bp #6396)
This is an automated backport of pull request #6396 done by Mergify
---


<details>
<summary>Mergify commands and options</summary>

<br />

More conditions and actions can be found in the [documentation](https://docs.mergify.io/).

You can also trigger Mergify actions by commenting on this pull request:

- `@Mergifyio refresh` will re-evaluate the rules
- `@Mergifyio rebase` will rebase this PR on its base branch
- `@Mergifyio update` will merge the base branch into this PR
- `@Mergifyio backport <destination>` will backport this PR on `<destination>` branch

Additionally, on Mergify [dashboard](https://dashboard.mergify.io/) you can:

- look at your merge queues
- generate the Mergify configuration with the config editor.

Finally, you can contact us on https://mergify.io/
</details>



Patch:
diff --git a/Documentation/ceph-cluster-crd.md b/Documentation/ceph-cluster-crd.md
index 4a6a25a1f..db652eab3 100755
--- a/Documentation/ceph-cluster-crd.md
+++ b/Documentation/ceph-cluster-crd.md
@@ -304,7 +304,9 @@ The selector keys are required to be `public` and `cluster` where each represent
 * `public`: client communications with the cluster (reads/writes)
 * `cluster`: internal Ceph replication network
 
-If you want to learn more, please read [Ceph Networking reference](https://docs.ceph.com/docs/master/rados/configuration/network-config-ref/).
+If you want to learn more, please read
+* [Ceph Networking reference](https://docs.ceph.com/docs/master/rados/configuration/network-config-ref/).
+* [Multus documentation](https://intel.github.io/multus-cni/doc/how-to-use.html)
 
 Based on the configuration, the operator will do the following:
 
@@ -314,14 +316,47 @@ Based on the configuration, the operator will do the following:
 In order to work, each selector value must match a `NetworkAttachmentDefinition` object name in Multus.
 For example, you can do:
 
-* `public`: "my-public-storage-network"
-* `cluster`: "my-replication-storage-network"
+* `public`: "rook-ceph/my-public-storage-network"
+* `cluster`: "rook-ceph/my-replication-storage-network"
 
 For `multus` network provider, an already working cluster with Multus networking is required. Network attachment definition that later will be attached to the cluster needs to be created before the Cluster CRD.
 The Network attachment definitions should be using whereabouts cni.
 If Rook cannot find the provided Network attachment definition it will fail running the Ceph OSD pods.
 You can add the Multus network attachment selection annotation selecting the created network attachment definition on `selectors`.
 
+A valid NetworkAttachmentDefinition will look like following:
+
+```yaml
+apiVersion: "k8s.cni.cncf.io/v1"
+kind: NetworkAttachmentDefinition
+metadata:
+  name: rook-public-nw
+spec:
+  config: '{
+      "cniVersion": "0.3.0",
+      "name": "public-nad",
+      "type": "macvlan",
+      "master": "ens5",
+      "mode": "bridge",
+      "ipam": {
+        "type": "whereabouts",
+        "range": "192.168.1.0/24"
+      }
+    }'
+```
+
+* Ensure that `master` matches the network interface of the host that you want to use.
+* The NAD should be referenced along with the namespace in which it is present like `public: <namespace>/<name of NAD>`.
+  e.g., the network attachment definition are in `rook-multus` namespace:
+
+```yaml
+  public: rook-multus/rook-public-nw
+  cluster: rook-multus/rook-cluster-nw
+```
+
+This is required in order to use the NAD across namespaces.
+* In Openshift, to use the NetworkAttachmentDefinition across namespaces, the NAD must be deployed in the default namespace and it can be referenced as `default/myNAD` where `default` is the namespace and `myNAD` is the network attachment definition.
+
 #### IPFamily
 
 Provide single-stack IPv4 or IPv6 protocol to assign corresponding addresses to pods and services. This field is optional. Possible inputs are IPv6 and IPv4. Empty value will be treated as IPv4. Kubernetes version should be at least v1.13 to run IPv6. Dual-stack is not supported by ceph.
diff --git a/cluster/examples/kubernetes/ceph/common.yaml b/cluster/examples/kubernetes/ceph/common.yaml
index 8114bdd1f..9c68238f1 100644
--- a/cluster/examples/kubernetes/ceph/common.yaml
+++ b/cluster/examples/kubernetes/ceph/common.yaml
@@ -129,12 +129,6 @@ rules:
   - create
   - update
   - delete
-- apiGroups:
-  - k8s.cni.cncf.io
-  resources:
-  - network-attachment-definitions
-  verbs:
-  - get
 ---
 # The cluster role for managing the Rook CRDs
 apiVersion: rbac.authorization.k8s.io/v1
@@ -248,6 +242,12 @@ rules:
   - delete
   - get
   - update
+- apiGroups:
+  - k8s.cni.cncf.io
+  resources:
+  - network-attachment-definitions
+  verbs:
+  - get
 ---
 # Aspects of ceph-mgr that require cluster-wide access
 kind: ClusterRole
diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go
index 23383f1d9..83c00e074 100755
--- a/pkg/operator/ceph/cluster/cluster.go
+++ b/pkg/operator/ceph/cluster/cluster.go
@@ -371,8 +371,13 @@ func (c *ClusterController) preClusterStartValidation(cluster *cluster) error {
 				continue
 			}
 
+			multusNamespace, nad := config.GetMultusNamespace(cluster.Spec.Network.Selectors[selector])
+			if multusNamespace == "" {
+				multusNamespace = cluster.Namespace
+			}
+
 			// Get network attachment definition
-			_, err := c.context.NetworkClient.NetworkAttachmentDefinitions(cluster.Namespace).Get(cluster.Spec.Network.Selectors[selector], metav1.GetOptions{})
+			_, err := c.context.NetworkClient.NetworkAttachmentDefinitions(multusNamespace).Get(nad, metav1.GetOptions{})
 			if err != nil {
 				if kerrors.IsNotFound(err) {
 					return errors.Wrapf(err, "specified network attachment definition for selector %q does not exist", selector)
diff --git a/pkg/operator/ceph/config/network.go b/pkg/operator/ceph/config/network.go
index 97a54709f..cc75bc928 100644
--- a/pkg/operator/ceph/config/network.go
+++ b/pkg/operator/ceph/config/network.go
@@ -19,6 +19,7 @@ package config
 
 import (
 	"fmt"
+	"strings"
 
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/clusterd"
@@ -52,11 +53,15 @@ func generateNetworkSettings(context *clusterd.Context, namespace string, networ
 			continue
 		}
 
+		multusNamespace, nad := GetMultusNamespace(networkSelectors[selectorKey])
+		if multusNamespace == "" {
+			multusNamespace = namespace
+		}
 		// Get network attachment definition
-		netDefinition, err := context.NetworkClient.NetworkAttachmentDefinitions(namespace).Get(networkSelectors[selectorKey], metav1.GetOptions{})
+		netDefinition, err := context.NetworkClient.NetworkAttachmentDefinitions(multusNamespace).Get(nad, metav1.GetOptions{})
 		if err != nil {
 			if kerrors.IsNotFound(err) {
-				return []Option{}, errors.Wrapf(err, "specified network attachment definition for selector %q does not exist", selectorKey)
+				return []Option{}, errors.Wrapf(err, "specified network attachment definition %q in namespace %q for selector %q does not exist", nad, namespace, selectorKey)
 			}
 			return []Option{}, errors.Wrapf(err, "failed to fetch network attachment definition for selector %q", selectorKey)
 		}
@@ -78,6 +83,14 @@ func generateNetworkSettings(context *clusterd.Context, namespace string, networ
 	return cephNetworks, nil
 }
 
+func GetMultusNamespace(nad string) (string, string) {
+	tmp := strings.Split(nad, "/")
+	if len(tmp) == 2 {
+		return tmp[0], tmp[1]
+	}
+	return "", nad
+}
+
 func getNetworkRange(netConfig k8sutil.NetworkAttachmentConfig) string {
 	if netConfig.Ipam.Subnet != "" {
 		return netConfig.Ipam.Subnet
diff --git a/pkg/operator/ceph/csi/csi.go b/pkg/operator/ceph/csi/csi.go
index d50b4cb76..91fae6ec1 100644
--- a/pkg/operator/ceph/csi/csi.go
+++ b/pkg/operator/ceph/csi/csi.go
@@ -20,7 +20,7 @@ import (
 	"strconv"
 
 	"github.com/pkg/errors"
-	rookclient "github.com/rook/rook/pkg/client/clientset/versioned"
+	"github.com/rook/rook/pkg/clusterd"
 	controllerutil "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -28,9 +28,9 @@ import (
 	"k8s.io/client-go/kubernetes"
 )
 
-func ValidateAndConfigureDrivers(clientset kubernetes.Interface, rookclientset rookclient.Interface, namespace, rookImage, securityAccount string, serverVersion *version.Info, ownerRef *metav1.OwnerReference) {
+func ValidateAndConfigureDrivers(context *clusterd.Context, namespace, rookImage, securityAccount string, serverVersion *version.Info, ownerRef *metav1.OwnerReference) {
 	if !AllowUnsupported {
-		if err := validateCSIVersion(clientset, namespace, rookImage, securityAccount, ownerRef); err != nil {
+		if err := validateCSIVersion(context.Clientset, namespace, rookImage, securityAccount, ownerRef); err != nil {
 			logger.Errorf("invalid csi version. %+v", err)
 			return
 		}
@@ -38,11 +38,11 @@ func ValidateAndConfigureDrivers(clientset kubernetes.Interface, rookclientset r
 		logger.Info("Skipping csi version check, since unsupported versions are allowed")
 	}
 
-	if err := startDrivers(clientset, rookclientset, namespace, serverVersion, ownerRef); err != nil {
+	if err := startDrivers(context, namespace, serverVersion, ownerRef); err != nil {
 		logger.Errorf("failed to start Ceph csi drivers. %v", err)
 		return
 	}
-	stopDrivers(clientset, namespace, serverVersion)
+	stopDrivers(context.Clientset, namespace, serverVersion)
 }
 
 func SetParams(clientset kubernetes.Interface) error {
diff --git a/pkg/operator/ceph/csi/spec.go b/pkg/operator/ceph/csi/spec.go
index 57d1c550d..b7fbd5a40 100644
--- a/pkg/operator/ceph/csi/spec.go
+++ b/pkg/operator/ceph/csi/spec.go
@@ -17,17 +17,20 @@ limitations under the License.
 package csi
 
 import (
+	"context"
 	"fmt"
 	"strconv"
 	"strings"
 	"time"
 
-	rookclient "github.com/rook/rook/pkg/client/clientset/versioned"
+	v1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	controllerutil "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/rook/rook/pkg/operator/k8sutil/cmdreporter"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/pkg/errors"
+	"github.com/rook/rook/pkg/clusterd"
 	apps "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	k8scsi "k8s.io/api/storage/v1beta1"
@@ -201,7 +204,7 @@ func ValidateCSIParam() error {
 	return nil
 }
 
-func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Interface, namespace string, ver *version.Info, ownerRef *metav1.OwnerReference) error {
+func startDrivers(context *clusterd.Context, namespace string, ver *version.Info, ownerRef *metav1.OwnerReference) error {
 	var (
 		err                                                   error
 		rbdPlugin, cephfsPlugin                               *apps.DaemonSet
@@ -225,7 +228,7 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 	tp.EnableCSIGRPCMetrics = fmt.Sprintf("%t", EnableCSIGRPCMetrics)
 
 	// If not set or set to anything but "false", the kernel client will be enabled
-	kClient, err := k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "CSI_FORCE_CEPHFS_KERNEL_CLIENT", "true")
+	kClient, err := k8sutil.GetOperatorSetting(context.Clientset, controllerutil.OperatorSettingConfigMapName, "CSI_FORCE_CEPHFS_KERNEL_CLIENT", "true")
 	if err != nil {
 		return errors.Wrap(err, "failed to load CSI_FORCE_CEPHFS_KERNEL_CLIENT setting")
 	}
@@ -235,38 +238,38 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 		tp.ForceCephFSKernelClient = "true"
 	}
 	// parse GRPC and Liveness ports
-	tp.CephFSGRPCMetricsPort, err = getPortFromConfig(clientset, "CSI_CEPHFS_GRPC_METRICS_PORT", DefaultCephFSGRPCMerticsPort)
+	tp.CephFSGRPCMetricsPort, err = getPortFromConfig(context.Clientset, "CSI_CEPHFS_GRPC_METRICS_PORT", DefaultCephFSGRPCMerticsPort)
 	if err != nil {
 		return errors.Wrap(err, "error getting CSI CephFS GRPC metrics port.")
 	}
-	tp.CephFSLivenessMetricsPort, err = getPortFromConfig(clientset, "CSI_CEPHFS_LIVENESS_METRICS_PORT", DefaultCephFSLivenessMerticsPort)
+	tp.CephFSLivenessMetricsPort, err = getPortFromConfig(context.Clientset, "CSI_CEPHFS_LIVENESS_METRICS_PORT", DefaultCephFSLivenessMerticsPort)
 	if err != nil {
 		return errors.Wrap(err, "error getting CSI CephFS liveness metrics port.")
 	}
 
-	tp.RBDGRPCMetricsPort, err = getPortFromConfig(clientset, "CSI_RBD_GRPC_METRICS_PORT", DefaultRBDGRPCMerticsPort)
+	tp.RBDGRPCMetricsPort, err = getPortFromConfig(context.Clientset, "CSI_RBD_GRPC_METRICS_PORT", DefaultRBDGRPCMerticsPort)
 	if err != nil {
 		return errors.Wrap(err, "error getting CSI RBD GRPC metrics port.")
 	}
-	tp.RBDLivenessMetricsPort, err = getPortFromConfig(clientset, "CSI_RBD_LIVENESS_METRICS_PORT", DefaultRBDLivenessMerticsPort)
+	tp.RBDLivenessMetricsPort, err = getPortFromConfig(context.Clientset, "CSI_RBD_LIVENESS_METRICS_PORT", DefaultRBDLivenessMerticsPort)
 	if err != nil {
 		return errors.Wrap(err, "error getting CSI RBD liveness metrics port.")
 	}
 
 	// default value `system-node-critical` is the highest available priority
-	tp.PluginPriorityClassName, err = k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "CSI_PLUGIN_PRIORITY_CLASSNAME", "")
+	tp.PluginPriorityClassName, err = k8sutil.GetOperatorSetting(context.Clientset, controllerutil.OperatorSettingConfigMapName, "CSI_PLUGIN_PRIORITY_CLASSNAME", "")
 	if err != nil {
 		return errors.Wrap(err, "failed to load CSI_PLUGIN_PRIORITY_CLASSNAME setting")
 	}
 
 	// default value `system-cluster-critical` is applied for some
 	// critical pods in cluster but less priority than plugin pods
-	tp.ProvisionerPriorityClassName, err = k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "CSI_PROVISIONER_PRIORITY_CLASSNAME", "")
+	tp.ProvisionerPriorityClassName, err = k8sutil.GetOperatorSetting(context.Clientset, controllerutil.OperatorSettingConfigMapName, "CSI_PROVISIONER_PRIORITY_CLASSNAME", "")
 	if err != nil {
 		return errors.Wrap(err, "failed to load CSI_PROVISIONER_PRIORITY_CLASSNAME setting")
 	}
 
-	updateStrategy, err := k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "CSI_CEPHFS_PLUGIN_UPDATE_STRATEGY", rollingUpdate)
+	updateStrategy, err := k8sutil.GetOperatorSetting(context.Clientset, controllerutil.OperatorSettingConfigMapName, "CSI_CEPHFS_PLUGIN_UPDATE_STRATEGY", rollingUpdate)
 	if err != nil {
 		return errors.Wrap(err, "failed to load CSI_CEPHFS_PLUGIN_UPDATE_STRATEGY setting")
 	}
@@ -276,7 +279,7 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 		tp.CephFSPluginUpdateStrategy = rollingUpdate
 	}
 
-	updateStrategy, err = k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "CSI_RBD_PLUGIN_UPDATE_STRATEGY", rollingUpdate)
+	updateStrategy, err = k8sutil.GetOperatorSetting(context.Clientset, controllerutil.OperatorSettingConfigMapName, "CSI_RBD_PLUGIN_UPDATE_STRATEGY", rollingUpdate)
 	if err != nil {
 		return errors.Wrap(err, "failed to load CSI_RBD_PLUGIN_UPDATE_STRATEGY setting")
 	}
@@ -288,7 +291,7 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 
 	logger.Infof("Kubernetes version is %s.%s", ver.Major, ver.Minor)
 
-	tp.ResizerImage, err = k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "ROOK_CSI_RESIZER_IMAGE", DefaultResizerImage)
+	tp.ResizerImage, err = k8sutil.GetOperatorSetting(context.Clientset, controllerutil.OperatorSettingConfigMapName, "ROOK_CSI_RESIZER_IMAGE", DefaultResizerImage)
 	if err != nil {
 		return errors.Wrap(err, "failed to load ROOK_CSI_RESIZER_IMAGE setting")
 	}
@@ -303,7 +306,7 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 		logger.Warning("CSI Block volume expansion requires Kubernetes version >=1.16.0")
 	}
 
-	logLevel, err := k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "CSI_LOG_LEVEL", "")
+	logLevel, err := k8sutil.GetOperatorSetting(context.Clientset, controllerutil.OperatorSettingConfigMapName, "CSI_LOG_LEVEL", "")
 	if err != nil {
 		// logging a warning and intentionally continuing with the default log level
 		logger.Warningf("failed to load CSI_LOG_LEVEL. Defaulting to %d. %v", defaultLogLevel, err)
@@ -319,7 +322,7 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 	}
 
 	tp.ProvisionerReplicas = 2
-	nodes, err := clientset.CoreV1().Nodes().List(metav1.ListOptions{})
+	nodes, err := context.Clientset.CoreV1().Nodes().List(metav1.ListOptions{})
 	if err == nil {
 		if len(nodes.Items) == 1 {
 			tp.ProvisionerReplicas = 1
@@ -364,24 +367,24 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 	}
 
 	// get provisioner toleration and node affinity
-	provisionerTolerations := getToleration(clientset, true)
-	provisionerNodeAffinity := getNodeAffinity(clientset, true)
+	provisionerTolerations := getToleration(context.Clientset, true)
+	provisionerNodeAffinity := getNodeAffinity(context.Clientset, true)
 	// get plugin toleration and node affinity
-	pluginTolerations := getToleration(clientset, false)
-	pluginNodeAffinity := getNodeAffinity(clientset, false)
+	pluginTolerations := getToleration(context.Clientset, false)
+	pluginNodeAffinity := getNodeAffinity(context.Clientset, false)
 	if rbdPlugin != nil {
 		applyToPodSpec(&rbdPlugin.Spec.Template.Spec, pluginNodeAffinity, pluginTolerations)
 		// apply resource request and limit to rbdplugin containers
-		applyResourcesToContainers(clientset, rbdPluginResource, &rbdPlugin.Spec.Template.Spec)
+		applyResourcesToContainers(context.Clientset, rbdPluginResource, &rbdPlugin.Spec.Template.Spec)
 		k8sutil.SetOwnerRef(&rbdPlugin.ObjectMeta, ownerRef)
-		multusApplied, err := applyCephClusterNetworkConfig(&rbdPlugin.Spec.Template.ObjectMeta, rookclientset)
+		multusApplied, err := applyCephClusterNetworkConfig(context, &rbdPlugin.Spec.Template.ObjectMeta)
 		if err != nil {
 			return errors.Wrapf(err, "failed to apply network config to rbd plugin daemonset: %+v", rbdPlugin)
 		}
 		if multusApplied {
 			rbdPlugin.Spec.Template.Spec.HostNetwork = false
 		}
-		err = k8sutil.CreateDaemonSet(csiRBDPlugin, namespace, clientset, rbdPlugin)
+		err = k8sutil.CreateDaemonSet(csiRBDPlugin, namespace, context.Clientset, rbdPlugin)
 		if err != nil {
 			return errors.Wrapf(err, "failed to start rbdplugin daemonset: %+v", rbdPlugin)
 		}
@@ -391,7 +394,7 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 	if rbdProvisionerDeployment != nil {
 		applyToPodSpec(&rbdProvisionerDeployment.Spec.Template.Spec, provisionerNodeAffinity, provisionerTolerations)
 		// apply resource request and limit to rbd provisioner containers
-		applyResourcesToContainers(clientset, rbdProvisionerResource, &rbdProvisionerDeployment.Spec.Template.Spec)
+		applyResourcesToContainers(context.Clientset, rbdProvisionerResource, &rbdProvisionerDeployment.Spec.Template.Spec)
 		k8sutil.SetOwnerRef(&rbdProvisionerDeployment.ObjectMeta, ownerRef)
 		antiAffinity := GetPodAntiAffinity("app", csiRBDProvisioner)
 		rbdProvisionerDeployment.Spec.Template.Spec.Affinity.PodAntiAffinity = &antiAffinity
@@ -399,11 +402,11 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 			Type: apps.RecreateDeploymentStrategyType,
 		}
 
-		_, err = applyCephClusterNetworkConfig(&rbdProvisionerDeployment.Spec.Template.ObjectMeta, rookclientset)
+		_, err = applyCephClusterNetworkConfig(context, &rbdProvisionerDeployment.Spec.Template.ObjectMeta)
 		if err != nil {
 			return errors.Wrapf(err, "failed to apply network config to rbd plugin provisioner deployment: %+v", rbdProvisionerDeployment)
 		}
-		err = k8sutil.CreateDeployment(clientset, csiRBDProvisioner, namespace, rbdProvisionerDeployment)
+		err = k8sutil.CreateDeployment(context.Clientset, csiRBDProvisioner, namespace, rbdProvisionerDeployment)
 		if err != nil {
 			return errors.Wrapf(err, "failed to start rbd provisioner deployment: %+v", rbdProvisionerDeployment)
 		}
@@ -412,7 +415,7 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 
 	if rbdService != nil {
 		k8sutil.SetOwnerRef(&rbdService.ObjectMeta, ownerRef)
-		_, err = k8sutil.CreateOrUpdateService(clientset, namespace, rbdService)
+		_, err = k8sutil.CreateOrUpdateService(context.Clientset, namespace, rbdService)
 		if err != nil {
 			return errors.Wrapf(err, "failed to create rbd service: %+v", rbdService)
 		}
@@ -421,16 +424,16 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 	if cephfsPlugin != nil {
 		applyToPodSpec(&cephfsPlugin.Spec.Template.Spec, pluginNodeAffinity, pluginTolerations)
 		// apply resource request and limit to cephfs plugin containers
-		applyResourcesToContainers(clientset, cephFSPluginResource, &cephfsPlugin.Spec.Template.Spec)
+		applyResourcesToContainers(context.Clientset, cephFSPluginResource, &cephfsPlugin.Spec.Template.Spec)
 		k8sutil.SetOwnerRef(&cephfsPlugin.ObjectMeta, ownerRef)
-		multusApplied, err := applyCephClusterNetworkConfig(&cephfsPlugin.Spec.Template.ObjectMeta, rookclientset)
+		multusApplied, err := applyCephClusterNetworkConfig(context, &cephfsPlugin.Spec.Template.ObjectMeta)
 		if err != nil {
 			return errors.Wrapf(err, "failed to apply network config to cephfs plugin daemonset: %+v", cephfsPlugin)
 		}
 		if multusApplied {
 			cephfsPlugin.Spec.Template.Spec.HostNetwork = false
 		}
-		err = k8sutil.CreateDaemonSet(csiCephFSPlugin, namespace, clientset, cephfsPlugin)
+		err = k8sutil.CreateDaemonSet(csiCephFSPlugin, namespace, context.Clientset, cephfsPlugin)
 		if err != nil {
 			return errors.Wrapf(err, "failed to start cephfs plugin daemonset: %+v", cephfsPlugin)
 		}
@@ -441,7 +444,7 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 		applyToPodSpec(&cephfsProvisionerDeployment.Spec.Template.Spec, provisionerNodeAffinity, provisionerTolerations)
 		// get resource details for cephfs provisioner
 		// apply resource request and limit to cephfs provisioner containers
-		applyResourcesToContainers(clientset, cephFSProvisionerResource, &cephfsProvisionerDeployment.Spec.Template.Spec)
+		applyResourcesToContainers(context.Clientset, cephFSProvisionerResource, &cephfsProvisionerDeployment.Spec.Template.Spec)
 		k8sutil.SetOwnerRef(&cephfsProvisionerDeployment.ObjectMeta, ownerRef)
 		antiAffinity := GetPodAntiAffinity("app", csiCephFSProvisioner)
 		cephfsProvisionerDeployment.Spec.Template.Spec.Affinity.PodAntiAffinity = &antiAffinity
@@ -449,11 +452,11 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 			Type: apps.RecreateDeploymentStrategyType,
 		}
 
-		_, err = applyCephClusterNetworkConfig(&cephfsProvisionerDeployment.Spec.Template.ObjectMeta, rookclientset)
+		_, err = applyCephClusterNetworkConfig(context, &cephfsProvisionerDeployment.Spec.Template.ObjectMeta)
 		if err != nil {
 			return errors.Wrapf(err, "failed to apply network config to cephfs plugin provisioner deployment: %+v", cephfsProvisionerDeployment)
 		}
-		err = k8sutil.CreateDeployment(clientset, csiCephFSProvisioner, namespace, cephfsProvisionerDeployment)
+		err = k8sutil.CreateDeployment(context.Clientset, csiCephFSProvisioner, namespace, cephfsProvisionerDeployment)
 		if err != nil {
 			return errors.Wrapf(err, "failed to start cephfs provisioner deployment: %+v", cephfsProvisionerDeployment)
 		}
@@ -461,20 +464,20 @@ func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Inter
 	}
 	if cephfsService != nil {
 		k8sutil.SetOwnerRef(&cephfsService.ObjectMeta, ownerRef)
-		_, err = k8sutil.CreateOrUpdateService(clientset, namespace, cephfsService)
+		_, err = k8sutil.CreateOrUpdateService(context.Clientset, namespace, cephfsService)
 		if err != nil {
 			return errors.Wrapf(err, "failed to create rbd service: %+v", cephfsService)
 		}
 	}
 
 	if EnableRBD {
-		err = createCSIDriverInfo(clientset, RBDDriverName)
+		err = createCSIDriverInfo(context.Clientset, RBDDriverName)
 		if err != nil {
 			return errors.Wrapf(err, "failed to create CSI driver object for %q", RBDDriverName)
 		}
 	}
 	if EnableCephFS {
-		err = createCSIDriverInfo(clientset, CephFSDriverName)
+		err = createCSIDriverInfo(context.Clientset, CephFSDriverName)
 		if err != nil {
 			return errors.Wrapf(err, "failed to create CSI driver object for %q", CephFSDriverName)
 		}
@@ -534,11 +537,12 @@ func deleteCSIDriverResources(
 	return succeeded
 }
 
-func applyCephClusterNetworkConfig(objectMeta *metav1.ObjectMeta, rookclientset rookclient.Interface) (bool, error) {
+func applyCephClusterNetworkConfig(ctx *clusterd.Context, objectMeta *metav1.ObjectMeta) (bool, error) {
 	var isMultusApplied bool
-	cephClusters, err := rookclientset.CephV1().CephClusters(objectMeta.Namespace).List(metav1.ListOptions{})
+	cephClusters := &v1.CephClusterList{}
+	err := ctx.Client.List(context.TODO(), cephClusters, &client.ListOptions{})
 	if err != nil {
-		return false, errors.Errorf("failed to find CephClusters in namespace %q", objectMeta.Namespace)
+		return false, errors.Wrap(err, "failed to find CephClusters")
 	}
 	for _, cephCluster := range cephClusters.Items {
 		if cephCluster.Spec.Network.IsMultus() {
diff --git a/pkg/operator/ceph/operator.go b/pkg/operator/ceph/operator.go
index 5b21fbb3c..3f1eecd01 100644
--- a/pkg/operator/ceph/operator.go
+++ b/pkg/operator/ceph/operator.go
@@ -251,7 +251,7 @@ func (o *Operator) updateDrivers() error {
 		return errors.Wrap(err, "invalid csi params")
 	}
 
-	go csi.ValidateAndConfigureDrivers(o.context.Clientset, o.context.RookClientset, o.operatorNamespace, o.rookImage, o.securityAccount, serverVersion, ownerRef)
+	go csi.ValidateAndConfigureDrivers(o.context, o.operatorNamespace, o.rookImage, o.securityAccount, serverVersion, ownerRef)
 	return nil
 }
 


Test patch:
diff --git a/pkg/operator/ceph/config/network_test.go b/pkg/operator/ceph/config/network_test.go
index 6eb4896c1..6f0c2b62c 100644
--- a/pkg/operator/ceph/config/network_test.go
+++ b/pkg/operator/ceph/config/network_test.go
@@ -196,3 +196,15 @@ func TestGetNetworkRange(t *testing.T) {
 	networkRange = getNetworkRange(netConfig)
 	assert.Equal(t, "192.168.0.0/24", networkRange)
 }
+
+func TestGetMultusNamespace(t *testing.T) {
+	// TEST 1: When namespace is specified with the NAD
+	namespace, nad := GetMultusNamespace("multus-ns/public-nad")
+	assert.Equal(t, "multus-ns", namespace)
+	assert.Equal(t, "public-nad", nad)
+
+	// TEST 2: When only NAD is specified
+	namespace, nad = GetMultusNamespace("public-nad")
+	assert.Empty(t, namespace)
+	assert.Equal(t, "public-nad", nad)
+}
diff --git a/pkg/operator/ceph/csi/spec_test.go b/pkg/operator/ceph/csi/spec_test.go
index 491386f06..c4b1e61b8 100644
--- a/pkg/operator/ceph/csi/spec_test.go
+++ b/pkg/operator/ceph/csi/spec_test.go
@@ -19,7 +19,8 @@ package csi
 import (
 	"testing"
 
-	rookclient "github.com/rook/rook/pkg/client/clientset/versioned"
+	rookfake "github.com/rook/rook/pkg/client/clientset/versioned/fake"
+	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/operator/test"
 
 	"github.com/stretchr/testify/assert"
@@ -39,11 +40,14 @@ func TestStartCSI(t *testing.T) {
 		SnapshotterImage: "image",
 	}
 	clientset := test.New(t, 3)
-	var rookclientset rookclient.Interface
+	context := &clusterd.Context{
+		Clientset:     clientset,
+		RookClientset: rookfake.NewSimpleClientset(),
+	}
 	serverVersion, err := clientset.Discovery().ServerVersion()
 	if err != nil {
 		assert.Nil(t, err)
 	}
-	err = startDrivers(clientset, rookclientset, "ns", serverVersion, nil)
+	err = startDrivers(context, "ns", serverVersion, nil)
 	assert.Nil(t, err)
 }