Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 29 additions & 18 deletions .github/actions/arc-job-report/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,10 @@ Expanded, it has:
[ARC Job Resources dashboard](https://montaapp.grafana.net/d/arc-job-resources).
Sharded jobs (`<name> - shard N`) are merged into one row with the time range
and a link per shard number; the slowest shard is bold.
- **Suggested runner changes:** only jobs where another catalog size would fit,
marked ⬇️ (smaller) or ⬆️ (bigger), with peak memory and CPU; the job name
links to its dashboard. Left out entirely when every runner fits.
- **Suggested runner changes:** only jobs where another catalog size is
suggested, marked ⬇️ (smaller) or ⬆️ (bigger) with the reason, the job's peak
memory, median and peak CPU and number of runs over the last 7 days; the job
name links to its dashboard. Left out entirely when every runner is right.
- **Run history:** one row per run (commit, ✅/❌, wall time, change vs the
previous run, slowest job), so speed changes can be traced to commits. The
history is stored as JSON in a hidden HTML comment and re-rendered on every
Expand Down Expand Up @@ -69,25 +70,35 @@ an ARC runner (`arc-*` / `self-hosted-*`) are skipped.

Metrics come from cAdvisor and kube-state-metrics in `mimir-internal`, scraped every **15 s**.

- **Peak memory is reliable.** It is the highest working-set sample over the job.
- **CPU is not, for short jobs.** It is the highest 1 min average, so short bursts
are averaged away. In service-ocpi, the CPU-based rule suggested 4 vCPU for the
integration shards, but a trial run made every shard 17–97 s slower. Trial a
smaller runner before moving to it.
- **Jobs under 2 minutes are left out of the suggestions.** They get too few
samples. There is no point waiting for them: a runner pod's samples stop once
its job ends and the pod is deleted.

The suggested size is the smallest catalog runner with memory ≥ 1.3 × peak and
vCPU ≥ 1.2 × the CPU p95.
Suggestions use the job's runs over the **last 7 days** (same repository and job
name), not just this run: cache hits make one run's peak misleading. Typecheck in
monorepo-typescript peaks at 1.5 GB in a typical run and 13.1 GB in the worst. The
history comes from two recording rules in monta-app/terraform
(`global/grafana/teams/sre/alerts_arc.tf`): `arc:job_memory_working_set_bytes:max1m`
and `arc:job_cpu_usage_cores:rate1m`. They join each pod's cAdvisor series to its
GitHub job once a minute, so the report never joins raw series at read time.

Memory and CPU are treated differently because the runners have a memory limit
and only a CPU request:

- **Memory is a hard floor.** A size must hold 1.3 × the job's 7-day peak, or the
job is OOM-killed. A job that needs more gets ⬆️ *needs memory*, and so does
any job that was OOM-killed in this run.
- **CPU only costs time.** A job whose median CPU is at least 85% of its runner's
vCPU gets ⬆️ *CPU-bound, faster*. A smaller size must still cover the job's
median CPU; it is *fits* if it also covers the peak, and *trial: may be slower*
if not. In service-ocpi, a 4 vCPU runner made every integration shard 17–97 s
slower, so trial a smaller runner before moving to it.
- **Smaller sizes need 5 runs** in the window, so rare jobs and new recording
rules don't produce a downsize from one run. Bigger sizes don't wait.

## Files

- `arc_job_report.py` — builds and posts the comment.
- `job_resources.py` — the per-job analysis. It is `infra/arc/scripts/job-resources.py`
from monta-app/kube-manifests, plus a `GRAFANA_TOKEN` path that queries Prometheus
through Grafana's datasource proxy instead of `gcx`. Keep the two in sync.
- `test_arc_job_report.py` — unit tests for the comment rendering:
- `job_resources.py` — the per-job analysis and the size suggestion. It queries
Prometheus through Grafana's datasource proxy, with `GRAFANA_TOKEN` in CI or
`gcx api` locally.
- `test_arc_job_report.py` — unit tests for the comment rendering and the suggestion:
`python3 -m unittest discover -s .github/actions/arc-job-report -p 'test_*.py'`

## Running locally
Expand Down
26 changes: 13 additions & 13 deletions .github/actions/arc-job-report/arc_job_report.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,12 +16,13 @@
import sys
from datetime import datetime

