diff --git a/scripts/baremetal-runner-report.py b/scripts/baremetal-runner-report.py index 481f452ed..0255914e3 100644 --- a/scripts/baremetal-runner-report.py +++ b/scripts/baremetal-runner-report.py @@ -11,30 +11,16 @@ import argparse import json +import os import subprocess import sys from collections import defaultdict +from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta, timezone from typing import Iterable BAREMETAL_LABEL_PREFIX = "oracle-bare-metal" -# (repo, workflow file path) — discovered via: -# gh search code 'oracle-bare-metal path:.github/workflows' --owner open-telemetry -WORKFLOWS = [ - ("open-telemetry/opentelemetry-go", "benchmark.yml"), - ("open-telemetry/opentelemetry-java", "benchmark-tags.yml"), - ("open-telemetry/opentelemetry-java", "benchmark.yml"), - ("open-telemetry/opentelemetry-python", "benchmarks.yml"), - ("open-telemetry/opentelemetry-rust", "benchmark.yml"), - ("open-telemetry/opentelemetry-js", "benchmark.yml"), - ("open-telemetry/opentelemetry-collector-contrib", "load-tests.yml"), - ("open-telemetry/otel-arrow", "pipeline-perf-on-label.yaml"), - ("open-telemetry/otel-arrow", "pipeline-perf-test-manual-pr.yaml"), - ("open-telemetry/otel-arrow", "pipeline-perf-test-continuous.yml"), - ("open-telemetry/otel-arrow", "pipeline-perf-test-nightly.yml"), -] - def gh_paginated(path: str, list_key: str) -> Iterable[dict]: proc = subprocess.run( @@ -42,6 +28,7 @@ def gh_paginated(path: str, list_key: str) -> Iterable[dict]: "-H", "Accept: application/vnd.github+json", "--jq", f".{list_key}[]", path], capture_output=True, text=True, encoding="utf-8", errors="replace", + timeout=120, ) if proc.returncode != 0: raise RuntimeError(f"gh api {path}: {proc.stderr}") @@ -51,6 +38,34 @@ def gh_paginated(path: str, list_key: str) -> Iterable[dict]: yield json.loads(line) +def discover_workflows(owner: str) -> list[tuple[str, str]]: + proc = subprocess.run( + [ + "gh", "search", "code", + f"{BAREMETAL_LABEL_PREFIX} path:.github/workflows", + "--language", "YAML", + "--owner", owner, + "--json", "repository,path", + "--limit", "1000", + ], + capture_output=True, text=True, encoding="utf-8", errors="replace", + timeout=120, + ) + if proc.returncode != 0: + raise RuntimeError(f"gh search code: {proc.stderr}") + workflows = set() + for result in json.loads(proc.stdout or "[]"): + repo = result.get("repository", {}).get("nameWithOwner") + path = result.get("path") + if not repo or not path: + continue + dirname, filename = os.path.split(path) + if dirname != ".github/workflows" or not filename.endswith((".yml", ".yaml")): + continue + workflows.add((repo, filename)) + return sorted(workflows) + + def parse_iso(ts: str) -> datetime: return datetime.fromisoformat(ts.replace("Z", "+00:00")) @@ -66,11 +81,72 @@ def fmt_dur(seconds: float) -> str: return f"{s}s" -def collect_jobs(since: datetime) -> list[dict]: +def workflow_url(repo: str, wf_file: str) -> str: + return f"https://github.com/{repo}/blob/main/.github/workflows/{wf_file}" + + +def collect_run_jobs(repo: str, wf_file: str, run: dict) -> tuple[list[dict], str | None]: + records: list[dict] = [] + run_id = run["id"] + wf_name = run.get("name") or wf_file + run_started_at = run.get("run_started_at") or run.get("created_at") + try: + jobs = list(gh_paginated( + f"/repos/{repo}/actions/runs/{run_id}/jobs?per_page=100", + "jobs", + )) + except Exception as e: + return records, f"{repo} {wf_file} run {run_id}: {str(e).strip()}" + for job in jobs: + labels = job.get("labels") or [] + if not any( + isinstance(lbl, str) and lbl.startswith(BAREMETAL_LABEL_PREFIX) + for lbl in labels + ): + continue + if job.get("conclusion") in ("skipped", "neutral"): + continue + if not (job.get("runner_name") and job.get("runner_group_id")): + continue + started, completed = job.get("started_at"), job.get("completed_at") + if not started or not completed: + continue + try: + started_dt = parse_iso(started) + dur = (parse_iso(completed) - started_dt).total_seconds() + except ValueError: + continue + if dur <= 0: + continue + qwait = 0.0 + if run_started_at: + try: + qwait = max( + 0.0, + (started_dt - parse_iso(run_started_at)).total_seconds(), + ) + except ValueError: + qwait = 0.0 + records.append({ + "repo": repo, + "wf_file": wf_file, + "wf_name": wf_name, + "run_id": run_id, + "started": started_dt, + "duration": dur, + "qwait": qwait, + }) + return records, None + + +def collect_jobs( + workflows: list[tuple[str, str]], since: datetime, jobs_concurrency: int, +) -> tuple[list[dict], list[str]]: """Return per-job records for every bare-metal job since `since`.""" since_q = since.strftime("%Y-%m-%dT%H:%M:%SZ") records: list[dict] = [] - for repo, wf_file in WORKFLOWS: + errors: list[str] = [] + for repo, wf_file in workflows: print(f" - {repo} :: {wf_file}", file=sys.stderr) runs_path = ( f"/repos/{repo}/actions/workflows/{wf_file}/runs" @@ -78,61 +154,26 @@ def collect_jobs(since: datetime) -> list[dict]: ) try: runs = list(gh_paginated(runs_path, "workflow_runs")) - except RuntimeError as e: - print(f" ! {e}", file=sys.stderr) + except Exception as e: + msg = str(e).strip() + errors.append(f"{repo} {wf_file}: {msg}") + print(f" ! {msg}", file=sys.stderr) continue - for run in runs: - run_id = run["id"] - wf_name = run.get("name") or wf_file - run_started_at = run.get("run_started_at") or run.get("created_at") - try: - jobs = list(gh_paginated( - f"/repos/{repo}/actions/runs/{run_id}/jobs?per_page=100", - "jobs", - )) - except RuntimeError as e: - print(f" ! run {run_id}: {e}", file=sys.stderr) - continue - for job in jobs: - labels = job.get("labels") or [] - if not any( - isinstance(lbl, str) and lbl.startswith(BAREMETAL_LABEL_PREFIX) - for lbl in labels - ): - continue - if job.get("conclusion") in ("skipped", "neutral"): - continue - if not (job.get("runner_name") and job.get("runner_group_id")): - continue - started, completed = job.get("started_at"), job.get("completed_at") - if not started or not completed: - continue - try: - dur = (parse_iso(completed) - parse_iso(started)).total_seconds() - except ValueError: - continue - if dur <= 0: - continue - qwait = 0.0 - if run_started_at: - try: - qwait = max( - 0.0, - (parse_iso(started) - parse_iso(run_started_at)) - .total_seconds(), - ) - except ValueError: - qwait = 0.0 - records.append({ - "repo": repo, - "wf_file": wf_file, - "wf_name": wf_name, - "run_id": run_id, - "started": parse_iso(started), - "duration": dur, - "qwait": qwait, - }) - return records + print(f" {len(runs)} workflow runs", file=sys.stderr) + with ThreadPoolExecutor(max_workers=jobs_concurrency) as executor: + futures = [ + executor.submit(collect_run_jobs, repo, wf_file, run) + for run in runs + ] + for index, future in enumerate(as_completed(futures), 1): + run_records, error = future.result() + records.extend(run_records) + if error: + errors.append(error) + print(f" ! {error}", file=sys.stderr) + if index % 50 == 0 or index == len(futures): + print(f" processed {index}/{len(futures)} runs", file=sys.stderr) + return records, errors def render_table(records: list[dict], since: datetime) -> str: @@ -164,9 +205,7 @@ def render_table(records: list[dict], since: datetime) -> str: avg = v["total"] / n_runs if n_runs else 0 max_run = max(v["run_totals"].values(), default=0.0) repo_short = repo.split("/", 1)[1] if "/" in repo else repo - wf_link = ( - f"https://github.com/{repo}/blob/main/.github/workflows/{wf_file}" - ) + wf_link = workflow_url(repo, wf_file) out.append( f"| {repo_short} | [`{wf_file}`]({wf_link}) | " f"{n_runs} | {fmt_dur(v['total'])} | {fmt_dur(avg)} | " @@ -175,25 +214,60 @@ def render_table(records: list[dict], since: datetime) -> str: return "\n".join(out) +def render_workflows(workflows: list[tuple[str, str]]) -> str: + lines = ["## Discovered workflow files", ""] + for repo, wf_file in workflows: + repo_short = repo.split("/", 1)[1] if "/" in repo else repo + lines.append(f"- {repo_short}: [`{wf_file}`]({workflow_url(repo, wf_file)})") + return "\n".join(lines) + + +def render_collection_notes(errors: list[str]) -> str: + if not errors: + return "## Collection notes\n\nNo API collection errors." + lines = ["## Collection notes", "", "The report skipped these API calls:"] + for error in errors: + lines.append(f"- {error}") + return "\n".join(lines) + + def main() -> int: p = argparse.ArgumentParser(description=__doc__) p.add_argument( "--windows", type=int, nargs="+", default=[7, 30], help="One or more day windows to emit tables for.", ) + p.add_argument( + "--jobs-concurrency", type=int, default=12, + help="Number of concurrent run job API calls.", + ) + p.add_argument("--owner", default="open-telemetry") p.add_argument("--output", default="baremetal-runner-report.md") args = p.parse_args() + if args.jobs_concurrency < 1: + p.error("--jobs-concurrency must be at least 1") now = datetime.now(timezone.utc) max_days = max(args.windows) since_max = now - timedelta(days=max_days) + workflows = discover_workflows(args.owner) + print(f"Discovered {len(workflows)} bare-metal workflow files", file=sys.stderr) print(f"Fetching jobs since {since_max.isoformat()} " f"(max window {max_days} days)", file=sys.stderr) - records = collect_jobs(since_max) + records, errors = collect_jobs(workflows, since_max, args.jobs_concurrency) print(f"Collected {len(records)} bare-metal job records", file=sys.stderr) - sections: list[str] = [] + sections: list[str] = [ + "# Bare-metal runner report", + "", + f"Generated: {now.isoformat()}", + "", + render_workflows(workflows), + "", + render_collection_notes(errors), + "", + ] for days in sorted(args.windows): since = now - timedelta(days=days) sections.append(f"## Last {days} days")