"""

defaults = dict(

osd=dict(

mon='allow profile osd',

mgr='allow profile osd',

osd='allow *',

),

mgr=dict(

mon='allow profile mgr',

osd='allow *',

mds='allow *',

),

mds=dict(

mon='allow *',

mgr='allow *',

osd='allow *',

mds='allow',

),

client=dict(

mon='allow rw',

mgr='allow r',

osd='allow rwx',

mds='allow',

),

)

for subsystem, capability in defaults[type_].items():

yield '--cap'

yield subsystem

"""

if ctx.archive is None:

return

with open(os.path.join(ctx.archive, 'info.yaml'), 'r+') as info_file:

info_yaml = yaml.safe_load(info_file)

info_file.seek(0)

if 'archive' in info_yaml:

info_yaml['archive'][key] = value

else:

info_yaml['archive'] = {key: value}

"""

# Add crash directory to job's archive

update_archive_setting(ctx, 'crash', '/var/lib/ceph/crash')

try:

yield

finally:

if ctx.archive is not None:

log.info('Archiving crash dumps...')

path = os.path.join(ctx.archive, 'remote')

try:

os.makedirs(path)

except OSError:

pass

for remote in ctx.cluster.remotes.keys():

sub = os.path.join(path, remote.shortname)

try:

os.makedirs(sub)

except OSError:

pass

try:

teuthology.pull_directory(remote, '/var/lib/ceph/crash',

os.path.join(sub, 'crash'))

except ReadError:

"""

log.info('Making ceph log dir writeable by non-root...')

run.wait(

ctx.cluster.run(

args=[

'sudo',

'chmod',

'777',

'/var/log/ceph',

],

wait=False,

)

)

log.info('Disabling ceph logrotate...')

run.wait(

ctx.cluster.run(

args=[

'sudo',

'rm', '-f', '--',

'/etc/logrotate.d/ceph',

],

wait=False,

)

)

log.info('Creating extra log directories...')

run.wait(

ctx.cluster.run(

args=[

'sudo',

'install', '-d', '-m0777', '--',

'/var/log/ceph/valgrind',

'/var/log/ceph/profiling-logger',

],

wait=False,

)

)

# Add logs directory to job's info log file

update_archive_setting(ctx, 'log', '/var/log/ceph')

class Rotater(object):

stop_event = gevent.event.Event()

def invoke_logrotate(self):

# 1) install ceph-test.conf in /etc/logrotate.d

# 2) continuously loop over logrotate invocation with ceph-test.conf

while not self.stop_event.is_set():

self.stop_event.wait(timeout=30)

try:

procs = ctx.cluster.run(

args=['sudo', 'logrotate', '/etc/logrotate.d/ceph-test.conf'],

wait=False,

stderr=StringIO()

)

run.wait(procs)

except exceptions.ConnectionLostError as e:

# Some tests may power off nodes during test, in which

# case we will see connection errors that we should ignore.

log.debug("Missed logrotate, node '{0}' is offline".format(

e.node))

except EOFError:

# Paramiko sometimes raises this when it fails to

# connect to a node during open_session. As with

# ConnectionLostError, we ignore this because nodes

# are allowed to get power cycled during tests.

log.debug("Missed logrotate, EOFError")

except SSHException:

log.debug("Missed logrotate, SSHException")

except run.CommandFailedError as e:

for p in procs:

if p.finished and p.exitstatus != 0:

err = p.stderr.getvalue()

if 'error: error renaming temp state file' in err:

log.info('ignoring transient state error: %s', e)

else:

raise

except socket.error as e:

if e.errno in (errno.EHOSTUNREACH, errno.ECONNRESET):

log.debug("Missed logrotate, host unreachable")

else:

raise

def begin(self):

self.thread = gevent.spawn(self.invoke_logrotate)

def end(self):

self.stop_event.set()

self.thread.get()

def write_rotate_conf(ctx, daemons):

testdir = teuthology.get_testdir(ctx)

remote_logrotate_conf = '%s/logrotate.ceph-test.conf' % testdir

rotate_conf_path = os.path.join(os.path.dirname(__file__), 'logrotate.conf')

with open(rotate_conf_path) as f:

conf = ""

for daemon, size in daemons.items():

log.info('writing logrotate stanza for {}'.format(daemon))

conf += f.read().format(daemon_type=daemon,

max_size=size)