from job_resources import ARC_RUNNER, analyse, jobs_for, sh
from job_resources import MEMORY_HEADROOM, MIN_RUNS, WINDOW, analyse_run, sh

MARKER = "<!-- arc-job-report -->"
HISTORY = re.compile(r"<!-- history (\[.*?\]) -->", re.S)
SHARD = re.compile(r"^(.*) - shard (\d+)$")
RESULT = {"success": "✅", "failure": "❌"}
WHY = {"memory": "needs memory", "faster": "CPU-bound, faster", "fits": "fits", "trial": "trial: may be slower"}
# ponytail: keeps the last 50 runs so the comment stays far below GitHub's 65,536-character limit
MAX_HISTORY = 50

Expand Down Expand Up @@ -79,17 +80,15 @@ def job_rows(results):


def change(r):
if r["duration_s"] < 120:
return None
runner, suggested = size(r["runner"]), size(r.get("suggested_runner") or r["runner"])
return None if suggested == runner else (runner, suggested)
return r.get("suggested_runner") and (size(r["runner"]), size(r["suggested_runner"]))


def suggestion_row(r):
runner, suggested = change(r)
arrow = "⬇️" if rank(suggested) < rank(runner) else "⬆️"
return (f"| [{name(r)}]({r['dashboard']}) | {arrow} {runner} → {suggested} | {r['cpu_cores']['max']:.1f} | "
f"{r['memory_gib']['peak']:.1f} GiB |")
h = r["history"]
return (f"| [{name(r)}]({r['dashboard']}) | {arrow} {runner} → {suggested} | {WHY[r['reason']]} | "
f"{h['peak_mem_gib']:.1f} GiB | {h['cpu_p50']:.1f} / {h['cpu_peak']:.1f} | {h['runs']} |")


