Revert "mgr: remove mgr sidecar"

This reverts commit ff75ec96b7.

Signed-off-by: Redouane Kachach <rkachach@redhat.com>
(cherry picked from commit 456a0c328c)

Documentation/CRDs/Cluster/ceph-cluster-crd.md

@@ -57,8 +57,9 @@ If this value is empty, each pod will get an ephemeral directory to store their
 For more details on the mons and when to choose a number other than `3`, see the [mon health doc](../../Storage-Configuration/Advanced/ceph-mon-health.md).
 * `mgr`: manager top level section
     * `count`: set number of ceph managers between `1` to `2`. The default value is 2.
-    If there are two managers, all mgr services will point to the active mgr. Due to a readiness probe,
-	the standby mgr pod stays in the `not-Ready` state until Ceph internally changes it to active state.
+    If there are two managers, it is important for all mgr services point to the active mgr and not the passive mgr. Therefore, Rook will
+    automatically update all services (in the cluster namespace) that have a label `app=rook-ceph-mgr` with a selector pointing to the
+    active mgr. This commonly applies to services for the dashboard or the prometheus metrics collector.
     * `modules`: is the list of Ceph manager modules to enable
 * `crashCollector`: The settings for crash collector daemon(s).
     * `disable`: is set to `true`, the crash collector will not run on any node where a Ceph daemon runs
