mgr/cephadm: adding HA support for mgmt-gateway and ouath2-proxy

adding HA support for mgmt-gateway and ouath2-proxy. In addition, logic
is add to prevent unnecessary daemons restart during mgr failover.
Previously, without the management gateway, some daemons, such as
Prometheus, had an explicit dependency on the manager because we
needed to point to the active manager. With the management gateway,
this explicit dependency is no longer necessary, as it automatically
handles routing requests to the active manager.

Signed-off-by: Redouane Kachach <[email protected]>

Author: rkachach

src/pybind/mgr/cephadm/module.py

@@ -819,30 +819,33 @@ def _get_security_config(self) -> Tuple[bool, bool, bool]:
         security_enabled = self.secure_monitoring_stack or mgmt_gw_enabled
         return security_enabled, mgmt_gw_enabled, oauth2_proxy_enabled
 
-    def get_mgmt_gw_internal_endpoint(self) -> Optional[str]:
+    def _get_mgmt_gw_endpoint(self, is_internal: bool) -> Optional[str]:
         mgmt_gw_daemons = self.cache.get_daemons_by_service('mgmt-gateway')
         if not mgmt_gw_daemons:
             return None
 
         dd = mgmt_gw_daemons[0]
         assert dd.hostname is not None
-        mgmt_gw_addr = self.get_fqdn(dd.hostname)
-        mgmt_gw_internal_endpoint = build_url(scheme='https', host=mgmt_gw_addr, port=MgmtGatewayService.INTERNAL_SERVICE_PORT)
-        return f'{mgmt_gw_internal_endpoint}/internal'
+        mgmt_gw_spec = cast(MgmtGatewaySpec, self.spec_store['mgmt-gateway'].spec)
+        mgmt_gw_addr = mgmt_gw_spec.virtual_ip if mgmt_gw_spec.virtual_ip is not None else self.get_fqdn(dd.hostname)
 
-    def get_mgmt_gw_external_endpoint(self) -> Optional[str]:
-        mgmt_gw_daemons = self.cache.get_daemons_by_service('mgmt-gateway')
-        if not mgmt_gw_daemons:
-            return None
+        if is_internal:
+            mgmt_gw_port: Optional[int] = MgmtGatewayService.INTERNAL_SERVICE_PORT
+            protocol = 'https'
+            endpoint_suffix = '/internal'
+        else:
+            mgmt_gw_port = dd.ports[0] if dd.ports else None
+            protocol = 'http' if mgmt_gw_spec.disable_https else 'https'
+            endpoint_suffix = ''
 
-        dd = mgmt_gw_daemons[0]
-        assert dd.hostname is not None
-        mgmt_gw_port = dd.ports[0] if dd.ports else None
-        mgmt_gw_addr = self.get_fqdn(dd.hostname)
-        mgmt_gw_spec = cast(MgmtGatewaySpec, self.spec_store['mgmt-gateway'].spec)
-        protocol = 'http' if mgmt_gw_spec.disable_https else 'https'
-        mgmt_gw_external_endpoint = build_url(scheme=protocol, host=mgmt_gw_addr, port=mgmt_gw_port)
-        return mgmt_gw_external_endpoint
+        mgmt_gw_endpoint = build_url(scheme=protocol, host=mgmt_gw_addr, port=mgmt_gw_port)
+        return f'{mgmt_gw_endpoint}{endpoint_suffix}'
+
+    def get_mgmt_gw_internal_endpoint(self) -> Optional[str]:
+        return self._get_mgmt_gw_endpoint(is_internal=True)
+
+    def get_mgmt_gw_external_endpoint(self) -> Optional[str]:
+        return self._get_mgmt_gw_endpoint(is_internal=False)
 
     def _get_cephadm_binary_path(self) -> str:
         import hashlib
@@ -3001,8 +3004,16 @@ def get_daemon_names(daemons: List[str]) -> List[str]:
                     daemon_names.append(dd.name())
             return daemon_names
 