def history_row(h, previous):
Expand Down Expand Up @@ -127,11 +126,13 @@ def render(run, results, previous_body):
"|---|---|",
*job_rows(results),
"",
*(["💡 **Suggested runner changes.** Memory is the exact peak; CPU is the max of 1 min averages, so "
"trial before downsizing on it. Jobs under 2 min have too few samples and are left out.",
*([f"💡 **Suggested runner changes.** Memory is a hard limit: a size must hold {MEMORY_HEADROOM} × the "
f"job's peak over the last {WINDOW}, or the job is OOM-killed. CPU only costs time: a smaller size "
"must cover the job's median CPU, and is a *trial* when its peak would not fit. "
f"Smaller sizes are only suggested after {MIN_RUNS} runs in the window.",
"",
"| Job | Change | CPU | Memory |",
"|---|---|---|---|",
f"| Job | Change | Why | Peak memory ({WINDOW}) | CPU median / peak | Runs ({WINDOW}) |",
"|---|---|---|---|---|---|",
*map(suggestion_row, changed),
""] if changed else []),
f"<!-- history {json.dumps(runs)} -->",
Expand All @@ -156,8 +157,7 @@ def main():
if not prs:
sys.exit(f"no pull request found for run {run_id}")
pr = prs[0]["number"]
results = [analyse(j) for j in jobs_for(args[0])
if j.get("runner_name") and ARC_RUNNER.match(j["runner_name"]) and j.get("completed_at")]
results = analyse_run(args[0])
comments = json.loads(sh("gh", "api", "--paginate", "--slurp", f"repos/{repo}/issues/{pr}/comments"))
existing = next((c for page in reversed(comments) for c in reversed(page)
if MARKER in c["body"] and c["user"]["type"] == "Bot"), None)
Expand Down
105 changes: 81 additions & 24 deletions .github/actions/arc-job-report/job_resources.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,14 +7,14 @@
Needs `gh` (authenticated) and either GRAFANA_TOKEN (a Grafana service-account token with read
access to the Prometheus datasource) or `gcx` (a context with that datasource).
A job's runner name (GitHub API `runner_name`) is its ARC pod name; metrics are scraped every 15 s.

Copied from monta-app/kube-manifests infra/arc/scripts/job-resources.py @ dd43b0172; the only
change is the GRAFANA_TOKEN query path in prom().
A job's history across runs comes from the arc:job_* recording rules (monta-app/terraform,
global/grafana/teams/sre/alerts_arc.tf).
"""
import json
import math
import os
import re
import statistics
import subprocess
import sys
import urllib.parse
Expand All @@ -33,23 +33,51 @@
("arc-x64-4cpu-12gb", "amd64", 4, 12),
("arc-x64-8cpu-24gb", "amd64", 8, 24),
]
JOB_MEMORY = "arc:job_memory_working_set_bytes:max1m"
JOB_CPU = "arc:job_cpu_usage_cores:rate1m"
WINDOW = "7d"
MIN_RUNS = 5
MEMORY_HEADROOM = 1.3
CPU_BOUND = 0.85


def sh(*args):
return subprocess.run(args, capture_output=True, text=True, check=True).stdout


def prom(expr, start, end, step=15):
def get(api, **params):
path = f"/api/datasources/proxy/uid/{DATASOURCE}/api/v1/{api}?{urllib.parse.urlencode(params)}"
token = os.environ.get("GRAFANA_TOKEN")
if token:
query = urllib.parse.urlencode({"query": expr, "start": start, "end": end, "step": step})
req = urllib.request.Request(f"{GRAFANA}/api/datasources/proxy/uid/{DATASOURCE}/api/v1/query_range?{query}",
headers={"Authorization": f"Bearer {token}"})
req = urllib.request.Request(GRAFANA + path, headers={"Authorization": f"Bearer {token}"})
with urllib.request.urlopen(req, timeout=30) as resp:
return json.load(resp)["data"]["result"]
out = sh("gcx", "metrics", "query", "-d", DATASOURCE, expr,
"--from", str(start), "--to", str(end), "--step", f"{step}s", "-o", "json")
return json.loads(out)["data"]["result"]
return json.loads(sh("gcx", "api", path))["data"]["result"]


def prom(expr, start, end, step=15):
return get("query_range", query=expr, start=start, end=end, step=step)


def label(value):
return value.replace("\\", "\\\\").replace('"', '\\"')


def history(repo, names):
if not names:
return {}
regex = "|".join(re.sub(r"([.*+?^${}()|\[\]\\])", r"\\\1", n) for n in sorted(set(names)))
sel = f'repository="{label(repo)}", job_name=~"{label(regex)}"'

def by_job(expr):
return {r["metric"]["job_name"]: float(r["value"][1]) for r in get("query", query=expr)}

mem = by_job(f"max by (job_name) (max_over_time({JOB_MEMORY}{{{sel}}}[{WINDOW}]))")
cpu_p50 = by_job(f"quantile by (job_name) (0.5, quantile_over_time(0.5, {JOB_CPU}{{{sel}}}[{WINDOW}]))")
cpu_peak = by_job(f"max by (job_name) (max_over_time({JOB_CPU}{{{sel}}}[{WINDOW}]))")
runs = by_job(f"count by (job_name) (count by (job_name, workflow_run_id) (count_over_time({JOB_MEMORY}{{{sel}}}[{WINDOW}])))")
return {n: {"runs": int(runs.get(n, 0)), "peak_mem_gib": mem[n] / 2**30,
"cpu_p50": cpu_p50.get(n), "cpu_peak": cpu_peak.get(n)} for n in mem}


def values(result):
Expand All @@ -75,15 +103,27 @@ def jobs_for(url):
return [json.loads(line) for line in out.splitlines() if line]


def suggest(arch, peak_mem_gib, p95_cpu):
need_mem, need_cpu = peak_mem_gib * 1.3, max(p95_cpu * 1.2, 1)
for name, a, cpu, mem in CATALOG:
if a == arch and mem >= need_mem and cpu >= need_cpu:
return name
return next((n for n, a, _, _ in reversed(CATALOG) if a == arch), None)


def analyse(job):
def suggest(runner, arch, peak_mem_gib, cpu_p50, cpu_peak, runs, oom):
sizes = [c for c in CATALOG if c[1] == arch]
now = next((c for c in sizes if runner.startswith(c[0] + "-")), None)
if not now:
return None
i = sizes.index(now)
fits = [c for c in sizes if c[3] >= peak_mem_gib * MEMORY_HEADROOM]
need = sizes.index(fits[0]) if fits else len(sizes) - 1
if oom:
need = max(need, i + 1)
if need > i:
return sizes[min(need, len(sizes) - 1)][0], "memory"
if cpu_p50 >= CPU_BOUND * now[2] and i + 1 < len(sizes):
return sizes[i + 1][0], "faster"
smaller = next((c for c in fits if c[2] >= cpu_p50), now)
if runs < MIN_RUNS or sizes.index(smaller) >= i:
return None
return smaller[0], "fits" if smaller[2] >= cpu_peak else "trial"


def analyse(job, hist):
pod, start, end = job["runner_name"], ts(job["started_at"]), ts(job["completed_at"])
sel = f'cluster="internal", namespace="arc", pod="{pod}"'
lo, hi = start - 60, end + 60
Expand All @@ -110,13 +150,22 @@ def analyse(job):
cpu_sorted = sorted(cpu)
p95 = cpu_sorted[min(len(cpu_sorted) - 1, math.ceil(0.95 * len(cpu_sorted)) - 1)]
peak_mem = max(mem) / 2**30
h = hist.get(job["name"], {})
runs = h.get("runs", 0)
week = {"runs": runs, "peak_mem_gib": round(max(peak_mem, h.get("peak_mem_gib", 0)), 2),
"cpu_p50": round(h["cpu_p50"] if runs >= MIN_RUNS and h.get("cpu_p50") is not None
else statistics.median(cpu), 2),
"cpu_peak": round(max(max(cpu), h.get("cpu_peak") or 0), 2)}
res.update({
"cpu_cores": {"avg": round(sum(cpu) / len(cpu), 2), "p95": round(p95, 2), "max": round(max(cpu), 2),
"request": max(req) if req else None},
"memory_gib": {"peak": round(peak_mem, 2), "limit": round(max(lim) / 2**30, 1) if lim else None},
"oom_kills": int(max(oom)) if oom else 0,
"suggested_runner": suggest(res["node"]["arch"], peak_mem, p95),
"history": week,
})
res["suggested_runner"], res["reason"] = suggest(
pod, res["node"]["arch"], week["peak_mem_gib"], week["cpu_p50"], week["cpu_peak"], runs,
res["oom_kills"]) or (None, None)
notes = []
if duration < 120:
notes.append("short job: few samples, low confidence")
Expand All @@ -129,12 +178,18 @@ def analyse(job):
return res


def analyse_run(url):
jobs = [j for j in jobs_for(url)
if j.get("runner_name") and ARC_RUNNER.match(j["runner_name"]) and j.get("completed_at")]
hist = history(re.search(r"github\.com/([^/]+/[^/]+)/", url).group(1), [j["name"] for j in jobs])
return [analyse(j, hist) for j in jobs]


def main():
args = [a for a in sys.argv[1:] if not a.startswith("--")]
if not args:
sys.exit(__doc__)
results = [analyse(j) for j in jobs_for(args[0])
if j.get("runner_name") and ARC_RUNNER.match(j["runner_name"]) and j.get("completed_at")]
results = analyse_run(args[0])
if "--json" in sys.argv:
print(json.dumps(results, indent=2))
return
Expand All @@ -145,10 +200,12 @@ def main():
print(f"\n{r['job']} ({r['duration_s']}s, {r['conclusion']})\n runner {r['runner']}\n"
f" node {n['instance_type']} {n['capacity']} ({n['vcpu']} vCPU)")
if "cpu_cores" in r:
c, m = r["cpu_cores"], r["memory_gib"]
c, m, h = r["cpu_cores"], r["memory_gib"], r["history"]
print(f" cpu avg {c['avg']} p95 {c['p95']} max {c['max']} cores (request {c['request']})\n"
f" memory peak {m['peak']} GiB of {m['limit']} GiB limit OOM kills: {r['oom_kills']}\n"
f" suggest {r['suggested_runner']}")
f" {WINDOW} {h['runs']} runs, peak memory {h['peak_mem_gib']} GiB, "
f"CPU median {h['cpu_p50']} peak {h['cpu_peak']} cores\n"
f" suggest {r['suggested_runner'] or 'keep'} {r['reason'] or ''}")
if r.get("note"):
print(f" note {r['note']}")
print(f" graphs {r['dashboard']}")
Expand Down
Loading