mgr/cephadm: extend support for drivegroups

qa/tasks/cephadm.py

@@ -576,13 +576,14 @@ def ceph_osds(ctx, config):
             devs = devs_by_remote[remote]
             assert devs   ## FIXME ##
             dev = devs.pop()
+            short_dev = dev.replace('/dev/', '')
             log.info('Deploying %s on %s with %s...' % (
                 osd, remote.shortname, dev))
             _shell(ctx, cluster_name, remote, [
                 'ceph-volume', 'lvm', 'zap', dev])
             _shell(ctx, cluster_name, remote, [
                 'ceph', 'orchestrator', 'osd', 'create',
-                remote.shortname + ':' + dev
+                remote.shortname + ':' + short_dev
             ])
             ctx.daemons.register_daemon(
                 remote, 'osd', id_,

src/pybind/mgr/ansible/module.py

@@ -348,14 +348,26 @@ def get_inventory(self, node_filter=None, refresh=False):
 
         return op
 
-    def create_osds(self, drive_group):
+    def create_osds(self, drive_groups):
         """Create one or more OSDs within a single Drive Group.
         If no host provided the operation affects all the host in the OSDS role
 
 
-        :param drive_group: (ceph.deployment.drive_group.DriveGroupSpec),
+        :param drive_groups: List[(ceph.deployment.drive_group.DriveGroupSpec)],
                             Drive group with the specification of drives to use
+
+        Caveat: Currently limited to a single DriveGroup.
+        The orchestrator_cli expects a single completion which
+        ideally represents a set of operations. This orchestrator
+        doesn't support this notion, yet. Hence it's only accepting
+        a single DriveGroup for now.
+            You can work around it by invoking:
+
+        $: ceph orchestrator osd create -i <dg.file>
+
+        multiple times. The drivegroup file must only contain one spec at a time.
         """
+        drive_group = drive_groups[0]
 
         # Transform drive group specification to Ansible playbook parameters
         host, osd_spec = dg_2_ansible(drive_group)

src/pybind/mgr/cephadm/Vagrantfile

@@ -17,8 +17,8 @@ Vagrant.configure("2") do |config|
     config.vm.define "osd#{i}" do |osd|
       osd.vm.hostname = "osd#{i}"
       osd.vm.provider :libvirt do |libvirt|
-        libvirt.storage :file, :size => '5G'
-        libvirt.storage :file, :size => '5G'
+        libvirt.storage :file, :size => '20G'
+        libvirt.storage :file, :size => '20G'
       end
     end
   end

src/pybind/mgr/cephadm/module.py

@@ -21,7 +21,10 @@
 import shutil
 import subprocess
 
-from ceph.deployment import inventory
+from ceph.deployment import inventory, translate
+from ceph.deployment.drive_group import DriveGroupSpecs
+from ceph.deployment.drive_selection import selector
+
 from mgr_module import MgrModule
 import mgr_util
 import orchestrator
@@ -1287,15 +1290,53 @@ def get_osd_uuid_map(self):
             r[str(o['osd'])] = o['uuid']
         return r
 
-    @async_completion
-    def _create_osd(self, all_hosts_, drive_group):
-        all_hosts = orchestrator.InventoryNode.get_host_names(all_hosts_)
-        assert len(drive_group.hosts(all_hosts)) == 1
-        assert len(drive_group.data_devices.paths) > 0
-        assert all(map(lambda p: isinstance(p, six.string_types),
-            drive_group.data_devices.paths))
-
-        host = drive_group.hosts(all_hosts)[0]
+    def call_inventory(self, hosts, drive_groups):
+        def call_create(inventory):
+            return self._prepare_deployment(hosts, drive_groups, inventory)
+
+        return self.get_inventory().then(call_create)
+
+    def create_osds(self, drive_groups):
+        return self.get_hosts().then(lambda hosts: self.call_inventory(hosts, drive_groups))
+
+    def _prepare_deployment(self,
+                            all_hosts,  # type: List[orchestrator.InventoryNode]
+                            drive_groups,  # type: List[DriveGroupSpecs]
+                            inventory_list  # type: List[orchestrator.InventoryNode]
+                            ):
+        # type: (...) -> orchestrator.Completion
+
+        for drive_group in drive_groups:
+            self.log.info("Processing DriveGroup {}".format(drive_group))
+            # 1) use fn_filter to determine matching_hosts
+            matching_hosts = drive_group.hosts([x.name for x in all_hosts])
+            # 2) Map the inventory to the InventoryNode object
+            # FIXME: lazy-load the inventory from a InventoryNode object;
+            #        this would save one call to the inventory(at least externally)
+
+            def _find_inv_for_host(hostname, inventory_list):
+                # This is stupid and needs to be loaded with the host
+                for _inventory in inventory_list:
+                    if _inventory.name == hostname:
+                        return _inventory
+                raise OrchestratorError("No inventory found for host: {}".format(hostname))
+
+            cmds = []
+            # 3) iterate over matching_host and call DriveSelection and to_ceph_volume
+            for host in matching_hosts:
+                inventory_for_host = _find_inv_for_host(host, inventory_list)
+                drive_selection = selector.DriveSelection(drive_group, inventory_for_host.devices)
+                cmd = translate.to_ceph_volume(drive_group, drive_selection).run()
+                if not cmd:
+                    self.log.info("No data_devices, skipping DriveGroup: {}".format(drive_group.name))
+                    continue
+                cmds.append((host, cmd))
+
+        return self._create_osd(cmds)
+
+    @async_map_completion
+    def _create_osd(self, host, cmd):
+
         self._require_hosts(host)
 
         # get bootstrap key
@@ -1314,20 +1355,16 @@ def _create_osd(self, all_hosts_, drive_group):
             'keyring': keyring,
         })
 
