ceph/src/pybind/mgr/nfs/cluster.py

   1 import ipaddress
   2 import logging
   3 import json
   4 import re
   5 import socket
   6 from typing import cast, Dict, List, Any, Union, Optional, TYPE_CHECKING, Tuple
   7
   8 from mgr_module import NFS_POOL_NAME as POOL_NAME
   9 from ceph.deployment.service_spec import NFSServiceSpec, PlacementSpec, IngressSpec
  10
  11 import orchestrator
  12
  13 from .exception import NFSInvalidOperation, ClusterNotFound
  14 from .utils import (available_clusters, restart_nfs_service, conf_obj_name,
  15                     user_conf_obj_name)
  16 from .export import NFSRados, exception_handler
  17
  18 if TYPE_CHECKING:
  19     from nfs.module import Module
  20     from mgr_module import MgrModule
  21
  22
  23 log = logging.getLogger(__name__)
  24
  25
  26 def resolve_ip(hostname: str) -> str:
  27     try:
  28         r = socket.getaddrinfo(hostname, None, flags=socket.AI_CANONNAME,
  29                                type=socket.SOCK_STREAM)
  30         # pick first v4 IP, if present
  31         for a in r:
  32             if a[0] == socket.AF_INET:
  33                 return a[4][0]
  34         return r[0][4][0]
  35     except socket.gaierror as e:
  36         raise NFSInvalidOperation(f"Cannot resolve IP for host {hostname}: {e}")
  37
  38
  39 def create_ganesha_pool(mgr: 'MgrModule') -> None:
  40     pool_list = [p['pool_name'] for p in mgr.get_osdmap().dump().get('pools', [])]
  41     if POOL_NAME not in pool_list:
  42         mgr.check_mon_command({'prefix': 'osd pool create', 'pool': POOL_NAME})
  43         mgr.check_mon_command({'prefix': 'osd pool application enable',
  44                                'pool': POOL_NAME,
  45                                'app': 'nfs'})
  46         log.debug("Successfully created nfs-ganesha pool %s", POOL_NAME)
  47
  48
  49 class NFSCluster:
  50     def __init__(self, mgr: 'Module') -> None:
  51         self.mgr = mgr
  52
  53     def _call_orch_apply_nfs(
  54             self,
  55             cluster_id: str,
  56             placement: Optional[str],
  57             virtual_ip: Optional[str] = None,
  58             port: Optional[int] = None,
  59     ) -> None:
  60         if not port:
  61             port = 2049   # default nfs port
  62         if virtual_ip:
  63             # nfs + ingress
  64             # run NFS on non-standard port
  65             spec = NFSServiceSpec(service_type='nfs', service_id=cluster_id,
  66                                   placement=PlacementSpec.from_string(placement),
  67                                   # use non-default port so we don't conflict with ingress
  68                                   port=10000 + port)   # semi-arbitrary, fix me someday
  69             completion = self.mgr.apply_nfs(spec)
  70             orchestrator.raise_if_exception(completion)
  71             ispec = IngressSpec(service_type='ingress',
  72                                 service_id='nfs.' + cluster_id,
  73                                 backend_service='nfs.' + cluster_id,
  74                                 frontend_port=port,
  75                                 monitor_port=7000 + port,   # semi-arbitrary, fix me someday
  76                                 virtual_ip=virtual_ip)
  77             completion = self.mgr.apply_ingress(ispec)
  78             orchestrator.raise_if_exception(completion)
  79         else:
  80             # standalone nfs
  81             spec = NFSServiceSpec(service_type='nfs', service_id=cluster_id,
  82                                   placement=PlacementSpec.from_string(placement),
  83                                   port=port)
  84             completion = self.mgr.apply_nfs(spec)
  85             orchestrator.raise_if_exception(completion)
  86         log.debug("Successfully deployed nfs daemons with cluster id %s and placement %s",
  87                   cluster_id, placement)
  88
  89     def create_empty_rados_obj(self, cluster_id: str) -> None:
  90         common_conf = conf_obj_name(cluster_id)
  91         self._rados(cluster_id).write_obj('', conf_obj_name(cluster_id))
  92         log.info("Created empty object:%s", common_conf)
  93
  94     def delete_config_obj(self, cluster_id: str) -> None:
  95         self._rados(cluster_id).remove_all_obj()
  96         log.info("Deleted %s object and all objects in %s",
  97                  conf_obj_name(cluster_id), cluster_id)
  98
  99     def create_nfs_cluster(
 100             self,
 101             cluster_id: str,
 102             placement: Optional[str],
 103             virtual_ip: Optional[str],
 104             ingress: Optional[bool] = None,
 105             port: Optional[int] = None,
 106     ) -> Tuple[int, str, str]:
 107
 108         try:
 109             if virtual_ip:
 110                 # validate virtual_ip value: ip_address throws a ValueError
 111                 # exception in case it's not a valid ipv4 or ipv6 address
 112                 ip = virtual_ip.split('/')[0]
 113                 ipaddress.ip_address(ip)
 114             if virtual_ip and not ingress:
 115                 raise NFSInvalidOperation('virtual_ip can only be provided with ingress enabled')
 116             if not virtual_ip and ingress:
 117                 raise NFSInvalidOperation('ingress currently requires a virtual_ip')
 118             invalid_str = re.search('[^A-Za-z0-9-_.]', cluster_id)
 119             if invalid_str:
 120                 raise NFSInvalidOperation(f"cluster id {cluster_id} is invalid. "
 121                                           f"{invalid_str.group()} is char not permitted")
 122
 123             create_ganesha_pool(self.mgr)
 124
 125             self.create_empty_rados_obj(cluster_id)
 126
 127             if cluster_id not in available_clusters(self.mgr):
 128                 self._call_orch_apply_nfs(cluster_id, placement, virtual_ip, port)
 129                 return 0, "NFS Cluster Created Successfully", ""
 130             return 0, "", f"{cluster_id} cluster already exists"
 131         except Exception as e:
 132             return exception_handler(e, f"NFS Cluster {cluster_id} could not be created")
 133
 134     def delete_nfs_cluster(self, cluster_id: str) -> Tuple[int, str, str]:
 135         try:
 136             cluster_list = available_clusters(self.mgr)
 137             if cluster_id in cluster_list:
 138                 self.mgr.export_mgr.delete_all_exports(cluster_id)
 139                 completion = self.mgr.remove_service('ingress.nfs.' + cluster_id)
 140                 orchestrator.raise_if_exception(completion)
 141                 completion = self.mgr.remove_service('nfs.' + cluster_id)
 142                 orchestrator.raise_if_exception(completion)
 143                 self.delete_config_obj(cluster_id)
 144                 return 0, "NFS Cluster Deleted Successfully", ""
 145             return 0, "", "Cluster does not exist"
 146         except Exception as e:
 147             return exception_handler(e, f"Failed to delete NFS Cluster {cluster_id}")
 148
 149     def list_nfs_cluster(self) -> Tuple[int, str, str]:
 150         try:
 151             return 0, '\n'.join(available_clusters(self.mgr)), ""
 152         except Exception as e:
 153             return exception_handler(e, "Failed to list NFS Cluster")
 154
 155     def _show_nfs_cluster_info(self, cluster_id: str) -> Dict[str, Any]:
 156         completion = self.mgr.list_daemons(daemon_type='nfs')
 157         # Here completion.result is a list DaemonDescription objects
 158         clusters = orchestrator.raise_if_exception(completion)
 159         backends: List[Dict[str, Union[Any]]] = []
 160
 161         for cluster in clusters:
 162             if cluster_id == cluster.service_id():
 163                 assert cluster.hostname
 164                 try:
 165                     if cluster.ip:
 166                         ip = cluster.ip
 167                     else:
 168                         c = self.mgr.get_hosts()
 169                         orchestrator.raise_if_exception(c)
 170                         hosts = [h for h in c.result or []
 171                                  if h.hostname == cluster.hostname]
 172                         if hosts:
 173                             ip = resolve_ip(hosts[0].addr)
 174                         else:
 175                             # sigh
 176                             ip = resolve_ip(cluster.hostname)
 177                     backends.append({
 178                         "hostname": cluster.hostname,
 179                         "ip": ip,
 180                         "port": cluster.ports[0] if cluster.ports else None
 181                     })
 182                 except orchestrator.OrchestratorError:
 183                     continue
 184
 185         r: Dict[str, Any] = {
 186             'virtual_ip': None,
 187             'backend': backends,
 188         }
 189         sc = self.mgr.describe_service(service_type='ingress')
 190         services = orchestrator.raise_if_exception(sc)
 191         for i in services:
 192             spec = cast(IngressSpec, i.spec)
 193             if spec.backend_service == f'nfs.{cluster_id}':
 194                 r['virtual_ip'] = i.virtual_ip.split('/')[0] if i.virtual_ip else None
 195                 if i.ports:
 196                     r['port'] = i.ports[0]
 197                     if len(i.ports) > 1:
 198                         r['monitor_port'] = i.ports[1]
 199         log.debug("Successfully fetched %s info: %s", cluster_id, r)
 200         return r
 201
 202     def show_nfs_cluster_info(self, cluster_id: Optional[str] = None) -> Tuple[int, str, str]:
 203         try:
 204             info_res = {}
 205             if cluster_id:
 206                 cluster_ls = [cluster_id]
 207             else:
 208                 cluster_ls = available_clusters(self.mgr)
 209
 210             for cluster_id in cluster_ls:
 211                 res = self._show_nfs_cluster_info(cluster_id)
 212                 if res:
 213                     info_res[cluster_id] = res
 214             return (0, json.dumps(info_res, indent=4), '')
 215         except Exception as e:
 216             return exception_handler(e, "Failed to show info for cluster")
 217
 218     def get_nfs_cluster_config(self, cluster_id: str) -> Tuple[int, str, str]:
 219         try:
 220             if cluster_id in available_clusters(self.mgr):
 221                 rados_obj = self._rados(cluster_id)
 222                 conf = rados_obj.read_obj(user_conf_obj_name(cluster_id))
 223                 return 0, conf or "", ""
 224             raise ClusterNotFound()
 225         except Exception as e:
 226             return exception_handler(e, f"Fetching NFS-Ganesha Config failed for {cluster_id}")
 227
 228     def set_nfs_cluster_config(self, cluster_id: str, nfs_config: str) -> Tuple[int, str, str]:
 229         try:
 230             if cluster_id in available_clusters(self.mgr):
 231                 rados_obj = self._rados(cluster_id)
 232                 if rados_obj.check_user_config():
 233                     return 0, "", "NFS-Ganesha User Config already exists"
 234                 rados_obj.write_obj(nfs_config, user_conf_obj_name(cluster_id),
 235                                     conf_obj_name(cluster_id))
 236                 log.debug("Successfully saved %s's user config: \n %s", cluster_id, nfs_config)
 237                 restart_nfs_service(self.mgr, cluster_id)
 238                 return 0, "NFS-Ganesha Config Set Successfully", ""
 239             raise ClusterNotFound()
 240         except NotImplementedError:
 241             return 0, "NFS-Ganesha Config Added Successfully "\
 242                 "(Manual Restart of NFS PODS required)", ""
 243         except Exception as e:
 244             return exception_handler(e, f"Setting NFS-Ganesha Config failed for {cluster_id}")
 245
 246     def reset_nfs_cluster_config(self, cluster_id: str) -> Tuple[int, str, str]:
 247         try:
 248             if cluster_id in available_clusters(self.mgr):
 249                 rados_obj = self._rados(cluster_id)
 250                 if not rados_obj.check_user_config():
 251                     return 0, "", "NFS-Ganesha User Config does not exist"
 252                 rados_obj.remove_obj(user_conf_obj_name(cluster_id),
 253                                      conf_obj_name(cluster_id))
 254                 restart_nfs_service(self.mgr, cluster_id)
 255                 return 0, "NFS-Ganesha Config Reset Successfully", ""
 256             raise ClusterNotFound()
 257         except NotImplementedError:
 258             return 0, "NFS-Ganesha Config Removed Successfully "\
 259                 "(Manual Restart of NFS PODS required)", ""
 260         except Exception as e:
 261             return exception_handler(e, f"Resetting NFS-Ganesha Config failed for {cluster_id}")
 262
 263     def _rados(self, cluster_id: str) -> NFSRados:
 264         """Return a new NFSRados object for the given cluster id."""
 265         return NFSRados(self.mgr.rados, cluster_id)