f.seek(0, 0)

for remote in ctx.cluster.remotes.keys():

remote.write_file(remote_logrotate_conf, BytesIO(conf.encode()))

remote.sh(

f'sudo mv {remote_logrotate_conf} /etc/logrotate.d/ceph-test.conf && '

'sudo chmod 0644 /etc/logrotate.d/ceph-test.conf && '

'sudo chown root.root /etc/logrotate.d/ceph-test.conf')

remote.chcon('/etc/logrotate.d/ceph-test.conf',

'system_u:object_r:etc_t:s0')

if ctx.config.get('log-rotate'):

daemons = ctx.config.get('log-rotate')

log.info('Setting up log rotation with ' + str(daemons))

write_rotate_conf(ctx, daemons)

logrotater = Rotater()

logrotater.begin()

try:

yield

finally:

if ctx.config.get('log-rotate'):

log.info('Shutting down logrotate')

logrotater.end()

ctx.cluster.sh('sudo rm /etc/logrotate.d/ceph-test.conf')

if ctx.archive is not None and \

not (ctx.config.get('archive-on-error') and ctx.summary['success']):

# and logs

log.info('Compressing logs...')

run.wait(

ctx.cluster.run(

args=[

'time',

'sudo',

'find',

'/var/log/ceph',

'-name',

'*.log',

'-print0',

run.Raw('|'),

'sudo',

'xargs',

'--max-args=1',

'--max-procs=0',

'--verbose',

'-0',

'--no-run-if-empty',

'--',

'gzip',

'-5',

'--verbose',

'--',

],

wait=False,

),

)

log.info('Archiving logs...')

path = os.path.join(ctx.archive, 'remote')

try:

os.makedirs(path)

except OSError:

pass

for remote in ctx.cluster.remotes.keys():

sub = os.path.join(path, remote.shortname)

try:

os.makedirs(sub)

except OSError:

pass

