[ceph.git] / ceph / qa / tasks / cephadm.py

"""
Ceph cluster task, deployed via cephadm orchestrator
"""
import argparse
import configobj
import contextlib
import logging
import os
import json
import re
import uuid
import yaml

from copy import deepcopy
from io import BytesIO, StringIO
from tarfile import ReadError
from tasks.ceph_manager import CephManager
from teuthology import misc as teuthology
from teuthology import contextutil
from teuthology import packaging
from teuthology.orchestra import run
from teuthology.orchestra.daemon import DaemonGroup
from teuthology.config import config as teuth_config
from textwrap import dedent
from tasks.cephfs.filesystem import MDSCluster, Filesystem
from tasks.util import chacra

# these items we use from ceph.py should probably eventually move elsewhere
from tasks.ceph import get_mons, healthy
from tasks.vip import subst_vip

CEPH_ROLE_TYPES = ['mon', 'mgr', 'osd', 'mds', 'rgw', 'prometheus']

log = logging.getLogger(__name__)


def _shell(ctx, cluster_name, remote, args, extra_cephadm_args=[], **kwargs):
    teuthology.get_testdir(ctx)
    return remote.run(
        args=[
            'sudo',
            ctx.cephadm,
            '--image', ctx.ceph[cluster_name].image,
            'shell',
            '-c', '/etc/ceph/{}.conf'.format(cluster_name),
            '-k', '/etc/ceph/{}.client.admin.keyring'.format(cluster_name),
            '--fsid', ctx.ceph[cluster_name].fsid,
            ] + extra_cephadm_args + [
            '--',
            ] + args,
        **kwargs
    )


def build_initial_config(ctx, config):
    cluster_name = config['cluster']

    path = os.path.join(os.path.dirname(__file__), 'cephadm.conf')
    conf = configobj.ConfigObj(path, file_error=True)

    conf.setdefault('global', {})
    conf['global']['fsid'] = ctx.ceph[cluster_name].fsid

    # overrides
    for section, keys in config.get('conf',{}).items():
        for key, value in keys.items():
            log.info(" override: [%s] %s = %s" % (section, key, value))
            if section not in conf:
                conf[section] = {}
            conf[section][key] = value

    return conf


def distribute_iscsi_gateway_cfg(ctx, conf_data):
    """
    Distribute common gateway config to get the IPs.
    These will help in iscsi clients with finding trusted_ip_list.
    """
    log.info('Distributing iscsi-gateway.cfg...')
    for remote, roles in ctx.cluster.remotes.items():
        remote.write_file(
            path='/etc/ceph/iscsi-gateway.cfg',
            data=conf_data,
            sudo=True)

def update_archive_setting(ctx, key, value):
    """
    Add logs directory to job's info log file
    """
    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}
        yaml.safe_dump(info_yaml, info_file, default_flow_style=False)


@contextlib.contextmanager
def normalize_hostnames(ctx):
    """
    Ensure we have short hostnames throughout, for consistency between
    remote.shortname and socket.gethostname() in cephadm.
    """
    log.info('Normalizing hostnames...')
    cluster = ctx.cluster.filter(lambda r: '.' in r.hostname)
    cluster.run(args=[
        'sudo',
        'hostname',
        run.Raw('$(hostname -s)'),
    ])

    try:
        yield
    finally:
        pass


@contextlib.contextmanager
def download_cephadm(ctx, config, ref):
    cluster_name = config['cluster']

    if config.get('cephadm_mode') != 'cephadm-package':
        if ctx.config.get('redhat'):
            _fetch_cephadm_from_rpm(ctx)
        # TODO: come up with a sensible way to detect if we need an "old, uncompiled"
        # cephadm
        elif 'cephadm_git_url' in config and 'cephadm_branch' in config:
            _fetch_cephadm_from_github(ctx, config, ref)
        else:
            _fetch_cephadm_from_chachra(ctx, config, cluster_name)

    try:
        yield
    finally:
        _rm_cluster(ctx, cluster_name)
        if config.get('cephadm_mode') == 'root':
            _rm_cephadm(ctx)


def _fetch_cephadm_from_rpm(ctx):
    log.info("Copying cephadm installed from an RPM package")
    # cephadm already installed from redhat.install task
    ctx.cluster.run(
        args=[
            'cp',
            run.Raw('$(which cephadm)'),
            ctx.cephadm,
            run.Raw('&&'),
            'ls', '-l',
            ctx.cephadm,
        ]
    )


def _fetch_cephadm_from_github(ctx, config, ref):
    ref = config.get('cephadm_branch', ref)
    git_url = config.get('cephadm_git_url', teuth_config.get_ceph_git_url())
    log.info('Downloading cephadm (repo %s ref %s)...' % (git_url, ref))
    if git_url.startswith('https://github.com/'):
        # git archive doesn't like https:// URLs, which we use with github.
        rest = git_url.split('https://github.com/', 1)[1]
        rest = re.sub(r'\.git/?$', '', rest).strip() # no .git suffix
        ctx.cluster.run(
            args=[
                'curl', '--silent',
                'https://raw.githubusercontent.com/' + rest + '/' + ref + '/src/cephadm/cephadm',
                run.Raw('>'),
                ctx.cephadm,
                run.Raw('&&'),
                'ls', '-l',
                ctx.cephadm,
            ],
        )
    else:
        ctx.cluster.run(
            args=[
                'git', 'clone', git_url, 'testrepo',
                run.Raw('&&'),
                'cd', 'testrepo',
                run.Raw('&&'),
                'git', 'show', f'{ref}:src/cephadm/cephadm',
                run.Raw('>'),
                ctx.cephadm,
                run.Raw('&&'),
                'ls', '-l', ctx.cephadm,
            ],
        )
    # sanity-check the resulting file and set executable bit
    cephadm_file_size = '$(stat -c%s {})'.format(ctx.cephadm)
    ctx.cluster.run(
        args=[
            'test', '-s', ctx.cephadm,
            run.Raw('&&'),
            'test', run.Raw(cephadm_file_size), "-gt", run.Raw('1000'),
            run.Raw('&&'),
            'chmod', '+x', ctx.cephadm,
        ],
    )


def _fetch_cephadm_from_chachra(ctx, config, cluster_name):
    log.info('Downloading "compiled" cephadm from cachra')
    bootstrap_remote = ctx.ceph[cluster_name].bootstrap_remote
    bp = packaging.get_builder_project()(
        config.get('project', 'ceph'),
        config,
        ctx=ctx,
        remote=bootstrap_remote,
    )
    log.info('builder_project result: %s' % (bp._result.json()))

    flavor = config.get('flavor', 'default')
    branch = config.get('branch')
    sha1 = config.get('sha1')

    # pull the cephadm binary from chacra
    url = chacra.get_binary_url(
            'cephadm',
            project=bp.project,
            distro=bp.distro.split('/')[0],
            release=bp.distro.split('/')[1],
            arch=bp.arch,
            flavor=flavor,
            branch=branch,
            sha1=sha1,
    )
    log.info("Discovered cachra url: %s", url)
    ctx.cluster.run(
        args=[
            'curl', '--silent', '-L', url,
            run.Raw('>'),
            ctx.cephadm,
            run.Raw('&&'),
            'ls', '-l',
            ctx.cephadm,
        ],
    )

    # sanity-check the resulting file and set executable bit
    cephadm_file_size = '$(stat -c%s {})'.format(ctx.cephadm)
    ctx.cluster.run(
        args=[
            'test', '-s', ctx.cephadm,
            run.Raw('&&'),
            'test', run.Raw(cephadm_file_size), "-gt", run.Raw('1000'),
            run.Raw('&&'),
            'chmod', '+x', ctx.cephadm,
        ],
    )


