jenkins-chef-ci-profiler.py

#!/usr/bin/env python3

from collections import defaultdict
from datetime import datetime, timedelta
import requests
from bisect import bisect_left

from typing import List, Dict, Any
from bisect import bisect

USERNAME = ""
PASSWORD = ""


# Recipe from https://docs.python.org/3/library/bisect.html
def find_le(a, x):
    "Find rightmost value less than or equal to x"
    i = bisect(a, x)
    if i:
        return a[i - 1]
    raise ValueError(f"Value {x} is below the lowest threshold.")


def fetch_log(url) -> List[str]:
    res = requests.get(url, auth=(USERNAME, PASSWORD))
    if res.status_code != 200:
        raise IOError("Something went wrong")

    # This should probably be returned instead.
    lines = list(res.text.splitlines())
    if len(lines) <= 1:
        raise ValueError("Insufficient lines")
    return lines


def get_datetime(line: str) -> datetime:
    return datetime.fromisoformat(line[1:20])


bucket_thresholds = [2, 4, 8, 16, 32, 64, 128, 256, 512]


def bucketize(lines: List[str]) -> Dict[int, List[str]]:
    buckets: Dict[Any, List[Any]] = defaultdict(list)
    times = [get_datetime(line) for line in lines]
    for index, (before_time, after_time, line) in enumerate(
        zip(times, times[1:], lines)
    ):
        time_taken = (after_time - before_time).total_seconds()
        try:
            bucket = find_le(bucket_thresholds, time_taken)
        except ValueError:
            # We don't care about the smallest bucket,
            # ie anything under the threshold of 2 seconds.
            continue
        else:
            buckets[bucket].append(line)
    return buckets


def reportize(buckets: Dict[int, List[str]]):
    return "".join(
        f"{threshold} => {len(self.bucket)}\n" for threshold, bucket in buckets.items()
    )


def run_JenkinsChefProfiler(repository, pr, build):
    url = f"https://kube-jenkins.inday.io/chef/job/cookbooks/job/{repository}/job/{pr}/{build}/consoleText"

    lines = fetch_log(url)
    buckets = bucketize(lines)
    print(reportize(buckets))


def main():
    run_JenkinsChefProfiler("oms", "PR-18", 3)


if __name__ == "__main__":
    main()