def wait_for_compaction(nodes=None, check_interval=30, idle_confirmations=3,
                        compaction_throughput=16, allowed_connection_errors=10):

    def compactionstats(nodes, check_interval):
        """Check for compactions via nodetool compactionstats"""
        consecutive_connection_errors = 0
        pattern = re.compile("(^|\n)pending tasks: 0")
        failure_pattern = re.compile("ConnectException")
        nodes = set(nodes)
        while True:
            results = execute(common.multi_nodetool, cmd="compactionstats")
            for node, output in results.iteritems():
                if pattern.search(output.strip()):
                    nodes.discard(node)
                elif failure_pattern.search(output.strip()):
                    consecutive_connection_errors += 1

            if consecutive_connection_errors > allowed_connection_errors:
                raise NodetoolException(
                    "Failed to connect via nodetool {consecutive_connection_errors} times in a row.".format(
                        consecutive_connection_errors=consecutive_connection_errors))

            if len(nodes) == 0:
                break
            logger.info("Waiting for compactions (compactionstats) on nodes:")
            for node in nodes:
                logger.info("{node} - {output}".format(node=node, output=results[node]))
            time.sleep(check_interval)

        assert len(nodes) == 0, ("Compactions (compactionstats) should have finished, but they didn't"
                            " on nodes: {nodes}. output: {results}".format(
                                nodes=nodes, output=results))

    def tpstats(nodes, check_interval):
        """Check for compactions via nodetool tpstats"""
        consecutive_connection_errors = 0
        stat_exists_pattern = re.compile("^CompactionExecutor", re.MULTILINE)
        no_compactions_pattern = re.compile("CompactionExecutor\W*0\W*0\W*[0-9]*\W*0", re.MULTILINE)
        failure_pattern = re.compile("ConnectException")

        nodes = set(nodes)
        while True:
            results = execute(common.multi_nodetool, cmd="tpstats")
            for node, output in results.iteritems():
                if stat_exists_pattern.search(output):
                    if no_compactions_pattern.search(output):
                        nodes.discard(node)
                elif failure_pattern.search(output.strip()):
                    consecutive_connection_errors += 1
                else:
                    logger.warn("CompactionExecutor not listed in nodetool tpstats, can't check for compactions this way.")
                    return

            if consecutive_connection_errors > allowed_connection_errors:
                raise NodetoolException(
                    "Failed to connect via nodetool {consecutive_connection_errors} times in a row.".format(
                        consecutive_connection_errors=consecutive_connection_errors))

            if len(nodes) == 0:
                break
            logger.info("Waiting for compactions (tpstats) on nodes: {nodes}".format(nodes=nodes))
            time.sleep(check_interval)

        assert len(nodes) == 0, ("Compactions (tpstats) should have finished, but they didn't"
                            " on nodes: {nodes}. output: {results}".format(
                                nodes=nodes, output=results))

    if nodes is None:
        nodes = set(common.fab.env.hosts)
    else:
        nodes = set(nodes)

    # Disable compaction throttling to speed things up:
    execute(common.multi_nodetool, cmd="setcompactionthroughput 0")

    # Perform checks multiple times to ensure compactions are really done:
    start = time.time()
    for i in range(idle_confirmations):
        compactionstats(nodes, check_interval)
        tpstats(nodes, check_interval)

    duration = time.time() - start

    # Re-enable compaction throttling:
    execute(common.multi_nodetool, cmd='setcompactionthroughput {compaction_throughput}'.format(**locals()))

    logger.info("Compactions finished on all nodes. Duration of checks: {duration}".format(**locals()))

    return duration
Macros should be useable with no arguments, and just a macro
dataset/ETHPy150Open datastax/cstar_perf/tool/cstar_perf/tool/benchmark.py/wait_for_compaction--crossover with--dataset/ETHPy150Open nalourie/django-macros/build/lib/macros/tests.py/MacrosTests.test_defining_macro_with_no_args