Circle CI Volume Stats

Circle CI Volume Stats

from collections import defaultdict
from datetime import datetime as dt
import os

from circleci.api import Api
import requests


def get_builds(cci):
    for offset in range(100):
        for build in cci.get_project_build_summary('toyota-connected', 'va-lexi', offset=30*offset):
            yield build['build_num'], build['status']


def get_failed_step(build_info):
    for step in build_info['steps']:
        if not step['actions'][0]['failed']:
            continue
        output_url = step['actions'][0]['output_url']
        break
    else:
        print(f"Expected to find a failed step for {build_info['build_num']}, but didn't")
        return ''

    resp = requests.get(output_url)
    resp.raise_for_status()
    return resp.text


def get_daemon_failures(volume_stats):
    daemon_failures = list()
    for volume_id in volume_stats:
        if volume_stats[volume_id]['daemon_failure']:
            daemon_failures.append(volume_id)

    return daemon_failures


def main():
    begin = dt.now()
    cci = Api(os.environ['CIRCLE_TOKEN'])
    volume_stats = defaultdict(lambda: defaultdict(list))

    for index, (build_num, build_status) in enumerate(get_builds(cci), 1):

        build_info = cci.get_build_info('toyota-connected', 'va-lexi', build_num)
        try:
            step_1_url = build_info['steps'][0]['actions'][0]['output_url']
        except IndexError:
            print(f"Something wrong with getting build info: {build_num}, skipping\n")
            continue
        resp = requests.get(step_1_url)
        resp.raise_for_status()

        if 'circleci/classic:latest' not in resp.text:
            continue

        spin_up_status = resp.json()[0]['message'].split('\n')

        try:
            volume_id = [s for s in spin_up_status if s.startswith('Using volume: ')][0].split(': ')[-1]
        except IndexError:
            print(f"Something wrong with getting volume id: {build_num}, skipping\n")
            continue
        volume_stats[volume_id][build_status].append(build_num)

        if build_info['failed']:
            failed_step_info = get_failed_step(build_info)
            if 'Is the docker daemon running?' in failed_step_info:
                volume_stats[volume_id]['daemon_failure'].append(build_num)
                daemon_failures = get_daemon_failures(volume_stats)
                print(f"Found one for volume {volume_id} on build {build_num}\n"
                      f"Current Stats for {volume_id}:\n"
                      f"Succes: {len(volume_stats[volume_id]['success'])}\n"
                      f"Failed: {len(volume_stats[volume_id]['failed'])}\n"
                      f"Failed daemon: {len(volume_stats[volume_id]['daemon_failure'])}\n"
                      f"Total builds evaluated: {index}\n"
                      f"List of volumes with daemon failure: {daemon_failures}\n"
                      f"Elapsed time: {dt.now() - begin}\n")
            else:
                volume_stats[volume_id]['no_daemon_failure'].append(build_num)

    from pprint import pprint as pp
    import pdb
    pdb.set_trace()
    pp('')


if __name__ == "__main__":
    main()