diff options
Diffstat (limited to 'src/pybind/mgr/rook')
| -rw-r--r-- | src/pybind/mgr/rook/__init__.py | 2 | ||||
| -rw-r--r-- | src/pybind/mgr/rook/module.py | 469 | ||||
| -rw-r--r-- | src/pybind/mgr/rook/rook_cluster.py | 442 |
3 files changed, 913 insertions, 0 deletions
diff --git a/src/pybind/mgr/rook/__init__.py b/src/pybind/mgr/rook/__init__.py new file mode 100644 index 00000000..b9f08fc8 --- /dev/null +++ b/src/pybind/mgr/rook/__init__.py @@ -0,0 +1,2 @@ + +from .module import RookOrchestrator diff --git a/src/pybind/mgr/rook/module.py b/src/pybind/mgr/rook/module.py new file mode 100644 index 00000000..4b49d681 --- /dev/null +++ b/src/pybind/mgr/rook/module.py @@ -0,0 +1,469 @@ +import threading +import functools +import os +import uuid +try: + from typing import List +except ImportError: + pass # just for type checking + +try: + from kubernetes import client, config + from kubernetes.client.rest import ApiException + + kubernetes_imported = True +except ImportError: + kubernetes_imported = False + client = None + config = None + +from mgr_module import MgrModule +import orchestrator + +from .rook_cluster import RookCluster + + +all_completions = [] + + +class RookReadCompletion(orchestrator.ReadCompletion): + """ + All reads are simply API calls: avoid spawning + huge numbers of threads by just running them + inline when someone calls wait() + """ + + def __init__(self, cb): + super(RookReadCompletion, self).__init__() + self.cb = cb + self._result = None + self._complete = False + + self.message = "<read op>" + + # XXX hacky global + global all_completions + all_completions.append(self) + + @property + def result(self): + return self._result + + @property + def is_complete(self): + return self._complete + + def execute(self): + self._result = self.cb() + self._complete = True + + +class RookWriteCompletion(orchestrator.WriteCompletion): + """ + Writes are a two-phase thing, firstly sending + the write to the k8s API (fast) and then waiting + for the corresponding change to appear in the + Ceph cluster (slow) + """ + # XXX kubernetes bindings call_api already usefully has + # a completion= param that uses threads. Maybe just + # use that? + def __init__(self, execute_cb, complete_cb, message): + super(RookWriteCompletion, self).__init__() + self.execute_cb = execute_cb + self.complete_cb = complete_cb + + # Executed means I executed my k8s API call, it may or may + # not have succeeded + self.executed = False + + # Result of k8s API call, this is set if executed==True + self._result = None + + self.effective = False + + self.id = str(uuid.uuid4()) + + self.message = message + + self.error = None + + # XXX hacky global + global all_completions + all_completions.append(self) + + def __str__(self): + return self.message + + @property + def result(self): + return self._result + + @property + def is_persistent(self): + return (not self.is_errored) and self.executed + + @property + def is_effective(self): + return self.effective + + @property + def is_errored(self): + return self.error is not None + + def execute(self): + if not self.executed: + self._result = self.execute_cb() + self.executed = True + + if not self.effective: + # TODO: check self.result for API errors + if self.complete_cb is None: + self.effective = True + else: + self.effective = self.complete_cb() + + +def deferred_read(f): + """ + Decorator to make RookOrchestrator methods return + a completion object that executes themselves. + """ + + @functools.wraps(f) + def wrapper(*args, **kwargs): + return RookReadCompletion(lambda: f(*args, **kwargs)) + + return wrapper + + +class RookEnv(object): + def __init__(self): + # POD_NAMESPACE already exist for Rook 0.9 + self.namespace = os.environ.get('POD_NAMESPACE', 'rook-ceph') + + # ROOK_CEPH_CLUSTER_CRD_NAME is new is Rook 1.0 + self.cluster_name = os.environ.get('ROOK_CEPH_CLUSTER_CRD_NAME', self.namespace) + + self.operator_namespace = os.environ.get('ROOK_OPERATOR_NAMESPACE', "rook-ceph-system") + self.crd_version = os.environ.get('ROOK_CEPH_CLUSTER_CRD_VERSION', 'v1') + self.api_name = "ceph.rook.io/" + self.crd_version + + def api_version_match(self): + return self.crd_version == 'v1' + + def has_namespace(self): + return 'POD_NAMESPACE' in os.environ + + +class RookOrchestrator(MgrModule, orchestrator.Orchestrator): + MODULE_OPTIONS = [ + # TODO: configure k8s API addr instead of assuming local + ] + + def wait(self, completions): + self.log.info("wait: completions={0}".format(completions)) + + incomplete = False + + # Our `wait` implementation is very simple because everything's + # just an API call. + for c in completions: + if not isinstance(c, RookReadCompletion) and \ + not isinstance(c, RookWriteCompletion): + raise TypeError( + "wait() requires list of completions, not {0}".format( + c.__class__ + )) + + if c.is_complete: + continue + + try: + c.execute() + except Exception as e: + if not isinstance(e, orchestrator.OrchestratorError): + self.log.exception("Completion {0} threw an exception:".format( + c.message + )) + c.exception = e + c._complete = True + + if not c.is_complete: + incomplete = True + + return not incomplete + + @staticmethod + def can_run(): + if not kubernetes_imported: + return False, "`kubernetes` python module not found" + if not RookEnv().api_version_match(): + return False, "Rook version unsupported." + return True, '' + + def available(self): + if not kubernetes_imported: + return False, "`kubernetes` python module not found" + elif not self._rook_env.has_namespace(): + return False, "ceph-mgr not running in Rook cluster" + + try: + self.k8s.list_namespaced_pod(self._rook_env.cluster_name) + except ApiException as e: + return False, "Cannot reach Kubernetes API: {}".format(e) + else: + return True, "" + + def __init__(self, *args, **kwargs): + super(RookOrchestrator, self).__init__(*args, **kwargs) + + self._initialized = threading.Event() + self._k8s = None + self._rook_cluster = None + self._rook_env = RookEnv() + + self._shutdown = threading.Event() + + def shutdown(self): + self._shutdown.set() + + @property + def k8s(self): + self._initialized.wait() + return self._k8s + + @property + def rook_cluster(self): + # type: () -> RookCluster + self._initialized.wait() + return self._rook_cluster + + def serve(self): + # For deployed clusters, we should always be running inside + # a Rook cluster. For development convenience, also support + # running outside (reading ~/.kube config) + + if self._rook_env.cluster_name: + config.load_incluster_config() + cluster_name = self._rook_env.cluster_name + else: + self.log.warning("DEVELOPMENT ONLY: Reading kube config from ~") + config.load_kube_config() + + cluster_name = "rook-ceph" + + # So that I can do port forwarding from my workstation - jcsp + from kubernetes.client import configuration + configuration.verify_ssl = False + + self._k8s = client.CoreV1Api() + + try: + # XXX mystery hack -- I need to do an API call from + # this context, or subsequent API usage from handle_command + # fails with SSLError('bad handshake'). Suspect some kind of + # thread context setup in SSL lib? + self._k8s.list_namespaced_pod(cluster_name) + except ApiException: + # Ignore here to make self.available() fail with a proper error message + pass + + self._rook_cluster = RookCluster( + self._k8s, + self._rook_env) + + self._initialized.set() + + while not self._shutdown.is_set(): + # XXX hack (or is it?) to kick all completions periodically, + # in case we had a caller that wait()'ed on them long enough + # to get persistence but not long enough to get completion + + global all_completions + self.wait(all_completions) + all_completions = [c for c in all_completions if not c.is_complete] + + self._shutdown.wait(5) + + # TODO: watch Rook for config changes to complain/update if + # things look a bit out of sync? + + @deferred_read + def get_inventory(self, node_filter=None, refresh=False): + node_list = None + if node_filter and node_filter.nodes: + # Explicit node list + node_list = node_filter.nodes + elif node_filter and node_filter.labels: + # TODO: query k8s API to resolve to node list, and pass + # it into RookCluster.get_discovered_devices + raise NotImplementedError() + + devs = self.rook_cluster.get_discovered_devices(node_list) + + result = [] + for node_name, node_devs in devs.items(): + devs = [] + for d in node_devs: + dev = orchestrator.InventoryDevice() + + # XXX CAUTION! https://github.com/rook/rook/issues/1716 + # Passing this through for the sake of completeness but it + # is not trustworthy! + dev.blank = d['empty'] + dev.type = 'hdd' if d['rotational'] else 'ssd' + dev.id = d['name'] + dev.size = d['size'] + + if d['filesystem'] == "" and not d['rotational']: + # Empty or partitioned SSD + partitioned_space = sum( + [p['size'] for p in d['Partitions']]) + dev.metadata_space_free = max(0, d[ + 'size'] - partitioned_space) + + devs.append(dev) + + result.append(orchestrator.InventoryNode(node_name, devs)) + + return result + + @deferred_read + def describe_service(self, service_type=None, service_id=None, node_name=None, refresh=False): + + if service_type not in ("mds", "osd", "mgr", "mon", "nfs", None): + raise orchestrator.OrchestratorValidationError(service_type + " unsupported") + + pods = self.rook_cluster.describe_pods(service_type, service_id, node_name) + + result = [] + for p in pods: + sd = orchestrator.ServiceDescription() + sd.nodename = p['nodename'] + sd.container_id = p['name'] + sd.service_type = p['labels']['app'].replace('rook-ceph-', '') + + if sd.service_type == "osd": + sd.service_instance = "%s" % p['labels']["ceph-osd-id"] + elif sd.service_type == "mds": + sd.service = p['labels']['rook_file_system'] + pfx = "{0}-".format(sd.service) + sd.service_instance = p['labels']['ceph_daemon_id'].replace(pfx, '', 1) + elif sd.service_type == "mon": + sd.service_instance = p['labels']["mon"] + elif sd.service_type == "mgr": + sd.service_instance = p['labels']["mgr"] + elif sd.service_type == "nfs": + sd.service = p['labels']['ceph_nfs'] + sd.service_instance = p['labels']['instance'] + sd.rados_config_location = self.rook_cluster.get_nfs_conf_url(sd.service, sd.service_instance) + elif sd.service_type == "rgw": + sd.service = p['labels']['rgw'] + sd.service_instance = p['labels']['ceph_daemon_id'] + else: + # Unknown type -- skip it + continue + + result.append(sd) + + return result + + def _service_add_decorate(self, typename, spec, func): + return RookWriteCompletion(lambda: func(spec), None, + "Creating {0} services for {1}".format(typename, spec.name)) + + def add_stateless_service(self, service_type, spec): + # assert isinstance(spec, orchestrator.StatelessServiceSpec) + if service_type == "mds": + return self._service_add_decorate("Filesystem", spec, + self.rook_cluster.add_filesystem) + elif service_type == "rgw" : + return self._service_add_decorate("RGW", spec, + self.rook_cluster.add_objectstore) + elif service_type == "nfs" : + return self._service_add_decorate("NFS", spec, + self.rook_cluster.add_nfsgw) + else: + raise NotImplementedError(service_type) + + def remove_stateless_service(self, service_type, service_id): + return RookWriteCompletion( + lambda: self.rook_cluster.rm_service(service_type, service_id), None, + "Removing {0} services for {1}".format(service_type, service_id)) + + def update_mons(self, num, hosts): + if hosts: + raise RuntimeError("Host list is not supported by rook.") + + return RookWriteCompletion( + lambda: self.rook_cluster.update_mon_count(num), None, + "Updating mon count to {0}".format(num)) + + def update_stateless_service(self, svc_type, spec): + # only nfs is currently supported + if svc_type != "nfs": + raise NotImplementedError(svc_type) + + num = spec.count + return RookWriteCompletion( + lambda: self.rook_cluster.update_nfs_count(spec.name, num), None, + "Updating NFS server count in {0} to {1}".format(spec.name, num)) + + def create_osds(self, drive_group, all_hosts): + # type: (orchestrator.DriveGroupSpec, List[str]) -> RookWriteCompletion + + assert len(drive_group.hosts(all_hosts)) == 1 + targets = [] + if drive_group.data_devices: + targets += drive_group.data_devices.paths + if drive_group.data_directories: + targets += drive_group.data_directories + + if not self.rook_cluster.node_exists(drive_group.hosts(all_hosts)[0]): + raise RuntimeError("Node '{0}' is not in the Kubernetes " + "cluster".format(drive_group.hosts(all_hosts))) + + # Validate whether cluster CRD can accept individual OSD + # creations (i.e. not useAllDevices) + if not self.rook_cluster.can_create_osd(): + raise RuntimeError("Rook cluster configuration does not " + "support OSD creation.") + + def execute(): + return self.rook_cluster.add_osds(drive_group, all_hosts) + + def is_complete(): + # Find OSD pods on this host + pod_osd_ids = set() + pods = self._k8s.list_namespaced_pod(self._rook_env.namespace, + label_selector="rook_cluster={},app=rook-ceph-osd".format(self._rook_env.cluster_name), + field_selector="spec.nodeName={0}".format( + drive_group.hosts(all_hosts)[0] + )).items + for p in pods: + pod_osd_ids.add(int(p.metadata.labels['ceph-osd-id'])) + + self.log.debug('pod_osd_ids={0}'.format(pod_osd_ids)) + + found = [] + osdmap = self.get("osd_map") + for osd in osdmap['osds']: + osd_id = osd['osd'] + if osd_id not in pod_osd_ids: + continue + + metadata = self.get_metadata('osd', "%s" % osd_id) + if metadata and metadata['devices'] in targets: + found.append(osd_id) + else: + self.log.info("ignoring osd {0} {1}".format( + osd_id, metadata['devices'] + )) + + return found is not None + + return RookWriteCompletion(execute, is_complete, + "Creating OSD on {0}:{1}".format( + drive_group.hosts(all_hosts)[0], targets + )) diff --git a/src/pybind/mgr/rook/rook_cluster.py b/src/pybind/mgr/rook/rook_cluster.py new file mode 100644 index 00000000..ef404075 --- /dev/null +++ b/src/pybind/mgr/rook/rook_cluster.py @@ -0,0 +1,442 @@ +""" +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 logging +import json +from contextlib import contextmanager + +from six.moves.urllib.parse import urljoin # pylint: disable=import-error + +# Optional kubernetes imports to enable MgrModule.can_run +# to behave cleanly. +try: + from kubernetes.client.rest import ApiException +except ImportError: + ApiException = None + +try: + import orchestrator + from rook.module import RookEnv + from typing import List +except ImportError: + pass # just used for type checking. + + +log = logging.getLogger(__name__) + + +class ApplyException(Exception): + """ + For failures to update the Rook CRDs, usually indicating + some kind of interference between our attempted update + and other conflicting activity. + """ + + +class RookCluster(object): + def __init__(self, k8s, rook_env): + self.rook_env = rook_env # type: RookEnv + self.k8s = k8s + + def rook_url(self, path): + 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, path, **kwargs): + full_path = self.rook_url(path) + log.debug("[%s] %s" % (verb, full_path)) + + return self.k8s.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, **kwargs): + return self.rook_api_call("GET", path, **kwargs) + + def rook_api_delete(self, path): + return self.rook_api_call("DELETE", path) + + def rook_api_patch(self, path, **kwargs): + return self.rook_api_call("PATCH", path, + header_params={"Content-Type": "application/json-patch+json"}, + **kwargs) + + def rook_api_post(self, path, **kwargs): + return self.rook_api_call("POST", path, **kwargs) + + def get_discovered_devices(self, nodenames=None): + # TODO: replace direct k8s calls with Rook API calls + # when they're implemented + label_selector = "app=rook-discover" + if nodenames is not None: + # FIXME: is there a practical or official limit on the + # number of entries in a label selector + label_selector += ", rook.io/node in ({0})".format( + ", ".join(nodenames)) + + try: + result = self.k8s.list_namespaced_config_map( + self.rook_env.operator_namespace, + label_selector=label_selector) + except ApiException as e: + log.exception("Failed to fetch device metadata: {0}".format(e)) + raise + + nodename_to_devices = {} + for i in result.items: + drives = json.loads(i.data['devices']) + nodename_to_devices[i.metadata.labels['rook.io/node']] = drives + + return nodename_to_devices + + def get_nfs_conf_url(self, nfs_cluster, instance): + # + # 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, service_id, nodename): + # Go query the k8s API about deployment, containers related to this + # filesystem + + # Inspect the Rook YAML, to decide whether this filesystem + # is Ceph-managed or Rook-managed + # TODO: extend Orchestrator interface to describe whether FS + # is manageable by us or not + + # 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 name> + # rook_file_system=<self.fs_name> + + label_filter = "rook_cluster={0}".format(self.rook_env.cluster_name) + if service_type != None: + label_filter += ",app=rook-ceph-{0}".format(service_type) + if service_id != None: + if service_type == "mds": + label_filter += ",rook_file_system={0}".format(service_id) + elif service_type == "osd": + # Label added in https://github.com/rook/rook/pull/1698 + label_filter += ",ceph-osd-id={0}".format(service_id) + elif service_type == "mon": + # label like mon=rook-ceph-mon0 + label_filter += ",mon={0}".format(service_id) + elif service_type == "mgr": + label_filter += ",mgr={0}".format(service_id) + elif service_type == "nfs": + label_filter += ",ceph_nfs={0}".format(service_id) + elif service_type == "rgw": + # TODO: rgw + pass + + field_filter = "" + if nodename != None: + field_filter = "spec.nodeName={0}".format(nodename) + + pods = self.k8s.list_namespaced_pod( + self.rook_env.namespace, + label_selector=label_filter, + field_selector=field_filter) + + # import json + # print json.dumps(pods.items[0]) + + pods_summary = [] + + for p in pods.items: + d = p.to_dict() + # p['metadata']['creationTimestamp'] + # p['metadata']['nodeName'] + pods_summary.append({ + "name": d['metadata']['name'], + "nodename": d['spec']['node_name'], + "labels": d['metadata']['labels'] + }) + pass + + return pods_summary + + @contextmanager + def ignore_409(self, what): + try: + yield + except ApiException as e: + if e.status == 409: + # Idempotent, succeed. + log.info("{} already exists".format(what)) + else: + raise + + def add_filesystem(self, spec): + # TODO use spec.placement + # TODO warn if spec.extended has entries we don't kow how + # to action. + + rook_fs = { + "apiVersion": self.rook_env.api_name, + "kind": "CephFilesystem", + "metadata": { + "name": spec.name, + "namespace": self.rook_env.namespace + }, + "spec": { + "onlyManageDaemons": True, + "metadataServer": { + "activeCount": spec.count, + "activeStandby": True + + } + } + } + + with self.ignore_409("CephFilesystem '{0}' already exists".format(spec.name)): + self.rook_api_post("cephfilesystems/", body=rook_fs) + + def add_nfsgw(self, spec): + # TODO use spec.placement + # TODO warn if spec.extended has entries we don't kow how + # to action. + + rook_nfsgw = { + "apiVersion": self.rook_env.api_name, + "kind": "CephNFS", + "metadata": { + "name": spec.name, + "namespace": self.rook_env.namespace + }, + "spec": { + "rados": { + "pool": spec.extended["pool"] + }, + "server": { + "active": spec.count, + } + } + } + + if "namespace" in spec.extended: + rook_nfsgw["spec"]["rados"]["namespace"] = spec.extended["namespace"] + + with self.ignore_409("NFS cluster '{0}' already exists".format(spec.name)): + self.rook_api_post("cephnfses/", body=rook_nfsgw) + + def add_objectstore(self, spec): + rook_os = { + "apiVersion": self.rook_env.api_name, + "kind": "CephObjectStore", + "metadata": { + "name": spec.name, + "namespace": self.rook_env.namespace + }, + "spec": { + "metadataPool": { + "failureDomain": "host", + "replicated": { + "size": 1 + } + }, + "dataPool": { + "failureDomain": "osd", + "replicated": { + "size": 1 + } + }, + "gateway": { + "type": "s3", + "port": 80, + "instances": 1, + "allNodes": False + } + } + } + + with self.ignore_409("CephObjectStore '{0}' already exists".format(spec.name)): + self.rook_api_post("cephobjectstores/", body=rook_os) + + def rm_service(self, service_type, service_id): + assert service_type in ("mds", "rgw", "nfs") + + if service_type == "mds": + rooktype = "cephfilesystems" + elif service_type == "rgw": + rooktype = "cephobjectstores" + elif service_type == "nfs": + rooktype = "cephnfses" + + objpath = "{0}/{1}".format(rooktype, service_id) + + try: + self.rook_api_delete(objpath) + except ApiException as e: + if e.status == 404: + log.info("{0} service '{1}' does not exist".format(service_type, service_id)) + # Idempotent, succeed. + else: + raise + + def can_create_osd(self): + 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): + try: + self.k8s.read_node(node_name, exact=False, export=True) + except ApiException as e: + if e.status == 404: + return False + else: + raise + else: + return True + + def update_mon_count(self, newcount): + patch = [{"op": "replace", "path": "/spec/mon/count", "value": newcount}] + + try: + self.rook_api_patch( + "cephclusters/{0}".format(self.rook_env.cluster_name), + body=patch) + except ApiException as e: + log.exception("API exception: {0}".format(e)) + raise ApplyException( + "Failed to update mon count in Cluster CRD: {0}".format(e)) + + return "Updated mon count to {0}".format(newcount) + + def update_nfs_count(self, svc_id, newcount): + patch = [{"op": "replace", "path": "/spec/server/active", "value": newcount}] + + try: + self.rook_api_patch( + "cephnfses/{0}".format(svc_id), + body=patch) + except ApiException as e: + log.exception("API exception: {0}".format(e)) + raise ApplyException( + "Failed to update NFS server count for {0}: {1}".format(svc_id, e)) + return "Updated NFS server count for {0} to {1}".format(svc_id, newcount) + + def add_osds(self, drive_group, all_hosts): + # type: (orchestrator.DriveGroupSpec, List[str]) -> str + """ + Rook currently (0.8) can only do single-drive OSDs, so we + treat all drive groups as just a list of individual OSDs. + """ + block_devices = drive_group.data_devices.paths if drive_group.data_devices else None + directories = drive_group.data_directories + + assert drive_group.objectstore in ("bluestore", "filestore") + + # The CRD looks something like this: + # nodes: + # - name: "gravel1.rockery" + # devices: + # - name: "sdb" + # config: + # storeType: bluestore + + current_cluster = self.rook_api_get( + "cephclusters/{0}".format(self.rook_env.cluster_name)) + + patch = [] + + # FIXME: this is all not really atomic, because jsonpatch doesn't + # let us do "test" operations that would check if items with + # matching names were in existing lists. + + if 'nodes' not in current_cluster['spec']['storage']: + patch.append({ + 'op': 'add', 'path': '/spec/storage/nodes', 'value': [] + }) + + current_nodes = current_cluster['spec']['storage'].get('nodes', []) + + if drive_group.hosts(all_hosts)[0] not in [n['name'] for n in current_nodes]: + pd = { "name": drive_group.hosts(all_hosts)[0], + "config": { "storeType": drive_group.objectstore }} + + if block_devices: + pd["devices"] = [{'name': d} for d in block_devices] + if directories: + pd["directories"] = [{'path': p} for p in directories] + + patch.append({ "op": "add", "path": "/spec/storage/nodes/-", "value": pd }) + else: + # Extend existing node + node_idx = None + current_node = None + for i, c in enumerate(current_nodes): + if c['name'] == drive_group.hosts(all_hosts)[0]: + current_node = c + node_idx = i + break + + assert node_idx is not None + assert current_node is not None + + new_devices = list(set(block_devices) - set([d['name'] for d in current_node['devices']])) + for n in new_devices: + patch.append({ + "op": "add", + "path": "/spec/storage/nodes/{0}/devices/-".format(node_idx), + "value": {'name': n} + }) + + new_dirs = list(set(directories) - set(current_node['directories'])) + for p in new_dirs: + patch.append({ + "op": "add", + "path": "/spec/storage/nodes/{0}/directories/-".format(node_idx), + "value": {'path': p} + }) + + if len(patch) == 0: + return "No change" + + try: + self.rook_api_patch( + "cephclusters/{0}".format(self.rook_env.cluster_name), + body=patch) + except ApiException as e: + log.exception("API exception: {0}".format(e)) + raise ApplyException( + "Failed to create OSD entries in Cluster CRD: {0}".format( + e)) + + return "Success" |