index policy metrics

def index_metrics(
        config, start, end, incremental=False, concurrency=5, accounts=None,
        period=3600, tag=None, index='policy-metrics', verbose=False):
    """index policy metrics"""
    logging.basicConfig(level=(verbose and logging.DEBUG or logging.INFO))
    logging.getLogger('botocore').setLevel(logging.WARNING)
    logging.getLogger('elasticsearch').setLevel(logging.WARNING)
    logging.getLogger('urllib3').setLevel(logging.WARNING)
    logging.getLogger('requests').setLevel(logging.WARNING)
    logging.getLogger('c7n.worker').setLevel(logging.INFO)

    with open(config) as fh:
        config = yaml.safe_load(fh.read())
    jsonschema.validate(config, CONFIG_SCHEMA)

    start, end = get_date_range(start, end)

    p_accounts = set()
    p_account_stats = {}
    i_time = i_points = 0
    t = time.time()

    with ProcessPoolExecutor(max_workers=concurrency) as w:
        futures = {}
        jobs = []
        # Filter
        for account in config.get('accounts'):
            if accounts and account['name'] not in accounts:
                continue
            if tag:
                found = False
                for t in account['tags'].values():
                    if tag == t:
                        found = True
                        break
                if not found:
                    continue
            p_accounts.add((account['name']))
            for region in account.get('regions'):
                for (p_start, p_end) in get_periods(start, end, period):
                    p = (config, index, region, account, p_start, p_end, period)
                    jobs.append(p)

        # by default we'll be effectively processing in order, but thats bumps
        # our concurrency into rate limits on metrics retrieval in a given account
        # region, go ahead and shuffle, at least with lucene, the non ordering
        # should have minimal impact on query perf (inverted index).

        random.shuffle(jobs)

        for j in jobs:
            log.debug("submit account:%s region:%s start:%s end:%s" % (
                j[3]['name'], j[2], j[4], j[5]))
            futures[w.submit(index_account_metrics, *j)] = j

        # Process completed
        for f in as_completed(futures):
            config, index, region, account, p_start, p_end, period = futures[f]
            if f.exception():
                log.warning("error account:%s region:%s error:%s",
                            account['name'], region, f.exception())
                continue
            rtime, rpoints = f.result()
            rstat = p_account_stats.setdefault(
                account['name'], {}).setdefault(region, {'points': 0})
            rstat['points'] += rpoints

            # log.info("complete account:%s, region:%s points:%s time:%0.2f",
            #         account['name'], region, rpoints, rtime)

            i_time += rtime
            i_points += rpoints

        log.info("complete accounts:%d points:%d itime:%0.2f time:%0.2f",
                 len(p_accounts), i_points, i_time, time.time() - t)