teuthology.pull_directory(remote, '/var/log/ceph',

"""

"""

try:

yield

finally:

valgrind_exception = None

valgrind_yaml = os.path.join(ctx.archive, 'valgrind.yaml')

for remote in ctx.cluster.remotes.keys():

scanner = ValgrindScanner(remote)

errors = scanner.scan_all_files('/var/log/ceph/valgrind/*')

scanner.write_summary(valgrind_yaml)

if errors and not valgrind_exception:

log.debug('valgrind exception message: %s', errors[0])

valgrind_exception = Exception(errors[0])

if config.get('expect_valgrind_errors'):

if not valgrind_exception:

raise Exception('expected valgrind issues and found none')

else:

if valgrind_exception:

cluster_name = config['cluster']

first_mon = teuthology.get_first_mon(ctx, config, cluster_name)

(mon_remote,) = ctx.cluster.only(first_mon).remotes.keys()

profile = config.get('crush_tunables', 'default')

log.info('Setting crush tunables to %s', profile)

mon_remote.run(

args=['sudo', 'ceph', '--cluster', cluster_name,

'osd', 'crush', 'tunables', profile])

cluster_name = config['cluster']

first_mon = teuthology.get_first_mon(ctx, config, cluster_name)

(mon_remote,) = ctx.cluster.only(first_mon).remotes.keys()

modules = config.get('mgr-modules', [])

for m in modules:

m = str(m)

cmd = [

'sudo',

'ceph',

'--cluster',

cluster_name,

'mgr',

'module',

'enable',

m,

]

log.info("enabling module %s", m)

mon_remote.run(args=cmd)

cluster_name = config['cluster']

first_mon = teuthology.get_first_mon(ctx, config, cluster_name)

(mon_remote,) = ctx.cluster.only(first_mon).remotes.keys()

configs = config.get('cluster-conf', {})

procs = []

for section, confs in configs.items():

section = str(section)

for k, v in confs.items():

k = str(k).replace(' ', '_') # pre-pacific compatibility

v = str(v)

cmd = [

'sudo',

'ceph',

'--cluster',

cluster_name,

'config',

'set',

section,

k,

v,

]

log.info("setting config [%s] %s = %s", section, k, v)

procs.append(mon_remote.run(args=cmd, wait=False))

log.debug("set %d configs", len(procs))

for p in procs:

log.debug("waiting for %s", p)

p.wait()

cmd = [

'sudo',

'ceph',

'--cluster',

cluster_name,

'config',

'dump',

]

mon_remote.run(args=cmd)

# enable crimson-osds if crimson

log.info("check_enable_crimson: {}".format(is_crimson(config)))

if is_crimson(config):

cluster_name = config['cluster']

first_mon = teuthology.get_first_mon(ctx, config, cluster_name)

(mon_remote,) = ctx.cluster.only(first_mon).remotes.keys()

log.info('check_enable_crimson: setting set-allow-crimson')

mon_remote.run(

args=[

'sudo', 'ceph', '--cluster', cluster_name,

'osd', 'set-allow-crimson', '--yes-i-really-mean-it'

]

)

first_mon = teuthology.get_first_mon(ctx, config, config['cluster'])

(mon,) = ctx.cluster.only(first_mon).remotes.keys()

if not hasattr(ctx, 'managers'):

ctx.managers = {}

ctx.managers[config['cluster']] = CephManager(

mon,

ctx=ctx,

logger=log.getChild('ceph_manager.' + config['cluster']),

cluster=config['cluster'],

)

cluster_name = config['cluster']

first_mon = teuthology.get_first_mon(ctx, config, cluster_name)

(mon_remote,) = ctx.cluster.only(first_mon).remotes.keys()

log.info('Waiting for OSDs to come up')

teuthology.wait_until_osds_up(

ctx,

cluster=ctx.cluster,

remote=mon_remote,

ceph_cluster=cluster_name,

)

if config.get('create_rbd_pool', True):

log.info('Creating RBD pool')

mon_remote.run(

args=['sudo', 'ceph', '--cluster', cluster_name,

'osd', 'pool', 'create', 'rbd', '8'])

mon_remote.run(

args=['rbd', '--cluster', cluster_name, 'pool', 'init', 'rbd'])

cluster_name = config['cluster']

first_mon = teuthology.get_first_mon(ctx, config, cluster_name)

(mon_remote,) = ctx.cluster.only(first_mon).remotes.keys()

mdss = ctx.cluster.only(teuthology.is_type('mds', cluster_name))

# If there are any MDSs, then create a filesystem for them to use

# Do this last because requires mon cluster to be up and running

if mdss.remotes:

log.info('Setting up CephFS filesystem(s)...')

cephfs_config = config.get('cephfs', {})

fs_configs = cephfs_config.pop('fs', [{'name': 'cephfs'}])

# wait for standbys to become available (slow due to valgrind, perhaps)

mdsc = MDSCluster(ctx)

mds_count = len(list(teuthology.all_roles_of_type(ctx.cluster, 'mds')))

with contextutil.safe_while(sleep=2,tries=150) as proceed:

while proceed():

if len(mdsc.get_standby_daemons()) >= mds_count:

break

fss = []

for fs_config in fs_configs:

assert isinstance(fs_config, dict)

name = fs_config.pop('name')

temp = deepcopy(cephfs_config)

teuthology.deep_merge(temp, fs_config)

subvols = config.get('subvols', None)

if subvols:

teuthology.deep_merge(temp, {'subvols': subvols})

fs = Filesystem(ctx, fs_config=temp, name=name, create=True)

fss.append(fs)

yield

for fs in fss:

fs.destroy()

else:

ctx.ceph[config['cluster']].thrashers = []

ctx.ceph[config['cluster']].watched_processes = []

ctx.ceph[config['cluster']].watchdog = DaemonWatchdog(ctx, config)

ctx.ceph[config['cluster']].watchdog.start()

mon_bind_msgr2=False,

mon_bind_addrvec=False):

"""

mons = {}

v1_ports = {}

v2_ports = {}

is_mon = teuthology.is_type('mon', cluster_name)

for idx, roles in enumerate(roles):

for role in roles:

if not is_mon(role):

continue

if ips[idx] not in v1_ports:

v1_ports[ips[idx]] = 6789

else:

v1_ports[ips[idx]] += 1

if mon_bind_msgr2:

if ips[idx] not in v2_ports:

v2_ports[ips[idx]] = 3300

addr = '{ip}'.format(ip=ips[idx])

else:

assert mon_bind_addrvec

v2_ports[ips[idx]] += 1

addr = '[v2:{ip}:{port2},v1:{ip}:{port1}]'.format(

ip=ips[idx],

port2=v2_ports[ips[idx]],

port1=v1_ports[ips[idx]],

)

elif mon_bind_addrvec:

addr = '[v1:{ip}:{port}]'.format(

ip=ips[idx],

port=v1_ports[ips[idx]],

)

else:

addr = '{ip}:{port}'.format(

ip=ips[idx],

port=v1_ports[ips[idx]],

)

mons[role] = addr

assert mons

"""

