diff --git a/Documentation/CRDs/Cluster/ceph-cluster-crd.md b/Documentation/CRDs/Cluster/ceph-cluster-crd.md index 2972b317b..acebd3f8d 100755 --- a/Documentation/CRDs/Cluster/ceph-cluster-crd.md +++ b/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-`: 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 diff --git a/cmd/rook/ceph/ceph.go b/cmd/rook/ceph/ceph.go index 6de44e3ca..31889758a 100644 --- a/cmd/rook/ceph/ceph.go +++ b/cmd/rook/ceph/ceph.go @@ -58,6 +58,7 @@ func init() { Cmd.AddCommand(cleanUpCmd, operatorCmd, osdCmd, + mgrCmd, configCmd) } diff --git a/cmd/rook/ceph/mgr.go b/cmd/rook/ceph/mgr.go new file mode 100644 index 000000000..99e774af2 --- /dev/null +++ b/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) + } +} diff --git a/deploy/charts/rook-ceph-cluster/values.yaml b/deploy/charts/rook-ceph-cluster/values.yaml index ae9fdb420..9c8195dd4 100644 --- a/deploy/charts/rook-ceph-cluster/values.yaml +++ b/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" diff --git a/deploy/examples/cluster.yaml b/deploy/examples/cluster.yaml index af5ea9c50..9bd50ec97 100644 --- a/deploy/examples/cluster.yaml +++ b/deploy/examples/cluster.yaml @@ -220,6 +220,7 @@ spec: # osd-ssd: # osd-nvme: # prepareosd: + # mgr-sidecar: # crashcollector: # logcollector: # cleanup: diff --git a/deploy/examples/monitoring/service-monitor.yaml b/deploy/examples/monitoring/service-monitor.yaml index 930d0dc34..77e82c959 100644 --- a/deploy/examples/monitoring/service-monitor.yaml +++ b/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 diff --git a/pkg/apis/ceph.rook.io/v1/resources.go b/pkg/apis/ceph.rook.io/v1/resources.go index 5ffbc80a2..1f621333e 100644 --- a/pkg/apis/ceph.rook.io/v1/resources.go +++ b/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] diff --git a/pkg/operator/ceph/cluster/cluster_external.go b/pkg/operator/ceph/cluster/cluster_external.go index a86c8f996..7e747138b 100644 --- a/pkg/operator/ceph/cluster/cluster_external.go +++ b/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 { diff --git a/pkg/operator/ceph/cluster/mgr/dashboard.go b/pkg/operator/ceph/cluster/mgr/dashboard.go index 8d2497a7a..7abf6c25a 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() error { - dashboardService, err := c.makeDashboardService(AppName) +func (c *Cluster) configureDashboardService(activeDaemon string) error { + dashboardService, err := c.makeDashboardService(AppName, activeDaemon) 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 8a568758f..965be9a6e 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() + 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{}) diff --git a/pkg/operator/ceph/cluster/mgr/mgr.go b/pkg/operator/ceph/cluster/mgr/mgr.go index 80a998917..efd6d2a60 100644 --- a/pkg/operator/ceph/cluster/mgr/mgr.go +++ b/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) diff --git a/pkg/operator/ceph/cluster/mgr/mgr_test.go b/pkg/operator/ceph/cluster/mgr/mgr_test.go index 66fa474ea..ad050cf41 100644 --- a/pkg/operator/ceph/cluster/mgr/mgr_test.go +++ b/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) { diff --git a/pkg/operator/ceph/cluster/mgr/spec.go b/pkg/operator/ceph/cluster/mgr/spec.go index 7a11c03ae..0c04000f8 100644 --- a/pkg/operator/ceph/cluster/mgr/spec.go +++ b/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 } diff --git a/pkg/operator/ceph/cluster/mgr/spec_test.go b/pkg/operator/ceph/cluster/mgr/spec_test.go index 11acf3367..612e7f639 100644 --- a/pkg/operator/ceph/cluster/mgr/spec_test.go +++ b/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) {