+++ /dev/null
-#
-# Copyright (C) 2015, 2016 Red Hat <contact@redhat.com>
-#
-# Author: Loic Dachary <loic@dachary.org>
-#
-# This program is free software; you can redistribute it and/or modify
-# it under the terms of the GNU Library Public License as published by
-# the Free Software Foundation; either version 2, or (at your option)
-# any later version.
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU Library Public License for more details.
-#
-# When debugging these tests (must be root), here are a few useful commands:
-#
-# export PATH=.:..:$PATH
-# ceph-disk.sh # run once to prepare the environment as it would be by teuthology
-# ln -sf /home/ubuntu/ceph/src/ceph-disk/ceph_disk/main.py $(which ceph-disk)
-# ln -sf /home/ubuntu/ceph/udev/95-ceph-osd.rules /lib/udev/rules.d/95-ceph-osd.rules
-# ln -sf /home/ubuntu/ceph/systemd/ceph-disk@.service /usr/lib/systemd/system/ceph-disk@.service
-# ceph-disk.conf will be silently ignored if it is a symbolic link or a hard link /var/log/upstart for logs
-# cp /home/ubuntu/ceph/src/upstart/ceph-disk.conf /etc/init/ceph-disk.conf
-# id=3 ; ceph-disk deactivate --deactivate-by-id $id ; ceph-disk destroy --purge --zap --destroy-by-id $id
-# py.test -s -v -k test_activate_dmcrypt_luks ceph-disk-test.py
-#
-# CentOS 7
-# udevadm monitor --property & tail -f /var/log/messages
-# udev rules messages are logged in /var/log/messages
-# systemctl stop ceph-osd@2
-# systemctl start ceph-osd@2
-#
-# udevadm monitor --property & tail -f /var/log/syslog /var/log/upstart/* # on Ubuntu 14.04
-# udevadm test --action=add /block/vdb/vdb1 # verify the udev rule is run as expected
-# udevadm control --reload # when changing the udev rules
-# sudo /usr/sbin/ceph-disk -v trigger /dev/vdb1 # activates if vdb1 is data
-#
-# integration tests coverage
-# pip install coverage
-# perl -pi -e 's|"ceph-disk |"coverage run --source=/usr/sbin/ceph-disk --append /usr/sbin/ceph-disk |' ceph-disk-test.py
-# rm -f .coverage ; py.test -s -v ceph-disk-test.py
-# coverage report --show-missing
-#
-import argparse
-import json
-import logging
-import configobj
-import os
-import pytest
-import re
-import subprocess
-import sys
-import tempfile
-import time
-import uuid
-
-LOG = logging.getLogger('CephDisk')
-
-
-class CephDisk:
-
- def __init__(self):
- self.conf = configobj.ConfigObj('/etc/ceph/ceph.conf')
-
- def save_conf(self):
- self.conf.write(open('/etc/ceph/ceph.conf', 'wb'))
-
- @staticmethod
- def helper(command):
- command = "ceph-helpers-root.sh " + command
- return CephDisk.sh(command)
-
- @staticmethod
- def sh(command):
- LOG.debug(":sh: " + command)
- proc = subprocess.Popen(
- args=command,
- stdout=subprocess.PIPE,
- stderr=subprocess.STDOUT,
- shell=True,
- bufsize=1)
- output, _ = proc.communicate()
- if proc.poll():
- LOG.warning(output.decode('utf-8'))
- raise subprocess.CalledProcessError(
- returncode=proc.returncode,
- cmd=command,
- output=output,
- )
- lines = []
- for line in output.decode('utf-8').split('\n'):
- if 'dangerous and experimental' in line:
- LOG.debug('SKIP dangerous and experimental')
- continue
- lines.append(line)
- LOG.debug(line.strip().encode('ascii', 'ignore'))
- return "".join(lines)
-
- def unused_disks(self, pattern='[vs]d.'):
- names = [x for x in os.listdir("/sys/block") if re.match(pattern, x)]
- if not names:
- return []
- disks = json.loads(
- self.sh("ceph-disk list --format json " + " ".join(names)))
- unused = []
- for disk in disks:
- if 'partitions' not in disk:
- unused.append(disk['path'])
- return unused
-
- def ensure_sd(self):
- LOG.debug(self.unused_disks('sd.'))
- if self.unused_disks('sd.'):
- return
- modprobe = "modprobe scsi_debug vpd_use_hostno=0 add_host=1 dev_size_mb=300 ; udevadm settle"
- try:
- self.sh(modprobe)
- except:
- self.helper("install linux-image-extra-3.13.0-61-generic")
- self.sh(modprobe)
-
- def unload_scsi_debug(self):
- self.sh("rmmod scsi_debug || true")
-
- def get_lockbox(self):
- disks = json.loads(self.sh("ceph-disk list --format json"))
- for disk in disks:
- if 'partitions' in disk:
- for partition in disk['partitions']:
- if partition.get('type') == 'lockbox':
- return partition
- raise Exception("no lockbox found " + str(disks))
-
- def get_osd_partition(self, uuid):
- disks = json.loads(self.sh("ceph-disk list --format json"))
- for disk in disks:
- if 'partitions' in disk:
- for partition in disk['partitions']:
- if partition.get('uuid') == uuid:
- return partition
- raise Exception("uuid = " + uuid + " not found in " + str(disks))
-
- def get_journal_partition(self, uuid):
- return self.get_space_partition('journal', uuid)
-
- def get_block_partition(self, uuid):
- return self.get_space_partition('block', uuid)
-
- def get_blockdb_partition(self, uuid):
- return self.get_space_partition('block.db', uuid)
-
- def get_blockwal_partition(self, uuid):
- return self.get_space_partition('block.wal', uuid)
-
- def get_space_partition(self, name, uuid):
- data_partition = self.get_osd_partition(uuid)
- space_dev = data_partition[name + '_dev']
- disks = json.loads(self.sh("ceph-disk list --format json"))
- for disk in disks:
- if 'partitions' in disk:
- for partition in disk['partitions']:
- if partition['path'] == space_dev:
- if name + '_for' in partition:
- assert partition[
- name + '_for'] == data_partition['path']
- return partition
- raise Exception(
- name + " for uuid = " + uuid + " not found in " + str(disks))
-
- def destroy_osd(self, uuid):
- id = self.sh("ceph osd create " + uuid).strip()
- self.sh("""
- set -xe
- ceph-disk --verbose deactivate --deactivate-by-id {id}
- ceph-disk --verbose destroy --purge --destroy-by-id {id} --zap
- """.format(id=id))
-
- def deactivate_osd(self, uuid):
- id = self.sh("ceph osd create " + uuid).strip()
- self.sh("""
- set -xe
- ceph-disk --verbose deactivate --once --deactivate-by-id {id}
- """.format(id=id))
-
- @staticmethod
- def osd_up_predicate(osds, uuid):
- for osd in osds:
- if osd['uuid'] == uuid and 'up' in osd['state']:
- return True
- return False
-
- @staticmethod
- def wait_for_osd_up(uuid):
- CephDisk.wait_for_osd(uuid, CephDisk.osd_up_predicate, 'up')
-
- @staticmethod
- def osd_down_predicate(osds, uuid):
- found = False
- for osd in osds:
- if osd['uuid'] == uuid:
- found = True
- if 'down' in osd['state'] or ['exists'] == osd['state']:
- return True
- return not found
-
- @staticmethod
- def wait_for_osd_down(uuid):
- CephDisk.wait_for_osd(uuid, CephDisk.osd_down_predicate, 'down')
-
- @staticmethod
- def wait_for_osd(uuid, predicate, info):
- LOG.info("wait_for_osd " + info + " " + uuid)
- for delay in (1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024):
- dump = json.loads(CephDisk.sh("ceph osd dump -f json"))
- if predicate(dump['osds'], uuid):
- return True
- time.sleep(delay)
- raise Exception('timeout waiting for osd ' + uuid + ' to be ' + info)
-
- def check_osd_status(self, uuid, space_name=None):
- data_partition = self.get_osd_partition(uuid)
- assert data_partition['type'] == 'data'
- assert data_partition['state'] == 'active'
- if space_name is not None:
- space_partition = self.get_space_partition(space_name, uuid)
- assert space_partition
-
-
-class TestCephDisk(object):
-
- def setup_class(self):
- logging.basicConfig(level=logging.DEBUG)
- c = CephDisk()
- if c.sh("lsb_release -si").strip() == 'CentOS':
- c.helper("install multipath-tools device-mapper-multipath")
- c.conf['global']['pid file'] = '/var/run/ceph/$cluster-$name.pid'
- #
- # Avoid json parsing interference
- #
- c.conf['global']['debug monc'] = 0
- #
- # objecstore
- #
- c.conf['global']['osd journal size'] = 100
- #
- # bluestore
- #
- c.conf['global']['bluestore fsck on mount'] = 'true'
- c.save_conf()
-
- def setup(self):
- c = CephDisk()
- for key in ('osd objectstore', 'osd dmcrypt type'):
- if key in c.conf['global']:
- del c.conf['global'][key]
- c.save_conf()
-
- def test_deactivate_reactivate_osd(self):
- c = CephDisk()
- disk = c.unused_disks()[0]
- osd_uuid = str(uuid.uuid1())
- c.sh("ceph-disk --verbose zap " + disk)
- c.sh("ceph-disk --verbose prepare --filestore --osd-uuid " + osd_uuid +
- " " + disk)
- c.wait_for_osd_up(osd_uuid)
- device = json.loads(c.sh("ceph-disk list --format json " + disk))[0]
- assert len(device['partitions']) == 2
- c.check_osd_status(osd_uuid, 'journal')
- data_partition = c.get_osd_partition(osd_uuid)
- c.sh("ceph-disk --verbose deactivate " + data_partition['path'])
- c.wait_for_osd_down(osd_uuid)
- c.sh("ceph-disk --verbose activate " + data_partition['path'] + " --reactivate")
- # check again
- c.wait_for_osd_up(osd_uuid)
- device = json.loads(c.sh("ceph-disk list --format json " + disk))[0]
- assert len(device['partitions']) == 2
- c.check_osd_status(osd_uuid, 'journal')
- c.helper("pool_read_write")
- c.destroy_osd(osd_uuid)
-
- def test_destroy_osd_by_id(self):
- c = CephDisk()
- disk = c.unused_disks()[0]
- osd_uuid = str(uuid.uuid1())
- c.sh("ceph-disk --verbose prepare --filestore --osd-uuid " + osd_uuid + " " + disk)
- c.wait_for_osd_up(osd_uuid)
- c.check_osd_status(osd_uuid)
- c.destroy_osd(osd_uuid)
-
- def test_destroy_osd_by_dev_path(self):
- c = CephDisk()
- disk = c.unused_disks()[0]
- osd_uuid = str(uuid.uuid1())
- c.sh("ceph-disk --verbose prepare --filestore --osd-uuid " + osd_uuid + " " + disk)
- c.wait_for_osd_up(osd_uuid)
- partition = c.get_osd_partition(osd_uuid)
- assert partition['type'] == 'data'
- assert partition['state'] == 'active'
- c.sh("ceph-disk --verbose deactivate " + partition['path'])
- c.wait_for_osd_down(osd_uuid)
- c.sh("ceph-disk --verbose destroy --purge " + partition['path'] + " --zap")
-
- def test_deactivate_reactivate_dmcrypt_plain(self):
- c = CephDisk()
- c.conf['global']['osd dmcrypt type'] = 'plain'
- c.save_conf()
- osd_uuid = self.activate_dmcrypt('ceph-disk-no-lockbox')
- data_partition = c.get_osd_partition(osd_uuid)
- c.sh("ceph-disk --verbose deactivate " + data_partition['path'])
- c.wait_for_osd_down(osd_uuid)
- c.sh("ceph-disk --verbose activate-journal " + data_partition['journal_dev'] +
- " --reactivate" + " --dmcrypt")
- c.wait_for_osd_up(osd_uuid)
- c.check_osd_status(osd_uuid, 'journal')
- c.destroy_osd(osd_uuid)
- c.save_conf()
-
- def test_deactivate_reactivate_dmcrypt_luks(self):
- c = CephDisk()
- osd_uuid = self.activate_dmcrypt('ceph-disk')
- data_partition = c.get_osd_partition(osd_uuid)
- lockbox_partition = c.get_lockbox()
- c.sh("ceph-disk --verbose deactivate " + data_partition['path'])
- c.wait_for_osd_down(osd_uuid)
- c.sh("ceph-disk --verbose trigger --sync " + lockbox_partition['path'])
- c.sh("ceph-disk --verbose activate-journal " + data_partition['journal_dev'] +
- " --reactivate" + " --dmcrypt")
- c.wait_for_osd_up(osd_uuid)
- c.check_osd_status(osd_uuid, 'journal')
- c.destroy_osd(osd_uuid)
-
- def test_activate_dmcrypt_plain_no_lockbox(self):
- c = CephDisk()
- c.conf['global']['osd dmcrypt type'] = 'plain'
- c.save_conf()
- osd_uuid = self.activate_dmcrypt('ceph-disk-no-lockbox')
- c.destroy_osd(osd_uuid)
- c.save_conf()
-
- def test_activate_dmcrypt_luks_no_lockbox(self):
- c = CephDisk()
- osd_uuid = self.activate_dmcrypt('ceph-disk-no-lockbox')
- c.destroy_osd(osd_uuid)
-
- def test_activate_dmcrypt_luks_with_lockbox(self):
- c = CephDisk()
- osd_uuid = self.activate_dmcrypt('ceph-disk')
- c.destroy_osd(osd_uuid)
-
- def test_activate_lockbox(self):
- c = CephDisk()
- osd_uuid = self.activate_dmcrypt('ceph-disk')
- lockbox = c.get_lockbox()
- assert lockbox['state'] == 'active'
- c.sh("umount " + lockbox['path'])
- lockbox = c.get_lockbox()
- assert lockbox['state'] == 'prepared'
- c.sh("ceph-disk --verbose trigger " + lockbox['path'])
- success = False
- for delay in (1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024):
- lockbox = c.get_lockbox()
- if lockbox['state'] == 'active':
- success = True
- break
- time.sleep(delay)
- if not success:
- raise Exception('timeout waiting for lockbox ' + lockbox['path'])
- c.destroy_osd(osd_uuid)
-
- def activate_dmcrypt(self, ceph_disk):
- c = CephDisk()
- disk = c.unused_disks()[0]
- osd_uuid = str(uuid.uuid1())
- journal_uuid = str(uuid.uuid1())
- c.sh("ceph-disk --verbose zap " + disk)
- c.sh(ceph_disk + " --verbose prepare --filestore " +
- " --osd-uuid " + osd_uuid +
- " --journal-uuid " + journal_uuid +
- " --dmcrypt " +
- " " + disk)
- c.wait_for_osd_up(osd_uuid)
- c.check_osd_status(osd_uuid, 'journal')
- return osd_uuid
-
- def test_trigger_dmcrypt_journal_lockbox(self):
- c = CephDisk()
- osd_uuid = self.activate_dmcrypt('ceph-disk')
- data_partition = c.get_osd_partition(osd_uuid)
- lockbox_partition = c.get_lockbox()
- c.deactivate_osd(osd_uuid)
- c.wait_for_osd_down(osd_uuid)
- with pytest.raises(subprocess.CalledProcessError):
- # fails because the lockbox is not mounted yet
- c.sh("ceph-disk --verbose trigger --sync " + data_partition['journal_dev'])
- c.sh("ceph-disk --verbose trigger --sync " + lockbox_partition['path'])
- c.wait_for_osd_up(osd_uuid)
- c.destroy_osd(osd_uuid)
-
- def test_trigger_dmcrypt_data_lockbox(self):
- c = CephDisk()
- osd_uuid = self.activate_dmcrypt('ceph-disk')
- data_partition = c.get_osd_partition(osd_uuid)
- lockbox_partition = c.get_lockbox()
- c.deactivate_osd(osd_uuid)
- c.wait_for_osd_down(osd_uuid)
- with pytest.raises(subprocess.CalledProcessError):
- # fails because the lockbox is not mounted yet
- c.sh("ceph-disk --verbose trigger --sync " + data_partition['path'])
- c.sh("ceph-disk --verbose trigger --sync " + lockbox_partition['path'])
- c.wait_for_osd_up(osd_uuid)
- c.destroy_osd(osd_uuid)
-
- def test_trigger_dmcrypt_lockbox(self):
- c = CephDisk()
- osd_uuid = self.activate_dmcrypt('ceph-disk')
- data_partition = c.get_osd_partition(osd_uuid)
- lockbox_partition = c.get_lockbox()
- c.deactivate_osd(osd_uuid)
- c.wait_for_osd_down(osd_uuid)
- c.sh("ceph-disk --verbose trigger --sync " + lockbox_partition['path'])
- c.wait_for_osd_up(osd_uuid)
- c.destroy_osd(osd_uuid)
-
- def test_activate_no_journal(self):
- c = CephDisk()
- disk = c.unused_disks()[0]
- osd_uuid = str(uuid.uuid1())
- c.sh("ceph-disk --verbose zap " + disk)
- c.conf['global']['osd objectstore'] = 'memstore'
- c.save_conf()
- c.sh("ceph-disk --verbose prepare --filestore --osd-uuid " + osd_uuid +
- " " + disk)
- c.wait_for_osd_up(osd_uuid)
- device = json.loads(c.sh("ceph-disk list --format json " + disk))[0]
- assert len(device['partitions']) == 1
- partition = device['partitions'][0]
- assert partition['type'] == 'data'
- assert partition['state'] == 'active'
- assert 'journal_dev' not in partition
- c.helper("pool_read_write")
- c.destroy_osd(osd_uuid)
- c.save_conf()
-
- def test_activate_with_journal_dev_no_symlink(self):
- c = CephDisk()
- disk = c.unused_disks()[0]
- osd_uuid = str(uuid.uuid1())
- c.sh("ceph-disk --verbose zap " + disk)
- c.sh("ceph-disk --verbose prepare --filestore --osd-uuid " + osd_uuid +
- " " + disk)
- c.wait_for_osd_up(osd_uuid)
- device = json.loads(c.sh("ceph-disk list --format json " + disk))[0]
- assert len(device['partitions']) == 2
- c.check_osd_status(osd_uuid, 'journal')
- c.helper("pool_read_write")
- c.destroy_osd(osd_uuid)
-
- def test_activate_bluestore(self):
- c = CephDisk()
- disk = c.unused_disks()[0]
- osd_uuid = str(uuid.uuid1())
- c.sh("ceph-disk --verbose zap " + disk)
- c.conf['global']['osd objectstore'] = 'bluestore'
- c.save_conf()
- c.sh("ceph-disk --verbose prepare --bluestore --osd-uuid " + osd_uuid +
- " " + disk)
- c.wait_for_osd_up(osd_uuid)
- device = json.loads(c.sh("ceph-disk list --format json " + disk))[0]
- assert len(device['partitions']) == 2
- c.check_osd_status(osd_uuid, 'block')
- c.helper("pool_read_write")
- c.destroy_osd(osd_uuid)
- c.sh("ceph-disk --verbose zap " + disk)
-
- def test_activate_bluestore_seperated_block_db_wal(self):
- c = CephDisk()
- disk1 = c.unused_disks()[0]
- disk2 = c.unused_disks()[1]
- osd_uuid = str(uuid.uuid1())
- c.sh("ceph-disk --verbose zap " + disk1 + " " + disk2)
- c.conf['global']['osd objectstore'] = 'bluestore'
- c.save_conf()
- c.sh("ceph-disk --verbose prepare --bluestore --osd-uuid " + osd_uuid +
- " " + disk1 + " --block.db " + disk2 + " --block.wal " + disk2)
- c.wait_for_osd_up(osd_uuid)
- device = json.loads(c.sh("ceph-disk list --format json " + disk1))[0]
- assert len(device['partitions']) == 2
- device = json.loads(c.sh("ceph-disk list --format json " + disk2))[0]
- assert len(device['partitions']) == 2
- c.check_osd_status(osd_uuid, 'block')
- c.check_osd_status(osd_uuid, 'block.wal')
- c.check_osd_status(osd_uuid, 'block.db')
- c.helper("pool_read_write")
- c.destroy_osd(osd_uuid)
- c.sh("ceph-disk --verbose zap " + disk1 + " " + disk2)
-
- def test_activate_bluestore_reuse_db_wal_partition(self):
- c = CephDisk()
- disks = c.unused_disks()
- block_disk = disks[0]
- db_wal_disk = disks[1]
- #
- # Create an OSD with two disks (one for block,
- # the other for block.db and block.wal ) and then destroy osd.
- #
- osd_uuid1 = str(uuid.uuid1())
- c.sh("ceph-disk --verbose zap " + block_disk + " " + db_wal_disk)
- c.conf['global']['osd objectstore'] = 'bluestore'
- c.save_conf()
- c.sh("ceph-disk --verbose prepare --bluestore --osd-uuid " +
- osd_uuid1 + " " + block_disk + " --block.db " + db_wal_disk +
- " --block.wal " + db_wal_disk)
- c.wait_for_osd_up(osd_uuid1)
- blockdb_partition = c.get_blockdb_partition(osd_uuid1)
- blockdb_path = blockdb_partition['path']
- blockwal_partition = c.get_blockwal_partition(osd_uuid1)
- blockwal_path = blockwal_partition['path']
- c.destroy_osd(osd_uuid1)
- c.sh("ceph-disk --verbose zap " + block_disk)
- #
- # Create another OSD with the block.db and block.wal partition
- # of the previous OSD
- #
- osd_uuid2 = str(uuid.uuid1())
- c.sh("ceph-disk --verbose prepare --bluestore --osd-uuid " +
- osd_uuid2 + " " + block_disk + " --block.db " + blockdb_path +
- " --block.wal " + blockwal_path)
- c.wait_for_osd_up(osd_uuid2)
- device = json.loads(c.sh("ceph-disk list --format json " + block_disk))[0]
- assert len(device['partitions']) == 2
- device = json.loads(c.sh("ceph-disk list --format json " + db_wal_disk))[0]
- assert len(device['partitions']) == 2
- c.check_osd_status(osd_uuid2, 'block')
- c.check_osd_status(osd_uuid2, 'block.wal')
- c.check_osd_status(osd_uuid2, 'block.db')
- blockdb_partition = c.get_blockdb_partition(osd_uuid2)
- blockwal_partition = c.get_blockwal_partition(osd_uuid2)
- #
- # Verify the previous OSD partition has been reused
- #
- assert blockdb_partition['path'] == blockdb_path
- assert blockwal_partition['path'] == blockwal_path
- c.destroy_osd(osd_uuid2)
- c.sh("ceph-disk --verbose zap " + block_disk + " " + db_wal_disk)
-
- def test_activate_with_journal_dev_is_symlink(self):
- c = CephDisk()
- disk = c.unused_disks()[0]
- osd_uuid = str(uuid.uuid1())
- tempdir = tempfile.mkdtemp()
- symlink = os.path.join(tempdir, 'osd')
- os.symlink(disk, symlink)
- c.sh("ceph-disk --verbose zap " + symlink)
- c.sh("ceph-disk --verbose prepare --filestore --osd-uuid " + osd_uuid +
- " " + symlink)
- c.wait_for_osd_up(osd_uuid)
- device = json.loads(c.sh("ceph-disk list --format json " + symlink))[0]
- assert len(device['partitions']) == 2
- data_partition = c.get_osd_partition(osd_uuid)
- assert data_partition['type'] == 'data'
- assert data_partition['state'] == 'active'
- journal_partition = c.get_journal_partition(osd_uuid)
- assert journal_partition
- c.helper("pool_read_write")
- c.destroy_osd(osd_uuid)
- c.sh("ceph-disk --verbose zap " + symlink)
- os.unlink(symlink)
- os.rmdir(tempdir)
-
- def test_activate_journal_file(self):
- c = CephDisk()
- disks = c.unused_disks()
- data_disk = disks[0]
- #
- # /var/lib/ceph/osd is required otherwise it may violate
- # restrictions enforced by systemd regarding the directories
- # which ceph-osd is allowed to read/write
- #
- tempdir = tempfile.mkdtemp(dir='/var/lib/ceph/osd')
- c.sh("chown ceph:ceph " + tempdir + " || true")
- journal_file = os.path.join(tempdir, 'journal')
- osd_uuid = str(uuid.uuid1())
- c.sh("ceph-disk --verbose prepare --filestore --osd-uuid " + osd_uuid +
- " " + data_disk + " " + journal_file)
- c.wait_for_osd_up(osd_uuid)
- device = json.loads(
- c.sh("ceph-disk list --format json " + data_disk))[0]
- assert len(device['partitions']) == 1
- partition = device['partitions'][0]
- assert journal_file == os.readlink(
- os.path.join(partition['mount'], 'journal'))
- c.check_osd_status(osd_uuid)
- c.helper("pool_read_write 1") # 1 == pool size
- c.destroy_osd(osd_uuid)
- c.sh("ceph-disk --verbose zap " + data_disk)
- os.unlink(journal_file)
- os.rmdir(tempdir)
-
- def test_activate_separated_journal(self):
- c = CephDisk()
- disks = c.unused_disks()
- data_disk = disks[0]
- journal_disk = disks[1]
- osd_uuid = self.activate_separated_journal(data_disk, journal_disk)
- c.helper("pool_read_write 1") # 1 == pool size
- c.destroy_osd(osd_uuid)
- c.sh("ceph-disk --verbose zap " + data_disk + " " + journal_disk)
-
- def test_activate_separated_journal_dev_is_symlink(self):
- c = CephDisk()
- disks = c.unused_disks()
- data_disk = disks[0]
- journal_disk = disks[1]
- tempdir = tempfile.mkdtemp()
- data_symlink = os.path.join(tempdir, 'osd')
- os.symlink(data_disk, data_symlink)
- journal_symlink = os.path.join(tempdir, 'journal')
- os.symlink(journal_disk, journal_symlink)
- osd_uuid = self.activate_separated_journal(
- data_symlink, journal_symlink)
- c.helper("pool_read_write 1") # 1 == pool size
- c.destroy_osd(osd_uuid)
- c.sh("ceph-disk --verbose zap " + data_symlink + " " + journal_symlink)
- os.unlink(data_symlink)
- os.unlink(journal_symlink)
- os.rmdir(tempdir)
-
- def activate_separated_journal(self, data_disk, journal_disk):
- c = CephDisk()
- osd_uuid = str(uuid.uuid1())
- c.sh("ceph-disk --verbose prepare --filestore --osd-uuid " + osd_uuid +
- " " + data_disk + " " + journal_disk)
- c.wait_for_osd_up(osd_uuid)
- device = json.loads(
- c.sh("ceph-disk list --format json " + data_disk))[0]
- assert len(device['partitions']) == 1
- c.check_osd_status(osd_uuid, 'journal')
- return osd_uuid
-
- #
- # Create an OSD and get a journal partition from a disk that
- # already contains a journal partition which is in use. Updates of
- # the kernel partition table may behave differently when a
- # partition is in use. See http://tracker.ceph.com/issues/7334 for
- # more information.
- #
- def test_activate_two_separated_journal(self):
- c = CephDisk()
- disks = c.unused_disks()
- data_disk = disks[0]
- other_data_disk = disks[1]
- journal_disk = disks[2]
- osd_uuid = self.activate_separated_journal(data_disk, journal_disk)
- other_osd_uuid = self.activate_separated_journal(
- other_data_disk, journal_disk)
- #
- # read/write can only succeed if the two osds are up because
- # the pool needs two OSD
- #
- c.helper("pool_read_write 2") # 2 == pool size
- c.destroy_osd(osd_uuid)
- c.destroy_osd(other_osd_uuid)
- c.sh("ceph-disk --verbose zap " + data_disk + " " +
- journal_disk + " " + other_data_disk)
-
- #
- # Create an OSD and reuse an existing journal partition
- #
- def test_activate_reuse_journal(self):
- c = CephDisk()
- disks = c.unused_disks()
- data_disk = disks[0]
- journal_disk = disks[1]
- #
- # Create an OSD with a separated journal and destroy it.
- #
- osd_uuid = self.activate_separated_journal(data_disk, journal_disk)
- journal_partition = c.get_journal_partition(osd_uuid)
- journal_path = journal_partition['path']
- c.destroy_osd(osd_uuid)
- c.sh("ceph-disk --verbose zap " + data_disk)
- osd_uuid = str(uuid.uuid1())
- #
- # Create another OSD with the journal partition of the previous OSD
- #
- c.sh("ceph-disk --verbose prepare --filestore --osd-uuid " + osd_uuid +
- " " + data_disk + " " + journal_path)
- c.helper("pool_read_write 1") # 1 == pool size
- c.wait_for_osd_up(osd_uuid)
- device = json.loads(
- c.sh("ceph-disk list --format json " + data_disk))[0]
- assert len(device['partitions']) == 1
- c.check_osd_status(osd_uuid)
- journal_partition = c.get_journal_partition(osd_uuid)
- #
- # Verify the previous OSD partition has been reused
- #
- assert journal_partition['path'] == journal_path
- c.destroy_osd(osd_uuid)
- c.sh("ceph-disk --verbose zap " + data_disk + " " + journal_disk)
-
- def test_activate_multipath(self):
- c = CephDisk()
- if c.sh("lsb_release -si").strip() != 'CentOS':
- pytest.skip(
- "see issue https://bugs.launchpad.net/ubuntu/+source/multipath-tools/+bug/1488688")
- c.ensure_sd()
- #
- # Figure out the name of the multipath device
- #
- disk = c.unused_disks('sd.')[0]
- c.sh("mpathconf --enable || true")
- c.sh("multipath " + disk)
- holders = os.listdir(
- "/sys/block/" + os.path.basename(disk) + "/holders")
- assert 1 == len(holders)
- name = open("/sys/block/" + holders[0] + "/dm/name").read()
- multipath = "/dev/mapper/" + name
- #
- # Prepare the multipath device
- #
- osd_uuid = str(uuid.uuid1())
- c.sh("ceph-disk --verbose zap " + multipath)
- c.sh("ceph-disk --verbose prepare --filestore --osd-uuid " + osd_uuid +
- " " + multipath)
- c.wait_for_osd_up(osd_uuid)
- device = json.loads(
- c.sh("ceph-disk list --format json " + multipath))[0]
- assert len(device['partitions']) == 2
- data_partition = c.get_osd_partition(osd_uuid)
- assert data_partition['type'] == 'data'
- assert data_partition['state'] == 'active'
- journal_partition = c.get_journal_partition(osd_uuid)
- assert journal_partition
- c.helper("pool_read_write")
- c.destroy_osd(osd_uuid)
- c.sh("udevadm settle")
- c.sh("multipath -F")
- c.unload_scsi_debug()
-
-
-class CephDiskTest(CephDisk):
-
- def main(self, argv):
- parser = argparse.ArgumentParser(
- 'ceph-disk-test',
- )
- parser.add_argument(
- '-v', '--verbose',
- action='store_true', default=None,
- help='be more verbose',
- )
- parser.add_argument(
- '--destroy-osd',
- help='stop, umount and destroy',
- )
- args = parser.parse_args(argv)
-
- if args.verbose:
- logging.basicConfig(level=logging.DEBUG)
-
- if args.destroy_osd:
- dump = json.loads(CephDisk.sh("ceph osd dump -f json"))
- osd_uuid = None
- for osd in dump['osds']:
- if str(osd['osd']) == args.destroy_osd:
- osd_uuid = osd['uuid']
- if osd_uuid:
- self.destroy_osd(osd_uuid)
- else:
- raise Exception("cannot find OSD " + args.destroy_osd +
- " ceph osd dump -f json")
- return
-
-if __name__ == '__main__':
- sys.exit(CephDiskTest().main(sys.argv[1:]))
Code Review / stor4nfv.git / blobdiff