path = os.path.join(os.path.dirname(__file__), 'ceph.conf.template')

conf = configobj.ConfigObj(path, file_error=True)

mon_hosts = []

for role, addr in mons.items():

mon_cluster, _, _ = teuthology.split_role(role)

if mon_cluster != cluster:

continue

name = teuthology.ceph_role(role)

conf.setdefault(name, {})

mon_hosts.append(addr)

conf.setdefault('global', {})

conf['global']['mon host'] = ','.join(mon_hosts)

# set up standby mds's

is_mds = teuthology.is_type('mds', cluster)

for roles_subset in roles:

for role in roles_subset:

if is_mds(role):

name = teuthology.ceph_role(role)

conf.setdefault(name, {})

path=None,

mon_bind_addrvec=False):

"""

addresses = list(mons.items())

assert addresses, "There are no monitors in config!"

log.debug('Ceph mon addresses: %s', addresses)

try:

log.debug('writing out conf {c}'.format(c=conf))

except:

log.debug('my conf logging attempt failed')

testdir = teuthology.get_testdir(ctx)

tmp_conf_path = '{tdir}/ceph.tmp.conf'.format(tdir=testdir)

conf_fp = BytesIO()

conf.write(conf_fp)

conf_fp.seek(0)

teuthology.write_file(remote, tmp_conf_path, conf_fp)

args = [

'adjust-ulimits',

'ceph-coverage',

'{tdir}/archive/coverage'.format(tdir=testdir),

'monmaptool',

'-c',

'{conf}'.format(conf=tmp_conf_path),

'--create',

'--clobber',

]

if mon_bind_addrvec:

args.extend(['--enable-all-features'])

for (role, addr) in addresses:

_, _, n = teuthology.split_role(role)

if mon_bind_addrvec and (',' in addr or 'v' in addr or ':' in addr):

args.extend(('--addv', n, addr))

else:

args.extend(('--add', n, addr))

if not path:

path = '{tdir}/monmap'.format(tdir=testdir)

args.extend([

'--print',

path

])

monmap_output = remote.sh(args)

fsid = re.search("generated fsid (.+)$",

monmap_output, re.MULTILINE).group(1)

teuthology.delete_file(remote, tmp_conf_path)

if type_ == 'osd' and is_crimson(config):

# teuthworker uses ubuntu:ubuntu to access the test nodes

create_log_cmd = \

f'sudo install -b -o ubuntu -g ubuntu /dev/null {log_path}'

return create_log_cmd, args + [run.Raw('2>>'), log_path]

else:

"""

if ctx.config.get('use_existing_cluster', False) is True:

log.info("'use_existing_cluster' is true; skipping cluster creation")

yield

testdir = teuthology.get_testdir(ctx)

cluster_name = config['cluster']

data_dir = '{tdir}/{cluster}.data'.format(tdir=testdir, cluster=cluster_name)

log.info('Creating ceph cluster %s...', cluster_name)

log.info('config %s', config)

log.info('ctx.config %s', ctx.config)

run.wait(

ctx.cluster.run(

args=[

'install', '-d', '-m0755', '--',

data_dir,

],

wait=False,

)

)

run.wait(

ctx.cluster.run(

args=[

'sudo',

'install', '-d', '-m0777', '--', '/var/run/ceph',

],

wait=False,

)

)

devs_to_clean = {}

remote_to_roles_to_devs = {}

osds = ctx.cluster.only(teuthology.is_type('osd', cluster_name))

for remote, roles_for_host in osds.remotes.items():

devs = teuthology.get_scratch_devices(remote)

roles_to_devs = assign_devs(

teuthology.cluster_roles_of_type(roles_for_host, 'osd', cluster_name), devs

)

devs_to_clean[remote] = []

log.info('osd dev map: {}'.format(roles_to_devs))

assert roles_to_devs, \

"remote {} has osd roles, but no osd devices were specified!".format(remote.hostname)

remote_to_roles_to_devs[remote] = roles_to_devs