def _rm_cluster(ctx, cluster_name):
    log.info('Removing cluster...')
    ctx.cluster.run(args=[
        'sudo',
        ctx.cephadm,
        'rm-cluster',
        '--fsid', ctx.ceph[cluster_name].fsid,
        '--force',
    ])


def _rm_cephadm(ctx):
    log.info('Removing cephadm ...')
    ctx.cluster.run(
        args=[
            'rm',
            '-rf',
            ctx.cephadm,
        ],
    )


@contextlib.contextmanager
def ceph_log(ctx, config):
    cluster_name = config['cluster']
    fsid = ctx.ceph[cluster_name].fsid

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


    try:
        yield

    except Exception:
        # we need to know this below
        ctx.summary['success'] = False
        raise

    finally:
        log.info('Checking cluster log for badness...')
        def first_in_ceph_log(pattern, excludes):
            """
            Find the first occurrence of the pattern specified in the Ceph log,
            Returns None if none found.

            :param pattern: Pattern scanned for.
            :param excludes: Patterns to ignore.
            :return: First line of text (or None if not found)
            """
            args = [
                'sudo',
                'egrep', pattern,
                '/var/log/ceph/{fsid}/ceph.log'.format(
                    fsid=fsid),
            ]
            if excludes:
                for exclude in excludes:
                    args.extend([run.Raw('|'), 'egrep', '-v', exclude])
            args.extend([
                run.Raw('|'), 'head', '-n', '1',
            ])
            r = ctx.ceph[cluster_name].bootstrap_remote.run(
                stdout=StringIO(),
                args=args,
            )
            stdout = r.stdout.getvalue()
            if stdout != '':
                return stdout
            return None

        if first_in_ceph_log('\[ERR\]|\[WRN\]|\[SEC\]',
                             config.get('log-ignorelist')) is not None:
            log.warning('Found errors (ERR|WRN|SEC) in cluster log')
            ctx.summary['success'] = False
            # use the most severe problem as the failure reason
            if 'failure_reason' not in ctx.summary:
                for pattern in ['\[SEC\]', '\[ERR\]', '\[WRN\]']:
                    match = first_in_ceph_log(pattern, config['log-ignorelist'])
                    if match is not None:
                        ctx.summary['failure_reason'] = \
                            '"{match}" in cluster log'.format(
                                match=match.rstrip('\n'),
                            )
                        break

        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=[
                        'sudo',
                        'find',
                        '/var/log/ceph',   # all logs, not just for the cluster
                        '/var/log/rbd-target-api', # ceph-iscsi
                        '-name',
                        '*.log',
                        '-print0',
                        run.Raw('|'),
                        'sudo',
                        'xargs',
                        '-0',
                        '--no-run-if-empty',
                        '--',
                        'gzip',
                        '--',
                    ],
                    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
                try:
                    teuthology.pull_directory(remote, '/var/log/ceph',  # everything
                                              os.path.join(sub, 'log'))
                except ReadError:
                    pass


@contextlib.contextmanager
def ceph_crash(ctx, config):
    """
    Gather crash dumps from /var/lib/ceph/$fsid/crash
    """
    cluster_name = config['cluster']
    fsid = ctx.ceph[cluster_name].fsid

    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/%s/crash' % fsid,
                                              os.path.join(sub, 'crash'))
                except ReadError:
                    pass


@contextlib.contextmanager
def pull_image(ctx, config):
    cluster_name = config['cluster']
    log.info(f'Pulling image {ctx.ceph[cluster_name].image} on all hosts...')
    run.wait(
        ctx.cluster.run(
            args=[
                'sudo',
                ctx.cephadm,
                '--image', ctx.ceph[cluster_name].image,
                'pull',
            ],
            wait=False,
        )
    )

    try:
        yield
    finally:
        pass

@contextlib.contextmanager
def setup_ca_signed_keys(ctx, config):
    # generate our ca key
    cluster_name = config['cluster']
    bootstrap_remote = ctx.ceph[cluster_name].bootstrap_remote
    bootstrap_remote.run(args=[
        'sudo', 'ssh-keygen', '-t', 'rsa', '-f', '/root/ca-key', '-N', ''
    ])

    # not using read_file here because it runs dd as a non-root
    # user and would hit permission issues
    r = bootstrap_remote.run(args=[
        'sudo', 'cat', '/root/ca-key.pub'
    ], stdout=StringIO())
    ca_key_pub_contents = r.stdout.getvalue()

    # make CA key accepted on each host
    for remote in ctx.cluster.remotes.keys():
        # write key to each host's /etc/ssh dir
        remote.run(args=[
            'sudo', 'echo', ca_key_pub_contents,
            run.Raw('|'),
            'sudo', 'tee', '-a', '/etc/ssh/ca-key.pub',
        ])
        # make sshd accept the CA signed key
        remote.run(args=[
            'sudo', 'echo', 'TrustedUserCAKeys /etc/ssh/ca-key.pub',
            run.Raw('|'),
            'sudo', 'tee', '-a', '/etc/ssh/sshd_config',
            run.Raw('&&'),
            'sudo', 'systemctl', 'restart', 'sshd',
        ])

    # generate a new key pair and sign the pub key to make a cert
    bootstrap_remote.run(args=[
        'sudo', 'ssh-keygen', '-t', 'rsa', '-f', '/root/cephadm-ssh-key', '-N', '',
        run.Raw('&&'),
        'sudo', 'ssh-keygen', '-s', '/root/ca-key', '-I', 'user_root', '-n', 'root', '-V', '+52w', '/root/cephadm-ssh-key',
    ])

    # for debugging, to make sure this setup has worked as intended
    for remote in ctx.cluster.remotes.keys():
        remote.run(args=[
            'sudo', 'cat', '/etc/ssh/ca-key.pub'
        ])
        remote.run(args=[
            'sudo', 'cat', '/etc/ssh/sshd_config',
            run.Raw('|'),
            'grep', 'TrustedUserCAKeys'
        ])
    bootstrap_remote.run(args=[
        'sudo', 'ls', '/root/'
    ])

    ctx.ca_signed_key_info = {}
    ctx.ca_signed_key_info['ca-key'] = '/root/ca-key'
    ctx.ca_signed_key_info['ca-key-pub'] = '/root/ca-key.pub'
    ctx.ca_signed_key_info['private-key'] = '/root/cephadm-ssh-key'
    ctx.ca_signed_key_info['ca-signed-cert'] = '/root/cephadm-ssh-key-cert.pub'

    try:
        yield
    finally:
        pass

