diff --git a/Documentation/CRDs/Cluster/ceph-cluster-crd.md b/Documentation/CRDs/Cluster/ceph-cluster-crd.md index d910acf0b..caa72913e 100755 --- a/Documentation/CRDs/Cluster/ceph-cluster-crd.md +++ b/Documentation/CRDs/Cluster/ceph-cluster-crd.md @@ -59,9 +59,10 @@ 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, 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. + If there are two managers, it is important for all mgr services point to the active mgr and not the standby mgr. Rook automatically + updates the label `mgr_role` on the mgr pods to be either `active` or `standby`. Therefore, services need just to add the label + `mgr_role=active` to their selector to point to the active mgr. This applies to all services that rely on the ceph mgr such as + 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 diff --git a/cmd/rook/ceph/mgr.go b/cmd/rook/ceph/mgr.go index 99e774af2..334c2ccd9 100644 --- a/cmd/rook/ceph/mgr.go +++ b/cmd/rook/ceph/mgr.go @@ -96,12 +96,13 @@ func runMgrSidecar(cmd *cobra.Command, args []string) error { clusterInfo.CephVersion = *version m := mgr.New(context, &clusterInfo, clusterSpec, "") + prevActiveMgr := "unknown" for { - err := m.ReconcileActiveMgrServices(daemonName) + prevActiveMgr, err = m.UpdateActiveMgrLabel(daemonName, prevActiveMgr) if err != nil { logger.Errorf("failed to reconcile services. %v", err) } else { - logger.Infof("successfully reconciled services. checking again in %ds", (int)(interval.Seconds())) + logger.Infof("successfully checked mgr_role label. checking again in %ds", (int)(interval.Seconds())) } time.Sleep(interval) } diff --git a/deploy/charts/library/templates/_cluster-role.tpl b/deploy/charts/library/templates/_cluster-role.tpl index 13de8de58..9006e2335 100644 --- a/deploy/charts/library/templates/_cluster-role.tpl +++ b/deploy/charts/library/templates/_cluster-role.tpl @@ -99,9 +99,12 @@ rules: resources: - deployments/scale - deployments + - pods verbs: - patch - delete + - get + - update - apiGroups: - '' resources: diff --git a/deploy/examples/common.yaml b/deploy/examples/common.yaml index 6254dfc7b..fea047050 100644 --- a/deploy/examples/common.yaml +++ b/deploy/examples/common.yaml @@ -793,9 +793,12 @@ rules: resources: - deployments/scale - deployments + - pods verbs: - patch - delete + - get + - update - apiGroups: - '' resources: diff --git a/deploy/examples/dashboard-external-http.yaml b/deploy/examples/dashboard-external-http.yaml index 27fc59866..82338367f 100644 --- a/deploy/examples/dashboard-external-http.yaml +++ b/deploy/examples/dashboard-external-http.yaml @@ -14,6 +14,7 @@ spec: targetPort: 7000 selector: app: rook-ceph-mgr + mgr_role: active rook_cluster: rook-ceph sessionAffinity: None type: NodePort diff --git a/deploy/examples/dashboard-external-https.yaml b/deploy/examples/dashboard-external-https.yaml index 9d30fadca..6a18dd30c 100644 --- a/deploy/examples/dashboard-external-https.yaml +++ b/deploy/examples/dashboard-external-https.yaml @@ -14,6 +14,7 @@ spec: targetPort: 8443 selector: app: rook-ceph-mgr + mgr_role: active rook_cluster: rook-ceph sessionAffinity: None type: NodePort diff --git a/deploy/examples/dashboard-loadbalancer.yaml b/deploy/examples/dashboard-loadbalancer.yaml index 98c0b9075..19fe8eb5c 100644 --- a/deploy/examples/dashboard-loadbalancer.yaml +++ b/deploy/examples/dashboard-loadbalancer.yaml @@ -14,6 +14,7 @@ spec: targetPort: 8443 selector: app: rook-ceph-mgr + mgr_role: active rook_cluster: rook-ceph sessionAffinity: None type: LoadBalancer diff --git a/deploy/examples/monitoring/service-monitor.yaml b/deploy/examples/monitoring/service-monitor.yaml index 77e82c959..44a7a4f7b 100644 --- a/deploy/examples/monitoring/service-monitor.yaml +++ b/deploy/examples/monitoring/service-monitor.yaml @@ -11,9 +11,9 @@ spec: - rook-ceph selector: matchLabels: + mgr_role: active app: rook-ceph-mgr rook_cluster: rook-ceph - ceph_daemon_id: a endpoints: - port: http-metrics path: /metrics diff --git a/pkg/operator/ceph/cluster/cluster_external.go b/pkg/operator/ceph/cluster/cluster_external.go index e5b5c2a93..c21e7f191 100644 --- a/pkg/operator/ceph/cluster/cluster_external.go +++ b/pkg/operator/ceph/cluster/cluster_external.go @@ -211,7 +211,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 } @@ -231,7 +231,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 { diff --git a/pkg/operator/ceph/cluster/mgr/dashboard.go b/pkg/operator/ceph/cluster/mgr/dashboard.go index 2fa57dd6d..cd0478a85 100644 --- a/pkg/operator/ceph/cluster/mgr/dashboard.go +++ b/pkg/operator/ceph/cluster/mgr/dashboard.go @@ -54,8 +54,8 @@ var ( dashboardInitWaitTime = 5 * time.Second ) -func (c *Cluster) configureDashboardService(activeDaemon string) error { - dashboardService, err := c.makeDashboardService(AppName, activeDaemon) +func (c *Cluster) configureDashboardService() error { + dashboardService, err := c.makeDashboardService(AppName) if err != nil { return err } diff --git a/pkg/operator/ceph/cluster/mgr/dashboard_test.go b/pkg/operator/ceph/cluster/mgr/dashboard_test.go index 6c86fa602..8b7f14a02 100644 --- a/pkg/operator/ceph/cluster/mgr/dashboard_test.go +++ b/pkg/operator/ceph/cluster/mgr/dashboard_test.go @@ -131,7 +131,7 @@ func TestStartSecureDashboard(t *testing.T) { } dashboardInitWaitTime = 0 - err := c.configureDashboardService("a") + err := c.configureDashboardService() 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("a") + err = c.configureDashboardService() 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("a") + err = c.configureDashboardService() 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("a") + err = c.configureDashboardService() assert.Nil(t, err) svc, err = c.context.Clientset.CoreV1().Services(clusterInfo.Namespace).Get(ctx, "rook-ceph-mgr-dashboard", metav1.GetOptions{}) diff --git a/pkg/operator/ceph/cluster/mgr/mgr.go b/pkg/operator/ceph/cluster/mgr/mgr.go index 5b0d67ee9..1efb2f005 100644 --- a/pkg/operator/ceph/cluster/mgr/mgr.go +++ b/pkg/operator/ceph/cluster/mgr/mgr.go @@ -51,6 +51,7 @@ const ( PgautoscalerModuleName = "pg_autoscaler" balancerModuleName = "balancer" balancerModuleMode = "upmap" + mgrRoleLabelName = "mgr_role" monitoringPath = "/etc/ceph-monitoring/" serviceMonitorFile = "service-monitor.yaml" // minimum amount of memory in MB to run the pod @@ -178,18 +179,7 @@ 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") @@ -198,10 +188,9 @@ func (c *Cluster) Start() error { // delete MGR PDB as the count is less than 2 c.deleteMgrPDB() } - if activeMgr != "" { - if err := c.reconcileServices(activeMgr); err != nil { - return errors.Wrap(err, "failed to enable mgr services") - } + + if err := c.reconcileServices(); err != nil { + return errors.Wrap(err, "failed to enable mgr services") } // configure the mgr modules @@ -222,35 +211,52 @@ func (c *Cluster) removeExtraMgrs(daemonIDs []string) { } } -// 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{}) +// UpdateActiveMgrLabel updates the mgr_role label value to either +// active or standby depending on the status of the ceph mgr running +// in the pod +func (c *Cluster) UpdateActiveMgrLabel(daemonNameToUpdate string, prevActiveMgr string) (string, error) { + + logger.Infof("Checking mgr_role label value of daemon %s (prev active mgr was %s)", daemonNameToUpdate, prevActiveMgr) + currActiveMgr, err := c.getActiveMgr() + if err != nil || currActiveMgr == "" { + logger.Infof("cannot update active mgr, no active mgr found. err=%v", err) + return "", err + } else if prevActiveMgr == currActiveMgr { + logger.Infof("active mgr is still the same (%s). No need to update mgr_role label on daemon %s.", currActiveMgr, daemonNameToUpdate) + return currActiveMgr, err + } + + pods, err := c.context.Clientset.CoreV1().Pods(c.clusterInfo.Namespace).List(c.clusterInfo.Context, metav1.ListOptions{ + LabelSelector: fmt.Sprintf("mgr=%s", daemonNameToUpdate), + }) 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("cannot get pod for mgr daemon %s", daemonNameToUpdate) + return "", err // force mrg_role update in the next call + } + + for i, pod := range pods.Items { + + labels := pod.GetLabels() + cephDaemonId := labels[controller.DaemonIDLabel] + newMgrRole := "standby" + if currActiveMgr == cephDaemonId { + newMgrRole = "active" + } + + currMgrRole, mgrHasLabel := labels[mgrRoleLabelName] + if !mgrHasLabel || currMgrRole != newMgrRole { + + logger.Infof("updating mgr_role label value of daemon %s to '%s'. New active mgr is %s.", daemonNameToUpdate, newMgrRole, currActiveMgr) + labels[mgrRoleLabelName] = newMgrRole + pod.SetLabels(labels) + _, err = c.context.Clientset.CoreV1().Pods(c.clusterInfo.Namespace).Update(c.clusterInfo.Context, &pods.Items[i], metav1.UpdateOptions{}) + if err != nil { + logger.Infof("cannot update the active mgr pod %q. err=%v", pods.Items[i].Name, err) + } } - logger.Infof("mgr service currently set to %q, checking if need to update to %q", currentDaemon, daemonNameToUpdate) } - 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) + return currActiveMgr, c.reconcileServices() } func (c *Cluster) getActiveMgr() (string, error) { @@ -261,16 +267,15 @@ func (c *Cluster) getActiveMgr() (string, error) { 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) +// reconcile the services, +func (c *Cluster) reconcileServices() error { - if err := c.configureDashboardService(activeDaemon); err != nil { + if err := c.configureDashboardService(); err != nil { return errors.Wrap(err, "failed to configure dashboard svc") } // create the metrics service - service, err := c.MakeMetricsService(AppName, activeDaemon, serviceMetricName) + service, err := c.MakeMetricsService(AppName, serviceMetricName) if err != nil { return err } @@ -280,41 +285,56 @@ func (c *Cluster) reconcileServices(activeDaemon string) error { // enable monitoring if `monitoring: enabled: true` if c.spec.Monitoring.Enabled { - if err := c.EnableServiceMonitor(activeDaemon); err != nil { + if err := c.EnableServiceMonitor(); err != nil { return errors.Wrap(err, "failed to enable service monitor") } } - return c.updateServiceSelectors(activeDaemon) + c.updateServiceSelectors() + return nil } -// 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 { +// For the upgrade scenario: we remove any selector DaemonIDLabel from the all +// the services since new mgr HA doesn't rely on this label anymore and we add +// the new "mgr_role=active" instead. +func (c *Cluster) updateServiceSelectors() { 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 { - return errors.Wrap(err, "failed to query mgr services to update") + logger.Errorf("failed to query mgr services to update labels: %v", err) + return } 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 } - // 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) + + updateService := false + + // Check if the service has a DaemonIDLabel (legacy mgr HA implementation) and remove it + _, hasDaemonLabel := service.Spec.Selector[controller.DaemonIDLabel] + if hasDaemonLabel { + logger.Infof("removing %s selector label on mgr service %q", controller.DaemonIDLabel, service.Name) + delete(service.Spec.Selector, controller.DaemonIDLabel) + updateService = true + } + + // In case the service doesn't have the new mgr_role DaemonIDLabel we add it + _, hasMgrRoleLabel := service.Spec.Selector[mgrRoleLabelName] + if !hasMgrRoleLabel { + logger.Infof("adding %s selector label on mgr service %q", mgrRoleLabelName, service.Name) + service.Spec.Selector[mgrRoleLabelName] = "active" + updateService = true + } + + if updateService { + 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) + } } } - return nil } func (c *Cluster) configureModules(daemonIDs []string) { @@ -480,7 +500,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(activeDaemon string) error { +func (c *Cluster) EnableServiceMonitor() error { serviceMonitor, err := k8sutil.GetServiceMonitor(path.Join(monitoringPath, serviceMonitorFile)) if err != nil { return errors.Wrap(err, "service monitor could not be enabled") @@ -497,7 +517,6 @@ func (c *Cluster) EnableServiceMonitor(activeDaemon string) 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(activeDaemon) applyMonitoringLabels(c, serviceMonitor) diff --git a/pkg/operator/ceph/cluster/mgr/mgr_test.go b/pkg/operator/ceph/cluster/mgr/mgr_test.go index 051c44575..51e1e0c63 100644 --- a/pkg/operator/ceph/cluster/mgr/mgr_test.go +++ b/pkg/operator/ceph/cluster/mgr/mgr_test.go @@ -31,6 +31,7 @@ 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" + "github.com/rook/rook/pkg/operator/k8sutil" 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" @@ -45,10 +46,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client/fake" ) -func TestStartMgr(t *testing.T) { - var deploymentsUpdated *[]*apps.Deployment - updateDeploymentAndWait, deploymentsUpdated = testopk8s.UpdateDeploymentAndWaitStub() - +func createNewCluster(t *testing.T) *Cluster { executor := &exectest.MockExecutor{ MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) { logger.Infof("Execute: %s %v", command, args) @@ -92,8 +90,17 @@ func TestStartMgr(t *testing.T) { c := New(ctx, clusterInfo, clusterSpec, "myversion") defer os.RemoveAll(c.spec.DataDirHostPath) + return c +} + +func TestStartMgr(t *testing.T) { + var deploymentsUpdated *[]*apps.Deployment + updateDeploymentAndWait, deploymentsUpdated = testopk8s.UpdateDeploymentAndWaitStub() + + c := createNewCluster(t) + // start a basic service - err = c.Start() + err := c.Start() assert.NoError(t, err) validateStart(t, c) assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated)) @@ -176,6 +183,49 @@ func validateServices(t *testing.T, c *Cluster) { } } +func TestActiveMgrLabels(t *testing.T) { + var deploymentsUpdated *[]*apps.Deployment + updateDeploymentAndWait, deploymentsUpdated = testopk8s.UpdateDeploymentAndWaitStub() + + c := createNewCluster(t) + err := c.Start() + assert.NoError(t, err) + validateStart(t, c) + + mgrNames := []string{"a", "b"} + for i := 0; i < c.spec.Mgr.Count; i++ { + // Simulate the Mgr pod having been created + daemonName := mgrNames[i] + mgrPod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{ + Name: fmt.Sprintf("rook-ceph-mgr-%s", daemonName), + Labels: map[string]string{k8sutil.AppAttr: "rook-ceph-mgr", + "instance": daemonName, + "ceph_daemon_id": daemonName, + "mgr": daemonName, + "rook_cluster": "ns"}}} + _, err = c.context.Clientset.CoreV1().Pods(c.clusterInfo.Namespace).Create(c.clusterInfo.Context, mgrPod, metav1.CreateOptions{}) + assert.NoError(t, err) + } + + for i := 0; i < c.spec.Mgr.Count; i++ { + daemonName := mgrNames[i] + _, err := c.context.Clientset.AppsV1().Deployments(c.clusterInfo.Namespace).Get(context.TODO(), fmt.Sprintf("rook-ceph-mgr-%s", daemonName), metav1.GetOptions{}) + assert.NoError(t, err) + _, err = c.UpdateActiveMgrLabel(daemonName, "") + assert.NoError(t, err) + pods, err := c.context.Clientset.CoreV1().Pods(c.clusterInfo.Namespace).List(c.clusterInfo.Context, metav1.ListOptions{LabelSelector: "app=rook-ceph-mgr"}) + assert.NoError(t, err) + assert.Contains(t, pods.Items[i].Labels, "mgr_role") + if daemonName == "a" { + assert.Equal(t, pods.Items[i].Labels["mgr_role"], "active") + } else if daemonName == "b" { + assert.Equal(t, pods.Items[i].Labels["mgr_role"], "standby") + } + } + + testopk8s.ClearDeploymentsUpdated(deploymentsUpdated) +} + func TestUpdateServiceSelectors(t *testing.T) { clientset := testop.New(t, 3) ctx := &clusterd.Context{Clientset: clientset} @@ -188,116 +238,54 @@ func TestUpdateServiceSelectors(t *testing.T) { } c := &Cluster{spec: spec, context: ctx, clusterInfo: clusterInfo} - t.Run("initial active daemon", func(t *testing.T) { - activeDaemon := "a" - err := c.reconcileServices(activeDaemon) - assert.NoError(t, err) - 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]) + // 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", + }}, } - } - assert.Equal(t, expectedUpdated, updated) - assert.Equal(t, expectedSkipped, skipped) -} + 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{}) + assert.NoError(t, err) + _, err2 := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Create(clusterInfo.Context, &svc2, metav1.CreateOptions{}) + assert.NoError(t, err2) -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} + // Check the the label has been only removed from svc1 + c.updateServiceSelectors() - // 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") + // Make sure we remove controller.DaemonIDLabel label from all services tagged as 'app: rook-ceph-mgr' + // and we add the selector label "mgr_role" set to "active" + updatedService1, err := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Get(clusterInfo.Context, "svc1", metav1.GetOptions{}) + assert.NoError(t, err) + assert.NotContains(t, updatedService1.Spec.Selector, controller.DaemonIDLabel) + assert.Contains(t, updatedService1.Spec.Selector, "mgr_role") + assert.Equal(t, updatedService1.Spec.Selector["mgr_role"], "active") - // 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) + // Make sure services not tagged as 'app: rook-ceph-mgr' are not modified + updatedService2, err := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Get(clusterInfo.Context, "svc2", metav1.GetOptions{}) + assert.NoError(t, err) + assert.Contains(t, updatedService2.Spec.Selector, controller.DaemonIDLabel) + assert.NotContains(t, updatedService2.Spec.Selector, "mgr_role") + }) } func TestConfigureModules(t *testing.T) { diff --git a/pkg/operator/ceph/cluster/mgr/spec.go b/pkg/operator/ceph/cluster/mgr/spec.go index a73997c52..de575c3f3 100644 --- a/pkg/operator/ceph/cluster/mgr/spec.go +++ b/pkg/operator/ceph/cluster/mgr/spec.go @@ -255,8 +255,10 @@ func (c *Cluster) makeCmdProxySidecarContainer(mgrConfig *mgrConfig) v1.Containe } // MakeMetricsService generates the Kubernetes service object for the monitoring service -func (c *Cluster) MakeMetricsService(name, activeDaemon, servicePortMetricName string) (*v1.Service, error) { - labels := c.selectorLabels(activeDaemon) +func (c *Cluster) MakeMetricsService(name, servicePortMetricName string) (*v1.Service, error) { + + labels := controller.AppLabels(AppName, c.clusterInfo.Namespace) + selectorLabels := c.buildSelectorLabels(labels) svc := &v1.Service{ ObjectMeta: metav1.ObjectMeta{ @@ -278,7 +280,7 @@ func (c *Cluster) MakeMetricsService(name, activeDaemon, servicePortMetricName s // If the cluster is external we don't need to add the selector if name != controller.ExternalMgrAppName { - svc.Spec.Selector = labels + svc.Spec.Selector = selectorLabels } err := c.clusterInfo.OwnerInfo.SetControllerReference(svc) @@ -288,8 +290,10 @@ func (c *Cluster) MakeMetricsService(name, activeDaemon, servicePortMetricName s return svc, nil } -func (c *Cluster) makeDashboardService(name, activeDaemon string) (*v1.Service, error) { - labels := c.selectorLabels(activeDaemon) +func (c *Cluster) makeDashboardService(name string) (*v1.Service, error) { + + labels := controller.AppLabels(AppName, c.clusterInfo.Namespace) + selectorLabels := c.buildSelectorLabels(labels) portName := "https-dashboard" if !c.spec.Dashboard.SSL { @@ -302,7 +306,7 @@ func (c *Cluster) makeDashboardService(name, activeDaemon string) (*v1.Service, Labels: labels, }, Spec: v1.ServiceSpec{ - Selector: labels, + Selector: selectorLabels, Type: v1.ServiceTypeClusterIP, Ports: []v1.ServicePort{ { @@ -352,10 +356,11 @@ func (c *Cluster) cephMgrOrchestratorModuleEnvs() []v1.EnvVar { return envVars } -func (c *Cluster) selectorLabels(activeDaemon string) map[string]string { - labels := controller.AppLabels(AppName, c.clusterInfo.Namespace) - if activeDaemon != "" { - labels[controller.DaemonIDLabel] = activeDaemon +func (c *Cluster) buildSelectorLabels(labels map[string]string) map[string]string { + selectorLabels := make(map[string]string) + for k, v := range labels { + selectorLabels[k] = v } - return labels + selectorLabels["mgr_role"] = "active" + return selectorLabels } diff --git a/pkg/operator/ceph/cluster/mgr/spec_test.go b/pkg/operator/ceph/cluster/mgr/spec_test.go index 612e7f639..84e5ab6a7 100644 --- a/pkg/operator/ceph/cluster/mgr/spec_test.go +++ b/pkg/operator/ceph/cluster/mgr/spec_test.go @@ -23,7 +23,6 @@ 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" @@ -113,14 +112,17 @@ func TestServiceSpec(t *testing.T) { clusterSpec := cephv1.ClusterSpec{} c := New(&clusterd.Context{Clientset: clientset}, clusterInfo, clusterSpec, "myversion") - s, err := c.MakeMetricsService("rook-mgr", "foo", serviceMetricName) + s, err := c.MakeMetricsService("rook-mgr", 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, 3, len(s.Labels)) + assert.Equal(t, 2, len(s.Labels)) + // Check service selector assert.Equal(t, 3, len(s.Spec.Selector)) - assert.Equal(t, "foo", s.Spec.Selector[controller.DaemonIDLabel]) + assert.Equal(t, s.Spec.Selector["app"], "rook-ceph-mgr") + assert.Equal(t, s.Spec.Selector["mgr_role"], "active") + assert.Equal(t, s.Spec.Selector["rook_cluster"], "ns") } func TestHostNetwork(t *testing.T) {