ceph-volume: use 'no workqueue' options with dmcrypt

CloudFlare engineers made some testing and realized that using
workqueues with encryption on flash devices has a bad effect.

See [1] for details.

With this patch it will make ceph-volume call crypsetup with
`--perf-no_read_workqueue` and `--perf-no_write_workqueue` options
when the device is not a rotational.

[1] https://blog.cloudflare.com/speeding-up-linux-disk-encryption/

Signed-off-by: Guillaume Abrioux <gabrioux@ibm.com>
Co-Authored-by: Stefan Kooman <stefan@kooman.org>
Signed-off-by: Guillaume Abrioux <gabrioux@ibm.com>

ceph.spec.in

@@ -902,6 +902,7 @@ Requires: parted
 Requires: util-linux
 Requires: xfsprogs
 Requires: python%{python3_pkgversion}-setuptools
+Requires: python%{python3_pkgversion}-packaging
 Requires: python%{python3_pkgversion}-ceph-common = %{_epoch_prefix}%{version}-%{release}
 %description volume
 This package contains a tool to deploy OSD with different devices like

src/ceph-volume/ceph_volume/__init__.py

@@ -14,8 +14,9 @@ class UnloadedConfig(object):
     def __getattr__(self, *a):
         raise RuntimeError("No valid ceph configuration file was loaded.")
 
-conf = namedtuple('config', ['ceph', 'cluster', 'verbosity', 'path', 'log_path'])
+conf = namedtuple('config', ['ceph', 'cluster', 'verbosity', 'path', 'log_path', 'dmcrypt_no_workqueue'])
 conf.ceph = UnloadedConfig()
+conf.dmcrypt_no_workqueue = None
 
 __version__ = "1.0.0"
 

src/ceph-volume/ceph_volume/devices/lvm/activate.py

@@ -69,6 +69,8 @@ def activate_bluestore(osd_lvs, no_systemd=False, no_tmpfs=False):
         raise RuntimeError('could not find a bluestore OSD to activate')
 
     is_encrypted = osd_block_lv.tags.get('ceph.encrypted', '0') == '1'
+    if is_encrypted and conf.dmcrypt_no_workqueue is None:
+        encryption_utils.set_dmcrypt_no_workqueue()
     dmcrypt_secret = None
     osd_id = osd_block_lv.tags['ceph.osd_id']
     conf.cluster = osd_block_lv.tags['ceph.cluster_name']

src/ceph-volume/ceph_volume/devices/lvm/batch.py

@@ -256,7 +256,7 @@ class Batch(object):
         )
         parser.add_argument(
             '--dmcrypt',
-            action='store_true',
+            action=arg_validators.DmcryptAction,
             help='Enable device encryption via dm-crypt',
         )
         parser.add_argument(

src/ceph-volume/ceph_volume/devices/lvm/common.py

@@ -73,7 +73,7 @@ common_args = {
         'default': "",
     },
     '--dmcrypt': {
-        'action': 'store_true',
+        'action': arg_validators.DmcryptAction,
         'help': 'Enable device encryption via dm-crypt',
     },
     '--no-systemd': {

src/ceph-volume/ceph_volume/devices/raw/common.py

@@ -46,7 +46,7 @@ def create_parser(prog, description):
     )
     parser.add_argument(
         '--dmcrypt',
-        action='store_true',
+        action=arg_validators.DmcryptAction,
         help='Enable device encryption via dm-crypt',
     )
     parser.add_argument(

src/ceph-volume/ceph_volume/util/arg_validators.py

@@ -4,11 +4,20 @@ import math
 from ceph_volume import terminal, decorators, process
 from ceph_volume.util.device import Device
 from ceph_volume.util import disk
-
+from ceph_volume.util.encryption import set_dmcrypt_no_workqueue
+from ceph_volume import process, conf
 
 def valid_osd_id(val):
     return str(int(val))
 
+class DmcryptAction(argparse._StoreTrueAction):
+    def __init__(self, *args, **kwargs):
+        super(DmcryptAction, self).__init__(*args, **kwargs)
+
+    def __call__(self, *args, **kwargs):
+        set_dmcrypt_no_workqueue()
+        super(DmcryptAction, self).__call__(*args, **kwargs)
+
 class ValidDevice(object):
 
     def __init__(self, as_string=False, gpt_ok=False):

src/ceph-volume/ceph_volume/util/device.py

@@ -121,13 +121,8 @@ class Device(object):
             # check if we are not a device mapper
             if "dm-" not in real_path:
                 self.path = real_path
-        if not sys_info.devices:
-            if self.path:
-                sys_info.devices = disk.get_devices(device=self.path)
-            else:
-                sys_info.devices = disk.get_devices()
-        if sys_info.devices.get(self.path, {}):
-            self.device_nodes = sys_info.devices[self.path]['device_nodes']
+        if not sys_info.devices.get(self.path):
+            sys_info.devices = disk.get_devices()
         self.sys_api = sys_info.devices.get(self.path, {})
         self.partitions = self._get_partitions()
         self.lv_api = None
@@ -143,6 +138,7 @@ class Device(object):
         self._is_lvm_member = None
         self.ceph_device = False
         self._parse()
+        self.device_nodes = sys_info.devices[self.path]['device_nodes']
         self.lsm_data = self.fetch_lsm(with_lsm)
 
         self.available_lvm, self.rejected_reasons_lvm = self._check_lvm_reject_reasons()

src/ceph-volume/ceph_volume/util/encryption.py

@@ -6,10 +6,28 @@ from ceph_volume.util import constants, system
 from ceph_volume.util.device import Device
 from .prepare import write_keyring
 from .disk import lsblk, device_family, get_part_entry_type
+from packaging import version
 
 logger = logging.getLogger(__name__)
 mlogger = terminal.MultiLogger(__name__)
 
+def set_dmcrypt_no_workqueue(target_version: str = '2.3.4') -> None:
+    """
+    set `conf.dmcrypt_no_workqueue` to `True` if the available
+    version of `cryptsetup` is greater or equal to `version`
+    """
+    command = ["cryptsetup", "--version"]
+    out, err, rc = process.call(command)
+    try:
+        if version.parse(out[0]) >= version.parse(f'cryptsetup {target_version}'):
+            conf.dmcrypt_no_workqueue = True
+    except IndexError:
+        mlogger.debug(f'cryptsetup version check: rc={rc} out={out} err={err}')
+        raise RuntimeError("Couldn't check the cryptsetup version.")
+
+def bypass_workqueue(device: str) -> bool:
+    return not Device(device).rotational and conf.dmcrypt_no_workqueue
+
 def get_key_size_from_conf():
     """
     Return the osd dmcrypt key size from config file.
@@ -79,6 +97,10 @@ def plain_open(key, device, mapping):
         '--key-size', '256',
     ]
 
+    if bypass_workqueue(device):
+        command.extend(['--perf-no_read_workqueue',
+                        '--perf-no_write_workqueue'])
+
     process.call(command, stdin=key, terminal_verbose=True, show_command=True)
 
 
@@ -103,6 +125,11 @@ def luks_open(key, device, mapping):
         device,
         mapping,
     ]
+
+    if bypass_workqueue(device):
+        command.extend(['--perf-no_read_workqueue',
+                        '--perf-no_write_workqueue'])
+
     process.call(command, stdin=key, terminal_verbose=True, show_command=True)