def create_users(ctx, config):
    """
    Create a main and an alternate s3 user.
    """
    assert isinstance(config, dict)
    log.info('Creating rgw users...')
    testdir = teuthology.get_testdir(ctx)
    users = {'s3 main': 'foo', 's3 alt': 'bar'}
    for client in config['clients']:
        s3tests_conf = config['s3tests_conf'][client]
        s3tests_conf.setdefault('fixtures', {})
        s3tests_conf['fixtures'].setdefault('bucket prefix', 'test-' + client + '-{random}-')
        for section, user in users.iteritems():
            _config_user(s3tests_conf, section, '{user}.{client}'.format(user=user, client=client))
            log.debug('Creating user {user} on {host}'.format(user=s3tests_conf[section]['user_id'], host=client))
            cluster_name, daemon_type, client_id = teuthology.split_role(client)
            client_with_id = daemon_type + '.' + client_id
            ctx.cluster.only(client).run(
                args=[
                    'adjust-ulimits',
                    'ceph-coverage',
                    '{tdir}/archive/coverage'.format(tdir=testdir),
                    'radosgw-admin',
                    '-n', client_with_id,
                    'user', 'create',
                    '--uid', s3tests_conf[section]['user_id'],
                    '--display-name', s3tests_conf[section]['display_name'],
                    '--access-key', s3tests_conf[section]['access_key'],
                    '--secret', s3tests_conf[section]['secret_key'],
                    '--email', s3tests_conf[section]['email'],
                    '--cluster', cluster_name,
                ],
            )
    try:
        yield
    finally:
        for client in config['clients']:
            for user in users.itervalues():
                uid = '{user}.{client}'.format(user=user, client=client)
                cluster_name, daemon_type, client_id = teuthology.split_role(client)
                client_with_id = daemon_type + '.' + client_id
                ctx.cluster.only(client).run(
                    args=[
                        'adjust-ulimits',
                        'ceph-coverage',
                        '{tdir}/archive/coverage'.format(tdir=testdir),
                        'radosgw-admin',
                        '-n', client_with_id,
                        'user', 'rm',
                        '--uid', uid,
                        '--purge-data',
                        '--cluster', cluster_name,
                        ],
                    )

def create_users(ctx, config):
    """
    Create rgw users to interact with the swift interface.
    """
    assert isinstance(config, dict)
    log.info('Creating rgw users...')
    testdir = teuthology.get_testdir(ctx)
    users = {'': 'foo', '2': 'bar'}
    for client in config['clients']:
        cluster_name, daemon_type, client_id = teuthology.split_role(client)
        testswift_conf = config['testswift_conf'][client]
        for suffix, user in users.iteritems():
            _config_user(testswift_conf, '{user}.{client}'.format(user=user, client=client), user, suffix)
            ctx.cluster.only(client).run(
                args=[
                    'adjust-ulimits',
                    'ceph-coverage',
                    '{tdir}/archive/coverage'.format(tdir=testdir),
                    'radosgw-admin',
                    '-n', client,
                    '--cluster', cluster_name,
                    'user', 'create',
                    '--subuser', '{account}:{user}'.format(account=testswift_conf['func_test']['account{s}'.format(s=suffix)],user=user),
                    '--display-name', testswift_conf['func_test']['display_name{s}'.format(s=suffix)],
                    '--secret', testswift_conf['func_test']['password{s}'.format(s=suffix)],
                    '--email', testswift_conf['func_test']['email{s}'.format(s=suffix)],
                    '--key-type', 'swift',
                    '--access', 'full',
                ],
            )
    try:
        yield
    finally:
        for client in config['clients']:
            for user in users.itervalues():
                uid = '{user}.{client}'.format(user=user, client=client)
                cluster_name, daemon_type, client_id = teuthology.split_role(client)
                ctx.cluster.only(client).run(
                    args=[
                        'adjust-ulimits',
                        'ceph-coverage',
                        '{tdir}/archive/coverage'.format(tdir=testdir),
                        'radosgw-admin',
                        '-n', client,
                        '--cluster', cluster_name,
                        'user', 'rm',
                        '--uid', uid,
                        '--purge-data',
                        ],
                    )