@contextlib.contextmanager
def ceph_bootstrap(ctx, config):
    """
    Bootstrap ceph cluster.

    :param ctx: the argparse.Namespace object
    :param config: the config dict
    """
    cluster_name = config['cluster']
    testdir = teuthology.get_testdir(ctx)
    fsid = ctx.ceph[cluster_name].fsid

    bootstrap_remote = ctx.ceph[cluster_name].bootstrap_remote
    first_mon = ctx.ceph[cluster_name].first_mon
    first_mon_role = ctx.ceph[cluster_name].first_mon_role
    mons = ctx.ceph[cluster_name].mons

    ctx.cluster.run(args=[
        'sudo', 'mkdir', '-p', '/etc/ceph',
        ]);
    ctx.cluster.run(args=[
        'sudo', 'chmod', '777', '/etc/ceph',
        ]);
    try:
        # write seed config
        log.info('Writing seed config...')
        conf_fp = BytesIO()
        seed_config = build_initial_config(ctx, config)
        seed_config.write(conf_fp)
        bootstrap_remote.write_file(
            path='{}/seed.{}.conf'.format(testdir, cluster_name),
            data=conf_fp.getvalue())
        log.debug('Final config:\n' + conf_fp.getvalue().decode())
        ctx.ceph[cluster_name].conf = seed_config

        # register initial daemons
        ctx.daemons.register_daemon(
            bootstrap_remote, 'mon', first_mon,
            cluster=cluster_name,
            fsid=fsid,
            logger=log.getChild('mon.' + first_mon),
            wait=False,
            started=True,
        )
        if not ctx.ceph[cluster_name].roleless:
            first_mgr = ctx.ceph[cluster_name].first_mgr
            ctx.daemons.register_daemon(
                bootstrap_remote, 'mgr', first_mgr,
                cluster=cluster_name,
                fsid=fsid,
                logger=log.getChild('mgr.' + first_mgr),
                wait=False,
                started=True,
            )

        # bootstrap
        log.info('Bootstrapping...')
        cmd = [
            'sudo',
            ctx.cephadm,
            '--image', ctx.ceph[cluster_name].image,
            '-v',
            'bootstrap',
            '--fsid', fsid,
            '--config', '{}/seed.{}.conf'.format(testdir, cluster_name),
            '--output-config', '/etc/ceph/{}.conf'.format(cluster_name),
            '--output-keyring',
            '/etc/ceph/{}.client.admin.keyring'.format(cluster_name),
        ]

        if not config.get("use-ca-signed-key", False):
            cmd += ['--output-pub-ssh-key', '{}/{}.pub'.format(testdir, cluster_name)]
        else:
            # ctx.ca_signed_key_info should have been set up in
            # setup_ca_signed_keys function which we expect to have
            # run before bootstrap if use-ca-signed-key is true
            signed_key_info = ctx.ca_signed_key_info
            cmd += [
                "--ssh-private-key", signed_key_info['private-key'],
                "--ssh-signed-cert", signed_key_info['ca-signed-cert'],
            ]

        if config.get("no_cgroups_split") is True:
            cmd.insert(cmd.index("bootstrap"), "--no-cgroups-split")

        if config.get('registry-login'):
            registry = config['registry-login']
            cmd += [
                "--registry-url", registry['url'],
                "--registry-username", registry['username'],
                "--registry-password", registry['password'],
            ]

        if not ctx.ceph[cluster_name].roleless:
            cmd += [
                '--mon-id', first_mon,
                '--mgr-id', first_mgr,
                '--orphan-initial-daemons',   # we will do it explicitly!
                '--skip-monitoring-stack',    # we'll provision these explicitly
            ]

        if mons[first_mon_role].startswith('['):
            cmd += ['--mon-addrv', mons[first_mon_role]]
        else:
            cmd += ['--mon-ip', mons[first_mon_role]]
        if config.get('skip_dashboard'):
            cmd += ['--skip-dashboard']
        if config.get('skip_monitoring_stack'):
            cmd += ['--skip-monitoring-stack']
        if config.get('single_host_defaults'):
            cmd += ['--single-host-defaults']
        if not config.get('avoid_pacific_features', False):
            cmd += ['--skip-admin-label']
        # bootstrap makes the keyring root 0600, so +r it for our purposes
        cmd += [
            run.Raw('&&'),
            'sudo', 'chmod', '+r',
            '/etc/ceph/{}.client.admin.keyring'.format(cluster_name),
        ]
        bootstrap_remote.run(args=cmd)

        # fetch keys and configs
        log.info('Fetching config...')
        ctx.ceph[cluster_name].config_file = \
            bootstrap_remote.read_file(f'/etc/ceph/{cluster_name}.conf')
        log.info('Fetching client.admin keyring...')
        ctx.ceph[cluster_name].admin_keyring = \
            bootstrap_remote.read_file(f'/etc/ceph/{cluster_name}.client.admin.keyring')
        log.info('Fetching mon keyring...')
        ctx.ceph[cluster_name].mon_keyring = \
            bootstrap_remote.read_file(f'/var/lib/ceph/{fsid}/mon.{first_mon}/keyring', sudo=True)

        if not config.get("use-ca-signed-key", False):
            # fetch ssh key, distribute to additional nodes
            log.info('Fetching pub ssh key...')
            ssh_pub_key = bootstrap_remote.read_file(
                f'{testdir}/{cluster_name}.pub').decode('ascii').strip()

            log.info('Installing pub ssh key for root users...')
            ctx.cluster.run(args=[
                'sudo', 'install', '-d', '-m', '0700', '/root/.ssh',
                run.Raw('&&'),
                'echo', ssh_pub_key,
                run.Raw('|'),
                'sudo', 'tee', '-a', '/root/.ssh/authorized_keys',
                run.Raw('&&'),
                'sudo', 'chmod', '0600', '/root/.ssh/authorized_keys',
            ])

        # set options
        if config.get('allow_ptrace', True):
            _shell(ctx, cluster_name, bootstrap_remote,
                   ['ceph', 'config', 'set', 'mgr', 'mgr/cephadm/allow_ptrace', 'true'])

        if not config.get('avoid_pacific_features', False):
            log.info('Distributing conf and client.admin keyring to all hosts + 0755')
            _shell(ctx, cluster_name, bootstrap_remote,
                   ['ceph', 'orch', 'client-keyring', 'set', 'client.admin',
                    '*', '--mode', '0755'],
                   check_status=False)

        # add other hosts
        for remote in ctx.cluster.remotes.keys():
            if remote == bootstrap_remote:
                continue

            # note: this may be redundant (see above), but it avoids
            # us having to wait for cephadm to do it.
            log.info('Writing (initial) conf and keyring to %s' % remote.shortname)
            remote.write_file(
                path='/etc/ceph/{}.conf'.format(cluster_name),
                data=ctx.ceph[cluster_name].config_file)
            remote.write_file(
                path='/etc/ceph/{}.client.admin.keyring'.format(cluster_name),
                data=ctx.ceph[cluster_name].admin_keyring)

            log.info('Adding host %s to orchestrator...' % remote.shortname)
            _shell(ctx, cluster_name, bootstrap_remote, [
                'ceph', 'orch', 'host', 'add',
                remote.shortname
            ])
            r = _shell(ctx, cluster_name, bootstrap_remote,
                       ['ceph', 'orch', 'host', 'ls', '--format=json'],
                       stdout=StringIO())
            hosts = [node['hostname'] for node in json.loads(r.stdout.getvalue())]
            assert remote.shortname in hosts

        yield

    finally:
        log.info('Cleaning up testdir ceph.* files...')
        ctx.cluster.run(args=[
            'rm', '-f',
            '{}/seed.{}.conf'.format(testdir, cluster_name),
            '{}/{}.pub'.format(testdir, cluster_name),
        ])

        log.info('Stopping all daemons...')

        # this doesn't block until they are all stopped...
        #ctx.cluster.run(args=['sudo', 'systemctl', 'stop', 'ceph.target'])

        # stop the daemons we know
        for role in ctx.daemons.resolve_role_list(None, CEPH_ROLE_TYPES, True):
            cluster, type_, id_ = teuthology.split_role(role)
            try:
                ctx.daemons.get_daemon(type_, id_, cluster).stop()
            except Exception:
                log.exception(f'Failed to stop "{role}"')
                raise

        # tear down anything left (but leave the logs behind)
        ctx.cluster.run(
            args=[
                'sudo',
                ctx.cephadm,
                'rm-cluster',
                '--fsid', fsid,
                '--force',
                '--keep-logs',
            ],
            check_status=False,  # may fail if upgrading from old cephadm
        )

        # clean up /etc/ceph
        ctx.cluster.run(args=[
            'sudo', 'rm', '-f',
            '/etc/ceph/{}.conf'.format(cluster_name),
            '/etc/ceph/{}.client.admin.keyring'.format(cluster_name),
        ])


