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

import pytest
import json
import random

from argparse import ArgumentError
from mock import MagicMock, patch

from ceph_volume.devices.lvm import batch
from ceph_volume.util import arg_validators


class TestBatch(object):

    def test_batch_instance(self, is_root):
        b = batch.Batch([])
        b.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")
        devices = [device1, device2]
        db_devices = [device2]
        with pytest.raises(Exception) as disjoint_ex:
            batch.ensure_disjoint_device_lists(devices, db_devices)
        assert 'Device lists are not disjoint' in str(disjoint_ex.value)

    @patch('ceph_volume.util.arg_validators.Device')
    def test_reject_partition(self, mocked_device):
        mocked_device.return_value = MagicMock(
            is_partition=True,
            has_gpt_headers=False,
        )
        with pytest.raises(ArgumentError):
            arg_validators.ValidBatchDevice()('foo')

    @pytest.mark.parametrize('format_', ['pretty', 'json', 'json-pretty'])
    def test_report(self, format_, factory, conf_ceph_stub, mock_device_generator):
        # just ensure reporting works
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        devs = [mock_device_generator() for _ in range(5)]
        args = factory(data_slots=1,
                       osds_per_device=1,
                       osd_ids=[],
                       report=True,
                       format=format_,
                       devices=devs,
                       db_devices=[],
                       wal_devices=[],
                       bluestore=True,
                       block_db_size="1G",
                       dmcrypt=True,
                      )
        b = batch.Batch([])
        plan = b.get_plan(args)
        b.args = args
        b.report(plan)

    @pytest.mark.parametrize('format_', ['json', 'json-pretty'])
    def test_json_report_valid_empty(self, format_, factory, conf_ceph_stub, mock_device_generator):
        # ensure json reports are valid when empty
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        devs = []
        args = factory(data_slots=1,
                       osds_per_device=1,
                       osd_ids=[],
                       report=True,
                       format=format_,
                       devices=devs,
                       db_devices=[],
                       wal_devices=[],
                       bluestore=True,
                       block_db_size="1G",
                       dmcrypt=True,
                      )
        b = batch.Batch([])
        plan = b.get_plan(args)
        b.args = args
        report = b._create_report(plan)
        json.loads(report)

    @pytest.mark.parametrize('format_', ['json', 'json-pretty'])
    def test_json_report_valid_empty_unavailable_fast(self, format_, factory, conf_ceph_stub, mock_device_generator):
        # ensure json reports are valid when empty
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        devs = [mock_device_generator() for _ in range(5)]
        fast_devs = [mock_device_generator()]
        fast_devs[0].available_lvm = False
        args = factory(data_slots=1,
                       osds_per_device=1,
                       osd_ids=[],
                       report=True,
                       format=format_,
                       devices=devs,
                       db_devices=fast_devs,
                       wal_devices=[],
                       bluestore=True,
                       block_db_size="1G",
                       dmcrypt=True,
                      )
        b = batch.Batch([])
        plan = b.get_plan(args)
        b.args = args
        report = b._create_report(plan)
        json.loads(report)


    @pytest.mark.parametrize('format_', ['json', 'json-pretty'])
    def test_json_report_valid_empty_unavailable_very_fast(self, format_, factory, conf_ceph_stub, mock_device_generator):
        # ensure json reports are valid when empty
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        devs = [mock_device_generator() for _ in range(5)]
        fast_devs = [mock_device_generator()]
        very_fast_devs = [mock_device_generator()]
        very_fast_devs[0].available_lvm = False
        args = factory(data_slots=1,
                       osds_per_device=1,
                       osd_ids=[],
                       report=True,
                       format=format_,
                       devices=devs,
                       db_devices=fast_devs,
                       wal_devices=very_fast_devs,
                       bluestore=True,
                       block_db_size="1G",
                       dmcrypt=True,
                      )
        b = batch.Batch([])
        plan = b.get_plan(args)
        b.args = args
        report = b._create_report(plan)
        json.loads(report)

    @pytest.mark.parametrize('rota', [0, 1])
    def test_batch_sort_full(self, factory, rota):
        device1 = factory(used_by_ceph=False, available=True, rotational=rota, abspath="/dev/sda")
        device2 = factory(used_by_ceph=False, available=True, rotational=rota, abspath="/dev/sdb")
        device3 = factory(used_by_ceph=False, available=True, rotational=rota, abspath="/dev/sdc")
        devices = [device1, device2, device3]
        args = factory(report=True,
                       devices=devices,
                       filestore=False,
                      )
        b = batch.Batch([])
        b.args = args
        b._sort_rotational_disks()
        assert len(b.args.devices) == 3

    @pytest.mark.parametrize('objectstore', ['bluestore', 'filestore'])
    def test_batch_sort_mixed(self, factory, objectstore):
        device1 = factory(used_by_ceph=False, available=True, rotational=1, abspath="/dev/sda")
        device2 = factory(used_by_ceph=False, available=True, rotational=1, abspath="/dev/sdb")
        device3 = factory(used_by_ceph=False, available=True, rotational=0, abspath="/dev/sdc")
        devices = [device1, device2, device3]
        args = factory(report=True,
                       devices=devices,
                       filestore=False if objectstore == 'bluestore' else True,
                      )
        b = batch.Batch([])
        b.args = args
        b._sort_rotational_disks()
        assert len(b.args.devices) == 2
        if objectstore == 'bluestore':
            assert len(b.args.db_devices) == 1
        else:
            assert len(b.args.journal_devices) == 1

    def test_get_physical_osds_return_len(self, factory,
                                          mock_devices_available,
                                          conf_ceph_stub,
                                          osds_per_device):
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        args = factory(data_slots=1, osds_per_device=osds_per_device,
                       osd_ids=[], dmcrypt=False)
        osds = batch.get_physical_osds(mock_devices_available, args)
        assert len(osds) == len(mock_devices_available) * osds_per_device

    def test_get_physical_osds_rel_size(self, factory,
                                          mock_devices_available,
                                          conf_ceph_stub,
                                          osds_per_device):
        args = factory(data_slots=1, osds_per_device=osds_per_device,
                       osd_ids=[], dmcrypt=False)
        osds = batch.get_physical_osds(mock_devices_available, args)
        for osd in osds:
            assert osd.data[1] == 1.0 / osds_per_device

    def test_get_physical_osds_abs_size(self, factory,
                                          mock_devices_available,
                                          conf_ceph_stub,
                                          osds_per_device):
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        args = factory(data_slots=1, osds_per_device=osds_per_device,
                       osd_ids=[], dmcrypt=False)
        osds = batch.get_physical_osds(mock_devices_available, args)
        for osd, dev in zip(osds, mock_devices_available):
            assert osd.data[2] == int(dev.vg_size[0] / osds_per_device)

    def test_get_physical_osds_osd_ids(self, factory,
                                          mock_devices_available,
                                          osds_per_device):
        pass

    def test_get_physical_fast_allocs_length(self, factory,
                                             conf_ceph_stub,
                                             mock_devices_available):
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        args = factory(block_db_slots=None, get_block_db_size=None)
        fast = batch.get_physical_fast_allocs(mock_devices_available,
                                              'block_db', 2, 2, args)
        assert len(fast) == 2

    @pytest.mark.parametrize('occupied_prior', range(7))
    @pytest.mark.parametrize('slots,num_devs',
                             [l for sub in [list(zip([x]*x, range(1, x + 1))) for x in range(1,7)] for l in sub])
    def test_get_physical_fast_allocs_length_existing(self,
                                                      num_devs,
                                                      slots,
                                                      occupied_prior,
                                                      factory,
                                                      conf_ceph_stub,
                                                      mock_device_generator):
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        occupied_prior = min(occupied_prior, slots)
        devs = [mock_device_generator() for _ in range(num_devs)]
        already_assigned = 0
        while already_assigned < occupied_prior:
            dev_i = random.randint(0, num_devs - 1)
            dev = devs[dev_i]
            if len(dev.lvs) < occupied_prior:
                dev.lvs.append('foo')
                dev.path = '/dev/bar'
                already_assigned = sum([len(d.lvs) for d in devs])
        args = factory(block_db_slots=None, get_block_db_size=None)
        expected_num_osds = max(len(devs) * slots - occupied_prior, 0)
        fast = batch.get_physical_fast_allocs(devs,
                                              'block_db', slots,
                                              expected_num_osds, args)
        assert len(fast) == expected_num_osds
        expected_assignment_on_used_devices = sum([slots - len(d.lvs) for d in devs if len(d.lvs) > 0])
        assert len([f for f in fast if f[0] == '/dev/bar']) == expected_assignment_on_used_devices
        assert len([f for f in fast if f[0] != '/dev/bar']) == expected_num_osds - expected_assignment_on_used_devices

    def test_get_lvm_osds_return_len(self, factory,
                                     mock_lv_device_generator,
                                     conf_ceph_stub,
                                     osds_per_device):
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        args = factory(data_slots=1, osds_per_device=osds_per_device,
                       osd_ids=[], dmcrypt=False)
        mock_lvs = [mock_lv_device_generator()]
        osds = batch.get_lvm_osds(mock_lvs, args)
        assert len(osds) == 1