log.info("remote_to_roles_to_devs: {}".format(remote_to_roles_to_devs))

for osd_role, dev_name in remote_to_roles_to_devs.items():

assert dev_name, "{} has no associated device!".format(osd_role)

log.info('Generating config...')

remotes_and_roles = ctx.cluster.remotes.items()

roles = [role_list for (remote, role_list) in remotes_and_roles]

ips = [host for (host, port) in

(remote.ssh.get_transport().getpeername() for (remote, role_list) in remotes_and_roles)]

mons = get_mons(

roles, ips, cluster_name,

mon_bind_msgr2=config.get('mon_bind_msgr2'),

mon_bind_addrvec=config.get('mon_bind_addrvec'),

)

conf = skeleton_config(

ctx, roles=roles, ips=ips, mons=mons, cluster=cluster_name,

)

for section, keys in config['conf'].items():

for key, value in keys.items():

log.info("[%s] %s = %s" % (section, key, value))

if section not in conf:

conf[section] = {}

conf[section][key] = value

if not hasattr(ctx, 'ceph'):

ctx.ceph = {}

ctx.ceph[cluster_name] = argparse.Namespace()

ctx.ceph[cluster_name].conf = conf

ctx.ceph[cluster_name].mons = mons

default_keyring = '/etc/ceph/{cluster}.keyring'.format(cluster=cluster_name)

keyring_path = config.get('keyring_path', default_keyring)

coverage_dir = '{tdir}/archive/coverage'.format(tdir=testdir)

firstmon = teuthology.get_first_mon(ctx, config, cluster_name)

log.info('Setting up %s...' % firstmon)

ctx.cluster.only(firstmon).run(

args=[

'sudo',

'adjust-ulimits',

'ceph-coverage',

coverage_dir,

'ceph-authtool',

'--create-keyring',

keyring_path,

],

)

ctx.cluster.only(firstmon).run(

args=[

'sudo',

'adjust-ulimits',

'ceph-coverage',

coverage_dir,

'ceph-authtool',

'--gen-key',

'--name=mon.',

keyring_path,

],

)

ctx.cluster.only(firstmon).run(

args=[

'sudo',

'chmod',

'0644',

keyring_path,

],

)

(mon0_remote,) = ctx.cluster.only(firstmon).remotes.keys()

monmap_path = '{tdir}/{cluster}.monmap'.format(tdir=testdir,

cluster=cluster_name)

fsid = create_simple_monmap(

ctx,

remote=mon0_remote,

conf=conf,

mons=mons,

path=monmap_path,

mon_bind_addrvec=config.get('mon_bind_addrvec'),

)

ctx.ceph[cluster_name].fsid = fsid

if not 'global' in conf:

conf['global'] = {}

conf['global']['fsid'] = fsid

default_conf_path = '/etc/ceph/{cluster}.conf'.format(cluster=cluster_name)

conf_path = config.get('conf_path', default_conf_path)

log.info('Writing %s for FSID %s...' % (conf_path, fsid))

write_conf(ctx, conf_path, cluster_name)

log.info('Creating admin key on %s...' % firstmon)

ctx.cluster.only(firstmon).run(

args=[

'sudo',

'adjust-ulimits',

'ceph-coverage',

coverage_dir,

'ceph-authtool',

'--gen-key',

'--name=client.admin',

'--cap', 'mon', 'allow *',

'--cap', 'osd', 'allow *',

'--cap', 'mds', 'allow *',

'--cap', 'mgr', 'allow *',

keyring_path,

],

)

log.info('Copying monmap to all nodes...')

keyring = mon0_remote.read_file(keyring_path)

monmap = mon0_remote.read_file(monmap_path)

for rem in ctx.cluster.remotes.keys():

# copy mon key and initial monmap

log.info('Sending monmap to node {remote}'.format(remote=rem))

rem.write_file(keyring_path, keyring, mode='0644', sudo=True)

rem.write_file(monmap_path, monmap)

log.info('Setting up mon nodes...')

mons = ctx.cluster.only(teuthology.is_type('mon', cluster_name))

if not config.get('skip_mgr_daemons', False):

log.info('Setting up mgr nodes...')

mgrs = ctx.cluster.only(teuthology.is_type('mgr', cluster_name))

for remote, roles_for_host in mgrs.remotes.items():

for role in teuthology.cluster_roles_of_type(roles_for_host, 'mgr',

cluster_name):

