1#!/usr/bin/env python32"""Observe a Git benchmark and its Docker cgroups without changing its exit code."""34import argparse5import datetime as dt6import json7import os8import pathlib9import signal10import subprocess11import sys12import threading13import time14import urllib.parse15import urllib.request161718def utc():19 return dt.datetime.now(dt.timezone.utc).isoformat()202122def write_json(path, value):23 path.write_text(json.dumps(value, sort_keys=True) + "\n", encoding="utf-8")242526def http_capture(base, token, path, output):27 if not base:28 return {"path": path, "status": "disabled"}29 url = base.rstrip("/") + path30 request = urllib.request.Request(url, headers={"Authorization": "Bearer " + token})31 try:32 with urllib.request.urlopen(request, timeout=3) as response:33 data = response.read()34 output.write_bytes(data)35 return {"path": path, "status": response.status, "bytes": len(data)}36 except Exception as exc: # diagnostics are supplementary to the benchmark37 return {"path": path, "status": "error", "error": str(exc)}383940def container_pid(name):41 try:42 result = subprocess.run(43 ["docker", "inspect", "--format", "{{.State.Pid}}", name],44 check=True, capture_output=True, text=True,45 )46 pid = int(result.stdout.strip())47 return pid if pid > 0 else None48 except (OSError, ValueError, subprocess.CalledProcessError):49 return None505152def cgroup_root(pid):53 try:54 for line in pathlib.Path(f"/proc/{pid}/cgroup").read_text().splitlines():55 hierarchy, _, path = line.partition("::")56 if hierarchy == "0":57 root = pathlib.Path("/sys/fs/cgroup") / path.lstrip("/")58 return root59 except OSError:60 pass61 return None626364def read_number(path):65 try:66 return int(path.read_text().strip())67 except (OSError, ValueError):68 return None697071def read_text(path):72 try:73 return path.read_text()74 except OSError:75 return None767778def sample_container(name):79 pid = container_pid(name)80 sample = {"time": utc(), "container": name, "pid": pid}81 if not pid:82 sample["error"] = "container PID unavailable"83 return sample84 root = cgroup_root(pid)85 if root is None:86 sample["error"] = "cgroup v2 path unavailable"87 return sample88 sample["cgroup"] = str(root)89 sample["cpu.stat"] = read_text(root / "cpu.stat")90 sample["memory.current"] = read_number(root / "memory.current")91 sample["memory.peak"] = read_number(root / "memory.peak")92 sample["memory.stat"] = read_text(root / "memory.stat")93 sample["io.stat"] = read_text(root / "io.stat")94 sample["pids.current"] = read_number(root / "pids.current")95 sample["process.status"] = read_text(pathlib.Path(f"/proc/{pid}/status"))96 sample["memory.events"] = read_text(root / "memory.events")97 return sample9899100def host_sample():101 return {102 "time": utc(),103 "meminfo": read_text(pathlib.Path("/proc/meminfo")),104 "loadavg": read_text(pathlib.Path("/proc/loadavg")),105 }106107108def capture_cpu(base, token, output, result):109 if not base:110 return111 request = urllib.request.Request(112 base.rstrip("/") + "/debug/pprof/profile?seconds=60",113 headers={"Authorization": "Bearer " + token},114 )115 try:116 with urllib.request.urlopen(request, timeout=65) as response:117 output.write_bytes(response.read())118 result["status"] = response.status119 except Exception as exc:120 result["status"] = "error"121 result["error"] = str(exc)122123124def main():125 parser = argparse.ArgumentParser(description=__doc__)126 parser.add_argument("--out", required=True, type=pathlib.Path)127 parser.add_argument("--container", action="append", default=[])128 parser.add_argument("--diagnostics", default="")129 parser.add_argument("--token", default="")130 parser.add_argument("--cpu-profile", action="store_true")131 parser.add_argument("command", nargs=argparse.REMAINDER)132 args = parser.parse_args()133 command = args.command[1:] if args.command and args.command[0] == "--" else args.command134 if not command:135 parser.error("a benchmark command is required")136 args.out.mkdir(parents=True, exist_ok=True)137 log_path = args.out / "harness.log"138 log = log_path.open("w", encoding="utf-8")139140 def log_line(text):141 line = f"[{utc()}] {text}\n"142 log.write(line)143 log.flush()144 sys.stdout.write(line)145 sys.stdout.flush()146147 write_json(args.out / "metadata.json", {148 "started": utc(), "command": command, "containers": args.container,149 "diagnostics": args.diagnostics or None, "cpu_profile": args.cpu_profile,150 "python": sys.version, "host": dict(zip(("sysname", "nodename", "release", "version", "machine"), os.uname())),151 })152 before = args.out / "diagnostics-before"153 after = args.out / "diagnostics-after"154 before.mkdir(exist_ok=True)155 after.mkdir(exist_ok=True)156 diag_paths = ["/metrics", "/debug/pprof/heap", "/debug/pprof/goroutine?debug=2"]157 before_results = [http_capture(args.diagnostics, args.token, path, before / path.rsplit("/", 1)[-1].split("?", 1)[0]) for path in diag_paths]158 write_json(args.out / "diagnostics-before.json", before_results)159 samples = (args.out / "samples.jsonl").open("w", encoding="utf-8")160 stop = threading.Event()161162 def monitor():163 while not stop.is_set():164 record = {"time": utc(), "host": host_sample(), "containers": [sample_container(name) for name in args.container]}165 samples.write(json.dumps(record, sort_keys=True) + "\n")166 samples.flush()167 stop.wait(0.5)168169 monitor_thread = threading.Thread(target=monitor, name="cgroup-monitor")170 monitor_thread.start()171 profile_result = {}172 profile_thread = None173 if args.cpu_profile and args.diagnostics:174 profile_thread = threading.Thread(target=capture_cpu, args=(args.diagnostics, args.token, args.out / "cpu.pprof", profile_result), name="cpu-profile")175 profile_thread.start()176 log_line("cpu profile started (bounded to 60 seconds)")177 log_line("phase benchmark-start")178 child = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1)179 try:180 assert child.stdout is not None181 for line in child.stdout:182 log_line(line.rstrip("\n"))183 exit_code = child.wait()184 except BaseException:185 child.send_signal(signal.SIGTERM)186 child.wait(timeout=10)187 raise188 finally:189 stop.set()190 monitor_thread.join(timeout=3)191 samples.close()192 log_line(f"phase benchmark-end exit={exit_code}")193 after_results = [http_capture(args.diagnostics, args.token, path, after / path.rsplit("/", 1)[-1].split("?", 1)[0]) for path in diag_paths]194 write_json(args.out / "diagnostics-after.json", after_results)195 if profile_thread is not None:196 profile_thread.join(timeout=67)197 if profile_thread.is_alive():198 profile_result.update({"status": "timeout"})199 write_json(args.out / "cpu-profile.json", profile_result)200 write_json(args.out / "result.json", {"finished": utc(), "exit_code": exit_code, "profile": profile_result})201 log.close()202 return exit_code203204205if __name__ == "__main__":206 raise SystemExit(main())