@contextlib.contextmanager
def ceph_mons(ctx, config):
    """
    Deploy any additional mons
    """
    cluster_name = config['cluster']
    fsid = ctx.ceph[cluster_name].fsid

    try:
        daemons = {}
        if config.get('add_mons_via_daemon_add'):
            # This is the old way of adding mons that works with the (early) octopus
            # cephadm scheduler.
            num_mons = 1
            for remote, roles in ctx.cluster.remotes.items():
                for mon in [r for r in roles
                            if teuthology.is_type('mon', cluster_name)(r)]:
                    c_, _, id_ = teuthology.split_role(mon)
                    if c_ == cluster_name and id_ == ctx.ceph[cluster_name].first_mon:
                        continue
                    log.info('Adding %s on %s' % (mon, remote.shortname))
                    num_mons += 1
                    _shell(ctx, cluster_name, remote, [
                        'ceph', 'orch', 'daemon', 'add', 'mon',
                        remote.shortname + ':' + ctx.ceph[cluster_name].mons[mon] + '=' + id_,
                    ])
                    ctx.daemons.register_daemon(
                        remote, 'mon', id_,
                        cluster=cluster_name,
                        fsid=fsid,
                        logger=log.getChild(mon),
                        wait=False,
                        started=True,
                    )
                    daemons[mon] = (remote, id_)

                    with contextutil.safe_while(sleep=1, tries=180) as proceed:
                        while proceed():
                            log.info('Waiting for %d mons in monmap...' % (num_mons))
                            r = _shell(
                                ctx=ctx,
                                cluster_name=cluster_name,
                                remote=remote,
                                args=[
                                    'ceph', 'mon', 'dump', '-f', 'json',
                                ],
                                stdout=StringIO(),
                            )
                            j = json.loads(r.stdout.getvalue())
                            if len(j['mons']) == num_mons:
                                break
        else:
            nodes = []
            for remote, roles in ctx.cluster.remotes.items():
                for mon in [r for r in roles
                            if teuthology.is_type('mon', cluster_name)(r)]:
                    c_, _, id_ = teuthology.split_role(mon)
                    log.info('Adding %s on %s' % (mon, remote.shortname))
                    nodes.append(remote.shortname
                                 + ':' + ctx.ceph[cluster_name].mons[mon]
                                 + '=' + id_)
                    if c_ == cluster_name and id_ == ctx.ceph[cluster_name].first_mon:
                        continue
                    daemons[mon] = (remote, id_)

            _shell(ctx, cluster_name, remote, [
                'ceph', 'orch', 'apply', 'mon',
                str(len(nodes)) + ';' + ';'.join(nodes)]
                   )
            for mgr, i in daemons.items():
                remote, id_ = i
                ctx.daemons.register_daemon(
                    remote, 'mon', id_,
                    cluster=cluster_name,
                    fsid=fsid,
                    logger=log.getChild(mon),
                    wait=False,
                    started=True,
                )

            with contextutil.safe_while(sleep=1, tries=180) as proceed:
                while proceed():
                    log.info('Waiting for %d mons in monmap...' % (len(nodes)))
                    r = _shell(
                        ctx=ctx,
                        cluster_name=cluster_name,
                        remote=remote,
                        args=[
                            'ceph', 'mon', 'dump', '-f', 'json',
                        ],
                        stdout=StringIO(),
                    )
                    j = json.loads(r.stdout.getvalue())
                    if len(j['mons']) == len(nodes):
                        break

        # refresh our (final) ceph.conf file
        bootstrap_remote = ctx.ceph[cluster_name].bootstrap_remote
        log.info('Generating final ceph.conf file...')
        r = _shell(
            ctx=ctx,
            cluster_name=cluster_name,
            remote=bootstrap_remote,
            args=[
                'ceph', 'config', 'generate-minimal-conf',
            ],
            stdout=StringIO(),
        )
        ctx.ceph[cluster_name].config_file = r.stdout.getvalue()

        yield

    finally:
        pass


@contextlib.contextmanager
def ceph_mgrs(ctx, config):
    """
    Deploy any additional mgrs
    """
    cluster_name = config['cluster']
    fsid = ctx.ceph[cluster_name].fsid

    try:
        nodes = []
        daemons = {}
        for remote, roles in ctx.cluster.remotes.items():
            for mgr in [r for r in roles
                        if teuthology.is_type('mgr', cluster_name)(r)]:
                c_, _, id_ = teuthology.split_role(mgr)
                log.info('Adding %s on %s' % (mgr, remote.shortname))
                nodes.append(remote.shortname + '=' + id_)
                if c_ == cluster_name and id_ == ctx.ceph[cluster_name].first_mgr:
                    continue
                daemons[mgr] = (remote, id_)
        if nodes:
            _shell(ctx, cluster_name, remote, [
                'ceph', 'orch', 'apply', 'mgr',
                str(len(nodes)) + ';' + ';'.join(nodes)]
            )
        for mgr, i in daemons.items():
            remote, id_ = i
            ctx.daemons.register_daemon(
                remote, 'mgr', id_,
                cluster=cluster_name,
                fsid=fsid,
                logger=log.getChild(mgr),
                wait=False,
                started=True,
            )

        yield

    finally:
        pass


