Merge pull request #11847 from rook/mergify/bp/release-1.11/pr-11829

mgr: revert readiness probe and go back to the original sidecar HA implementation (backport #11829)

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

@@ -28,6 +28,7 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	"github.com/rook/rook/pkg/operator/ceph/config"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
@@ -77,7 +78,8 @@ func New(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, spec ce
 	}
 }
 
-var waitForPodsWithLabelToRun = k8sutil.WaitForPodsWithLabelToRun
+var waitForDeploymentToStart = k8sutil.WaitForDeploymentToStart
+var updateDeploymentAndWait = mon.UpdateCephDeploymentAndWait
 
 // for backward compatibility, default to 1 mgr
 func (c *Cluster) getReplicas() int {
@@ -149,16 +151,8 @@ func (c *Cluster) Start() error {
 			}
 			logger.Infof("deployment for mgr %s already exists. updating if needed", resourceName)
 
-			_, err = c.context.Clientset.AppsV1().Deployments(c.clusterInfo.Namespace).Update(c.clusterInfo.Context, d, metav1.UpdateOptions{})
-			if err != nil {
-				return errors.Wrapf(err, "failed to update mgr deployment %q", resourceName)
-			}
-			if len(daemonIDs) > 1 {
-				// wait for the updated mgr to start running if we have more than one mgr
-				daemonLabels := fmt.Sprintf("%s=%s,%s=%s", controller.DaemonTypeLabel, cephv1.KeyMgr, controller.DaemonIDLabel, daemonID)
-				if err := waitForPodsWithLabelToRun(c.clusterInfo.Context, c.context.Clientset, c.clusterInfo.Namespace, daemonLabels); err != nil {
-					return errors.Wrapf(err, "failed to wait for mgr pod %q to run", daemonID)
-				}
+			if err := updateDeploymentAndWait(c.context, c.clusterInfo, d, config.MgrType, mgrConfig.DaemonID, c.spec.SkipUpgradeChecks, false); err != nil {
+				logger.Errorf("failed to update mgr deployment %q. %v", resourceName, err)
 			}
 		} else {
 			// wait for the new deployment
@@ -170,18 +164,31 @@ func (c *Cluster) Start() error {
 	// If we're waiting for the mgr deployments to start, it is a clean deployment
 	if len(deploymentsToWaitFor) > 0 {
 		config.DisableInsecureGlobalID(c.context, c.clusterInfo)
+	}
 
-		// Wait for the new mgr pods to start
-		mgrLabel := fmt.Sprintf("%s=%s", controller.DaemonTypeLabel, cephv1.KeyMgr)
-		if err := waitForPodsWithLabelToRun(c.clusterInfo.Context, c.context.Clientset, c.clusterInfo.Namespace, mgrLabel); err != nil {
-			return errors.Wrap(err, "failed to wait for mgr pods to start")
+	// If the mgr is newly created, wait for it to start before continuing with the service and
+	// module configuration
+	for _, d := range deploymentsToWaitFor {
+		if err := waitForDeploymentToStart(c.clusterInfo.Context, c.context, d); err != nil {
+			return errors.Wrapf(err, "failed to wait for mgr %q to start", d.Name)
 		}
 	}
 
 	// 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")
@@ -190,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
@@ -213,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
 	}
@@ -231,41 +279,41 @@ 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")
 		}
 	}
 
-	c.updateServiceSelectors()
-	return nil
+	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() {
+// 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 {
-		logger.Errorf("failed to query mgr services to update: %v", err)
-		return
+		return errors.Wrap(err, "failed to query mgr services to update")
 	}
 	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
 }
 
 func (c *Cluster) configureModules(daemonIDs []string) {
@@ -382,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")
@@ -399,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

@@ -31,20 +31,23 @@ import (
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
+	testopk8s "github.com/rook/rook/pkg/operator/k8sutil/test"
 	testop "github.com/rook/rook/pkg/operator/test"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	apps "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	policyv1 "k8s.io/api/policy/v1"
 	policyv1beta1 "k8s.io/api/policy/v1beta1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
 func TestStartMgr(t *testing.T) {
+	var deploymentsUpdated *[]*apps.Deployment
+	updateDeploymentAndWait, deploymentsUpdated = testopk8s.UpdateDeploymentAndWaitStub()
 
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
@@ -55,8 +58,8 @@ func TestStartMgr(t *testing.T) {
 			return "{\"key\":\"mysecurekey\"}", nil
 		},
 	}
-	waitForPodsWithLabelToRun = func(ctx context.Context, clientset kubernetes.Interface, namespace, label string) error {
-		logger.Infof("simulated mgr deployment waiting for deployment")
+	waitForDeploymentToStart = func(ctx context.Context, clusterdContext *clusterd.Context, deployment *apps.Deployment) error {
+		logger.Infof("simulated mgr deployment starting")
 		return nil
 	}
 
@@ -93,12 +96,16 @@ func TestStartMgr(t *testing.T) {
 	err = c.Start()
 	assert.NoError(t, err)
 	validateStart(t, c)
+	assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
+	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
 
 	c.spec.Dashboard.URLPrefix = "/test"
 	c.spec.Dashboard.Port = 12345
 	err = c.Start()
 	assert.NoError(t, err)
 	validateStart(t, c)
+	assert.ElementsMatch(t, []string{"rook-ceph-mgr-a"}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
+	testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
 
 	// starting with more replicas
 	c.spec.Mgr.Count = 2
@@ -131,7 +138,13 @@ func validateStart(t *testing.T, c *Cluster) {
 		assert.Equal(t, map[string]string{"my": "annotation"}, d.Spec.Template.Annotations)
 		assert.Contains(t, d.Spec.Template.Labels, "my-label-key")
 		assert.Equal(t, "my-priority-class", d.Spec.Template.Spec.PriorityClassName)
-		assert.Equal(t, 1, len(d.Spec.Template.Spec.Containers))
+		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)
+		}
 	}
 
 	// verify we have exactly the expected number of deployments and not extra
@@ -175,51 +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) {
-		svc1 := corev1.Service{
-			ObjectMeta: metav1.ObjectMeta{
-				Name: "svc1",
-				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",
-				}},
-		}
-		svc2 := corev1.Service{
-			ObjectMeta: metav1.ObjectMeta{Name: "svc2"},
-			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, &svc1, metav1.CreateOptions{})
+	t.Run("initial active daemon", func(t *testing.T) {
+		activeDaemon := "a"
+		err := c.reconcileServices(activeDaemon)
 		assert.NoError(t, err)
-		_, err2 := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Create(clusterInfo.Context, &svc2, metav1.CreateOptions{})
-		assert.NoError(t, err2)
-
-		// Check the the label has been only removed from svc1
-		c.updateServiceSelectors()
-
-		updatedService1, err := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Get(clusterInfo.Context, "svc1", metav1.GetOptions{})
-		assert.NoError(t, err)
-		_, hasDaemonLabel := updatedService1.Spec.Selector[controller.DaemonIDLabel]
-		assert.Equal(t, hasDaemonLabel, false)
-
-		updatedService2, err := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Get(clusterInfo.Context, "svc2", metav1.GetOptions{})
-		assert.NoError(t, err)
-		_, hasDaemonLabel = updatedService2.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())
 
@@ -190,7 +191,6 @@ func (c *Cluster) makeMgrDaemonContainer(mgrConfig *mgrConfig) v1.Container {
 		SecurityContext: controller.PodSecurityContext(),
 		StartupProbe:    controller.GenerateStartupProbeExecDaemon(config.MgrType, mgrConfig.DaemonID),
 		LivenessProbe:   controller.GenerateLivenessProbeExecDaemon(config.MgrType, mgrConfig.DaemonID),
-		ReadinessProbe:  controller.GenerateMgrReadinessProbeExecDaemon(config.MgrType, mgrConfig.DaemonID),
 		WorkingDir:      config.VarLogCephDir,
 	}
 
@@ -208,6 +208,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 +255,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 +288,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,6 +352,10 @@ func (c *Cluster) cephMgrOrchestratorModuleEnvs() []v1.EnvVar {
 	return envVars
 }
 
-func (c *Cluster) selectorLabels() map[string]string {
-	return controller.AppLabels(AppName, c.clusterInfo.Namespace)
+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) {

pkg/operator/ceph/controller/spec.go

@@ -52,15 +52,13 @@ const (
 	daemonSocketsSubPath                    = "/exporter"
 	logCollector                            = "log-collector"
 	DaemonIDLabel                           = "ceph_daemon_id"
-	DaemonTypeLabel                         = "ceph_daemon_type"
+	daemonTypeLabel                         = "ceph_daemon_type"
 	ExternalMgrAppName                      = "rook-ceph-mgr-external"
 	ExternalCephExporterName                = "rook-ceph-exporter-external"
 	ServiceExternalMetricName               = "http-external-metrics"
 	CephUserID                              = 167
 	livenessProbeTimeoutSeconds       int32 = 5
 	livenessProbeInitialDelaySeconds  int32 = 10
-	readinessProbePeriodSeconds       int32 = 5
-	readinessProbeInitialDelaySeconds int32 = 5
 	startupProbeFailuresDaemonDefault int32 = 6 // multiply by 10 = effective startup timeout
 	// The OSD requires a long timeout in case the OSD is taking extra time to
 	// scrub data during startup. We don't want the probe to disrupt the OSD update
@@ -425,7 +423,7 @@ func CephDaemonAppLabels(appName, namespace, daemonType, daemonID, parentName, r
 
 	// New labels cannot be applied to match selectors during upgrade
 	if includeNewLabels {
-		labels[DaemonTypeLabel] = daemonType
+		labels[daemonTypeLabel] = daemonType
 		k8sutil.AddRecommendedLabels(labels, "ceph-"+daemonType, parentName, resourceKind, daemonID)
 	}
 	labels[DaemonIDLabel] = daemonID
@@ -591,81 +589,44 @@ func StoredLogAndCrashVolumeMount(varLogCephDir, varLibCephCrashDir string) []v1
 	}
 }
 
-// Generate an exec ProbeHandler
-func GenProbeHandler(command []string) v1.ProbeHandler {
-	return v1.ProbeHandler{
-		Exec: &v1.ExecAction{
-			Command: command,
-		},
-	}
-}
-
 // GenerateLivenessProbeExecDaemon generates a liveness probe that makes sure a daemon has a socket,
 // that it can be called, and that it returns 0
 func GenerateLivenessProbeExecDaemon(daemonType, daemonID string) *v1.Probe {
-	// Run with env -i to clean env variables in the exec context
-	// This avoids conflict with the CEPH_ARGS env
-	//
-	// Example:
-	// env -i sh -c "ceph --admin-daemon /run/ceph/ceph-osd.0.asok status"
-	//
-	// Ceph gives pretty un-diagnostic error message when `ceph status` or `ceph mon_status` command fails.
-	// Add a clear message after Ceph's to help.
-	// ref: https://github.com/rook/rook/issues/9846
 	confDaemon := getDaemonConfig(daemonType, daemonID)
-	command := []string{
-		"env",
-		"-i",
-		"sh",
-		"-c",
-		fmt.Sprintf(`outp="$(ceph --admin-daemon %s %s 2>&1)"
+
+	return &v1.Probe{
+		ProbeHandler: v1.ProbeHandler{
+			Exec: &v1.ExecAction{
+				// Run with env -i to clean env variables in the exec context
+				// This avoids conflict with the CEPH_ARGS env
+				//
+				// Example:
+				// env -i sh -c "ceph --admin-daemon /run/ceph/ceph-osd.0.asok status"
+				//
+				// Ceph gives pretty un-diagnostic error message when `ceph status` or `ceph mon_status` command fails.
+				// Add a clear message after Ceph's to help.
+				// ref: https://github.com/rook/rook/issues/9846
+				Command: []string{
+					"env",
+					"-i",
+					"sh",
+					"-c",
+					fmt.Sprintf(`outp="$(ceph --admin-daemon %s %s 2>&1)"
 rc=$?
 if [ $rc -ne 0 ]; then
   echo "ceph daemon health check failed with the following output:"
   echo "$outp" | sed -e 's/^/> /g'
   exit $rc
 fi`,
-			confDaemon.buildSocketPath(), confDaemon.buildAdminSocketCommand()),
-	}
-
-	return &v1.Probe{
-		ProbeHandler:        GenProbeHandler(command),
+						confDaemon.buildSocketPath(), confDaemon.buildAdminSocketCommand()),
+				},
+			},
+		},
 		InitialDelaySeconds: livenessProbeInitialDelaySeconds,
 		TimeoutSeconds:      livenessProbeTimeoutSeconds,
 	}
 }
 
-// GenerateMgrReadinessProbeExecDaemon generates a readiness probe that makes sure
-// a mgr daemon is ready to process traffic
-func GenerateMgrReadinessProbeExecDaemon(daemonType, daemonID string) *v1.Probe {
-	// Run with env -i to clean env variables in the exec context
-	// This avoids conflict with the CEPH_ARGS env
-	//
-	// Example:
-	// env -i sh -c "ceph --admin-daemon /run/ceph/ceph-mgr.a.asok mgr_status"
-	//
-	confDaemon := getDaemonConfig(daemonType, daemonID)
-	command := []string{
-		"env",
-		"-i",
-		"sh",
-		"-c",
-		fmt.Sprintf(`ceph --admin-daemon %s mgr_status &>/dev/null
-rc=$?
-if [ $rc -ne 0 ]; then
-  echo "Ceph standby manager does not process requests."
-  exit $rc
-fi`,
-			confDaemon.buildSocketPath()),
-	}
-
-	return &v1.Probe{
-		ProbeHandler:        GenProbeHandler(command),
-		InitialDelaySeconds: readinessProbeInitialDelaySeconds,
-		PeriodSeconds:       readinessProbePeriodSeconds,
-	}
-}
-
 // GenerateStartupProbeExecDaemon generates a startup probe that makes sure a daemon has a socket,
 // that it can be called, and that it returns 0
 func GenerateStartupProbeExecDaemon(daemonType, daemonID string) *v1.Probe {

pkg/operator/ceph/controller/spec_test.go

@@ -171,24 +171,6 @@ fi`}
 	assert.Equal(t, livenessProbeTimeoutSeconds, probe.TimeoutSeconds)
 }
 
-func TestGenerateReadinessProbeExecDaemon(t *testing.T) {
-	daemonID := "a"
-	probe := GenerateMgrReadinessProbeExecDaemon(config.MgrType, daemonID)
-	expectedCommand := []string{"env",
-		"-i",
-		"sh",
-		"-c",
-		`ceph --admin-daemon /run/ceph/ceph-mgr.a.asok mgr_status &>/dev/null
-rc=$?
-if [ $rc -ne 0 ]; then
-  echo "Ceph standby manager does not process requests."
-  exit $rc
-fi`}
-
-	assert.Equal(t, expectedCommand, probe.ProbeHandler.Exec.Command)
-	assert.Equal(t, readinessProbeInitialDelaySeconds, probe.InitialDelaySeconds)
-}
-
 func TestDaemonFlags(t *testing.T) {
 	testcases := []struct {
 		label       string

pkg/operator/k8sutil/deployment.go

@@ -144,25 +144,6 @@ func WaitForDeploymentToStart(ctx context.Context, clusterdContext *clusterd.Con
 	return fmt.Errorf("gave up waiting for deployment %q to update", deployment.Name)
 }
 
-func WaitForPodsWithLabelToRun(ctx context.Context, clientset kubernetes.Interface, namespace, label string) error {
-	// wait for the pods to run for a maximum of 300 sec
-	sleepTime := 5
-	attempts := 60
-	for i := 0; i < attempts; i++ {
-		logger.Infof("waiting for all pods with label %s to be in running state", label)
-		time.Sleep(time.Duration(sleepTime) * time.Second)
-		allRunning, err := PodsWithLabelAreAllRunning(ctx, clientset, namespace, label)
-		if err != nil {
-			return errors.Wrapf(err, "failed to query pods with label %s to be in running state", label)
-		}
-		if allRunning {
-			logger.Infof("all pods with label %s are running", label)
-			return nil
-		}
-	}
-	return fmt.Errorf("gave up waiting for all pods with label %s to run", label)
-}
-
 // DeploymentNames returns a list of the names of deployments in the deployment list
 func DeploymentNames(deployments *appsv1.DeploymentList) (names []string) {
 	for _, d := range deployments.Items {

pkg/operator/k8sutil/pod.go

@@ -226,8 +226,7 @@ func podStatusWithLabel(ctx context.Context, clientset kubernetes.Interface, nam
 	running := 0
 	notRunning := 0
 	for _, pod := range pods.Items {
-		// The pod must be in running state and not deleted
-		if pod.Status.Phase == v1.PodRunning && pod.DeletionTimestamp.IsZero() {
+		if pod.Status.Phase == v1.PodRunning {
 			running++
 		} else {
 			notRunning++