[ceph.git] / ceph / src / pybind / mgr / nfs / cluster.py

import logging
import json
import re
import socket
from typing import cast, Dict, List, Any, Union, Optional, TYPE_CHECKING, Tuple

from mgr_module import NFS_POOL_NAME as POOL_NAME
from ceph.deployment.service_spec import NFSServiceSpec, PlacementSpec, IngressSpec

import orchestrator

from .exception import NFSInvalidOperation, ClusterNotFound
from .utils import (available_clusters, restart_nfs_service, conf_obj_name,
                    user_conf_obj_name)
from .export import NFSRados, exception_handler

if TYPE_CHECKING:
    from nfs.module import Module
    from mgr_module import MgrModule


log = logging.getLogger(__name__)


def resolve_ip(hostname: str) -> str:
    try:
        r = socket.getaddrinfo(hostname, None, flags=socket.AI_CANONNAME,
                               type=socket.SOCK_STREAM)
        # pick first v4 IP, if present
        for a in r:
            if a[0] == socket.AF_INET:
                return a[4][0]
        return r[0][4][0]
    except socket.gaierror as e:
        raise NFSInvalidOperation(f"Cannot resolve IP for host {hostname}: {e}")


def create_ganesha_pool(mgr: 'MgrModule') -> None:
    pool_list = [p['pool_name'] for p in mgr.get_osdmap().dump().get('pools', [])]
    if POOL_NAME not in pool_list:
        mgr.check_mon_command({'prefix': 'osd pool create', 'pool': POOL_NAME})
        mgr.check_mon_command({'prefix': 'osd pool application enable',
                               'pool': POOL_NAME,
                               'app': 'nfs'})
        log.debug("Successfully created nfs-ganesha pool %s", POOL_NAME)


