measure-idle-cpu.py

import atexit
import statistics
import sys
import time
from subprocess import Popen

import psutil


def jupyterhub(interval):
    """Start jupyterhub to measure idle cpu"""
    cmd = [
        sys.executable,
        "-m",
        "jupyterhub",
        "--log-level=ERROR",
        "--ConfigurableHTTPProxy.log_level=error",
        "--db=sqlite:///:memory:",
    ]
    if interval:
        cmd.append(
            f"--PeriodicMetricsCollector.event_loop_interval_resolution={interval}"
        )
    else:
        cmd.append("--PeriodicMetricsCollector.event_loop_interval_enabled=False")
    p = Popen(cmd)
    atexit.register(lambda *args: p.terminate)
    return p


def measure(interval):
    """Complete one measurement"""
    samples = []
    with jupyterhub(interval) as hub:
        # wait for startup to complete
        time.sleep(2)
        # print(f"Measuring {hub.pid}", file=sys.stderr)
        p = psutil.Process(hub.pid)
        start_time = time.perf_counter()
        cpu_times = p.cpu_times()
        start_cpu = cpu_times.system + cpu_times.user
        # measure for a while
        for i in range(100):
            samples.append(p.cpu_percent())
            time.sleep(0.1)
        cpu_times = p.cpu_times()
        end_time = time.perf_counter()
        end_cpu = cpu_times.system + cpu_times.user

        hub.terminate()
    avg_cpu = 100 * (end_cpu - start_cpu) / (end_time - start_time)
    deciles = statistics.quantiles(samples, n=10)
    quartiles = statistics.quantiles(samples, n=4)
    return (interval, avg_cpu, deciles[1], quartiles[1])


def collect_measurements(intervals):
    measurements = []
    print("interval\tavg_cpu\t90%\t75%")
    for interval in intervals:
        measurement = measure(interval)
        print(measurement[0], end="")
        for n in measurement[1:]:
            print(f"\t{n:.1f}%", end="")
        print()
        measurements.append(measurement)


if __name__ == "__main__":
    collect_measurements([0.02, 0.05, 0.1, 0.2, 0.5])