@contextlib.contextmanager
def ceph_osds(ctx, config):
    """
    Deploy OSDs
    """
    cluster_name = config['cluster']
    fsid = ctx.ceph[cluster_name].fsid

    try:
        log.info('Deploying OSDs...')

        # provision OSDs in numeric order
        id_to_remote = {}
        devs_by_remote = {}
        for remote, roles in ctx.cluster.remotes.items():
            devs_by_remote[remote] = teuthology.get_scratch_devices(remote)
            for osd in [r for r in roles
                        if teuthology.is_type('osd', cluster_name)(r)]:
                _, _, id_ = teuthology.split_role(osd)
                id_to_remote[int(id_)] = (osd, remote)

        cur = 0
        for osd_id in sorted(id_to_remote.keys()):
            osd, remote = id_to_remote[osd_id]
            _, _, id_ = teuthology.split_role(osd)
            assert int(id_) == cur
            devs = devs_by_remote[remote]
            assert devs   ## FIXME ##
            dev = devs.pop()
            if all(_ in dev for _ in ('lv', 'vg')):
                short_dev = dev.replace('/dev/', '')
            else:
                short_dev = dev
            log.info('Deploying %s on %s with %s...' % (
                osd, remote.shortname, dev))
            _shell(ctx, cluster_name, remote, [
                'ceph-volume', 'lvm', 'zap', dev])
            add_osd_args = ['ceph', 'orch', 'daemon', 'add', 'osd',
                            remote.shortname + ':' + short_dev]
            osd_method = config.get('osd_method')
            if osd_method:
                add_osd_args.append(osd_method)
            _shell(ctx, cluster_name, remote, add_osd_args)
            ctx.daemons.register_daemon(
                remote, 'osd', id_,
                cluster=cluster_name,
                fsid=fsid,
                logger=log.getChild(osd),
                wait=False,
                started=True,
            )
            cur += 1

        if cur == 0:
            _shell(ctx, cluster_name, remote, [
                'ceph', 'orch', 'apply', 'osd', '--all-available-devices',
            ])
            # expect the number of scratch devs
            num_osds = sum(map(len, devs_by_remote.values()))
            assert num_osds
        else:
            # expect the number of OSDs we created
            num_osds = cur

        log.info(f'Waiting for {num_osds} OSDs to come up...')
        with contextutil.safe_while(sleep=1, tries=120) as proceed:
            while proceed():
                p = _shell(ctx, cluster_name, ctx.ceph[cluster_name].bootstrap_remote,
                           ['ceph', 'osd', 'stat', '-f', 'json'], stdout=StringIO())
                j = json.loads(p.stdout.getvalue())
                if int(j.get('num_up_osds', 0)) == num_osds:
                    break;

        if not hasattr(ctx, 'managers'):
            ctx.managers = {}
        ctx.managers[cluster_name] = CephManager(
            ctx.ceph[cluster_name].bootstrap_remote,
            ctx=ctx,
            logger=log.getChild('ceph_manager.' + cluster_name),
            cluster=cluster_name,
            cephadm=True,
        )

        yield
    finally:
        pass


@contextlib.contextmanager
def ceph_mdss(ctx, config):
    """
    Deploy MDSss
    """
    cluster_name = config['cluster']
    fsid = ctx.ceph[cluster_name].fsid

    nodes = []
    daemons = {}
    for remote, roles in ctx.cluster.remotes.items():
        for role in [r for r in roles
                    if teuthology.is_type('mds', cluster_name)(r)]:
            c_, _, id_ = teuthology.split_role(role)
            log.info('Adding %s on %s' % (role, remote.shortname))
            nodes.append(remote.shortname + '=' + id_)
            daemons[role] = (remote, id_)
    if nodes:
        _shell(ctx, cluster_name, remote, [
            'ceph', 'orch', 'apply', 'mds',
            'all',
            str(len(nodes)) + ';' + ';'.join(nodes)]
        )
    for role, i in daemons.items():
        remote, id_ = i
        ctx.daemons.register_daemon(
            remote, 'mds', id_,
            cluster=cluster_name,
            fsid=fsid,
            logger=log.getChild(role),
            wait=False,
            started=True,
        )

    yield

@contextlib.contextmanager
def cephfs_setup(ctx, config):
    mdss = list(teuthology.all_roles_of_type(ctx.cluster, 'mds'))

    # 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 len(mdss) > 0:
        log.info('Setting up CephFS filesystem(s)...')
        cephfs_config = config.get('cephfs', {})
        fs_configs =  cephfs_config.pop('fs', [{'name': 'cephfs'}])
        set_allow_multifs = len(fs_configs) > 1

        # wait for standbys to become available (slow due to valgrind, perhaps)
        mdsc = MDSCluster(ctx)
        with contextutil.safe_while(sleep=2,tries=150) as proceed:
            while proceed():
                if len(mdsc.get_standby_daemons()) >= len(mdss):
                    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)
            if set_allow_multifs:
                fs.set_allow_multifs()
                set_allow_multifs = False
            fss.append(fs)

        yield

        for fs in fss:
            fs.destroy()
    else:
        yield

@contextlib.contextmanager
def ceph_monitoring(daemon_type, ctx, config):
    """
    Deploy prometheus, node-exporter, etc.
    """
    cluster_name = config['cluster']
    fsid = ctx.ceph[cluster_name].fsid

    nodes = []
    daemons = {}
    for remote, roles in ctx.cluster.remotes.items():
        for role in [r for r in roles
                    if teuthology.is_type(daemon_type, cluster_name)(r)]:
            c_, _, id_ = teuthology.split_role(role)
            log.info('Adding %s on %s' % (role, remote.shortname))
            nodes.append(remote.shortname + '=' + id_)
            daemons[role] = (remote, id_)
    if nodes:
        _shell(ctx, cluster_name, remote, [
            'ceph', 'orch', 'apply', daemon_type,
            str(len(nodes)) + ';' + ';'.join(nodes)]
        )
    for role, i in daemons.items():
        remote, id_ = i
        ctx.daemons.register_daemon(
            remote, daemon_type, id_,
            cluster=cluster_name,
            fsid=fsid,
            logger=log.getChild(role),
            wait=False,
            started=True,
        )

    yield


@contextlib.contextmanager
def ceph_rgw(ctx, config):
    """
    Deploy rgw
    """
    cluster_name = config['cluster']
    fsid = ctx.ceph[cluster_name].fsid

    nodes = {}
    daemons = {}
    for remote, roles in ctx.cluster.remotes.items():
        for role in [r for r in roles
                    if teuthology.is_type('rgw', cluster_name)(r)]:
            c_, _, id_ = teuthology.split_role(role)
            log.info('Adding %s on %s' % (role, remote.shortname))
            svc = '.'.join(id_.split('.')[0:2])
            if svc not in nodes:
                nodes[svc] = []
            nodes[svc].append(remote.shortname + '=' + id_)
            daemons[role] = (remote, id_)

    for svc, nodes in nodes.items():
        _shell(ctx, cluster_name, remote, [
            'ceph', 'orch', 'apply', 'rgw', svc,
             '--placement',
             str(len(nodes)) + ';' + ';'.join(nodes)]
        )
    for role, i in daemons.items():
        remote, id_ = i
        ctx.daemons.register_daemon(
            remote, 'rgw', id_,
            cluster=cluster_name,
            fsid=fsid,
            logger=log.getChild(role),
            wait=False,
            started=True,
        )

    yield