class NFSCluster:
    def __init__(self, mgr: 'Module') -> None:
        self.mgr = mgr

    def _call_orch_apply_nfs(
            self,
            cluster_id: str,
            placement: Optional[str],
            virtual_ip: Optional[str] = None,
            port: Optional[int] = None,
    ) -> None:
        if not port:
            port = 2049   # default nfs port
        if virtual_ip:
            # nfs + ingress
            # run NFS on non-standard port
            spec = NFSServiceSpec(service_type='nfs', service_id=cluster_id,
                                  placement=PlacementSpec.from_string(placement),
                                  # use non-default port so we don't conflict with ingress
                                  port=10000 + port)   # semi-arbitrary, fix me someday
            completion = self.mgr.apply_nfs(spec)
            orchestrator.raise_if_exception(completion)
            ispec = IngressSpec(service_type='ingress',
                                service_id='nfs.' + cluster_id,
                                backend_service='nfs.' + cluster_id,
                                frontend_port=port,
                                monitor_port=7000 + port,   # semi-arbitrary, fix me someday
                                virtual_ip=virtual_ip)
            completion = self.mgr.apply_ingress(ispec)
            orchestrator.raise_if_exception(completion)
        else:
            # standalone nfs
            spec = NFSServiceSpec(service_type='nfs', service_id=cluster_id,
                                  placement=PlacementSpec.from_string(placement),
                                  port=port)
            completion = self.mgr.apply_nfs(spec)
            orchestrator.raise_if_exception(completion)
        log.debug("Successfully deployed nfs daemons with cluster id %s and placement %s",
                  cluster_id, placement)

    def create_empty_rados_obj(self, cluster_id: str) -> None:
        common_conf = conf_obj_name(cluster_id)
        self._rados(cluster_id).write_obj('', conf_obj_name(cluster_id))
        log.info("Created empty object:%s", common_conf)

    def delete_config_obj(self, cluster_id: str) -> None:
        self._rados(cluster_id).remove_all_obj()
        log.info("Deleted %s object and all objects in %s",
                 conf_obj_name(cluster_id), cluster_id)

    def create_nfs_cluster(
            self,
            cluster_id: str,
            placement: Optional[str],
            virtual_ip: Optional[str],
            ingress: Optional[bool] = None,
            port: Optional[int] = None,
    ) -> Tuple[int, str, str]:
        try:
            if virtual_ip and not ingress:
                raise NFSInvalidOperation('virtual_ip can only be provided with ingress enabled')
            if not virtual_ip and ingress:
                raise NFSInvalidOperation('ingress currently requires a virtual_ip')
            invalid_str = re.search('[^A-Za-z0-9-_.]', cluster_id)
            if invalid_str:
                raise NFSInvalidOperation(f"cluster id {cluster_id} is invalid. "
                                          f"{invalid_str.group()} is char not permitted")

            create_ganesha_pool(self.mgr)

            self.create_empty_rados_obj(cluster_id)

            if cluster_id not in available_clusters(self.mgr):
                self._call_orch_apply_nfs(cluster_id, placement, virtual_ip, port)
                return 0, "NFS Cluster Created Successfully", ""
            return 0, "", f"{cluster_id} cluster already exists"
        except Exception as e:
            return exception_handler(e, f"NFS Cluster {cluster_id} could not be created")

    def delete_nfs_cluster(self, cluster_id: str) -> Tuple[int, str, str]:
        try:
            cluster_list = available_clusters(self.mgr)
            if cluster_id in cluster_list:
                self.mgr.export_mgr.delete_all_exports(cluster_id)
                completion = self.mgr.remove_service('ingress.nfs.' + cluster_id)
                orchestrator.raise_if_exception(completion)
                completion = self.mgr.remove_service('nfs.' + cluster_id)
                orchestrator.raise_if_exception(completion)
                self.delete_config_obj(cluster_id)
                return 0, "NFS Cluster Deleted Successfully", ""
            return 0, "", "Cluster does not exist"
        except Exception as e:
            return exception_handler(e, f"Failed to delete NFS Cluster {cluster_id}")

    def list_nfs_cluster(self) -> Tuple[int, str, str]:
        try:
            return 0, '\n'.join(available_clusters(self.mgr)), ""
        except Exception as e:
            return exception_handler(e, "Failed to list NFS Cluster")

    def _show_nfs_cluster_info(self, cluster_id: str) -> Dict[str, Any]:
        completion = self.mgr.list_daemons(daemon_type='nfs')
        # Here completion.result is a list DaemonDescription objects
        clusters = orchestrator.raise_if_exception(completion)
        backends: List[Dict[str, Union[Any]]] = []

        for cluster in clusters:
            if cluster_id == cluster.service_id():
                assert cluster.hostname
                try:
                    if cluster.ip:
                        ip = cluster.ip
                    else:
                        c = self.mgr.get_hosts()
                        orchestrator.raise_if_exception(c)
                        hosts = [h for h in c.result or []
                                 if h.hostname == cluster.hostname]
                        if hosts:
                            ip = resolve_ip(hosts[0].addr)
                        else:
                            # sigh
                            ip = resolve_ip(cluster.hostname)
                    backends.append({
                        "hostname": cluster.hostname,
                        "ip": ip,
                        "port": cluster.ports[0] if cluster.ports else None
                    })
                except orchestrator.OrchestratorError:
                    continue

        r: Dict[str, Any] = {
            'virtual_ip': None,
            'backend': backends,
        }
        sc = self.mgr.describe_service(service_type='ingress')
        services = orchestrator.raise_if_exception(sc)
        for i in services:
            spec = cast(IngressSpec, i.spec)
            if spec.backend_service == f'nfs.{cluster_id}':
                r['virtual_ip'] = i.virtual_ip.split('/')[0] if i.virtual_ip else None
                if i.ports:
                    r['port'] = i.ports[0]
                    if len(i.ports) > 1:
                        r['monitor_port'] = i.ports[1]
        log.debug("Successfully fetched %s info: %s", cluster_id, r)
        return r

    def show_nfs_cluster_info(self, cluster_id: Optional[str] = None) -> Tuple[int, str, str]:
        try:
            info_res = {}
            if cluster_id:
                cluster_ls = [cluster_id]
            else:
                cluster_ls = available_clusters(self.mgr)

            for cluster_id in cluster_ls:
                res = self._show_nfs_cluster_info(cluster_id)
                if res:
                    info_res[cluster_id] = res
            return (0, json.dumps(info_res, indent=4), '')
        except Exception as e:
            return exception_handler(e, "Failed to show info for cluster")

    def get_nfs_cluster_config(self, cluster_id: str) -> Tuple[int, str, str]:
        try:
            if cluster_id in available_clusters(self.mgr):
                rados_obj = self._rados(cluster_id)
                conf = rados_obj.read_obj(user_conf_obj_name(cluster_id))
                return 0, conf or "", ""
            raise ClusterNotFound()
        except Exception as e:
            return exception_handler(e, f"Fetching NFS-Ganesha Config failed for {cluster_id}")

    def set_nfs_cluster_config(self, cluster_id: str, nfs_config: str) -> Tuple[int, str, str]:
        try:
            if cluster_id in available_clusters(self.mgr):
                rados_obj = self._rados(cluster_id)
                if rados_obj.check_user_config():
                    return 0, "", "NFS-Ganesha User Config already exists"
                rados_obj.write_obj(nfs_config, user_conf_obj_name(cluster_id),
                                    conf_obj_name(cluster_id))
                log.debug("Successfully saved %s's user config: \n %s", cluster_id, nfs_config)
                restart_nfs_service(self.mgr, cluster_id)
                return 0, "NFS-Ganesha Config Set Successfully", ""
            raise ClusterNotFound()
        except NotImplementedError:
            return 0, "NFS-Ganesha Config Added Successfully "\
                "(Manual Restart of NFS PODS required)", ""
        except Exception as e:
            return exception_handler(e, f"Setting NFS-Ganesha Config failed for {cluster_id}")

    def reset_nfs_cluster_config(self, cluster_id: str) -> Tuple[int, str, str]:
        try:
            if cluster_id in available_clusters(self.mgr):
                rados_obj = self._rados(cluster_id)
                if not rados_obj.check_user_config():
                    return 0, "", "NFS-Ganesha User Config does not exist"
                rados_obj.remove_obj(user_conf_obj_name(cluster_id),
                                     conf_obj_name(cluster_id))
                restart_nfs_service(self.mgr, cluster_id)
                return 0, "NFS-Ganesha Config Reset Successfully", ""
            raise ClusterNotFound()
        except NotImplementedError:
            return 0, "NFS-Ganesha Config Removed Successfully "\
                "(Manual Restart of NFS PODS required)", ""
        except Exception as e:
            return exception_handler(e, f"Resetting NFS-Ganesha Config failed for {cluster_id}")

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