Adding upstream version 14.2.21.upstream/14.2.21upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

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"