@@ -504,6 +505,9 @@ You can set resource requests/limits for Rook components through the [Resource R
 * `osd-<deviceClass>`: Set resource requests/limits for OSDs on a specific device class. Rook will automatically detect `hdd`,
   `ssd`, or `nvme` device classes. Custom device classes can also be set.
 * `mgr`: Set resource requests/limits for MGRs
+* `mgr-sidecar`: Set resource requests/limits for the MGR sidecar, which is only created when `mgr.count: 2`.
+  The sidecar requires very few resources since it only executes every 15 seconds to query Ceph for the active
+  mgr and update the mgr services if the active mgr changed.
 * `prepareosd`: Set resource requests/limits for OSD prepare job
 * `crashcollector`: Set resource requests/limits for crash. This pod runs wherever there is a Ceph pod running.
 It scrapes for Ceph daemon core dumps and sends them to the Ceph manager crash module so that core dumps are centralized and can be easily listed/accessed.
@@ -518,6 +522,7 @@ If a user configures a limit or request value that is too low, Rook will still r
 * `mgr`: 512MB
 * `osd`: 2048MB
 * `crashcollector`: 60MB
+* `mgr-sidecar`: 100MB limit, 40MB requests
 * `prepareosd`: no limits (see the note)
 
 !!! note

cmd/rook/ceph/ceph.go

@@ -58,6 +58,7 @@ func init() {
 	Cmd.AddCommand(cleanUpCmd,
 		operatorCmd,
 		osdCmd,
+		mgrCmd,
 		configCmd)
 }
 

cmd/rook/ceph/mgr.go

@@ -0,0 +1,108 @@
+/*
+Copyright 2021 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 ceph
+
+import (
+	"path"
+	"time"
+
+	"github.com/rook/rook/cmd/rook/rook"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/operator/ceph/cluster/mgr"
+	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
+	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/util/flags"
+	"github.com/spf13/cobra"
+)
+
+var mgrCmd = &cobra.Command{
+	Use: "mgr",
+}
+var mgrSidecarCmd = &cobra.Command{
+	Use: "watch-active",
+}
+var (
+	updateMgrServicesInterval string
+	daemonName                string
+	clusterSpec               cephv1.ClusterSpec
+	rawCephVersion            string
+)
+
+func init() {
+	addCephFlags(mgrCmd)
+
+	// add the subcommands to the parent mgr command
+	mgrCmd.AddCommand(mgrSidecarCmd)
+
+	mgrSidecarCmd.Flags().BoolVar(&clusterSpec.Dashboard.Enabled, "dashboard-enabled", false, "whether the dashboard is enabled")
+	mgrSidecarCmd.Flags().BoolVar(&clusterSpec.Monitoring.Enabled, "monitoring-enabled", false, "whether the monitoring is enabled")
+	mgrSidecarCmd.Flags().StringVar(&updateMgrServicesInterval, "update-interval", "", "the interval at which to update the mgr services")
+	mgrSidecarCmd.Flags().StringVar(&ownerRefID, "cluster-id", "", "the UID of the cluster CR that owns this cluster")
+	mgrSidecarCmd.Flags().StringVar(&clusterName, "cluster-name", "", "the name of the cluster CR that owns this cluster")
+	mgrSidecarCmd.Flags().StringVar(&daemonName, "daemon-name", "", "the name of the local mgr daemon")
+	mgrSidecarCmd.Flags().StringVar(&rawCephVersion, "ceph-version", "", "the version of ceph")
+
+	flags.SetFlagsFromEnv(mgrCmd.Flags(), rook.RookEnvVarPrefix)
+	flags.SetFlagsFromEnv(mgrSidecarCmd.Flags(), rook.RookEnvVarPrefix)
+	mgrSidecarCmd.RunE = runMgrSidecar
+}
+
+// Start the mgr daemon sidecar
+func runMgrSidecar(cmd *cobra.Command, args []string) error {
+	rook.SetLogLevel()
+	clusterInfo.Context = cmd.Context()
+
+	if err := readCephSecret(path.Join(mon.CephSecretMountPath, mon.CephSecretFilename)); err != nil {
+		rook.TerminateFatal(err)
+	}
+
+	context := createContext()
+	clusterInfo.Monitors = mon.ParseMonEndpoints(cfg.monEndpoints)
+	rook.LogStartupInfo(mgrSidecarCmd.Flags())
+
+	ownerRef := opcontroller.ClusterOwnerRef(clusterName, ownerRefID)
+	clusterInfo.OwnerInfo = k8sutil.NewOwnerInfoWithOwnerRef(&ownerRef, clusterInfo.Namespace)
+
+	if err := client.WriteCephConfig(context, &clusterInfo); err != nil {
+		rook.TerminateFatal(err)
+	}
+
+	interval, err := time.ParseDuration(updateMgrServicesInterval)
+	if err != nil {
+		rook.TerminateFatal(err)
+	}
+
+	version, err := cephver.ExtractCephVersion(rawCephVersion)
+	if err != nil {
+		rook.TerminateFatal(err)
+	}
+	clusterInfo.CephVersion = *version
+
+	m := mgr.New(context, &clusterInfo, clusterSpec, "")
+	for {
+		err := m.ReconcileActiveMgrServices(daemonName)
+		if err != nil {
+			logger.Errorf("failed to reconcile services. %v", err)
+		} else {
+			logger.Infof("successfully reconciled services. checking again in %ds", (int)(interval.Seconds()))
+		}
+		time.Sleep(interval)
+	}
+}

deploy/charts/rook-ceph-cluster/values.yaml

@@ -303,6 +303,13 @@ cephClusterSpec:
       requests:
         cpu: "500m"
         memory: "50Mi"
+    mgr-sidecar:
+      limits:
+        cpu: "500m"
+        memory: "100Mi"
+      requests:
+        cpu: "100m"
+        memory: "40Mi"
     crashcollector:
       limits:
         cpu: "500m"

deploy/examples/cluster.yaml

@@ -220,6 +220,7 @@ spec:
   #   osd-ssd:
   #   osd-nvme:
   #   prepareosd:
+  #   mgr-sidecar:
   #   crashcollector:
   #   logcollector:
   #   cleanup:

deploy/examples/monitoring/service-monitor.yaml

@@ -13,6 +13,7 @@ spec:
     matchLabels:
       app: rook-ceph-mgr
       rook_cluster: rook-ceph
+      ceph_daemon_id: a
   endpoints:
   - port: http-metrics
     path: /metrics

pkg/apis/ceph.rook.io/v1/resources.go

@@ -25,6 +25,8 @@ const (
 	ResourcesKeyMon = "mon"
 	// ResourcesKeyMgr represents the name of resource in the CR for a mgr
 	ResourcesKeyMgr = "mgr"
+	// ResourcesKeyMgrSidecar represents the name of resource in the CR for a mgr
+	ResourcesKeyMgrSidecar = "mgr-sidecar"
 	// ResourcesKeyOSD represents the name of a resource in the CR for all OSDs
 	ResourcesKeyOSD = "osd"
 	// ResourcesKeyPrepareOSD represents the name of resource in the CR for the osd prepare job
@@ -50,6 +52,11 @@ func GetMgrResources(p ResourceSpec) v1.ResourceRequirements {
 	return p[ResourcesKeyMgr]
 }
 
+// GetMgrSidecarResources returns the placement for the MGR sidecar container
+func GetMgrSidecarResources(p ResourceSpec) v1.ResourceRequirements {
+	return p[ResourcesKeyMgrSidecar]
+}
+
 // GetMonResources returns the placement for the monitors
 func GetMonResources(p ResourceSpec) v1.ResourceRequirements {
 	return p[ResourcesKeyMon]

pkg/operator/ceph/cluster/cluster_external.go

@@ -210,7 +210,7 @@ func (c *ClusterController) configureExternalClusterMonitoring(context *clusterd
 	)
 
 	// Create external monitoring Service
-	service, err := manager.MakeMetricsService(opcontroller.ExternalMgrAppName, opcontroller.ServiceExternalMetricName)
+	service, err := manager.MakeMetricsService(opcontroller.ExternalMgrAppName, "", opcontroller.ServiceExternalMetricName)
 	if err != nil {
 		return err
 	}
@@ -230,7 +230,7 @@ func (c *ClusterController) configureExternalClusterMonitoring(context *clusterd
 	// Deploy external ServiceMonitor
 	logger.Info("creating external service monitor")
 	// servicemonitor takes some metadata from the service for easy mapping
-	err = manager.EnableServiceMonitor()
+	err = manager.EnableServiceMonitor("")
 	if err != nil {
 		logger.Errorf("failed to enable external service monitor. %v", err)
 	} else {

pkg/operator/ceph/cluster/mgr/dashboard.go

@@ -54,8 +54,8 @@ var (
 	dashboardInitWaitTime = 5 * time.Second
 )
 
-func (c *Cluster) configureDashboardService() error {
-	dashboardService, err := c.makeDashboardService(AppName)
+func (c *Cluster) configureDashboardService(activeDaemon string) error {
+	dashboardService, err := c.makeDashboardService(AppName, activeDaemon)
 	if err != nil {
 		return err
 	}

pkg/operator/ceph/cluster/mgr/dashboard_test.go

@@ -131,7 +131,7 @@ func TestStartSecureDashboard(t *testing.T) {
 	}
 
 	dashboardInitWaitTime = 0
-	err := c.configureDashboardService()
+	err := c.configureDashboardService("a")
 	assert.NoError(t, err)
 	err = c.configureDashboardModules()
 	assert.NoError(t, err)
@@ -148,7 +148,7 @@ func TestStartSecureDashboard(t *testing.T) {
 
 	// disable the dashboard
 	c.spec.Dashboard.Enabled = false
-	err = c.configureDashboardService()
+	err = c.configureDashboardService("a")
 	assert.Nil(t, err)
 	err = c.configureDashboardModules()
 	assert.NoError(t, err)
@@ -163,7 +163,7 @@ func TestStartSecureDashboard(t *testing.T) {
 	// Set the port to something over 1024 and confirm the port and targetPort are the same
 	c.spec.Dashboard.Enabled = true
 	c.spec.Dashboard.Port = 1025
-	err = c.configureDashboardService()
+	err = c.configureDashboardService("a")
 	assert.Nil(t, err)
 
 	svc, err = c.context.Clientset.CoreV1().Services(clusterInfo.Namespace).Get(ctx, "rook-ceph-mgr-dashboard", metav1.GetOptions{})
@@ -175,7 +175,7 @@ func TestStartSecureDashboard(t *testing.T) {
 	// Fall back to the default port
 	c.spec.Dashboard.Enabled = true
 	c.spec.Dashboard.Port = 0
-	err = c.configureDashboardService()
+	err = c.configureDashboardService("a")
 	assert.Nil(t, err)
 
 	svc, err = c.context.Clientset.CoreV1().Services(clusterInfo.Namespace).Get(ctx, "rook-ceph-mgr-dashboard", metav1.GetOptions{})

pkg/operator/ceph/cluster/mgr/mgr.go

@@ -177,7 +177,18 @@ func (c *Cluster) Start() error {
 	// check if any extra mgrs need to be removed
 	c.removeExtraMgrs(daemonIDs)
 
+	activeMgr := daemonIDs[0]
 	if len(daemonIDs) > 1 {
+		// When multiple mgrs are running, the mgr sidecar for the active mgr
+		// will create the services. However, the sidecar will only reconcile all
+		// the services when the active mgr changes. Here as part of the regular reconcile
+		// we trigger reconciling all the services to ensure they are current.
+		activeMgr, err = c.getActiveMgr()
+		if err != nil || activeMgr == "" {
+			activeMgr = ""
+			logger.Infof("cannot reconcile mgr services, no active mgr found. err=%v", err)
+		}
+
 		// reconcile mgr PDB
 		if err := c.reconcileMgrPDB(); err != nil {
 			return errors.Wrap(err, "failed to reconcile mgr PDB")
@@ -186,9 +197,10 @@ func (c *Cluster) Start() error {
 		// delete MGR PDB as the count is less than 2
 		c.deleteMgrPDB()
 	}
-
-	if err := c.reconcileServices(); err != nil {
-		return errors.Wrap(err, "failed to enable mgr services")
+	if activeMgr != "" {
+		if err := c.reconcileServices(activeMgr); err != nil {
+			return errors.Wrap(err, "failed to enable mgr services")
+		}
 	}
 
 	// configure the mgr modules
@@ -209,15 +221,55 @@ func (c *Cluster) removeExtraMgrs(daemonIDs []string) {
 	}
 }
 
-// reconcile the services, if the active mgr is not detected, use the default mgr
-func (c *Cluster) reconcileServices() error {
+// ReconcileActiveMgrServices reconciles the services if the active mgr is the one running
+// in the sidecar
+func (c *Cluster) ReconcileActiveMgrServices(daemonNameToUpdate string) error {
+	// If the services are already set to this daemon, no need to attempt to update
+	svc, err := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Get(c.clusterInfo.Context, AppName, metav1.GetOptions{})
+	if err != nil {
+		logger.Errorf("failed to check current mgr service, proceeding to update. %v", err)
+	} else {
+		currentDaemon := svc.Spec.Selector[controller.DaemonIDLabel]
+		if currentDaemon == daemonNameToUpdate {
+			logger.Infof("mgr services already set to daemon %q, no need to update", daemonNameToUpdate)
+			return nil
+		}
+		logger.Infof("mgr service currently set to %q, checking if need to update to %q", currentDaemon, daemonNameToUpdate)
+	}
 
-	if err := c.configureDashboardService(); err != nil {
+	activeName, err := c.getActiveMgr()
+	if err != nil {
+		return err
+	}
+	if activeName == "" {
+		return errors.New("active mgr not found")
+	}
+	if daemonNameToUpdate != activeName {
+		logger.Infof("no need for the mgr update since the active mgr is %q, rather than the local mgr %q", activeName, daemonNameToUpdate)
+		return nil
+	}
+
+	return c.reconcileServices(activeName)
+}
+
+func (c *Cluster) getActiveMgr() (string, error) {
+	mgrStat, err := cephclient.CephMgrStat(c.context, c.clusterInfo)
+	if err != nil {
+		return "", errors.Wrap(err, "failed to get mgr stat for the active mgr")
+	}
+	return mgrStat.ActiveName, nil
+}
+
+// reconcile the services, if the active mgr is not detected, use the default mgr
+func (c *Cluster) reconcileServices(activeDaemon string) error {
+	logger.Infof("setting services to point to mgr %q", activeDaemon)
+
+	if err := c.configureDashboardService(activeDaemon); err != nil {
 		return errors.Wrap(err, "failed to configure dashboard svc")
 	}
 
 	// create the metrics service
-	service, err := c.MakeMetricsService(AppName, serviceMetricName)
+	service, err := c.MakeMetricsService(AppName, activeDaemon, serviceMetricName)
 	if err != nil {
 		return err
 	}
@@ -227,17 +279,17 @@ func (c *Cluster) reconcileServices() error {
 
 	// enable monitoring if `monitoring: enabled: true`
 	if c.spec.Monitoring.Enabled {
-		if err := c.EnableServiceMonitor(); err != nil {
+		if err := c.EnableServiceMonitor(activeDaemon); err != nil {
 			return errors.Wrap(err, "failed to enable service monitor")
 		}
 	}
 
-	return c.updateServiceSelectors()
+	return c.updateServiceSelectors(activeDaemon)
 }
 
-// For the upgrade scenario: remove any selector DaemonIDLabel from the all
-// the services since new mgr HA doesn't rely on this label anymore.
-func (c *Cluster) updateServiceSelectors() error {
+// Make a best effort to update the services that have been labeled for being updated
+// when the mgr has changed. They might be services for node ports, ingress, etc
+func (c *Cluster) updateServiceSelectors(activeDaemon string) error {
 	selector := metav1.ListOptions{LabelSelector: "app=rook-ceph-mgr"}
 	services, err := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).List(c.clusterInfo.Context, selector)
 	if err != nil {
@@ -245,20 +297,21 @@ func (c *Cluster) updateServiceSelectors() error {
 	}
 	for i, service := range services.Items {
 		if service.Spec.Selector == nil {
+			service.Spec.Selector = map[string]string{}
+		}
+		// Update the selector on the service to point to the active mgr
+		if service.Spec.Selector[controller.DaemonIDLabel] == activeDaemon {
+			logger.Infof("no need to update service %q", service.Name)
 			continue
 		}
-		// Check if the service has a DaemonIDLabel (legacy mgr HA implementation) and remove it
-		_, hasDaemonLabel := service.Spec.Selector[controller.DaemonIDLabel]
-		if hasDaemonLabel {
-			delete(service.Spec.Selector, controller.DaemonIDLabel)
-			logger.Infof("removing %s selector label on mgr service %q", controller.DaemonIDLabel, service.Name)
-			if _, err := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Update(c.clusterInfo.Context, &services.Items[i], metav1.UpdateOptions{}); err != nil {
-				logger.Errorf("failed to update service %q. %v", service.Name, err)
-			} else {
-				logger.Infof("service %q successfully updated", service.Name)
-			}
+		// Update the service to point to the new active mgr
+		service.Spec.Selector[controller.DaemonIDLabel] = activeDaemon
+		logger.Infof("updating selector on mgr service %q to active mgr %q", service.Name, activeDaemon)
+		if _, err := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Update(c.clusterInfo.Context, &services.Items[i], metav1.UpdateOptions{}); err != nil {
+			logger.Errorf("failed to update service %q. %v", service.Name, err)
+		} else {
+			logger.Infof("service %q successfully updated to active mgr %q", service.Name, activeDaemon)
 		}
-
 	}
 	return nil
 }
@@ -377,7 +430,7 @@ func wellKnownModule(name string) bool {
 }
 
 // EnableServiceMonitor add a servicemonitor that allows prometheus to scrape from the monitoring endpoint of the cluster
-func (c *Cluster) EnableServiceMonitor() error {
+func (c *Cluster) EnableServiceMonitor(activeDaemon string) error {
 	serviceMonitor, err := k8sutil.GetServiceMonitor(path.Join(monitoringPath, serviceMonitorFile))
 	if err != nil {
 		return errors.Wrap(err, "service monitor could not be enabled")
@@ -394,7 +447,7 @@ func (c *Cluster) EnableServiceMonitor() error {
 		return errors.Wrapf(err, "failed to set owner reference to service monitor %q", serviceMonitor.Name)
 	}
 	serviceMonitor.Spec.NamespaceSelector.MatchNames = []string{c.clusterInfo.Namespace}
-	serviceMonitor.Spec.Selector.MatchLabels = c.selectorLabels()
+	serviceMonitor.Spec.Selector.MatchLabels = c.selectorLabels(activeDaemon)
 
 	applyMonitoringLabels(c, serviceMonitor)
 

pkg/operator/ceph/cluster/mgr/mgr_test.go

@@ -140,6 +140,10 @@ func validateStart(t *testing.T, c *Cluster) {
 		assert.Equal(t, "my-priority-class", d.Spec.Template.Spec.PriorityClassName)
 		if c.spec.Mgr.Count == 1 {
 			assert.Equal(t, 1, len(d.Spec.Template.Spec.Containers))
+		} else {
+			// The sidecar container is only there when multiple mgrs are enabled
+			assert.Equal(t, 2, len(d.Spec.Template.Spec.Containers))
+			assert.Equal(t, "watch-active", d.Spec.Template.Spec.Containers[1].Name)
 		}
 	}
 
@@ -184,52 +188,116 @@ func TestUpdateServiceSelectors(t *testing.T) {
 	}
 	c := &Cluster{spec: spec, context: ctx, clusterInfo: clusterInfo}
 
-	// Make sure we remove the daemon_id label from the selector
-	// of all services with a label "app=rook-ceph-mgr"
-	t.Run("remove daemon_id from mgr services", func(t *testing.T) {
-		svc_1 := corev1.Service{
-			ObjectMeta: metav1.ObjectMeta{
-				Name: "svc_1",
-				Labels: map[string]string{
-					"app":                    "rook-ceph-mgr",
-					"svc":                    "rook-ceph-mgr",
-					controller.DaemonIDLabel: "a",
-				},
-			},
-			Spec: corev1.ServiceSpec{
-				Selector: map[string]string{
-					"app":                    "rook-ceph-mgr",
-					controller.DaemonIDLabel: "a",
-				}},
-		}
-		svc_2 := corev1.Service{
-			ObjectMeta: metav1.ObjectMeta{Name: "svc_2"},
-			Spec: corev1.ServiceSpec{
-				Selector: map[string]string{
-					"app":                    "rook-ceph-mgr",
-					controller.DaemonIDLabel: "a",
-				}},
-		}
-		_, err := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Create(clusterInfo.Context, &svc_1, metav1.CreateOptions{})
+	t.Run("initial active daemon", func(t *testing.T) {
+		activeDaemon := "a"
+		err := c.reconcileServices(activeDaemon)
 		assert.NoError(t, err)
-		_, err_2 := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Create(clusterInfo.Context, &svc_2, metav1.CreateOptions{})
-		assert.NoError(t, err_2)
-
-		// Check the the label has been only removed from svc_1
-		err = c.updateServiceSelectors()
-		assert.NoError(t, err)
-
-		updated_svc_1, err := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Get(clusterInfo.Context, "svc_1", metav1.GetOptions{})
-		assert.NoError(t, err)
-		_, hasDaemonLabel := updated_svc_1.Spec.Selector[controller.DaemonIDLabel]
-		assert.Equal(t, hasDaemonLabel, false)
-
-		updated_svc_2, err := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Get(clusterInfo.Context, "svc_2", metav1.GetOptions{})
-		assert.NoError(t, err)
-		_, hasDaemonLabel = updated_svc_2.Spec.Selector[controller.DaemonIDLabel]
-		assert.Equal(t, hasDaemonLabel, true)
-
+		validateServiceActiveDaemon(t, c, activeDaemon, 2, 0)
 	})
+
+	t.Run("update active daemon", func(t *testing.T) {
+		activeDaemon := "b"
+		err := c.updateServiceSelectors(activeDaemon)
+		assert.NoError(t, err)
+		validateServiceActiveDaemon(t, c, activeDaemon, 2, 0)
+	})
+
+	t.Run("skip non-mgr services", func(t *testing.T) {
+		svc := corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "mysvc"}}
+		_, err := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Create(clusterInfo.Context, &svc, metav1.CreateOptions{})
+		assert.NoError(t, err)
+
+		activeDaemon := "c"
+		err = c.updateServiceSelectors(activeDaemon)
+		assert.NoError(t, err)
+		validateServiceActiveDaemon(t, c, activeDaemon, 2, 1)
+	})
+}
+
+func validateServiceActiveDaemon(t *testing.T, c *Cluster, activeDaemon string, expectedUpdated, expectedSkipped int) {
+	services, err := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).List(c.clusterInfo.Context, metav1.ListOptions{})
+	assert.NoError(t, err)
+	skipped := 0
+	updated := 0
+	for _, service := range services.Items {
+		if service.Labels["app"] == "rook-ceph-mgr" {
+			updated++
+			assert.Equal(t, activeDaemon, service.Spec.Selector[controller.DaemonIDLabel])
+		} else {
+			skipped++
+			assert.Equal(t, "", service.Spec.Selector[controller.DaemonIDLabel])
+		}
+	}
+	assert.Equal(t, expectedUpdated, updated)
+	assert.Equal(t, expectedSkipped, skipped)
+}
+
+func TestMgrSidecarReconcile(t *testing.T) {
+	activeMgr := "a"
+	calledMgrStat := false
+	spec := cephv1.ClusterSpec{
+		Mgr: cephv1.MgrSpec{Count: 1},
+		Dashboard: cephv1.DashboardSpec{
+			Enabled: true,
+			Port:    7000,
+		},
+	}
+	executor := &exectest.MockExecutor{
+		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			logger.Infof("Command: %s %v", command, args)
+			if args[1] == "stat" {
+				calledMgrStat = true
+			}
+			return fmt.Sprintf(`{"active_name":"%s"}`, activeMgr), nil
+		},
+	}
+	clientset := testop.New(t, 3)
+	configDir := t.TempDir()
+	ctx := &clusterd.Context{
+		Executor:  executor,
+		ConfigDir: configDir,
+		Clientset: clientset,
+	}
+	clusterInfo := cephclient.AdminTestClusterInfo("mycluster")
+	clusterInfo.SetName("test")
+	c := &Cluster{spec: spec, context: ctx, clusterInfo: clusterInfo}
+
+	// Update services according to the active mgr
+	calledMgrStat = false
+	clusterInfo.CephVersion = cephver.CephVersion{Major: 17, Minor: 2, Build: 0}
+	err := c.ReconcileActiveMgrServices(activeMgr)
+	assert.NoError(t, err)
+	assert.True(t, calledMgrStat)
+	validateServices(t, c)
+	validateServiceMatches(t, c, "a")
+
+	// nothing is created or updated when the requested mgr is not the active mgr
+	clusterInfo.CephVersion = cephver.CephVersion{Major: 16, Minor: 2, Build: 5}
+	err = c.ReconcileActiveMgrServices("b")
+	assert.NoError(t, err)
+	assert.True(t, calledMgrStat)
+	_, err = c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Get(context.TODO(), "rook-ceph-mgr", metav1.GetOptions{})
+	assert.True(t, kerrors.IsNotFound(err))
+
+	// nothing is updated when the requested mgr is not the active mgr
+	activeMgr = "b"
+	err = c.ReconcileActiveMgrServices("b")
+	assert.NoError(t, err)
+	validateServices(t, c)
+	validateServiceMatches(t, c, "b")
+}
+
+func validateServiceMatches(t *testing.T, c *Cluster, expectedActive string) {
+	// The service labels should match the active mgr
+	svc, err := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Get(context.TODO(), "rook-ceph-mgr", metav1.GetOptions{})
+	assert.NoError(t, err)
+	matchDaemon, ok := svc.Spec.Selector["ceph_daemon_id"]
+	assert.True(t, ok)
+	assert.Equal(t, expectedActive, matchDaemon)
+
+	// clean up the service for the next test
+	err = c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Delete(context.TODO(), "rook-ceph-mgr", metav1.DeleteOptions{})
+	assert.NoError(t, err)
 }
 
 func TestConfigureModules(t *testing.T) {

pkg/operator/ceph/cluster/mgr/spec.go

@@ -70,8 +70,9 @@ func (c *Cluster) makeDeployment(mgrConfig *mgrConfig) (*apps.Deployment, error)
 	}
 	cephv1.GetMgrPlacement(c.spec.Placement).ApplyToPodSpec(&podSpec.Spec)
 
-	// Require anti affinity only if there are multiple mgrs
+	// Run the sidecar and require anti affinity only if there are multiple mgrs
 	if c.spec.Mgr.Count > 1 {
+		podSpec.Spec.Containers = append(podSpec.Spec.Containers, c.makeMgrSidecarContainer(mgrConfig))
 		matchLabels := controller.AppLabels(AppName, c.clusterInfo.Namespace)
 		podSpec.Spec.Volumes = append(podSpec.Spec.Volumes, mon.CephSecretVolume())
 
@@ -208,6 +209,35 @@ func (c *Cluster) makeMgrDaemonContainer(mgrConfig *mgrConfig) v1.Container {
 	return container
 }
 
+func (c *Cluster) makeMgrSidecarContainer(mgrConfig *mgrConfig) v1.Container {
+	envVars := []v1.EnvVar{
+		{Name: "ROOK_CLUSTER_ID", Value: string(c.clusterInfo.OwnerInfo.GetUID())},
+		{Name: "ROOK_CLUSTER_NAME", Value: string(c.clusterInfo.NamespacedName().Name)},
+		k8sutil.PodIPEnvVar(k8sutil.PrivateIPEnvVar),
+		k8sutil.PodIPEnvVar(k8sutil.PublicIPEnvVar),
+		mon.PodNamespaceEnvVar(c.clusterInfo.Namespace),
+		mon.EndpointEnvVar(),
+		mon.CephUsernameEnvVar(),
+		k8sutil.ConfigOverrideEnvVar(),
+		{Name: "ROOK_DASHBOARD_ENABLED", Value: strconv.FormatBool(c.spec.Dashboard.Enabled)},
+		{Name: "ROOK_MONITORING_ENABLED", Value: strconv.FormatBool(c.spec.Monitoring.Enabled)},
+		{Name: "ROOK_UPDATE_INTERVAL", Value: "15s"},
+		{Name: "ROOK_DAEMON_NAME", Value: mgrConfig.DaemonID},
+		{Name: "ROOK_CEPH_VERSION", Value: "ceph version " + c.clusterInfo.CephVersion.String()},
+	}
+
+	return v1.Container{
+		Args:            []string{"ceph", "mgr", "watch-active"},
+		Name:            "watch-active",
+		Image:           c.rookVersion,
+		ImagePullPolicy: controller.GetContainerImagePullPolicy(c.spec.CephVersion.ImagePullPolicy),
+		Env:             envVars,
+		Resources:       cephv1.GetMgrSidecarResources(c.spec.Resources),
+		SecurityContext: controller.PrivilegedContext(true),
+		VolumeMounts:    []v1.VolumeMount{mon.CephSecretVolumeMount()},
+	}
+}
+
 func (c *Cluster) makeCmdProxySidecarContainer(mgrConfig *mgrConfig) v1.Container {
 	_, adminKeyringVolMount := keyring.Volume().Admin(), keyring.VolumeMount().Admin()
 	container := v1.Container{
@@ -226,8 +256,8 @@ func (c *Cluster) makeCmdProxySidecarContainer(mgrConfig *mgrConfig) v1.Containe
 }
 
 // MakeMetricsService generates the Kubernetes service object for the monitoring service
-func (c *Cluster) MakeMetricsService(name, servicePortMetricName string) (*v1.Service, error) {
-	labels := c.selectorLabels()
+func (c *Cluster) MakeMetricsService(name, activeDaemon, servicePortMetricName string) (*v1.Service, error) {
+	labels := c.selectorLabels(activeDaemon)
 
 	svc := &v1.Service{
 		ObjectMeta: metav1.ObjectMeta{
@@ -259,8 +289,8 @@ func (c *Cluster) MakeMetricsService(name, servicePortMetricName string) (*v1.Se
 	return svc, nil
 }
 
-func (c *Cluster) makeDashboardService(name string) (*v1.Service, error) {
-	labels := c.selectorLabels()
+func (c *Cluster) makeDashboardService(name, activeDaemon string) (*v1.Service, error) {
+	labels := c.selectorLabels(activeDaemon)
 
 	portName := "https-dashboard"
 	if !c.spec.Dashboard.SSL {
@@ -323,7 +353,10 @@ func (c *Cluster) cephMgrOrchestratorModuleEnvs() []v1.EnvVar {
 	return envVars
 }
 
-func (c *Cluster) selectorLabels() map[string]string {
+func (c *Cluster) selectorLabels(activeDaemon string) map[string]string {
 	labels := controller.AppLabels(AppName, c.clusterInfo.Namespace)
+	if activeDaemon != "" {
+		labels[controller.DaemonIDLabel] = activeDaemon
+	}
 	return labels
 }

pkg/operator/ceph/cluster/mgr/spec_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/pkg/operator/ceph/config"
+	"github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/test"
 	optest "github.com/rook/rook/pkg/operator/test"
 	"github.com/stretchr/testify/assert"
@@ -112,13 +113,14 @@ func TestServiceSpec(t *testing.T) {
 	clusterSpec := cephv1.ClusterSpec{}
 	c := New(&clusterd.Context{Clientset: clientset}, clusterInfo, clusterSpec, "myversion")
 
-	s, err := c.MakeMetricsService("rook-mgr", serviceMetricName)
+	s, err := c.MakeMetricsService("rook-mgr", "foo", serviceMetricName)
 	assert.NoError(t, err)
 	assert.NotNil(t, s)
 	assert.Equal(t, "rook-mgr", s.Name)
 	assert.Equal(t, 1, len(s.Spec.Ports))
-	assert.Equal(t, 2, len(s.Labels))
-	assert.Equal(t, 2, len(s.Spec.Selector))
+	assert.Equal(t, 3, len(s.Labels))
+	assert.Equal(t, 3, len(s.Spec.Selector))
+	assert.Equal(t, "foo", s.Spec.Selector[controller.DaemonIDLabel])
 }
 
 func TestHostNetwork(t *testing.T) {