summaryrefslogtreecommitdiffstats
path: root/src/pybind/mgr/rook/rook_cluster.py
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--src/pybind/mgr/rook/rook_cluster.py1591
1 files changed, 1591 insertions, 0 deletions
diff --git a/src/pybind/mgr/rook/rook_cluster.py b/src/pybind/mgr/rook/rook_cluster.py
new file mode 100644
index 000000000..5c7c9fc04
--- /dev/null
+++ b/src/pybind/mgr/rook/rook_cluster.py
@@ -0,0 +1,1591 @@
+"""
+This module wrap's Rook + Kubernetes APIs to expose the calls
+needed to implement an orchestrator module. While the orchestrator
+module exposes an async API, this module simply exposes blocking API
+call methods.
+
+This module is runnable outside of ceph-mgr, useful for testing.
+"""
+import datetime
+import threading
+import logging
+from contextlib import contextmanager
+from time import sleep
+import re
+from orchestrator import OrchResult
+
+import jsonpatch
+from urllib.parse import urljoin
+import json
+
+# Optional kubernetes imports to enable MgrModule.can_run
+# to behave cleanly.
+from urllib3.exceptions import ProtocolError
+
+from ceph.deployment.inventory import Device
+from ceph.deployment.drive_group import DriveGroupSpec
+from ceph.deployment.service_spec import ServiceSpec, NFSServiceSpec, RGWSpec, PlacementSpec, HostPlacementSpec
+from ceph.utils import datetime_now
+from ceph.deployment.drive_selection.matchers import SizeMatcher
+from nfs.cluster import create_ganesha_pool
+from nfs.module import Module
+from nfs.export import NFSRados
+from mgr_module import NFS_POOL_NAME
+from mgr_util import merge_dicts
+
+from typing import Optional, Tuple, TypeVar, List, Callable, Any, cast, Generic, \
+ Iterable, Dict, Iterator, Type
+
+try:
+ from kubernetes import client, watch
+ from kubernetes.client.rest import ApiException
+except ImportError:
+ class ApiException(Exception): # type: ignore
+ status = 0
+
+from .rook_client.ceph import cephfilesystem as cfs
+from .rook_client.ceph import cephnfs as cnfs
+from .rook_client.ceph import cephobjectstore as cos
+from .rook_client.ceph import cephcluster as ccl
+from .rook_client.ceph import cephrbdmirror as crbdm
+from .rook_client._helper import CrdClass
+
+import orchestrator
+
+try:
+ from rook.module import RookEnv, RookOrchestrator
+except ImportError:
+ pass # just used for type checking.
+
+
+T = TypeVar('T')
+FuncT = TypeVar('FuncT', bound=Callable)
+
+CrdClassT = TypeVar('CrdClassT', bound=CrdClass)
+
+
+log = logging.getLogger(__name__)
+
+
+def __urllib3_supports_read_chunked() -> bool:
+ # There is a bug in CentOS 7 as it ships a urllib3 which is lower
+ # than required by kubernetes-client
+ try:
+ from urllib3.response import HTTPResponse
+ return hasattr(HTTPResponse, 'read_chunked')
+ except ImportError:
+ return False
+
+
+_urllib3_supports_read_chunked = __urllib3_supports_read_chunked()
+
+class ApplyException(orchestrator.OrchestratorError):
+ """
+ For failures to update the Rook CRDs, usually indicating
+ some kind of interference between our attempted update
+ and other conflicting activity.
+ """
+
+
+def threaded(f: Callable[..., None]) -> Callable[..., threading.Thread]:
+ def wrapper(*args: Any, **kwargs: Any) -> threading.Thread:
+ t = threading.Thread(target=f, args=args, kwargs=kwargs)
+ t.start()
+ return t
+
+ return cast(Callable[..., threading.Thread], wrapper)
+
+
+class DefaultFetcher():
+ def __init__(self, storage_class_name: str, coreV1_api: 'client.CoreV1Api', rook_env: 'RookEnv'):
+ self.storage_class_name = storage_class_name
+ self.coreV1_api = coreV1_api
+ self.rook_env = rook_env
+ self.pvs_in_sc: List[client.V1PersistentVolumeList] = []
+
+ def fetch(self) -> None:
+ self.inventory: KubernetesResource[client.V1PersistentVolumeList] = KubernetesResource(self.coreV1_api.list_persistent_volume)
+ self.pvs_in_sc = [i for i in self.inventory.items if i.spec.storage_class_name == self.storage_class_name]
+
+ def convert_size(self, size_str: str) -> int:
+ units = ("", "Ki", "Mi", "Gi", "Ti", "Pi", "Ei", "", "K", "M", "G", "T", "P", "E")
+ coeff_and_unit = re.search('(\d+)(\D+)', size_str)
+ assert coeff_and_unit is not None
+ coeff = int(coeff_and_unit[1])
+ unit = coeff_and_unit[2]
+ try:
+ factor = units.index(unit) % 7
+ except ValueError:
+ log.error("PV size format invalid")
+ raise
+ size = coeff * (2 ** (10 * factor))
+ return size
+
+ def devices(self) -> Dict[str, List[Device]]:
+ nodename_to_devices: Dict[str, List[Device]] = {}
+ for i in self.pvs_in_sc:
+ node, device = self.device(i)
+ if node not in nodename_to_devices:
+ nodename_to_devices[node] = []
+ nodename_to_devices[node].append(device)
+ return nodename_to_devices
+
+ def device(self, i: 'client.V1PersistentVolume') -> Tuple[str, Device]:
+ node = 'N/A'
+ if i.spec.node_affinity:
+ terms = i.spec.node_affinity.required.node_selector_terms
+ if len(terms) == 1 and len(terms[0].match_expressions) == 1 and terms[0].match_expressions[0].key == 'kubernetes.io/hostname' and len(terms[0].match_expressions[0].values) == 1:
+ node = terms[0].match_expressions[0].values[0]
+ size = self.convert_size(i.spec.capacity['storage'])
+ path = i.spec.host_path.path if i.spec.host_path else i.spec.local.path if i.spec.local else ('/dev/' + i.metadata.annotations['storage.openshift.com/device-name']) if i.metadata.annotations and 'storage.openshift.com/device-name' in i.metadata.annotations else ''
+ state = i.spec.volume_mode == 'Block' and i.status.phase == 'Available'
+ pv_name = i.metadata.name
+ device = Device(
+ path = path,
+ sys_api = dict(
+ size = size,
+ node = node,
+ pv_name = pv_name
+ ),
+ available = state,
+ )
+ return (node, device)
+
+
+class LSOFetcher(DefaultFetcher):
+ def __init__(self, storage_class: 'str', coreV1_api: 'client.CoreV1Api', rook_env: 'RookEnv', customObjects_api: 'client.CustomObjectsApi', nodenames: 'Optional[List[str]]' = None):
+ super().__init__(storage_class, coreV1_api, rook_env)
+ self.customObjects_api = customObjects_api
+ self.nodenames = nodenames
+
+ def fetch(self) -> None:
+ super().fetch()
+ self.discovery: KubernetesCustomResource = KubernetesCustomResource(self.customObjects_api.list_cluster_custom_object,
+ group="local.storage.openshift.io",
+ version="v1alpha1",
+ plural="localvolumediscoveryresults")
+
+ def predicate(self, item: 'client.V1ConfigMapList') -> bool:
+ if self.nodenames is not None:
+ return item['spec']['nodeName'] in self.nodenames
+ else:
+ return True
+
+ def devices(self) -> Dict[str, List[Device]]:
+ try:
+ lso_discovery_results = [i for i in self.discovery.items if self.predicate(i)]
+ except ApiException as dummy_e:
+ log.error("Failed to fetch device metadata")
+ raise
+ self.lso_devices = {}
+ for i in lso_discovery_results:
+ drives = i['status']['discoveredDevices']
+ for drive in drives:
+ self.lso_devices[drive['deviceID'].split('/')[-1]] = drive
+ nodename_to_devices: Dict[str, List[Device]] = {}
+ for i in self.pvs_in_sc:
+ node, device = (None, None)
+ if (not i.metadata.annotations) or ('storage.openshift.com/device-id' not in i.metadata.annotations) or (i.metadata.annotations['storage.openshift.com/device-id'] not in self.lso_devices):
+ node, device = super().device(i)
+ else:
+ node, device = self.device(i)
+ if node not in nodename_to_devices:
+ nodename_to_devices[node] = []
+ nodename_to_devices[node].append(device)
+ return nodename_to_devices
+
+ def device(self, i: Any) -> Tuple[str, Device]:
+ node = i.metadata.labels['kubernetes.io/hostname']
+ device_discovery = self.lso_devices[i.metadata.annotations['storage.openshift.com/device-id']]
+ pv_name = i.metadata.name
+ vendor: str = device_discovery['model'].split()[0] if len(device_discovery['model'].split()) >= 1 else ''
+ model: str = ' '.join(device_discovery['model'].split()[1:]) if len(device_discovery['model'].split()) > 1 else ''
+ device = Device(
+ path = device_discovery['path'],
+ sys_api = dict(
+ size = device_discovery['size'],
+ rotational = '1' if device_discovery['property']=='Rotational' else '0',
+ node = node,
+ pv_name = pv_name,
+ model = model,
+ vendor = vendor
+ ),
+ available = device_discovery['status']['state']=='Available',
+ device_id = device_discovery['deviceID'].split('/')[-1],
+ lsm_data = dict(
+ serialNum = device_discovery['serial']
+ )
+ )
+ return (node, device)
+
+
+class PDFetcher(DefaultFetcher):
+ """ Physical Devices Fetcher"""
+ def __init__(self, coreV1_api: 'client.CoreV1Api', rook_env: 'RookEnv'):
+ super().__init__('', coreV1_api, rook_env)
+
+ def fetch(self) -> None:
+ """ Collect the devices information from k8s configmaps"""
+ self.dev_cms: KubernetesResource = KubernetesResource(self.coreV1_api.list_namespaced_config_map,
+ namespace=self.rook_env.operator_namespace,
+ label_selector='app=rook-discover')
+
+ def devices(self) -> Dict[str, List[Device]]:
+ """ Return the list of devices found"""
+ node_devices: Dict[str, List[Device]] = {}
+ for i in self.dev_cms.items:
+ devices_list: List[Device] = []
+ for d in json.loads(i.data['devices']):
+ devices_list.append(self.device(d)[1])
+ node_devices[i.metadata.labels['rook.io/node']] = devices_list
+
+ return node_devices
+
+ def device(self, devData: Dict[str,str]) -> Tuple[str, Device]:
+ """ Build an orchestrator device """
+ if 'cephVolumeData' in devData and devData['cephVolumeData']:
+ return "", Device.from_json(json.loads(devData['cephVolumeData']))
+ else:
+ return "", Device(
+ path='/dev/' + devData['name'],
+ sys_api=dict(
+ rotational='1' if devData['rotational'] else '0',
+ size=devData['size']
+ ),
+ available=False,
+ rejected_reasons=['device data coming from ceph-volume not provided'],
+ )
+
+
+class KubernetesResource(Generic[T]):
+ def __init__(self, api_func: Callable, **kwargs: Any) -> None:
+ """
+ Generic kubernetes Resource parent class
+
+ The api fetch and watch methods should be common across resource types,
+
+ Exceptions in the runner thread are propagated to the caller.
+
+ :param api_func: kubernetes client api function that is passed to the watcher
+ :param filter_func: signature: ``(Item) -> bool``.
+ """
+ self.kwargs = kwargs
+ self.api_func = api_func
+
+ # ``_items`` is accessed by different threads. I assume assignment is atomic.
+ self._items: Dict[str, T] = dict()
+ self.thread = None # type: Optional[threading.Thread]
+ self.exception: Optional[Exception] = None
+ if not _urllib3_supports_read_chunked:
+ logging.info('urllib3 is too old. Fallback to full fetches')
+
+ def _fetch(self) -> str:
+ """ Execute the requested api method as a one-off fetch"""
+ response = self.api_func(**self.kwargs)
+ metadata = response.metadata
+ self._items = {item.metadata.name: item for item in response.items}
+ log.info('Full fetch of {}. result: {}'.format(self.api_func, len(self._items)))
+ return metadata.resource_version
+
+ @property
+ def items(self) -> Iterable[T]:
+ """
+ Returns the items of the request.
+ Creates the watcher as a side effect.
+ :return:
+ """
+ if self.exception:
+ e = self.exception
+ self.exception = None
+ raise e # Propagate the exception to the user.
+ if not self.thread or not self.thread.is_alive():
+ resource_version = self._fetch()
+ if _urllib3_supports_read_chunked:
+ # Start a thread which will use the kubernetes watch client against a resource
+ log.debug("Attaching resource watcher for k8s {}".format(self.api_func))
+ self.thread = self._watch(resource_version)
+
+ return self._items.values()
+
+ def get_item_name(self, item: Any) -> Any:
+ try:
+ return item.metadata.name
+ except AttributeError:
+ raise AttributeError(
+ "{} doesn't contain a metadata.name. Unable to track changes".format(
+ self.api_func))
+ @threaded
+ def _watch(self, res_ver: Optional[str]) -> None:
+ """ worker thread that runs the kubernetes watch """
+
+ self.exception = None
+
+ w = watch.Watch()
+
+ try:
+ # execute generator to continually watch resource for changes
+ for event in w.stream(self.api_func, resource_version=res_ver, watch=True,
+ **self.kwargs):
+ self.health = ''
+ item = event['object']
+ name = self.get_item_name(item)
+
+ log.info('{} event: {}'.format(event['type'], name))
+
+ if event['type'] in ('ADDED', 'MODIFIED'):
+ self._items = merge_dicts(self._items, {name: item})
+ elif event['type'] == 'DELETED':
+ self._items = {k:v for k,v in self._items.items() if k != name}
+ elif event['type'] == 'BOOKMARK':
+ pass
+ elif event['type'] == 'ERROR':
+ raise ApiException(str(event))
+ else:
+ raise KeyError('Unknown watch event {}'.format(event['type']))
+ except ProtocolError as e:
+ if 'Connection broken' in str(e):
+ log.info('Connection reset.')
+ return
+ raise
+ except ApiException as e:
+ log.exception('K8s API failed. {}'.format(self.api_func))
+ self.exception = e
+ raise
+ except Exception as e:
+ log.exception("Watcher failed. ({})".format(self.api_func))
+ self.exception = e
+ raise
+
+class KubernetesCustomResource(KubernetesResource):
+ def _fetch(self) -> str:
+ response = self.api_func(**self.kwargs)
+ metadata = response['metadata']
+ self._items = {item['metadata']['name']: item for item in response['items']}
+ log.info('Full fetch of {}. result: {}'.format(self.api_func, len(self._items)))
+ return metadata['resourceVersion']
+
+ def get_item_name(self, item: Any) -> Any:
+ try:
+ return item['metadata']['name']
+ except AttributeError:
+ raise AttributeError(
+ "{} doesn't contain a metadata.name. Unable to track changes".format(
+ self.api_func))
+
+class DefaultCreator():
+ def __init__(self, inventory: 'Dict[str, List[Device]]', coreV1_api: 'client.CoreV1Api', storage_class_name: 'str'):
+ self.coreV1_api = coreV1_api
+ self.storage_class_name = storage_class_name
+ self.inventory = inventory
+
+ def device_to_device_set(self, drive_group: DriveGroupSpec, d: Device) -> ccl.StorageClassDeviceSetsItem:
+ device_set = ccl.StorageClassDeviceSetsItem(
+ name=d.sys_api['pv_name'],
+ volumeClaimTemplates= ccl.VolumeClaimTemplatesList(),
+ count=1,
+ encrypted=drive_group.encrypted,
+ portable=False
+ )
+ device_set.volumeClaimTemplates.append(
+ ccl.VolumeClaimTemplatesItem(
+ metadata=ccl.Metadata(
+ name="data"
+ ),
+ spec=ccl.Spec(
+ storageClassName=self.storage_class_name,
+ volumeMode="Block",
+ accessModes=ccl.CrdObjectList(["ReadWriteOnce"]),
+ resources={
+ "requests":{
+ "storage": 1
+ }
+ },
+ volumeName=d.sys_api['pv_name']
+ )
+ )
+ )
+ return device_set
+
+ def filter_devices(self, rook_pods: KubernetesResource, drive_group: DriveGroupSpec, matching_hosts: List[str]) -> List[Device]:
+ device_list = []
+ assert drive_group.data_devices is not None
+ sizematcher: Optional[SizeMatcher] = None
+ if drive_group.data_devices.size:
+ sizematcher = SizeMatcher('size', drive_group.data_devices.size)
+ limit = getattr(drive_group.data_devices, 'limit', None)
+ count = 0
+ all = getattr(drive_group.data_devices, 'all', None)
+ paths = [device.path for device in drive_group.data_devices.paths]
+ osd_list = []
+ for pod in rook_pods.items:
+ if (
+ hasattr(pod, 'metadata')
+ and hasattr(pod.metadata, 'labels')
+ and 'osd' in pod.metadata.labels
+ and 'ceph.rook.io/DeviceSet' in pod.metadata.labels
+ ):
+ osd_list.append(pod.metadata.labels['ceph.rook.io/DeviceSet'])
+ for _, node in self.inventory.items():
+ for device in node:
+ if device.sys_api['pv_name'] in osd_list:
+ count += 1
+ for _, node in self.inventory.items():
+ for device in node:
+ if not limit or (count < limit):
+ if device.available:
+ if (
+ all
+ or (
+ device.sys_api['node'] in matching_hosts
+ and ((sizematcher != None) or sizematcher.compare(device))
+ and (
+ not drive_group.data_devices.paths
+ or (device.path in paths)
+ )
+ )
+ ):
+ device_list.append(device)
+ count += 1
+
+ return device_list
+
+ def add_osds(self, rook_pods: KubernetesResource, drive_group: DriveGroupSpec, matching_hosts: List[str]) -> Any:
+ to_create = self.filter_devices(rook_pods, drive_group,matching_hosts)
+ assert drive_group.data_devices is not None
+ def _add_osds(current_cluster, new_cluster):
+ # type: (ccl.CephCluster, ccl.CephCluster) -> ccl.CephCluster
+ if not hasattr(new_cluster.spec, 'storage') or not new_cluster.spec.storage:
+ new_cluster.spec.storage = ccl.Storage()
+
+ if not hasattr(new_cluster.spec.storage, 'storageClassDeviceSets') or not new_cluster.spec.storage.storageClassDeviceSets:
+ new_cluster.spec.storage.storageClassDeviceSets = ccl.StorageClassDeviceSetsList()
+
+ existing_scds = [
+ scds.name for scds in new_cluster.spec.storage.storageClassDeviceSets
+ ]
+ for device in to_create:
+ new_scds = self.device_to_device_set(drive_group, device)
+ if new_scds.name not in existing_scds:
+ new_cluster.spec.storage.storageClassDeviceSets.append(new_scds)
+ return new_cluster
+ return _add_osds
+
+class LSOCreator(DefaultCreator):
+ def filter_devices(self, rook_pods: KubernetesResource, drive_group: DriveGroupSpec, matching_hosts: List[str]) -> List[Device]:
+ device_list = []
+ assert drive_group.data_devices is not None
+ sizematcher = None
+ if drive_group.data_devices.size:
+ sizematcher = SizeMatcher('size', drive_group.data_devices.size)
+ limit = getattr(drive_group.data_devices, 'limit', None)
+ all = getattr(drive_group.data_devices, 'all', None)
+ paths = [device.path for device in drive_group.data_devices.paths]
+ vendor = getattr(drive_group.data_devices, 'vendor', None)
+ model = getattr(drive_group.data_devices, 'model', None)
+ count = 0
+ osd_list = []
+ for pod in rook_pods.items:
+ if (
+ hasattr(pod, 'metadata')
+ and hasattr(pod.metadata, 'labels')
+ and 'osd' in pod.metadata.labels
+ and 'ceph.rook.io/DeviceSet' in pod.metadata.labels
+ ):
+ osd_list.append(pod.metadata.labels['ceph.rook.io/DeviceSet'])
+ for _, node in self.inventory.items():
+ for device in node:
+ if device.sys_api['pv_name'] in osd_list:
+ count += 1
+ for _, node in self.inventory.items():
+ for device in node:
+ if not limit or (count < limit):
+ if device.available:
+ if (
+ all
+ or (
+ device.sys_api['node'] in matching_hosts
+ and ((sizematcher != None) or sizematcher.compare(device))
+ and (
+ not drive_group.data_devices.paths
+ or device.path in paths
+ )
+ and (
+ not vendor
+ or device.sys_api['vendor'] == vendor
+ )
+ and (
+ not model
+ or device.sys_api['model'].startsWith(model)
+ )
+ )
+ ):
+ device_list.append(device)
+ count += 1
+ return device_list
+
+class DefaultRemover():
+ def __init__(
+ self,
+ coreV1_api: 'client.CoreV1Api',
+ batchV1_api: 'client.BatchV1Api',
+ appsV1_api: 'client.AppsV1Api',
+ osd_ids: List[str],
+ replace_flag: bool,
+ force_flag: bool,
+ mon_command: Callable,
+ patch: Callable,
+ rook_env: 'RookEnv',
+ inventory: Dict[str, List[Device]]
+ ):
+ self.batchV1_api = batchV1_api
+ self.appsV1_api = appsV1_api
+ self.coreV1_api = coreV1_api
+
+ self.osd_ids = osd_ids
+ self.replace_flag = replace_flag
+ self.force_flag = force_flag
+
+ self.mon_command = mon_command
+
+ self.patch = patch
+ self.rook_env = rook_env
+
+ self.inventory = inventory
+ self.osd_pods: KubernetesResource = KubernetesResource(self.coreV1_api.list_namespaced_pod,
+ namespace=self.rook_env.namespace,
+ label_selector='app=rook-ceph-osd')
+ self.jobs: KubernetesResource = KubernetesResource(self.batchV1_api.list_namespaced_job,
+ namespace=self.rook_env.namespace,
+ label_selector='app=rook-ceph-osd-prepare')
+ self.pvcs: KubernetesResource = KubernetesResource(self.coreV1_api.list_namespaced_persistent_volume_claim,
+ namespace=self.rook_env.namespace)
+
+
+ def remove_device_sets(self) -> str:
+ self.to_remove: Dict[str, int] = {}
+ self.pvc_to_remove: List[str] = []
+ for pod in self.osd_pods.items:
+ if (
+ hasattr(pod, 'metadata')
+ and hasattr(pod.metadata, 'labels')
+ and 'osd' in pod.metadata.labels
+ and pod.metadata.labels['osd'] in self.osd_ids
+ ):
+ if pod.metadata.labels['ceph.rook.io/DeviceSet'] in self.to_remove:
+ self.to_remove[pod.metadata.labels['ceph.rook.io/DeviceSet']] = self.to_remove[pod.metadata.labels['ceph.rook.io/DeviceSet']] + 1
+ else:
+ self.to_remove[pod.metadata.labels['ceph.rook.io/DeviceSet']] = 1
+ self.pvc_to_remove.append(pod.metadata.labels['ceph.rook.io/pvc'])
+ def _remove_osds(current_cluster, new_cluster):
+ # type: (ccl.CephCluster, ccl.CephCluster) -> ccl.CephCluster
+ assert new_cluster.spec.storage is not None and new_cluster.spec.storage.storageClassDeviceSets is not None
+ for _set in new_cluster.spec.storage.storageClassDeviceSets:
+ if _set.name in self.to_remove:
+ if _set.count == self.to_remove[_set.name]:
+ new_cluster.spec.storage.storageClassDeviceSets.remove(_set)
+ else:
+ _set.count = _set.count - self.to_remove[_set.name]
+ return new_cluster
+ return self.patch(ccl.CephCluster, 'cephclusters', self.rook_env.cluster_name, _remove_osds)
+
+ def check_force(self) -> None:
+ if not self.force_flag:
+ safe_args = {'prefix': 'osd safe-to-destroy',
+ 'ids': [str(x) for x in self.osd_ids]}
+ ret, out, err = self.mon_command(safe_args)
+ if ret != 0:
+ raise RuntimeError(err)
+
+ def set_osds_down(self) -> None:
+ down_flag_args = {
+ 'prefix': 'osd down',
+ 'ids': [str(x) for x in self.osd_ids]
+ }
+ ret, out, err = self.mon_command(down_flag_args)
+ if ret != 0:
+ raise RuntimeError(err)
+
+ def scale_deployments(self) -> None:
+ for osd_id in self.osd_ids:
+ self.appsV1_api.patch_namespaced_deployment_scale(namespace=self.rook_env.namespace,
+ name='rook-ceph-osd-{}'.format(osd_id),
+ body=client.V1Scale(spec=client.V1ScaleSpec(replicas=0)))
+
+ def set_osds_out(self) -> None:
+ out_flag_args = {
+ 'prefix': 'osd out',
+ 'ids': [str(x) for x in self.osd_ids]
+ }
+ ret, out, err = self.mon_command(out_flag_args)
+ if ret != 0:
+ raise RuntimeError(err)
+
+ def delete_deployments(self) -> None:
+ for osd_id in self.osd_ids:
+ self.appsV1_api.delete_namespaced_deployment(namespace=self.rook_env.namespace,
+ name='rook-ceph-osd-{}'.format(osd_id),
+ propagation_policy='Foreground')
+
+ def clean_up_prepare_jobs_and_pvc(self) -> None:
+ for job in self.jobs.items:
+ if job.metadata.labels['ceph.rook.io/pvc'] in self.pvc_to_remove:
+ self.batchV1_api.delete_namespaced_job(name=job.metadata.name, namespace=self.rook_env.namespace,
+ propagation_policy='Foreground')
+ self.coreV1_api.delete_namespaced_persistent_volume_claim(name=job.metadata.labels['ceph.rook.io/pvc'],
+ namespace=self.rook_env.namespace,
+ propagation_policy='Foreground')
+
+ def purge_osds(self) -> None:
+ for id in self.osd_ids:
+ purge_args = {
+ 'prefix': 'osd purge-actual',
+ 'id': int(id),
+ 'yes_i_really_mean_it': True
+ }
+ ret, out, err = self.mon_command(purge_args)
+ if ret != 0:
+ raise RuntimeError(err)
+
+ def destroy_osds(self) -> None:
+ for id in self.osd_ids:
+ destroy_args = {
+ 'prefix': 'osd destroy-actual',
+ 'id': int(id),
+ 'yes_i_really_mean_it': True
+ }
+ ret, out, err = self.mon_command(destroy_args)
+ if ret != 0:
+ raise RuntimeError(err)
+
+ def remove(self) -> str:
+ try:
+ self.check_force()
+ except Exception as e:
+ log.exception("Error checking if OSDs are safe to destroy")
+ return f"OSDs not safe to destroy or unable to check if they are safe to destroy: {e}"
+ try:
+ remove_result = self.remove_device_sets()
+ except Exception as e:
+ log.exception("Error patching ceph cluster CRD")
+ return f"Not possible to modify Ceph cluster CRD: {e}"
+ try:
+ self.scale_deployments()
+ self.delete_deployments()
+ self.clean_up_prepare_jobs_and_pvc()
+ except Exception as e:
+ log.exception("Ceph cluster CRD patched, but error cleaning environment")
+ return f"Error cleaning environment after removing OSDs from Ceph cluster CRD: {e}"
+ try:
+ self.set_osds_down()
+ self.set_osds_out()
+ if self.replace_flag:
+ self.destroy_osds()
+ else:
+ self.purge_osds()
+ except Exception as e:
+ log.exception("OSDs removed from environment, but not able to remove OSDs from Ceph cluster")
+ return f"Error removing OSDs from Ceph cluster: {e}"
+
+ return remove_result
+
+
+
+class RookCluster(object):
+ # import of client.CoreV1Api must be optional at import time.
+ # Instead allow mgr/rook to be imported anyway.
+ def __init__(
+ self,
+ coreV1_api: 'client.CoreV1Api',
+ batchV1_api: 'client.BatchV1Api',
+ customObjects_api: 'client.CustomObjectsApi',
+ storageV1_api: 'client.StorageV1Api',
+ appsV1_api: 'client.AppsV1Api',
+ rook_env: 'RookEnv',
+ storage_class_name: 'str'
+ ):
+ self.rook_env = rook_env # type: RookEnv
+ self.coreV1_api = coreV1_api # client.CoreV1Api
+ self.batchV1_api = batchV1_api
+ self.customObjects_api = customObjects_api
+ self.storageV1_api = storageV1_api # client.StorageV1Api
+ self.appsV1_api = appsV1_api # client.AppsV1Api
+ self.storage_class_name = storage_class_name # type: str
+
+ # TODO: replace direct k8s calls with Rook API calls
+ self.available_storage_classes : KubernetesResource = KubernetesResource(self.storageV1_api.list_storage_class)
+ self.configured_storage_classes = self.list_storage_classes()
+
+ self.rook_pods: KubernetesResource[client.V1Pod] = KubernetesResource(self.coreV1_api.list_namespaced_pod,
+ namespace=self.rook_env.namespace,
+ label_selector="rook_cluster={0}".format(
+ self.rook_env.namespace))
+ self.nodes: KubernetesResource[client.V1Node] = KubernetesResource(self.coreV1_api.list_node)
+
+ def rook_url(self, path: str) -> str:
+ prefix = "/apis/ceph.rook.io/%s/namespaces/%s/" % (
+ self.rook_env.crd_version, self.rook_env.namespace)
+ return urljoin(prefix, path)
+
+ def rook_api_call(self, verb: str, path: str, **kwargs: Any) -> Any:
+ full_path = self.rook_url(path)
+ log.debug("[%s] %s" % (verb, full_path))
+
+ return self.coreV1_api.api_client.call_api(
+ full_path,
+ verb,
+ auth_settings=['BearerToken'],
+ response_type="object",
+ _return_http_data_only=True,
+ _preload_content=True,
+ **kwargs)
+
+ def rook_api_get(self, path: str, **kwargs: Any) -> Any:
+ return self.rook_api_call("GET", path, **kwargs)
+
+ def rook_api_delete(self, path: str) -> Any:
+ return self.rook_api_call("DELETE", path)
+
+ def rook_api_patch(self, path: str, **kwargs: Any) -> Any:
+ return self.rook_api_call("PATCH", path,
+ header_params={"Content-Type": "application/json-patch+json"},
+ **kwargs)
+
+ def rook_api_post(self, path: str, **kwargs: Any) -> Any:
+ return self.rook_api_call("POST", path, **kwargs)
+
+ def list_storage_classes(self) -> List[str]:
+ try:
+ crd = self.customObjects_api.get_namespaced_custom_object(
+ group="ceph.rook.io",
+ version="v1",
+ namespace=self.rook_env.namespace,
+ plural="cephclusters",
+ name=self.rook_env.cluster_name)
+
+ sc_devicesets = crd['spec']['storage']['storageClassDeviceSets']
+ sc_names = [vct['spec']['storageClassName'] for sc in sc_devicesets for vct in sc['volumeClaimTemplates']]
+ log.info(f"the cluster has the following configured sc: {sc_names}")
+ return sc_names
+ except Exception as e:
+ log.error(f"unable to list storage classes: {e}")
+ return []
+
+ # TODO: remove all the calls to code that uses rook_cluster.storage_class_name
+ def get_storage_class(self) -> 'client.V1StorageClass':
+ matching_sc = [i for i in self.available_storage_classes.items if self.storage_class_name == i.metadata.name]
+ if len(matching_sc) == 0:
+ log.error(f"No storage class exists matching configured Rook orchestrator storage class which currently is <{self.storage_class_name}>. This storage class can be set in ceph config (mgr/rook/storage_class)")
+ raise Exception('No storage class exists matching name provided in ceph config at mgr/rook/storage_class')
+ return matching_sc[0]
+
+ def get_discovered_devices(self, nodenames: Optional[List[str]] = None) -> Dict[str, List[Device]]:
+ discovered_devices: Dict[str, List[Device]] = {}
+ op_settings = self.coreV1_api.read_namespaced_config_map(name="rook-ceph-operator-config", namespace=self.rook_env.operator_namespace).data
+ fetcher: Optional[DefaultFetcher] = None
+ if op_settings.get('ROOK_ENABLE_DISCOVERY_DAEMON', 'false').lower() == 'true':
+ fetcher = PDFetcher(self.coreV1_api, self.rook_env)
+ fetcher.fetch()
+ discovered_devices = fetcher.devices()
+ else:
+ active_storage_classes = [sc for sc in self.available_storage_classes.items if sc.metadata.name in self.configured_storage_classes]
+ for sc in active_storage_classes:
+ if sc.metadata.labels and ('local.storage.openshift.io/owner-name' in sc.metadata.labels):
+ fetcher = LSOFetcher(sc.metadata.name, self.coreV1_api, self.customObjects_api, nodenames)
+ else:
+ fetcher = DefaultFetcher(sc.metadata.name, self.coreV1_api, self.rook_env)
+ fetcher.fetch()
+ discovered_devices.update(fetcher.devices())
+
+ return discovered_devices
+
+ def get_osds(self) -> List:
+ osd_pods: KubernetesResource = KubernetesResource(self.coreV1_api.list_namespaced_pod,
+ namespace=self.rook_env.namespace,
+ label_selector='app=rook-ceph-osd')
+ return list(osd_pods.items)
+
+ def get_nfs_conf_url(self, nfs_cluster: str, instance: str) -> Optional[str]:
+ #
+ # Fetch cephnfs object for "nfs_cluster" and then return a rados://
+ # URL for the instance within that cluster. If the fetch fails, just
+ # return None.
+ #
+ try:
+ ceph_nfs = self.rook_api_get("cephnfses/{0}".format(nfs_cluster))
+ except ApiException as e:
+ log.info("Unable to fetch cephnfs object: {}".format(e.status))
+ return None
+
+ pool = ceph_nfs['spec']['rados']['pool']
+ namespace = ceph_nfs['spec']['rados'].get('namespace', None)
+
+ if namespace == None:
+ url = "rados://{0}/conf-{1}.{2}".format(pool, nfs_cluster, instance)
+ else:
+ url = "rados://{0}/{1}/conf-{2}.{3}".format(pool, namespace, nfs_cluster, instance)
+ return url
+
+ def describe_pods(self,
+ service_type: Optional[str],
+ service_id: Optional[str],
+ nodename: Optional[str]) -> List[Dict[str, Any]]:
+ """
+ Go query the k8s API about deployment, containers related to this
+ filesystem
+
+ Example Rook Pod labels for a mgr daemon:
+ Labels: app=rook-ceph-mgr
+ pod-template-hash=2171958073
+ rook_cluster=rook
+ And MDS containers additionally have `rook_filesystem` label
+
+ Label filter is rook_cluster=<cluster namespace>
+ rook_file_system=<self.fs_name>
+ """
+ def predicate(item):
+ # type: (client.V1Pod) -> bool
+ metadata = item.metadata
+ if service_type is not None:
+ if metadata.labels['app'] != "rook-ceph-{0}".format(service_type):
+ return False
+
+ if service_id is not None:
+ try:
+ k, v = {
+ "mds": ("rook_file_system", service_id),
+ "osd": ("ceph-osd-id", service_id),
+ "mon": ("mon", service_id),
+ "mgr": ("mgr", service_id),
+ "nfs": ("nfs", service_id),
+ "rgw": ("ceph_rgw", service_id),
+ }[service_type]
+ except KeyError:
+ raise orchestrator.OrchestratorValidationError(
+ '{} not supported'.format(service_type))
+ if metadata.labels[k] != v:
+ return False
+
+ if nodename is not None:
+ if item.spec.node_name != nodename:
+ return False
+ return True
+
+ refreshed = datetime_now()
+ pods = [i for i in self.rook_pods.items if predicate(i)]
+
+ pods_summary = []
+ prefix = 'sha256:'
+
+ for p in pods:
+ d = p.to_dict()
+
+ image_name = None
+ for c in d['spec']['containers']:
+ # look at the first listed container in the pod...
+ image_name = c['image']
+ break
+
+ ls = d['status'].get('container_statuses')
+ if not ls:
+ # ignore pods with no containers
+ continue
+ image_id = ls[0]['image_id']
+ image_id = image_id.split(prefix)[1] if prefix in image_id else image_id
+
+ s = {
+ "name": d['metadata']['name'],
+ "hostname": d['spec']['node_name'],
+ "labels": d['metadata']['labels'],
+ 'phase': d['status']['phase'],
+ 'container_image_name': image_name,
+ 'container_image_id': image_id,
+ 'refreshed': refreshed,
+ # these may get set below...
+ 'started': None,
+ 'created': None,
+ }
+
+ # note: we want UTC
+ if d['metadata'].get('creation_timestamp', None):
+ s['created'] = d['metadata']['creation_timestamp'].astimezone(
+ tz=datetime.timezone.utc)
+ if d['status'].get('start_time', None):
+ s['started'] = d['status']['start_time'].astimezone(
+ tz=datetime.timezone.utc)
+
+ pods_summary.append(s)
+
+ return pods_summary
+
+ def remove_pods(self, names: List[str]) -> List[str]:
+ pods = [i for i in self.rook_pods.items]
+ for p in pods:
+ d = p.to_dict()
+ daemon_type = d['metadata']['labels']['app'].replace('rook-ceph-','')
+ daemon_id = d['metadata']['labels']['ceph_daemon_id']
+ name = daemon_type + '.' + daemon_id
+ if name in names:
+ self.coreV1_api.delete_namespaced_pod(
+ d['metadata']['name'],
+ self.rook_env.namespace,
+ body=client.V1DeleteOptions()
+ )
+ return [f'Removed Pod {n}' for n in names]
+
+ def get_node_names(self) -> List[str]:
+ return [i.metadata.name for i in self.nodes.items]
+
+ @contextmanager
+ def ignore_409(self, what: str) -> Iterator[None]:
+ try:
+ yield
+ except ApiException as e:
+ if e.status == 409:
+ # Idempotent, succeed.
+ log.info("{} already exists".format(what))
+ else:
+ raise
+
+ def apply_filesystem(self, spec: ServiceSpec, num_replicas: int,
+ leaf_type: str) -> str:
+ # TODO use spec.placement
+ # TODO warn if spec.extended has entries we don't kow how
+ # to action.
+ all_hosts = self.get_hosts()
+ def _update_fs(new: cfs.CephFilesystem) -> cfs.CephFilesystem:
+ new.spec.metadataServer.activeCount = spec.placement.count or 1
+ new.spec.metadataServer.placement = cfs.Placement(
+ nodeAffinity=cfs.NodeAffinity(
+ requiredDuringSchedulingIgnoredDuringExecution=cfs.RequiredDuringSchedulingIgnoredDuringExecution(
+ nodeSelectorTerms=cfs.NodeSelectorTermsList(
+ [placement_spec_to_node_selector(spec.placement, all_hosts)]
+ )
+ )
+ )
+ )
+ return new
+ def _create_fs() -> cfs.CephFilesystem:
+ fs = cfs.CephFilesystem(
+ apiVersion=self.rook_env.api_name,
+ metadata=dict(
+ name=spec.service_id,
+ namespace=self.rook_env.namespace,
+ ),
+ spec=cfs.Spec(
+ dataPools=cfs.DataPoolsList(
+ {
+ cfs.DataPoolsItem(
+ failureDomain=leaf_type,
+ replicated=cfs.Replicated(
+ size=num_replicas
+ )
+ )
+ }
+ ),
+ metadataPool=cfs.MetadataPool(
+ failureDomain=leaf_type,
+ replicated=cfs.Replicated(
+ size=num_replicas
+ )
+ ),
+ metadataServer=cfs.MetadataServer(
+ activeCount=spec.placement.count or 1,
+ activeStandby=True,
+ placement=
+ cfs.Placement(
+ nodeAffinity=cfs.NodeAffinity(
+ requiredDuringSchedulingIgnoredDuringExecution=cfs.RequiredDuringSchedulingIgnoredDuringExecution(
+ nodeSelectorTerms=cfs.NodeSelectorTermsList(
+ [placement_spec_to_node_selector(spec.placement, all_hosts)]
+ )
+ )
+ )
+ )
+ )
+ )
+ )
+ return fs
+ assert spec.service_id is not None
+ return self._create_or_patch(
+ cfs.CephFilesystem, 'cephfilesystems', spec.service_id,
+ _update_fs, _create_fs)
+
+ def get_matching_node(self, host: str) -> Any:
+ matching_node = None
+ for node in self.nodes.items:
+ if node.metadata.labels['kubernetes.io/hostname'] == host:
+ matching_node = node
+ return matching_node
+
+ def add_host_label(self, host: str, label: str) -> OrchResult[str]:
+ matching_node = self.get_matching_node(host)
+ if matching_node == None:
+ return OrchResult(None, RuntimeError(f"Cannot add {label} label to {host}: host not found in cluster"))
+ matching_node.metadata.labels['ceph-label/'+ label] = ""
+ self.coreV1_api.patch_node(host, matching_node)
+ return OrchResult(f'Added {label} label to {host}')
+
+ def remove_host_label(self, host: str, label: str) -> OrchResult[str]:
+ matching_node = self.get_matching_node(host)
+ if matching_node == None:
+ return OrchResult(None, RuntimeError(f"Cannot remove {label} label from {host}: host not found in cluster"))
+ matching_node.metadata.labels.pop('ceph-label/' + label, None)
+ self.coreV1_api.patch_node(host, matching_node)
+ return OrchResult(f'Removed {label} label from {host}')
+
+ def apply_objectstore(self, spec: RGWSpec, num_replicas: int, leaf_type: str) -> str:
+ assert spec.service_id is not None
+
+ name = spec.service_id
+
+ if '.' in spec.service_id:
+ # rook does not like . in the name. this is could
+ # there because it is a legacy rgw spec that was named
+ # like $realm.$zone, except that I doubt there were any
+ # users of this code. Instead, focus on future users and
+ # translate . to - (fingers crossed!) instead.
+ name = spec.service_id.replace('.', '-')
+
+ all_hosts = self.get_hosts()
+ def _create_zone() -> cos.CephObjectStore:
+ port = None
+ secure_port = None
+ if spec.ssl:
+ secure_port = spec.get_port()
+ else:
+ port = spec.get_port()
+ object_store = cos.CephObjectStore(
+ apiVersion=self.rook_env.api_name,
+ metadata=dict(
+ name=name,
+ namespace=self.rook_env.namespace
+ ),
+ spec=cos.Spec(
+ gateway=cos.Gateway(
+ port=port,
+ securePort=secure_port,
+ instances=spec.placement.count or 1,
+ placement=cos.Placement(
+ cos.NodeAffinity(
+ requiredDuringSchedulingIgnoredDuringExecution=cos.RequiredDuringSchedulingIgnoredDuringExecution(
+ nodeSelectorTerms=cos.NodeSelectorTermsList(
+ [
+ placement_spec_to_node_selector(spec.placement, all_hosts)
+ ]
+ )
+ )
+ )
+ )
+ ),
+ dataPool=cos.DataPool(
+ failureDomain=leaf_type,
+ replicated=cos.Replicated(
+ size=num_replicas
+ )
+ ),
+ metadataPool=cos.MetadataPool(
+ failureDomain=leaf_type,
+ replicated=cos.Replicated(
+ size=num_replicas
+ )
+ )
+ )
+ )
+ if spec.rgw_zone:
+ object_store.spec.zone=cos.Zone(
+ name=spec.rgw_zone
+ )
+ return object_store
+
+
+ def _update_zone(new: cos.CephObjectStore) -> cos.CephObjectStore:
+ if new.spec.gateway:
+ new.spec.gateway.instances = spec.placement.count or 1
+ else:
+ new.spec.gateway=cos.Gateway(
+ instances=spec.placement.count or 1
+ )
+ return new
+ return self._create_or_patch(
+ cos.CephObjectStore, 'cephobjectstores', name,
+ _update_zone, _create_zone)
+
+ def apply_nfsgw(self, spec: NFSServiceSpec, mgr: 'RookOrchestrator') -> str:
+ # TODO use spec.placement
+ # TODO warn if spec.extended has entries we don't kow how
+ # to action.
+ # TODO Number of pods should be based on the list of hosts in the
+ # PlacementSpec.
+ assert spec.service_id, "service id in NFS service spec cannot be an empty string or None " # for mypy typing
+ service_id = spec.service_id
+ mgr_module = cast(Module, mgr)
+ count = spec.placement.count or 1
+ def _update_nfs(new: cnfs.CephNFS) -> cnfs.CephNFS:
+ new.spec.server.active = count
+ return new
+
+ def _create_nfs() -> cnfs.CephNFS:
+ rook_nfsgw = cnfs.CephNFS(
+ apiVersion=self.rook_env.api_name,
+ metadata=dict(
+ name=spec.service_id,
+ namespace=self.rook_env.namespace,
+ ),
+ spec=cnfs.Spec(
+ rados=cnfs.Rados(
+ namespace=service_id,
+ pool=NFS_POOL_NAME,
+ ),
+ server=cnfs.Server(
+ active=count
+ )
+ )
+ )
+
+
+ return rook_nfsgw
+
+ create_ganesha_pool(mgr)
+ NFSRados(mgr_module.rados, service_id).write_obj('', f'conf-nfs.{spec.service_id}')
+ return self._create_or_patch(cnfs.CephNFS, 'cephnfses', service_id,
+ _update_nfs, _create_nfs)
+
+ def rm_service(self, rooktype: str, service_id: str) -> str:
+ self.customObjects_api.delete_namespaced_custom_object(group="ceph.rook.io", version="v1",
+ namespace=self.rook_env.namespace,
+ plural=rooktype, name=service_id)
+ objpath = "{0}/{1}".format(rooktype, service_id)
+ return f'Removed {objpath}'
+
+ def get_resource(self, resource_type: str) -> Iterable:
+ custom_objects: KubernetesCustomResource = KubernetesCustomResource(self.customObjects_api.list_namespaced_custom_object,
+ group="ceph.rook.io",
+ version="v1",
+ namespace=self.rook_env.namespace,
+ plural=resource_type)
+ return custom_objects.items
+
+ def can_create_osd(self) -> bool:
+ current_cluster = self.rook_api_get(
+ "cephclusters/{0}".format(self.rook_env.cluster_name))
+ use_all_nodes = current_cluster['spec'].get('useAllNodes', False)
+
+ # If useAllNodes is set, then Rook will not be paying attention
+ # to anything we put in 'nodes', so can't do OSD creation.
+ return not use_all_nodes
+
+ def node_exists(self, node_name: str) -> bool:
+ return node_name in self.get_node_names()
+
+ def update_mon_count(self, newcount: Optional[int]) -> str:
+ def _update_mon_count(current, new):
+ # type: (ccl.CephCluster, ccl.CephCluster) -> ccl.CephCluster
+ if newcount is None:
+ raise orchestrator.OrchestratorError('unable to set mon count to None')
+ if not new.spec.mon:
+ raise orchestrator.OrchestratorError("mon attribute not specified in new spec")
+ new.spec.mon.count = newcount
+ return new
+ return self._patch(ccl.CephCluster, 'cephclusters', self.rook_env.cluster_name, _update_mon_count)
+
+ def add_osds(self, drive_group, matching_hosts):
+ # type: (DriveGroupSpec, List[str]) -> str
+ assert drive_group.objectstore in ("bluestore", "filestore")
+ assert drive_group.service_id
+ storage_class = self.get_storage_class()
+ inventory = self.get_discovered_devices()
+ creator: Optional[DefaultCreator] = None
+ if (
+ storage_class.metadata.labels
+ and 'local.storage.openshift.io/owner-name' in storage_class.metadata.labels
+ ):
+ creator = LSOCreator(inventory, self.coreV1_api, self.storage_class_name)
+ else:
+ creator = DefaultCreator(inventory, self.coreV1_api, self.storage_class_name)
+ return self._patch(
+ ccl.CephCluster,
+ 'cephclusters',
+ self.rook_env.cluster_name,
+ creator.add_osds(self.rook_pods, drive_group, matching_hosts)
+ )
+
+ def remove_osds(self, osd_ids: List[str], replace: bool, force: bool, mon_command: Callable) -> str:
+ inventory = self.get_discovered_devices()
+ self.remover = DefaultRemover(
+ self.coreV1_api,
+ self.batchV1_api,
+ self.appsV1_api,
+ osd_ids,
+ replace,
+ force,
+ mon_command,
+ self._patch,
+ self.rook_env,
+ inventory
+ )
+ return self.remover.remove()
+
+ def get_hosts(self) -> List[orchestrator.HostSpec]:
+ ret = []
+ for node in self.nodes.items:
+ spec = orchestrator.HostSpec(
+ node.metadata.name,
+ addr='/'.join([addr.address for addr in node.status.addresses]),
+ labels=[label.split('/')[1] for label in node.metadata.labels if label.startswith('ceph-label')],
+ )
+ ret.append(spec)
+ return ret
+
+ def create_zap_job(self, host: str, path: str) -> None:
+ body = client.V1Job(
+ api_version="batch/v1",
+ metadata=client.V1ObjectMeta(
+ name="rook-ceph-device-zap",
+ namespace=self.rook_env.namespace
+ ),
+ spec=client.V1JobSpec(
+ template=client.V1PodTemplateSpec(
+ spec=client.V1PodSpec(
+ containers=[
+ client.V1Container(
+ name="device-zap",
+ image="rook/ceph:master",
+ command=["bash"],
+ args=["-c", f"ceph-volume raw list {path} && dd if=/dev/zero of=\"{path}\" bs=1M count=1 oflag=direct,dsync || ceph-volume lvm zap --destroy {path}"],
+ env=[
+ client.V1EnvVar(
+ name="ROOK_CEPH_USERNAME",
+ value_from=client.V1EnvVarSource(
+ secret_key_ref=client.V1SecretKeySelector(
+ key="ceph-username",
+ name="rook-ceph-mon"
+ )
+ )
+ ),
+ client.V1EnvVar(
+ name="ROOK_CEPH_SECRET",
+ value_from=client.V1EnvVarSource(
+ secret_key_ref=client.V1SecretKeySelector(
+ key="ceph-secret",
+ name="rook-ceph-mon"
+ )
+ )
+ )
+ ],
+ security_context=client.V1SecurityContext(
+ run_as_user=0,
+ privileged=True
+ ),
+ volume_mounts=[
+ client.V1VolumeMount(
+ mount_path="/etc/ceph",
+ name="ceph-conf-emptydir"
+ ),
+ client.V1VolumeMount(
+ mount_path="/etc/rook",
+ name="rook-config"
+ ),
+ client.V1VolumeMount(
+ mount_path="/dev",
+ name="devices"
+ )
+ ]
+ )
+ ],
+ volumes=[
+ client.V1Volume(
+ name="ceph-conf-emptydir",
+ empty_dir=client.V1EmptyDirVolumeSource()
+ ),
+ client.V1Volume(
+ name="rook-config",
+ empty_dir=client.V1EmptyDirVolumeSource()
+ ),
+ client.V1Volume(
+ name="devices",
+ host_path=client.V1HostPathVolumeSource(
+ path="/dev"
+ )
+ ),
+ ],
+ node_selector={
+ "kubernetes.io/hostname": host
+ },
+ restart_policy="Never"
+ )
+ )
+ )
+ )
+ self.batchV1_api.create_namespaced_job(self.rook_env.namespace, body)
+
+ def rbd_mirror(self, spec: ServiceSpec) -> None:
+ service_id = spec.service_id or "default-rbd-mirror"
+ all_hosts = self.get_hosts()
+ def _create_rbd_mirror() -> crbdm.CephRBDMirror:
+ return crbdm.CephRBDMirror(
+ apiVersion=self.rook_env.api_name,
+ metadata=dict(
+ name=service_id,
+ namespace=self.rook_env.namespace,
+ ),
+ spec=crbdm.Spec(
+ count=spec.placement.count or 1,
+ placement=crbdm.Placement(
+ nodeAffinity=crbdm.NodeAffinity(
+ requiredDuringSchedulingIgnoredDuringExecution=crbdm.RequiredDuringSchedulingIgnoredDuringExecution(
+ nodeSelectorTerms=crbdm.NodeSelectorTermsList(
+ [
+ placement_spec_to_node_selector(spec.placement, all_hosts)
+ ]
+ )
+ )
+ )
+ )
+ )
+ )
+ def _update_rbd_mirror(new: crbdm.CephRBDMirror) -> crbdm.CephRBDMirror:
+ new.spec.count = spec.placement.count or 1
+ new.spec.placement = crbdm.Placement(
+ nodeAffinity=crbdm.NodeAffinity(
+ requiredDuringSchedulingIgnoredDuringExecution=crbdm.RequiredDuringSchedulingIgnoredDuringExecution(
+ nodeSelectorTerms=crbdm.NodeSelectorTermsList(
+ [
+ placement_spec_to_node_selector(spec.placement, all_hosts)
+ ]
+ )
+ )
+ )
+ )
+ return new
+ self._create_or_patch(crbdm.CephRBDMirror, 'cephrbdmirrors', service_id, _update_rbd_mirror, _create_rbd_mirror)
+ def _patch(self, crd: Type, crd_name: str, cr_name: str, func: Callable[[CrdClassT, CrdClassT], CrdClassT]) -> str:
+ current_json = self.rook_api_get(
+ "{}/{}".format(crd_name, cr_name)
+ )
+
+ current = crd.from_json(current_json)
+ new = crd.from_json(current_json) # no deepcopy.
+
+ new = func(current, new)
+
+ patch = list(jsonpatch.make_patch(current_json, new.to_json()))
+
+ log.info('patch for {}/{}: \n{}'.format(crd_name, cr_name, patch))
+
+ if len(patch) == 0:
+ return "No change"
+
+ try:
+ self.rook_api_patch(
+ "{}/{}".format(crd_name, cr_name),
+ body=patch)
+ except ApiException as e:
+ log.exception("API exception: {0}".format(e))
+ raise ApplyException(
+ "Failed to update {}/{}: {}".format(crd_name, cr_name, e))
+
+ return "Success"
+
+ def _create_or_patch(self,
+ crd: Type,
+ crd_name: str,
+ cr_name: str,
+ update_func: Callable[[CrdClassT], CrdClassT],
+ create_func: Callable[[], CrdClassT]) -> str:
+ try:
+ current_json = self.rook_api_get(
+ "{}/{}".format(crd_name, cr_name)
+ )
+ except ApiException as e:
+ if e.status == 404:
+ current_json = None
+ else:
+ raise
+
+ if current_json:
+ new = crd.from_json(current_json) # no deepcopy.
+
+ new = update_func(new)
+
+ patch = list(jsonpatch.make_patch(current_json, new.to_json()))
+
+ log.info('patch for {}/{}: \n{}'.format(crd_name, cr_name, patch))
+
+ if len(patch) == 0:
+ return "No change"
+
+ try:
+ self.rook_api_patch(
+ "{}/{}".format(crd_name, cr_name),
+ body=patch)
+ except ApiException as e:
+ log.exception("API exception: {0}".format(e))
+ raise ApplyException(
+ "Failed to update {}/{}: {}".format(crd_name, cr_name, e))
+ return "Updated"
+ else:
+ new = create_func()
+ with self.ignore_409("{} {} already exists".format(crd_name,
+ cr_name)):
+ self.rook_api_post("{}/".format(crd_name),
+ body=new.to_json())
+ return "Created"
+ def get_ceph_image(self) -> str:
+ try:
+ api_response = self.coreV1_api.list_namespaced_pod(self.rook_env.namespace,
+ label_selector="app=rook-ceph-mon",
+ timeout_seconds=10)
+ if api_response.items:
+ return api_response.items[-1].spec.containers[0].image
+ else:
+ raise orchestrator.OrchestratorError(
+ "Error getting ceph image. Cluster without monitors")
+ except ApiException as e:
+ raise orchestrator.OrchestratorError("Error getting ceph image: {}".format(e))
+
+
+ def _execute_blight_job(self, ident_fault: str, on: bool, loc: orchestrator.DeviceLightLoc) -> str:
+ operation_id = str(hash(loc))
+ message = ""
+
+ # job definition
+ job_metadata = client.V1ObjectMeta(name=operation_id,
+ namespace= self.rook_env.namespace,
+ labels={"ident": operation_id})
+ pod_metadata = client.V1ObjectMeta(labels={"ident": operation_id})
+ pod_container = client.V1Container(name="ceph-lsmcli-command",
+ security_context=client.V1SecurityContext(privileged=True),
+ image=self.get_ceph_image(),
+ command=["lsmcli",],
+ args=['local-disk-%s-led-%s' % (ident_fault,'on' if on else 'off'),
+ '--path', loc.path or loc.dev,],
+ volume_mounts=[client.V1VolumeMount(name="devices", mount_path="/dev"),
+ client.V1VolumeMount(name="run-udev", mount_path="/run/udev")])
+ pod_spec = client.V1PodSpec(containers=[pod_container],
+ active_deadline_seconds=30, # Max time to terminate pod
+ restart_policy="Never",
+ node_selector= {"kubernetes.io/hostname": loc.host},
+ volumes=[client.V1Volume(name="devices",
+ host_path=client.V1HostPathVolumeSource(path="/dev")),
+ client.V1Volume(name="run-udev",
+ host_path=client.V1HostPathVolumeSource(path="/run/udev"))])
+ pod_template = client.V1PodTemplateSpec(metadata=pod_metadata,
+ spec=pod_spec)
+ job_spec = client.V1JobSpec(active_deadline_seconds=60, # Max time to terminate job
+ ttl_seconds_after_finished=10, # Alfa. Lifetime after finishing (either Complete or Failed)
+ backoff_limit=0,
+ template=pod_template)
+ job = client.V1Job(api_version="batch/v1",
+ kind="Job",
+ metadata=job_metadata,
+ spec=job_spec)
+
+ # delete previous job if it exists
+ try:
+ try:
+ api_response = self.batchV1_api.delete_namespaced_job(operation_id,
+ self.rook_env.namespace,
+ propagation_policy="Background")
+ except ApiException as e:
+ if e.status != 404: # No problem if the job does not exist
+ raise
+
+ # wait until the job is not present
+ deleted = False
+ retries = 0
+ while not deleted and retries < 10:
+ api_response = self.batchV1_api.list_namespaced_job(self.rook_env.namespace,
+ label_selector="ident=%s" % operation_id,
+ timeout_seconds=10)
+ deleted = not api_response.items
+ if retries > 5:
+ sleep(0.1)
+ retries += 1
+ if retries == 10 and not deleted:
+ raise orchestrator.OrchestratorError(
+ "Light <{}> in <{}:{}> cannot be executed. Cannot delete previous job <{}>".format(
+ on, loc.host, loc.path or loc.dev, operation_id))
+
+ # create the job
+ api_response = self.batchV1_api.create_namespaced_job(self.rook_env.namespace, job)
+
+ # get the result
+ finished = False
+ while not finished:
+ api_response = self.batchV1_api.read_namespaced_job(operation_id,
+ self.rook_env.namespace)
+ finished = api_response.status.succeeded or api_response.status.failed
+ if finished:
+ message = api_response.status.conditions[-1].message
+
+ # get the result of the lsmcli command
+ api_response=self.coreV1_api.list_namespaced_pod(self.rook_env.namespace,
+ label_selector="ident=%s" % operation_id,
+ timeout_seconds=10)
+ if api_response.items:
+ pod_name = api_response.items[-1].metadata.name
+ message = self.coreV1_api.read_namespaced_pod_log(pod_name,
+ self.rook_env.namespace)
+
+ except ApiException as e:
+ log.exception('K8s API failed. {}'.format(e))
+ raise
+
+ # Finally, delete the job.
+ # The job uses <ttl_seconds_after_finished>. This makes that the TTL controller delete automatically the job.
+ # This feature is in Alpha state, so extra explicit delete operations trying to delete the Job has been used strategically
+ try:
+ api_response = self.batchV1_api.delete_namespaced_job(operation_id,
+ self.rook_env.namespace,
+ propagation_policy="Background")
+ except ApiException as e:
+ if e.status != 404: # No problem if the job does not exist
+ raise
+
+ return message
+
+ def blink_light(self, ident_fault, on, locs):
+ # type: (str, bool, List[orchestrator.DeviceLightLoc]) -> List[str]
+ return [self._execute_blight_job(ident_fault, on, loc) for loc in locs]
+
+def placement_spec_to_node_selector(spec: PlacementSpec, all_hosts: List) -> ccl.NodeSelectorTermsItem:
+ all_hostnames = [hs.hostname for hs in all_hosts]
+ res = ccl.NodeSelectorTermsItem(matchExpressions=ccl.MatchExpressionsList())
+ if spec.host_pattern and spec.host_pattern != "*":
+ raise RuntimeError("The Rook orchestrator only supports a host_pattern of * for placements")
+ if spec.label:
+ res.matchExpressions.append(
+ ccl.MatchExpressionsItem(
+ key="ceph-label/" + spec.label,
+ operator="Exists"
+ )
+ )
+ if spec.hosts:
+ host_list = [h.hostname for h in spec.hosts if h.hostname in all_hostnames]
+ res.matchExpressions.append(
+ ccl.MatchExpressionsItem(
+ key="kubernetes.io/hostname",
+ operator="In",
+ values=ccl.CrdObjectList(host_list)
+ )
+ )
+ if spec.host_pattern == "*" or (not spec.label and not spec.hosts and not spec.host_pattern):
+ res.matchExpressions.append(
+ ccl.MatchExpressionsItem(
+ key="kubernetes.io/hostname",
+ operator="Exists",
+ )
+ )
+ return res
+
+def node_selector_to_placement_spec(node_selector: ccl.NodeSelectorTermsItem) -> PlacementSpec:
+ res = PlacementSpec()
+ for expression in node_selector.matchExpressions:
+ if expression.key.startswith("ceph-label/"):
+ res.label = expression.key.split('/')[1]
+ elif expression.key == "kubernetes.io/hostname":
+ if expression.operator == "Exists":
+ res.host_pattern = "*"
+ elif expression.operator == "In":
+ res.hosts = [HostPlacementSpec(hostname=value, network='', name='')for value in expression.values]
+ return res
generated by cgit v1.2.3 (git 2.39.1) at 2024-08-27 13:42:55 +0000