class TestBatchOsd(object):

    def test_osd_class_ctor(self):
        osd = batch.Batch.OSD('/dev/data', 1, '5G', 1, 1, None)
        assert osd.data == batch.Batch.OSD.VolSpec('/dev/data',
                                                   1,
                                                   '5G',
                                                   1,
                                                   'data')
    def test_add_fast(self):
        osd = batch.Batch.OSD('/dev/data', 1, '5G', 1, 1, None)
        osd.add_fast_device('/dev/db', 1, '5G', 1, 'block_db')
        assert osd.fast == batch.Batch.OSD.VolSpec('/dev/db',
                                                   1,
                                                   '5G',
                                                   1,
                                                   'block_db')

    def test_add_very_fast(self):
        osd = batch.Batch.OSD('/dev/data', 1, '5G', 1, 1, None)
        osd.add_very_fast_device('/dev/wal', 1, '5G', 1)
        assert osd.very_fast == batch.Batch.OSD.VolSpec('/dev/wal',
                                                        1,
                                                        '5G',
                                                        1,
                                                        'block_wal')
Commit	Line	Data
494da23a	1	import pytest
f6b5b4d7	2	import json
f91f0fd5 TL	3	import random
	4
	5	from argparse import ArgumentError
	6	from mock import MagicMock, patch
	7