-        devices = drive_group.data_devices.paths
-        for device in devices:
-            out, err, code = self._run_cephadm(
-                host, 'osd', 'ceph-volume',
-                [
-                    '--config-and-keyring', '-',
-                    '--',
-                    'lvm', 'prepare',
-                    "--cluster-fsid", self._cluster_fsid,
-                    "--{}".format(drive_group.objectstore),
-                    "--data", device,
-                ],
-                stdin=j)
-            self.log.debug('ceph-volume prepare: %s' % out)
+        before_osd_uuid_map = self.get_osd_uuid_map()
+
+        split_cmd = cmd.split(' ')
+        _cmd = ['--config-and-keyring', '-', '--']
+        _cmd.extend(split_cmd)
+        out, err, code = self._run_cephadm(
+            host, 'osd', 'ceph-volume',
+            _cmd,
+            stdin=j)
+        self.log.debug('ceph-volume prepare: %s' % out)
 
         # check result
         out, err, code = self._run_cephadm(
@@ -1346,8 +1383,8 @@ def _create_osd(self, all_hosts_, drive_group):
                 if osd['tags']['ceph.cluster_fsid'] != fsid:
                     self.log.debug('mismatched fsid, skipping %s' % osd)
                     continue
-                if len(list(set(devices) & set(osd['devices']))) == 0 and osd.get('lv_path') not in devices:
-                    self.log.debug('mismatched devices, skipping %s' % osd)
+                if osd_id in before_osd_uuid_map:
+                    # this osd existed before we ran prepare
                     continue
                 if osd_id not in osd_uuid_map:
                     self.log.debug('osd id %d does not exist in cluster' % osd_id)
@@ -1365,22 +1402,6 @@ def _create_osd(self, all_hosts_, drive_group):
 
         return "Created osd(s) on host '{}'".format(host)
 
-    def create_osds(self, drive_group):
-        """
-        Create a new osd.
-
-        The orchestrator CLI currently handles a narrow form of drive
-        specification defined by a single block device using bluestore.
-
-        :param drive_group: osd specification
-
-        TODO:
-          - support full drive_group specification
-          - support batch creation
-        """
-
-        return self.get_hosts().then(lambda hosts: self._create_osd(hosts, drive_group))
-
     @with_services('osd')
     def remove_osds(self, osd_ids, services):
         # type: (List[str], List[orchestrator.ServiceDescription]) -> AsyncCompletion
@@ -2093,4 +2114,3 @@ def assign_nodes(self):
             logger.info('Assigning nodes to spec: {}'.format(candidates))
             self.spec.placement.set_hosts(candidates)
             return None
-

src/pybind/mgr/cephadm/tests/test_cephadm.py

@@ -124,9 +124,9 @@ def test_mgr_update(self, _send_command, _get_connection, cephadm_module):
     @mock.patch("cephadm.module.CephadmOrchestrator._get_connection")
     def test_create_osds(self, _send_command, _get_connection, cephadm_module):
         with self._with_host(cephadm_module, 'test'):
-            dg = DriveGroupSpec('test', DeviceSelection(paths=['']))
-            c = cephadm_module.create_osds(dg)
-            assert wait(cephadm_module, c) == "Created osd(s) on host 'test'"
+            dg = DriveGroupSpec('test', data_devices=DeviceSelection(paths=['']))
+            c = cephadm_module.create_osds([dg])
+            assert wait(cephadm_module, c) == ["Created osd(s) on host 'test'"]
 
     @mock.patch("cephadm.module.CephadmOrchestrator._run_cephadm", _run_cephadm(
         json.dumps([

src/pybind/mgr/orchestrator_cli/module.py

@@ -1,6 +1,7 @@
 import datetime
 import errno
 import json
+import yaml
 from functools import wraps
 
 from ceph.deployment.inventory import Device
@@ -15,7 +16,7 @@
 
 
 from ceph.deployment.drive_group import DriveGroupSpec, DriveGroupValidationError, \
-    DeviceSelection
+    DeviceSelection, DriveGroupSpecs
 from mgr_module import MgrModule, CLICommand, HandleCommandResult
 
 import orchestrator
@@ -342,13 +343,14 @@ def _create_osd(self, svc_arg=None, inbuf=None):
 
         usage = """
 Usage:
-  ceph orchestrator osd create -i <json_file>
+  ceph orchestrator osd create -i <json_file/yaml_file>
   ceph orchestrator osd create host:device1,device2,...
 """
 
         if inbuf:
             try:
-                drive_group = DriveGroupSpec.from_json(json.loads(inbuf))
+                dgs = DriveGroupSpecs(yaml.load(inbuf))
+                drive_groups = dgs.drive_groups
             except ValueError as e:
                 msg = 'Failed to read JSON input: {}'.format(str(e)) + usage
                 return HandleCommandResult(-errno.EINVAL, stderr=msg)
@@ -362,14 +364,13 @@ def _create_osd(self, svc_arg=None, inbuf=None):
                 return HandleCommandResult(-errno.EINVAL, stderr=msg)
 
             devs = DeviceSelection(paths=block_devices)
-            drive_group = DriveGroupSpec(node_name, data_devices=devs)
+            drive_groups = [DriveGroupSpec(node_name, data_devices=devs)]
         else:
             return HandleCommandResult(-errno.EINVAL, stderr=usage)
 
-        completion = self.create_osds(drive_group)
+        completion = self.create_osds(drive_groups)
         self._orchestrator_wait([completion])
         orchestrator.raise_if_exception(completion)
-        self.log.warning(str(completion.result))
         return HandleCommandResult(stdout=completion.result_str())
 
     @orchestrator._cli_write_command(

src/pybind/mgr/requirements.txt

@@ -3,4 +3,5 @@ mock; python_version <= '3.3'
 ipaddress; python_version < '3.3'
 ../../python-common
 kubernetes
-requests-mock
+requests-mock
+pyyaml

src/pybind/mgr/rook/module.py

@@ -369,8 +369,22 @@ def update_nfs(self, spec):
             mgr=self
         )
 
-    def create_osds(self, drive_group):
-        # type: (DriveGroupSpec) -> RookCompletion
+    def create_osds(self, drive_groups):
+        # type: (List[DriveGroupSpec]) -> RookCompletion
+        """ Creates OSDs from a drive group specification.
+
+        Caveat: Currently limited to a single DriveGroup.
+        The orchestrator_cli expects a single completion which
+        ideally represents a set of operations. This orchestrator
+        doesn't support this notion, yet. Hence it's only accepting
+        a single DriveGroup for now.
+        You can work around it by invoking:
+
+        $: ceph orchestrator osd create -i <dg.file>
+
+        multiple times. The drivegroup file must only contain one spec at a time.
+        """
+        drive_group = drive_groups[0]
 
         targets = []  # type: List[str]
         if drive_group.data_devices:

src/pybind/mgr/test_orchestrator/module.py

@@ -13,6 +13,7 @@
 import six
 
 from ceph.deployment import inventory
+from ceph.deployment.drive_group import DriveGroupSpec
 from mgr_module import CLICommand, HandleCommandResult
 from mgr_module import MgrModule
 
@@ -183,8 +184,23 @@ def describe_service(self, service_type=None, service_id=None, node_name=None, r
 
         return result
 
-    def create_osds(self, drive_group):
-        # type: (orchestrator.DriveGroupSpec) -> TestCompletion
+    def create_osds(self, drive_groups):
+        # type: (List[DriveGroupSpec]) -> TestCompletion
+        """ Creates OSDs from a drive group specification.
+
+        Caveat: Currently limited to a single DriveGroup.
+        The orchestrator_cli expects a single completion which
+        ideally represents a set of operations. This orchestrator
+        doesn't support this notion, yet. Hence it's only accepting
+        a single DriveGroup for now.
+        You can work around it by invoking:
+
+        $: ceph orchestrator osd create -i <dg.file>
+
+        multiple times. The drivegroup file must only contain one spec at a time.
+        """
+        drive_group = drive_groups[0]
+
         def run(all_hosts):
             drive_group.validate(orchestrator.InventoryNode.get_host_names(all_hosts))
         return self.get_hosts().then(run).then(