@contextlib.contextmanager
def ceph_iscsi(ctx, config):
    """
    Deploy iSCSIs
    """
    cluster_name = config['cluster']
    fsid = ctx.ceph[cluster_name].fsid

    nodes = []
    daemons = {}
    ips = []

    for remote, roles in ctx.cluster.remotes.items():
        for role in [r for r in roles
                     if teuthology.is_type('iscsi', cluster_name)(r)]:
            c_, _, id_ = teuthology.split_role(role)
            log.info('Adding %s on %s' % (role, remote.shortname))
            nodes.append(remote.shortname + '=' + id_)
            daemons[role] = (remote, id_)
            ips.append(remote.ip_address)
    trusted_ip_list = ','.join(ips)
    if nodes:
        poolname = 'datapool'
        # ceph osd pool create datapool 3 3 replicated
        _shell(ctx, cluster_name, remote, [
            'ceph', 'osd', 'pool', 'create',
            poolname, '3', '3', 'replicated']
        )

        _shell(ctx, cluster_name, remote, [
            'rbd', 'pool', 'init', poolname]
        )

        # ceph orch apply iscsi datapool (admin)user (admin)password
        _shell(ctx, cluster_name, remote, [
            'ceph', 'orch', 'apply', 'iscsi',
            poolname, 'admin', 'admin',
            '--trusted_ip_list', trusted_ip_list,
            '--placement', str(len(nodes)) + ';' + ';'.join(nodes)]
        )

        # used by iscsi client to identify valid gateway ip's
        conf_data = dedent(f"""
        [config]
        trusted_ip_list = {trusted_ip_list}
        """)
        distribute_iscsi_gateway_cfg(ctx, conf_data)

    for role, i in daemons.items():
        remote, id_ = i
        ctx.daemons.register_daemon(
            remote, 'iscsi', id_,
            cluster=cluster_name,
            fsid=fsid,
            logger=log.getChild(role),
            wait=False,
            started=True,
        )

    yield


@contextlib.contextmanager
def ceph_clients(ctx, config):
    cluster_name = config['cluster']

    log.info('Setting up client nodes...')
    clients = ctx.cluster.only(teuthology.is_type('client', cluster_name))
    for remote, roles_for_host in clients.remotes.items():
        for role in teuthology.cluster_roles_of_type(roles_for_host, 'client',
                                                     cluster_name):
            name = teuthology.ceph_role(role)
            client_keyring = '/etc/ceph/{0}.{1}.keyring'.format(cluster_name,
                                                                name)
            r = _shell(
                ctx=ctx,
                cluster_name=cluster_name,
                remote=remote,
                args=[
                    'ceph', 'auth',
                    'get-or-create', name,
                    'mon', 'allow *',
                    'osd', 'allow *',
                    'mds', 'allow *',
                    'mgr', 'allow *',
                ],
                stdout=StringIO(),
            )
            keyring = r.stdout.getvalue()
            remote.sudo_write_file(client_keyring, keyring, mode='0644')
    yield


@contextlib.contextmanager
def ceph_initial():
    try:
        yield
    finally:
        log.info('Teardown complete')


## public methods
@contextlib.contextmanager
def stop(ctx, config):
    """
    Stop ceph daemons

    For example::
      tasks:
      - ceph.stop: [mds.*]

      tasks:
      - ceph.stop: [osd.0, osd.2]

      tasks:
      - ceph.stop:
          daemons: [osd.0, osd.2]

    """
    if config is None:
        config = {}
    elif isinstance(config, list):
        config = {'daemons': config}

    daemons = ctx.daemons.resolve_role_list(
        config.get('daemons', None), CEPH_ROLE_TYPES, True)
    clusters = set()

    for role in daemons:
        cluster, type_, id_ = teuthology.split_role(role)
        ctx.daemons.get_daemon(type_, id_, cluster).stop()
        clusters.add(cluster)

#    for cluster in clusters:
#        ctx.ceph[cluster].watchdog.stop()
#        ctx.ceph[cluster].watchdog.join()

    yield


def shell(ctx, config):
    """
    Execute (shell) commands
    """
    cluster_name = config.get('cluster', 'ceph')

    args = []
    for k in config.pop('env', []):
        args.extend(['-e', k + '=' + ctx.config.get(k, '')])
    for k in config.pop('volumes', []):
        args.extend(['-v', k])

    if 'all-roles' in config and len(config) == 1:
        a = config['all-roles']
        roles = teuthology.all_roles(ctx.cluster)
        config = dict((id_, a) for id_ in roles if not id_.startswith('host.'))
    elif 'all-hosts' in config and len(config) == 1:
        a = config['all-hosts']
        roles = teuthology.all_roles(ctx.cluster)
        config = dict((id_, a) for id_ in roles if id_.startswith('host.'))

    for role, cmd in config.items():
        (remote,) = ctx.cluster.only(role).remotes.keys()
        log.info('Running commands on role %s host %s', role, remote.name)
        if isinstance(cmd, list):
            for c in cmd:
                _shell(ctx, cluster_name, remote,
                       ['bash', '-c', subst_vip(ctx, c)],
                       extra_cephadm_args=args)
        else:
            assert isinstance(cmd, str)
            _shell(ctx, cluster_name, remote,
                   ['bash', '-ex', '-c', subst_vip(ctx, cmd)],
                   extra_cephadm_args=args)


def apply(ctx, config):
    """
    Apply spec
    
      tasks:
        - cephadm.apply:
            specs:
            - service_type: rgw
              service_id: foo
              spec:
                rgw_frontend_port: 8000
            - service_type: rgw
              service_id: bar
              spec:
                rgw_frontend_port: 9000
                zone: bar
                realm: asdf

    """
    cluster_name = config.get('cluster', 'ceph')

    specs = config.get('specs', [])
    y = subst_vip(ctx, yaml.dump_all(specs))

    log.info(f'Applying spec(s):\n{y}')
    _shell(
        ctx, cluster_name, ctx.ceph[cluster_name].bootstrap_remote,
        ['ceph', 'orch', 'apply', '-i', '-'],
        stdin=y,
    )


