Repository: rook/rook
Pull request #15400 (2025-02-14 12:15:36)

Implement external arbiter mon
Implements #14733, follow-up to discussion in #15210

First draft of ext mon implementaion. Ready for review. I was able to test it locally and it seems working for diffrent use-cases like adding/removing ext mon id to spec, upscaling/downcaling internal mons when ext mon is presented. 

Feature adds the following behaviour: If mon if is presented in `spec.mon.externalMonIDs: []` then rook will not manage it and **not** include such montiror in `monCount`. If external mon is also in quorum, then rook will add its ip address to all secrets and configMaps with mon endopoints. All other exising behaviour remains unchaged.

TODO:
 

- [x] add tests
- [x]  add docs


arttor: 
> 
> One question... With this feature, now `mon.count` would be 4 in stretch clusters, plus an external arbiter, right? There is a check somewhere that enforces 5 mons for a stretch cluster that likely still needs to be changed.

I was thinking to use it only for **non-stretched** cluster and current implementation follows this logic:
-  `mon.count` is only responsible for mons which are managed by Rook. If `mon.count` is different from number of mons running in local rook cluster, then rook will upscale/downscale local mons and ignore number of external monitors.
- if external mon is down rook will do nothing. the only thing rook does about external mon is: if mon id is in `externalMonIDs` rook will not remove it from quorum and add its ip to mon endpoints configmaps/secrets.


arttor: PR is ready for review. Added tests and docs and will do another round of local tests just in case.
arttor: Updated docs, tested edge-cases locally. Ready for review/merge.
arttor: applied doc suggestions and squashed commits.