-        alertmanager_user, alertmanager_password = self._get_alertmanager_credentials()
-        prometheus_user, prometheus_password = self._get_prometheus_credentials()
+        prom_cred_hash = None
+        alertmgr_cred_hash = None
+        security_enabled, mgmt_gw_enabled, _ = self._get_security_config()
+        if security_enabled:
+            alertmanager_user, alertmanager_password = self._get_alertmanager_credentials()
+            prometheus_user, prometheus_password = self._get_prometheus_credentials()
+            if prometheus_user and prometheus_password:
+                prom_cred_hash = f'{utils.md5_hash(prometheus_user + prometheus_password)}'
+            if alertmanager_user and alertmanager_password:
+                alertmgr_cred_hash = f'{utils.md5_hash(alertmanager_user + alertmanager_password)}'
 
         deps = []
         if daemon_type == 'haproxy':
@@ -3049,9 +3060,10 @@ def get_daemon_names(daemons: List[str]) -> List[str]:
             else:
                 deps = [self.get_mgr_ip()]
         elif daemon_type == 'prometheus':
-            # for prometheus we add the active mgr as an explicit dependency,
-            # this way we force a redeploy after a mgr failover
-            deps.append(self.get_active_mgr().name())
+            if not mgmt_gw_enabled:
+                # for prometheus we add the active mgr as an explicit dependency,
+                # this way we force a redeploy after a mgr failover
+                deps.append(self.get_active_mgr().name())
             deps.append(str(self.get_module_option_ex('prometheus', 'server_port', 9283)))
             deps.append(str(self.service_discovery_port))
             # prometheus yaml configuration file (generated by prometheus.yml.j2) contains
@@ -3068,22 +3080,20 @@ def get_daemon_names(daemons: List[str]) -> List[str]:
             deps += [d.name() for d in self.cache.get_daemons_by_service('ceph-exporter')]
             deps += [d.name() for d in self.cache.get_daemons_by_service('mgmt-gateway')]
             deps += [d.name() for d in self.cache.get_daemons_by_service('oauth2-proxy')]
-            security_enabled, _, _ = self._get_security_config()
-            if security_enabled:
-                if prometheus_user and prometheus_password:
-                    deps.append(f'{hash(prometheus_user + prometheus_password)}')
-                if alertmanager_user and alertmanager_password:
-                    deps.append(f'{hash(alertmanager_user + alertmanager_password)}')
+            if prom_cred_hash is not None:
+                deps.append(prom_cred_hash)
+            if alertmgr_cred_hash is not None:
+                deps.append(alertmgr_cred_hash)
         elif daemon_type == 'grafana':
             deps += get_daemon_names(['prometheus', 'loki', 'mgmt-gateway', 'oauth2-proxy'])
-            security_enabled, _, _ = self._get_security_config()
-            if security_enabled and prometheus_user and prometheus_password:
-                deps.append(f'{hash(prometheus_user + prometheus_password)}')
+            if prom_cred_hash is not None:
+                deps.append(prom_cred_hash)
         elif daemon_type == 'alertmanager':
-            deps += get_daemon_names(['mgr', 'alertmanager', 'snmp-gateway', 'mgmt-gateway', 'oauth2-proxy'])
-            security_enabled, _, _ = self._get_security_config()
-            if security_enabled and alertmanager_user and alertmanager_password:
-                deps.append(f'{hash(alertmanager_user + alertmanager_password)}')
+            deps += get_daemon_names(['alertmanager', 'snmp-gateway', 'mgmt-gateway', 'oauth2-proxy'])
+            if not mgmt_gw_enabled:
+                deps += get_daemon_names(['mgr'])
+            if alertmgr_cred_hash is not None:
+                deps.append(alertmgr_cred_hash)
         elif daemon_type == 'promtail':
             deps += get_daemon_names(['loki'])
         elif daemon_type in ['ceph-exporter', 'node-exporter']:
@@ -3097,7 +3107,7 @@ def get_daemon_names(daemons: List[str]) -> List[str]:
         elif daemon_type == 'mgmt-gateway':
             # url_prefix for monitoring daemons depends on the presence of mgmt-gateway
             # while dashboard urls depend on the mgr daemons
-            deps += get_daemon_names(['mgr', 'grafana', 'prometheus', 'alertmanager', 'oauth2-proxy'])
+            deps += get_daemon_names(['grafana', 'prometheus', 'alertmanager', 'oauth2-proxy'])
         else:
             # this daemon type doesn't need deps mgmt
             pass