def wait_for_service(ctx, config):
    """
    Wait for a service to be fully started

      tasks:
        - cephadm.wait_for_service:
            service: rgw.foo
            timeout: 60    # defaults to 300

    """
    cluster_name = config.get('cluster', 'ceph')
    timeout = config.get('timeout', 300)
    service = config.get('service')
    assert service

    log.info(
        f'Waiting for {cluster_name} service {service} to start (timeout {timeout})...'
    )
    with contextutil.safe_while(sleep=1, tries=timeout) as proceed:
        while proceed():
            r = _shell(
                ctx=ctx,
                cluster_name=cluster_name,
                remote=ctx.ceph[cluster_name].bootstrap_remote,
                args=[
                    'ceph', 'orch', 'ls', '-f', 'json',
                ],
                stdout=StringIO(),
            )
            j = json.loads(r.stdout.getvalue())
            svc = None
            for s in j:
                if s['service_name'] == service:
                    svc = s
                    break
            if svc:
                log.info(
                    f"{service} has {s['status']['running']}/{s['status']['size']}"
                )
                if s['status']['running'] == s['status']['size']:
                    break


@contextlib.contextmanager
def tweaked_option(ctx, config):
    """
    set an option, and then restore it with its original value

    Note, due to the way how tasks are executed/nested, it's not suggested to
    use this method as a standalone task. otherwise, it's likely that it will
    restore the tweaked option at the /end/ of 'tasks' block.
    """
    saved_options = {}
    # we can complicate this when necessary
    options = ['mon-health-to-clog']
    type_, id_ = 'mon', '*'
    cluster = config.get('cluster', 'ceph')
    manager = ctx.managers[cluster]
    if id_ == '*':
        get_from = next(teuthology.all_roles_of_type(ctx.cluster, type_))
    else:
        get_from = id_
    for option in options:
        if option not in config:
            continue
        value = 'true' if config[option] else 'false'
        option = option.replace('-', '_')
        old_value = manager.get_config(type_, get_from, option)
        if value != old_value:
            saved_options[option] = old_value
            manager.inject_args(type_, id_, option, value)
    yield
    for option, value in saved_options.items():
        manager.inject_args(type_, id_, option, value)


@contextlib.contextmanager
def restart(ctx, config):
    """
   restart ceph daemons

   For example::
      tasks:
      - ceph.restart: [all]

   For example::
      tasks:
      - ceph.restart: [osd.0, mon.1, mds.*]

   or::

      tasks:
      - ceph.restart:
          daemons: [osd.0, mon.1]
          wait-for-healthy: false
          wait-for-osds-up: true

    :param ctx: Context
    :param config: Configuration
    """
    if config is None:
        config = {}
    elif isinstance(config, list):
        config = {'daemons': config}

    daemons = ctx.daemons.resolve_role_list(
        config.get('daemons', None), CEPH_ROLE_TYPES, True)
    clusters = set()

    log.info('daemons %s' % daemons)
    with tweaked_option(ctx, config):
        for role in daemons:
            cluster, type_, id_ = teuthology.split_role(role)
            d = ctx.daemons.get_daemon(type_, id_, cluster)
            assert d, 'daemon %s does not exist' % role
            d.stop()
            if type_ == 'osd':
                ctx.managers[cluster].mark_down_osd(id_)
            d.restart()
            clusters.add(cluster)

    if config.get('wait-for-healthy', True):
        for cluster in clusters:
            healthy(ctx=ctx, config=dict(cluster=cluster))
    if config.get('wait-for-osds-up', False):
        for cluster in clusters:
            ctx.managers[cluster].wait_for_all_osds_up()
    yield


@contextlib.contextmanager
def distribute_config_and_admin_keyring(ctx, config):
    """
    Distribute a sufficient config and keyring for clients
    """
    cluster_name = config['cluster']
    log.info('Distributing (final) config and client.admin keyring...')
    for remote, roles in ctx.cluster.remotes.items():
        remote.write_file(
            '/etc/ceph/{}.conf'.format(cluster_name),
            ctx.ceph[cluster_name].config_file,
            sudo=True)
        remote.write_file(
            path='/etc/ceph/{}.client.admin.keyring'.format(cluster_name),
            data=ctx.ceph[cluster_name].admin_keyring,
            sudo=True)
    try:
        yield
    finally:
        ctx.cluster.run(args=[
            'sudo', 'rm', '-f',
            '/etc/ceph/{}.conf'.format(cluster_name),
            '/etc/ceph/{}.client.admin.keyring'.format(cluster_name),
        ])


@contextlib.contextmanager
def crush_setup(ctx, config):
    cluster_name = config['cluster']

    profile = config.get('crush_tunables', 'default')
    log.info('Setting crush tunables to %s', profile)
    _shell(ctx, cluster_name, ctx.ceph[cluster_name].bootstrap_remote,
        args=['ceph', 'osd', 'crush', 'tunables', profile])
    yield


@contextlib.contextmanager
def create_rbd_pool(ctx, config):
    if config.get('create_rbd_pool', False):
      cluster_name = config['cluster']
      log.info('Waiting for OSDs to come up')
      teuthology.wait_until_osds_up(
          ctx,
          cluster=ctx.cluster,
          remote=ctx.ceph[cluster_name].bootstrap_remote,
          ceph_cluster=cluster_name,
      )
      log.info('Creating RBD pool')
      _shell(ctx, cluster_name, ctx.ceph[cluster_name].bootstrap_remote,
          args=['sudo', 'ceph', '--cluster', cluster_name,
                'osd', 'pool', 'create', 'rbd', '8'])
      _shell(ctx, cluster_name, ctx.ceph[cluster_name].bootstrap_remote,
          args=['sudo', 'ceph', '--cluster', cluster_name,
                'osd', 'pool', 'application', 'enable',
                'rbd', 'rbd', '--yes-i-really-mean-it'
          ])
    yield


@contextlib.contextmanager
def _bypass():
    yield