def wait_for_failure(ctx, config):
    """
    Wait for a failure of a ceph daemon

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

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

      tasks:
      - ceph.wait_for_failure:
          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)
    for role in daemons:
        cluster, type_, id_ = teuthology.split_role(role)
        try:
            ctx.daemons.get_daemon(type_, id_, cluster).wait()
        except:
            log.info("Saw expected daemon failure.  Continuing.")
            pass
        else:
            raise RuntimeError("daemon %s did not fail" % role)

    yield

def task(ctx, config):
    """
    Test monitor recovery from OSD
    """
    if config is None:
        config = {}
    assert isinstance(config, dict), \
        'task only accepts a dict for configuration'

    first_mon = teuthology.get_first_mon(ctx, config)
    (mon,) = ctx.cluster.only(first_mon).remotes.iterkeys()
    manager = ceph_manager.CephManager(
        mon,
        ctx=ctx,
        logger=log.getChild('ceph_manager'))

    mons = ctx.cluster.only(teuthology.is_type('mon'))
    # note down the first cluster_name and mon_id
    # we will recover it later on
    cluster_name, _, mon_id = teuthology.split_role(first_mon)
    _nuke_mons(manager, mons, mon_id)
    default_keyring = '/etc/ceph/{cluster}.keyring'.format(
        cluster=cluster_name)
    keyring_path = config.get('keyring_path', default_keyring)
    _rebuild_db(ctx, manager, cluster_name, mon, mon_id, keyring_path)
    _revive_mons(manager, mons, mon_id, keyring_path)
    _revive_mgrs(ctx, manager)
    _revive_osds(ctx, manager)

def _revive_mons(manager, mons, recovered, keyring_path):
    # revive monitors
    # the initial monmap is in the ceph.conf, so we are good.
    n_mons = 0
    is_mon = teuthology.is_type('mon')
    for remote, roles in mons.remotes.iteritems():
        for role in roles:
            if not is_mon(role):
                continue
            cluster, _, m = teuthology.split_role(role)
            if recovered != m:
                log.info('running mkfs on {cluster}:mon.{mon}'.format(
                    cluster=cluster,
                    mon=m))
                remote.run(
                    args=[
                        'sudo',
                        'ceph-mon',
                        '--cluster', cluster,
                        '--mkfs',
                        '-i', m,
                        '--keyring', keyring_path])
            log.info('reviving mon.{0}'.format(m))
            manager.revive_mon(m)
            n_mons += 1
    manager.wait_for_mon_quorum_size(n_mons, timeout=30)

def _delete_dir(ctx, role, created_mountpoint):
    """
    Delete file used by this role, and delete the directory that this
    role appeared in.

    :param ctx: Context
    :param role: "role.#" where # is used for the role id.
    """
    cluster, _, id_ = misc.split_role(role)
    remote = get_remote_for_role(ctx, role)
    mnt = _client_mountpoint(ctx, cluster, id_)
    client = os.path.join(mnt, 'client.{id}'.format(id=id_))

    # Remove the directory inside the mount where the workunit ran
    remote.run(
        args=[
            'sudo',
            'rm',
            '-rf',
            '--',
            client,
        ],
    )
    log.info("Deleted dir {dir}".format(dir=client))

    # If the mount was an artificially created dir, delete that too
    if created_mountpoint:
        remote.run(
            args=[
                'rmdir',
                '--',
                mnt,
            ],
        )
        log.info("Deleted artificial mount point {dir}".format(dir=client))

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)
    for role in daemons:
        cluster, type_, id_ = teuthology.split_role(role)
        ctx.daemons.get_daemon(type_, id_, cluster).stop()

    yield

def osd_scrub_pgs(ctx, config):
    """
    Scrub pgs when we exit.

    First make sure all pgs are active and clean.
    Next scrub all osds.
    Then periodically check until all pgs have scrub time stamps that
    indicate the last scrub completed.  Time out if no progess is made
    here after two minutes.
    """
    retries = 12
    delays = 10
    cluster_name = config["cluster"]
    manager = ctx.managers[cluster_name]
    all_clean = False
    for _ in range(0, retries):
        stats = manager.get_pg_stats()
        states = [stat["state"] for stat in stats]
        if len(set(states)) == 1 and states[0] == "active+clean":
            all_clean = True
            break
        log.info("Waiting for all osds to be active and clean.")
        time.sleep(delays)
    if not all_clean:
        log.info("Scrubbing terminated -- not all pgs were active and clean.")
        return
    check_time_now = time.localtime()
    time.sleep(1)
    all_roles = teuthology.all_roles(ctx.cluster)
    for role in teuthology.cluster_roles_of_type(all_roles, "osd", cluster_name):
        log.info("Scrubbing {osd}".format(osd=role))
        _, _, id_ = teuthology.split_role(role)
        manager.raw_cluster_cmd("osd", "deep-scrub", id_)
    prev_good = 0
    gap_cnt = 0
    loop = True
    while loop:
        stats = manager.get_pg_stats()
        timez = [stat["last_scrub_stamp"] for stat in stats]
        loop = False
        thiscnt = 0
        for tmval in timez:
            pgtm = time.strptime(tmval[0 : tmval.find(".")], "%Y-%m-%d %H:%M:%S")
            if pgtm > check_time_now:
                thiscnt += 1
            else:
                loop = True
        if thiscnt > prev_good:
            prev_good = thiscnt
            gap_cnt = 0
        else:
            gap_cnt += 1
            if gap_cnt > retries:
                log.info("Exiting scrub checking -- not all pgs scrubbed.")
                return
        if loop:
            log.info("Still waiting for all pgs to be scrubbed.")
            time.sleep(delays)

def start_apache(ctx, config, on_client = None, except_client = None):
    """
    Start apache on remote sites.
    """
    log.info('Starting apache...')
    testdir = teuthology.get_testdir(ctx)
    apaches = {}
    clients_to_run = [on_client]
    if on_client is None:
        clients_to_run = config.keys()
    for client in clients_to_run:
        cluster_name, daemon_type, client_id = teuthology.split_role(client)
        client_with_cluster = cluster_name + '.' + daemon_type + '.' + client_id
        if client == except_client:
            continue
        (remote,) = ctx.cluster.only(client).remotes.keys()
        system_type = teuthology.get_system_type(remote)
        if system_type == 'deb':
            apache_name = 'apache2'
        else:
            try:
                remote.run(
                    args=[
                        'stat',
                        '/usr/sbin/httpd.worker',
                    ],
                )
                apache_name = '/usr/sbin/httpd.worker'
            except CommandFailedError:
                apache_name = '/usr/sbin/httpd'

        proc = remote.run(
            args=[
                'adjust-ulimits',
                'daemon-helper',
                'kill',
                apache_name,
                '-X',
                '-f',
                '{tdir}/apache/apache.{client_with_cluster}.conf'.format(tdir=testdir,
                                                            client_with_cluster=client_with_cluster),
                ],
            logger=log.getChild(client),
            stdin=run.PIPE,
            wait=False,
            )
        apaches[client_with_cluster] = proc

    try:
        yield
    finally:
        log.info('Stopping apache...')
        for client, proc in apaches.iteritems():
            proc.stdin.close()

        run.wait(apaches.itervalues())

def _revive_osds(ctx, manager):
    is_osd = teuthology.is_type('osd')
    osds = ctx.cluster.only(is_osd)
    for _, roles in osds.remotes.iteritems():
        for role in roles:
            if not is_osd(role):
                continue
            _, _, osd_id = teuthology.split_role(role)
            log.info('reviving osd.{0}'.format(osd_id))
            manager.revive_osd(osd_id)

def _revive_mgrs(ctx, manager):
    is_mgr = teuthology.is_type('mgr')
    mgrs = ctx.cluster.only(is_mgr)
    for _, roles in mgrs.remotes.iteritems():
        for role in roles:
            if not is_mgr(role):
                continue
            _, _, mgr_id = teuthology.split_role(role)
            log.info('reviving mgr.{0}'.format(mgr_id))
            manager.revive_mgr(mgr_id)

def create_apache_dirs(ctx, config, on_client = None, except_client = None):
    """
    Remotely create apache directories.  Delete when finished.
    """
    log.info('Creating apache directories...')
    log.debug('client is %r', on_client)
    testdir = teuthology.get_testdir(ctx)
    clients_to_create_as = [on_client]
    if on_client is None:
        clients_to_create_as = config.keys()
    for client in clients_to_create_as:
        if client == except_client:
            continue
    	cluster_name, daemon_type, client_id = teuthology.split_role(client)
    	client_with_cluster = cluster_name + '.' + daemon_type + '.' + client_id
        ctx.cluster.only(client).run(
            args=[
                'mkdir',
                '-p',
                '{tdir}/apache/htdocs.{client_with_cluster}'.format(tdir=testdir,
                                                       client_with_cluster=client_with_cluster),
                '{tdir}/apache/tmp.{client_with_cluster}/fastcgi_sock'.format(
                    tdir=testdir,
                    client_with_cluster=client_with_cluster),
                run.Raw('&&'),
                'mkdir',
                '{tdir}/archive/apache.{client_with_cluster}'.format(tdir=testdir,
                                                        client_with_cluster=client_with_cluster),
                ],
            )
    try:
        yield
    finally:
        log.info('Cleaning up apache directories...')
        for client in clients_to_create_as:
            ctx.cluster.only(client).run(
                args=[
                    'rm',
                    '-rf',
                    '{tdir}/apache/tmp.{client_with_cluster}'.format(tdir=testdir,
                                                        client_with_cluster=client_with_cluster),
                    run.Raw('&&'),
                    'rmdir',
                    '{tdir}/apache/htdocs.{client_with_cluster}'.format(tdir=testdir,
                                                           client_with_cluster=client_with_cluster),
                    ],
                )
        for client in clients_to_create_as:
            ctx.cluster.only(client).run(
                args=[
                    'rmdir',
                    '{tdir}/apache'.format(tdir=testdir),
                    ],
                check_status=False,  # only need to remove once per host
                )

    def setup(self):
        super(RBDMirror, self).setup()
        try:
            self.client = self.config['client']
        except KeyError:
            raise ConfigError('rbd-mirror requires a client to connect with')

        self.cluster_name, type_, self.client_id = misc.split_role(self.client)

        if type_ != 'client':
            msg = 'client role ({0}) must be a client'.format(self.client)
            raise ConfigError(msg)

        self.remote = get_remote_for_role(self.ctx, self.client)

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()
    for role in daemons:
        cluster, type_, id_ = teuthology.split_role(role)
        ctx.daemons.get_daemon(type_, id_, cluster).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:
            wait_for_osds_up(ctx=ctx, config=dict(cluster=cluster))
    manager = ctx.managers["ceph"]
    for dmon in daemons:
        if "." in dmon:
            dm_parts = dmon.split(".")
            if dm_parts[1].isdigit():
                if dm_parts[0] == "osd":
                    manager.mark_down_osd(int(dm_parts[1]))
    yield

def extract_zone_cluster_name(zone_config):
    """ return the cluster (must be common to all zone endpoints) """
    cluster_name = None
    endpoints = zone_config.get('endpoints')
    if not endpoints:
        raise ConfigError('zone %s missing \'endpoints\' list' % \
                          zone_config['name'])
    for role in endpoints:
        name, _, _ = misc.split_role(role)
        if not cluster_name:
            cluster_name = name
        elif cluster_name != name:
            raise ConfigError('all zone %s endpoints must be in the same cluster' % \
                              zone_config['name'])
    return cluster_name

def get_zone_host_and_port(ctx, client, zone):
    cluster_name, daemon_type, client_id = teuthology.split_role(client)
    client_with_id = daemon_type + '.' + client_id
    _, period = rgwadmin(ctx, client, check_status=True,
                         cmd=['period', 'get'])
    period_map = period['period_map']
    zonegroups = period_map['zonegroups']
    for zonegroup in zonegroups:
        for zone_info in zonegroup['zones']:
            if zone_info['name'] == zone:
                endpoint = urlparse(zone_info['endpoints'][0])
                host, port = endpoint.hostname, endpoint.port
                if port is None:
                    port = 80
                return host, port
    assert False, 'no endpoint for zone {zone} found'.format(zone=zone)

def validate_config(ctx, config):
    """
    Perform some simple validation on task configuration.
    Raises exceptions.ConfigError if an error is found.
    """
    # check for osds from multiple clusters on the same host
    for remote, roles_for_host in ctx.cluster.remotes.items():
        last_cluster = None
        last_role = None
        for role in roles_for_host:
            role_cluster, role_type, _ = teuthology.split_role(role)
            if role_type != "osd":
                continue
            if last_cluster and last_cluster != role_cluster:
                msg = "Host should not have osds (%s and %s) from multiple clusters" % (last_role, role)
                raise exceptions.ConfigError(msg)
            last_cluster = role_cluster
            last_role = role

def extract_sync_client_data(ctx, client_name):
    """
    Extract synchronized client rgw zone and rgw region information.

    :param ctx: Context passed to the s3tests task
    :param name: Name of client that we are synching with
    """
    return_region_name = None
    return_dict = None
    cluster_name, daemon_type, client_id = teuthology.split_role(client_name)
    client = ctx.ceph[cluster_name].conf.get(client_name, None)
    if client:
        current_client_zone = client.get('rgw zone', None)
        if current_client_zone:
            (endpoint_host, endpoint_port) = ctx.rgw.role_endpoints.get(client_name, (None, None))
            # pull out the radosgw_agent stuff
            regions = ctx.rgw.regions
            for region in regions:
                log.debug('jbuck, region is {region}'.format(region=region))
                region_data = ctx.rgw.regions[region]
                log.debug('region data is {region}'.format(region=region_data))
                zones = region_data['zones']
                for zone in zones:
                    if current_client_zone in zone:
                        return_region_name = region
                        return_dict = dict()
                        return_dict['api_name'] = region_data['api name']
                        return_dict['is_master'] = region_data['is master']
                        return_dict['port'] = endpoint_port
                        return_dict['host'] = endpoint_host

                        # The s3tests expect the sync_agent_[addr|port} to be
                        # set on the non-master node for some reason
                        if not region_data['is master']:
                            (rgwagent_host, rgwagent_port) = ctx.radosgw_agent.endpoint
                            (return_dict['sync_agent_addr'], _) = ctx.rgw.role_endpoints[rgwagent_host]
                            return_dict['sync_agent_port'] = rgwagent_port

        else: #if client_zone:
            log.debug('No zone info for {host}'.format(host=client_name))
    else: # if client
        log.debug('No ceph conf for {host}'.format(host=client_name))

    return return_region_name, return_dict

def create_pools(ctx, config):
    """Create replicated or erasure coded data pools for rgw."""

    log.info('Creating data pools')
    for client in config.keys():
        (remote,) = ctx.cluster.only(client).remotes.iterkeys()
        data_pool = '.rgw.buckets'
        cluster_name, daemon_type, client_id = teuthology.split_role(client)

        if ctx.rgw.ec_data_pool:
            create_ec_pool(remote, data_pool, client, 64,
                           ctx.rgw.erasure_code_profile, cluster_name)
        else:
            create_replicated_pool(remote, data_pool, 64, cluster_name)
        if ctx.rgw.cache_pools:
            create_cache_pool(remote, data_pool, data_pool + '.cache', 64,
                              64*1024*1024, cluster_name)
    log.debug('Pools created')
    yield

def get_master_zone(ctx, client):
    cluster_name, daemon_type, client_id = teuthology.split_role(client)
    client_with_id = daemon_type + '.' + client_id
    _, period = rgwadmin(ctx, client, check_status=True,
                         cmd=['period', 'get'])
    period_map = period['period_map']
    zonegroups = period_map['zonegroups']
    for zonegroup in zonegroups:
        is_master = (zonegroup['is_master'] == "true")
        log.info('zonegroup={z} is_master={ism}'.format(z=zonegroup, ism=is_master))
        if not is_master:
          continue
        master_zone = zonegroup['master_zone']
        log.info('master_zone=%s' % master_zone)
        for zone_info in zonegroup['zones']:
            if zone_info['name'] == master_zone:
                return master_zone
    log.info('couldn\'t find master zone')
    return None