Patch:
diff --git a/Documentation/CRDs/Cluster/ceph-cluster-crd.md b/Documentation/CRDs/Cluster/ceph-cluster-crd.md
index 51b4e4fc1..eea150e38 100755
--- a/Documentation/CRDs/Cluster/ceph-cluster-crd.md
+++ b/Documentation/CRDs/Cluster/ceph-cluster-crd.md
@@ -150,6 +150,13 @@ A specific will contain a specific release of Ceph as well as security fixes fro
 * `failureDomainLabel`: The label that is expected on each node where the mons
     are expected to be deployed. The labels must be found in the list of
     well-known [topology labels](#osd-topology).
+* `externalMonIDs`: ID list of external mons deployed outside of Rook cluster
+    and not managed by Rook. If set, Rook will not remove external mons from quorum
+    and populate external mons addresses to mon endpoints for CSI.
+    This parameter is supported only for local Rook cluster running in normal mode,
+    meaning that it will be ignored for external cluster (`spec.external.enabled: true`)
+    or for `stretchedCluster`.
+    For more details see [external mons](../../Storage-Configuration/Advanced/ceph-mon-health.md#external-monitors).
 * `zones`: The failure domain names where the Mons are expected to be deployed.
     There must be **at least three zones** specified in the list. Each zone can be
     backed by a different storage class by specifying the `volumeClaimTemplate`.
diff --git a/Documentation/CRDs/specification.md b/Documentation/CRDs/specification.md
index 99e2000fd..ca91f6d29 100644
--- a/Documentation/CRDs/specification.md
+++ b/Documentation/CRDs/specification.md
@@ -8777,6 +8777,22 @@ VolumeClaimTemplate
 <p>VolumeClaimTemplate is the PVC definition</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>externalMonIDs</code><br/>
+<em>
+[]string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>ExternalMonIDs - optional list of monitor IDs which are deployed externally and not managed by Rook.
+If set, Rook will not remove mons with given IDs from quorum.
+This parameter is used only for local Rook cluster running in normal mode
+and will be ignored if external or stretched mode is used.
+leading</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="ceph.rook.io/v1.MonZoneSpec">MonZoneSpec
diff --git a/Documentation/Storage-Configuration/Advanced/ceph-mon-health.md b/Documentation/Storage-Configuration/Advanced/ceph-mon-health.md
index f39c931b3..5e69b9ac2 100644
--- a/Documentation/Storage-Configuration/Advanced/ceph-mon-health.md
+++ b/Documentation/Storage-Configuration/Advanced/ceph-mon-health.md
@@ -124,3 +124,43 @@ CephCluster CR:
 - `spec.network.hostNetwork`: When enabled or disabled, Rook fails over all monitors, configuring them to enable or disable host networking.
 - `spec.network.Provider` : When updated from being empty to "host", Rook fails over all monitors, configuring them to enable or disable host networking.
 - `spec.network.multiClusterService`: When enabled or disabled, Rook fails over all monitors, configuring them to start (or stop) using service IPs compatible with the multi-cluster service.
+
+## External Monitors
+
+!!! attention
+    This feature is experimental.
+
+It is possible to have both Rook-managed and external monitors in the same Rook cluster.
+One use case for this is 2 datacenter aka 2-AZ (Availability Zone) setup. For 2-AZ setup, Zone outage will lead to loss of the k8s control plane and half of the worker nodes hosting Rook mons.
+In this case remaining half of the Rook cluster will not be able to form quorum and will be in a stuck state even if the other half of worker nodes are still up.
+To avoid this situation, external mons can be used to form quorum and keep the cluster running.
+
+If there are external monitors, Rook must be aware of them, otherwise Rook will remove the unknown mons from the quorum. This is done by setting the `mon.externalMonIDs` field in the CephCluster CR. The `mon.count` ignores the number of `mon.externalMonIDs`. For example, if `mon.count = 2`, Rook will create two internal mons no matter how many external mons are in the cluster and no matter what their health state might be. External monitors are supported only for the local Rook cluster running in normal mode. The external mons will be ignored for external clusters and stretch clusters.
+
+Here is a step-by-step guide on how to add external monitors to a Rook cluster:
+
+1. Create a CephCluster CR with the `mon.externalMonIDs` field set to the external monitor IDs. For example:
+
+    ```yaml
+    spec:
+      mon:
+        # Spawn 2 Mons - one Mon in each AZ managed by Rook
+        count: 2
+        allowMultiplePerNode: false
+        # ID of external Mon
+        externalMonIDs:
+        - ext-mon-1 
+    ```
+
+    This will tell Rook to create two internal monitors in the cluster and to keep the external monitor with the ID `ext-mon-1` if it is found in the quorum.
+    It is also possible to add `externalMonIDs` to an existing Cluster.
+2. Wait until the Rook cluster and internal mons are up and running.
+3. Manually deploy an external monitor with the ID `ext-mon-1` outside of the Rook cluster and its availability zones. See [ceph guide](https://docs.ceph.com/en/latest/rados/operations/add-or-rm-mons/#adding-removing-monitors) on deploying monitors.
+4. Move the external Mon to [disallow-mode](https://docs.ceph.com/en/reef/rados/operations/change-mon-elections/#rados-operations-disallow-mode) to make sure that it won't be elected as a leader. The purpose of external mode is maintaining quorum to elect a new leader in case of a zone outage. The external Mon will likely have higher latency to the cluster so it should not be elected.
+5. Check that the external monitor is in the quorum by running `ceph status` or `ceph quorum_status` from the toolbox.
+6. Check that the external mon is added to the Rook mon endpoints:
+
+    ```console
+    $ kubectl -n rook-ceph get cm rook-ceph-mon-endpoints -o jsonpath='{.data.data}'
+    a=10.100.68.61:6789,b=10.103.201.172:6789,ext-mon-1=10.102.136.102:6789
+    ```
diff --git a/PendingReleaseNotes.md b/PendingReleaseNotes.md
index 252f70b28..faf45c2b6 100644
--- a/PendingReleaseNotes.md
+++ b/PendingReleaseNotes.md
@@ -13,3 +13,5 @@ Object:
     See https://github.com/rook/rook/pull/15376 for more information.
 
 ## Features
+
+- Support external mons for local Rook cluster (see [#14733](https://github.com/rook/rook/issues/14733)).
diff --git a/cmd/rook/ceph/mgr.go b/cmd/rook/ceph/mgr.go
index 372beb449..8b8834de1 100644
--- a/cmd/rook/ceph/mgr.go
+++ b/cmd/rook/ceph/mgr.go
@@ -79,7 +79,7 @@ func runMgrSidecar(cmd *cobra.Command, args []string) error {
 	}
 
 	context := createContext()
-	clusterInfo.Monitors = opcontroller.ParseMonEndpoints(cfg.monEndpoints)
+	clusterInfo.InternalMonitors = opcontroller.ParseMonEndpoints(cfg.monEndpoints)
 	rook.LogStartupInfo(mgrSidecarCmd.Flags())
 
 	ownerRef := opcontroller.ClusterOwnerRef(clusterName, ownerRefID)
diff --git a/cmd/rook/ceph/osd.go b/cmd/rook/ceph/osd.go
index d4f0caf5f..c231c4b95 100644
--- a/cmd/rook/ceph/osd.go
+++ b/cmd/rook/ceph/osd.go
@@ -344,7 +344,7 @@ func commonOSDInit(cmd *cobra.Command) {
 	rook.SetLogLevel()
 	rook.LogStartupInfo(cmd.Flags())
 
-	clusterInfo.Monitors = opcontroller.ParseMonEndpoints(cfg.monEndpoints)
+	clusterInfo.InternalMonitors = opcontroller.ParseMonEndpoints(cfg.monEndpoints)
 }
 
 // use zone/region/hostname labels in the crushmap
diff --git a/deploy/charts/rook-ceph/templates/resources.yaml b/deploy/charts/rook-ceph/templates/resources.yaml
index 53a806f02..1eda882a4 100644
--- a/deploy/charts/rook-ceph/templates/resources.yaml
+++ b/deploy/charts/rook-ceph/templates/resources.yaml
@@ -1789,6 +1789,16 @@ spec:
                       maximum: 9
                       minimum: 0
                       type: integer
+                    externalMonIDs:
+                      description: |-
+                        ExternalMonIDs - optional list of monitor IDs which are deployed externally and not managed by Rook.
+                        If set, Rook will not remove mons with given IDs from quorum.
+                        This parameter is used only for local Rook cluster running in normal mode
+                        and will be ignored if external or stretched mode is used.
+                        leading
+                      items:
+                        type: string
+                      type: array
                     failureDomainLabel:
                       type: string
                     stretchCluster:
diff --git a/deploy/examples/crds.yaml b/deploy/examples/crds.yaml
index 25b576af4..9446d8900 100644
--- a/deploy/examples/crds.yaml
+++ b/deploy/examples/crds.yaml
@@ -1787,6 +1787,16 @@ spec:
                       maximum: 9
                       minimum: 0
                       type: integer
+                    externalMonIDs:
+                      description: |-
+                        ExternalMonIDs - optional list of monitor IDs which are deployed externally and not managed by Rook.
+                        If set, Rook will not remove mons with given IDs from quorum.
+                        This parameter is used only for local Rook cluster running in normal mode
+                        and will be ignored if external or stretched mode is used.
+                        leading
+                      items:
+                        type: string
+                      type: array
                     failureDomainLabel:
                       type: string
                     stretchCluster:
diff --git a/pkg/apis/ceph.rook.io/v1/types.go b/pkg/apis/ceph.rook.io/v1/types.go
index de8c70cce..c6076056a 100755
--- a/pkg/apis/ceph.rook.io/v1/types.go
+++ b/pkg/apis/ceph.rook.io/v1/types.go
@@ -626,6 +626,13 @@ type MonSpec struct {
 	// +kubebuilder:pruning:PreserveUnknownFields
 	// +optional
 	VolumeClaimTemplate *VolumeClaimTemplate `json:"volumeClaimTemplate,omitempty"`
+	// ExternalMonIDs - optional list of monitor IDs which are deployed externally and not managed by Rook.
+	// If set, Rook will not remove mons with given IDs from quorum.
+	// This parameter is used only for local Rook cluster running in normal mode
+	// and will be ignored if external or stretched mode is used.
+	// leading
+	// +optional
+	ExternalMonIDs []string `json:"externalMonIDs,omitempty"`
 }
 
 // VolumeClaimTemplate is a simplified version of K8s corev1's PVC. It has no type meta or status.
diff --git a/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go b/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go
index 90d36b973..b71595f86 100644
--- a/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go
+++ b/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go
@@ -3346,6 +3346,11 @@ func (in *MonSpec) DeepCopyInto(out *MonSpec) {
 		*out = new(VolumeClaimTemplate)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.ExternalMonIDs != nil {
+		in, out := &in.ExternalMonIDs, &out.ExternalMonIDs
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
 	return
 }
 
diff --git a/pkg/daemon/ceph/client/config.go b/pkg/daemon/ceph/client/config.go
index f2cc6119b..278c37445 100644
--- a/pkg/daemon/ceph/client/config.go
+++ b/pkg/daemon/ceph/client/config.go
@@ -249,7 +249,7 @@ func PopulateMonHostMembers(clusterInfo *ClusterInfo) ([]string, []string) {
 	var monMembers []string
 	var monHosts []string
 
-	for _, monitor := range clusterInfo.Monitors {
+	for _, monitor := range clusterInfo.AllMonitors() {
 		if monitor.OutOfQuorum {
 			logger.Warningf("skipping adding mon %q to config file, detected out of quorum", monitor.Name)
 			continue
diff --git a/pkg/daemon/ceph/client/info.go b/pkg/daemon/ceph/client/info.go
index 70e49bc48..79d273471 100644
--- a/pkg/daemon/ceph/client/info.go
+++ b/pkg/daemon/ceph/client/info.go
@@ -39,10 +39,13 @@ type ClusterInfo struct {
 	FSID          string
 	MonitorSecret string
 	CephCred      CephCred
-	Monitors      map[string]*MonInfo
-	CephVersion   cephver.CephVersion
-	Namespace     string
-	OwnerInfo     *k8sutil.OwnerInfo
+	// InternalMonitors - montiros managed by Rook or external monitors when Rook manages external cluster.
+	InternalMonitors map[string]*MonInfo
+	// ExternalMons - external montiros listed in CephCluster.spec.mon.externalMonIDs when Rook managing local cluster.
+	ExternalMons map[string]*MonInfo
+	CephVersion  cephver.CephVersion
+	Namespace    string
+	OwnerInfo    *k8sutil.OwnerInfo
 	// Hide the name of the cluster since in 99% of uses we want to use the cluster namespace.
 	// If the CR name is needed, access it through the NamespacedName() method.
 	name              string
@@ -59,6 +62,20 @@ type ClusterInfo struct {
 	Context context.Context
 }
 
+func (c *ClusterInfo) AllMonitors() map[string]*MonInfo {
+	if len(c.ExternalMons) == 0 {
+		return c.InternalMonitors
+	}
+	res := make(map[string]*MonInfo, len(c.InternalMonitors)+len(c.ExternalMons))
+	for id, mon := range c.InternalMonitors {
+		res[id] = mon
+	}
+	for id, mon := range c.ExternalMons {
+		res[id] = mon
+	}
+	return res
+}
+
 // MonInfo is a collection of information about a Ceph mon.
 type MonInfo struct {
 	Name     string `json:"name"`
diff --git a/pkg/daemon/ceph/client/mirror.go b/pkg/daemon/ceph/client/mirror.go
index d9c2dc8a8..a0795c892 100644
--- a/pkg/daemon/ceph/client/mirror.go
+++ b/pkg/daemon/ceph/client/mirror.go
@@ -441,7 +441,7 @@ func CreateRBDMirrorBootstrapPeerWithoutPool(context *clusterd.Context, clusterI
 	logger.Infof("successfully created rbd-mirror bootstrap peer token for cluster %q", clusterInfo.NamespacedName().Name)
 
 	mons := sets.New[string]()
-	for _, mon := range clusterInfo.Monitors {
+	for _, mon := range clusterInfo.AllMonitors() {
 		mons.Insert(mon.Endpoint)
 	}
 
diff --git a/pkg/operator/ceph/cluster/cluster_external.go b/pkg/operator/ceph/cluster/cluster_external.go
index 37f825263..b9b14121b 100644
--- a/pkg/operator/ceph/cluster/cluster_external.go
+++ b/pkg/operator/ceph/cluster/cluster_external.go
@@ -109,7 +109,7 @@ func (c *ClusterController) configureExternalCephCluster(cluster *cluster) error
 	}
 
 	// update the msgr2 flag
-	for _, m := range cluster.ClusterInfo.Monitors {
+	for _, m := range cluster.ClusterInfo.InternalMonitors {
 		// m.Endpoint=10.1.115.104:3300
 		monPort := util.GetPortFromEndpoint(m.Endpoint)
 		if monPort == client.Msgr2port {
@@ -123,7 +123,7 @@ func (c *ClusterController) configureExternalCephCluster(cluster *cluster) error
 	}
 
 	// Save CSI configmap
-	monEndpoints := csi.MonEndpoints(cluster.ClusterInfo.Monitors, cluster.Spec.RequireMsgr2())
+	monEndpoints := csi.MonEndpoints(cluster.ClusterInfo.InternalMonitors, cluster.Spec.RequireMsgr2())
 	csiConfigEntry := &csi.CSIClusterConfigEntry{
 		Namespace: cluster.ClusterInfo.Namespace,
 		ClusterInfo: cephcsi.ClusterInfo{
diff --git a/pkg/operator/ceph/cluster/mon/health.go b/pkg/operator/ceph/cluster/mon/health.go
index 050d88f43..fe1042ae8 100644
--- a/pkg/operator/ceph/cluster/mon/health.go
+++ b/pkg/operator/ceph/cluster/mon/health.go
@@ -1,5 +1,4 @@
-/*
-Copyright 2018 The Rook Authors. All rights reserved.
+/* Copyright 2018 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.
@@ -20,6 +19,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"slices"
 	"strings"
 	"time"
 
@@ -202,6 +202,12 @@ func (c *Cluster) checkHealth(ctx context.Context) error {
 	}
 	logger.Debugf("Mon quorum status: %+v", quorumStatus)
 
+	// handle external Mons
+	quorumStatus, err = c.reconcileExternalMons(ctx, quorumStatus)
+	if err != nil {
+		return errors.Wrap(err, "failed to check external mons health")
+	}
+
 	// Use a local mon count in case the user updates the crd in another goroutine.
 	// We need to complete a health check with a consistent value.
 	desiredMonCount := c.spec.Mon.Count
@@ -209,7 +215,7 @@ func (c *Cluster) checkHealth(ctx context.Context) error {
 
 	// Source of truth of which mons should exist is our *clusterInfo*
 	monsNotFound := map[string]interface{}{}
-	for _, mon := range c.ClusterInfo.Monitors {
+	for _, mon := range c.ClusterInfo.InternalMonitors {
 		monsNotFound[mon.Name] = struct{}{}
 	}
 
@@ -314,7 +320,7 @@ func (c *Cluster) checkHealth(ctx context.Context) error {
 	// handle all mons that haven't been in the Ceph mon map
 	for mon := range monsNotFound {
 		logger.Warningf("mon %s NOT found in ceph mon map, failover", mon)
-		c.failMon(len(c.ClusterInfo.Monitors), desiredMonCount, mon)
+		c.failMon(len(c.ClusterInfo.InternalMonitors), desiredMonCount, mon)
 		// only deal with one "not found in ceph mon map" mon per health check
 		return nil
 	}
@@ -351,24 +357,24 @@ func (c *Cluster) checkHealth(ctx context.Context) error {
 
 	// failover mon if `multiClusterService` is enabled but mon service is not exported
 	if allMonsInQuorum && c.spec.Network.MultiClusterService.Enabled {
-		for _, mon := range c.ClusterInfo.Monitors {
+		for _, mon := range c.ClusterInfo.InternalMonitors {
 			monResourceName := resourceName(mon.Name)
 			isAlreadyExported, err := k8sutil.IsServiceExported(c.ClusterInfo.Context, c.context, monResourceName, c.ClusterInfo.Namespace)
 			if err != nil {
 				return errors.Wrapf(err, "failed to check if the service %q is already exported", mon.Name)
 			}
 			if !isAlreadyExported {
-				c.failMon(len(c.ClusterInfo.Monitors), desiredMonCount, mon.Name)
+				c.failMon(len(c.ClusterInfo.InternalMonitors), desiredMonCount, mon.Name)
 				return nil
 			}
 		}
 	}
 
 	// failover any mons present in the mon fail over list
-	for _, mon := range c.ClusterInfo.Monitors {
+	for _, mon := range c.ClusterInfo.InternalMonitors {
 		if c.monsToFailover.Has(mon.Name) {
 			logger.Infof("fail over mon %q from the mon fail over list", mon.Name)
-			c.failMon(len(c.ClusterInfo.Monitors), desiredMonCount, mon.Name)
+			c.failMon(len(c.ClusterInfo.InternalMonitors), desiredMonCount, mon.Name)
 			c.monsToFailover.Delete(mon.Name)
 			return nil
 		}
@@ -377,12 +383,102 @@ func (c *Cluster) checkHealth(ctx context.Context) error {
 	return nil
 }
 
+// reconcileExternalMons handling external monitors defined in CephCluster.spec.mon.externalMonIDs when Rook managing local cluster.
+func (c *Cluster) reconcileExternalMons(ctx context.Context, quorumStatus cephclient.MonStatusResponse) (cephclient.MonStatusResponse, error) {
+	if len(c.spec.Mon.ExternalMonIDs) != 0 {
+		if c.spec.External.Enable {
+			logger.Warning("ignore external mon IDs in cluster spec: cluster is running in external mode")
+			return quorumStatus, nil
+		}
+		if c.spec.IsStretchCluster() {
+			logger.Warning("ignore external mon IDs in cluster spec: cluster is running in stretch mode")
+			return quorumStatus, nil
+		}
+	}
+
+	extMonIDs := c.spec.Mon.ExternalMonIDs
+	if c.ClusterInfo.ExternalMons == nil {
+		c.ClusterInfo.ExternalMons = make(map[string]*cephclient.MonInfo, len(extMonIDs))
+	}
+
+	extMonsChanged := false
+	for extID := range c.ClusterInfo.ExternalMons {
+		if slices.Contains(extMonIDs, extID) {
+			continue
+		}
+		// existing external mon was removed from Cluster CRD spec:
+		// remove it from CLusterInfo
+		logger.Debugf("existing external mon %q was removed from spec: removing it", extID)
+		delete(c.ClusterInfo.ExternalMons, extID)
+		extMonsChanged = true
+	}
+
+	// handle external monitors if configured in cluster CRD:
+	logger.Debugf("external mon IDs: %v", extMonIDs)
+	for _, extID := range extMonIDs {
+		monStatus, inQuorum := getMonByID(extID, quorumStatus)
+		if inQuorum {
+			logger.Debugf("external mon %q in quorum", extID)
+		} else {
+			logger.Debugf("external mon %q not in quorum %+v, %+v", extID, quorumStatus.Quorum, quorumStatus.MonMap.Mons)
+		}
+		_, inInfo := c.ClusterInfo.ExternalMons[extID]
+		if inQuorum && !inInfo {
+			// add newly discovered external mon to cluster info:
+			monInfo := monStatusToInfo(monStatus)
+			c.ClusterInfo.ExternalMons[extID] = monInfo
+			extMonsChanged = true
+			logger.Infof("new external mon %q found: %s, adding it", extID, monInfo.Endpoint)
+		} else if !inQuorum && inInfo {
+			// remove external mon from cluster info if it is out of quorum:
+			delete(c.ClusterInfo.ExternalMons, extID)
+			extMonsChanged = true
+			logger.Infof("new external mon %q not in quorum: removing it", extID)
+		}
+	}
+	if extMonsChanged {
+		// update config if external mon was removed or added:
+		if err := c.saveMonConfig(); err != nil {
+			return cephclient.MonStatusResponse{}, errors.Wrap(err, "failed to save mon config after adding/removing external mon")
+		}
+	}
+
+	// now, remove processed external mons from ceph quorum status response
+	// to not affect existing logic processing internal mons that are within the mon.count
+	quorumStatus = removeMonsFromQuorumStatusResponse(quorumStatus, extMonIDs)
+	return quorumStatus, nil
+}
+
+func removeMonsFromQuorumStatusResponse(quorumStatus cephclient.MonStatusResponse, idsToRemove []string) cephclient.MonStatusResponse {
+	var removeFromQuorum []int
+	var keepMons []cephclient.MonMapEntry
+
+	for _, mon := range quorumStatus.MonMap.Mons {
+		if !slices.Contains(idsToRemove, mon.Name) {
+			keepMons = append(keepMons, mon)
+			continue
+		}
+		removeFromQuorum = append(removeFromQuorum, mon.Rank)
+	}
+
+	var keepQuorum []int
+	for _, rank := range quorumStatus.Quorum {
+		if slices.Contains(removeFromQuorum, rank) {
+			continue
+		}
+		keepQuorum = append(keepQuorum, rank)
+	}
+	quorumStatus.MonMap.Mons = keepMons
+	quorumStatus.Quorum = keepQuorum
+	return quorumStatus
+}
+
 func (c *Cluster) trackMonInOrOutOfQuorum(monName string, inQuorum bool) (bool, error) {
 	updateNeeded := false
 	var monsOutOfQuorum []string
 	if monName == "" {
 		// All mons are in quorum, so make sure no mons are marked out of quorum
-		for monName, mon := range c.ClusterInfo.Monitors {
+		for monName, mon := range c.ClusterInfo.InternalMonitors {
 			if mon.OutOfQuorum {
 				logger.Infof("resetting mon %q to be back in quorum", monName)
 				mon.OutOfQuorum = false
@@ -390,7 +486,7 @@ func (c *Cluster) trackMonInOrOutOfQuorum(monName string, inQuorum bool) (bool,
 			}
 		}
 	} else {
-		mon, ok := c.ClusterInfo.Monitors[monName]
+		mon, ok := c.ClusterInfo.InternalMonitors[monName]
 		if !ok {
 			logger.Infof("mon %q not found to keep track of being out of quorum", monName)
 			return false, nil
@@ -675,7 +771,7 @@ func (c *Cluster) failoverMon(name string) error {
 			m.PublicIP = monService.Spec.ClusterIP
 		}
 	}
-	c.ClusterInfo.Monitors[m.DaemonName] = cephclient.NewMonInfo(m.DaemonName, m.PublicIP, m.Port)
+	c.ClusterInfo.InternalMonitors[m.DaemonName] = cephclient.NewMonInfo(m.DaemonName, m.PublicIP, m.Port)
 
 	// Start the deployment
 	newMonMightBeInQuorum = true
@@ -717,7 +813,7 @@ func (c *Cluster) removeMonWithOptionalQuorum(daemonName string, shouldRemoveFro
 			logger.Errorf("failed to remove mon %q from quorum. %v", daemonName, err)
 		}
 	}
-	delete(c.ClusterInfo.Monitors, daemonName)
+	delete(c.ClusterInfo.InternalMonitors, daemonName)
 	delete(c.mapping.Schedule, daemonName)
 
 	if err := c.saveMonConfig(); err != nil {
@@ -813,9 +909,9 @@ func (c *Cluster) addOrRemoveExternalMonitor(status cephclient.MonStatusResponse
 	// clearing the content of clusterinfo monitors
 	// and populate oldClusterInfoMonitors with monitors from clusterinfo
 	// later c.ClusterInfo.Monitors get populated again
-	for monName, mon := range c.ClusterInfo.Monitors {
+	for monName, mon := range c.ClusterInfo.InternalMonitors {
 		oldClusterInfoMonitors[mon.Name] = mon
-		delete(c.ClusterInfo.Monitors, monName)
+		delete(c.ClusterInfo.InternalMonitors, monName)
 	}
 	logger.Debugf("ClusterInfo is now Empty, refilling it from status.MonMap.Mons")
 
@@ -831,17 +927,9 @@ func (c *Cluster) addOrRemoveExternalMonitor(status cephclient.MonStatusResponse
 		if _, ok := oldClusterInfoMonitors[mon.Name]; !ok {
 			// If the mon is part of the quorum
 			if inQuorum {
-				// let's add it to ClusterInfo
-				// FYI mon.PublicAddr is "10.97.171.131:6789/0"
-				// so we need to remove '/0'
-				endpointSlash := strings.Split(mon.PublicAddr, "/")
-				endpoint := endpointSlash[0]
-
-				// find IP and Port of that Mon
-				monIP := cephutil.GetIPFromEndpoint(endpoint)
-				monPort := cephutil.GetPortFromEndpoint(endpoint)
-				logger.Infof("new external mon %q found: %s, adding it", mon.Name, endpoint)
-				c.ClusterInfo.Monitors[mon.Name] = cephclient.NewMonInfo(mon.Name, monIP, monPort)
+				info := monStatusToInfo(mon)
+				logger.Infof("new external mon %q found: %s, adding it", mon.Name, info.Endpoint)
+				c.ClusterInfo.InternalMonitors[mon.Name] = info
 			} else {
 				logger.Debugf("mon %q is not in quorum and not in ClusterInfo", mon.Name)
 			}
@@ -857,7 +945,7 @@ func (c *Cluster) addOrRemoveExternalMonitor(status cephclient.MonStatusResponse
 			} else {
 				// this mon was in clusterInfo and is still in the quorum
 				// add it again
-				c.ClusterInfo.Monitors[mon.Name] = oldClusterInfoMonitors[mon.Name]
+				c.ClusterInfo.InternalMonitors[mon.Name] = oldClusterInfoMonitors[mon.Name]
 				logger.Debugf("everything is fine mon %q in the clusterInfo and its quorum status is %v", mon.Name, inQuorum)
 			}
 		}
@@ -865,20 +953,33 @@ func (c *Cluster) addOrRemoveExternalMonitor(status cephclient.MonStatusResponse
 	// compare old clusterInfo with new ClusterInfo
 	// if length differ -> the are different
 	// then check if all elements are the same
-	if len(oldClusterInfoMonitors) != len(c.ClusterInfo.Monitors) {
+	if len(oldClusterInfoMonitors) != len(c.ClusterInfo.InternalMonitors) {
 		changed = true
 	} else {
-		for _, mon := range c.ClusterInfo.Monitors {
+		for _, mon := range c.ClusterInfo.InternalMonitors {
 			if old, ok := oldClusterInfoMonitors[mon.Name]; !ok || *old != *mon {
 				changed = true
 			}
 		}
 	}
 
-	logger.Debugf("ClusterInfo.Monitors is %+v", c.ClusterInfo.Monitors)
+	logger.Debugf("ClusterInfo.Monitors is %+v", c.ClusterInfo.InternalMonitors)
 	return changed, nil
 }
 
+func monStatusToInfo(mon cephclient.MonMapEntry) *cephclient.MonInfo {
+	// let's add it to ClusterInfo
+	// FYI mon.PublicAddr is "10.97.171.131:6789/0"
+	// so we need to remove '/0'
+	endpointSlash := strings.Split(mon.PublicAddr, "/")
+	endpoint := endpointSlash[0]
+
+	// find IP and Port of that Mon
+	monIP := cephutil.GetIPFromEndpoint(endpoint)
+	monPort := cephutil.GetPortFromEndpoint(endpoint)
+	return cephclient.NewMonInfo(mon.Name, monIP, monPort)
+}
+
 func (c *Cluster) evictMonIfMultipleOnSameNode() error {
 	if c.spec.Mon.AllowMultiplePerNode {
 		logger.Debug("skipping check for multiple mons on same node since multiple mons are allowed")
diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go
index df8fc7ff3..4d059b39c 100644
--- a/pkg/operator/ceph/cluster/mon/mon.go
+++ b/pkg/operator/ceph/cluster/mon/mon.go
@@ -58,6 +58,8 @@ const (
 	EndpointConfigMapName = "rook-ceph-mon-endpoints"
 	// EndpointDataKey is the name of the key inside the mon configmap to get the endpoints
 	EndpointDataKey = "data"
+	// EndpointExternalMonsKey key in EndpointConfigMapName configmap containing IDs of external mons
+	EndpointExternalMonsKey = "externalMons"
 	// 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
@@ -303,7 +305,7 @@ func (c *Cluster) startMons(targetCount int) error {
 		}
 	}
 
-	logger.Debugf("mon endpoints used are: %s", flattenMonEndpoints(c.ClusterInfo.Monitors))
+	logger.Debugf("mon endpoints used are: %s", flattenMonEndpoints(c.ClusterInfo.AllMonitors()))
 
 	// reconcile mon PDB
 	if err := c.reconcileMonPDB(); err != nil {
@@ -484,7 +486,7 @@ func (c *Cluster) ensureMonsRunning(mons []*monConfig, i, targetCount int, requi
 	// Calculate how many mons we expected to exist after this method is completed.
 	// If we are adding a new mon, we expect one more than currently exist.
 	// If we haven't created all the desired mons already, we will be adding a new one with this iteration
-	expectedMonCount := len(c.ClusterInfo.Monitors)
+	expectedMonCount := len(c.ClusterInfo.InternalMonitors)
 	if expectedMonCount < targetCount {
 		expectedMonCount++
 	}
@@ -548,13 +550,12 @@ func (c *Cluster) initClusterInfo(cephVersion cephver.CephVersion, clusterName s
 }
 
 func (c *Cluster) initMonConfig(size int) (int, []*monConfig, error) {
-
 	// initialize the mon pod info for mons that have been previously created
 	mons := c.clusterInfoToMonConfig()
 
 	// initialize mon info if we don't have enough mons (at first startup)
-	existingCount := len(c.ClusterInfo.Monitors)
-	for i := len(c.ClusterInfo.Monitors); i < size; i++ {
+	existingCount := len(c.ClusterInfo.InternalMonitors)
+	for i := len(c.ClusterInfo.InternalMonitors); i < size; i++ {
 		c.maxMonID++
 		zone, err := c.findAvailableZone(mons)
 		if err != nil {
@@ -572,7 +573,7 @@ func (c *Cluster) clusterInfoToMonConfig() []*monConfig {
 
 func (c *Cluster) clusterInfoToMonConfigWithExclude(excludedMon string) []*monConfig {
 	mons := []*monConfig{}
-	for _, monitor := range c.ClusterInfo.Monitors {
+	for _, monitor := range c.ClusterInfo.InternalMonitors {
 		if monitor.Name == excludedMon {
 			// Skip a mon if it is being failed over
 			continue
@@ -851,7 +852,7 @@ func (c *Cluster) initMonIPs(mons []*monConfig) error {
 				}
 			}
 		}
-		c.ClusterInfo.Monitors[m.DaemonName] = cephclient.NewMonInfo(m.DaemonName, m.PublicIP, m.Port)
+		c.ClusterInfo.InternalMonitors[m.DaemonName] = cephclient.NewMonInfo(m.DaemonName, m.PublicIP, m.Port)
 	}
 
 	return nil
@@ -1160,7 +1161,7 @@ func (c *Cluster) saveMonConfig() error {
 		return errors.Wrap(err, "failed to write connection config for new mons")
 	}
 
-	monEndpoints := csi.MonEndpoints(c.ClusterInfo.Monitors, c.spec.RequireMsgr2())
+	monEndpoints := csi.MonEndpoints(c.ClusterInfo.AllMonitors(), c.spec.RequireMsgr2())
 	csiConfigEntry := &csi.CSIClusterConfigEntry{
 		Namespace: c.ClusterInfo.Namespace,
 		ClusterInfo: cephcsi.ClusterInfo{
@@ -1173,7 +1174,7 @@ func (c *Cluster) saveMonConfig() error {
 		return errors.Wrap(err, "failed to update csi cluster config")
 	}
 
-	if csi.EnableCSIOperator() && len(c.ClusterInfo.Monitors) > 0 {
+	if csi.EnableCSIOperator() && len(c.ClusterInfo.AllMonitors()) > 0 {
 		err := csi.CreateUpdateCephConnection(c.context.Client, c.ClusterInfo, c.spec)
 		if err != nil {
 			return errors.Wrap(err, "failed to create/update cephConnection")
@@ -1207,7 +1208,7 @@ func (c *Cluster) persistExpectedMonDaemons() error {
 	}
 
 	csiConfigValue, err := csi.FormatCsiClusterConfig(
-		c.Namespace, c.ClusterInfo.Monitors)
+		c.Namespace, c.ClusterInfo.AllMonitors())
 	if err != nil {
 		return errors.Wrap(err, "failed to format csi config")
 	}
@@ -1219,14 +1220,21 @@ func (c *Cluster) persistExpectedMonDaemons() error {
 
 	// preserve the mons detected out of quorum
 	var monsOutOfQuorum []string
-	for monName, mon := range c.ClusterInfo.Monitors {
+	for monName, mon := range c.ClusterInfo.InternalMonitors {
 		if mon.OutOfQuorum {
 			monsOutOfQuorum = append(monsOutOfQuorum, monName)
 		}
 	}
+	extMonIDs := make([]string, 0, len(c.ClusterInfo.ExternalMons))
+	if c.ClusterInfo.ExternalMons != nil {
+		for monID := range c.ClusterInfo.ExternalMons {
+			extMonIDs = append(extMonIDs, monID)
+		}
+	}
 
 	configMap.Data = map[string]string{
-		EndpointDataKey: flattenMonEndpoints(c.ClusterInfo.Monitors),
+		EndpointDataKey:         flattenMonEndpoints(c.ClusterInfo.AllMonitors()),
+		EndpointExternalMonsKey: strings.Join(extMonIDs, ","),
 		// persist the maxMonID that was previously stored in the configmap. We are likely saving info
 		// about scheduling of the mons, but we only want to update the maxMonID once a new mon has
 		// actually been started. If the operator is restarted or the reconcile is otherwise restarted,
diff --git a/pkg/operator/ceph/cluster/mon/util.go b/pkg/operator/ceph/cluster/mon/util.go
index f688a3794..39b20e582 100644
--- a/pkg/operator/ceph/cluster/mon/util.go
+++ b/pkg/operator/ceph/cluster/mon/util.go
@@ -17,6 +17,7 @@ limitations under the License.
 package mon
 
 import (
+	"slices"
 	"strings"
 
 	"github.com/rook/rook/pkg/daemon/ceph/client"
@@ -34,6 +35,17 @@ func monInQuorum(monitor client.MonMapEntry, quorum []int) bool {
 	return false
 }
 
+func getMonByID(monID string, monMap client.MonStatusResponse) (info client.MonMapEntry, inQuorum bool) {
+	for _, mon := range monMap.MonMap.Mons {
+		if mon.Name != monID {
+			continue
+		}
+		monRank := mon.Rank
+		return mon, slices.Contains(monMap.Quorum, monRank)
+	}
+	return client.MonMapEntry{}, false
+}
+
 // convert the mon name to the numeric mon ID
 func fullNameToIndex(name string) (int, error) {
 	// remove the "rook-ceph-mon" prefix
diff --git a/pkg/operator/ceph/controller/cluster_info.go b/pkg/operator/ceph/controller/cluster_info.go
index 2da3191ca..7a4d3851d 100644
--- a/pkg/operator/ceph/controller/cluster_info.go
+++ b/pkg/operator/ceph/controller/cluster_info.go
@@ -23,6 +23,7 @@ import (
 	"os"
 	"path"
 	"path/filepath"
+	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -56,6 +57,8 @@ const (
 	EndpointConfigMapName = "rook-ceph-mon-endpoints"
 	// EndpointDataKey is the name of the key inside the mon configmap to get the endpoints
 	EndpointDataKey = "data"
+	// EndpointExternalMonsKey key in EndpointConfigMapName configmap containing IDs of external mons
+	EndpointExternalMonsKey = "externalMons"
 	// OutOfQuorumKey is the name of the key for tracking mons detected out of quorum
 	OutOfQuorumKey = "outOfQuorum"
 	// MaxMonIDKey is the name of the max mon id used
@@ -148,7 +151,7 @@ func CreateOrLoadClusterInfo(clusterdContext *clusterd.Context, context context.
 	}
 
 	// get the existing monitor config
-	clusterInfo.Monitors, maxMonID, monMapping, err = loadMonConfig(clusterdContext.Clientset, namespace)
+	clusterInfo.ExternalMons, clusterInfo.InternalMonitors, maxMonID, monMapping, err = loadMonConfig(clusterdContext.Clientset, namespace)
 	if err != nil {
 		return nil, maxMonID, monMapping, errors.Wrap(err, "failed to get mon config")
 	}
@@ -276,33 +279,34 @@ func UpdateMonsOutOfQuorum(clientset kubernetes.Interface, namespace string, mon
 }
 
 // loadMonConfig returns the monitor endpoints and maxMonID
-func loadMonConfig(clientset kubernetes.Interface, namespace string) (map[string]*cephclient.MonInfo, int, *Mapping, error) {
+func loadMonConfig(clientset kubernetes.Interface, namespace string) (extMons map[string]*cephclient.MonInfo, internalMons map[string]*cephclient.MonInfo, maxMonID int, monMapping *Mapping, err error) {
 	ctx := context.TODO()
-	monEndpointMap := map[string]*cephclient.MonInfo{}
-	maxMonID := -1
-	monMapping := &Mapping{
+	extMons = map[string]*cephclient.MonInfo{}
+	internalMons = 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
+			return nil, nil, maxMonID, monMapping, err
 		}
 		// If the config map was not found, initialize the empty set of monitors
-		return monEndpointMap, maxMonID, monMapping, nil
+		return extMons, internalMons, maxMonID, monMapping, nil
 	}
 
 	// Parse the monitor List
 	if info, ok := cm.Data[EndpointDataKey]; ok {
-		monEndpointMap = ParseMonEndpoints(info)
+		internalMons = ParseMonEndpoints(info)
 	}
 
 	// Parse the mons that were detected out of quorum
 	if outOfQuorum, ok := cm.Data[OutOfQuorumKey]; ok && len(outOfQuorum) > 0 {
 		monNames := strings.Split(outOfQuorum, ",")
 		for _, monName := range monNames {
-			if monInfo, ok := monEndpointMap[monName]; ok {
+			if monInfo, ok := internalMons[monName]; ok {
 				monInfo.OutOfQuorum = true
 			} else {
 				logger.Warningf("did not find mon %q to set it out of quorum in the cluster info", monName)
@@ -322,7 +326,7 @@ func loadMonConfig(clientset kubernetes.Interface, namespace string) (map[string
 	}
 
 	// Make sure the max id is consistent with the current monitors
-	for _, m := range monEndpointMap {
+	for _, m := range internalMons {
 		id, _ := fullNameToIndex(m.Name)
 		if maxMonID < id {
 			maxMonID = id
@@ -336,9 +340,19 @@ func loadMonConfig(clientset kubernetes.Interface, namespace string) (map[string
 	if err != nil {
 		logger.Errorf("invalid JSON in mon mapping. %v", err)
 	}
+	// filter external mons:
+	if extMonIDsStr, ok := cm.Data[EndpointExternalMonsKey]; ok && extMonIDsStr != "" {
+		extMonIDs := strings.Split(extMonIDsStr, ",")
+		for monID, mon := range internalMons {
+			if slices.Contains(extMonIDs, monID) {
+				extMons[monID] = mon
+				delete(internalMons, monID)
+			}
+		}
+	}
 
-	logger.Debugf("loaded: maxMonID=%d, mons=%+v, assignment=%+v", maxMonID, monEndpointMap, monMapping)
-	return monEndpointMap, maxMonID, monMapping, nil
+	logger.Debugf("loaded: maxMonID=%d, extMons=%+v, mons=%+v, assignment=%+v", maxMonID, extMons, internalMons, monMapping)
+	return extMons, internalMons, maxMonID, monMapping, nil
 }
 
 // convert the mon name to the numeric mon ID
@@ -413,7 +427,7 @@ func PopulateExternalClusterInfo(cephClusterSpec *cephv1.ClusterSpec, context *c
 			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)
+		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.AllMonitors())
 		clusterInfo.OwnerInfo = ownerInfo
 
 		return clusterInfo, nil
diff --git a/pkg/operator/ceph/csi/ceph_connection.go b/pkg/operator/ceph/csi/ceph_connection.go
index 518b09d2f..61d89c57b 100644
--- a/pkg/operator/ceph/csi/ceph_connection.go
+++ b/pkg/operator/ceph/csi/ceph_connection.go
@@ -91,7 +91,7 @@ func generateCephConnSpec(c client.Client, clusterInfo *cephclient.ClusterInfo,
 		csiClusterConnSpec.RbdMirrorDaemonCount = cephRBDMirrorList.Items[0].Spec.Count
 	}
 
-	for _, mon := range clusterInfo.Monitors {
+	for _, mon := range clusterInfo.AllMonitors() {
 		csiClusterConnSpec.Monitors = append(csiClusterConnSpec.Monitors, mon.Endpoint)
 	}
 
diff --git a/pkg/operator/ceph/file/subvolumegroup/controller.go b/pkg/operator/ceph/file/subvolumegroup/controller.go
index e5605785a..2969f00b7 100644
--- a/pkg/operator/ceph/file/subvolumegroup/controller.go
+++ b/pkg/operator/ceph/file/subvolumegroup/controller.go
@@ -298,7 +298,7 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) updateClusterConfig(cephFilesyst
 	csiClusterConfigEntry := csi.CSIClusterConfigEntry{
 		Namespace: r.clusterInfo.Namespace,
 		ClusterInfo: cephcsi.ClusterInfo{
-			Monitors: csi.MonEndpoints(r.clusterInfo.Monitors, cephCluster.Spec.RequireMsgr2()),
+			Monitors: csi.MonEndpoints(r.clusterInfo.AllMonitors(), cephCluster.Spec.RequireMsgr2()),
 			CephFS: cephcsi.CephFS{
 				SubvolumeGroup:     getSubvolumeGroupName(cephFilesystemSubVolumeGroup),
 				KernelMountOptions: r.clusterInfo.CSIDriverSpec.CephFS.KernelMountOptions,
diff --git a/pkg/operator/ceph/pool/radosnamespace/controller.go b/pkg/operator/ceph/pool/radosnamespace/controller.go
index bbac0c9cc..14974bdf6 100644
--- a/pkg/operator/ceph/pool/radosnamespace/controller.go
+++ b/pkg/operator/ceph/pool/radosnamespace/controller.go
@@ -323,7 +323,7 @@ func (r *ReconcileCephBlockPoolRadosNamespace) updateClusterConfig(cephBlockPool
 	csiClusterConfigEntry := csi.CSIClusterConfigEntry{
 		Namespace: r.clusterInfo.Namespace,
 		ClusterInfo: cephcsi.ClusterInfo{
-			Monitors: csi.MonEndpoints(r.clusterInfo.Monitors, cephCluster.Spec.RequireMsgr2()),
+			Monitors: csi.MonEndpoints(r.clusterInfo.AllMonitors(), cephCluster.Spec.RequireMsgr2()),
 			RBD: cephcsi.RBD{
 				RadosNamespace: getRadosNamespaceName(cephBlockPoolRadosNamespace),
 			},


Test patch:
diff --git a/pkg/daemon/ceph/client/config_test.go b/pkg/daemon/ceph/client/config_test.go
index cfb2afa20..cf7810312 100644
--- a/pkg/daemon/ceph/client/config_test.go
+++ b/pkg/daemon/ceph/client/config_test.go
@@ -37,7 +37,7 @@ func TestCreateDefaultCephConfig(t *testing.T) {
 		FSID:          "id",
 		MonitorSecret: "monsecret",
 		Namespace:     "foo-cluster",
-		Monitors: map[string]*MonInfo{
+		InternalMonitors: map[string]*MonInfo{
 			"node0": {Name: "mon0", Endpoint: "10.0.0.1:6789"},
 			"node1": {Name: "mon1", Endpoint: "10.0.0.2:6789"},
 		},
@@ -89,7 +89,7 @@ func TestGenerateConfigFile(t *testing.T) {
 		FSID:          "myfsid",
 		MonitorSecret: "monsecret",
 		Namespace:     ns,
-		Monitors: map[string]*MonInfo{
+		InternalMonitors: map[string]*MonInfo{
 			"node0": {Name: "mon0", Endpoint: "10.0.0.1:6789"},
 		},
 		CephCred: CephCred{Username: "admin", Secret: "mysecret"},
@@ -112,9 +112,9 @@ func TestGenerateConfigFile(t *testing.T) {
 }
 
 func verifyConfig(t *testing.T, cephConfig *CephConfig, cluster *ClusterInfo, loggingLevel int) {
-	monMembers := make([]string, len(cluster.Monitors))
+	monMembers := make([]string, len(cluster.InternalMonitors))
 	i := 0
-	for _, expectedMon := range cluster.Monitors {
+	for _, expectedMon := range cluster.InternalMonitors {
 		contained := false
 		monMembers[i] = expectedMon.Name
 		for _, actualMon := range strings.Split(cephConfig.MonMembers, " ") {
diff --git a/pkg/daemon/ceph/client/test/info.go b/pkg/daemon/ceph/client/test/info.go
index 4aaa57ac3..c5081e243 100644
--- a/pkg/daemon/ceph/client/test/info.go
+++ b/pkg/daemon/ceph/client/test/info.go
@@ -51,14 +51,15 @@ func CreateTestClusterInfo(monCount int) *client.ClusterInfo {
 			Username: client.AdminUsername,
 			Secret:   "adminkey",
 		},
-		Monitors:  map[string]*client.MonInfo{},
-		OwnerInfo: ownerInfo,
-		Context:   context.TODO(),
+		InternalMonitors: map[string]*client.MonInfo{},
+		ExternalMons:     map[string]*client.MonInfo{},
+		OwnerInfo:        ownerInfo,
+		Context:          context.TODO(),
 	}
 	mons := []string{"a", "b", "c", "d", "e"}
 	for i := 0; i < monCount; i++ {
 		id := mons[i]
-		c.Monitors[id] = &client.MonInfo{
+		c.InternalMonitors[id] = &client.MonInfo{
 			Name:     id,
 			Endpoint: fmt.Sprintf("1.2.3.%d:3300", (i + 1)),
 		}
diff --git a/pkg/operator/ceph/cluster/mon/health_test.go b/pkg/operator/ceph/cluster/mon/health_test.go
index 90a778dbc..5a9452a75 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"
+	"github.com/rook/rook/pkg/operator/ceph/controller"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	testopk8s "github.com/rook/rook/pkg/operator/k8sutil/test"
 	"github.com/rook/rook/pkg/operator/test"
@@ -74,7 +75,7 @@ func TestCheckHealth(t *testing.T) {
 	assert.NotNil(t, err)
 
 	c.spec.Mon.Count = 3
-	logger.Infof("initial mons: %v", c.ClusterInfo.Monitors)
+	logger.Infof("initial mons: %v", c.ClusterInfo.InternalMonitors)
 	c.waitForStart = false
 
 	c.mapping.Schedule["f"] = &opcontroller.MonScheduleInfo{
@@ -92,7 +93,7 @@ func TestCheckHealth(t *testing.T) {
 	c.ClusterInfo.Context = ctx
 	err = c.checkHealth(ctx)
 	assert.Nil(t, err)
-	logger.Infof("mons after checkHealth: %v", c.ClusterInfo.Monitors)
+	logger.Infof("mons after checkHealth: %v", c.ClusterInfo.InternalMonitors)
 	assert.ElementsMatch(t, []string{"rook-ceph-mon-a", "rook-ceph-mon-f"}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
 	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
 
@@ -105,8 +106,8 @@ func TestCheckHealth(t *testing.T) {
 		"g",
 	}
 	for _, monName := range newMons {
-		_, ok := c.ClusterInfo.Monitors[monName]
-		assert.True(t, ok, fmt.Sprintf("mon %s not found in monitor list. %v", monName, c.ClusterInfo.Monitors))
+		_, ok := c.ClusterInfo.InternalMonitors[monName]
+		assert.True(t, ok, fmt.Sprintf("mon %s not found in monitor list. %v", monName, c.ClusterInfo.InternalMonitors))
 	}
 
 	deployments, err := clientset.AppsV1().Deployments(c.Namespace).List(ctx, metav1.ListOptions{})
@@ -153,7 +154,7 @@ func TestCheckHealth(t *testing.T) {
 func TestRemoveExtraMon(t *testing.T) {
 	endpoint := "1.2.3.4:6789"
 	c := &Cluster{mapping: &opcontroller.Mapping{}}
-	c.ClusterInfo = &cephclient.ClusterInfo{Monitors: map[string]*cephclient.MonInfo{
+	c.ClusterInfo = &cephclient.ClusterInfo{InternalMonitors: map[string]*cephclient.MonInfo{
 		"a": {Name: "a", Endpoint: endpoint},
 		"b": {Name: "b", Endpoint: endpoint},
 		"c": {Name: "c", Endpoint: endpoint},
@@ -182,7 +183,7 @@ func TestRemoveExtraMon(t *testing.T) {
 		{Name: "y"},
 		{Name: "z"},
 	}}
-	c.ClusterInfo.Monitors["e"] = &cephclient.MonInfo{Name: "e", Endpoint: endpoint}
+	c.ClusterInfo.InternalMonitors["e"] = &cephclient.MonInfo{Name: "e", Endpoint: endpoint}
 	c.mapping.Schedule["a"].Zone = "x"
 	c.mapping.Schedule["b"].Zone = "y"
 	c.mapping.Schedule["c"].Zone = "y"
@@ -218,7 +219,7 @@ func TestTrackMonsOutOfQuorum(t *testing.T) {
 		context:   &clusterd.Context{Clientset: clientset, ConfigDir: tempDir},
 		ownerInfo: ownerInfo,
 		Namespace: "ns"}
-	c.ClusterInfo = &cephclient.ClusterInfo{Monitors: map[string]*cephclient.MonInfo{
+	c.ClusterInfo = &cephclient.ClusterInfo{InternalMonitors: map[string]*cephclient.MonInfo{
 		"a": {Name: "a", Endpoint: endpoint},
 		"b": {Name: "b", Endpoint: endpoint},
 		"c": {Name: "c", Endpoint: endpoint},
@@ -323,7 +324,7 @@ func TestScaleMonDeployment(t *testing.T) {
 
 	name := "a"
 	c.spec.Mon.Count = 3
-	logger.Infof("initial mons: %v", c.ClusterInfo.Monitors[name])
+	logger.Infof("initial mons: %v", c.ClusterInfo.InternalMonitors[name])
 	monConfig := &monConfig{ResourceName: resourceName(name), DaemonName: name, DataPathMap: &config.DataPathMap{}}
 	d, err := c.makeDeployment(monConfig, false)
 	require.NoError(t, err)
@@ -437,7 +438,7 @@ func TestAddRemoveMons(t *testing.T) {
 	// checking the health will increase the mons as desired all in one go
 	err := c.checkHealth(ctx)
 	assert.Nil(t, err)
-	assert.Equal(t, 5, len(c.ClusterInfo.Monitors), fmt.Sprintf("mons: %v", c.ClusterInfo.Monitors))
+	assert.Equal(t, 5, len(c.ClusterInfo.InternalMonitors), fmt.Sprintf("mons: %v", c.ClusterInfo.InternalMonitors))
 	assert.ElementsMatch(t, []string{
 		// b is created first, no updates
 		"rook-ceph-mon-b",                    // b updated when c created
@@ -448,39 +449,39 @@ func TestAddRemoveMons(t *testing.T) {
 	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
 
 	// reducing the mon count to 3 will reduce the mon count once each time we call checkHealth
-	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.ClusterInfo.Monitors)
+	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.ClusterInfo.InternalMonitors)
 	c.spec.Mon.Count = 3
 	err = c.checkHealth(ctx)
 	assert.Nil(t, err)
-	assert.Equal(t, 4, len(c.ClusterInfo.Monitors))
+	assert.Equal(t, 4, len(c.ClusterInfo.InternalMonitors))
 	// No updates in unit tests w/ workaround
 	assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
 	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
 
 	// after the second call we will be down to the expected count of 3
-	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.ClusterInfo.Monitors)
+	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.ClusterInfo.InternalMonitors)
 	err = c.checkHealth(ctx)
 	assert.Nil(t, err)
-	assert.Equal(t, 3, len(c.ClusterInfo.Monitors))
+	assert.Equal(t, 3, len(c.ClusterInfo.InternalMonitors))
 	// No updates in unit tests w/ workaround
 	assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
 	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
 
 	// now attempt to reduce the mons down to quorum size 1
-	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.ClusterInfo.Monitors)
+	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.ClusterInfo.InternalMonitors)
 	c.spec.Mon.Count = 1
 	err = c.checkHealth(ctx)
 	assert.Nil(t, err)
-	assert.Equal(t, 2, len(c.ClusterInfo.Monitors))
+	assert.Equal(t, 2, len(c.ClusterInfo.InternalMonitors))
 	// No updates in unit tests w/ workaround
 	assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
 	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
 
 	// cannot reduce from quorum size of 2 to 1
-	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.ClusterInfo.Monitors)
+	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.ClusterInfo.InternalMonitors)
 	err = c.checkHealth(ctx)
 	assert.Nil(t, err)
-	assert.Equal(t, 2, len(c.ClusterInfo.Monitors))
+	assert.Equal(t, 2, len(c.ClusterInfo.InternalMonitors))
 	// No updates in unit tests w/ workaround
 	assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
 	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
@@ -511,7 +512,7 @@ func TestAddOrRemoveExternalMonitor(t *testing.T) {
 	changed, err = c.addOrRemoveExternalMonitor(fakeResp)
 	assert.NoError(t, err)
 	assert.False(t, changed)
-	assert.Equal(t, 1, len(c.ClusterInfo.Monitors))
+	assert.Equal(t, 1, len(c.ClusterInfo.InternalMonitors))
 
 	//
 	// TEST 2
@@ -524,7 +525,7 @@ func TestAddOrRemoveExternalMonitor(t *testing.T) {
 	assert.NoError(t, err)
 	assert.True(t, changed)
 	// ClusterInfo should shrink to 1
-	assert.Equal(t, 1, len(c.ClusterInfo.Monitors))
+	assert.Equal(t, 1, len(c.ClusterInfo.InternalMonitors))
 
 	//
 	// TEST 3
@@ -545,7 +546,7 @@ func TestAddOrRemoveExternalMonitor(t *testing.T) {
 	assert.NoError(t, err)
 	assert.True(t, changed)
 	// ClusterInfo should now have 2 monitors
-	assert.Equal(t, 2, len(c.ClusterInfo.Monitors))
+	assert.Equal(t, 2, len(c.ClusterInfo.InternalMonitors))
 }
 
 func TestNewHealthChecker(t *testing.T) {
@@ -610,3 +611,522 @@ func TestUpdateMonInterval(t *testing.T) {
 		assert.Equal(t, time.Minute, h.interval)
 	})
 }
+
+func Test_removeMonsFromQuorumStatusResponse(t *testing.T) {
+	type args struct {
+		quorumStatus cephclient.MonStatusResponse
+		idsToRemove  []string
+	}
+	tests := []struct {
+		name string
+		args args
+		want cephclient.MonStatusResponse
+	}{
+		{
+			name: "remove one mon",
+			args: args{
+				quorumStatus: cephclient.MonStatusResponse{
+					Quorum: []int{0, 1, 2},
+					MonMap: struct {
+						Mons []cephclient.MonMapEntry `json:"mons"`
+					}{
+						Mons: []cephclient.MonMapEntry{
+							{
+								Name: "a",
+								Rank: 0,
+							},
+							{
+								Name: "b",
+								Rank: 1,
+							},
+							{
+								Name: "c",
+								Rank: 2,
+							},
+						},
+					},
+				},
+				idsToRemove: []string{"b"},
+			},
+			want: cephclient.MonStatusResponse{
+				Quorum: []int{0, 2},
+				MonMap: struct {
+					Mons []cephclient.MonMapEntry `json:"mons"`
+				}{
+					Mons: []cephclient.MonMapEntry{
+						{
+							Name: "a",
+							Rank: 0,
+						},
+						{
+							Name: "c",
+							Rank: 2,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "not remove if not present",
+			args: args{
+				quorumStatus: cephclient.MonStatusResponse{
+					Quorum: []int{0, 1, 2},
+					MonMap: struct {
+						Mons []cephclient.MonMapEntry `json:"mons"`
+					}{
+						Mons: []cephclient.MonMapEntry{
+							{
+								Name: "a",
+								Rank: 0,
+							},
+							{
+								Name: "b",
+								Rank: 1,
+							},
+							{
+								Name: "c",
+								Rank: 2,
+							},
+						},
+					},
+				},
+				idsToRemove: []string{"e"},
+			},
+			want: cephclient.MonStatusResponse{
+				Quorum: []int{0, 1, 2},
+				MonMap: struct {
+					Mons []cephclient.MonMapEntry `json:"mons"`
+				}{
+					Mons: []cephclient.MonMapEntry{
+						{
+							Name: "a",
+							Rank: 0,
+						},
+						{
+							Name: "b",
+							Rank: 1,
+						},
+						{
+							Name: "c",
+							Rank: 2,
+						},
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := removeMonsFromQuorumStatusResponse(tt.args.quorumStatus, tt.args.idsToRemove); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("removeMonsFromQuorumStatusResponse() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestExternalMons_notInSpec_InQuorum(t *testing.T) {
+	// 1. setup test
+	ctx := context.TODO()
+	var deploymentsUpdated *[]*apps.Deployment
+	updateDeploymentAndWait, deploymentsUpdated = testopk8s.UpdateDeploymentAndWaitStub()
+
+	monQuorumResponse := clienttest.MonInQuorumResponse()
+	executor := &exectest.MockExecutor{
+		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			logger.Infof("executing command: %s %+v", command, args)
+			if args[0] == "auth" && args[1] == "get-or-create-key" {
+				return "{\"key\":\"mysecurekey\"}", nil
+			}
+			return monQuorumResponse, nil
+		},
+	}
+	clientset := test.New(t, 1)
+	configDir := t.TempDir()
+	context := &clusterd.Context{
+		Clientset: clientset,
+		ConfigDir: configDir,
+		Executor:  executor,
+	}
+	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
+	c := New(ctx, context, "ns", cephv1.ClusterSpec{}, ownerInfo)
+	setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 5, AllowMultiplePerNode: true}, "myversion")
+	c.maxMonID = 0 // "a" is max mon id
+	c.waitForStart = false
+
+	// checking the health will increase the mons as desired all in one go
+	err := c.checkHealth(ctx)
+	assert.Nil(t, err)
+	assert.Equal(t, 5, len(c.ClusterInfo.InternalMonitors), fmt.Sprintf("mons: %v", c.ClusterInfo.InternalMonitors))
+	assert.ElementsMatch(t, []string{
+		// b is created first, no updates
+		"rook-ceph-mon-b",                    // b updated when c created
+		"rook-ceph-mon-b", "rook-ceph-mon-c", // b and c updated when d created
+		"rook-ceph-mon-b", "rook-ceph-mon-c", "rook-ceph-mon-d", // etc.
+		"rook-ceph-mon-b", "rook-ceph-mon-c", "rook-ceph-mon-d", "rook-ceph-mon-e"},
+		testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
+	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
+
+	inital5Mons := make(map[string]*cephclient.MonInfo)
+	for k, v := range c.ClusterInfo.InternalMonitors {
+		inital5Mons[k] = v
+	}
+
+	// 2. add external mon to quorum but not in spec:
+
+	mons := make(map[string]*cephclient.MonInfo)
+	for k, v := range inital5Mons {
+		mons[k] = v
+	}
+	// add unknown mon to quorum:
+	mons["ext-mon-id"] = &cephclient.MonInfo{Name: "ext-mon-id", Endpoint: "0.0.0.0:6789"}
+	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(mons)
+
+	// internal mons and deployments has not changed
+	// and unknown mon was removed from quorum
+	err = c.checkHealth(ctx)
+	assert.Nil(t, err)
+	assert.Equal(t, 5, len(c.ClusterInfo.InternalMonitors))
+	assert.Empty(t, c.ClusterInfo.ExternalMons)
+
+	// No updates in unit tests w/ workaround
+	assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
+	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
+
+	// check that unknown mon is not in endpoint configmap
+	cm, err := c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(ctx, EndpointConfigMapName, metav1.GetOptions{})
+	assert.Nil(t, err)
+	assert.Empty(t, cm.Data[EndpointExternalMonsKey])
+	monsFromCM := controller.ParseMonEndpoints(cm.Data[EndpointDataKey])
+	assert.Len(t, monsFromCM, 5)
+	for id, mon := range monsFromCM {
+		assert.Equal(t, inital5Mons[id].Name, mon.Name)
+		assert.Equal(t, inital5Mons[id].Endpoint, mon.Endpoint)
+		assert.Equal(t, inital5Mons[id].OutOfQuorum, mon.OutOfQuorum)
+	}
+
+	// 3. downscale mons to 4:
+	c.spec.Mon.Count = 4
+	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.ClusterInfo.InternalMonitors)
+
+	err = c.checkHealth(ctx)
+	assert.Nil(t, err)
+	// todo fix
+	assert.Equal(t, 4, len(c.ClusterInfo.InternalMonitors))
+	assert.Empty(t, c.ClusterInfo.ExternalMons)
+	// No updates in unit tests w/ workaround
+	assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
+	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
+
+	// check that unknown mon is not in endpoint configmap
+	cm, err = c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(ctx, EndpointConfigMapName, metav1.GetOptions{})
+	assert.Nil(t, err)
+	assert.Empty(t, cm.Data[EndpointExternalMonsKey])
+	monsFromCM = controller.ParseMonEndpoints(cm.Data[EndpointDataKey])
+	assert.Len(t, monsFromCM, 4)
+	for id, mon := range monsFromCM {
+		assert.Equal(t, inital5Mons[id].Name, mon.Name)
+		assert.Equal(t, inital5Mons[id].Endpoint, mon.Endpoint)
+		assert.Equal(t, inital5Mons[id].OutOfQuorum, mon.OutOfQuorum)
+	}
+
+	// 4. upscale back to 5:
+	c.spec.Mon.Count = 5
+	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.ClusterInfo.InternalMonitors)
+
+	err = c.checkHealth(ctx)
+	assert.Nil(t, err)
+	assert.Equal(t, 5, len(c.ClusterInfo.InternalMonitors))
+	assert.Empty(t, c.ClusterInfo.ExternalMons)
+	// No updates in unit tests w/ workaround
+	assert.Len(t, testopk8s.DeploymentNamesUpdated(deploymentsUpdated), 4)
+	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
+
+	// check that unknown mon is not in endpoint configmap
+	cm, err = c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(ctx, EndpointConfigMapName, metav1.GetOptions{})
+	assert.Nil(t, err)
+	assert.Empty(t, cm.Data[EndpointExternalMonsKey])
+	monsFromCM = controller.ParseMonEndpoints(cm.Data[EndpointDataKey])
+	assert.Len(t, monsFromCM, 5)
+	for id, mon := range monsFromCM {
+		assert.Equal(t, c.ClusterInfo.InternalMonitors[id].Name, mon.Name)
+		assert.Equal(t, c.ClusterInfo.InternalMonitors[id].Endpoint, mon.Endpoint)
+		assert.Equal(t, c.ClusterInfo.InternalMonitors[id].OutOfQuorum, mon.OutOfQuorum)
+	}
+
+}
+
+func TestExternalMons_inSpec_notInQuorum(t *testing.T) {
+	// 1. setup test
+	ctx := context.TODO()
+	var deploymentsUpdated *[]*apps.Deployment
+	updateDeploymentAndWait, deploymentsUpdated = testopk8s.UpdateDeploymentAndWaitStub()
+
+	monQuorumResponse := clienttest.MonInQuorumResponse()
+	executor := &exectest.MockExecutor{
+		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			logger.Infof("executing command: %s %+v", command, args)
+			if args[0] == "auth" && args[1] == "get-or-create-key" {
+				return "{\"key\":\"mysecurekey\"}", nil
+			}
+			return monQuorumResponse, nil
+		},
+	}
+	clientset := test.New(t, 1)
+	configDir := t.TempDir()
+	context := &clusterd.Context{
+		Clientset: clientset,
+		ConfigDir: configDir,
+		Executor:  executor,
+	}
+	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
+	c := New(ctx, context, "ns", cephv1.ClusterSpec{}, ownerInfo)
+	setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 5, AllowMultiplePerNode: true}, "myversion")
+	c.maxMonID = 0 // "a" is max mon id
+	c.waitForStart = false
+
+	// checking the health will increase the mons as desired all in one go
+	err := c.checkHealth(ctx)
+	assert.Nil(t, err)
+	assert.Equal(t, 5, len(c.ClusterInfo.InternalMonitors), fmt.Sprintf("mons: %v", c.ClusterInfo.InternalMonitors))
+	assert.ElementsMatch(t, []string{
+		// b is created first, no updates
+		"rook-ceph-mon-b",                    // b updated when c created
+		"rook-ceph-mon-b", "rook-ceph-mon-c", // b and c updated when d created
+		"rook-ceph-mon-b", "rook-ceph-mon-c", "rook-ceph-mon-d", // etc.
+		"rook-ceph-mon-b", "rook-ceph-mon-c", "rook-ceph-mon-d", "rook-ceph-mon-e"},
+		testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
+	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
+
+	inital5Mons := make(map[string]*cephclient.MonInfo)
+	for k, v := range c.ClusterInfo.InternalMonitors {
+		inital5Mons[k] = v
+	}
+
+	// 2. add external mon id to spec but not to quorum
+	c.spec.Mon.ExternalMonIDs = []string{"ext-mon-id"}
+
+	// don't add ext mon to quorum:
+	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(inital5Mons)
+
+	err = c.checkHealth(ctx)
+	assert.Nil(t, err)
+	assert.Equal(t, 5, len(c.ClusterInfo.InternalMonitors))
+	assert.Empty(t, c.ClusterInfo.ExternalMons)
+
+	// No updates in unit tests w/ workaround
+	assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
+	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
+
+	// check that unknown mon is not in endpoint configmap
+	cm, err := c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(ctx, EndpointConfigMapName, metav1.GetOptions{})
+	assert.Nil(t, err)
+	assert.Empty(t, cm.Data[EndpointExternalMonsKey])
+	monsFromCM := controller.ParseMonEndpoints(cm.Data[EndpointDataKey])
+	assert.Len(t, monsFromCM, 5)
+	for id, mon := range monsFromCM {
+		assert.Equal(t, inital5Mons[id].Name, mon.Name)
+		assert.Equal(t, inital5Mons[id].Endpoint, mon.Endpoint)
+		assert.Equal(t, inital5Mons[id].OutOfQuorum, mon.OutOfQuorum)
+	}
+
+	// 3. downscale mons to 4:
+	c.spec.Mon.Count = 4
+
+	err = c.checkHealth(ctx)
+	assert.Nil(t, err)
+	assert.Equal(t, 4, len(c.ClusterInfo.InternalMonitors))
+	assert.Empty(t, c.ClusterInfo.ExternalMons)
+	// No updates in unit tests w/ workaround
+	assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
+	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
+
+	// check that unknown mon is not in endpoint configmap
+	cm, err = c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(ctx, EndpointConfigMapName, metav1.GetOptions{})
+	assert.Nil(t, err)
+	assert.Empty(t, cm.Data[EndpointExternalMonsKey])
+	monsFromCM = controller.ParseMonEndpoints(cm.Data[EndpointDataKey])
+	assert.Len(t, monsFromCM, 4)
+	for id, mon := range monsFromCM {
+		assert.Equal(t, inital5Mons[id].Name, mon.Name)
+		assert.Equal(t, inital5Mons[id].Endpoint, mon.Endpoint)
+		assert.Equal(t, inital5Mons[id].OutOfQuorum, mon.OutOfQuorum)
+	}
+
+	// 4. upscale back to 5:
+	c.spec.Mon.Count = 5
+	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.ClusterInfo.InternalMonitors)
+
+	err = c.checkHealth(ctx)
+	assert.Nil(t, err)
+	assert.Equal(t, 5, len(c.ClusterInfo.InternalMonitors))
+	assert.Empty(t, c.ClusterInfo.ExternalMons)
+	// No updates in unit tests w/ workaround
+	assert.Len(t, testopk8s.DeploymentNamesUpdated(deploymentsUpdated), 4)
+	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
+
+	// check that unknown mon is not in endpoint configmap
+	cm, err = c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(ctx, EndpointConfigMapName, metav1.GetOptions{})
+	assert.Nil(t, err)
+	assert.Empty(t, cm.Data[EndpointExternalMonsKey])
+	monsFromCM = controller.ParseMonEndpoints(cm.Data[EndpointDataKey])
+	assert.Len(t, monsFromCM, 5)
+	for id, mon := range monsFromCM {
+		assert.Equal(t, c.ClusterInfo.InternalMonitors[id].Name, mon.Name)
+		assert.Equal(t, c.ClusterInfo.InternalMonitors[id].Endpoint, mon.Endpoint)
+		assert.Equal(t, c.ClusterInfo.InternalMonitors[id].OutOfQuorum, mon.OutOfQuorum)
+	}
+
+}
+
+func TestExternalMons_inSpec_inQuorum(t *testing.T) {
+	// 1. setup test
+	ctx := context.TODO()
+	var deploymentsUpdated *[]*apps.Deployment
+	updateDeploymentAndWait, deploymentsUpdated = testopk8s.UpdateDeploymentAndWaitStub()
+
+	monQuorumResponse := clienttest.MonInQuorumResponse()
+	executor := &exectest.MockExecutor{
+		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			logger.Infof("executing command: %s %+v", command, args)
+			if args[0] == "auth" && args[1] == "get-or-create-key" {
+				return "{\"key\":\"mysecurekey\"}", nil
+			}
+			return monQuorumResponse, nil
+		},
+	}
+	clientset := test.New(t, 1)
+	configDir := t.TempDir()
+	context := &clusterd.Context{
+		Clientset: clientset,
+		ConfigDir: configDir,
+		Executor:  executor,
+	}
+	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
+	c := New(ctx, context, "ns", cephv1.ClusterSpec{}, ownerInfo)
+	setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 5, AllowMultiplePerNode: true}, "myversion")
+	c.maxMonID = 0 // "a" is max mon id
+	c.waitForStart = false
+
+	// checking the health will increase the mons as desired all in one go
+	err := c.checkHealth(ctx)
+	assert.Nil(t, err)
+	assert.Equal(t, 5, len(c.ClusterInfo.InternalMonitors), fmt.Sprintf("mons: %v", c.ClusterInfo.InternalMonitors))
+	assert.ElementsMatch(t, []string{
+		// b is created first, no updates
+		"rook-ceph-mon-b",                    // b updated when c created
+		"rook-ceph-mon-b", "rook-ceph-mon-c", // b and c updated when d created
+		"rook-ceph-mon-b", "rook-ceph-mon-c", "rook-ceph-mon-d", // etc.
+		"rook-ceph-mon-b", "rook-ceph-mon-c", "rook-ceph-mon-d", "rook-ceph-mon-e"},
+		testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
+	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
+
+	inital5Mons := make(map[string]*cephclient.MonInfo)
+	for k, v := range c.ClusterInfo.InternalMonitors {
+		inital5Mons[k] = v
+	}
+
+	// 2. add external mon id to spec
+	c.spec.Mon.ExternalMonIDs = []string{"ext-mon-id"}
+
+	// add ext mon to quorum:
+	mons := make(map[string]*cephclient.MonInfo)
+	for k, v := range inital5Mons {
+		mons[k] = v
+	}
+	mons["ext-mon-id"] = &cephclient.MonInfo{Name: "ext-mon-id", Endpoint: "0.0.0.0:6789"}
+	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(mons)
+
+	// internal mons and deployments has not changed
+	err = c.checkHealth(ctx)
+	assert.Nil(t, err)
+	assert.Equal(t, 5, len(c.ClusterInfo.InternalMonitors))
+	// external mon is in quorum
+	assert.Len(t, c.ClusterInfo.ExternalMons, 1)
+	assert.Equal(t, "ext-mon-id", c.ClusterInfo.ExternalMons["ext-mon-id"].Name)
+
+	// No updates in unit tests w/ workaround
+	assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
+	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
+
+	// check that ext mon is in endpoint configmap
+	cm, err := c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(ctx, EndpointConfigMapName, metav1.GetOptions{})
+	assert.Nil(t, err)
+	assert.Equal(t, "ext-mon-id", cm.Data[EndpointExternalMonsKey])
+	monsFromCM := controller.ParseMonEndpoints(cm.Data[EndpointDataKey])
+	assert.Len(t, monsFromCM, 6)
+	for id, mon := range monsFromCM {
+		if id == "ext-mon-id" {
+			assert.Equal(t, c.ClusterInfo.ExternalMons[id].Name, mon.Name)
+			assert.Equal(t, c.ClusterInfo.ExternalMons[id].Endpoint, mon.Endpoint)
+			assert.Equal(t, c.ClusterInfo.ExternalMons[id].OutOfQuorum, mon.OutOfQuorum)
+		} else {
+			assert.Equal(t, inital5Mons[id].Name, mon.Name)
+			assert.Equal(t, inital5Mons[id].Endpoint, mon.Endpoint)
+			assert.Equal(t, inital5Mons[id].OutOfQuorum, mon.OutOfQuorum)
+		}
+	}
+
+	// 3. downscale mons to 4:
+	c.spec.Mon.Count = 4
+
+	err = c.checkHealth(ctx)
+	assert.Nil(t, err)
+	assert.Equal(t, 4, len(c.ClusterInfo.InternalMonitors))
+	assert.Len(t, c.ClusterInfo.ExternalMons, 1)
+	// No updates in unit tests w/ workaround
+	assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
+	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
+
+	// check that unknown mon is not in endpoint configmap
+	cm, err = c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(ctx, EndpointConfigMapName, metav1.GetOptions{})
+	assert.Nil(t, err)
+	assert.Equal(t, "ext-mon-id", cm.Data[EndpointExternalMonsKey])
+	monsFromCM = controller.ParseMonEndpoints(cm.Data[EndpointDataKey])
+	assert.Len(t, monsFromCM, 5)
+	for id, mon := range monsFromCM {
+		if id == "ext-mon-id" {
+			assert.Equal(t, c.ClusterInfo.ExternalMons[id].Name, mon.Name)
+			assert.Equal(t, c.ClusterInfo.ExternalMons[id].Endpoint, mon.Endpoint)
+			assert.Equal(t, c.ClusterInfo.ExternalMons[id].OutOfQuorum, mon.OutOfQuorum)
+		} else {
+			assert.Equal(t, inital5Mons[id].Name, mon.Name)
+			assert.Equal(t, inital5Mons[id].Endpoint, mon.Endpoint)
+			assert.Equal(t, inital5Mons[id].OutOfQuorum, mon.OutOfQuorum)
+		}
+	}
+
+	// 4. upscale back to 5:
+	c.spec.Mon.Count = 5
+	mons = make(map[string]*cephclient.MonInfo)
+	for k, v := range c.ClusterInfo.InternalMonitors {
+		mons[k] = v
+	}
+	mons["ext-mon-id"] = &cephclient.MonInfo{Name: "ext-mon-id", Endpoint: "0.0.0.0:6789"}
+	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(mons)
+
+	err = c.checkHealth(ctx)
+	assert.Nil(t, err)
+	assert.Equal(t, 5, len(c.ClusterInfo.InternalMonitors))
+	assert.Len(t, c.ClusterInfo.ExternalMons, 1)
+	// No updates in unit tests w/ workaround
+	assert.Len(t, testopk8s.DeploymentNamesUpdated(deploymentsUpdated), 4)
+	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
+
+	// check that unknown mon is not in endpoint configmap
+	cm, err = c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Get(ctx, EndpointConfigMapName, metav1.GetOptions{})
+	assert.Nil(t, err)
+	assert.Equal(t, "ext-mon-id", cm.Data[EndpointExternalMonsKey])
+	monsFromCM = controller.ParseMonEndpoints(cm.Data[EndpointDataKey])
+	assert.Len(t, monsFromCM, 6)
+	for id, mon := range monsFromCM {
+		if id == "ext-mon-id" {
+			assert.Equal(t, c.ClusterInfo.ExternalMons[id].Name, mon.Name)
+			assert.Equal(t, c.ClusterInfo.ExternalMons[id].Endpoint, mon.Endpoint)
+			assert.Equal(t, c.ClusterInfo.ExternalMons[id].OutOfQuorum, mon.OutOfQuorum)
+		} else {
+			assert.Equal(t, c.ClusterInfo.InternalMonitors[id].Name, mon.Name)
+			assert.Equal(t, c.ClusterInfo.InternalMonitors[id].Endpoint, mon.Endpoint)
+			assert.Equal(t, c.ClusterInfo.InternalMonitors[id].OutOfQuorum, mon.OutOfQuorum)
+		}
+	}
+}
diff --git a/pkg/operator/ceph/cluster/mon/mon_test.go b/pkg/operator/ceph/cluster/mon/mon_test.go
index b4c5e0f5b..a481ec4d7 100644
--- a/pkg/operator/ceph/cluster/mon/mon_test.go
+++ b/pkg/operator/ceph/cluster/mon/mon_test.go
@@ -288,8 +288,8 @@ func TestPersistMons(t *testing.T) {
 	assert.Equal(t, map[string]string{"key": "value"}, cm.Annotations)
 
 	// Persist mon b, and remove mon a for simply testing the configmap is updated
-	c.ClusterInfo.Monitors["b"] = &cephclient.MonInfo{Name: "b", Endpoint: "4.5.6.7:3300"}
-	delete(c.ClusterInfo.Monitors, "a")
+	c.ClusterInfo.InternalMonitors["b"] = &cephclient.MonInfo{Name: "b", Endpoint: "4.5.6.7:3300"}
+	delete(c.ClusterInfo.InternalMonitors, "a")
 	err = c.persistExpectedMonDaemons()
 	assert.NoError(t, err)
 
@@ -317,7 +317,7 @@ func TestSaveMonEndpoints(t *testing.T) {
 	assert.Equal(t, "-1", cm.Data[opcontroller.MaxMonIDKey])
 
 	// update the config map
-	c.ClusterInfo.Monitors["a"].Endpoint = "2.3.4.5:6789"
+	c.ClusterInfo.InternalMonitors["a"].Endpoint = "2.3.4.5:6789"
 	c.maxMonID = 2
 	c.mapping.Schedule["a"] = &opcontroller.MonScheduleInfo{
 		Name:     "node0",
diff --git a/pkg/operator/ceph/config/store_test.go b/pkg/operator/ceph/config/store_test.go
index 902a72e4c..e9a2962b7 100644
--- a/pkg/operator/ceph/config/store_test.go
+++ b/pkg/operator/ceph/config/store_test.go
@@ -45,25 +45,25 @@ func TestStore(t *testing.T) {
 		sec, e := clientset.CoreV1().Secrets(ns).Get(ctxt, StoreName, metav1.GetOptions{})
 		assert.NoError(t, e)
 		mh := strings.Split(sec.StringData["mon_host"], ",") // list of mon ip:port pairs in cluster
-		expectedEndpoints := len(ci.Monitors)
+		expectedEndpoints := len(ci.InternalMonitors)
 		if mon1EndpointsEnabled {
 			expectedEndpoints *= 2
 		}
-		assert.Equal(t, expectedEndpoints, len(mh), ci.Monitors["a"].Endpoint) // we need to pass x2 since we split on "," above and that returns msgr1 and msgr2 addresses
-		mim := strings.Split(sec.StringData["mon_initial_members"], ",")       // list of mon ids in cluster
-		assert.Equal(t, len(ci.Monitors), len(mim))
+		assert.Equal(t, expectedEndpoints, len(mh), ci.InternalMonitors["a"].Endpoint) // we need to pass x2 since we split on "," above and that returns msgr1 and msgr2 addresses
+		mim := strings.Split(sec.StringData["mon_initial_members"], ",")               // list of mon ids in cluster
+		assert.Equal(t, len(ci.InternalMonitors), len(mim))
 		// make sure every mon has its id/ip:port in mon_initial_members/mon_host
 		for _, id := range mim {
 			// cannot use "assert.Contains(t, mh, ci.Monitors[id].Endpoint)"
 			// it looks like the value is not found but if present, it might be confused by the brackets
 			contains := false
 			for _, c := range mh {
-				if strings.Contains(c, ci.Monitors[id].Endpoint) {
+				if strings.Contains(c, ci.InternalMonitors[id].Endpoint) {
 					contains = true
 				}
 			}
 			assert.True(t, contains)
-			assert.Contains(t, mim, ci.Monitors[id].Name)
+			assert.Contains(t, mim, ci.InternalMonitors[id].Name)
 		}
 	}
 
@@ -80,14 +80,14 @@ func TestStore(t *testing.T) {
 
 	// Now run the same test for v1 endpoints
 	mon1EndpointsEnabled = true
-	for _, mon := range i1.Monitors {
+	for _, mon := range i1.InternalMonitors {
 		mon.Endpoint = "1.2.3.4:6789"
 	}
 	err = s.CreateOrUpdate(i1)
 	assert.NoError(t, err)
 	assertConfigStore(i1)
 
-	for _, mon := range i3.Monitors {
+	for _, mon := range i3.InternalMonitors {
 		mon.Endpoint = "1.2.3.4:6789"
 	}
 	err = s.CreateOrUpdate(i3)