ceph/qa/workunits/ceph-disk/ceph-disk-test.py

   1 #
   2 # Copyright (C) 2015, 2016 Red Hat <contact@redhat.com>
   3 #
   4 # Author: Loic Dachary <loic@dachary.org>
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU Library Public License as published by
   8 # the Free Software Foundation; either version 2, or (at your option)
   9 # any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful,
  12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 # GNU Library Public License for more details.
  15 #
  16 # When debugging these tests (must be root), here are a few useful commands:
  17 #
  18 #  export PATH=.:..:$PATH
  19 #  ceph-disk.sh # run once to prepare the environment as it would be by teuthology
  20 #  ln -sf /home/ubuntu/ceph/src/ceph-disk/ceph_disk/main.py $(which ceph-disk)
  21 #  ln -sf /home/ubuntu/ceph/udev/95-ceph-osd.rules /lib/udev/rules.d/95-ceph-osd.rules
  22 #  ln -sf /home/ubuntu/ceph/systemd/ceph-disk@.service /usr/lib/systemd/system/ceph-disk@.service
  23 #  ceph-disk.conf will be silently ignored if it is a symbolic link or a hard link /var/log/upstart for logs
  24 #  cp /home/ubuntu/ceph/src/upstart/ceph-disk.conf /etc/init/ceph-disk.conf
  25 #  id=3 ; ceph-disk deactivate --deactivate-by-id $id ; ceph-disk destroy --zap --destroy-by-id $id
  26 #  py.test -s -v -k test_activate_dmcrypt_luks ceph-disk-test.py
  27 #
  28 #  CentOS 7
  29 #    udevadm monitor --property & tail -f /var/log/messages
  30 #    udev rules messages are logged in /var/log/messages
  31 #    systemctl stop ceph-osd@2
  32 #    systemctl start ceph-osd@2
  33 #
  34 #  udevadm monitor --property & tail -f /var/log/syslog /var/log/upstart/*  # on Ubuntu 14.04
  35 #  udevadm test --action=add /block/vdb/vdb1 # verify the udev rule is run as expected
  36 #  udevadm control --reload # when changing the udev rules
  37 #  sudo /usr/sbin/ceph-disk -v trigger /dev/vdb1 # activates if vdb1 is data
  38 #
  39 #  integration tests coverage
  40 #  pip install coverage
  41 #  perl -pi -e 's|"ceph-disk |"coverage run --source=/usr/sbin/ceph-disk --append /usr/sbin/ceph-disk |' ceph-disk-test.py
  42 #  rm -f .coverage ; py.test -s -v ceph-disk-test.py
  43 #  coverage report --show-missing
  44 #
  45 import argparse
  46 import json
  47 import logging
  48 import configobj
  49 import os
  50 import pytest
  51 import re
  52 import subprocess
  53 import sys
  54 import tempfile
  55 import time
  56 import uuid
  57
  58 LOG = logging.getLogger('CephDisk')
  59
  60
  61 class CephDisk:
  62
  63     def __init__(self):
  64         self.conf = configobj.ConfigObj('/etc/ceph/ceph.conf')
  65
  66     def save_conf(self):
  67         self.conf.write(open('/etc/ceph/ceph.conf', 'wb'))
  68
  69     @staticmethod
  70     def helper(command):
  71         command = "ceph-helpers-root.sh " + command
  72         return CephDisk.sh(command)
  73
  74     @staticmethod
  75     def sh(command):
  76         LOG.debug(":sh: " + command)
  77         proc = subprocess.Popen(
  78             args=command,
  79             stdout=subprocess.PIPE,
  80             stderr=subprocess.STDOUT,
  81             shell=True,
  82             bufsize=1)
  83         lines = []
  84         with proc.stdout:
  85             for line in iter(proc.stdout.readline, b''):
  86                 line = line.decode('utf-8')
  87                 if 'dangerous and experimental' in line:
  88                     LOG.debug('SKIP dangerous and experimental')
  89                     continue
  90                 lines.append(line)
  91                 LOG.debug(line.strip().encode('ascii', 'ignore'))
  92         if proc.wait() != 0:
  93             raise subprocess.CalledProcessError(
  94                 returncode=proc.returncode,
  95                 cmd=command
  96             )
  97         return "".join(lines)
  98
  99     def unused_disks(self, pattern='[vs]d.'):
 100         names = [x for x in os.listdir("/sys/block") if re.match(pattern, x)]
 101         if not names:
 102             return []
 103         disks = json.loads(
 104             self.sh("ceph-disk list --format json " + " ".join(names)))
 105         unused = []
 106         for disk in disks:
 107             if 'partitions' not in disk:
 108                 unused.append(disk['path'])
 109         return unused
 110
 111     def ensure_sd(self):
 112         LOG.debug(self.unused_disks('sd.'))
 113         if self.unused_disks('sd.'):
 114             return
 115         modprobe = "modprobe scsi_debug vpd_use_hostno=0 add_host=1 dev_size_mb=200 ; udevadm settle"
 116         try:
 117             self.sh(modprobe)
 118         except:
 119             self.helper("install linux-image-extra-3.13.0-61-generic")
 120             self.sh(modprobe)
 121
 122     def unload_scsi_debug(self):
 123         self.sh("rmmod scsi_debug || true")
 124
 125     def get_lockbox(self):
 126         disks = json.loads(self.sh("ceph-disk list --format json"))
 127         for disk in disks:
 128             if 'partitions' in disk:
 129                 for partition in disk['partitions']:
 130                     if partition.get('type') == 'lockbox':
 131                         return partition
 132         raise Exception("no lockbox found " + str(disks))
 133
 134     def get_osd_partition(self, uuid):
 135         disks = json.loads(self.sh("ceph-disk list --format json"))
 136         for disk in disks:
 137             if 'partitions' in disk:
 138                 for partition in disk['partitions']:
 139                     if partition.get('uuid') == uuid:
 140                         return partition
 141         raise Exception("uuid = " + uuid + " not found in " + str(disks))
 142
 143     def get_journal_partition(self, uuid):
 144         return self.get_space_partition('journal', uuid)
 145
 146     def get_block_partition(self, uuid):
 147         return self.get_space_partition('block', uuid)
 148
 149     def get_blockdb_partition(self, uuid):
 150         return self.get_space_partition('block.db', uuid)
 151
 152     def get_blockwal_partition(self, uuid):
 153         return self.get_space_partition('block.wal', uuid)
 154
 155     def get_space_partition(self, name, uuid):
 156         data_partition = self.get_osd_partition(uuid)
 157         space_dev = data_partition[name + '_dev']
 158         disks = json.loads(self.sh("ceph-disk list --format json"))
 159         for disk in disks:
 160             if 'partitions' in disk:
 161                 for partition in disk['partitions']:
 162                     if partition['path'] == space_dev:
 163                         if name + '_for' in partition:
 164                             assert partition[
 165                                 name + '_for'] == data_partition['path']
 166                         return partition
 167         raise Exception(
 168             name + " for uuid = " + uuid + " not found in " + str(disks))
 169
 170     def destroy_osd(self, uuid):
 171         id = self.sh("ceph osd create " + uuid).strip()
 172         self.sh("""
 173         set -xe
 174         ceph-disk --verbose deactivate --deactivate-by-id {id}
 175         ceph-disk --verbose destroy --destroy-by-id {id} --zap
 176         """.format(id=id))
 177
 178     def deactivate_osd(self, uuid):
 179         id = self.sh("ceph osd create " + uuid).strip()
 180         self.sh("""
 181         set -xe
 182         ceph-disk --verbose deactivate --once --deactivate-by-id {id}
 183         """.format(id=id))
 184
 185     @staticmethod
 186     def osd_up_predicate(osds, uuid):
 187         for osd in osds:
 188             if osd['uuid'] == uuid and 'up' in osd['state']:
 189                 return True
 190         return False
 191
 192     @staticmethod
 193     def wait_for_osd_up(uuid):
 194         CephDisk.wait_for_osd(uuid, CephDisk.osd_up_predicate, 'up')
 195
 196     @staticmethod
 197     def osd_down_predicate(osds, uuid):
 198         found = False
 199         for osd in osds:
 200             if osd['uuid'] == uuid:
 201                 found = True
 202                 if 'down' in osd['state'] or ['exists'] == osd['state']:
 203                     return True
 204         return not found
 205
 206     @staticmethod
 207     def wait_for_osd_down(uuid):
 208         CephDisk.wait_for_osd(uuid, CephDisk.osd_down_predicate, 'down')
 209
 210     @staticmethod
 211     def wait_for_osd(uuid, predicate, info):
 212         LOG.info("wait_for_osd " + info + " " + uuid)
 213         for delay in (1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024):
 214             dump = json.loads(CephDisk.sh("ceph osd dump -f json"))
 215             if predicate(dump['osds'], uuid):
 216                 return True
 217             time.sleep(delay)
 218         raise Exception('timeout waiting for osd ' + uuid + ' to be ' + info)
 219
 220     def check_osd_status(self, uuid, space_name=None):
 221         data_partition = self.get_osd_partition(uuid)
 222         assert data_partition['type'] == 'data'
 223         assert data_partition['state'] == 'active'
 224         if space_name is not None:
 225             space_partition = self.get_space_partition(space_name, uuid)
 226             assert space_partition
 227
 228
 229 class TestCephDisk(object):
 230
 231     def setup_class(self):
 232         logging.basicConfig(level=logging.DEBUG)
 233         c = CephDisk()
 234         if c.sh("lsb_release -si").strip() == 'CentOS':
 235             c.helper("install multipath-tools device-mapper-multipath")
 236         c.conf['global']['pid file'] = '/var/run/ceph/$cluster-$name.pid'
 237         #
 238         # Avoid json parsing interference
 239         #
 240         c.conf['global']['debug monc'] = 0
 241         #
 242         # objecstore
 243         #
 244         c.conf['global']['osd journal size'] = 100
 245         #
 246         # bluestore
 247         #
 248         c.conf['global']['enable experimental unrecoverable data corrupting features'] = '*'
 249         c.conf['global']['bluestore fsck on mount'] = 'true'
 250         c.save_conf()
 251
 252     def setup(self):
 253         c = CephDisk()
 254         for key in ('osd objectstore', 'osd dmcrypt type'):
 255             if key in c.conf['global']:
 256                 del c.conf['global'][key]
 257         c.save_conf()
 258
 259     def test_deactivate_reactivate_osd(self):
 260         c = CephDisk()
 261         disk = c.unused_disks()[0]
 262         osd_uuid = str(uuid.uuid1())
 263         c.sh("ceph-disk --verbose zap " + disk)
 264         c.sh("ceph-disk --verbose prepare --osd-uuid " + osd_uuid +
 265              " " + disk)
 266         c.wait_for_osd_up(osd_uuid)
 267         device = json.loads(c.sh("ceph-disk list --format json " + disk))[0]
 268         assert len(device['partitions']) == 2
 269         c.check_osd_status(osd_uuid, 'journal')
 270         data_partition = c.get_osd_partition(osd_uuid)
 271         c.sh("ceph-disk --verbose deactivate " + data_partition['path'])
 272         c.wait_for_osd_down(osd_uuid)
 273         c.sh("ceph-disk --verbose activate " + data_partition['path'] + " --reactivate")
 274         # check again
 275         c.wait_for_osd_up(osd_uuid)
 276         device = json.loads(c.sh("ceph-disk list --format json " + disk))[0]
 277         assert len(device['partitions']) == 2
 278         c.check_osd_status(osd_uuid, 'journal')
 279         c.helper("pool_read_write")
 280         c.destroy_osd(osd_uuid)
 281
 282     def test_destroy_osd_by_id(self):
 283         c = CephDisk()
 284         disk = c.unused_disks()[0]
 285         osd_uuid = str(uuid.uuid1())
 286         c.sh("ceph-disk --verbose prepare --osd-uuid " + osd_uuid + " " + disk)
 287         c.wait_for_osd_up(osd_uuid)
 288         c.check_osd_status(osd_uuid)
 289         c.destroy_osd(osd_uuid)
 290
 291     def test_destroy_osd_by_dev_path(self):
 292         c = CephDisk()
 293         disk = c.unused_disks()[0]
 294         osd_uuid = str(uuid.uuid1())
 295         c.sh("ceph-disk --verbose prepare --osd-uuid " + osd_uuid + " " + disk)
 296         c.wait_for_osd_up(osd_uuid)
 297         partition = c.get_osd_partition(osd_uuid)
 298         assert partition['type'] == 'data'
 299         assert partition['state'] == 'active'
 300         c.sh("ceph-disk --verbose deactivate " + partition['path'])
 301         c.wait_for_osd_down(osd_uuid)
 302         c.sh("ceph-disk --verbose destroy " + partition['path'] + " --zap")
 303
 304     def test_deactivate_reactivate_dmcrypt_plain(self):
 305         c = CephDisk()
 306         c.conf['global']['osd dmcrypt type'] = 'plain'
 307         c.save_conf()
 308         osd_uuid = self.activate_dmcrypt('ceph-disk-no-lockbox')
 309         data_partition = c.get_osd_partition(osd_uuid)
 310         c.sh("ceph-disk --verbose deactivate " + data_partition['path'])
 311         c.wait_for_osd_down(osd_uuid)
 312         c.sh("ceph-disk --verbose activate-journal " + data_partition['journal_dev'] +
 313              " --reactivate" + " --dmcrypt")
 314         c.wait_for_osd_up(osd_uuid)
 315         c.check_osd_status(osd_uuid, 'journal')
 316         c.destroy_osd(osd_uuid)
 317         c.save_conf()
 318
 319     def test_deactivate_reactivate_dmcrypt_luks(self):
 320         c = CephDisk()
 321         osd_uuid = self.activate_dmcrypt('ceph-disk')
 322         data_partition = c.get_osd_partition(osd_uuid)
 323         lockbox_partition = c.get_lockbox()
 324         c.sh("ceph-disk --verbose deactivate " + data_partition['path'])
 325         c.wait_for_osd_down(osd_uuid)
 326         c.sh("ceph-disk --verbose trigger --sync " + lockbox_partition['path'])
 327         c.sh("ceph-disk --verbose activate-journal " + data_partition['journal_dev'] +
 328              " --reactivate" + " --dmcrypt")
 329         c.wait_for_osd_up(osd_uuid)
 330         c.check_osd_status(osd_uuid, 'journal')
 331         c.destroy_osd(osd_uuid)
 332
 333     def test_activate_dmcrypt_plain_no_lockbox(self):
 334         c = CephDisk()
 335         c.conf['global']['osd dmcrypt type'] = 'plain'
 336         c.save_conf()
 337         osd_uuid = self.activate_dmcrypt('ceph-disk-no-lockbox')
 338         c.destroy_osd(osd_uuid)
 339         c.save_conf()
 340
 341     def test_activate_dmcrypt_luks_no_lockbox(self):
 342         c = CephDisk()
 343         osd_uuid = self.activate_dmcrypt('ceph-disk-no-lockbox')
 344         c.destroy_osd(osd_uuid)
 345
 346     def test_activate_dmcrypt_luks_with_lockbox(self):
 347         c = CephDisk()
 348         osd_uuid = self.activate_dmcrypt('ceph-disk')
 349         c.destroy_osd(osd_uuid)
 350
 351     def test_activate_lockbox(self):
 352         c = CephDisk()
 353         osd_uuid = self.activate_dmcrypt('ceph-disk')
 354         lockbox = c.get_lockbox()
 355         assert lockbox['state'] == 'active'
 356         c.sh("umount " + lockbox['path'])
 357         lockbox = c.get_lockbox()
 358         assert lockbox['state'] == 'prepared'
 359         c.sh("ceph-disk --verbose trigger " + lockbox['path'])
 360         success = False
 361         for delay in (1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024):
 362             lockbox = c.get_lockbox()
 363             if lockbox['state'] == 'active':
 364                 success = True
 365                 break
 366             time.sleep(delay)
 367         if not success:
 368             raise Exception('timeout waiting for lockbox ' + lockbox['path'])
 369         c.destroy_osd(osd_uuid)
 370
 371     def activate_dmcrypt(self, ceph_disk):
 372         c = CephDisk()
 373         disk = c.unused_disks()[0]
 374         osd_uuid = str(uuid.uuid1())
 375         journal_uuid = str(uuid.uuid1())
 376         c.sh("ceph-disk --verbose zap " + disk)
 377         c.sh(ceph_disk + " --verbose prepare " +
 378              " --osd-uuid " + osd_uuid +
 379              " --journal-uuid " + journal_uuid +
 380              " --dmcrypt " +
 381              " " + disk)
 382         c.wait_for_osd_up(osd_uuid)
 383         c.check_osd_status(osd_uuid, 'journal')
 384         return osd_uuid
 385
 386     def test_trigger_dmcrypt_journal_lockbox(self):
 387         c = CephDisk()
 388         osd_uuid = self.activate_dmcrypt('ceph-disk')
 389         data_partition = c.get_osd_partition(osd_uuid)
 390         lockbox_partition = c.get_lockbox()
 391         c.deactivate_osd(osd_uuid)
 392         c.wait_for_osd_down(osd_uuid)
 393         with pytest.raises(subprocess.CalledProcessError):
 394             # fails because the lockbox is not mounted yet
 395             c.sh("ceph-disk --verbose trigger --sync " + data_partition['journal_dev'])
 396         c.sh("ceph-disk --verbose trigger --sync " + lockbox_partition['path'])
 397         c.wait_for_osd_up(osd_uuid)
 398         c.destroy_osd(osd_uuid)
 399
 400     def test_trigger_dmcrypt_data_lockbox(self):
 401         c = CephDisk()
 402         osd_uuid = self.activate_dmcrypt('ceph-disk')
 403         data_partition = c.get_osd_partition(osd_uuid)
 404         lockbox_partition = c.get_lockbox()
 405         c.deactivate_osd(osd_uuid)
 406         c.wait_for_osd_down(osd_uuid)
 407         with pytest.raises(subprocess.CalledProcessError):
 408             # fails because the lockbox is not mounted yet
 409             c.sh("ceph-disk --verbose trigger --sync " + data_partition['path'])
 410         c.sh("ceph-disk --verbose trigger --sync " + lockbox_partition['path'])
 411         c.wait_for_osd_up(osd_uuid)
 412         c.destroy_osd(osd_uuid)
 413
 414     def test_trigger_dmcrypt_lockbox(self):
 415         c = CephDisk()
 416         osd_uuid = self.activate_dmcrypt('ceph-disk')
 417         data_partition = c.get_osd_partition(osd_uuid)
 418         lockbox_partition = c.get_lockbox()
 419         c.deactivate_osd(osd_uuid)
 420         c.wait_for_osd_down(osd_uuid)
 421         c.sh("ceph-disk --verbose trigger --sync " + lockbox_partition['path'])
 422         c.wait_for_osd_up(osd_uuid)
 423         c.destroy_osd(osd_uuid)
 424
 425     def test_activate_no_journal(self):
 426         c = CephDisk()
 427         disk = c.unused_disks()[0]
 428         osd_uuid = str(uuid.uuid1())
 429         c.sh("ceph-disk --verbose zap " + disk)
 430         c.conf['global']['osd objectstore'] = 'memstore'
 431         c.save_conf()
 432         c.sh("ceph-disk --verbose prepare --osd-uuid " + osd_uuid +
 433              " " + disk)
 434         c.wait_for_osd_up(osd_uuid)
 435         device = json.loads(c.sh("ceph-disk list --format json " + disk))[0]
 436         assert len(device['partitions']) == 1
 437         partition = device['partitions'][0]
 438         assert partition['type'] == 'data'
 439         assert partition['state'] == 'active'
 440         assert 'journal_dev' not in partition
 441         c.helper("pool_read_write")
 442         c.destroy_osd(osd_uuid)
 443         c.save_conf()
 444
 445     def test_activate_with_journal_dev_no_symlink(self):
 446         c = CephDisk()
 447         disk = c.unused_disks()[0]
 448         osd_uuid = str(uuid.uuid1())
 449         c.sh("ceph-disk --verbose zap " + disk)
 450         c.sh("ceph-disk --verbose prepare --osd-uuid " + osd_uuid +
 451              " " + disk)
 452         c.wait_for_osd_up(osd_uuid)
 453         device = json.loads(c.sh("ceph-disk list --format json " + disk))[0]
 454         assert len(device['partitions']) == 2
 455         c.check_osd_status(osd_uuid, 'journal')
 456         c.helper("pool_read_write")
 457         c.destroy_osd(osd_uuid)
 458
 459     def test_activate_bluestore(self):
 460         c = CephDisk()
 461         disk = c.unused_disks()[0]
 462         osd_uuid = str(uuid.uuid1())
 463         c.sh("ceph-disk --verbose zap " + disk)
 464         c.conf['global']['osd objectstore'] = 'bluestore'
 465         c.save_conf()
 466         c.sh("ceph-disk --verbose prepare --bluestore --osd-uuid " + osd_uuid +
 467              " " + disk)
 468         c.wait_for_osd_up(osd_uuid)
 469         device = json.loads(c.sh("ceph-disk list --format json " + disk))[0]
 470         assert len(device['partitions']) == 2
 471         c.check_osd_status(osd_uuid, 'block')
 472         c.helper("pool_read_write")
 473         c.destroy_osd(osd_uuid)
 474         c.sh("ceph-disk --verbose zap " + disk)
 475
 476     def test_activate_bluestore_seperated_block_db_wal(self):
 477         c = CephDisk()
 478         disk1 = c.unused_disks()[0]
 479         disk2 = c.unused_disks()[1]
 480         osd_uuid = str(uuid.uuid1())
 481         c.sh("ceph-disk --verbose zap " + disk1 + " " + disk2)
 482         c.conf['global']['osd objectstore'] = 'bluestore'
 483         c.save_conf()
 484         c.sh("ceph-disk --verbose prepare --bluestore --osd-uuid " + osd_uuid +
 485              " " + disk1 + " --block.db " + disk2 + " --block.wal " + disk2)
 486         c.wait_for_osd_up(osd_uuid)
 487         device = json.loads(c.sh("ceph-disk list --format json " + disk1))[0]
 488         assert len(device['partitions']) == 2
 489         device = json.loads(c.sh("ceph-disk list --format json " + disk2))[0]
 490         assert len(device['partitions']) == 2
 491         c.check_osd_status(osd_uuid, 'block')
 492         c.check_osd_status(osd_uuid, 'block.wal')
 493         c.check_osd_status(osd_uuid, 'block.db')
 494         c.helper("pool_read_write")
 495         c.destroy_osd(osd_uuid)
 496         c.sh("ceph-disk --verbose zap " + disk1 + " " + disk2)
 497
 498     def test_activate_bluestore_reuse_db_wal_partition(self):
 499         c = CephDisk()
 500         disks = c.unused_disks()
 501         block_disk = disks[0]
 502         db_wal_disk = disks[1]
 503         #
 504         # Create an OSD with two disks (one for block,
 505         # the other for block.db and block.wal ) and then destroy osd.
 506         #
 507         osd_uuid1 = str(uuid.uuid1())
 508         c.sh("ceph-disk --verbose zap " + block_disk + " " + db_wal_disk)
 509         c.conf['global']['osd objectstore'] = 'bluestore'
 510         c.save_conf()
 511         c.sh("ceph-disk --verbose prepare --bluestore --osd-uuid " +
 512              osd_uuid1 + " " + block_disk + " --block.db " + db_wal_disk +
 513              " --block.wal " + db_wal_disk)
 514         c.wait_for_osd_up(osd_uuid1)
 515         blockdb_partition = c.get_blockdb_partition(osd_uuid1)
 516         blockdb_path = blockdb_partition['path']
 517         blockwal_partition = c.get_blockwal_partition(osd_uuid1)
 518         blockwal_path = blockwal_partition['path']
 519         c.destroy_osd(osd_uuid1)
 520         c.sh("ceph-disk --verbose zap " + block_disk)
 521         #
 522         # Create another OSD with the block.db and block.wal partition
 523         # of the previous OSD
 524         #
 525         osd_uuid2 = str(uuid.uuid1())
 526         c.sh("ceph-disk --verbose prepare --bluestore --osd-uuid " +
 527              osd_uuid2 + " " + block_disk + " --block.db " + blockdb_path +
 528              " --block.wal " + blockwal_path)
 529         c.wait_for_osd_up(osd_uuid2)
 530         device = json.loads(c.sh("ceph-disk list --format json " + block_disk))[0]
 531         assert len(device['partitions']) == 2
 532         device = json.loads(c.sh("ceph-disk list --format json " + db_wal_disk))[0]
 533         assert len(device['partitions']) == 2
 534         c.check_osd_status(osd_uuid2, 'block')
 535         c.check_osd_status(osd_uuid2, 'block.wal')
 536         c.check_osd_status(osd_uuid2, 'block.db')
 537         blockdb_partition = c.get_blockdb_partition(osd_uuid2)
 538         blockwal_partition = c.get_blockwal_partition(osd_uuid2)
 539         #
 540         # Verify the previous OSD partition has been reused
 541         #
 542         assert blockdb_partition['path'] == blockdb_path
 543         assert blockwal_partition['path'] == blockwal_path
 544         c.destroy_osd(osd_uuid2)
 545         c.sh("ceph-disk --verbose zap " + block_disk + " " + db_wal_disk)
 546
 547     def test_activate_with_journal_dev_is_symlink(self):
 548         c = CephDisk()
 549         disk = c.unused_disks()[0]
 550         osd_uuid = str(uuid.uuid1())
 551         tempdir = tempfile.mkdtemp()
 552         symlink = os.path.join(tempdir, 'osd')
 553         os.symlink(disk, symlink)
 554         c.sh("ceph-disk --verbose zap " + symlink)
 555         c.sh("ceph-disk --verbose prepare --osd-uuid " + osd_uuid +
 556              " " + symlink)
 557         c.wait_for_osd_up(osd_uuid)
 558         device = json.loads(c.sh("ceph-disk list --format json " + symlink))[0]
 559         assert len(device['partitions']) == 2
 560         data_partition = c.get_osd_partition(osd_uuid)
 561         assert data_partition['type'] == 'data'
 562         assert data_partition['state'] == 'active'
 563         journal_partition = c.get_journal_partition(osd_uuid)
 564         assert journal_partition
 565         c.helper("pool_read_write")
 566         c.destroy_osd(osd_uuid)
 567         c.sh("ceph-disk --verbose zap " + symlink)
 568         os.unlink(symlink)
 569         os.rmdir(tempdir)
 570
 571     def test_activate_journal_file(self):
 572         c = CephDisk()
 573         disks = c.unused_disks()
 574         data_disk = disks[0]
 575         #
 576         # /var/lib/ceph/osd is required otherwise it may violate
 577         # restrictions enforced by systemd regarding the directories
 578         # which ceph-osd is allowed to read/write
 579         #
 580         tempdir = tempfile.mkdtemp(dir='/var/lib/ceph/osd')
 581         c.sh("chown ceph:ceph " + tempdir + " || true")
 582         journal_file = os.path.join(tempdir, 'journal')
 583         osd_uuid = str(uuid.uuid1())
 584         c.sh("ceph-disk --verbose prepare --osd-uuid " + osd_uuid +
 585              " " + data_disk + " " + journal_file)
 586         c.wait_for_osd_up(osd_uuid)
 587         device = json.loads(
 588             c.sh("ceph-disk list --format json " + data_disk))[0]
 589         assert len(device['partitions']) == 1
 590         partition = device['partitions'][0]
 591         assert journal_file == os.readlink(
 592             os.path.join(partition['mount'], 'journal'))
 593         c.check_osd_status(osd_uuid)
 594         c.helper("pool_read_write 1")  # 1 == pool size
 595         c.destroy_osd(osd_uuid)
 596         c.sh("ceph-disk --verbose zap " + data_disk)
 597         os.unlink(journal_file)
 598         os.rmdir(tempdir)
 599
 600     def test_activate_separated_journal(self):
 601         c = CephDisk()
 602         disks = c.unused_disks()
 603         data_disk = disks[0]
 604         journal_disk = disks[1]
 605         osd_uuid = self.activate_separated_journal(data_disk, journal_disk)
 606         c.helper("pool_read_write 1")  # 1 == pool size
 607         c.destroy_osd(osd_uuid)
 608         c.sh("ceph-disk --verbose zap " + data_disk + " " + journal_disk)
 609
 610     def test_activate_separated_journal_dev_is_symlink(self):
 611         c = CephDisk()
 612         disks = c.unused_disks()
 613         data_disk = disks[0]
 614         journal_disk = disks[1]
 615         tempdir = tempfile.mkdtemp()
 616         data_symlink = os.path.join(tempdir, 'osd')
 617         os.symlink(data_disk, data_symlink)
 618         journal_symlink = os.path.join(tempdir, 'journal')
 619         os.symlink(journal_disk, journal_symlink)
 620         osd_uuid = self.activate_separated_journal(
 621             data_symlink, journal_symlink)
 622         c.helper("pool_read_write 1")  # 1 == pool size
 623         c.destroy_osd(osd_uuid)
 624         c.sh("ceph-disk --verbose zap " + data_symlink + " " + journal_symlink)
 625         os.unlink(data_symlink)
 626         os.unlink(journal_symlink)
 627         os.rmdir(tempdir)
 628
 629     def activate_separated_journal(self, data_disk, journal_disk):
 630         c = CephDisk()
 631         osd_uuid = str(uuid.uuid1())
 632         c.sh("ceph-disk --verbose prepare --osd-uuid " + osd_uuid +
 633              " " + data_disk + " " + journal_disk)
 634         c.wait_for_osd_up(osd_uuid)
 635         device = json.loads(
 636             c.sh("ceph-disk list --format json " + data_disk))[0]
 637         assert len(device['partitions']) == 1
 638         c.check_osd_status(osd_uuid, 'journal')
 639         return osd_uuid
 640
 641     #
 642     # Create an OSD and get a journal partition from a disk that
 643     # already contains a journal partition which is in use. Updates of
 644     # the kernel partition table may behave differently when a
 645     # partition is in use. See http://tracker.ceph.com/issues/7334 for
 646     # more information.
 647     #
 648     def test_activate_two_separated_journal(self):
 649         c = CephDisk()
 650         disks = c.unused_disks()
 651         data_disk = disks[0]
 652         other_data_disk = disks[1]
 653         journal_disk = disks[2]
 654         osd_uuid = self.activate_separated_journal(data_disk, journal_disk)
 655         other_osd_uuid = self.activate_separated_journal(
 656             other_data_disk, journal_disk)
 657         #
 658         # read/write can only succeed if the two osds are up because
 659         # the pool needs two OSD
 660         #
 661         c.helper("pool_read_write 2")  # 2 == pool size
 662         c.destroy_osd(osd_uuid)
 663         c.destroy_osd(other_osd_uuid)
 664         c.sh("ceph-disk --verbose zap " + data_disk + " " +
 665              journal_disk + " " + other_data_disk)
 666
 667     #
 668     # Create an OSD and reuse an existing journal partition
 669     #
 670     def test_activate_reuse_journal(self):
 671         c = CephDisk()
 672         disks = c.unused_disks()
 673         data_disk = disks[0]
 674         journal_disk = disks[1]
 675         #
 676         # Create an OSD with a separated journal and destroy it.
 677         #
 678         osd_uuid = self.activate_separated_journal(data_disk, journal_disk)
 679         journal_partition = c.get_journal_partition(osd_uuid)
 680         journal_path = journal_partition['path']
 681         c.destroy_osd(osd_uuid)
 682         c.sh("ceph-disk --verbose zap " + data_disk)
 683         osd_uuid = str(uuid.uuid1())
 684         #
 685         # Create another OSD with the journal partition of the previous OSD
 686         #
 687         c.sh("ceph-disk --verbose prepare --osd-uuid " + osd_uuid +
 688              " " + data_disk + " " + journal_path)
 689         c.helper("pool_read_write 1")  # 1 == pool size
 690         c.wait_for_osd_up(osd_uuid)
 691         device = json.loads(
 692             c.sh("ceph-disk list --format json " + data_disk))[0]
 693         assert len(device['partitions']) == 1
 694         c.check_osd_status(osd_uuid)
 695         journal_partition = c.get_journal_partition(osd_uuid)
 696         #
 697         # Verify the previous OSD partition has been reused
 698         #
 699         assert journal_partition['path'] == journal_path
 700         c.destroy_osd(osd_uuid)
 701         c.sh("ceph-disk --verbose zap " + data_disk + " " + journal_disk)
 702
 703     def test_activate_multipath(self):
 704         c = CephDisk()
 705         if c.sh("lsb_release -si").strip() != 'CentOS':
 706             pytest.skip(
 707                 "see issue https://bugs.launchpad.net/ubuntu/+source/multipath-tools/+bug/1488688")
 708         c.ensure_sd()
 709         #
 710         # Figure out the name of the multipath device
 711         #
 712         disk = c.unused_disks('sd.')[0]
 713         c.sh("mpathconf --enable || true")
 714         c.sh("multipath " + disk)
 715         holders = os.listdir(
 716             "/sys/block/" + os.path.basename(disk) + "/holders")
 717         assert 1 == len(holders)
 718         name = open("/sys/block/" + holders[0] + "/dm/name").read()
 719         multipath = "/dev/mapper/" + name
 720         #
 721         # Prepare the multipath device
 722         #
 723         osd_uuid = str(uuid.uuid1())
 724         c.sh("ceph-disk --verbose zap " + multipath)
 725         c.sh("ceph-disk --verbose prepare --osd-uuid " + osd_uuid +
 726              " " + multipath)
 727         c.wait_for_osd_up(osd_uuid)
 728         device = json.loads(
 729             c.sh("ceph-disk list --format json " + multipath))[0]
 730         assert len(device['partitions']) == 2
 731         data_partition = c.get_osd_partition(osd_uuid)
 732         assert data_partition['type'] == 'data'
 733         assert data_partition['state'] == 'active'
 734         journal_partition = c.get_journal_partition(osd_uuid)
 735         assert journal_partition
 736         c.helper("pool_read_write")
 737         c.destroy_osd(osd_uuid)
 738         c.sh("udevadm settle")
 739         c.sh("multipath -F")
 740         c.unload_scsi_debug()
 741
 742
 743 class CephDiskTest(CephDisk):
 744
 745     def main(self, argv):
 746         parser = argparse.ArgumentParser(
 747             'ceph-disk-test',
 748         )
 749         parser.add_argument(
 750             '-v', '--verbose',
 751             action='store_true', default=None,
 752             help='be more verbose',
 753         )
 754         parser.add_argument(
 755             '--destroy-osd',
 756             help='stop, umount and destroy',
 757         )
 758         args = parser.parse_args(argv)
 759
 760         if args.verbose:
 761             logging.basicConfig(level=logging.DEBUG)
 762
 763         if args.destroy_osd:
 764             dump = json.loads(CephDisk.sh("ceph osd dump -f json"))
 765             osd_uuid = None
 766             for osd in dump['osds']:
 767                 if str(osd['osd']) == args.destroy_osd:
 768                     osd_uuid = osd['uuid']
 769             if osd_uuid:
 770                 self.destroy_osd(osd_uuid)
 771             else:
 772                 raise Exception("cannot find OSD " + args.destroy_osd +
 773                                 " ceph osd dump -f json")
 774             return
 775
 776 if __name__ == '__main__':
 777     sys.exit(CephDiskTest().main(sys.argv[1:]))