91327a77	8	from ceph_volume.devices.lvm import batch
f91f0fd5	9	from ceph_volume.util import arg_validators
91327a77 AA	10
91327a77 AA	11
11fdf7f2	12	class TestBatch(object):
f64942e4 AA	13
	14	def test_batch_instance(self, is_root):
	15	b = batch.Batch([])
	16	b.main()
	17
494da23a TL	18	def test_disjoint_device_lists(self, factory):
	19	device1 = factory(used_by_ceph=False, available=True, abspath="/dev/sda")
	20	device2 = factory(used_by_ceph=False, available=True, abspath="/dev/sdb")
f91f0fd5 TL	21	devices = [device1, device2]
f91f0fd5 TL	22	db_devices = [device2]
494da23a	23	with pytest.raises(Exception) as disjoint_ex:
f91f0fd5	24	batch.ensure_disjoint_device_lists(devices, db_devices)
494da23a TL	25	assert 'Device lists are not disjoint' in str(disjoint_ex.value)
494da23a TL	26
f91f0fd5 TL	27	@patch('ceph_volume.util.arg_validators.Device')
	28	def test_reject_partition(self, mocked_device):
	29	mocked_device.return_value = MagicMock(
	30	is_partition=True,
	31	has_gpt_headers=False,
	32	)
	33	with pytest.raises(ArgumentError):
	34	arg_validators.ValidBatchDevice()('foo')
f64942e4	35
f91f0fd5 TL	36	@pytest.mark.parametrize('format_', ['pretty', 'json', 'json-pretty'])
	37	def test_report(self, format_, factory, conf_ceph_stub, mock_device_generator):
	38	# just ensure reporting works
	39	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
	40	devs = [mock_device_generator() for _ in range(5)]
	41	args = factory(data_slots=1,
	42	osds_per_device=1,
	43	osd_ids=[],
	44	report=True,
	45	format=format_,
	46	devices=devs,
	47	db_devices=[],
	48	wal_devices=[],
	49	bluestore=True,
	50	block_db_size="1G",
	51	dmcrypt=True,
	52	)
	53	b = batch.Batch([])
	54	plan = b.get_plan(args)
	55	b.args = args
	56	b.report(plan)
91327a77	57
f91f0fd5 TL	58	@pytest.mark.parametrize('format_', ['json', 'json-pretty'])
	59	def test_json_report_valid_empty(self, format_, factory, conf_ceph_stub, mock_device_generator):
	60	# ensure json reports are valid when empty
	61	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
	62	devs = []
	63	args = factory(data_slots=1,
	64	osds_per_device=1,
	65	osd_ids=[],
	66	report=True,
	67	format=format_,
	68	devices=devs,
	69	db_devices=[],
	70	wal_devices=[],
	71	bluestore=True,
	72	block_db_size="1G",
	73	dmcrypt=True,
	74	)
	75	b = batch.Batch([])
	76	plan = b.get_plan(args)
	77	b.args = args
	78	report = b._create_report(plan)
	79	json.loads(report)