src/pybind/mgr/cephadm/services/ingress.py

@@ -241,7 +241,12 @@ def keepalived_generate_config(
         if spec.keepalived_password:
             password = spec.keepalived_password
 
-        daemons = self.mgr.cache.get_daemons_by_service(spec.service_name())
+        if spec.keepalive_only:
+            # when keepalive_only instead of haproxy, we have to monitor the backend service daemons
+            if spec.backend_service is not None:
+                daemons = self.mgr.cache.get_daemons_by_service(spec.backend_service)
+        else:
+            daemons = self.mgr.cache.get_daemons_by_service(spec.service_name())
 
         if not daemons and not spec.keepalive_only:
             raise OrchestratorError(
@@ -297,6 +302,10 @@ def _get_valid_interface_and_ip(vip: str, host: str) -> Tuple[str, str]:
                     port = d.ports[1]   # monitoring port
                     host_ip = d.ip or self.mgr.inventory.get_addr(d.hostname)
                     script = f'/usr/bin/curl {build_url(scheme="http", host=host_ip, port=port)}/health'
+                elif d.daemon_type == 'mgmt-gateway':
+                    mgmt_gw_port = d.ports[0] if d.ports else None
+                    host_ip = d.ip or self.mgr.inventory.get_addr(d.hostname)
+                    script = f'/usr/bin/curl -k {build_url(scheme="https", host=host_ip, port=mgmt_gw_port)}/health'
         assert script
 
         states = []

src/pybind/mgr/cephadm/services/mgmt_gateway.py

@@ -1,5 +1,5 @@
 import logging
-from typing import List, Any, Tuple, Dict, cast, Optional
+from typing import List, Any, Tuple, Dict, cast, TYPE_CHECKING
 
 from orchestrator import DaemonDescription
 from ceph.deployment.service_spec import MgmtGatewaySpec, GrafanaSpec
@@ -36,10 +36,11 @@ def get_active_daemon(self, daemon_descrs: List[DaemonDescription]) -> DaemonDes
         # if empty list provided, return empty Daemon Desc
         return DaemonDescription()
 
-    def get_oauth2_service_url(self) -> Optional[str]:
-        # TODO(redo): check how can we create several servers for HA
-        oauth2_servers = self.get_service_endpoints('oauth2-proxy')
-        return f'https://{oauth2_servers[0]}' if oauth2_servers else None
+    def get_mgmt_gw_ips(self, svc_spec: MgmtGatewaySpec, daemon_spec: CephadmDaemonDeploySpec) -> List[str]:
+        mgmt_gw_ips = [self.mgr.inventory.get_addr(daemon_spec.host)]
+        if svc_spec.virtual_ip is not None:
+            mgmt_gw_ips.append(svc_spec.virtual_ip)
+        return mgmt_gw_ips
 
     def config_dashboard(self, daemon_descrs: List[DaemonDescription]) -> None:
         # we adjust the standby behaviour so rev-proxy can pick correctly the active instance
@@ -56,9 +57,9 @@ def get_external_certificates(self, svc_spec: MgmtGatewaySpec, daemon_spec: Ceph
                 key = svc_spec.ssl_certificate_key
             else:
                 # not provided on the spec, let's generate self-sigend certificates
-                addr = self.mgr.inventory.get_addr(daemon_spec.host)
+                ips = self.get_mgmt_gw_ips(svc_spec, daemon_spec)
                 host_fqdn = self.mgr.get_fqdn(daemon_spec.host)
-                cert, key = self.mgr.cert_mgr.generate_cert(host_fqdn, addr)
+                cert, key = self.mgr.cert_mgr.generate_cert(host_fqdn, ips)
             # save certificates
             if cert and key:
                 self.mgr.cert_key_store.save_cert('mgmt_gw_cert', cert)
@@ -67,10 +68,18 @@ def get_external_certificates(self, svc_spec: MgmtGatewaySpec, daemon_spec: Ceph
                 logger.error("Failed to obtain certificate and key from mgmt-gateway.")
         return cert, key
 
-    def get_internal_certificates(self, daemon_spec: CephadmDaemonDeploySpec) -> Tuple[str, str]:
-        node_ip = self.mgr.inventory.get_addr(daemon_spec.host)
+    def get_internal_certificates(self, svc_spec: MgmtGatewaySpec, daemon_spec: CephadmDaemonDeploySpec) -> Tuple[str, str]:
+        ips = self.get_mgmt_gw_ips(svc_spec, daemon_spec)
         host_fqdn = self.mgr.get_fqdn(daemon_spec.host)
-        return self.mgr.cert_mgr.generate_cert(host_fqdn, node_ip)
+        return self.mgr.cert_mgr.generate_cert(host_fqdn, ips)
+
+    def get_service_discovery_endpoints(self) -> List[str]:
+        sd_endpoints = []
+        for dd in self.mgr.cache.get_daemons_by_service('mgr'):
+            assert dd.hostname is not None
+            addr = dd.ip if dd.ip else self.mgr.inventory.get_addr(dd.hostname)
+            sd_endpoints.append(f"{addr}:{self.mgr.service_discovery_port}")
+        return sd_endpoints
 
     def get_mgmt_gateway_deps(self) -> List[str]:
         # url_prefix for the following services depends on the presence of mgmt-gateway
@@ -79,10 +88,6 @@ def get_mgmt_gateway_deps(self) -> List[str]:
         deps += [d.name() for d in self.mgr.cache.get_daemons_by_service('alertmanager')]
         deps += [d.name() for d in self.mgr.cache.get_daemons_by_service('grafana')]
         deps += [d.name() for d in self.mgr.cache.get_daemons_by_service('oauth2-proxy')]
-        for dd in self.mgr.cache.get_daemons_by_service('mgr'):
-            # we consider mgr a dep even if the dashboard is disabled
-            # in order to be consistent with _calc_daemon_deps().
-            deps.append(dd.name())
 
         return deps
 
@@ -94,6 +99,8 @@ def generate_config(self, daemon_spec: CephadmDaemonDeploySpec) -> Tuple[Dict[st
         prometheus_endpoints = self.get_service_endpoints('prometheus')
         alertmanager_endpoints = self.get_service_endpoints('alertmanager')
         grafana_endpoints = self.get_service_endpoints('grafana')
+        oauth2_proxy_endpoints = self.get_service_endpoints('oauth2-proxy')
+        service_discovery_endpoints = self.get_service_discovery_endpoints()
         try:
             grafana_spec = cast(GrafanaSpec, self.mgr.spec_store['grafana'].spec)
             grafana_protocol = grafana_spec.protocol
@@ -104,7 +111,9 @@ def generate_config(self, daemon_spec: CephadmDaemonDeploySpec) -> Tuple[Dict[st
             'dashboard_endpoints': dashboard_endpoints,
             'prometheus_endpoints': prometheus_endpoints,
             'alertmanager_endpoints': alertmanager_endpoints,
-            'grafana_endpoints': grafana_endpoints
+            'grafana_endpoints': grafana_endpoints,
+            'oauth2_proxy_endpoints': oauth2_proxy_endpoints,
+            'service_discovery_endpoints': service_discovery_endpoints
         }
         server_context = {
             'spec': svc_spec,
@@ -117,11 +126,12 @@ def generate_config(self, daemon_spec: CephadmDaemonDeploySpec) -> Tuple[Dict[st
             'prometheus_endpoints': prometheus_endpoints,
             'alertmanager_endpoints': alertmanager_endpoints,
             'grafana_endpoints': grafana_endpoints,
-            'oauth2_proxy_url': self.get_oauth2_service_url(),
+            'service_discovery_endpoints': service_discovery_endpoints,
+            'enable_oauth2_proxy': bool(oauth2_proxy_endpoints),
         }
 
         cert, key = self.get_external_certificates(svc_spec, daemon_spec)
-        internal_cert, internal_pkey = self.get_internal_certificates(daemon_spec)
+        internal_cert, internal_pkey = self.get_internal_certificates(svc_spec, daemon_spec)
         daemon_config = {
             "files": {
                 "nginx.conf": self.mgr.template.render(self.SVC_TEMPLATE_PATH, main_context),