Repository: rook/rook
Pull request #9925 (2022-03-17 14:24:59)

core: fix csi-cephfsplugin pod restart on non-hostnetworking env
**Description of your changes:**

Implementation of the design proposed in
https://github.com/rook/rook/pull/9903.

Signed-off-by: Sébastien Han <seb@redhat.com>

<!-- Please take a look at our [Contributing](https://rook.io/docs/rook/latest/development-flow.html)
documentation before submitting a Pull Request!
Thank you for contributing to Rook! -->


**Checklist:**

- [ ] Helm intgegration
- [ ] Doc? for Helm?


mergify[bot]: This pull request has merge conflicts that must be resolved before it can be merged. @leseb please rebase it. https://rook.io/docs/rook/latest/development-flow.html#updating-your-fork
leseb: CI is green and we can see the line:

```
I0414 13:24:29.893223    1214 cephcmds.go:71] ID: 8 Req-ID: 0001-0009-rook-ceph-0000000000000003-2b429e4e-bbf6-11ec-959f-dee386aefe14 command succeeded: nsenter [--net=/var/lib/kubelet/plugins/rook-ceph.rbd.csi.ceph.com/695adefe-af3c-4ba5-91cd-00d4d3180653.net.ns -- rbd --id csi-rbd-node -m 10.102.248.24:6789 --keyfile=***stripped*** map replicapool/csi-vol-2b429e4e-bbf6-11ec-959f-dee386aefe14 --device-type krbd --options noudev]
```

Which does all the magic :) 
mergify[bot]: This pull request has merge conflicts that must be resolved before it can be merged. @leseb please rebase it. https://rook.io/docs/rook/latest/development-flow.html#updating-your-fork
leseb: Need rebase on https://github.com/rook/rook/pull/10150 to work.

Patch:
diff --git a/pkg/daemon/ceph/cleanup/hostpath.go b/pkg/daemon/ceph/cleanup/hostpath.go
index 88eb33789..5c65c6435 100644
--- a/pkg/daemon/ceph/cleanup/hostpath.go
+++ b/pkg/daemon/ceph/cleanup/hostpath.go
@@ -23,7 +23,7 @@ import (
 	"path/filepath"
 
 	"github.com/pkg/errors"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
+	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 )
 
 // StartHostPathCleanup is the main entrypoint function to clean up dataDirHostPath and monitor store
