ceph-volume: fix type annotation in objectore

This commit addresses some python type annotations errors
in `objectstore` code.

Signed-off-by: Guillaume Abrioux <[email protected]>

Author: guits

src/ceph-volume/ceph_volume/api/lvm.py

@@ -835,6 +835,7 @@ def __init__(self, **kw: Any) -> None:
         self.lv_name: str = ''
         self.lv_uuid: str = ''
         self.vg_name: str = ''
+        self.lv_tags: Dict[str, Any] = {}
         for k, v in kw.items():
             setattr(self, k, v)
         self.lv_api = kw

src/ceph-volume/ceph_volume/objectstore/baseobjectstore.py

@@ -64,7 +64,7 @@ def prepare_data_device(self,
                             osd_uuid: str) -> Optional["Volume"]:
         raise NotImplementedError()
 
-    def safe_prepare(self, args: "argparse.Namespace") -> None:
+    def safe_prepare(self, args: Optional["argparse.Namespace"] = None) -> None:
         raise NotImplementedError()
 
     def add_objectstore_opts(self) -> None:

src/ceph-volume/ceph_volume/objectstore/bluestore.py

@@ -101,6 +101,9 @@ def osd_mkfs(self) -> None:
                 if self.method == 'raw':
                     path = self.args.__dict__.get(dev_type, None)
                 else:
-                    path = self.block_lv.tags.get(dev_type, None)
+                    if self.block_lv is not None:
+                        path = self.block_lv.tags.get(dev_type, None)
+                    else:
+                        raise RuntimeError('Unexpected error while running bluestore mkfs.')
                 if path is not None:
                     CephLuks2(path).config_luks2({'subsystem': f'ceph_fsid={self.osd_fsid}'})

src/ceph-volume/ceph_volume/objectstore/lvmbluestore.py

@@ -121,10 +121,11 @@ def safe_prepare(self,
         except ValueError:
             lv = None
 
-        if api.is_ceph_device(lv):
-            logger.info("device {} is already used".format(self.args.data))
-            raise RuntimeError("skipping {}, it is already prepared".format(
-                self.args.data))
+        if lv is not None:
+            if api.is_ceph_device(lv):
+                logger.info("device {} is already used".format(self.args.data))
+                raise RuntimeError("skipping {}, it is already prepared".format(
+                    self.args.data))
         try:
             self.prepare()
         except Exception:
@@ -253,23 +254,26 @@ def setup_device(self,
             lv_type = "osd-{}".format(device_type)
             name_uuid = system.generate_uuid()
             kwargs = {
+                'name_prefix': lv_type,
+                'uuid': name_uuid,
+                'vg': None,
                 'device': device_name,
+                'slots': slots,
+                'extents': None,
+                'size': None,
                 'tags': tags,
-                'slots': slots
             }
             # TODO use get_block_db_size and co here to get configured size in
             # conf file
             if size != 0:
                 kwargs['size'] = size
-            lv = api.create_lv(
-                lv_type,
-                name_uuid,
-                **kwargs)
-            path = lv.lv_path
-            tags['ceph.{}_device'.format(device_type)] = path
-            tags['ceph.{}_uuid'.format(device_type)] = lv.lv_uuid
-            lv_uuid = lv.lv_uuid
-            lv.set_tags(tags)
+            lv = api.create_lv(**kwargs)
+            if lv is not None:
+                path = lv.lv_path
+                lv_uuid = lv.lv_uuid
+                tags['ceph.{}_device'.format(device_type)] = path
+                tags['ceph.{}_uuid'.format(device_type)] = lv_uuid
+                lv.set_tags(tags)
         else:
             # otherwise assume this is a regular disk partition
             name_uuid = self.get_ptuuid(device_name)
@@ -282,8 +286,7 @@ def setup_device(self,
     def get_osd_device_path(self,
                             osd_lvs: List["Volume"],
                             device_type: str,
-                            dmcrypt_secret: Optional[str] =
-                            None) -> Optional[str]:
+                            dmcrypt_secret: str = '') -> Optional[str]:
         """
         ``device_type`` can be one of ``db``, ``wal`` or ``block`` so that we
         can query LVs on system and fallback to querying the uuid if that is
@@ -303,7 +306,7 @@ def get_osd_device_path(self,
             logger.debug('Found block device (%s) with encryption: %s',
                          osd_block_lv.name, is_encrypted)
             uuid_tag = 'ceph.%s_uuid' % device_type
-            device_uuid = osd_block_lv.tags.get(uuid_tag)
+            device_uuid = osd_block_lv.tags.get(uuid_tag, '')
             if not device_uuid:
                 return None
 

src/ceph-volume/ceph_volume/objectstore/rawbluestore.py

@@ -22,7 +22,7 @@ def __init__(self, args: "argparse.Namespace") -> None:
         super().__init__(args)
         self.method = 'raw'
         self.devices: List[str] = getattr(args, 'devices', [])
-        self.osd_id = getattr(self.args, 'osd_id', None)
+        self.osd_id = getattr(self.args, 'osd_id', '')
         self.osd_fsid = getattr(self.args, 'osd_fsid', '')
         self.block_device_path = getattr(self.args, 'data', '')
         self.db_device_path = getattr(self.args, 'block_db', '')
@@ -164,7 +164,7 @@ def activate(self) -> None:
         activated_any: bool = False
 
         for d in disk.lsblk_all(abspath=True):
-            device: str = d.get('NAME')
+            device: str = d.get('NAME', '')
             luks2 = encryption_utils.CephLuks2(device)
             if luks2.is_ceph_encrypted:
                 if luks2.is_tpm2_enrolled and self.osd_fsid == luks2.osd_fsid: