def download(ctx, config):
    """
    Download the Swift API.
    """
    testdir = teuthology.get_testdir(ctx)
    assert isinstance(config, list)
    log.info('Downloading swift...')
    for client in config:
        ctx.cluster.only(client).run(
            args=[
                'git', 'clone',
                teuth_config.ceph_git_base_url + 'swift.git',
                '{tdir}/swift'.format(tdir=testdir),
                ],
            )
    try:
        yield
    finally:
        log.info('Removing swift...')
        testdir = teuthology.get_testdir(ctx)
        for client in config:
            ctx.cluster.only(client).run(
                args=[
                    'rm',
                    '-rf',
                    '{tdir}/swift'.format(tdir=testdir),
                    ],
                )

def download(ctx, config):
    """
    Download the Swift API.
    """
    testdir = teuthology.get_testdir(ctx)
    assert isinstance(config, dict)
    log.info('Downloading swift...')
    for (client, cconf) in config.items():
        ctx.cluster.only(client).run(
            args=[
                'git', 'clone',
                '-b', cconf.get('force-branch', 'ceph-master'),
                teuth_config.ceph_git_base_url + 'swift.git',
                '{tdir}/swift'.format(tdir=testdir),
                ],
            )
    try:
        yield
    finally:
        log.info('Removing swift...')
        testdir = teuthology.get_testdir(ctx)
        for (client, _) in config.items():
            ctx.cluster.only(client).run(
                args=[
                    'rm',
                    '-rf',
                    '{tdir}/swift'.format(tdir=testdir),
                    ],
                )

def download(ctx, config):
    testdir = teuthology.get_testdir(ctx)
    assert isinstance(config, list)
    log.info('Downloading swift...')
    for client in config:
        ctx.cluster.only(client).run(
            args=[
                'git', 'clone',
                'git://ceph.com/git/swift.git',
                '{tdir}/swift'.format(tdir=testdir),
                ],
            )
    try:
        yield
    finally:
        log.info('Removing swift...')
        testdir = teuthology.get_testdir(ctx)
        for client in config:
            ctx.cluster.only(client).run(
                args=[
                    'rm',
                    '-rf',
                    '{tdir}/swift'.format(tdir=testdir),
                    ],
                )

def download(ctx, config):
    assert isinstance(config, dict)
    log.info("Downloading s3-tests...")
    testdir = teuthology.get_testdir(ctx)
    for (client, cconf) in config.items():
        branch = cconf.get("force-branch", None)
        if not branch:
            branch = cconf.get("branch", "master")
        sha1 = cconf.get("sha1")
        ctx.cluster.only(client).run(
            args=[
                "git",
                "clone",
                "-b",
                branch,
                teuth_config.ceph_git_base_url + "s3-tests.git",
                "{tdir}/s3-tests".format(tdir=testdir),
            ]
        )
        if sha1 is not None:
            ctx.cluster.only(client).run(
                args=["cd", "{tdir}/s3-tests".format(tdir=testdir), run.Raw("&&"), "git", "reset", "--hard", sha1]
            )
    try:
        yield
    finally:
        log.info("Removing s3-tests...")
        testdir = teuthology.get_testdir(ctx)
        for client in config:
            ctx.cluster.only(client).run(args=["rm", "-rf", "{tdir}/s3-tests".format(tdir=testdir)])

def configure(ctx, config, hadoops):
    tempdir = teuthology.get_testdir(ctx)

    log.info("Writing Hadoop slaves file...")
    for remote in hadoops.remotes:
        path, data = get_slaves_data(ctx)
        teuthology.write_file(remote, path, StringIO(data))

    log.info("Writing Hadoop masters file...")
    for remote in hadoops.remotes:
        path, data = get_masters_data(ctx)
        teuthology.write_file(remote, path, StringIO(data))

    log.info("Writing Hadoop core-site.xml file...")
    for remote in hadoops.remotes:
        path, data = get_core_site_data(ctx, config)
        teuthology.write_file(remote, path, StringIO(data))

    log.info("Writing Hadoop yarn-site.xml file...")
    for remote in hadoops.remotes:
        path, data = get_yarn_site_data(ctx)
        teuthology.write_file(remote, path, StringIO(data))

    log.info("Writing Hadoop hdfs-site.xml file...")
    for remote in hadoops.remotes:
        path, data = get_hdfs_site_data(ctx)
        teuthology.write_file(remote, path, StringIO(data))

    log.info("Writing Hadoop mapred-site.xml file...")
    for remote in hadoops.remotes:
        path, data = get_mapred_site_data(ctx)
        teuthology.write_file(remote, path, StringIO(data))

    log.info("Setting JAVA_HOME in hadoop-env.sh...")
    for remote in hadoops.remotes:
        path = "{tdir}/hadoop/etc/hadoop/hadoop-env.sh".format(tdir=tempdir)
        if remote.os.package_type == 'rpm':
            data = "JAVA_HOME=/usr/lib/jvm/java\n"
        elif remote.os.package_type == 'deb':
            data = "JAVA_HOME=/usr/lib/jvm/default-java\n"
        else:
            raise UnsupportedPackageTypeError(remote)
        teuthology.prepend_lines_to_file(remote, path, data)

    if config.get('hdfs', False):
        log.info("Formatting HDFS...")
        testdir = teuthology.get_testdir(ctx)
        hadoop_dir = "{tdir}/hadoop/".format(tdir=testdir)
        masters = ctx.cluster.only(teuthology.is_type('hadoop.master'))
        assert len(masters.remotes) == 1
        master = masters.remotes.keys()[0]
        master.run(
            args = [
                hadoop_dir + "bin/hadoop",
                "namenode",
                "-format"
            ],
            wait = True,
            )

def binaries(ctx, config):
    path = config.get("path")

    if path is None:
        # fetch Apache Hadoop from gitbuilder
        log.info("Fetching and unpacking Apache Hadoop binaries from gitbuilder...")
        apache_sha1, apache_hadoop_bindir_url = teuthology.get_ceph_binary_url(
            package="apache-hadoop",
            branch=config.get("apache_branch"),
            tag=config.get("tag"),
            sha1=config.get("sha1"),
            flavor=config.get("flavor"),
            format=config.get("format"),
            dist=config.get("dist"),
            arch=config.get("arch"),
        )
        log.info("apache_hadoop_bindir_url %s" % (apache_hadoop_bindir_url))
        ctx.summary["apache-hadoop-sha1"] = apache_sha1

        # fetch Inktank Hadoop from gitbuilder
        log.info("Fetching and unpacking Inktank Hadoop binaries from gitbuilder...")
        inktank_sha1, inktank_hadoop_bindir_url = teuthology.get_ceph_binary_url(
            package="hadoop",
            branch=config.get("inktank_branch"),
            tag=config.get("tag"),
            sha1=config.get("sha1"),
            flavor=config.get("flavor"),
            format=config.get("format"),
            dist=config.get("dist"),
            arch=config.get("arch"),
        )
        log.info("inktank_hadoop_bindir_url %s" % (inktank_hadoop_bindir_url))
        ctx.summary["inktank-hadoop-sha1"] = inktank_sha1

    else:
        raise Exception("The hadoop task does not support the path argument at present")

    with parallel() as p:
        hadoopNodes = ctx.cluster.only(teuthology.is_type("hadoop"))
        # these can happen independently
        for remote in hadoopNodes.remotes.iterkeys():
            p.spawn(_node_binaries, ctx, config, remote, inktank_hadoop_bindir_url, apache_hadoop_bindir_url)

    try:
        yield
    finally:
        log.info("Removing hadoop binaries...")
        run.wait(
            ctx.cluster.run(
                args=["rm", "-rf", "--", "{tdir}/apache_hadoop".format(tdir=teuthology.get_testdir(ctx))], wait=False
            )
        )
        run.wait(
            ctx.cluster.run(
                args=["rm", "-rf", "--", "{tdir}/inktank_hadoop".format(tdir=teuthology.get_testdir(ctx))], wait=False
            )
        )

def download(ctx, config):
    """
    Download the s3 tests from the git builder.
    Remove downloaded s3 file upon exit.

    The context passed in should be identical to the context
    passed in to the main task.
    """
    assert isinstance(config, dict)
    log.info('Downloading s3-tests...')
    testdir = teuthology.get_testdir(ctx)
    s3_branches = [ 'giant', 'firefly', 'firefly-original', 'hammer' ]
    for (client, cconf) in config.items():
        branch = cconf.get('force-branch', None)
        if not branch:
            ceph_branch = ctx.config.get('branch')
            suite_branch = ctx.config.get('suite_branch', ceph_branch)
            if suite_branch in s3_branches:
                branch = cconf.get('branch', suite_branch)
	    else:
                branch = cconf.get('branch', 'ceph-' + suite_branch)
        if not branch:
            raise ValueError(
                "Could not determine what branch to use for s3tests!")
        else:
            log.info("Using branch '%s' for s3tests", branch)
        sha1 = cconf.get('sha1')
        git_remote = cconf.get('git_remote', None) or teuth_config.ceph_git_base_url
        ctx.cluster.only(client).run(
            args=[
                'git', 'clone',
                '-b', branch,
                git_remote + 's3-tests.git',
                '{tdir}/s3-tests'.format(tdir=testdir),
                ],
            )
        if sha1 is not None:
            ctx.cluster.only(client).run(
                args=[
                    'cd', '{tdir}/s3-tests'.format(tdir=testdir),
                    run.Raw('&&'),
                    'git', 'reset', '--hard', sha1,
                    ],
                )
    try:
        yield
    finally:
        log.info('Removing s3-tests...')
        testdir = teuthology.get_testdir(ctx)
        for client in config:
            ctx.cluster.only(client).run(
                args=[
                    'rm',
                    '-rf',
                    '{tdir}/s3-tests'.format(tdir=testdir),
                    ],
                )

def download(ctx, config):
    testdir = teuthology.get_testdir(ctx)
    assert isinstance(config, list)
    log.info("Downloading swift...")
    for client in config:
        ctx.cluster.only(client).run(
            args=["git", "clone", teuth_config.ceph_git_base_url + "swift.git", "{tdir}/swift".format(tdir=testdir)]
        )
    try:
        yield
    finally:
        log.info("Removing swift...")
        testdir = teuthology.get_testdir(ctx)
        for client in config:
            ctx.cluster.only(client).run(args=["rm", "-rf", "{tdir}/swift".format(tdir=testdir)])

def rgwadmin(ctx, client, cmd):
    log.info('radosgw-admin: %s' % cmd)
    testdir = teuthology.get_testdir(ctx)
    pre = [
        'adjust-ulimits',
        'ceph-coverage',
        '{tdir}/archive/coverage'.format(tdir=testdir),
        'radosgw-admin',
        '--log-to-stderr',
        '--format', 'json',
        ]
    pre.extend(cmd)
    (remote,) = ctx.cluster.only(client).remotes.iterkeys()
    proc = remote.run(
        args=pre,
        check_status=False,
        stdout=StringIO(),
        stderr=StringIO(),
        )
    r = proc.exitstatus
    out = proc.stdout.getvalue()
    j = None
    if not r and out != '':
        try:
            j = json.loads(out)
            log.info(' json result: %s' % j)
        except ValueError:
            j = out
            log.info(' raw result: %s' % j)
    return (r, j)

def cephfs_setup(ctx, config):
    testdir = teuthology.get_testdir(ctx)
    coverage_dir = '{tdir}/archive/coverage'.format(tdir=testdir)

    first_mon = teuthology.get_first_mon(ctx, config)
    (mon_remote,) = ctx.cluster.only(first_mon).remotes.iterkeys()
    mdss = ctx.cluster.only(teuthology.is_type('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 mdss.remotes:
        log.info('Setting up CephFS filesystem...')

        ceph_fs = Filesystem(ctx)
        if not ceph_fs.legacy_configured():
            ceph_fs.create()

        is_active_mds = lambda role: role.startswith('mds.') and not role.endswith('-s') and role.find('-s-') == -1
        all_roles = [item for remote_roles in mdss.remotes.values() for item in remote_roles]
        num_active = len([r for r in all_roles if is_active_mds(r)])
        mon_remote.run(args=[
            'adjust-ulimits',
            'ceph-coverage',
            coverage_dir,
            'ceph',
            'mds', 'set_max_mds', str(num_active)])

    yield

def rgwadmin(ctx, client, cmd, stdin=StringIO(), check_status=False):
    log.info("rgwadmin: {client} : {cmd}".format(client=client, cmd=cmd))
    testdir = teuthology.get_testdir(ctx)
    pre = [
        "adjust-ulimits",
        "ceph-coverage".format(tdir=testdir),
        "{tdir}/archive/coverage".format(tdir=testdir),
        "radosgw-admin".format(tdir=testdir),
        "--log-to-stderr",
        "--format",
        "json",
        "-n",
        client,
    ]
    pre.extend(cmd)
    log.info("rgwadmin: cmd=%s" % pre)
    (remote,) = ctx.cluster.only(client).remotes.iterkeys()
    proc = remote.run(args=pre, check_status=check_status, stdout=StringIO(), stderr=StringIO(), stdin=stdin)
    r = proc.exitstatus
    out = proc.stdout.getvalue()
    j = None
    if not r and out != "":
        try:
            j = json.loads(out)
            log.info(" json result: %s" % j)
        except ValueError:
            j = out
            log.info(" raw result: %s" % j)
    return (r, j)

def run_tests(ctx, config, run_stages):
    """
    Run the ragweed after everything is set up.

    :param ctx: Context passed to task
    :param config: specific configuration information
    """
    assert isinstance(config, dict)
    testdir = teuthology.get_testdir(ctx)
    attrs = ["!fails_on_rgw"]
    for client, client_config in config.iteritems():
        stages = string.join(run_stages[client], ',')
        args = [
            'RAGWEED_CONF={tdir}/archive/ragweed.{client}.conf'.format(tdir=testdir, client=client),
            'RAGWEED_STAGES={stages}'.format(stages=stages),
            'BOTO_CONFIG={tdir}/boto.cfg'.format(tdir=testdir),
            '{tdir}/ragweed/virtualenv/bin/nosetests'.format(tdir=testdir),
            '-w',
            '{tdir}/ragweed'.format(tdir=testdir),
            '-v',
            '-a', ','.join(attrs),
            ]
        if client_config is not None and 'extra_args' in client_config:
            args.extend(client_config['extra_args'])

        ctx.cluster.only(client).run(
            args=args,
            label="ragweed tests against rgw"
            )
    yield

def _run_tests(ctx, role):
    """
    For each role, check to make sure it's a client, then run the cram on that client

    :param ctx: Context
    :param role: Roles
    """
    assert isinstance(role, basestring)
    PREFIX = 'client.'
    assert role.startswith(PREFIX)
    id_ = role[len(PREFIX):]
    (remote,) = ctx.cluster.only(role).remotes.iterkeys()
    ceph_ref = ctx.summary.get('ceph-sha1', 'master')

    testdir = teuthology.get_testdir(ctx)
    log.info('Running tests for %s...', role)
    remote.run(
        args=[
            run.Raw('CEPH_REF={ref}'.format(ref=ceph_ref)),
            run.Raw('CEPH_ID="{id}"'.format(id=id_)),
            'adjust-ulimits',
            'ceph-coverage',
            '{tdir}/archive/coverage'.format(tdir=testdir),
            '{tdir}/virtualenv/bin/cram'.format(tdir=testdir),
            '-v', '--',
            run.Raw('{tdir}/archive/cram.{role}/*.t'.format(tdir=testdir, role=role)),
            ],
        logger=log.getChild(role),
        )

def _socket_command(ctx, remote, socket_path, command, args):
    """
    Run an admin socket command and return the result as a string.
    """
    json_fp = StringIO()
    testdir = teuthology.get_testdir(ctx)
    max_tries = 60
    while True:
        proc = remote.run(
            args=[
                'sudo',
                '{tdir}/adjust-ulimits'.format(tdir=testdir),
                'ceph-coverage',
                '{tdir}/archive/coverage'.format(tdir=testdir),
                'ceph',
                '--admin-daemon', socket_path,
                command,
                ] + args,
            stdout=json_fp,
            )
        if proc.exitstatus == 0:
            break
        assert max_tries > 0
        max_tries -= 1
        log.info('ceph cli returned an error, command not registered yet?  sleeping and retrying ...')
        time.sleep(1)
    out = json_fp.getvalue()
    json_fp.close()
    log.debug('admin socket command %s returned %s', command, out)
    return json.loads(out)

 def generate_cbt_config(self):
     mon_hosts = self.hosts_of_type('mon')
     osd_hosts = self.hosts_of_type('osd')
     client_hosts = self.hosts_of_type('client')
     cluster_config = dict(
         user=self.config.get('cluster', {}).get('user', 'ubuntu'),
         head=mon_hosts[0],
         osds=osd_hosts,
         mons=mon_hosts,
         clients=client_hosts,
         osds_per_node=self.config.get('cluster', {}).get('osds_per_node', 1),
         rebuild_every_test=False,
         use_existing=True,
         iterations=self.config.get('cluster', {}).get('iterations', 1),
         tmp_dir='/tmp/cbt',
         pool_profiles=self.config.get('cluster', {}).get('pool_profiles'),
         )
     benchmark_config = self.config.get('benchmarks')
     benchmark_type = benchmark_config.keys()[0]
     if benchmark_type == 'librbdfio':
       testdir = misc.get_testdir(self.ctx)
       benchmark_config['librbdfio']['cmd_path'] = os.path.join(testdir, 'fio/fio')
     return dict(
         cluster=cluster_config,
         benchmarks=benchmark_config,
         )

def download_ceph_deploy(ctx, config):
    """
    Downloads ceph-deploy from the ceph.com git mirror and (by default)
    switches to the master branch. If the `ceph-deploy-branch` is specified, it
    will use that instead.
    """
    log.info("Downloading ceph-deploy...")
    testdir = teuthology.get_testdir(ctx)
    ceph_admin = teuthology.get_first_mon(ctx, config)
    default_cd_branch = {"ceph-deploy-branch": "master"}
    ceph_deploy_branch = config.get("ceph-deploy", default_cd_branch).get("ceph-deploy-branch")

    ctx.cluster.only(ceph_admin).run(
        args=[
            "git",
            "clone",
            "-b",
            ceph_deploy_branch,
            teuth_config.ceph_git_base_url + "ceph-deploy.git",
            "{tdir}/ceph-deploy".format(tdir=testdir),
        ]
    )
    ctx.cluster.only(ceph_admin).run(
        args=["cd", "{tdir}/ceph-deploy".format(tdir=testdir), run.Raw("&&"), "./bootstrap"]
    )

    try:
        yield
    finally:
        log.info("Removing ceph-deploy ...")
        ctx.cluster.only(ceph_admin).run(args=["rm", "-rf", "{tdir}/ceph-deploy".format(tdir=testdir)])

def task(ctx, config):
    """
    Run an autotest test on the ceph cluster.

    Only autotest client tests are supported.

    The config is a mapping from role name to list of tests to run on
    that client.

    For example::

        tasks:
        - ceph:
        - ceph-fuse: [client.0, client.1]
        - autotest:
            client.0: [dbench]
            client.1: [bonnie]

    You can also specify a list of tests to run on all clients::

        tasks:
        - ceph:
        - ceph-fuse:
        - autotest:
            all: [dbench]
    """
    assert isinstance(config, dict)
    config = teuthology.replace_all_with_clients(ctx.cluster, config)
    log.info('Setting up autotest...')
    testdir = teuthology.get_testdir(ctx)
    with parallel() as p:
        for role in config.iterkeys():
            (remote,) = ctx.cluster.only(role).remotes.keys()
            p.spawn(_download, testdir, remote)

    log.info('Making a separate scratch dir for every client...')
    for role in config.iterkeys():
        assert isinstance(role, basestring)
        PREFIX = 'client.'
        assert role.startswith(PREFIX)
        id_ = role[len(PREFIX):]
        (remote,) = ctx.cluster.only(role).remotes.iterkeys()
        mnt = os.path.join(testdir, 'mnt.{id}'.format(id=id_))
        scratch = os.path.join(mnt, 'client.{id}'.format(id=id_))
        remote.run(
            args=[
                'sudo',
                'install',
                '-d',
                '-m', '0755',
                '--owner={user}'.format(user='ubuntu'), #TODO
                '--',
                scratch,
                ],
            )

    with parallel() as p:
        for role, tests in config.iteritems():
            (remote,) = ctx.cluster.only(role).remotes.keys()
            p.spawn(_run_tests, testdir, remote, role, tests)

 def osd_admin_socket(self, osdnum, command, check_status=True):
     """
     Remotely start up ceph specifying the admin socket
     """
     testdir = teuthology.get_testdir(self.ctx)
     remote = None
     for _remote, roles_for_host in self.ctx.cluster.remotes.iteritems():
         for id_ in teuthology.roles_of_type(roles_for_host, 'osd'):
             if int(id_) == int(osdnum):
                 remote = _remote
     assert remote is not None
     args = [
         'sudo',
         'adjust-ulimits',
         'ceph-coverage',
         '{tdir}/archive/coverage'.format(tdir=testdir),
         'ceph',
         '--admin-daemon',
         '/var/run/ceph/ceph-osd.{id}.asok'.format(id=osdnum),
         ]
     args.extend(command)
     return remote.run(
         args=args,
         stdout=StringIO(),
         wait=True,
         check_status=check_status
         )

def create_dirs(ctx, config):
    """
    Handle directory creation and cleanup
    """
    testdir = teuthology.get_testdir(ctx)
    for client, client_config in config.iteritems():
        assert 'test' in client_config, 'You must specify a test to run'
        (remote,) = ctx.cluster.only(client).remotes.keys()
        remote.run(
            args=[
                'install', '-d', '-m0755', '--',
                '{tdir}/qemu'.format(tdir=testdir),
                '{tdir}/archive/qemu'.format(tdir=testdir),
                ]
            )
    try:
        yield
    finally:
        for client, client_config in config.iteritems():
            assert 'test' in client_config, 'You must specify a test to run'
            (remote,) = ctx.cluster.only(client).remotes.keys()
            remote.run(
                args=[
                    'rmdir', '{tdir}/qemu'.format(tdir=testdir), run.Raw('||'), 'true',
                    ]
                )

def download_image(ctx, config):
    """Downland base image, remove image file when done"""
    log.info('downloading base image')
    testdir = teuthology.get_testdir(ctx)
    for client, client_config in config.iteritems():
        (remote,) = ctx.cluster.only(client).remotes.keys()
        base_file = '{tdir}/qemu/base.{client}.qcow2'.format(tdir=testdir, client=client)
        remote.run(
            args=[
                'wget', '-nv', '-O', base_file, DEFAULT_IMAGE_URL,
                ]
            )
    try:
        yield
    finally:
        log.debug('cleaning up base image files')
        for client in config.iterkeys():
            base_file = '{tdir}/qemu/base.{client}.qcow2'.format(
                tdir=testdir,
                client=client,
                )
            (remote,) = ctx.cluster.only(client).remotes.keys()
            remote.run(
                args=[
                    'rm', '-f', base_file,
                    ],
                )