91327a77	80
f91f0fd5 TL	81	@pytest.mark.parametrize('format_', ['json', 'json-pretty'])
	82	def test_json_report_valid_empty_unavailable_fast(self, format_, factory, conf_ceph_stub, mock_device_generator):
	83	# ensure json reports are valid when empty
	84	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
	85	devs = [mock_device_generator() for _ in range(5)]
	86	fast_devs = [mock_device_generator()]
	87	fast_devs[0].available_lvm = False
	88	args = factory(data_slots=1,
	89	osds_per_device=1,
	90	osd_ids=[],
	91	report=True,
	92	format=format_,
	93	devices=devs,
	94	db_devices=fast_devs,
	95	wal_devices=[],
	96	bluestore=True,
	97	block_db_size="1G",
	98	dmcrypt=True,
	99	)
92f5a8d4	100	b = batch.Batch([])
f91f0fd5	101	plan = b.get_plan(args)
92f5a8d4	102	b.args = args
f91f0fd5 TL	103	report = b._create_report(plan)
	104	json.loads(report)
	105
	106
	107	@pytest.mark.parametrize('format_', ['json', 'json-pretty'])
	108	def test_json_report_valid_empty_unavailable_very_fast(self, format_, factory, conf_ceph_stub, mock_device_generator):
	109	# ensure json reports are valid when empty
	110	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
	111	devs = [mock_device_generator() for _ in range(5)]
	112	fast_devs = [mock_device_generator()]
	113	very_fast_devs = [mock_device_generator()]
	114	very_fast_devs[0].available_lvm = False
	115	args = factory(data_slots=1,
	116	osds_per_device=1,
	117	osd_ids=[],
	118	report=True,
	119	format=format_,
	120	devices=devs,
	121	db_devices=fast_devs,
	122	wal_devices=very_fast_devs,
	123	bluestore=True,
	124	block_db_size="1G",
	125	dmcrypt=True,
	126	)
	127	b = batch.Batch([])
	128	plan = b.get_plan(args)
	129	b.args = args
	130	report = b._create_report(plan)
	131	json.loads(report)
	132
	133	@pytest.mark.parametrize('rota', [0, 1])
	134	def test_batch_sort_full(self, factory, rota):
	135	device1 = factory(used_by_ceph=False, available=True, rotational=rota, abspath="/dev/sda")
	136	device2 = factory(used_by_ceph=False, available=True, rotational=rota, abspath="/dev/sdb")
	137	device3 = factory(used_by_ceph=False, available=True, rotational=rota, abspath="/dev/sdc")
	138	devices = [device1, device2, device3]
	139	args = factory(report=True,
	140	devices=devices,
	141	filestore=False,
	142	)