_, _, id_ = teuthology.split_role(role)

mgr_dir = DATA_PATH.format(

type_='mgr', cluster=cluster_name, id_=id_)

remote.run(

args=[

'sudo',

'mkdir',

'-p',

mgr_dir,

run.Raw('&&'),

'sudo',

'adjust-ulimits',

'ceph-coverage',

coverage_dir,

'ceph-authtool',

'--create-keyring',

'--gen-key',

'--name=mgr.{id}'.format(id=id_),

mgr_dir + '/keyring',

],

)

log.info('Setting up mds nodes...')

mdss = ctx.cluster.only(teuthology.is_type('mds', cluster_name))

for remote, roles_for_host in mdss.remotes.items():

for role in teuthology.cluster_roles_of_type(roles_for_host, 'mds',

cluster_name):

_, _, id_ = teuthology.split_role(role)

mds_dir = DATA_PATH.format(

type_='mds', cluster=cluster_name, id_=id_)

remote.run(

args=[

'sudo',

'mkdir',

'-p',

mds_dir,

run.Raw('&&'),

'sudo',

'adjust-ulimits',

'ceph-coverage',

coverage_dir,

'ceph-authtool',

'--create-keyring',

'--gen-key',

'--name=mds.{id}'.format(id=id_),

mds_dir + '/keyring',

],

)

remote.run(args=[

'sudo', 'chown', '-R', 'ceph:ceph', mds_dir

])

cclient.create_keyring(ctx, cluster_name)

log.info('Running mkfs on osd nodes...')

if not hasattr(ctx, 'disk_config'):

ctx.disk_config = argparse.Namespace()

if not hasattr(ctx.disk_config, 'remote_to_roles_to_dev'):

ctx.disk_config.remote_to_roles_to_dev = {}

if not hasattr(ctx.disk_config, 'remote_to_roles_to_dev_mount_options'):

ctx.disk_config.remote_to_roles_to_dev_mount_options = {}

if not hasattr(ctx.disk_config, 'remote_to_roles_to_dev_fstype'):

ctx.disk_config.remote_to_roles_to_dev_fstype = {}

teuthology.deep_merge(ctx.disk_config.remote_to_roles_to_dev, remote_to_roles_to_devs)

log.info("ctx.disk_config.remote_to_roles_to_dev: {r}".format(r=str(ctx.disk_config.remote_to_roles_to_dev)))

for remote, roles_for_host in osds.remotes.items():

roles_to_devs = remote_to_roles_to_devs[remote]

for role in teuthology.cluster_roles_of_type(roles_for_host, 'osd', cluster_name):

_, _, id_ = teuthology.split_role(role)

mnt_point = DATA_PATH.format(

type_='osd', cluster=cluster_name, id_=id_)

remote.run(

args=[

'sudo',

'mkdir',

'-p',

mnt_point,

])

log.info('roles_to_devs: {}'.format(roles_to_devs))

log.info('role: {}'.format(role))

if roles_to_devs.get(role):

dev = roles_to_devs[role]

fs = config.get('fs')

package = None

mkfs_options = config.get('mkfs_options')

mount_options = config.get('mount_options')

if fs == 'btrfs':

# package = 'btrfs-tools'

if mount_options is None:

mount_options = ['noatime', 'user_subvol_rm_allowed']

if mkfs_options is None:

mkfs_options = ['-m', 'single',

'-l', '32768',

'-n', '32768']

if fs == 'xfs':

# package = 'xfsprogs'

if mount_options is None:

mount_options = ['noatime']

if mkfs_options is None:

mkfs_options = ['-f', '-i', 'size=2048']

if fs == 'ext4' or fs == 'ext3':

if mount_options is None:

mount_options = ['noatime', 'user_xattr']

if mount_options is None:

mount_options = []

if mkfs_options is None:

mkfs_options = []

mkfs = ['mkfs.%s' % fs] + mkfs_options

log.info('%s on %s on %s' % (mkfs, dev, remote))

if package is not None:

remote.sh('sudo apt-get install -y %s' % package)

try:

remote.run(args=['yes', run.Raw('|')] + ['sudo'] + mkfs + [dev])

except run.CommandFailedError:

if fs != 'btrfs':

raise

# Newer btrfs-tools doesn't prompt for overwrite, use -f

if '-f' not in mount_options: