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

   1 import ipaddress
   2 import logging
   3 import random
   4 import string
   5 from typing import List, Dict, Any, Tuple, cast, Optional
   6
   7 from ceph.deployment.service_spec import IngressSpec
   8 from mgr_util import build_url
   9 from cephadm.utils import resolve_ip
  10 from orchestrator import OrchestratorError
  11 from cephadm.services.cephadmservice import CephadmDaemonDeploySpec, CephService
  12
  13 logger = logging.getLogger(__name__)
  14
  15
  16 class IngressService(CephService):
  17     TYPE = 'ingress'
  18
  19     def primary_daemon_type(self) -> str:
  20         return 'haproxy'
  21
  22     def per_host_daemon_type(self) -> Optional[str]:
  23         return 'keepalived'
  24
  25     def prepare_create(
  26             self,
  27             daemon_spec: CephadmDaemonDeploySpec,
  28     ) -> CephadmDaemonDeploySpec:
  29         if daemon_spec.daemon_type == 'haproxy':
  30             return self.haproxy_prepare_create(daemon_spec)
  31         if daemon_spec.daemon_type == 'keepalived':
  32             return self.keepalived_prepare_create(daemon_spec)
  33         assert False, "unexpected daemon type"
  34
  35     def generate_config(
  36             self,
  37             daemon_spec: CephadmDaemonDeploySpec
  38     ) -> Tuple[Dict[str, Any], List[str]]:
  39         if daemon_spec.daemon_type == 'haproxy':
  40             return self.haproxy_generate_config(daemon_spec)
  41         else:
  42             return self.keepalived_generate_config(daemon_spec)
  43         assert False, "unexpected daemon type"
  44
  45     def haproxy_prepare_create(
  46             self,
  47             daemon_spec: CephadmDaemonDeploySpec,
  48     ) -> CephadmDaemonDeploySpec:
  49         assert daemon_spec.daemon_type == 'haproxy'
  50
  51         daemon_id = daemon_spec.daemon_id
  52         host = daemon_spec.host
  53         spec = cast(IngressSpec, self.mgr.spec_store[daemon_spec.service_name].spec)
  54
  55         logger.debug('prepare_create haproxy.%s on host %s with spec %s' % (
  56             daemon_id, host, spec))
  57
  58         daemon_spec.final_config, daemon_spec.deps = self.haproxy_generate_config(daemon_spec)
  59
  60         return daemon_spec
  61
  62     def haproxy_generate_config(
  63             self,
  64             daemon_spec: CephadmDaemonDeploySpec,
  65     ) -> Tuple[Dict[str, Any], List[str]]:
  66         spec = cast(IngressSpec, self.mgr.spec_store[daemon_spec.service_name].spec)
  67         assert spec.backend_service
  68         if spec.backend_service not in self.mgr.spec_store:
  69             raise RuntimeError(
  70                 f'{spec.service_name()} backend service {spec.backend_service} does not exist')
  71         backend_spec = self.mgr.spec_store[spec.backend_service].spec
  72         daemons = self.mgr.cache.get_daemons_by_service(spec.backend_service)
  73         deps = [d.name() for d in daemons]
  74
  75         # generate password?
  76         pw_key = f'{spec.service_name()}/monitor_password'
  77         password = self.mgr.get_store(pw_key)
  78         if password is None:
  79             if not spec.monitor_password:
  80                 password = ''.join(random.choice(string.ascii_lowercase) for _ in range(20))
  81                 self.mgr.set_store(pw_key, password)
  82         else:
  83             if spec.monitor_password:
  84                 self.mgr.set_store(pw_key, None)
  85         if spec.monitor_password:
  86             password = spec.monitor_password
  87
  88         if backend_spec.service_type == 'nfs':
  89             mode = 'tcp'
  90             by_rank = {d.rank: d for d in daemons if d.rank is not None}
  91             servers = []
  92
  93             # try to establish how many ranks we *should* have
  94             num_ranks = backend_spec.placement.count
  95             if not num_ranks:
  96                 num_ranks = 1 + max(by_rank.keys())
  97
  98             for rank in range(num_ranks):
  99                 if rank in by_rank:
 100                     d = by_rank[rank]
 101                     assert(d.ports)
 102                     servers.append({
 103                         'name': f"{spec.backend_service}.{rank}",
 104                         'ip': d.ip or resolve_ip(self.mgr.inventory.get_addr(str(d.hostname))),
 105                         'port': d.ports[0],
 106                     })
 107                 else:
 108                     # offline/missing server; leave rank in place
 109                     servers.append({
 110                         'name': f"{spec.backend_service}.{rank}",
 111                         'ip': '0.0.0.0',
 112                         'port': 0,
 113                     })
 114         else:
 115             mode = 'http'
 116             servers = [
 117                 {
 118                     'name': d.name(),
 119                     'ip': d.ip or resolve_ip(self.mgr.inventory.get_addr(str(d.hostname))),
 120                     'port': d.ports[0],
 121                 } for d in daemons if d.ports
 122             ]
 123
 124         haproxy_conf = self.mgr.template.render(
 125             'services/ingress/haproxy.cfg.j2',
 126             {
 127                 'spec': spec,
 128                 'mode': mode,
 129                 'servers': servers,
 130                 'user': spec.monitor_user or 'admin',
 131                 'password': password,
 132                 'ip': str(spec.virtual_ip).split('/')[0] or daemon_spec.ip or '*',
 133                 'frontend_port': daemon_spec.ports[0] if daemon_spec.ports else spec.frontend_port,
 134                 'monitor_port': daemon_spec.ports[1] if daemon_spec.ports else spec.monitor_port,
 135             }
 136         )
 137         config_files = {
 138             'files': {
 139                 "haproxy.cfg": haproxy_conf,
 140             }
 141         }
 142         if spec.ssl_cert:
 143             ssl_cert = spec.ssl_cert
 144             if isinstance(ssl_cert, list):
 145                 ssl_cert = '\n'.join(ssl_cert)
 146             config_files['files']['haproxy.pem'] = ssl_cert
 147
 148         return config_files, sorted(deps)
 149
 150     def keepalived_prepare_create(
 151             self,
 152             daemon_spec: CephadmDaemonDeploySpec,
 153     ) -> CephadmDaemonDeploySpec:
 154         assert daemon_spec.daemon_type == 'keepalived'
 155
 156         daemon_id = daemon_spec.daemon_id
 157         host = daemon_spec.host
 158         spec = cast(IngressSpec, self.mgr.spec_store[daemon_spec.service_name].spec)
 159
 160         logger.debug('prepare_create keepalived.%s on host %s with spec %s' % (
 161             daemon_id, host, spec))
 162
 163         daemon_spec.final_config, daemon_spec.deps = self.keepalived_generate_config(daemon_spec)
 164
 165         return daemon_spec
 166
 167     def keepalived_generate_config(
 168             self,
 169             daemon_spec: CephadmDaemonDeploySpec,
 170     ) -> Tuple[Dict[str, Any], List[str]]:
 171         spec = cast(IngressSpec, self.mgr.spec_store[daemon_spec.service_name].spec)
 172         assert spec.backend_service
 173
 174         # generate password?
 175         pw_key = f'{spec.service_name()}/keepalived_password'
 176         password = self.mgr.get_store(pw_key)
 177         if password is None:
 178             if not spec.keepalived_password:
 179                 password = ''.join(random.choice(string.ascii_lowercase) for _ in range(20))
 180                 self.mgr.set_store(pw_key, password)
 181         else:
 182             if spec.keepalived_password:
 183                 self.mgr.set_store(pw_key, None)
 184         if spec.keepalived_password:
 185             password = spec.keepalived_password
 186
 187         daemons = self.mgr.cache.get_daemons_by_service(spec.service_name())
 188
 189         if not daemons:
 190             raise OrchestratorError(
 191                 f'Failed to generate keepalived.conf: No daemons deployed for {spec.service_name()}')
 192
 193         deps = sorted([d.name() for d in daemons if d.daemon_type == 'haproxy'])
 194
 195         host = daemon_spec.host
 196         hosts = sorted(list(set([host] + [str(d.hostname) for d in daemons])))
 197
 198         # interface
 199         bare_ip = str(spec.virtual_ip).split('/')[0]
 200         interface = None
 201         for subnet, ifaces in self.mgr.cache.networks.get(host, {}).items():
 202             if ifaces and ipaddress.ip_address(bare_ip) in ipaddress.ip_network(subnet):
 203                 interface = list(ifaces.keys())[0]
 204                 logger.info(
 205                     f'{bare_ip} is in {subnet} on {host} interface {interface}'
 206                 )
 207                 break
 208         # try to find interface by matching spec.virtual_interface_networks
 209         if not interface and spec.virtual_interface_networks:
 210             for subnet, ifaces in self.mgr.cache.networks.get(host, {}).items():
 211                 if subnet in spec.virtual_interface_networks:
 212                     interface = list(ifaces.keys())[0]
 213                     logger.info(
 214                         f'{spec.virtual_ip} will be configured on {host} interface '
 215                         f'{interface} (which has guiding subnet {subnet})'
 216                     )
 217                     break
 218         if not interface:
 219             raise OrchestratorError(
 220                 f"Unable to identify interface for {spec.virtual_ip} on {host}"
 221             )
 222
 223         # script to monitor health
 224         script = '/usr/bin/false'
 225         for d in daemons:
 226             if d.hostname == host:
 227                 if d.daemon_type == 'haproxy':
 228                     assert d.ports
 229                     port = d.ports[1]   # monitoring port
 230                     script = f'/usr/bin/curl {build_url(scheme="http", host=d.ip or "localhost", port=port)}/health'
 231         assert script
 232
 233         # set state. first host in placement is master all others backups
 234         state = 'BACKUP'
 235         if hosts[0] == host:
 236             state = 'MASTER'
 237
 238         # remove host, daemon is being deployed on from hosts list for
 239         # other_ips in conf file and converter to ips
 240         if host in hosts:
 241             hosts.remove(host)
 242         other_ips = [resolve_ip(self.mgr.inventory.get_addr(h)) for h in hosts]
 243
 244         keepalived_conf = self.mgr.template.render(
 245             'services/ingress/keepalived.conf.j2',
 246             {
 247                 'spec': spec,
 248                 'script': script,
 249                 'password': password,
 250                 'interface': interface,
 251                 'state': state,
 252                 'other_ips': other_ips,
 253                 'host_ip': resolve_ip(self.mgr.inventory.get_addr(host)),
 254             }
 255         )
 256
 257         config_file = {
 258             'files': {
 259                 "keepalived.conf": keepalived_conf,
 260             }
 261         }
 262
 263         return config_file, deps