ceph-volume: util/prepare fix osd_id_available()

The current check only allows to request an OSD id that exists but
marked as 'destroyed'.
With this small fix, we can now use `--osd-id` with an id that doesn't
exist at all.

Fixes: https://tracker.ceph.com/issues/50880

Signed-off-by: Guillaume Abrioux <gabrioux@redhat.com>

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

@@ -325,6 +325,7 @@ class Batch(object):
             nargs='*',
             default=[],
             help='Reuse existing OSD ids',
+            type=common.valid_osd_id
         )
         self.args = parser.parse_args(argv)
         self.parser = parser

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

@@ -3,6 +3,8 @@ from ceph_volume import process, conf
 from ceph_volume import terminal
 import argparse
 
+def valid_osd_id(val):
+    return str(int(val))
 
 def rollback_osd(args, osd_id=None):
     """
@@ -56,6 +58,7 @@ common_args = {
     '--osd-id': {
         'help': 'Reuse an existing OSD id',
         'default': None,
+        'type': valid_osd_id,
     },
     '--osd-fsid': {
         'help': 'Reuse an existing OSD fsid',

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

@@ -8,6 +8,7 @@ from ceph_volume.util.device import Device
 from ceph_volume import decorators, terminal, process
 from ceph_volume.api import lvm as api
 from ceph_volume.systemd import systemctl
+from ceph_volume.devices.lvm.common import valid_osd_id
 
 
 logger = logging.getLogger(__name__)
@@ -275,7 +276,7 @@ class Migrate(object):
     # (in the order of precedence, stop on the first match)
     # if source list has DB volume - target device replaces it.
     # if source list has WAL volume - target device replace it.
-    # if source list has slow volume only - operation isn’t permitted,
+    # if source list has slow volume only - operation isn't permitted,
     #  requires explicit allocation via new-db/new-wal command.detects which
     def get_target_type_by_source(self, devices):
         ret = None
@@ -447,6 +448,7 @@ class Migrate(object):
             '--osd-id',
             required=True,
             help='Specify an OSD ID to detect associated devices for zapping',
+            type=valid_osd_id
         )
 
         parser.add_argument(
@@ -545,6 +547,7 @@ class NewVolume(object):
             '--osd-id',
             required=True,
             help='Specify an OSD ID to attach new volume to',
+            type=valid_osd_id,
         )
 
         parser.add_argument(

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

@@ -10,6 +10,7 @@ from ceph_volume.api import lvm as api
 from ceph_volume.util import system, encryption, disk, arg_validators, str_to_int, merge_dict
 from ceph_volume.util.device import Device
 from ceph_volume.systemd import systemctl
+from ceph_volume.devices.lvm.common import valid_osd_id
 
 logger = logging.getLogger(__name__)
 mlogger = terminal.MultiLogger(__name__)
@@ -376,6 +377,7 @@ class Zap(object):
 
         parser.add_argument(
             '--osd-id',
+            type=valid_osd_id,
             help='Specify an OSD ID to detect associated devices for zapping',
         )
 

src/ceph-volume/ceph_volume/tests/devices/lvm/test_batch.py

@@ -15,6 +15,10 @@ class TestBatch(object):
         b = batch.Batch([])
         b.main()
 
+    def test_invalid_osd_ids_passed(self):
+        with pytest.raises(SystemExit):
+            batch.Batch(argv=['--osd-ids', '1', 'foo']).main()
+
     def test_disjoint_device_lists(self, factory):
         device1 = factory(used_by_ceph=False, available=True, abspath="/dev/sda")
         device2 = factory(used_by_ceph=False, available=True, abspath="/dev/sdb")

src/ceph-volume/ceph_volume/tests/devices/lvm/test_migrate.py

@@ -992,13 +992,17 @@ class TestNew(object):
 
 class TestMigrate(object):
 
+    def test_invalid_osd_id_passed(self, is_root):
+        with pytest.raises(SystemExit):
+            migrate.Migrate(argv=['--osd-fsid', '123', '--from', 'data', '--target', 'foo', '--osd-id', 'foo']).main()
+
     mock_volume = None
     def mock_get_lv_by_fullname(self, *args, **kwargs):
         return self.mock_volume
 
     mock_process_input = []
     def mock_process(self, *args, **kwargs):
-        self.mock_process_input.append(args[0]);
+        self.mock_process_input.append(args[0])
         return ('', '', 0)
 
     mock_single_volumes = {}

src/ceph-volume/ceph_volume/tests/devices/lvm/test_prepare.py

@@ -158,6 +158,10 @@ class TestPrepare(object):
                                                     'ceph.data_uuid': 'fake-uuid',
                                                     'ceph.data_device': '/dev/sdx'})
 
+    def test_invalid_osd_id_passed(self):
+        with pytest.raises(SystemExit):
+            lvm.prepare.Prepare(argv=['--osd-id', 'foo']).main()
+
 
 class TestActivate(object):
 

src/ceph-volume/ceph_volume/tests/devices/lvm/test_zap.py

@@ -7,6 +7,11 @@ from ceph_volume.api import lvm as api
 from ceph_volume.devices.lvm import zap
 
 
+class TestZap(object):
+    def test_invalid_osd_id_passed(self):
+        with pytest.raises(SystemExit):
+            zap.Zap(argv=['--osd-id', 'foo']).main()
+
 class TestFindAssociatedDevices(object):
 
     def test_no_lvs_found_that_match_id(self, monkeypatch, device_info):

src/ceph-volume/ceph_volume/tests/util/test_prepare.py

@@ -33,16 +33,7 @@ class TestOSDIDAvailable(object):
         stdout = ['', json.dumps(stdout)]
         monkeypatch.setattr('ceph_volume.process.call', lambda *a, **kw: (stdout, '', 0))
         result = prepare.osd_id_available(1)
-        assert not result
-
-    def test_invalid_osd_id(self, monkeypatch):
-        stdout = dict(nodes=[
-            dict(id=0),
-        ])
-        stdout = ['', json.dumps(stdout)]
-        monkeypatch.setattr('ceph_volume.process.call', lambda *a, **kw: (stdout, '', 0))
-        result = prepare.osd_id_available("foo")
-        assert not result
+        assert result
 
     def test_returns_true_when_id_is_destroyed(self, monkeypatch):
         stdout = dict(nodes=[

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

@@ -183,6 +183,7 @@ def osd_id_available(osd_id):
     """
     if osd_id is None:
         return False
+
     bootstrap_keyring = '/var/lib/ceph/bootstrap-osd/%s.keyring' % conf.cluster
     stdout, stderr, returncode = process.call(
         [
@@ -202,7 +203,7 @@ def osd_id_available(osd_id):
     output = json.loads(''.join(stdout).strip())
     osds = output['nodes']
     osd = [osd for osd in osds if str(osd['id']) == str(osd_id)]
-    if osd and osd[0].get('status') == "destroyed":
+    if not osd or (osd and osd[0].get('status') == "destroyed"):
         return True
     return False