@@ -78,7 +78,7 @@ func secretKeyMatch(monDir, monSecret string) (bool, error) {
 	if err != nil {
 		return false, errors.Wrapf(err, "failed to read keyring %q for the mon directory %q", keyringDirPath, monDir)
 	}
-	extractedKey, err := mon.ExtractKey(string(contents))
+	extractedKey, err := opcontroller.ExtractKey(string(contents))
 	if err != nil {
 		return false, errors.Wrapf(err, "failed to extract secret key from the keyring %q for the mon directory %q", keyringDirPath, monDir)
 	}
diff --git a/pkg/operator/ceph/client/controller.go b/pkg/operator/ceph/client/controller.go
index f7238fd1b..f3b45df67 100644
--- a/pkg/operator/ceph/client/controller.go
+++ b/pkg/operator/ceph/client/controller.go
@@ -37,7 +37,6 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
 	"github.com/rook/rook/pkg/operator/k8sutil"
@@ -177,7 +176,7 @@ func (r *ReconcileCephClient) reconcile(request reconcile.Request) (reconcile.Re
 	}
 
 	// Populate clusterInfo during each reconcile
-	r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
+	r.clusterInfo, _, _, err = opcontroller.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
 	if err != nil {
 		return reconcile.Result{}, *cephClient, errors.Wrap(err, "failed to populate cluster info")
 	}
diff --git a/pkg/operator/ceph/cluster/cleanup.go b/pkg/operator/ceph/cluster/cleanup.go
index dc5df20dd..89d007686 100644
--- a/pkg/operator/ceph/cluster/cleanup.go
+++ b/pkg/operator/ceph/cluster/cleanup.go
@@ -249,7 +249,7 @@ func (c *ClusterController) getCephHosts(namespace string) ([]string, error) {
 }
 
 func (c *ClusterController) getCleanUpDetails(namespace string) (string, string, error) {
-	clusterInfo, _, _, err := mon.LoadClusterInfo(c.context, c.OpManagerCtx, namespace)
+	clusterInfo, _, _, err := controller.LoadClusterInfo(c.context, c.OpManagerCtx, namespace)
 	if err != nil {
 		return "", "", errors.Wrap(err, "failed to get cluster info")
 	}
diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go
index 313a95df9..e764afa51 100755
--- a/pkg/operator/ceph/cluster/cluster.go
+++ b/pkg/operator/ceph/cluster/cluster.go
@@ -181,9 +181,9 @@ func (c *ClusterController) initializeCluster(cluster *cluster) error {
 			return errors.Wrap(err, "failed to configure external ceph cluster")
 		}
 	} else {
-		clusterInfo, _, _, err := mon.LoadClusterInfo(c.context, c.OpManagerCtx, cluster.Namespace)
+		clusterInfo, _, _, err := controller.LoadClusterInfo(c.context, c.OpManagerCtx, cluster.Namespace)
 		if err != nil {
-			if errors.Is(err, mon.ClusterInfoNoClusterNoSecret) {
+			if errors.Is(err, controller.ClusterInfoNoClusterNoSecret) {
 				logger.Info("clusterInfo not yet found, must be a new cluster.")
 			} else {
 				return errors.Wrap(err, "failed to load cluster info")
diff --git a/pkg/operator/ceph/cluster/cluster_external.go b/pkg/operator/ceph/cluster/cluster_external.go
index 8f67c7da8..56fa7ce1f 100644
--- a/pkg/operator/ceph/cluster/cluster_external.go
+++ b/pkg/operator/ceph/cluster/cluster_external.go
@@ -49,7 +49,7 @@ func (c *ClusterController) configureExternalCephCluster(cluster *cluster) error
 	// loop until we find the secret necessary to connect to the external cluster
 	// then populate clusterInfo
 
-	cluster.ClusterInfo, err = mon.PopulateExternalClusterInfo(c.context, c.OpManagerCtx, c.namespacedName.Namespace, cluster.ownerInfo)
+	cluster.ClusterInfo, err = opcontroller.PopulateExternalClusterInfo(c.context, c.OpManagerCtx, c.namespacedName.Namespace, cluster.ownerInfo)
 	if err != nil {
 		return errors.Wrap(err, "failed to populate external cluster info")
 	}
diff --git a/pkg/operator/ceph/cluster/mon/config.go b/pkg/operator/ceph/cluster/mon/config.go
index 57860d9c1..7e5dfbafd 100644
--- a/pkg/operator/ceph/cluster/mon/config.go
+++ b/pkg/operator/ceph/cluster/mon/config.go
@@ -17,28 +17,13 @@ limitations under the License.
 package mon
 
 import (
-	"context"
-	"encoding/json"
 	"fmt"
-	"io/ioutil"
-	"os"
 	"path"
-	"path/filepath"
-	"strconv"
 	"strings"
-	"time"
 
-	"github.com/google/uuid"
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
-	"github.com/rook/rook/pkg/operator/k8sutil"
-	"github.com/rook/rook/pkg/util/exec"
-	"github.com/rook/rook/pkg/util/sys"
-	v1 "k8s.io/api/core/v1"
-	kerrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
 )
 
 const (
@@ -52,12 +37,6 @@ const (
 	caps mon = "allow *"
 
 %s`
-
-	externalConnectionRetry = 60 * time.Second
-)
-
-var (
-	ClusterInfoNoClusterNoSecret = errors.New("not expected to create new cluster info and did not find existing secret")
 )
 
 func (c *Cluster) genMonSharedKeyring() string {
@@ -79,91 +58,6 @@ func dataDirRelativeHostPath(monName string) string {
 	return path.Join(monHostDir, "data")
 }
 
-// LoadClusterInfo constructs or loads a clusterinfo and returns it along with the maxMonID
-func LoadClusterInfo(ctx *clusterd.Context, context context.Context, namespace string) (*cephclient.ClusterInfo, int, *Mapping, error) {
-	return CreateOrLoadClusterInfo(ctx, context, namespace, nil)
-}
-
-// CreateOrLoadClusterInfo constructs or loads a clusterinfo and returns it along with the maxMonID
-func CreateOrLoadClusterInfo(clusterdContext *clusterd.Context, context context.Context, namespace string, ownerInfo *k8sutil.OwnerInfo) (*cephclient.ClusterInfo, int, *Mapping, error) {
-	var clusterInfo *cephclient.ClusterInfo
-	maxMonID := -1
-	monMapping := &Mapping{
-		Schedule: map[string]*MonScheduleInfo{},
-	}
-
-	secrets, err := clusterdContext.Clientset.CoreV1().Secrets(namespace).Get(context, AppName, metav1.GetOptions{})
-	if err != nil {
-		if !kerrors.IsNotFound(err) {
-			return nil, maxMonID, monMapping, errors.Wrap(err, "failed to get mon secrets")
-		}
-		if ownerInfo == nil {
-			return nil, maxMonID, monMapping, ClusterInfoNoClusterNoSecret
-		}
-
-		clusterInfo, err = createNamedClusterInfo(clusterdContext, namespace)
-		if err != nil {
-			return nil, maxMonID, monMapping, errors.Wrap(err, "failed to create initial cluster info")
-		}
-		clusterInfo.Context = context
-
-		err = createClusterAccessSecret(clusterdContext.Clientset, namespace, clusterInfo, ownerInfo)
-		if err != nil {
-			return nil, maxMonID, monMapping, err
-		}
-	} else {
-		clusterInfo = &cephclient.ClusterInfo{
-			Namespace:     namespace,
-			FSID:          string(secrets.Data[fsidSecretNameKey]),
-			MonitorSecret: string(secrets.Data[monSecretNameKey]),
-			Context:       context,
-		}
-		if cephUsername, ok := secrets.Data[cephUsernameKey]; ok {
-			clusterInfo.CephCred.Username = string(cephUsername)
-			clusterInfo.CephCred.Secret = string(secrets.Data[cephUserSecretKey])
-		} else if adminSecretKey, ok := secrets.Data[adminSecretNameKey]; ok {
-			clusterInfo.CephCred.Username = cephclient.AdminUsername
-			clusterInfo.CephCred.Secret = string(adminSecretKey)
-
-			secrets.Data[cephUsernameKey] = []byte(cephclient.AdminUsername)
-			secrets.Data[cephUserSecretKey] = adminSecretKey
-			if _, err = clusterdContext.Clientset.CoreV1().Secrets(namespace).Update(context, secrets, metav1.UpdateOptions{}); err != nil {
-				return nil, maxMonID, monMapping, errors.Wrap(err, "failed to update mon secrets")
-			}
-		} else {
-			return nil, maxMonID, monMapping, errors.New("failed to find either the cluster admin key or the username")
-		}
-		logger.Debugf("found existing monitor secrets for cluster %s", clusterInfo.Namespace)
-	}
-
-	// get the existing monitor config
-	clusterInfo.Monitors, maxMonID, monMapping, err = loadMonConfig(context, clusterdContext.Clientset, namespace)
-	if err != nil {
-		return nil, maxMonID, monMapping, errors.Wrap(err, "failed to get mon config")
-	}
-
-	// If an admin key was provided we don't need to load the other resources
-	// Some people might want to give the admin key
-	// The necessary users/keys/secrets will be created by Rook
-	// This is also done to allow backward compatibility
-	if clusterInfo.CephCred.Username == cephclient.AdminUsername && clusterInfo.CephCred.Secret != adminSecretNameKey {
-		return clusterInfo, maxMonID, monMapping, nil
-	}
-
-	// If the admin secret is "admin-secret", look for the deprecated secret that has the external creds
-	if clusterInfo.CephCred.Secret == adminSecretNameKey {
-		secret, err := clusterdContext.Clientset.CoreV1().Secrets(namespace).Get(context, OperatorCreds, metav1.GetOptions{})
-		if err != nil {
-			return nil, maxMonID, monMapping, err
-		}
-		// Populate external credential
-		clusterInfo.CephCred.Username = string(secret.Data["userID"])
-		clusterInfo.CephCred.Secret = string(secret.Data["userKey"])
-	}
-
-	return clusterInfo, maxMonID, monMapping, nil
-}
-
 // WriteConnectionConfig save monitor connection config to disk
 func WriteConnectionConfig(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo) error {
 	// write the latest config to the config dir
@@ -173,184 +67,3 @@ func WriteConnectionConfig(context *clusterd.Context, clusterInfo *cephclient.Cl
 
 	return nil
 }
-
-// loadMonConfig returns the monitor endpoints and maxMonID
-func loadMonConfig(ctx context.Context, clientset kubernetes.Interface, namespace string) (map[string]*cephclient.MonInfo, int, *Mapping, error) {
-	monEndpointMap := map[string]*cephclient.MonInfo{}
-	maxMonID := -1
-	monMapping := &Mapping{
-		Schedule: map[string]*MonScheduleInfo{},
-	}
-
-	cm, err := clientset.CoreV1().ConfigMaps(namespace).Get(ctx, EndpointConfigMapName, metav1.GetOptions{})
-	if err != nil {
-		if !kerrors.IsNotFound(err) {
-			return nil, maxMonID, monMapping, err
-		}
-		// If the config map was not found, initialize the empty set of monitors
-		return monEndpointMap, maxMonID, monMapping, nil
-	}
-
-	// Parse the monitor List
-	if info, ok := cm.Data[EndpointDataKey]; ok {
-		monEndpointMap = ParseMonEndpoints(info)
-	}
-
-	// Parse the max monitor id
-	storedMaxMonID := -1
-	if id, ok := cm.Data[MaxMonIDKey]; ok {
-		storedMaxMonID, err = strconv.Atoi(id)
-		if err != nil {
-			logger.Errorf("invalid max mon id %q. %v", id, err)
-		} else {
-			maxMonID = storedMaxMonID
-		}
-	}
-
-	// Make sure the max id is consistent with the current monitors
-	for _, m := range monEndpointMap {
-		id, _ := fullNameToIndex(m.Name)
-		if maxMonID < id {
-			maxMonID = id
-		}
-	}
-	if maxMonID != storedMaxMonID {
-		logger.Infof("updating obsolete maxMonID %d to actual value %d", storedMaxMonID, maxMonID)
-	}
-
-	err = json.Unmarshal([]byte(cm.Data[MappingKey]), &monMapping)
-	if err != nil {
-		logger.Errorf("invalid JSON in mon mapping. %v", err)
-	}
-
-	logger.Debugf("loaded: maxMonID=%d, mons=%+v, assignment=%+v", maxMonID, monEndpointMap, monMapping)
-	return monEndpointMap, maxMonID, monMapping, nil
-}
-
-func createClusterAccessSecret(clientset kubernetes.Interface, namespace string, clusterInfo *cephclient.ClusterInfo, ownerInfo *k8sutil.OwnerInfo) error {
-	logger.Infof("creating mon secrets for a new cluster")
-	var err error
-
-	// store the secrets for internal usage of the rook pods
-	secrets := map[string][]byte{
-		fsidSecretNameKey: []byte(clusterInfo.FSID),
-		monSecretNameKey:  []byte(clusterInfo.MonitorSecret),
-		cephUsernameKey:   []byte(clusterInfo.CephCred.Username),
-		cephUserSecretKey: []byte(clusterInfo.CephCred.Secret),
-	}
-	secret := &v1.Secret{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:       AppName,
-			Namespace:  namespace,
-			Finalizers: []string{DisasterProtectionFinalizerName},
-		},
-		Data: secrets,
-		Type: k8sutil.RookType,
-	}
-	err = ownerInfo.SetControllerReference(secret)
-	if err != nil {
-		return errors.Wrapf(err, "failed to set owner reference to mon secret %q", secret.Name)
-	}
-	if _, err = clientset.CoreV1().Secrets(namespace).Create(clusterInfo.Context, secret, metav1.CreateOptions{}); err != nil {
-		return errors.Wrap(err, "failed to save mon secrets")
-	}
-
-	return nil
-}
-
-// create new cluster info (FSID, shared keys)
-func createNamedClusterInfo(context *clusterd.Context, namespace string) (*cephclient.ClusterInfo, error) {
-	fsid, err := uuid.NewRandom()
-	if err != nil {
-		return nil, err
-	}
-
-	dir := path.Join(context.ConfigDir, namespace)
-	if err = os.MkdirAll(dir, 0744); err != nil {
-		return nil, errors.Wrapf(err, "failed to create dir %s", dir)
-	}
-
-	// generate the mon secret
-	monSecret, err := genSecret(context.Executor, dir, "mon.", []string{"--cap", "mon", "'allow *'"})
-	if err != nil {
-		return nil, err
-	}
-
-	// generate the admin secret if one was not provided at the command line
-	args := []string{
-		"--cap", "mon", "'allow *'",
-		"--cap", "osd", "'allow *'",
-		"--cap", "mgr", "'allow *'",
-		"--cap", "mds", "'allow'"}
-	adminSecret, err := genSecret(context.Executor, dir, cephclient.AdminUsername, args)
-	if err != nil {
-		return nil, err
-	}
-
-	return &cephclient.ClusterInfo{
-		FSID:          fsid.String(),
-		MonitorSecret: monSecret,
-		Namespace:     namespace,
-		CephCred: cephclient.CephCred{
-			Username: cephclient.AdminUsername,
-			Secret:   adminSecret,
-		},
-	}, nil
-}
-
-func genSecret(executor exec.Executor, configDir, name string, args []string) (string, error) {
-	path := path.Join(configDir, fmt.Sprintf("%s.keyring", name))
-	path = strings.Replace(path, "..", ".", 1)
-	base := []string{
-		"--create-keyring",
-		path,
-		"--gen-key",
-		"-n", name,
-	}
-	args = append(base, args...)
-	_, err := executor.ExecuteCommandWithOutput("ceph-authtool", args...)
-	if err != nil {
-		return "", errors.Wrap(err, "failed to gen secret")
-	}
-
-	contents, err := ioutil.ReadFile(filepath.Clean(path))
-	if err != nil {
-		return "", errors.Wrap(err, "failed to read secret file")
-	}
-	return ExtractKey(string(contents))
-}
-
-// ExtractKey retrieves mon secret key from the keyring file
-func ExtractKey(contents string) (string, error) {
-	secret := ""
-	slice := strings.Fields(sys.Grep(string(contents), "key"))
-	if len(slice) >= 3 {
-		secret = slice[2]
-	}
-	if secret == "" {
-		return "", errors.New("failed to parse secret")
-	}
-	return secret, nil
-}
-
-// PopulateExternalClusterInfo Add validation in the code to fail if the external cluster has no
-// OSDs keep waiting
-func PopulateExternalClusterInfo(context *clusterd.Context, ctx context.Context, namespace string, ownerInfo *k8sutil.OwnerInfo) (*cephclient.ClusterInfo, error) {
-	for {
-		// Checking for the context makes sure we don't loop forever with a canceled context
-		if ctx.Err() != nil {
-			return nil, ctx.Err()
-		}
-		clusterInfo, _, _, err := CreateOrLoadClusterInfo(context, ctx, namespace, nil)
-		if err != nil {
-			logger.Warningf("waiting for connection info of the external cluster. retrying in %s.", externalConnectionRetry.String())
-			logger.Debugf("%v", err)
-			time.Sleep(externalConnectionRetry)
-			continue
-		}
-		logger.Infof("found the cluster info to connect to the external cluster. will use %q to check health and monitor status. mons=%+v", clusterInfo.CephCred.Username, clusterInfo.Monitors)
-		clusterInfo.OwnerInfo = ownerInfo
-
-		return clusterInfo, nil
-	}
-}
diff --git a/pkg/operator/ceph/cluster/mon/env.go b/pkg/operator/ceph/cluster/mon/env.go
index ff4fa230b..5e72a757b 100644
--- a/pkg/operator/ceph/cluster/mon/env.go
+++ b/pkg/operator/ceph/cluster/mon/env.go
@@ -17,6 +17,7 @@ limitations under the License.
 package mon
 
 import (
+	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	v1 "k8s.io/api/core/v1"
 )
@@ -34,18 +35,18 @@ func EndpointEnvVar() v1.EnvVar {
 
 // SecretEnvVar is the mon secret environment var
 func SecretEnvVar() v1.EnvVar {
-	ref := &v1.SecretKeySelector{LocalObjectReference: v1.LocalObjectReference{Name: AppName}, Key: monSecretNameKey}
+	ref := &v1.SecretKeySelector{LocalObjectReference: v1.LocalObjectReference{Name: AppName}, Key: opcontroller.MonSecretNameKey}
 	return v1.EnvVar{Name: "ROOK_MON_SECRET", ValueFrom: &v1.EnvVarSource{SecretKeyRef: ref}}
 }
 
 // CephUsernameEnvVar is the ceph username environment var
 func CephUsernameEnvVar() v1.EnvVar {
-	ref := &v1.SecretKeySelector{LocalObjectReference: v1.LocalObjectReference{Name: AppName}, Key: cephUsernameKey}
+	ref := &v1.SecretKeySelector{LocalObjectReference: v1.LocalObjectReference{Name: AppName}, Key: opcontroller.CephUsernameKey}
 	return v1.EnvVar{Name: "ROOK_CEPH_USERNAME", ValueFrom: &v1.EnvVarSource{SecretKeyRef: ref}}
 }
 
 // CephSecretEnvVar is the ceph secret environment var
 func CephSecretEnvVar() v1.EnvVar {
-	ref := &v1.SecretKeySelector{LocalObjectReference: v1.LocalObjectReference{Name: AppName}, Key: cephUserSecretKey}
+	ref := &v1.SecretKeySelector{LocalObjectReference: v1.LocalObjectReference{Name: AppName}, Key: opcontroller.CephUserSecretKey}
 	return v1.EnvVar{Name: "ROOK_CEPH_SECRET", ValueFrom: &v1.EnvVarSource{SecretKeyRef: ref}}
 }
diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go
index efc9a4096..72a1bbf2d 100644
--- a/pkg/operator/ceph/cluster/mon/mon.go
+++ b/pkg/operator/ceph/cluster/mon/mon.go
@@ -55,22 +55,11 @@ const (
 	EndpointConfigMapName = "rook-ceph-mon-endpoints"
 	// EndpointDataKey is the name of the key inside the mon configmap to get the endpoints
 	EndpointDataKey = "data"
-	// MaxMonIDKey is the name of the max mon id used
-	MaxMonIDKey = "maxMonId"
-	// MappingKey is the name of the mapping for the mon->node and node->port
-	MappingKey = "mapping"
-
 	// AppName is the name of the secret storing cluster mon.admin key, fsid and name
 	AppName = "rook-ceph-mon"
 	//nolint:gosec // OperatorCreds is the name of the secret
-	OperatorCreds     = "rook-ceph-operator-creds"
-	monClusterAttr    = "mon_cluster"
-	fsidSecretNameKey = "fsid"
-	monSecretNameKey  = "mon-secret"
-	// AdminSecretName is the name of the admin secret
-	adminSecretNameKey = "admin-secret"
-	cephUsernameKey    = "ceph-username"
-	cephUserSecretKey  = "ceph-secret"
+	OperatorCreds  = "rook-ceph-operator-creds"
+	monClusterAttr = "mon_cluster"
 
 	// DefaultMonCount Default mon count for a cluster
 	DefaultMonCount = 3
@@ -119,7 +108,7 @@ type Cluster struct {
 	maxMonID           int
 	waitForStart       bool
 	monTimeoutList     map[string]time.Time
-	mapping            *Mapping
+	mapping            *controller.Mapping
 	ownerInfo          *k8sutil.OwnerInfo
 	isUpgrade          bool
 	arbiterMon         string
@@ -142,21 +131,6 @@ type monConfig struct {
 	DataPathMap *config.DataPathMap
 }
 
-// Mapping is mon node and port mapping
-type Mapping struct {
-	// This isn't really node info since it could also be for zones, but we leave it as "node" for backward compatibility.
-	Schedule map[string]*MonScheduleInfo `json:"node"`
-}
-
-// MonScheduleInfo contains name and address of a node.
-type MonScheduleInfo struct {
-	// Name of the node. **json names are capitalized for backwards compat**
-	Name     string `json:"Name,omitempty"`
-	Hostname string `json:"Hostname,omitempty"`
-	Address  string `json:"Address,omitempty"`
-	Zone     string `json:"zone,omitempty"`
-}
-
 type SchedulingResult struct {
 	Node             *v1.Node
 	CanaryDeployment *apps.Deployment
@@ -172,8 +146,8 @@ func New(ctx context.Context, clusterdContext *clusterd.Context, namespace strin
 		maxMonID:       -1,
 		waitForStart:   true,
 		monTimeoutList: map[string]time.Time{},
-		mapping: &Mapping{
-			Schedule: map[string]*MonScheduleInfo{},
+		mapping: &controller.Mapping{
+			Schedule: map[string]*controller.MonScheduleInfo{},
 		},
 		ownerInfo: ownerInfo,
 		ClusterInfo: &cephclient.ClusterInfo{
@@ -490,7 +464,7 @@ func (c *Cluster) initClusterInfo(cephVersion cephver.CephVersion, clusterName s
 
 	context := c.ClusterInfo.Context
 	// get the cluster info from secret
-	c.ClusterInfo, c.maxMonID, c.mapping, err = CreateOrLoadClusterInfo(c.context, context, c.Namespace, c.ownerInfo)
+	c.ClusterInfo, c.maxMonID, c.mapping, err = controller.CreateOrLoadClusterInfo(c.context, context, c.Namespace, c.ownerInfo)
 	if err != nil {
 		return errors.Wrap(err, "failed to get cluster info")
 	}
@@ -879,7 +853,7 @@ func (c *Cluster) assignMons(mons []*monConfig) error {
 			// store nil in the node mapping to indicate that an explicit node
 			// placement is not being made. otherwise, the node choice will map
 			// directly to a node selector on the monitor pod.
-			var schedule *MonScheduleInfo
+			var schedule *controller.MonScheduleInfo
 			if c.spec.Network.IsHost() || c.monVolumeClaimTemplate(mon) == nil {
 				logger.Infof("mon %s assigned to node %s", mon.DaemonName, nodeChoice.Name)
 				schedule, err = getNodeInfoFromNode(*nodeChoice)
@@ -894,7 +868,7 @@ func (c *Cluster) assignMons(mons []*monConfig) error {
 
 			if c.spec.IsStretchCluster() {
 				if schedule == nil {
-					schedule = &MonScheduleInfo{}
+					schedule = &controller.MonScheduleInfo{}
 				}
 				logger.Infof("mon %q is assigned to zone %q", mon.DaemonName, mon.Zone)
 				schedule.Zone = mon.Zone
@@ -1094,9 +1068,9 @@ func (c *Cluster) persistExpectedMonDaemons() error {
 		// actually been started. If the operator is restarted or the reconcile is otherwise restarted,
 		// we want to calculate the mon scheduling next time based on the committed maxMonID, rather
 		// than only a mon scheduling, which may not have completed.
-		MaxMonIDKey:   maxMonID,
-		MappingKey:    string(monMapping),
-		csi.ConfigKey: csiConfigValue,
+		controller.MaxMonIDKey: maxMonID,
+		controller.MappingKey:  string(monMapping),
+		csi.ConfigKey:          csiConfigValue,
 	}
 
 	if _, err := c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Create(c.ClusterInfo.Context, configMap, metav1.CreateOptions{}); err != nil {
@@ -1119,7 +1093,7 @@ func (c *Cluster) getStoredMaxMonID() (string, error) {
 		return "", errors.Wrap(err, "could not load maxMonId")
 	}
 	if err == nil {
-		if val, ok := configmap.Data[MaxMonIDKey]; ok {
+		if val, ok := configmap.Data[controller.MaxMonIDKey]; ok {
 			return val, nil
 		}
 	}
@@ -1143,7 +1117,7 @@ func (c *Cluster) commitMaxMonID(monName string) error {
 	}
 
 	// set the new max key if greater
-	existingMax, err := strconv.Atoi(configmap.Data[MaxMonIDKey])
+	existingMax, err := strconv.Atoi(configmap.Data[controller.MaxMonIDKey])
 	if err != nil {
 		return errors.Wrap(err, "failed to read existing maxMonId")
 	}
@@ -1154,7 +1128,7 @@ func (c *Cluster) commitMaxMonID(monName string) error {
 	}
 
 	logger.Infof("updating maxMonID from %d to %d after committing mon %q", existingMax, committedMonID, monName)
-	configmap.Data[MaxMonIDKey] = strconv.Itoa(committedMonID)
+	configmap.Data[controller.MaxMonIDKey] = strconv.Itoa(committedMonID)
 
 	if _, err = c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Update(c.ClusterInfo.Context, configmap, metav1.UpdateOptions{}); err != nil {
 		return errors.Wrap(err, "failed to update mon endpoint config map for the maxMonID")
@@ -1220,7 +1194,7 @@ func (c *Cluster) updateMon(m *monConfig, d *apps.Deployment) error {
 //           - if HostPath -> leave node selector as is
 //           - if PVC      -> remove node selector, if present
 //
-func (c *Cluster) startMon(m *monConfig, schedule *MonScheduleInfo) error {
+func (c *Cluster) startMon(m *monConfig, schedule *controller.MonScheduleInfo) error {
 	// check if the monitor deployment already exists. if the deployment does
 	// exist, also determine if it using pvc storage.
 	pvcExists := false
diff --git a/pkg/operator/ceph/cluster/mon/node.go b/pkg/operator/ceph/cluster/mon/node.go
index 7a231e372..9c353c66e 100644
--- a/pkg/operator/ceph/cluster/mon/node.go
+++ b/pkg/operator/ceph/cluster/mon/node.go
@@ -18,11 +18,12 @@ package mon
 
 import (
 	"github.com/pkg/errors"
+	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	v1 "k8s.io/api/core/v1"
 )
 
-func getNodeInfoFromNode(n v1.Node) (*MonScheduleInfo, error) {
-	nr := &MonScheduleInfo{
+func getNodeInfoFromNode(n v1.Node) (*opcontroller.MonScheduleInfo, error) {
+	nr := &opcontroller.MonScheduleInfo{
 		Name:     n.Name,
 		Hostname: n.Labels[v1.LabelHostname],
 	}
diff --git a/pkg/operator/ceph/cluster/mon/predicate.go b/pkg/operator/ceph/cluster/mon/predicate.go
index b858973ce..023e92ec9 100644
--- a/pkg/operator/ceph/cluster/mon/predicate.go
+++ b/pkg/operator/ceph/cluster/mon/predicate.go
@@ -21,6 +21,7 @@ import (
 	"reflect"
 	"sort"
 
+	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	corev1 "k8s.io/api/core/v1"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
@@ -62,14 +63,14 @@ func wereMonEndpointsUpdated(oldCMData, newCMData map[string]string) bool {
 	if oldMapping, ok := oldCMData["mapping"]; ok {
 		if newMapping, ok := newCMData["mapping"]; ok {
 			// Unmarshal both into a type
-			var oldMappingToGo Mapping
+			var oldMappingToGo opcontroller.Mapping
 			err := json.Unmarshal([]byte(oldMapping), &oldMappingToGo)
 			if err != nil {
 				logger.Debugf("failed to unmarshal new. %v", err)
 				return false
 			}
 
-			var newMappingToGo Mapping
+			var newMappingToGo opcontroller.Mapping
 			err = json.Unmarshal([]byte(newMapping), &newMappingToGo)
 			if err != nil {
 				logger.Debugf("failed to unmarshal new. %v", err)
diff --git a/pkg/operator/ceph/cluster/rbd/controller.go b/pkg/operator/ceph/cluster/rbd/controller.go
index a841e552d..931d957fd 100644
--- a/pkg/operator/ceph/cluster/rbd/controller.go
+++ b/pkg/operator/ceph/cluster/rbd/controller.go
@@ -27,7 +27,6 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
 	"github.com/rook/rook/pkg/operator/k8sutil"
@@ -188,7 +187,7 @@ func (r *ReconcileCephRBDMirror) reconcile(request reconcile.Request) (reconcile
 
 	// Populate clusterInfo
 	// Always populate it during each reconcile
-	r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
+	r.clusterInfo, _, _, err = opcontroller.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
 	if err != nil {
 		return opcontroller.ImmediateRetryResult, *cephRBDMirror, errors.Wrap(err, "failed to populate cluster info")
 	}
diff --git a/pkg/operator/ceph/cluster/version.go b/pkg/operator/ceph/cluster/version.go
index a3308ed59..1fc7d1a9f 100644
--- a/pkg/operator/ceph/cluster/version.go
+++ b/pkg/operator/ceph/cluster/version.go
@@ -143,7 +143,7 @@ func (c *cluster) validateCephVersion(version *cephver.CephVersion) error {
 	// state we are in and if we should upgrade or not
 
 	// Try to load clusterInfo so we can compare the running version with the one from the spec image
-	clusterInfo, _, _, err := mon.LoadClusterInfo(c.context, c.ClusterInfo.Context, c.Namespace)
+	clusterInfo, _, _, err := controller.LoadClusterInfo(c.context, c.ClusterInfo.Context, c.Namespace)
 	if err == nil {
 		clusterInfo.Context = c.ClusterInfo.Context
 		// Write connection info (ceph config file and keyring) for ceph commands
diff --git a/pkg/operator/ceph/controller/cluster_info.go b/pkg/operator/ceph/controller/cluster_info.go
new file mode 100644
index 000000000..e2c778558
--- /dev/null
+++ b/pkg/operator/ceph/controller/cluster_info.go
@@ -0,0 +1,378 @@
+/*
+Copyright 2022 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/pkg/errors"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/clusterd"
+	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/operator/k8sutil"
+	"github.com/rook/rook/pkg/util/exec"
+	"github.com/rook/rook/pkg/util/sys"
+	v1 "k8s.io/api/core/v1"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+const (
+	// OperatorCreds is the name of the secret
+	//nolint:gosec // since this is not leaking any hardcoded credentials, it's just the secret name
+	OperatorCreds     = "rook-ceph-operator-creds"
+	fsidSecretNameKey = "fsid"
+	MonSecretNameKey  = "mon-secret"
+	// AdminSecretName is the name of the admin secret
+	adminSecretNameKey = "admin-secret"
+	CephUsernameKey    = "ceph-username"
+	CephUserSecretKey  = "ceph-secret"
+	// EndpointConfigMapName is the name of the configmap with mon endpoints
+	EndpointConfigMapName = "rook-ceph-mon-endpoints"
+	// EndpointDataKey is the name of the key inside the mon configmap to get the endpoints
+	EndpointDataKey = "data"
+	// MaxMonIDKey is the name of the max mon id used
+	MaxMonIDKey = "maxMonId"
+	// MappingKey is the name of the mapping for the mon->node and node->port
+	MappingKey = "mapping"
+	// AppName is the name of the secret storing cluster mon.admin key, fsid and name
+	AppName                         = "rook-ceph-mon"
+	DisasterProtectionFinalizerName = cephv1.CustomResourceGroup + "/disaster-protection"
+)
+
+var (
+	ClusterInfoNoClusterNoSecret = errors.New("not expected to create new cluster info and did not find existing secret")
+	externalConnectionRetry      = 60 * time.Second
+)
+
+// Mapping is mon node and port mapping
+type Mapping struct {
+	// This isn't really node info since it could also be for zones, but we leave it as "node" for backward compatibility.
+	Schedule map[string]*MonScheduleInfo `json:"node"`
+}
+
+// MonScheduleInfo contains name and address of a node.
+type MonScheduleInfo struct {
+	// Name of the node. **json names are capitalized for backwards compat**
+	Name     string `json:"Name,omitempty"`
+	Hostname string `json:"Hostname,omitempty"`
+	Address  string `json:"Address,omitempty"`
+	Zone     string `json:"zone,omitempty"`
+}
+
+// LoadClusterInfo constructs or loads a clusterinfo and returns it along with the maxMonID
+func LoadClusterInfo(ctx *clusterd.Context, context context.Context, namespace string) (*cephclient.ClusterInfo, int, *Mapping, error) {
+	return CreateOrLoadClusterInfo(ctx, context, namespace, nil)
+}
+
+// CreateOrLoadClusterInfo constructs or loads a clusterinfo and returns it along with the maxMonID
+func CreateOrLoadClusterInfo(clusterdContext *clusterd.Context, context context.Context, namespace string, ownerInfo *k8sutil.OwnerInfo) (*cephclient.ClusterInfo, int, *Mapping, error) {
+	var clusterInfo *cephclient.ClusterInfo
+	maxMonID := -1
+	monMapping := &Mapping{
+		Schedule: map[string]*MonScheduleInfo{},
+	}
+
+	secrets, err := clusterdContext.Clientset.CoreV1().Secrets(namespace).Get(context, AppName, metav1.GetOptions{})
+	if err != nil {
+		if !kerrors.IsNotFound(err) {
+			return nil, maxMonID, monMapping, errors.Wrap(err, "failed to get mon secrets")
+		}
+		if ownerInfo == nil {
+			return nil, maxMonID, monMapping, ClusterInfoNoClusterNoSecret
+		}
+
+		clusterInfo, err = createNamedClusterInfo(clusterdContext, namespace)
+		if err != nil {
+			return nil, maxMonID, monMapping, errors.Wrap(err, "failed to create initial cluster info")
+		}
+		clusterInfo.Context = context
+
+		err = createClusterAccessSecret(clusterdContext.Clientset, namespace, clusterInfo, ownerInfo)
+		if err != nil {
+			return nil, maxMonID, monMapping, err
+		}
+	} else {
+		clusterInfo = &cephclient.ClusterInfo{
+			Namespace:     namespace,
+			FSID:          string(secrets.Data[fsidSecretNameKey]),
+			MonitorSecret: string(secrets.Data[MonSecretNameKey]),
+			Context:       context,
+		}
+		if cephUsername, ok := secrets.Data[CephUsernameKey]; ok {
+			clusterInfo.CephCred.Username = string(cephUsername)
+			clusterInfo.CephCred.Secret = string(secrets.Data[CephUserSecretKey])
+		} else if adminSecretKey, ok := secrets.Data[adminSecretNameKey]; ok {
+			clusterInfo.CephCred.Username = cephclient.AdminUsername
+			clusterInfo.CephCred.Secret = string(adminSecretKey)
+
+			secrets.Data[CephUsernameKey] = []byte(cephclient.AdminUsername)
+			secrets.Data[CephUserSecretKey] = adminSecretKey
+			if _, err = clusterdContext.Clientset.CoreV1().Secrets(namespace).Update(context, secrets, metav1.UpdateOptions{}); err != nil {
+				return nil, maxMonID, monMapping, errors.Wrap(err, "failed to update mon secrets")
+			}
+		} else {
+			return nil, maxMonID, monMapping, errors.New("failed to find either the cluster admin key or the username")
+		}
+		logger.Debugf("found existing monitor secrets for cluster %s", clusterInfo.Namespace)
+	}
+
+	// get the existing monitor config
+	clusterInfo.Monitors, maxMonID, monMapping, err = loadMonConfig(clusterdContext.Clientset, namespace)
+	if err != nil {
+		return nil, maxMonID, monMapping, errors.Wrap(err, "failed to get mon config")
+	}
+
+	// If an admin key was provided we don't need to load the other resources
+	// Some people might want to give the admin key
+	// The necessary users/keys/secrets will be created by Rook
+	// This is also done to allow backward compatibility
+	if clusterInfo.CephCred.Username == cephclient.AdminUsername && clusterInfo.CephCred.Secret != adminSecretNameKey {
+		return clusterInfo, maxMonID, monMapping, nil
+	}
+
+	// If the admin secret is "admin-secret", look for the deprecated secret that has the external creds
+	if clusterInfo.CephCred.Secret == adminSecretNameKey {
+		secret, err := clusterdContext.Clientset.CoreV1().Secrets(namespace).Get(context, OperatorCreds, metav1.GetOptions{})
+		if err != nil {
+			return nil, maxMonID, monMapping, err
+		}
+		// Populate external credential
+		clusterInfo.CephCred.Username = string(secret.Data["userID"])
+		clusterInfo.CephCred.Secret = string(secret.Data["userKey"])
+	}
+
+	return clusterInfo, maxMonID, monMapping, nil
+}
+
+// create new cluster info (FSID, shared keys)
+func createNamedClusterInfo(context *clusterd.Context, namespace string) (*cephclient.ClusterInfo, error) {
+	fsid, err := uuid.NewRandom()
+	if err != nil {
+		return nil, err
+	}
+
+	dir := path.Join(context.ConfigDir, namespace)
+	if err = os.MkdirAll(dir, 0744); err != nil {
+		return nil, errors.Wrapf(err, "failed to create dir %s", dir)
+	}
+
+	// generate the mon secret
+	monSecret, err := genSecret(context.Executor, dir, "mon.", []string{"--cap", "mon", "'allow *'"})
+	if err != nil {
+		return nil, err
+	}
+
+	// generate the admin secret if one was not provided at the command line
+	args := []string{
+		"--cap", "mon", "'allow *'",
+		"--cap", "osd", "'allow *'",
+		"--cap", "mgr", "'allow *'",
+		"--cap", "mds", "'allow'"}
+	adminSecret, err := genSecret(context.Executor, dir, cephclient.AdminUsername, args)
+	if err != nil {
+		return nil, err
+	}
+
+	return &cephclient.ClusterInfo{
+		FSID:          fsid.String(),
+		MonitorSecret: monSecret,
+		Namespace:     namespace,
+		CephCred: cephclient.CephCred{
+			Username: cephclient.AdminUsername,
+			Secret:   adminSecret,
+		},
+	}, nil
+}
+func genSecret(executor exec.Executor, configDir, name string, args []string) (string, error) {
+	path := path.Join(configDir, fmt.Sprintf("%s.keyring", name))
+	path = strings.Replace(path, "..", ".", 1)
+	base := []string{
+		"--create-keyring",
+		path,
+		"--gen-key",
+		"-n", name,
+	}
+	args = append(base, args...)
+	_, err := executor.ExecuteCommandWithOutput("ceph-authtool", args...)
+	if err != nil {
+		return "", errors.Wrap(err, "failed to gen secret")
+	}
+
+	contents, err := ioutil.ReadFile(filepath.Clean(path))
+	if err != nil {
+		return "", errors.Wrap(err, "failed to read secret file")
+	}
+	return ExtractKey(string(contents))
+}
+
+// ExtractKey retrieves mon secret key from the keyring file
+func ExtractKey(contents string) (string, error) {
+	secret := ""
+	slice := strings.Fields(sys.Grep(string(contents), "key"))
+	if len(slice) >= 3 {
+		secret = slice[2]
+	}
+	if secret == "" {
+		return "", errors.New("failed to parse secret")
+	}
+	return secret, nil
+}
+
+// loadMonConfig returns the monitor endpoints and maxMonID
+func loadMonConfig(clientset kubernetes.Interface, namespace string) (map[string]*cephclient.MonInfo, int, *Mapping, error) {
+	ctx := context.TODO()
+	monEndpointMap := map[string]*cephclient.MonInfo{}
+	maxMonID := -1
+	monMapping := &Mapping{
+		Schedule: map[string]*MonScheduleInfo{},
+	}
+
+	cm, err := clientset.CoreV1().ConfigMaps(namespace).Get(ctx, EndpointConfigMapName, metav1.GetOptions{})
+	if err != nil {
+		if !kerrors.IsNotFound(err) {
+			return nil, maxMonID, monMapping, err
+		}
+		// If the config map was not found, initialize the empty set of monitors
+		return monEndpointMap, maxMonID, monMapping, nil
+	}
+
+	// Parse the monitor List
+	if info, ok := cm.Data[EndpointDataKey]; ok {
+		monEndpointMap = ParseMonEndpoints(info)
+	}
+
+	// Parse the max monitor id
+	storedMaxMonID := -1
+	if id, ok := cm.Data[MaxMonIDKey]; ok {
+		storedMaxMonID, err = strconv.Atoi(id)
+		if err != nil {
+			logger.Errorf("invalid max mon id %q. %v", id, err)
+		} else {
+			maxMonID = storedMaxMonID
+		}
+	}
+
+	// Make sure the max id is consistent with the current monitors
+	for _, m := range monEndpointMap {
+		id, _ := fullNameToIndex(m.Name)
+		if maxMonID < id {
+			maxMonID = id
+		}
+	}
+	if maxMonID != storedMaxMonID {
+		logger.Infof("updating obsolete maxMonID %d to actual value %d", storedMaxMonID, maxMonID)
+	}
+
+	err = json.Unmarshal([]byte(cm.Data[MappingKey]), &monMapping)
+	if err != nil {
+		logger.Errorf("invalid JSON in mon mapping. %v", err)
+	}
+
+	logger.Debugf("loaded: maxMonID=%d, mons=%+v, assignment=%+v", maxMonID, monEndpointMap, monMapping)
+	return monEndpointMap, maxMonID, monMapping, nil
+}
+
+// convert the mon name to the numeric mon ID
+func fullNameToIndex(name string) (int, error) {
+	// remove the "rook-ceph-mon" prefix
+	name = strings.TrimPrefix(name, AppName)
+	// remove the "-" prefix
+	name = strings.TrimPrefix(name, "-")
+	return k8sutil.NameToIndex(name)
+}
+
+func createClusterAccessSecret(clientset kubernetes.Interface, namespace string, clusterInfo *cephclient.ClusterInfo, ownerInfo *k8sutil.OwnerInfo) error {
+	logger.Infof("creating mon secrets for a new cluster")
+	var err error
+
+	// store the secrets for internal usage of the rook pods
+	secrets := map[string][]byte{
+		fsidSecretNameKey: []byte(clusterInfo.FSID),
+		MonSecretNameKey:  []byte(clusterInfo.MonitorSecret),
+		CephUsernameKey:   []byte(clusterInfo.CephCred.Username),
+		CephUserSecretKey: []byte(clusterInfo.CephCred.Secret),
+	}
+	secret := &v1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:       AppName,
+			Namespace:  namespace,
+			Finalizers: []string{DisasterProtectionFinalizerName},
+		},
+		Data: secrets,
+		Type: k8sutil.RookType,
+	}
+	err = ownerInfo.SetControllerReference(secret)
+	if err != nil {
+		return errors.Wrapf(err, "failed to set owner reference to mon secret %q", secret.Name)
+	}
+	if _, err = clientset.CoreV1().Secrets(namespace).Create(clusterInfo.Context, secret, metav1.CreateOptions{}); err != nil {
+		return errors.Wrap(err, "failed to save mon secrets")
+	}
+
+	return nil
+}
+
+// ParseMonEndpoints parses a flattened representation of mons and endpoints in the form
+// <mon-name>=<mon-endpoint> and returns a list of Ceph mon configs.
+func ParseMonEndpoints(input string) map[string]*cephclient.MonInfo {
+	logger.Infof("parsing mon endpoints: %s", input)
+	mons := map[string]*cephclient.MonInfo{}
+	rawMons := strings.Split(input, ",")
+	for _, rawMon := range rawMons {
+		parts := strings.Split(rawMon, "=")
+		if len(parts) != 2 {
+			logger.Warningf("ignoring invalid monitor %s", rawMon)
+			continue
+		}
+		mons[parts[0]] = &cephclient.MonInfo{Name: parts[0], Endpoint: parts[1]}
+	}
+	return mons
+}
+
+// PopulateExternalClusterInfo Add validation in the code to fail if the external cluster has no
+// OSDs keep waiting
+func PopulateExternalClusterInfo(context *clusterd.Context, ctx context.Context, namespace string, ownerInfo *k8sutil.OwnerInfo) (*cephclient.ClusterInfo, error) {
+	for {
+		// Checking for the context makes sure we don't loop forever with a canceled context
+		if ctx.Err() != nil {
+			return nil, ctx.Err()
+		}
+		clusterInfo, _, _, err := CreateOrLoadClusterInfo(context, ctx, namespace, nil)
+		if err != nil {
+			logger.Warningf("waiting for connection info of the external cluster. retrying in %s.", externalConnectionRetry.String())
+			logger.Debugf("%v", err)
+			time.Sleep(externalConnectionRetry)
+			continue
+		}
+		logger.Infof("found the cluster info to connect to the external cluster. will use %q to check health and monitor status. mons=%+v", clusterInfo.CephCred.Username, clusterInfo.Monitors)
+		clusterInfo.OwnerInfo = ownerInfo
+
+		return clusterInfo, nil
+	}
+}
diff --git a/pkg/operator/ceph/csi/cluster_config.go b/pkg/operator/ceph/csi/cluster_config.go
index 6f241acb2..2f7c47015 100644
--- a/pkg/operator/ceph/csi/cluster_config.go
+++ b/pkg/operator/ceph/csi/cluster_config.go
@@ -42,11 +42,18 @@ type CsiClusterConfigEntry struct {
 	Monitors       []string       `json:"monitors"`
 	Namespace      string         `json:"namespace"`
 	CephFS         *CsiCephFSSpec `json:"cephFS,omitempty"`
+	RBD            *CsiRBDSpec    `json:"rbd,omitempty"`
 	RadosNamespace string         `json:"radosNamespace,omitempty"`
 }
 
 type CsiCephFSSpec struct {
-	SubvolumeGroup string `json:"subvolumeGroup,omitempty"`
+	NetNamespaceFilePath string `json:"netNamespaceFilePath,omitempty"`
+	SubvolumeGroup       string `json:"subvolumeGroup,omitempty"`
+}
+
+type CsiRBDSpec struct {
+	NetNamespaceFilePath string `json:"netNamespaceFilePath,omitempty"`
+	RadosNamespace       string `json:"radosNamespace,omitempty"`
 }
 
 type csiClusterConfig []CsiClusterConfigEntry
@@ -132,9 +139,14 @@ func updateCsiClusterConfig(curr, clusterKey string, newCsiClusterConfigEntry *C
 				break
 			}
 			centry.Monitors = newCsiClusterConfigEntry.Monitors
-			if newCsiClusterConfigEntry.CephFS != nil && newCsiClusterConfigEntry.CephFS.SubvolumeGroup != "" {
+			if newCsiClusterConfigEntry.CephFS != nil && (newCsiClusterConfigEntry.CephFS.SubvolumeGroup != "" || newCsiClusterConfigEntry.CephFS.NetNamespaceFilePath != "") {
 				centry.CephFS = newCsiClusterConfigEntry.CephFS
 			}
+			if newCsiClusterConfigEntry.RBD != nil && (newCsiClusterConfigEntry.RBD.RadosNamespace != "" || newCsiClusterConfigEntry.RBD.NetNamespaceFilePath != "") {
+				centry.RBD = newCsiClusterConfigEntry.RBD
+			}
+			// This maintains backward compatibility for existing clusters, from now on the
+			// preferred way is to use RBD.RadosNamespace
 			if newCsiClusterConfigEntry.RadosNamespace != "" {
 				centry.RadosNamespace = newCsiClusterConfigEntry.RadosNamespace
 			}
@@ -150,13 +162,13 @@ func updateCsiClusterConfig(curr, clusterKey string, newCsiClusterConfigEntry *C
 			centry.ClusterID = clusterKey
 			centry.Namespace = newCsiClusterConfigEntry.Namespace
 			centry.Monitors = newCsiClusterConfigEntry.Monitors
+			if newCsiClusterConfigEntry.RBD != nil && (newCsiClusterConfigEntry.RBD.RadosNamespace != "" || newCsiClusterConfigEntry.CephFS.NetNamespaceFilePath != "") {
+				centry.RBD = newCsiClusterConfigEntry.RBD
+			}
 			// Add a condition not to fill with empty values
-			if newCsiClusterConfigEntry.CephFS != nil && newCsiClusterConfigEntry.CephFS.SubvolumeGroup != "" {
+			if newCsiClusterConfigEntry.CephFS != nil && (newCsiClusterConfigEntry.CephFS.SubvolumeGroup != "" || newCsiClusterConfigEntry.CephFS.NetNamespaceFilePath != "") {
 				centry.CephFS = newCsiClusterConfigEntry.CephFS
 			}
-			if newCsiClusterConfigEntry.RadosNamespace != "" {
-				centry.RadosNamespace = newCsiClusterConfigEntry.RadosNamespace
-			}
 			cc = append(cc, centry)
 		}
 	}
@@ -215,6 +227,12 @@ func SaveClusterConfig(clientset kubernetes.Interface, clusterNamespace string,
 	// fetch current ConfigMap contents
 	configMap, err := clientset.CoreV1().ConfigMaps(csiNamespace).Get(clusterInfo.Context, ConfigName, metav1.GetOptions{})
 	if err != nil {
+		if k8serrors.IsNotFound(err) {
+			err = CreateCsiConfigMap(clusterInfo.Context, csiNamespace, clientset, clusterInfo.OwnerInfo)
+			if err != nil {
+				return errors.Wrap(err, "failed creating csi config map")
+			}
+		}
 		return errors.Wrap(err, "failed to fetch current csi config map")
 	}
 
diff --git a/pkg/operator/ceph/csi/controller.go b/pkg/operator/ceph/csi/controller.go
index 44b334d48..322b29098 100644
--- a/pkg/operator/ceph/csi/controller.go
+++ b/pkg/operator/ceph/csi/controller.go
@@ -21,6 +21,7 @@ import (
 	"os"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -49,6 +50,14 @@ type ReconcileCSI struct {
 	context          *clusterd.Context
 	opManagerContext context.Context
 	opConfig         opcontroller.OperatorConfig
+	multusClusters   []ClusterDetail
+}
+
+// ClusterDetail is a struct that holds the information of a cluster, it knows its internals (like
+// FSID through clusterInfo) and the Kubernetes object that represents it (like the CephCluster CRD)
+type ClusterDetail struct {
+	cluster     *cephv1.CephCluster
+	clusterInfo *cephclient.ClusterInfo
 }
 
 // Add creates a new Ceph CSI Controller and adds it to the Manager. The Manager will set fields on the Controller
@@ -64,6 +73,7 @@ func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerCont
 		context:          context,
 		opConfig:         opConfig,
 		opManagerContext: opManagerContext,
+		multusClusters:   []ClusterDetail{},
 	}
 }
 
@@ -139,17 +149,36 @@ func (r *ReconcileCSI) reconcile(request reconcile.Request) (reconcile.Result, e
 		}
 
 		return reconcile.Result{}, nil
-	} else {
-		for _, cluster := range cephClusters.Items {
-			if !cluster.DeletionTimestamp.IsZero() {
-				logger.Debug("ceph cluster is being deleting, no need to reconcile the csi driver")
-				return reconcile.Result{}, nil
-			}
+	}
+	for i, cluster := range cephClusters.Items {
+		if !cluster.DeletionTimestamp.IsZero() {
+			logger.Debugf("ceph cluster %q is being deleting, no need to reconcile the csi driver", request.NamespacedName)
+			return reconcile.Result{}, nil
+		}
 
-			if !cluster.Spec.External.Enable && cluster.Spec.CleanupPolicy.HasDataDirCleanPolicy() {
-				logger.Debug("ceph cluster has cleanup policy, the cluster will soon go away, no need to reconcile the csi driver")
-				return reconcile.Result{}, nil
+		if !cluster.Spec.External.Enable && cluster.Spec.CleanupPolicy.HasDataDirCleanPolicy() {
+			logger.Debugf("ceph cluster %q has cleanup policy, the cluster will soon go away, no need to reconcile the csi driver", cluster.Name)
+			return reconcile.Result{}, nil
+		}
+
+		// Do we have a multus cluster?
+		// If so deploy the plugin holder with the fsid attached
+		if cluster.Spec.Network.IsMultus() {
+			// Load cluster info for later use in updating the ceph-csi configmap
+			clusterInfo, _, _, err := opcontroller.LoadClusterInfo(r.context, r.opManagerContext, cluster.Namespace)
+			if err != nil {
+				// This avoids a requeue with exponential backoff and allows the controller to reconcile
+				// more quickly when the cluster is ready.
+				if errors.Is(err, opcontroller.ClusterInfoNoClusterNoSecret) {
+					return opcontroller.WaitForRequeueIfCephClusterNotReady, nil
+				}
+				return opcontroller.ImmediateRetryResult, errors.Wrapf(err, "failed to populate cluster info for cluster %q", cluster.Name)
 			}
+
+			logger.Debugf("cluster %q is running on multus, deploying the ceph-csi plugin holder", cluster.Name)
+			r.multusClusters = append(r.multusClusters, ClusterDetail{cluster: &cephClusters.Items[i], clusterInfo: clusterInfo})
+		} else {
+			logger.Debugf("not a multus cluster %q, not deploying the ceph-csi plugin holder", request.NamespacedName)
 		}
 	}
 
diff --git a/pkg/operator/ceph/csi/spec.go b/pkg/operator/ceph/csi/spec.go
index 27440ca80..8ce509571 100644
--- a/pkg/operator/ceph/csi/spec.go
+++ b/pkg/operator/ceph/csi/spec.go
@@ -22,14 +22,18 @@ import (
 	"fmt"
 	"time"
 
+	opcontroller "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"
 	apps "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	k8scsi "k8s.io/api/storage/v1beta1"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/version"
 )
 
@@ -81,6 +85,15 @@ type templateParam struct {
 	Namespace string
 }
 
+type driverDetails struct {
+	name           string
+	fullName       string
+	holderTemplate string
+	toleration     string
+	nodeAffinity   string
+	resource       string
+}
+
 var (
 	CSIParam Param
 
@@ -122,6 +135,8 @@ var (
 	// Local package template path for RBD
 	//go:embed template/rbd/csi-rbdplugin.yaml
 	RBDPluginTemplatePath string
+	//go:embed template/rbd/csi-rbdplugin-holder.yaml
+	RBDPluginHolderTemplatePath string
 	//go:embed template/rbd/csi-rbdplugin-provisioner-dep.yaml
 	RBDProvisionerDepTemplatePath string
 	//go:embed template/rbd/csi-rbdplugin-svc.yaml
@@ -130,6 +145,8 @@ var (
 	// Local package template path for CephFS
 	//go:embed template/cephfs/csi-cephfsplugin.yaml
 	CephFSPluginTemplatePath string
+	//go:embed template/cephfs/csi-cephfsplugin-holder.yaml
+	CephFSPluginHolderTemplatePath string
 	//go:embed template/cephfs/csi-cephfsplugin-provisioner-dep.yaml
 	CephFSProvisionerDepTemplatePath string
 	//go:embed template/cephfs/csi-cephfsplugin-svc.yaml
@@ -216,6 +233,11 @@ const (
 	csiRBDProvisioner    = "csi-rbdplugin-provisioner"
 	csiCephFSProvisioner = "csi-cephfsplugin-provisioner"
 	csiNFSProvisioner    = "csi-nfsplugin-provisioner"
+
+	RBDDriverShortName    = "rbd"
+	CephFSDriverShortName = "cephfs"
+	rbdDriverSuffix       = "rbd.csi.ceph.com"
+	cephFSDriverSuffix    = "cephfs.csi.ceph.com"
 )
 
 func CSIEnabled() bool {
@@ -247,6 +269,8 @@ func (r *ReconcileCSI) startDrivers(ver *version.Info, ownerInfo *k8sutil.OwnerI
 		rbdService, cephfsService                                                       *corev1.Service
 	)
 
+	enabledDrivers := make([]driverDetails, 0)
+
 	tp := templateParam{
 		Param:     CSIParam,
 		Namespace: r.opConfig.OperatorNamespace,
@@ -254,8 +278,8 @@ func (r *ReconcileCSI) startDrivers(ver *version.Info, ownerInfo *k8sutil.OwnerI
 
 	tp.DriverNamePrefix = fmt.Sprintf("%s.", r.opConfig.OperatorNamespace)
 
-	CephFSDriverName = tp.DriverNamePrefix + "cephfs.csi.ceph.com"
-	RBDDriverName = tp.DriverNamePrefix + "rbd.csi.ceph.com"
+	CephFSDriverName = tp.DriverNamePrefix + cephFSDriverSuffix
+	RBDDriverName = tp.DriverNamePrefix + rbdDriverSuffix
 	NFSDriverName = tp.DriverNamePrefix + "nfs.csi.ceph.com"
 
 	if CustomCSICephConfigExists {
@@ -300,6 +324,14 @@ func (r *ReconcileCSI) startDrivers(ver *version.Info, ownerInfo *k8sutil.OwnerI
 			return errors.Wrap(err, "failed to load rbd plugin service template")
 		}
 		rbdService.Namespace = r.opConfig.OperatorNamespace
+		enabledDrivers = append(enabledDrivers, driverDetails{
+			name:           RBDDriverShortName,
+			fullName:       RBDDriverName,
+			holderTemplate: RBDPluginHolderTemplatePath,
+			nodeAffinity:   rbdPluginNodeAffinityEnv,
+			toleration:     rbdPluginTolerationsEnv,
+			resource:       rbdPluginResource,
+		})
 	}
 	if EnableCephFS {
 		cephfsPlugin, err = templateToDaemonSet("cephfsplugin", CephFSPluginTemplatePath, tp)
@@ -317,6 +349,14 @@ func (r *ReconcileCSI) startDrivers(ver *version.Info, ownerInfo *k8sutil.OwnerI
 			return errors.Wrap(err, "failed to load cephfs plugin service template")
 		}
 		cephfsService.Namespace = r.opConfig.OperatorNamespace
+		enabledDrivers = append(enabledDrivers, driverDetails{
+			name:           CephFSDriverShortName,
+			fullName:       CephFSDriverName,
+			holderTemplate: CephFSPluginHolderTemplatePath,
+			nodeAffinity:   cephFSPluginNodeAffinityEnv,
+			toleration:     cephFSPluginTolerationsEnv,
+			resource:       cephFSPluginResource,
+		})
 	}
 
 	if EnableNFS {
@@ -330,14 +370,26 @@ func (r *ReconcileCSI) startDrivers(ver *version.Info, ownerInfo *k8sutil.OwnerI
 			return errors.Wrap(err, "failed to load nfs provisioner deployment template")
 		}
 	}
+	multusApplied := len(r.multusClusters) > 0
 
 	// get common provisioner tolerations and node affinity
 	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(r.opConfig.Parameters, pluginTolerationsEnv, []corev1.Toleration{})
 	pluginNodeAffinity := getNodeAffinity(r.opConfig.Parameters, pluginNodeAffinityEnv, &corev1.NodeAffinity{})
 
+	if multusApplied && !v.SupportsMultus() {
+		return errors.Errorf("multus is applied but the csi version %q does not support it, need at least %q", v.String(), multusSupportedVersion.String())
+	}
+
+	// Deploy the CSI Holder DaemonSet if Multus is enabled
+	err = r.configureHolders(enabledDrivers, tp, pluginTolerations, pluginNodeAffinity)
+	if err != nil {
+		return errors.Wrap(err, "failed to configure holder")
+	}
+
 	if rbdPlugin != nil {
 		// get RBD plugin tolerations and node affinity, defaults to common tolerations and node affinity if not specified
 		rbdPluginTolerations := getToleration(r.opConfig.Parameters, rbdPluginTolerationsEnv, pluginTolerations)
@@ -350,14 +402,14 @@ func (r *ReconcileCSI) startDrivers(ver *version.Info, ownerInfo *k8sutil.OwnerI
 		if err != nil {
 			return errors.Wrapf(err, "failed to set owner reference to rbd plugin daemonset %q", rbdPlugin.Name)
 		}
-		multusApplied, err := r.applyCephClusterNetworkConfig(r.opManagerContext, &rbdPlugin.Spec.Template.ObjectMeta)
+		_, err := r.applyCephClusterNetworkConfig(r.opManagerContext, &rbdPlugin.Spec.Template.ObjectMeta)
 		if err != nil {
 			return errors.Wrapf(err, "failed to apply network config to rbd plugin daemonset %q", rbdPlugin.Name)
 		}
 		if multusApplied {
 			rbdPlugin.Spec.Template.Spec.HostNetwork = false
 		}
-		err = k8sutil.CreateDaemonSet(r.opManagerContext, csiRBDPlugin, r.opConfig.OperatorNamespace, r.context.Clientset, rbdPlugin)
+		err = k8sutil.CreateDaemonSet(r.opManagerContext, r.opConfig.OperatorNamespace, r.context.Clientset, rbdPlugin)
 		if err != nil {
 			return errors.Wrapf(err, "failed to start rbdplugin daemonset %q", rbdPlugin.Name)
 		}
@@ -424,8 +476,11 @@ func (r *ReconcileCSI) startDrivers(ver *version.Info, ownerInfo *k8sutil.OwnerI
 		}
 		if multusApplied {
 			cephfsPlugin.Spec.Template.Spec.HostNetwork = false
+			// HostPID is used to communicate with the network namespace
+			cephfsPlugin.Spec.Template.Spec.HostPID = true
 		}
-		err = k8sutil.CreateDaemonSet(r.opManagerContext, csiCephFSPlugin, r.opConfig.OperatorNamespace, r.context.Clientset, cephfsPlugin)
+
+		err = k8sutil.CreateDaemonSet(r.opManagerContext, r.opConfig.OperatorNamespace, r.context.Clientset, cephfsPlugin)
 		if err != nil {
 			return errors.Wrapf(err, "failed to start cephfs plugin daemonset %q", cephfsPlugin.Name)
 		}
@@ -455,6 +510,7 @@ func (r *ReconcileCSI) startDrivers(ver *version.Info, ownerInfo *k8sutil.OwnerI
 		if err != nil {
 			return errors.Wrapf(err, "failed to apply network config to cephfs plugin provisioner deployment %q", cephfsProvisionerDeployment.Name)
 		}
+
 		_, err = k8sutil.CreateOrUpdateDeployment(r.opManagerContext, r.context.Clientset, cephfsProvisionerDeployment)
 		if err != nil {
 			return errors.Wrapf(err, "failed to start cephfs provisioner deployment %q", cephfsProvisionerDeployment.Name)
@@ -492,7 +548,7 @@ func (r *ReconcileCSI) startDrivers(ver *version.Info, ownerInfo *k8sutil.OwnerI
 		if multusApplied {
 			nfsPlugin.Spec.Template.Spec.HostNetwork = false
 		}
-		err = k8sutil.CreateDaemonSet(r.opManagerContext, csiNFSPlugin, r.opConfig.OperatorNamespace, r.context.Clientset, nfsPlugin)
+		err = k8sutil.CreateDaemonSet(r.opManagerContext, r.opConfig.OperatorNamespace, r.context.Clientset, nfsPlugin)
 		if err != nil {
 			return errors.Wrapf(err, "failed to start nfs plugin daemonset %q", nfsPlugin.Name)
 		}
@@ -683,3 +739,124 @@ func (r *ReconcileCSI) validateCSIVersion(ownerInfo *k8sutil.OwnerInfo) (*CephCS
 	}
 	return version, nil
 }
+
+func (r *ReconcileCSI) configureHolders(enabledDrivers []driverDetails, tp templateParam, pluginTolerations []corev1.Toleration, pluginNodeAffinity *corev1.NodeAffinity) error {
+	for _, multusCluster := range r.multusClusters {
+		for _, driver := range enabledDrivers {
+			err := r.configureHolder(driver, multusCluster, tp, pluginTolerations, pluginNodeAffinity)
+			if err != nil {
+				return errors.Wrapf(err, "failed to configure holder %q for %q/%q", driver.name, multusCluster.cluster.Name, multusCluster.cluster.Namespace)
+			}
+		}
+	}
+
+	return nil
+}
+
+func (r *ReconcileCSI) configureHolder(driver driverDetails, multusCluster ClusterDetail, tp templateParam, pluginTolerations []corev1.Toleration, pluginNodeAffinity *corev1.NodeAffinity) error {
+	cephPluginHolder, err := templateToDaemonSet("cephpluginholder", driver.holderTemplate, tp)
+	if err != nil {
+		return errors.Wrapf(err, "failed to load ceph %q plugin holder template", driver.fullName)
+	}
+
+	// DO NOT set owner reference on ceph plugin holder daemonset, this DS must never restart unless
+	// the entire node is rebooted
+	holderPluginTolerations := getToleration(r.opConfig.Parameters, driver.toleration, pluginTolerations)
+	holderPluginNodeAffinity := getNodeAffinity(r.opConfig.Parameters, driver.nodeAffinity, pluginNodeAffinity)
+	// apply driver's plugin tolerations and node affinity
+	applyToPodSpec(&cephPluginHolder.Spec.Template.Spec, holderPluginNodeAffinity, holderPluginTolerations)
+
+	// apply resource request and limit from corresponding plugin container
+	applyResourcesToContainers(r.opConfig.Parameters, driver.resource, &cephPluginHolder.Spec.Template.Spec)
+
+	// Append the CEPH_CLUSTER_NAMESPACE env var so that the main container can use it to create the network
+	// namespace symlink to the Kubelet plugin directory
+	cephPluginHolder.Spec.Template.Spec.Containers[0].Env = append(
+		cephPluginHolder.Spec.Template.Spec.Containers[0].Env,
+		corev1.EnvVar{
+			Name:  "CEPH_CLUSTER_NAMESPACE",
+			Value: multusCluster.cluster.Namespace,
+		},
+	)
+
+	// Append the driver name so that the symlink file goes into the right location on the
+	// kubelet plugin directory
+	cephPluginHolder.Spec.Template.Spec.Containers[0].Env = append(
+		cephPluginHolder.Spec.Template.Spec.Containers[0].Env,
+		corev1.EnvVar{
+			Name:  "ROOK_CEPH_CSI_DRIVER_NAME",
+			Value: driver.fullName,
+		},
+	)
+
+	// Make the DS name unique per Ceph cluster
+	cephPluginHolder.Name = fmt.Sprintf("%s-%s", cephPluginHolder.Name, multusCluster.cluster.Name)
+	cephPluginHolder.Spec.Template.Name = fmt.Sprintf("%s-%s", cephPluginHolder.Spec.Template.Name, multusCluster.cluster.Name)
+	cephPluginHolder.Spec.Template.Spec.Containers[0].Name = fmt.Sprintf("%s-%s", cephPluginHolder.Spec.Template.Spec.Containers[0].Name, multusCluster.cluster.Name)
+
+	// Add default labels
+	k8sutil.AddRookVersionLabelToDaemonSet(cephPluginHolder)
+
+	// Apply Multus annotations to daemonset spec
+	err = k8sutil.ApplyMultus(multusCluster.cluster.Spec.Network, &cephPluginHolder.Spec.Template.ObjectMeta)
+	if err != nil {
+		return errors.Wrapf(err, "failed to apply multus configuration for holder %q in cluster %q", cephPluginHolder.Name, multusCluster.cluster.Namespace)
+	}
+
+	// Finally create the DaemonSet
+	_, err = r.context.Clientset.AppsV1().DaemonSets(r.opConfig.OperatorNamespace).Create(r.opManagerContext, cephPluginHolder, metav1.CreateOptions{})
+	if err != nil {
+		if kerrors.IsAlreadyExists(err) {
+			logger.Debugf("holder %q already exists for cluster %q, it should never be updated", cephPluginHolder.Name, multusCluster.cluster.Namespace)
+		} else {
+			return errors.Wrapf(err, "failed to start ceph plugin holder daemonset %q", cephPluginHolder.Name)
+		}
+	}
+
+	clusterConfigEntry := &CsiClusterConfigEntry{
+		Monitors: MonEndpoints(multusCluster.clusterInfo.Monitors),
+		RBD:      &CsiRBDSpec{},
+		CephFS:   &CsiCephFSSpec{},
+	}
+	netNamespaceFilePath := generateNetNamespaceFilePath(CSIParam.KubeletDirPath, driver.fullName, multusCluster.cluster.Namespace)
+	if driver.name == RBDDriverShortName {
+		clusterConfigEntry.RBD.NetNamespaceFilePath = netNamespaceFilePath
+	}
+	if driver.name == CephFSDriverShortName {
+		clusterConfigEntry.CephFS.NetNamespaceFilePath = netNamespaceFilePath
+	}
+	// Save the path of the network namespace file for ceph-csi to use
+	err = SaveClusterConfig(r.context.Clientset, multusCluster.cluster.Namespace, multusCluster.clusterInfo, clusterConfigEntry)
+	if err != nil {
+		return errors.Wrapf(err, "failed to save cluster config for csi holder %q", driver.fullName)
+	}
+	return nil
+}
+
+func GenerateNetNamespaceFilePath(ctx context.Context, client client.Client, clusterNamespace, opNamespace, driverName string) (string, error) {
+	var driverSuffix string
+	opNamespaceName := types.NamespacedName{Name: opcontroller.OperatorSettingConfigMapName, Namespace: opNamespace}
+	opConfig := &corev1.ConfigMap{}
+	err := client.Get(ctx, opNamespaceName, opConfig)
+	if err != nil && !kerrors.IsNotFound(err) {
+		return "", errors.Wrap(err, "failed to get operator's configmap")
+	}
+
+	switch driverName {
+	case RBDDriverShortName:
+		driverSuffix = rbdDriverSuffix
+	case CephFSDriverShortName:
+		driverSuffix = cephFSDriverSuffix
+	default:
+		return "", errors.Errorf("unsupported driver name %q", driverName)
+	}
+
+	kubeletDirPath := k8sutil.GetValue(opConfig.Data, "ROOK_CSI_KUBELET_DIR_PATH", DefaultKubeletDirPath)
+	driverFullName := fmt.Sprintf("%s.%s", opNamespace, driverSuffix)
+
+	return generateNetNamespaceFilePath(kubeletDirPath, driverFullName, clusterNamespace), nil
+}
+
+func generateNetNamespaceFilePath(kubeletDirPath, driverFullName, clusterNamespace string) string {
+	return fmt.Sprintf("%s/plugins/%s/%s.net.ns", kubeletDirPath, driverFullName, clusterNamespace)
+}
diff --git a/pkg/operator/ceph/csi/template/cephfs/csi-cephfsplugin-holder.yaml b/pkg/operator/ceph/csi/template/cephfs/csi-cephfsplugin-holder.yaml
new file mode 100644
index 000000000..686f6d1e5
--- /dev/null
+++ b/pkg/operator/ceph/csi/template/cephfs/csi-cephfsplugin-holder.yaml
@@ -0,0 +1,63 @@
+# This DaemonSet is only active when the cluster is deployed with Multus enabled. Hopefully in the
+# future it will be part of every Ceph-CSI deployment. Its role is to "hold" the Ceph Filesystem
+# mounts that are created by the CSI driver. Essentially, it exposes its network namespace into the
+# Kubelet plugin directly and Ceph-CSI picks it up. When mounting, the Ceph-CSI driver will use the
+# network namespace of the DaemonSet instead of its own (the plugin pod). This pod is not expected
+# to be updated nor restarted unless the node reboots.
+kind: DaemonSet
+apiVersion: apps/v1
+metadata:
+  name: csi-cephfsplugin-holder
+  namespace: {{ .Namespace }}
+spec:
+  selector:
+    matchLabels:
+      app: csi-cephfsplugin-holder
+  updateStrategy:
+    # DO NOT CHANGE THE STRATEGY
+    # The pod is not supposed to be restarted on updates, it should never receive updates too.
+    # If it does then the connectivity to the Ceph cluster will be lost and the client I/O will
+    # fail.
+    type: OnDelete
+  template:
+    metadata:
+      labels:
+        app: csi-cephfsplugin-holder
+        {{ range $key, $value := .CSICephFSPodLabels }}
+        {{ $key }}: "{{ $value }}"
+        {{ end }}
+    spec:
+      # HostPID is needed to expose the correct process ID network namespace and not the process namespace
+      hostPID: true
+      {{ if .PluginPriorityClassName }}
+      priorityClassName: {{ .PluginPriorityClassName }}
+      {{ end }}
+      containers:
+        - name: csi-cephfsplugin-holder
+          # This is necessary for the Bidirectional mount propagation
+          securityContext:
+            privileged: true
+          image: {{ .CSIPluginImage }}
+          command:
+            - "/bin/sh"
+            - "-c"
+          # the third $ allows the "$$" to be interpreted so it can return the current bash process ID
+          args:
+            - "ln
+              --force
+              --symbolic
+              --verbose
+              /proc/$$$/ns/net
+              {{ .KubeletDirPath }}/plugins/$ROOK_CEPH_CSI_DRIVER_NAME/$CEPH_CLUSTER_NAMESPACE.net.ns
+              &&
+              sleep
+              infinity"
+          volumeMounts:
+            - name: plugin-mount-dir
+              mountPath: "{{ .KubeletDirPath }}/plugins"
+              mountPropagation: "Bidirectional"
+      volumes:
+        - name: plugin-mount-dir
+          hostPath:
+            path: "{{ .KubeletDirPath }}/plugins"
+            type: Directory
diff --git a/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin-holder.yaml b/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin-holder.yaml
new file mode 100644
index 000000000..6bf9b080f
--- /dev/null
+++ b/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin-holder.yaml
@@ -0,0 +1,63 @@
+# This DaemonSet is only active when the cluster is deployed with Multus enabled. Hopefully in the
+# future it will be part of every Ceph-CSI deployment. Its role is to "hold" the network RBD block
+# mapping that are created by the CSI driver. Essentially, it exposes its network namespace into the
+# Kubelet plugin directly and Ceph-CSI picks it up. When mapping an RBD image, the Ceph-CSI driver
+# will use the network namespace of the DaemonSet instead of its own (the plugin pod). This pod is
+# not expected to be updated nor restarted unless the node reboots.
+kind: DaemonSet
+apiVersion: apps/v1
+metadata:
+  name: csi-rbdplugin-holder
+  namespace: {{ .Namespace }}
+spec:
+  selector:
+    matchLabels:
+      app: csi-rbdplugin-holder
+  updateStrategy:
+    # DO NOT CHANGE THE STRATEGY
+    # The pod is not supposed to be restarted on updates, it should never receive updates too.
+    # If it does then the connectivity to the Ceph cluster will be lost and the client I/O will
+    # fail.
+    type: OnDelete
+  template:
+    metadata:
+      labels:
+        app: csi-rbdplugin-holder
+        {{ range $key, $value := .CSIRBDPodLabels }}
+        {{ $key }}: "{{ $value }}"
+        {{ end }}
+    spec:
+      # HostPID is needed to expose the correct process ID network namespace and not the process namespace
+      hostPID: true
+      {{ if .PluginPriorityClassName }}
+      priorityClassName: {{ .PluginPriorityClassName }}
+      {{ end }}
+      containers:
+        - name: csi-rbdplugin-holder
+          # This is necessary for the Bidirectional mount propagation
+          securityContext:
+            privileged: true
+          image: {{ .CSIPluginImage }}
+          command:
+            - "/bin/sh"
+            - "-c"
+          # the third $ allows the "$$" to be interpreted so it can return the current bash process ID
+          args:
+            - "ln
+              --force
+              --symbolic
+              --verbose
+              /proc/$$$/ns/net
+              {{ .KubeletDirPath }}/plugins/$ROOK_CEPH_CSI_DRIVER_NAME/$CEPH_CLUSTER_NAMESPACE.net.ns
+              &&
+              sleep
+              infinity"
+          volumeMounts:
+            - name: plugin-mount-dir
+              mountPath: "{{ .KubeletDirPath }}/plugins"
+              mountPropagation: "Bidirectional"
+      volumes:
+        - name: plugin-mount-dir
+          hostPath:
+            path: "{{ .KubeletDirPath }}/plugins"
+            type: Directory
diff --git a/pkg/operator/ceph/csi/version.go b/pkg/operator/ceph/csi/version.go
index 19ea488d5..1a7f21b9e 100644
--- a/pkg/operator/ceph/csi/version.go
+++ b/pkg/operator/ceph/csi/version.go
@@ -40,6 +40,9 @@ var (
 	// custom ceph.conf is supported in v3.5.0+
 	cephConfSupportedVersion = CephCSIVersion{3, 5, 0}
 
+	// multus fix with the extra holder pod is supported with at least v3.6.1
+	multusSupportedVersion = CephCSIVersion{3, 6, 1}
+
 	// for parsing the output of `cephcsi`
 	versionCSIPattern = regexp.MustCompile(`v(\d+)\.(\d+)\.(\d+)`)
 )
@@ -127,3 +130,14 @@ func (v *CephCSIVersion) SupportsCustomCephConf() bool {
 
 	return v.isAtLeast(&cephConfSupportedVersion)
 }
+
+// SupportsMultus checks if the csi image has support for calling "nsenter" while executing
+// mount/map commands. This is needed for Multus scenarios.
+func (v *CephCSIVersion) SupportsMultus() bool {
+	// if AllowUnsupported is set also a csi-image greater than the supported ones are allowed
+	if AllowUnsupported {
+		return true
+	}
+
+	return v.isAtLeast(&multusSupportedVersion)
+}
diff --git a/pkg/operator/ceph/file/controller.go b/pkg/operator/ceph/file/controller.go
index 6ab072db5..fe9e133f3 100644
--- a/pkg/operator/ceph/file/controller.go
+++ b/pkg/operator/ceph/file/controller.go
@@ -240,7 +240,7 @@ func (r *ReconcileCephFilesystem) reconcile(request reconcile.Request) (reconcil
 
 	// Populate clusterInfo
 	// Always populate it during each reconcile
-	clusterInfo, _, _, err := mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
+	clusterInfo, _, _, err := opcontroller.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
 	if err != nil {
 		return reconcile.Result{}, *cephFilesystem, errors.Wrap(err, "failed to populate cluster info")
 	}
diff --git a/pkg/operator/ceph/file/mirror/controller.go b/pkg/operator/ceph/file/mirror/controller.go
index c4cf593f5..de7999116 100644
--- a/pkg/operator/ceph/file/mirror/controller.go
+++ b/pkg/operator/ceph/file/mirror/controller.go
@@ -27,7 +27,6 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
 	"github.com/rook/rook/pkg/operator/k8sutil"
@@ -178,7 +177,7 @@ func (r *ReconcileFilesystemMirror) reconcile(request reconcile.Request) (reconc
 	r.cephClusterSpec = &cephCluster.Spec
 
 	// Populate clusterInfo
-	r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
+	r.clusterInfo, _, _, err = opcontroller.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
 	if err != nil {
 		return opcontroller.ImmediateRetryResult, *filesystemMirror, errors.Wrap(err, "failed to populate cluster info")
 	}
diff --git a/pkg/operator/ceph/file/subvolumegroup/controller.go b/pkg/operator/ceph/file/subvolumegroup/controller.go
index fe3f1403f..fd00744d5 100644
--- a/pkg/operator/ceph/file/subvolumegroup/controller.go
+++ b/pkg/operator/ceph/file/subvolumegroup/controller.go
@@ -38,7 +38,6 @@ import (
 	"github.com/coreos/pkg/capnslog"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/csi"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
@@ -70,21 +69,23 @@ type ReconcileCephFilesystemSubVolumeGroup struct {
 	context          *clusterd.Context
 	clusterInfo      *cephclient.ClusterInfo
 	opManagerContext context.Context
+	opConfig         opcontroller.OperatorConfig
 }
 
 // Add creates a new CephFilesystemSubVolumeGroup Controller and adds it to the Manager. The Manager will set fields on the Controller
 // and Start it when the Manager is Started.
 func Add(mgr manager.Manager, context *clusterd.Context, opManagerContext context.Context, opConfig opcontroller.OperatorConfig) error {
-	return add(mgr, newReconciler(mgr, context, opManagerContext))
+	return add(mgr, newReconciler(mgr, context, opManagerContext, opConfig))
 }
 
 // newReconciler returns a new reconcile.Reconciler
-func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerContext context.Context) reconcile.Reconciler {
+func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerContext context.Context, opConfig opcontroller.OperatorConfig) reconcile.Reconciler {
 	return &ReconcileCephFilesystemSubVolumeGroup{
 		client:           mgr.GetClient(),
 		scheme:           mgr.GetScheme(),
 		context:          context,
 		opManagerContext: opManagerContext,
+		opConfig:         opConfig,
 	}
 }
 
@@ -133,7 +134,7 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) reconcile(request reconcile.Requ
 		return reconcile.Result{}, errors.Wrap(err, "failed to get cephFilesystemSubVolumeGroup")
 	}
 	// update observedGeneration local variable with current generation value,
-	// because generation can be changed before reconile got completed
+	// because generation can be changed before reconcile got completed
 	// CR status will be updated at end of reconcile, so to reflect the reconcile has finished
 	observedGeneration := cephFilesystemSubVolumeGroup.ObjectMeta.Generation
 
@@ -171,7 +172,7 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) reconcile(request reconcile.Requ
 	}
 
 	// Populate clusterInfo during each reconcile
-	r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
+	r.clusterInfo, _, _, err = opcontroller.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
 	if err != nil {
 		return reconcile.Result{}, errors.Wrap(err, "failed to populate cluster info")
 	}
@@ -211,7 +212,7 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) reconcile(request reconcile.Requ
 
 	if cephCluster.Spec.External.Enable {
 		logger.Debug("external subvolume group creation is not supported, create it manually, the controller will assume it's there")
-		err = r.updateClusterConfig(cephFilesystemSubVolumeGroup)
+		err = r.updateClusterConfig(cephFilesystemSubVolumeGroup, cephCluster)
 		if err != nil {
 			return reconcile.Result{}, errors.Wrap(err, "failed to save cluster config")
 		}
@@ -250,17 +251,17 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) reconcile(request reconcile.Requ
 		return reconcile.Result{}, errors.Wrapf(err, "failed to create or update ceph filesystem subvolume group %q", cephFilesystemSubVolumeGroup.Name)
 	}
 
-	err = r.updateClusterConfig(cephFilesystemSubVolumeGroup)
+	err = r.updateClusterConfig(cephFilesystemSubVolumeGroup, cephCluster)
 	if err != nil {
 		return reconcile.Result{}, errors.Wrap(err, "failed to save cluster config")
 	}
 	r.updateStatus(observedGeneration, request.NamespacedName, cephv1.ConditionReady)
 	// Return and do not requeue
-	logger.Debug("done reconciling cephFilesystemSubVolumeGroup %q", namespacedName)
+	logger.Debugf("done reconciling cephFilesystemSubVolumeGroup %q", namespacedName)
 	return reconcile.Result{}, nil
 }
 
-func (r *ReconcileCephFilesystemSubVolumeGroup) updateClusterConfig(cephFilesystemSubVolumeGroup *cephv1.CephFilesystemSubVolumeGroup) error {
+func (r *ReconcileCephFilesystemSubVolumeGroup) updateClusterConfig(cephFilesystemSubVolumeGroup *cephv1.CephFilesystemSubVolumeGroup, cephCluster cephv1.CephCluster) error {
 	// Update CSI config map
 	// If the mon endpoints change, the mon health check go routine will take care of updating the
 	// config map, so no special care is needed in this controller
@@ -271,6 +272,17 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) updateClusterConfig(cephFilesyst
 			SubvolumeGroup: cephFilesystemSubVolumeGroup.Name,
 		},
 	}
+
+	// If the cluster has Multus enabled we need to append the network namespace of the driver's
+	// holder DaemonSet in the csi configmap
+	if cephCluster.Spec.Network.IsMultus() {
+		netNamespaceFilePath, err := csi.GenerateNetNamespaceFilePath(r.opManagerContext, r.client, cephCluster.ClusterName, r.opConfig.OperatorNamespace, csi.CephFSDriverShortName)
+		if err != nil {
+			return errors.Wrap(err, "failed to generate cephfs net namespace file path")
+		}
+		csiClusterConfigEntry.CephFS.NetNamespaceFilePath = netNamespaceFilePath
+	}
+
 	err := csi.SaveClusterConfig(r.context.Clientset, buildClusterID(cephFilesystemSubVolumeGroup), r.clusterInfo, &csiClusterConfigEntry)
 	if err != nil {
 		return errors.Wrap(err, "failed to save cluster config")
diff --git a/pkg/operator/ceph/nfs/controller.go b/pkg/operator/ceph/nfs/controller.go
index f90b46c21..e887cda5f 100644
--- a/pkg/operator/ceph/nfs/controller.go
+++ b/pkg/operator/ceph/nfs/controller.go
@@ -27,7 +27,6 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	"github.com/rook/rook/pkg/operator/ceph/config"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
@@ -197,7 +196,7 @@ func (r *ReconcileCephNFS) reconcile(request reconcile.Request) (reconcile.Resul
 
 	// Populate clusterInfo
 	// Always populate it during each reconcile
-	r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
+	r.clusterInfo, _, _, err = opcontroller.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
 	if err != nil {
 		return reconcile.Result{}, *cephNFS, errors.Wrap(err, "failed to populate cluster info")
 	}
diff --git a/pkg/operator/ceph/object/bucket/controller.go b/pkg/operator/ceph/object/bucket/controller.go
index 184f5a994..339c1e332 100644
--- a/pkg/operator/ceph/object/bucket/controller.go
+++ b/pkg/operator/ceph/object/bucket/controller.go
@@ -30,7 +30,6 @@ import (
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	v1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
@@ -148,11 +147,11 @@ func (r *ReconcileBucket) reconcile(request reconcile.Request) (reconcile.Result
 	}
 
 	// Populate clusterInfo during each reconcile
-	clusterInfo, _, _, err := mon.LoadClusterInfo(r.context, r.opManagerContext, cephCluster.Namespace)
+	clusterInfo, _, _, err := opcontroller.LoadClusterInfo(r.context, r.opManagerContext, cephCluster.Namespace)
 	if err != nil {
 		// This avoids a requeue with exponential backoff and allows the controller to reconcile
 		// more quickly when the cluster is ready.
-		if errors.Is(err, mon.ClusterInfoNoClusterNoSecret) {
+		if errors.Is(err, opcontroller.ClusterInfoNoClusterNoSecret) {
 			return opcontroller.WaitForRequeueIfOperatorNotInitialized, nil
 		}
 		return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to populate cluster info")
diff --git a/pkg/operator/ceph/object/controller.go b/pkg/operator/ceph/object/controller.go
index 4c1a82df1..9beafea9d 100644
--- a/pkg/operator/ceph/object/controller.go
+++ b/pkg/operator/ceph/object/controller.go
@@ -30,7 +30,6 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	"github.com/rook/rook/pkg/operator/ceph/config"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
@@ -234,7 +233,7 @@ func (r *ReconcileCephObjectStore) reconcile(request reconcile.Request) (reconci
 	}
 
 	// Populate clusterInfo during each reconcile
-	r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
+	r.clusterInfo, _, _, err = opcontroller.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
 	if err != nil {
 		return reconcile.Result{}, *cephObjectStore, errors.Wrap(err, "failed to populate cluster info")
 	}
diff --git a/pkg/operator/ceph/object/notification/controller.go b/pkg/operator/ceph/object/notification/controller.go
index b5d57ee1f..571e93ad9 100644
--- a/pkg/operator/ceph/object/notification/controller.go
+++ b/pkg/operator/ceph/object/notification/controller.go
@@ -27,7 +27,6 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/object"
 	"github.com/rook/rook/pkg/operator/ceph/object/bucket"
@@ -246,7 +245,7 @@ func getReadyCluster(client client.Client, opManagerContext context.Context, con
 		logger.Debug("Ceph cluster not yet present.")
 		return nil, nil, nil
 	}
-	clusterInfo, _, _, err := mon.LoadClusterInfo(&context, opManagerContext, cephCluster.Namespace)
+	clusterInfo, _, _, err := opcontroller.LoadClusterInfo(&context, opManagerContext, cephCluster.Namespace)
 	if err != nil {
 		return nil, nil, errors.Wrap(err, "failed to populate cluster info")
 	}
diff --git a/pkg/operator/ceph/object/realm/controller.go b/pkg/operator/ceph/object/realm/controller.go
index d14f2b210..9785f2a3c 100644
--- a/pkg/operator/ceph/object/realm/controller.go
+++ b/pkg/operator/ceph/object/realm/controller.go
@@ -25,7 +25,6 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -169,7 +168,7 @@ func (r *ReconcileObjectRealm) reconcile(request reconcile.Request) (reconcile.R
 	}
 
 	// Populate clusterInfo during each reconcile
-	r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
+	r.clusterInfo, _, _, err = opcontroller.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
 	if err != nil {
 		return reconcile.Result{}, *cephObjectRealm, errors.Wrap(err, "failed to populate cluster info")
 	}
diff --git a/pkg/operator/ceph/object/topic/controller.go b/pkg/operator/ceph/object/topic/controller.go
index 46d0b8a2d..608cc17e5 100644
--- a/pkg/operator/ceph/object/topic/controller.go
+++ b/pkg/operator/ceph/object/topic/controller.go
@@ -25,7 +25,6 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
 	"github.com/rook/rook/pkg/operator/k8sutil"
@@ -148,7 +147,7 @@ func (r *ReconcileBucketTopic) reconcile(request reconcile.Request) (reconcile.R
 	r.clusterSpec = &cephCluster.Spec
 
 	// Populate clusterInfo during each reconcile
-	r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, cephCluster.Namespace)
+	r.clusterInfo, _, _, err = opcontroller.LoadClusterInfo(r.context, r.opManagerContext, cephCluster.Namespace)
 	if err != nil {
 		return reconcile.Result{}, errors.Wrap(err, "failed to populate cluster info")
 	}
diff --git a/pkg/operator/ceph/object/user/controller.go b/pkg/operator/ceph/object/user/controller.go
index 4ff1f7896..d9b287059 100644
--- a/pkg/operator/ceph/object/user/controller.go
+++ b/pkg/operator/ceph/object/user/controller.go
@@ -23,7 +23,6 @@ import (
 	"reflect"
 
 	"github.com/ceph/go-ceph/rgw/admin"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -188,7 +187,7 @@ func (r *ReconcileObjectStoreUser) reconcile(request reconcile.Request) (reconci
 	r.cephClusterSpec = &cephCluster.Spec
 
 	// Populate clusterInfo during each reconcile
-	r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
+	r.clusterInfo, _, _, err = opcontroller.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
 	if err != nil {
 		return reconcile.Result{}, *cephObjectStoreUser, errors.Wrap(err, "failed to populate cluster info")
 	}
diff --git a/pkg/operator/ceph/object/zone/controller.go b/pkg/operator/ceph/object/zone/controller.go
index 58db4e225..90d163636 100644
--- a/pkg/operator/ceph/object/zone/controller.go
+++ b/pkg/operator/ceph/object/zone/controller.go
@@ -24,7 +24,6 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
 	v1 "k8s.io/api/core/v1"
@@ -168,7 +167,7 @@ func (r *ReconcileObjectZone) reconcile(request reconcile.Request) (reconcile.Re
 	}
 
 	// Populate clusterInfo during each reconcile
-	r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
+	r.clusterInfo, _, _, err = opcontroller.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
 	if err != nil {
 		return reconcile.Result{}, *cephObjectZone, errors.Wrap(err, "failed to populate cluster info")
 	}
diff --git a/pkg/operator/ceph/object/zonegroup/controller.go b/pkg/operator/ceph/object/zonegroup/controller.go
index feeb3bcf4..e31848145 100644
--- a/pkg/operator/ceph/object/zonegroup/controller.go
+++ b/pkg/operator/ceph/object/zonegroup/controller.go
@@ -24,7 +24,6 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -162,7 +161,7 @@ func (r *ReconcileObjectZoneGroup) reconcile(request reconcile.Request) (reconci
 	}
 
 	// Populate clusterInfo during each reconcile
-	r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
+	r.clusterInfo, _, _, err = opcontroller.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
 	if err != nil {
 		return reconcile.Result{}, errors.Wrap(err, "failed to populate cluster info")
 	}
diff --git a/pkg/operator/ceph/pool/controller.go b/pkg/operator/ceph/pool/controller.go
index 9197ec77d..9c168d436 100644
--- a/pkg/operator/ceph/pool/controller.go
+++ b/pkg/operator/ceph/pool/controller.go
@@ -202,7 +202,7 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
 	}
 
 	// Populate clusterInfo during each reconcile
-	clusterInfo, _, _, err := mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
+	clusterInfo, _, _, err := opcontroller.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
 	if err != nil {
 		return opcontroller.ImmediateRetryResult, *cephBlockPool, errors.Wrap(err, "failed to populate cluster info")
 	}
diff --git a/pkg/operator/ceph/pool/radosnamespace/controller.go b/pkg/operator/ceph/pool/radosnamespace/controller.go
index 866a3b682..ddd3459d2 100644
--- a/pkg/operator/ceph/pool/radosnamespace/controller.go
+++ b/pkg/operator/ceph/pool/radosnamespace/controller.go
@@ -27,7 +27,6 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/csi"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
@@ -68,22 +67,24 @@ type ReconcileCephBlockPoolRadosNamespace struct {
 	context          *clusterd.Context
 	clusterInfo      *cephclient.ClusterInfo
 	opManagerContext context.Context
+	opConfig         opcontroller.OperatorConfig
 }
 
 // Add creates a new CephBlockPoolRadosNamespace Controller and adds it to the
 // Manager. The Manager will set fields on the Controller and Start it when the
 // Manager is Started.
 func Add(mgr manager.Manager, context *clusterd.Context, opManagerContext context.Context, opConfig opcontroller.OperatorConfig) error {
-	return add(mgr, newReconciler(mgr, context, opManagerContext))
+	return add(mgr, newReconciler(mgr, context, opManagerContext, opConfig))
 }
 
 // newReconciler returns a new reconcile.Reconciler
-func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerContext context.Context) reconcile.Reconciler {
+func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerContext context.Context, opConfig opcontroller.OperatorConfig) reconcile.Reconciler {
 	return &ReconcileCephBlockPoolRadosNamespace{
 		client:           mgr.GetClient(),
 		scheme:           mgr.GetScheme(),
 		context:          context,
 		opManagerContext: opManagerContext,
+		opConfig:         opConfig,
 	}
 }
 
@@ -167,7 +168,7 @@ func (r *ReconcileCephBlockPoolRadosNamespace) reconcile(request reconcile.Reque
 	}
 
 	// Populate clusterInfo during each reconcile
-	r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
+	r.clusterInfo, _, _, err = opcontroller.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
 	if err != nil {
 		return reconcile.Result{}, errors.Wrap(err, "failed to populate cluster info")
 	}
@@ -205,7 +206,7 @@ func (r *ReconcileCephBlockPoolRadosNamespace) reconcile(request reconcile.Reque
 
 	if cephCluster.Spec.External.Enable {
 		logger.Debugf("external rados namespace %q creation is not supported, create it manually, the controller will assume it's there", namespacedName)
-		err = r.updateClusterConfig(cephBlockPoolRadosNamespace)
+		err = r.updateClusterConfig(cephBlockPoolRadosNamespace, cephCluster)
 		if err != nil {
 			return reconcile.Result{}, errors.Wrap(err, "failed to save cluster config")
 		}
@@ -241,7 +242,7 @@ func (r *ReconcileCephBlockPoolRadosNamespace) reconcile(request reconcile.Reque
 		return reconcile.Result{}, errors.Wrapf(err, "failed to create or update ceph pool rados namespace %q", cephBlockPoolRadosNamespace.Name)
 	}
 
-	err = r.updateClusterConfig(cephBlockPoolRadosNamespace)
+	err = r.updateClusterConfig(cephBlockPoolRadosNamespace, cephCluster)
 	if err != nil {
 		return reconcile.Result{}, errors.Wrap(err, "failed to save cluster config")
 	}
@@ -252,19 +253,34 @@ func (r *ReconcileCephBlockPoolRadosNamespace) reconcile(request reconcile.Reque
 	return reconcile.Result{}, nil
 }
 
-func (r *ReconcileCephBlockPoolRadosNamespace) updateClusterConfig(cephBlockPoolRadosNamespace *cephv1.CephBlockPoolRadosNamespace) error {
+func (r *ReconcileCephBlockPoolRadosNamespace) updateClusterConfig(cephBlockPoolRadosNamespace *cephv1.CephBlockPoolRadosNamespace, cephCluster cephv1.CephCluster) error {
 	// Update CSI config map
 	// If the mon endpoints change, the mon health check go routine will take care of updating the
 	// config map, so no special care is needed in this controller
 	csiClusterConfigEntry := csi.CsiClusterConfigEntry{
-		Namespace:      r.clusterInfo.Namespace,
-		Monitors:       csi.MonEndpoints(r.clusterInfo.Monitors),
+		Namespace: r.clusterInfo.Namespace,
+		Monitors:  csi.MonEndpoints(r.clusterInfo.Monitors),
+		RBD: &csi.CsiRBDSpec{
+			RadosNamespace: cephBlockPoolRadosNamespace.Name,
+		},
 		RadosNamespace: cephBlockPoolRadosNamespace.Name,
 	}
+
+	if cephCluster.Spec.Network.IsMultus() {
+		// Build the network namespace config to be injected into the csi config
+		netNamespaceFilePath, err := csi.GenerateNetNamespaceFilePath(r.opManagerContext, r.client, cephCluster.ClusterName, r.opConfig.OperatorNamespace, csi.RBDDriverShortName)
+		if err != nil {
+			return errors.Wrap(err, "failed to generate rbd net namespace file path")
+		}
+		csiClusterConfigEntry.RBD.NetNamespaceFilePath = netNamespaceFilePath
+	}
+
+	// Save cluster config in the csi config map
 	err := csi.SaveClusterConfig(r.context.Clientset, buildClusterID(cephBlockPoolRadosNamespace), r.clusterInfo, &csiClusterConfigEntry)
 	if err != nil {
 		return errors.Wrap(err, "failed to save cluster config")
 	}
+
 	return nil
 }
 
diff --git a/pkg/operator/k8sutil/daemonset.go b/pkg/operator/k8sutil/daemonset.go
index c17894176..c6892de88 100644
--- a/pkg/operator/k8sutil/daemonset.go
+++ b/pkg/operator/k8sutil/daemonset.go
@@ -27,14 +27,17 @@ import (
 )
 
 // CreateDaemonSet creates
-func CreateDaemonSet(ctx context.Context, name, namespace string, clientset kubernetes.Interface, ds *appsv1.DaemonSet) error {
+func CreateDaemonSet(ctx context.Context, namespace string, clientset kubernetes.Interface, ds *appsv1.DaemonSet) error {
+	if ds == nil {
+		return fmt.Errorf("failed to create daemonset: daemonset is nil")
+	}
 	_, err := clientset.AppsV1().DaemonSets(namespace).Create(ctx, ds, metav1.CreateOptions{})
 	if err != nil {
 		if k8serrors.IsAlreadyExists(err) {
 			_, err = clientset.AppsV1().DaemonSets(namespace).Update(ctx, ds, metav1.UpdateOptions{})
 		}
 		if err != nil {
-			return fmt.Errorf("failed to start %s daemonset: %+v\n%+v", name, err, ds)
+			return fmt.Errorf("failed to start %s daemonset: %+v\n%+v", ds.Name, err, ds)
 		}
 	}
 	return err


Test patch:
diff --git a/.github/workflows/canary-integration-test.yml b/.github/workflows/canary-integration-test.yml
index ccb3afe5f..54b337156 100644
--- a/.github/workflows/canary-integration-test.yml
+++ b/.github/workflows/canary-integration-test.yml
@@ -941,7 +941,16 @@ jobs:
       run: tests/scripts/github-action-helper.sh wait_for_prepare_pod
 
     - name: wait for ceph to be ready
-      run: tests/scripts/github-action-helper.sh wait_for_ceph_to_be_ready osd 2
+      run: IS_MULTUS=true tests/scripts/github-action-helper.sh wait_for_ceph_to_be_ready osd_multus 2
+
+    - name: wait for ceph-csi configmap to be updated with network namespace
+      run: tests/scripts/github-action-helper.sh wait_for_ceph_csi_configmap_to_be_updated
+
+    - name: test ceph-csi-rbd plugin restart
+      run: tests/scripts/github-action-helper.sh test_csi_rbd_workload
+
+    - name: test ceph-csi-cephfs plugin restart
+      run: tests/scripts/github-action-helper.sh test_csi_cephfs_workload
 
     - name: collect common logs
       if: always()
diff --git a/pkg/operator/ceph/cluster/mon/health_test.go b/pkg/operator/ceph/cluster/mon/health_test.go
index 329607f3a..4e149f4c8 100644
--- a/pkg/operator/ceph/cluster/mon/health_test.go
+++ b/pkg/operator/ceph/cluster/mon/health_test.go
@@ -29,6 +29,7 @@ import (
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	clienttest "github.com/rook/rook/pkg/daemon/ceph/client/test"
 	"github.com/rook/rook/pkg/operator/ceph/config"
+	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/version"
 	testopk8s "github.com/rook/rook/pkg/operator/k8sutil/test"
 	"github.com/rook/rook/pkg/operator/test"
@@ -77,7 +78,7 @@ func TestCheckHealth(t *testing.T) {
 	logger.Infof("initial mons: %v", c.ClusterInfo.Monitors)
 	c.waitForStart = false
 
-	c.mapping.Schedule["f"] = &MonScheduleInfo{
+	c.mapping.Schedule["f"] = &opcontroller.MonScheduleInfo{
 		Name:    "node0",
 		Address: "",
 	}
@@ -305,10 +306,10 @@ func TestCheckHealthNotFound(t *testing.T) {
 	setCommonMonProperties(c, 2, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
 	c.waitForStart = false
 
-	c.mapping.Schedule["a"] = &MonScheduleInfo{
+	c.mapping.Schedule["a"] = &opcontroller.MonScheduleInfo{
 		Name: "node0",
 	}
-	c.mapping.Schedule["b"] = &MonScheduleInfo{
+	c.mapping.Schedule["b"] = &opcontroller.MonScheduleInfo{
 		Name: "node0",
 	}
 	c.maxMonID = 4
diff --git a/pkg/operator/ceph/cluster/mon/mon_test.go b/pkg/operator/ceph/cluster/mon/mon_test.go
index 572fe37df..8ba3919ef 100644
--- a/pkg/operator/ceph/cluster/mon/mon_test.go
+++ b/pkg/operator/ceph/cluster/mon/mon_test.go
@@ -34,6 +34,7 @@ import (
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	clienttest "github.com/rook/rook/pkg/daemon/ceph/client/test"
 	"github.com/rook/rook/pkg/operator/ceph/config"
+	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/rook/rook/pkg/operator/test"
@@ -114,8 +115,8 @@ func newCluster(context *clusterd.Context, namespace string, allowMultiplePerNod
 		maxMonID:       -1,
 		waitForStart:   false,
 		monTimeoutList: map[string]time.Time{},
-		mapping: &Mapping{
-			Schedule: map[string]*MonScheduleInfo{},
+		mapping: &opcontroller.Mapping{
+			Schedule: map[string]*opcontroller.MonScheduleInfo{},
 		},
 		ownerInfo: ownerInfo,
 	}
@@ -151,7 +152,7 @@ func TestStartMonDeployment(t *testing.T) {
 
 	// Start mon a on a specific node since there is no volumeClaimTemplate
 	m := &monConfig{ResourceName: "rook-ceph-mon-a", DaemonName: "a", Port: 3300, PublicIP: "1.2.3.4", DataPathMap: &config.DataPathMap{}}
-	schedule := &MonScheduleInfo{Hostname: "host-a", Zone: "zonea"}
+	schedule := &opcontroller.MonScheduleInfo{Hostname: "host-a", Zone: "zonea"}
 	err = c.startMon(m, schedule)
 	assert.NoError(t, err)
 	deployment, err := c.context.Clientset.AppsV1().Deployments(c.Namespace).Get(c.ClusterInfo.Context, m.ResourceName, metav1.GetOptions{})
@@ -160,7 +161,7 @@ func TestStartMonDeployment(t *testing.T) {
 
 	// Start mon b on any node in a zone since there is a volumeClaimTemplate
 	m = &monConfig{ResourceName: "rook-ceph-mon-b", DaemonName: "b", Port: 3300, PublicIP: "1.2.3.5", DataPathMap: &config.DataPathMap{}}
-	schedule = &MonScheduleInfo{Hostname: "host-b", Zone: "zoneb"}
+	schedule = &opcontroller.MonScheduleInfo{Hostname: "host-b", Zone: "zoneb"}
 	c.spec.Mon.VolumeClaimTemplate = &v1.PersistentVolumeClaim{}
 	err = c.startMon(m, schedule)
 	assert.NoError(t, err)
@@ -308,13 +309,13 @@ func TestSaveMonEndpoints(t *testing.T) {
 	cm, err := c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(ctx, EndpointConfigMapName, metav1.GetOptions{})
 	assert.NoError(t, err)
 	assert.Equal(t, "a=1.2.3.1:3300", cm.Data[EndpointDataKey])
-	assert.Equal(t, `{"node":{}}`, cm.Data[MappingKey])
-	assert.Equal(t, "-1", cm.Data[MaxMonIDKey])
+	assert.Equal(t, `{"node":{}}`, cm.Data[opcontroller.MappingKey])
+	assert.Equal(t, "-1", cm.Data[opcontroller.MaxMonIDKey])
 
 	// update the config map
 	c.ClusterInfo.Monitors["a"].Endpoint = "2.3.4.5:6789"
 	c.maxMonID = 2
-	c.mapping.Schedule["a"] = &MonScheduleInfo{
+	c.mapping.Schedule["a"] = &opcontroller.MonScheduleInfo{
 		Name:     "node0",
 		Address:  "1.1.1.1",
 		Hostname: "myhost",
@@ -325,11 +326,11 @@ func TestSaveMonEndpoints(t *testing.T) {
 	cm, err = c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(ctx, EndpointConfigMapName, metav1.GetOptions{})
 	assert.NoError(t, err)
 	assert.Equal(t, "a=2.3.4.5:6789", cm.Data[EndpointDataKey])
-	assert.Equal(t, `{"node":{"a":{"Name":"node0","Hostname":"myhost","Address":"1.1.1.1"}}}`, cm.Data[MappingKey])
-	assert.Equal(t, "-1", cm.Data[MaxMonIDKey])
+	assert.Equal(t, `{"node":{"a":{"Name":"node0","Hostname":"myhost","Address":"1.1.1.1"}}}`, cm.Data[opcontroller.MappingKey])
+	assert.Equal(t, "-1", cm.Data[opcontroller.MaxMonIDKey])
 
 	// Update the maxMonID to some random value
-	cm.Data[MaxMonIDKey] = "23"
+	cm.Data[opcontroller.MaxMonIDKey] = "23"
 	_, err = c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Update(ctx, cm, metav1.UpdateOptions{})
 	assert.NoError(t, err)
 	// Confirm the maxMonId will be persisted and not updated to anything else.
@@ -338,7 +339,7 @@ func TestSaveMonEndpoints(t *testing.T) {
 	assert.NoError(t, err)
 	cm, err = c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(ctx, EndpointConfigMapName, metav1.GetOptions{})
 	assert.NoError(t, err)
-	assert.Equal(t, "23", cm.Data[MaxMonIDKey])
+	assert.Equal(t, "23", cm.Data[opcontroller.MaxMonIDKey])
 }
 
 func TestMaxMonID(t *testing.T) {
diff --git a/pkg/operator/ceph/cluster/mon/node_test.go b/pkg/operator/ceph/cluster/mon/node_test.go
index c5bb7005a..39b45367d 100644
--- a/pkg/operator/ceph/cluster/mon/node_test.go
+++ b/pkg/operator/ceph/cluster/mon/node_test.go
@@ -24,6 +24,7 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	clienttest "github.com/rook/rook/pkg/daemon/ceph/client/test"
+	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/rook/rook/pkg/operator/test"
@@ -214,7 +215,7 @@ func TestGetNodeInfoFromNode(t *testing.T) {
 	node.Status = v1.NodeStatus{}
 	node.Status.Addresses = []v1.NodeAddress{}
 
-	var info *MonScheduleInfo
+	var info *opcontroller.MonScheduleInfo
 	_, err = getNodeInfoFromNode(*node)
 	assert.NotNil(t, err)
 
diff --git a/pkg/operator/ceph/cluster/mon/config_test.go b/pkg/operator/ceph/controller/cluster_info_test.go
similarity index 96%
rename from pkg/operator/ceph/cluster/mon/config_test.go
rename to pkg/operator/ceph/controller/cluster_info_test.go
index 03d04b069..c22a9022f 100644
--- a/pkg/operator/ceph/cluster/mon/config_test.go
+++ b/pkg/operator/ceph/controller/cluster_info_test.go
@@ -1,5 +1,5 @@
 /*
-Copyright 2019 The Rook Authors. All rights reserved.
+Copyright 2022 The Rook Authors. All rights reserved.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package mon
+package controller
 
 import (
 	"context"
@@ -23,13 +23,12 @@ import (
 	"os"
 	"testing"
 
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/pkg/operator/test"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -73,8 +72,8 @@ func TestCreateClusterSecrets(t *testing.T) {
 
 	// For backward compatibility check that the admin secret can be loaded as previously specified
 	// Update the secret as if created in an old cluster
-	delete(secret.Data, cephUserSecretKey)
-	delete(secret.Data, cephUsernameKey)
+	delete(secret.Data, CephUserSecretKey)
+	delete(secret.Data, CephUsernameKey)
 	secret.Data[adminSecretNameKey] = []byte(adminSecret)
 	_, err = clientset.CoreV1().Secrets(namespace).Update(ctx, secret, metav1.UpdateOptions{})
 	assert.NoError(t, err)
diff --git a/pkg/operator/ceph/csi/cluster_config_test.go b/pkg/operator/ceph/csi/cluster_config_test.go
index df974bae6..9899767c1 100644
--- a/pkg/operator/ceph/csi/cluster_config_test.go
+++ b/pkg/operator/ceph/csi/cluster_config_test.go
@@ -27,6 +27,14 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 		Namespace: "rook-ceph-1",
 		Monitors:  []string{"1.2.3.4:5000"},
 	}
+	csiClusterConfigEntryMultus := CsiClusterConfigEntry{
+		Namespace: "rook-ceph-1",
+		Monitors:  []string{"1.2.3.4:5000"},
+		RBD: &CsiRBDSpec{
+			NetNamespaceFilePath: "/var/run/netns/rook-ceph-1",
+			RadosNamespace:       "rook-ceph-1",
+		},
+	}
 	csiClusterConfigEntry2 := CsiClusterConfigEntry{
 		Namespace: "rook-ceph-2",
 		Monitors:  []string{"20.1.1.1:5000", "20.1.1.2:5000", "20.1.1.3:5000"},
@@ -35,7 +43,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 		Namespace: "rook-ceph-3",
 		Monitors:  []string{"10.1.1.1:5000", "10.1.1.2:5000", "10.1.1.3:5000"},
 		CephFS: &CsiCephFSSpec{
-			SubvolumeGroup: "mygroup",
+			SubvolumeGroup: "my-group",
 		},
 	}
 
@@ -45,8 +53,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	t.Run("add a simple mons list", func(t *testing.T) {
 		s, err = updateCsiClusterConfig("[]", "rook-ceph-1", &csiClusterConfigEntry)
 		assert.NoError(t, err)
-		assert.Equal(t, s,
-			`[{"clusterID":"rook-ceph-1","monitors":["1.2.3.4:5000"],"namespace":"rook-ceph-1"}]`)
+		assert.Equal(t, `[{"clusterID":"rook-ceph-1","monitors":["1.2.3.4:5000"],"namespace":"rook-ceph-1"}]`, s)
 	})
 
 	t.Run("add a 2nd mon to the current cluster", func(t *testing.T) {
@@ -114,7 +121,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 		assert.Equal(t, "baba", cc[2].ClusterID)
 		assert.Equal(t, "10.1.1.1:5000", cc[2].Monitors[0])
 		assert.Equal(t, 3, len(cc[2].Monitors))
-		assert.Equal(t, "mygroup", cc[2].CephFS.SubvolumeGroup)
+		assert.Equal(t, "my-group", cc[2].CephFS.SubvolumeGroup)
 
 	})
 
@@ -126,7 +133,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 		assert.NoError(t, err)
 		assert.Equal(t, 3, len(cc))
 		assert.Equal(t, 4, len(cc[2].Monitors))
-		assert.Equal(t, "mygroup", cc[2].CephFS.SubvolumeGroup)
+		assert.Equal(t, "my-group", cc[2].CephFS.SubvolumeGroup)
 	})
 
 	t.Run("remove subvolumegroup", func(t *testing.T) {
@@ -141,7 +148,7 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 	t.Run("add subvolumegroup and mons after", func(t *testing.T) {
 		csiClusterConfigEntry4 := CsiClusterConfigEntry{
 			CephFS: &CsiCephFSSpec{
-				SubvolumeGroup: "mygroup2",
+				SubvolumeGroup: "my-group2",
 			},
 		}
 		s, err = updateCsiClusterConfig(s, "quatre", &csiClusterConfigEntry4)
@@ -150,14 +157,14 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 		assert.NoError(t, err)
 		assert.Equal(t, 3, len(cc), cc)
 		assert.Equal(t, 0, len(cc[2].Monitors))
-		assert.Equal(t, "mygroup2", cc[2].CephFS.SubvolumeGroup, cc)
+		assert.Equal(t, "my-group2", cc[2].CephFS.SubvolumeGroup, cc)
 
 		csiClusterConfigEntry4.Monitors = []string{"10.1.1.1:5000", "10.1.1.2:5000", "10.1.1.3:5000"}
 		s, err = updateCsiClusterConfig(s, "quatre", &csiClusterConfigEntry4)
 		assert.NoError(t, err)
 		cc, err = parseCsiClusterConfig(s)
 		assert.NoError(t, err)
-		assert.Equal(t, "mygroup2", cc[2].CephFS.SubvolumeGroup)
+		assert.Equal(t, "my-group2", cc[2].CephFS.SubvolumeGroup)
 		assert.Equal(t, 3, len(cc[2].Monitors))
 	})
 
@@ -200,6 +207,9 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 			Namespace:      clusterIDofCluster1,
 			Monitors:       csiCluster1ConfigEntry.Monitors,
 			RadosNamespace: radosNSofCluster1,
+			RBD: &CsiRBDSpec{
+				RadosNamespace: radosNSofCluster1,
+			},
 		}
 		s, err = updateCsiClusterConfig(s, radosNSofCluster1, &radosNsCsiCluster1Config)
 		assert.NoError(t, err)
@@ -207,7 +217,9 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 		assert.NoError(t, err)
 		assert.Equal(t, 3, len(cc), cc)
 		assert.Equal(t, 1, len(cc[2].Monitors))
-		assert.Equal(t, radosNSofCluster1, cc[2].RadosNamespace, cc)
+		// Now the configuration of new entries goes into RBD.RadosNamespace so it should be empty
+		assert.Empty(t, cc[2].RadosNamespace, cc)
+		assert.Equal(t, radosNSofCluster1, cc[2].RBD.RadosNamespace, cc)
 
 		// update mon IP's and check is it updating for all clusterID's
 		csiCluster1ConfigEntry.Monitors = append(csiCluster1ConfigEntry.Monitors, "1.2.3.10:5000")
@@ -304,6 +316,13 @@ func TestUpdateCsiClusterConfig(t *testing.T) {
 			assert.Equal(t, 1, len(cc[i].Monitors))
 		}
 	})
+
+	t.Run("test multus cluster", func(t *testing.T) {
+		s, err = updateCsiClusterConfig("[]", "rook-ceph-1", &csiClusterConfigEntryMultus)
+		assert.NoError(t, err)
+		assert.Equal(t, `[{"clusterID":"rook-ceph-1","monitors":["1.2.3.4:5000"],"namespace":"rook-ceph-1","rbd":{"netNamespaceFilePath":"/var/run/netns/rook-ceph-1","radosNamespace":"rook-ceph-1"}}]`, s)
+	})
+
 }
 
 func contains(src, dest []string) bool {
diff --git a/pkg/operator/ceph/csi/controller_test.go b/pkg/operator/ceph/csi/controller_test.go
index 821f2c85b..edbae6fff 100644
--- a/pkg/operator/ceph/csi/controller_test.go
+++ b/pkg/operator/ceph/csi/controller_test.go
@@ -155,4 +155,86 @@ func TestCephCSIController(t *testing.T) {
 		assert.NoError(t, err)
 		assert.Equal(t, 2, len(ds.Items), ds)
 	})
+
+	t.Run("success ceph csi deployment with multus", func(t *testing.T) {
+		fakeClientSet := test.New(t, 1)
+		test.SetFakeKubernetesVersion(fakeClientSet, "v1.21.0")
+		c := &clusterd.Context{
+			Clientset:     fakeClientSet,
+			RookClientset: rookclient.NewSimpleClientset(),
+		}
+		_, err := c.Clientset.CoreV1().Pods(namespace).Create(ctx, test.FakeOperatorPod(namespace), metav1.CreateOptions{})
+		assert.NoError(t, err)
+		_, err = c.Clientset.AppsV1().ReplicaSets(namespace).Create(context.TODO(), test.FakeReplicaSet(namespace), metav1.CreateOptions{})
+		assert.NoError(t, err)
+		cephCluster := &cephv1.CephCluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      namespace,
+				Namespace: namespace,
+			},
+			Spec: cephv1.ClusterSpec{
+				Network: cephv1.NetworkSpec{
+					Provider:  "multus",
+					Selectors: map[string]string{"public": "public-net", "cluster": "cluster-net"},
+				},
+			},
+			Status: cephv1.ClusterStatus{
+				Phase: "",
+				CephVersion: &cephv1.ClusterVersion{
+					Version: "14.2.9-0",
+				},
+				CephStatus: &cephv1.CephStatus{
+					FSID: "89d00089-d6e6-4445-b596-96898ce31793",
+				},
+			},
+		}
+		// Mock clusterInfo
+		secrets := map[string][]byte{
+			"fsid":         []byte(name),
+			"mon-secret":   []byte("monsecret"),
+			"admin-secret": []byte("adminsecret"),
+		}
+		secret := &v1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "rook-ceph-mon",
+				Namespace: namespace,
+			},
+			Data: secrets,
+			Type: k8sutil.RookType,
+		}
+		_, err = c.Clientset.CoreV1().Secrets(namespace).Create(ctx, secret, metav1.CreateOptions{})
+		assert.NoError(t, err)
+		s := scheme.Scheme
+		s.AddKnownTypes(cephv1.SchemeGroupVersion, &cephv1.CephCluster{}, &cephv1.CephClusterList{})
+
+		object := []runtime.Object{
+			cephCluster,
+		}
+		// Create a fake client to mock API calls.
+		cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
+		c.Client = cl
+		// // Create a ReconcileCSI object with the scheme and fake client.
+		r := &ReconcileCSI{
+			client:  cl,
+			context: c,
+			opConfig: controller.OperatorConfig{
+				OperatorNamespace: namespace,
+				Image:             "rook",
+				ServiceAccount:    "foo",
+			},
+		}
+
+		res, err := r.Reconcile(ctx, req)
+		assert.NoError(t, err)
+		assert.False(t, res.Requeue)
+
+		ds, err := c.Clientset.AppsV1().DaemonSets(namespace).List(ctx, metav1.ListOptions{})
+		assert.NoError(t, err)
+		assert.Equal(t, 4, len(ds.Items), ds)
+		assert.Equal(t, "csi-cephfsplugin", ds.Items[0].Name)
+		assert.Equal(t, "csi-cephfsplugin-holder-rook-ceph", ds.Items[1].Name)
+		assert.Equal(t, "csi-rbdplugin", ds.Items[2].Name)
+		assert.Equal(t, "csi-rbdplugin-holder-rook-ceph", ds.Items[3].Name)
+		assert.Equal(t, "public-net", ds.Items[1].Spec.Template.Annotations["k8s.v1.cni.cncf.io/networks"], ds.Items[1].Spec.Template.Annotations)
+	})
 }
diff --git a/pkg/operator/ceph/csi/spec_test.go b/pkg/operator/ceph/csi/spec_test.go
new file mode 100644
index 000000000..d530e25e1
--- /dev/null
+++ b/pkg/operator/ceph/csi/spec_test.go
@@ -0,0 +1,130 @@
+/*
+Copyright 2022 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package csi
+
+import (
+	"context"
+	_ "embed"
+	"os"
+	"testing"
+
+	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/clusterd"
+	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
+	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
+	"github.com/rook/rook/pkg/operator/k8sutil"
+	testop "github.com/rook/rook/pkg/operator/test"
+	"github.com/stretchr/testify/assert"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+)
+
+func TestReconcileCSI_configureHolders(t *testing.T) {
+	r := &ReconcileCSI{
+		context: &clusterd.Context{
+			Clientset:     testop.New(t, 1),
+			RookClientset: rookclient.NewSimpleClientset(),
+		},
+		opManagerContext: context.TODO(),
+		opConfig:         opcontroller.OperatorConfig{},
+		multusClusters:   []ClusterDetail{},
+	}
+
+	t.Run("no clusters, noop", func(t *testing.T) {
+		err := r.configureHolders([]driverDetails{}, templateParam{}, []v1.Toleration{}, nil)
+		assert.NoError(t, err)
+	})
+
+	t.Run("one multus cluster", func(t *testing.T) {
+		r.opConfig.OperatorNamespace = "rook-ceph"
+		driverDetails := []driverDetails{
+			{
+				name:           "rbd",
+				fullName:       "rbd.csi.ceph.com",
+				holderTemplate: CephFSPluginHolderTemplatePath,
+				nodeAffinity:   cephFSPluginNodeAffinityEnv,
+				toleration:     cephFSPluginTolerationsEnv,
+			},
+		}
+		tp := templateParam{
+			Param:     CSIParam,
+			Namespace: r.opConfig.OperatorNamespace,
+		}
+
+		r.multusClusters = []ClusterDetail{
+			{
+				cluster: &cephv1.CephCluster{
+					ObjectMeta: metav1.ObjectMeta{Namespace: "rook-ceph", Name: "my-cluster"},
+					Spec:       cephv1.ClusterSpec{},
+				},
+				clusterInfo: &cephclient.ClusterInfo{Monitors: map[string]*cephclient.MonInfo{"a": {Name: "a", Endpoint: "10.0.0.1:6789"}}},
+			},
+		}
+
+		os.Setenv(k8sutil.PodNamespaceEnvVar, "rook-ceph")
+		_, err := r.context.Clientset.CoreV1().ConfigMaps("rook-ceph").Create(r.opManagerContext, &v1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: ConfigName, Namespace: "rook-ceph"}, Data: map[string]string{}}, metav1.CreateOptions{})
+		assert.NoError(t, err)
+
+		err = r.configureHolders(driverDetails, tp, []v1.Toleration{}, nil)
+		assert.NoError(t, err)
+	})
+}
+
+func TestGenerateNetNamespaceFilePath(t *testing.T) {
+	ctx := context.TODO()
+
+	t.Run("generate with no op configmap available and non supported driver name", func(t *testing.T) {
+		client := fake.NewClientBuilder().Build()
+		netNsFilePath, err := GenerateNetNamespaceFilePath(ctx, client, "rook-ceph", "rook-ceph", "foo")
+		assert.Error(t, err)
+		assert.Empty(t, "", netNsFilePath)
+	})
+
+	t.Run("generate with no op configmap available for rbd", func(t *testing.T) {
+		client := fake.NewClientBuilder().Build()
+		netNsFilePath, err := GenerateNetNamespaceFilePath(ctx, client, "rook-ceph", "rook-ceph", "rbd")
+		assert.NoError(t, err)
+		assert.Equal(t, "/var/lib/kubelet/plugins/rook-ceph.rbd.csi.ceph.com/rook-ceph.net.ns", netNsFilePath)
+	})
+
+	t.Run("generate with no op configmap available for cephfs", func(t *testing.T) {
+		client := fake.NewClientBuilder().Build()
+		netNsFilePath, err := GenerateNetNamespaceFilePath(ctx, client, "rook-ceph", "rook-ceph", "cephfs")
+		assert.NoError(t, err)
+		assert.Equal(t, "/var/lib/kubelet/plugins/rook-ceph.cephfs.csi.ceph.com/rook-ceph.net.ns", netNsFilePath)
+	})
+
+	t.Run("generate with op configmap for cephfs", func(t *testing.T) {
+		opCm := &v1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      opcontroller.OperatorSettingConfigMapName,
+				Namespace: "rook-ceph",
+			},
+			Data: map[string]string{"ROOK_CSI_KUBELET_DIR_PATH": "/foo"},
+		}
+		object := []runtime.Object{
+			opCm,
+		}
+		client := fake.NewClientBuilder().WithRuntimeObjects(object...).Build()
+		netNsFilePath, err := GenerateNetNamespaceFilePath(ctx, client, "rook-ceph", "rook-ceph", "cephfs")
+		assert.NoError(t, err)
+		assert.Equal(t, "/foo/plugins/rook-ceph.cephfs.csi.ceph.com/rook-ceph.net.ns", netNsFilePath)
+	})
+}
diff --git a/pkg/operator/ceph/csi/version_test.go b/pkg/operator/ceph/csi/version_test.go
index e3a774350..c1ef2bbcd 100644
--- a/pkg/operator/ceph/csi/version_test.go
+++ b/pkg/operator/ceph/csi/version_test.go
@@ -27,6 +27,7 @@ var (
 	testReleaseV340        = CephCSIVersion{3, 4, 0}
 	testReleaseV350        = CephCSIVersion{3, 5, 0}
 	testReleaseV360        = CephCSIVersion{3, 6, 0}
+	testReleaseV361        = CephCSIVersion{3, 6, 1}
 	testVersionUnsupported = CephCSIVersion{4, 0, 0}
 )
 
@@ -112,3 +113,23 @@ func TestSupportsCustomCephConf(t *testing.T) {
 	ret = testReleaseV360.SupportsCustomCephConf()
 	assert.True(t, ret)
 }
+
+func TestSupportsMultus(t *testing.T) {
+	t.Run("AllowUnsupported=true regardless of the version", func(t *testing.T) {
+		AllowUnsupported = true
+		ret := testMinVersion.SupportsMultus()
+		assert.True(t, ret)
+	})
+
+	t.Run("AllowUnsupported=false and version 3.5 is too old", func(t *testing.T) {
+		AllowUnsupported = false
+		ret := testMinVersion.SupportsMultus()
+		assert.False(t, ret)
+	})
+
+	t.Run("AllowUnsupported=false and version 3.6.1 is fine", func(t *testing.T) {
+		AllowUnsupported = false
+		ret := testReleaseV361.SupportsMultus()
+		assert.True(t, ret)
+	})
+}
diff --git a/pkg/operator/ceph/file/subvolumegroup/controller_test.go b/pkg/operator/ceph/file/subvolumegroup/controller_test.go
index 3e9460c41..0cf0d1312 100644
--- a/pkg/operator/ceph/file/subvolumegroup/controller_test.go
+++ b/pkg/operator/ceph/file/subvolumegroup/controller_test.go
@@ -283,6 +283,51 @@ func TestCephClientController(t *testing.T) {
 		assert.Contains(t, cm.Data[csi.ConfigKey], "clusterID")
 		assert.Contains(t, cm.Data[csi.ConfigKey], "group-a")
 	})
+
+	t.Run("success - test with multus ceph cluster", func(t *testing.T) {
+		cephCluster.Spec.External.Enable = false
+		cephCluster.Spec.Network.Provider = "multus"
+		cephCluster.Status.Phase = cephv1.ConditionReady
+		cephCluster.Status.CephStatus.Health = "HEALTH_OK"
+		objects := []runtime.Object{
+			cephFilesystemSubVolumeGroup,
+			cephCluster,
+			cephFilesystem,
+		}
+		// Create a fake client to mock API calls.
+		cl = fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(objects...).Build()
+		c.Client = cl
+
+		s.AddKnownTypes(cephv1.SchemeGroupVersion, &cephv1.CephBlockPoolList{}, &v1.ConfigMap{})
+		// Create a ReconcileCephFilesystemSubVolumeGroup object with the scheme and fake client.
+		r = &ReconcileCephFilesystemSubVolumeGroup{
+			client:           cl,
+			scheme:           s,
+			context:          c,
+			opManagerContext: ctx,
+		}
+
+		// Enable CSI
+		csi.EnableRBD = true
+		os.Setenv("POD_NAMESPACE", namespace)
+		// Create CSI config map
+		ownerRef := &metav1.OwnerReference{}
+		ownerInfo := k8sutil.NewOwnerInfoWithOwnerRef(ownerRef, "")
+		err := csi.CreateCsiConfigMap(ctx, namespace, c.Clientset, ownerInfo)
+		assert.NoError(t, err)
+
+		res, err := r.Reconcile(ctx, req)
+		assert.NoError(t, err)
+		assert.False(t, res.Requeue)
+
+		// test that csi configmap is created
+		cm, err := c.Clientset.CoreV1().ConfigMaps(namespace).Get(ctx, csi.ConfigName, metav1.GetOptions{})
+		assert.NoError(t, err)
+		assert.NotEmpty(t, cm.Data[csi.ConfigKey])
+		assert.Contains(t, cm.Data[csi.ConfigKey], "clusterID")
+		assert.Contains(t, cm.Data[csi.ConfigKey], "group-a")
+		assert.Contains(t, cm.Data[csi.ConfigKey], "netNamespaceFilePath")
+	})
 }
 
 func Test_buildClusterID(t *testing.T) {
diff --git a/tests/scripts/collect-logs.sh b/tests/scripts/collect-logs.sh
index 5cb8ee13b..ad814cdec 100755
--- a/tests/scripts/collect-logs.sh
+++ b/tests/scripts/collect-logs.sh
@@ -27,7 +27,7 @@ for NAMESPACE in "${NAMESPACES[@]}"; do
   mkdir "${NS_DIR}"
 
   # describe every one of the k8s resources in the namespace which rook commonly uses
-  for KIND in 'pod' 'deployment' 'job' 'daemonset'; do
+  for KIND in 'pod' 'deployment' 'job' 'daemonset' 'secret' 'cm'; do
     kubectl -n "$NAMESPACE" get "$KIND" -o wide > "${NS_DIR}"/"$KIND"-list.txt
     for resource in $(kubectl -n "$NAMESPACE" get "$KIND" -o jsonpath='{.items[*].metadata.name}'); do
       kubectl -n "$NAMESPACE" describe "$KIND" "$resource" > "${NS_DIR}"/"$KIND"-describe--"$resource".txt
diff --git a/tests/scripts/github-action-helper.sh b/tests/scripts/github-action-helper.sh
index 6582739d7..bf54f1352 100755
--- a/tests/scripts/github-action-helper.sh
+++ b/tests/scripts/github-action-helper.sh
@@ -427,6 +427,50 @@ function deploy_multus_cluster() {
   deploy_manifest_with_local_build toolbox.yaml
   sed -i "s|#deviceFilter:|deviceFilter: ${BLOCK/\/dev\/}|g" cluster-multus-test.yaml
   kubectl create -f cluster-multus-test.yaml
+  kubectl create -f filesystem-test.yaml
+}
+
+function wait_for_ceph_csi_configmap_to_be_updated {
+  timeout 60 bash <<EOF
+until [[ $(kubectl -n rook-ceph get configmap rook-ceph-csi-config  -o jsonpath="{.data.csi-cluster-config-json}" | jq .[0].rbd.netNamespaceFilePath) != "null" ]]; do
+  echo "waiting for ceph csi configmap to be updated with rbd.netNamespaceFilePath"
+  sleep 5
+done
+EOF
+  timeout 60 bash <<EOF
+until [[ $(kubectl -n rook-ceph get configmap rook-ceph-csi-config  -o jsonpath="{.data.csi-cluster-config-json}" | jq .[0].cephFS.netNamespaceFilePath) != "null" ]]; do
+  echo "waiting for ceph csi configmap to be updated with cephFS.netNamespaceFilePath"
+  sleep 5
+done
+EOF
+}
+
+function test_csi_rbd_workload {
+  cd deploy/examples/csi/rbd
+  sed -i 's|size: 3|size: 1|g' storageclass.yaml
+  sed -i 's|requireSafeReplicaSize: true|requireSafeReplicaSize: false|g' storageclass.yaml
+  kubectl create -f storageclass.yaml
+  kubectl create -f pvc.yaml
+  kubectl create -f pod.yaml
+  timeout 45 sh -c 'until kubectl exec -t pod/csirbd-demo-pod -- dd if=/dev/random of=/var/lib/www/html/test bs=1M count=1; do echo "waiting for test pod to be ready" && sleep 1; done'
+  kubectl exec -t pod/csirbd-demo-pod -- dd if=/dev/random of=/var/lib/www/html/test oflag=direct bs=1M count=1
+  kubectl -n rook-ceph logs ds/csi-rbdplugin -c csi-rbdplugin
+  kubectl -n rook-ceph delete "$(kubectl -n rook-ceph get pod --selector=app=csi-rbdplugin --field-selector=status.phase=Running -o name)"
+  kubectl exec -t pod/csirbd-demo-pod -- dd if=/dev/random of=/var/lib/www/html/test1 oflag=direct bs=1M count=1
+  kubectl exec -t pod/csirbd-demo-pod -- ls -alh /var/lib/www/html/
+}
+
+function test_csi_cephfs_workload {
+  cd deploy/examples/csi/cephfs
+  kubectl create -f storageclass.yaml
+  kubectl create -f pvc.yaml
+  kubectl create -f pod.yaml
+  timeout 45 sh -c 'until kubectl exec -t pod/csicephfs-demo-pod -- dd if=/dev/random of=/var/lib/www/html/test bs=1M count=1; do echo "waiting for test pod to be ready" && sleep 1; done'
+  kubectl exec -t pod/csicephfs-demo-pod -- dd if=/dev/random of=/var/lib/www/html/test oflag=direct bs=1M count=1
+  kubectl -n rook-ceph logs ds/csi-cephfsplugin -c csi-cephfsplugin
+  kubectl -n rook-ceph delete "$(kubectl -n rook-ceph get pod --selector=app=csi-cephfsplugin --field-selector=status.phase=Running -o name)"
+  kubectl exec -t pod/csicephfs-demo-pod -- dd if=/dev/random of=/var/lib/www/html/test1 oflag=direct bs=1M count=1
+  kubectl exec -t pod/csicephfs-demo-pod -- ls -alh /var/lib/www/html/
 }
 
 FUNCTION="$1"
diff --git a/tests/scripts/validate_cluster.sh b/tests/scripts/validate_cluster.sh
index d0b8bf799..ac170130d 100755
--- a/tests/scripts/validate_cluster.sh
+++ b/tests/scripts/validate_cluster.sh
@@ -88,11 +88,22 @@ function test_demo_pool {
 }
 
 function test_csi {
-  timeout 360 bash <<-'EOF'
-    until [[ "$(kubectl -n rook-ceph get pods --field-selector=status.phase=Running|grep -c ^csi-)" -ge 4 ]]; do
-      echo "waiting for csi pods to be ready"
-      sleep 5
-    done
+  timeout 360 bash -x <<-'EOF'
+    echo $IS_MULTUS
+    if [ -z $IS_MULTUS ]; then
+      until [[ "$(kubectl -n rook-ceph get pods --field-selector=status.phase=Running|grep -c ^csi-)" -eq 4 ]]; do
+        echo "waiting for csi pods to be ready"
+        sleep 5
+      done
+    else
+      until [[ "$(kubectl -n rook-ceph get pods --field-selector=status.phase=Running|grep -c ^csi-)" -eq 6 ]]; do
+        echo "waiting for csi pods to be ready with multus"
+        sleep 5
+      done
+      echo "verifying csi holder interfaces (multus ones must be present)"
+      kubectl -n rook-ceph exec -t ds/csi-rbdplugin-holder-my-cluster -- grep net /proc/net/dev
+      kubectl -n rook-ceph exec -t ds/csi-cephfsplugin-holder-my-cluster -- grep net /proc/net/dev
+    fi
 EOF
 }
 
@@ -105,6 +116,13 @@ function test_nfs {
 EOF
 }
 
+function test_multus_osd {
+  for i in $(seq 1 2); do
+    kubectl -n rook-ceph exec -t deploy/rook-ceph-osd-0 -c osd -- grep net"$i" /proc/net/dev
+    kubectl -n rook-ceph exec -t deploy/rook-ceph-osd-0 -c osd -- grep net"$i" /proc/net/dev
+  done
+}
+
 ########
 # MAIN #
 ########
@@ -136,6 +154,10 @@ for daemon in $daemons_list; do
     osd)
       test_demo_osd
       ;;
+    osd_multus)
+      test_demo_osd
+      test_multus_osd
+      ;;
     mds)
       test_demo_mds
       ;;