@contextlib.contextmanager
def initialize_config(ctx, config):
    cluster_name = config['cluster']
    testdir = teuthology.get_testdir(ctx)

    ctx.ceph[cluster_name].thrashers = []
    # fixme: setup watchdog, ala ceph.py

    ctx.ceph[cluster_name].roleless = False  # see below

    first_ceph_cluster = False
    if not hasattr(ctx, 'daemons'):
        first_ceph_cluster = True

    # cephadm mode?
    if 'cephadm_mode' not in config:
        config['cephadm_mode'] = 'root'
    assert config['cephadm_mode'] in ['root', 'cephadm-package']
    if config['cephadm_mode'] == 'root':
        ctx.cephadm = testdir + '/cephadm'
    else:
        ctx.cephadm = 'cephadm'  # in the path

    if first_ceph_cluster:
        # FIXME: this is global for all clusters
        ctx.daemons = DaemonGroup(
            use_cephadm=ctx.cephadm)

    # uuid
    fsid = str(uuid.uuid1())
    log.info('Cluster fsid is %s' % fsid)
    ctx.ceph[cluster_name].fsid = fsid

    # mon ips
    log.info('Choosing monitor IPs and ports...')
    remotes_and_roles = ctx.cluster.remotes.items()
    ips = [host for (host, port) in
           (remote.ssh.get_transport().getpeername() for (remote, role_list) in remotes_and_roles)]

    if config.get('roleless', False):
        # mons will be named after hosts
        first_mon = None
        max_mons = config.get('max_mons', 5)
        for remote, _ in remotes_and_roles:
            ctx.cluster.remotes[remote].append('mon.' + remote.shortname)
            if not first_mon:
                first_mon = remote.shortname
                bootstrap_remote = remote
            max_mons -= 1
            if not max_mons:
                break
        log.info('No mon roles; fabricating mons')

    roles = [role_list for (remote, role_list) in ctx.cluster.remotes.items()]

    ctx.ceph[cluster_name].mons = get_mons(
        roles, ips, cluster_name,
        mon_bind_msgr2=config.get('mon_bind_msgr2', True),
        mon_bind_addrvec=config.get('mon_bind_addrvec', True),
    )
    log.info('Monitor IPs: %s' % ctx.ceph[cluster_name].mons)

    if config.get('roleless', False):
        ctx.ceph[cluster_name].roleless = True
        ctx.ceph[cluster_name].bootstrap_remote = bootstrap_remote
        ctx.ceph[cluster_name].first_mon = first_mon
        ctx.ceph[cluster_name].first_mon_role = 'mon.' + first_mon
    else:
        first_mon_role = sorted(ctx.ceph[cluster_name].mons.keys())[0]
        _, _, first_mon = teuthology.split_role(first_mon_role)
        (bootstrap_remote,) = ctx.cluster.only(first_mon_role).remotes.keys()
        log.info('First mon is mon.%s on %s' % (first_mon,
                                                bootstrap_remote.shortname))
        ctx.ceph[cluster_name].bootstrap_remote = bootstrap_remote
        ctx.ceph[cluster_name].first_mon = first_mon
        ctx.ceph[cluster_name].first_mon_role = first_mon_role

        others = ctx.cluster.remotes[bootstrap_remote]
        mgrs = sorted([r for r in others
                       if teuthology.is_type('mgr', cluster_name)(r)])
        if not mgrs:
            raise RuntimeError('no mgrs on the same host as first mon %s' % first_mon)
        _, _, first_mgr = teuthology.split_role(mgrs[0])
        log.info('First mgr is %s' % (first_mgr))
        ctx.ceph[cluster_name].first_mgr = first_mgr
    yield


@contextlib.contextmanager
def task(ctx, config):
    """
    Deploy ceph cluster using cephadm

    For example, teuthology.yaml can contain the 'defaults' section:

        defaults:
          cephadm:
            containers:
              image: 'quay.io/ceph-ci/ceph'

    Using overrides makes it possible to customize it per run.
    The equivalent 'overrides' section looks like:

        overrides:
          cephadm:
            containers:
              image: 'quay.io/ceph-ci/ceph'
            registry-login:
              url:  registry-url
              username: registry-user
              password: registry-password

    :param ctx: the argparse.Namespace object
    :param config: the config dict
    """
    if config is None:
        config = {}

    assert isinstance(config, dict), \
        "task only supports a dictionary for configuration"

    overrides = ctx.config.get('overrides', {})
    teuthology.deep_merge(config, overrides.get('ceph', {}))
    teuthology.deep_merge(config, overrides.get('cephadm', {}))
    log.info('Config: ' + str(config))

    # set up cluster context
    if not hasattr(ctx, 'ceph'):
        ctx.ceph = {}
    if 'cluster' not in config:
        config['cluster'] = 'ceph'
    cluster_name = config['cluster']
    if cluster_name not in ctx.ceph:
        ctx.ceph[cluster_name] = argparse.Namespace()
        ctx.ceph[cluster_name].bootstrapped = False

    # image
    teuth_defaults = teuth_config.get('defaults', {})
    cephadm_defaults = teuth_defaults.get('cephadm', {})
    containers_defaults = cephadm_defaults.get('containers', {})
    container_image_name = containers_defaults.get('image', None)

    containers = config.get('containers', {})
    container_image_name = containers.get('image', container_image_name)

    if not hasattr(ctx.ceph[cluster_name], 'image'):
        ctx.ceph[cluster_name].image = config.get('image')
    ref = ctx.config.get("branch", "main")
    if not ctx.ceph[cluster_name].image:
        if not container_image_name:
            raise Exception("Configuration error occurred. "
                            "The 'image' value is undefined for 'cephadm' task. "
                            "Please provide corresponding options in the task's "
                            "config, task 'overrides', or teuthology 'defaults' "
                            "section.")
        sha1 = config.get('sha1')
        flavor = config.get('flavor', 'default')

        if sha1:
            if flavor == "crimson":
                ctx.ceph[cluster_name].image = container_image_name + ':' + sha1 + '-' + flavor
            else:
                ctx.ceph[cluster_name].image = container_image_name + ':' + sha1
            ref = sha1
        else:
            # fall back to using the branch value
            ctx.ceph[cluster_name].image = container_image_name + ':' + ref
    log.info('Cluster image is %s' % ctx.ceph[cluster_name].image)


    with contextutil.nested(
            #if the cluster is already bootstrapped bypass corresponding methods
            lambda: _bypass() if (ctx.ceph[cluster_name].bootstrapped) \
                              else initialize_config(ctx=ctx, config=config),
            lambda: ceph_initial(),
            lambda: normalize_hostnames(ctx=ctx),
            lambda: _bypass() if (ctx.ceph[cluster_name].bootstrapped) \
                              else download_cephadm(ctx=ctx, config=config, ref=ref),
            lambda: ceph_log(ctx=ctx, config=config),
            lambda: ceph_crash(ctx=ctx, config=config),
            lambda: pull_image(ctx=ctx, config=config),
            lambda: _bypass() if not (config.get('use-ca-signed-key', False)) \
                              else setup_ca_signed_keys(ctx, config),
            lambda: _bypass() if (ctx.ceph[cluster_name].bootstrapped) \
                              else ceph_bootstrap(ctx, config),
            lambda: crush_setup(ctx=ctx, config=config),
            lambda: ceph_mons(ctx=ctx, config=config),
            lambda: distribute_config_and_admin_keyring(ctx=ctx, config=config),
            lambda: ceph_mgrs(ctx=ctx, config=config),
            lambda: ceph_osds(ctx=ctx, config=config),
            lambda: ceph_mdss(ctx=ctx, config=config),
            lambda: cephfs_setup(ctx=ctx, config=config),
            lambda: ceph_rgw(ctx=ctx, config=config),
            lambda: ceph_iscsi(ctx=ctx, config=config),
            lambda: ceph_monitoring('prometheus', ctx=ctx, config=config),
            lambda: ceph_monitoring('node-exporter', ctx=ctx, config=config),
            lambda: ceph_monitoring('alertmanager', ctx=ctx, config=config),
            lambda: ceph_monitoring('grafana', ctx=ctx, config=config),
            lambda: ceph_clients(ctx=ctx, config=config),
            lambda: create_rbd_pool(ctx=ctx, config=config),
    ):
        try:
            if config.get('wait-for-healthy', True):
                healthy(ctx=ctx, config=config)

            log.info('Setup complete, yielding')
            yield

        finally:
            log.info('Teardown begin')