f6b5b4d7 TL	143	b = batch.Batch([])
f6b5b4d7 TL	144	b.args = args
f91f0fd5 TL	145	b._sort_rotational_disks()
	146	assert len(b.args.devices) == 3
	147
	148	@pytest.mark.parametrize('objectstore', ['bluestore', 'filestore'])
	149	def test_batch_sort_mixed(self, factory, objectstore):
	150	device1 = factory(used_by_ceph=False, available=True, rotational=1, abspath="/dev/sda")
	151	device2 = factory(used_by_ceph=False, available=True, rotational=1, abspath="/dev/sdb")
	152	device3 = factory(used_by_ceph=False, available=True, rotational=0, abspath="/dev/sdc")
	153	devices = [device1, device2, device3]
	154	args = factory(report=True,
	155	devices=devices,
	156	filestore=False if objectstore == 'bluestore' else True,
	157	)
	158	b = batch.Batch([])
	159	b.args = args
	160	b._sort_rotational_disks()
	161	assert len(b.args.devices) == 2
	162	if objectstore == 'bluestore':
	163	assert len(b.args.db_devices) == 1
	164	else:
	165	assert len(b.args.journal_devices) == 1
	166
	167	def test_get_physical_osds_return_len(self, factory,
	168	mock_devices_available,
	169	conf_ceph_stub,
	170	osds_per_device):
	171	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
	172	args = factory(data_slots=1, osds_per_device=osds_per_device,
	173	osd_ids=[], dmcrypt=False)
	174	osds = batch.get_physical_osds(mock_devices_available, args)
	175	assert len(osds) == len(mock_devices_available) * osds_per_device
	176
	177	def test_get_physical_osds_rel_size(self, factory,
	178	mock_devices_available,
	179	conf_ceph_stub,
	180	osds_per_device):
	181	args = factory(data_slots=1, osds_per_device=osds_per_device,
	182	osd_ids=[], dmcrypt=False)
	183	osds = batch.get_physical_osds(mock_devices_available, args)
	184	for osd in osds:
	185	assert osd.data[1] == 1.0 / osds_per_device
	186
	187	def test_get_physical_osds_abs_size(self, factory,
	188	mock_devices_available,
	189	conf_ceph_stub,
	190	osds_per_device):
	191	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
	192	args = factory(data_slots=1, osds_per_device=osds_per_device,
	193	osd_ids=[], dmcrypt=False)
	194	osds = batch.get_physical_osds(mock_devices_available, args)
	195	for osd, dev in zip(osds, mock_devices_available):
	196	assert osd.data[2] == int(dev.vg_size[0] / osds_per_device)
	197
	198	def test_get_physical_osds_osd_ids(self, factory,
	199	mock_devices_available,
	200	osds_per_device):
	201	pass
	202
	203	def test_get_physical_fast_allocs_length(self, factory,
	204	conf_ceph_stub,
	205	mock_devices_available):
	206	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
	207	args = factory(block_db_slots=None, get_block_db_size=None)
	208	fast = batch.get_physical_fast_allocs(mock_devices_available,
209	'block_db', 2, 2, args)
210	assert len(fast) == 2
211
212	@pytest.mark.parametrize('occupied_prior', range(7))
213	@pytest.mark.parametrize('slots,num_devs',
214	[l for sub in [list(zip([x]*x, range(1, x + 1))) for x in range(1,7)] for l in sub])
215	def test_get_physical_fast_allocs_length_existing(self,
216	num_devs,
217	slots,
218	occupied_prior,
219	factory,
220	conf_ceph_stub,
221	mock_device_generator):
222	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
223	occupied_prior = min(occupied_prior, slots)
224	devs = [mock_device_generator() for _ in range(num_devs)]
225	already_assigned = 0
226	while already_assigned < occupied_prior:
227	dev_i = random.randint(0, num_devs - 1)
228	dev = devs[dev_i]
229	if len(dev.lvs) < occupied_prior:
230	dev.lvs.append('foo')
231	dev.path = '/dev/bar'
232	already_assigned = sum([len(d.lvs) for d in devs])
233	args = factory(block_db_slots=None, get_block_db_size=None)
234	expected_num_osds = max(len(devs) * slots - occupied_prior, 0)
235	fast = batch.get_physical_fast_allocs(devs,
236	'block_db', slots,
237	expected_num_osds, args)
238	assert len(fast) == expected_num_osds
239	expected_assignment_on_used_devices = sum([slots - len(d.lvs) for d in devs if len(d.lvs) > 0])
240	assert len([f for f in fast if f[0] == '/dev/bar']) == expected_assignment_on_used_devices
241	assert len([f for f in fast if f[0] != '/dev/bar']) == expected_num_osds - expected_assignment_on_used_devices
242
243	def test_get_lvm_osds_return_len(self, factory,
244	mock_lv_device_generator,
245	conf_ceph_stub,
246	osds_per_device):
247	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
248	args = factory(data_slots=1, osds_per_device=osds_per_device,
249	osd_ids=[], dmcrypt=False)
250	mock_lvs = [mock_lv_device_generator()]
251	osds = batch.get_lvm_osds(mock_lvs, args)
252	assert len(osds) == 1
253
254
255	class TestBatchOsd(object):
256
257	def test_osd_class_ctor(self):
258	osd = batch.Batch.OSD('/dev/data', 1, '5G', 1, 1, None)
259	assert osd.data == batch.Batch.OSD.VolSpec('/dev/data',
260	1,
261	'5G',
262	1,
263	'data')
264	def test_add_fast(self):
265	osd = batch.Batch.OSD('/dev/data', 1, '5G', 1, 1, None)
266	osd.add_fast_device('/dev/db', 1, '5G', 1, 'block_db')
267	assert osd.fast == batch.Batch.OSD.VolSpec('/dev/db',
268	1,
269	'5G',
270	1,
271	'block_db')
272
273	def test_add_very_fast(self):
274	osd = batch.Batch.OSD('/dev/data', 1, '5G', 1, 1, None)
275	osd.add_very_fast_device('/dev/wal', 1, '5G', 1)
276	assert osd.very_fast == batch.Batch.OSD.VolSpec('/dev/wal',
277	1,
278	'5G',
279	1,
280	'block_wal')