diff --git a/product/skills/ci-job-timing/SKILL.md b/product/skills/ci-job-timing/SKILL.md index c79b4261..4f21a0da 100644 --- a/product/skills/ci-job-timing/SKILL.md +++ b/product/skills/ci-job-timing/SKILL.md @@ -21,7 +21,18 @@ fields, so the script reads the message text ("took N seconds") instead. ## Steps -1. Build the query and its link (`` is this skill's folder): +0. Context first: why does the job start late? What does it wait for, and which + prerequisite finished last? (`` is this skill's folder.) + `python3 /scripts/ci_job_timing.py context --job "" --pipelines 10 --job-ids ` + Get `` (comma-separated) from Datadog first: `search_datadog_ci_pipeline_events`, + `ci_level=job`, query `@ci.job.name:"" @ci.status:success`, `limit=10`; use the + `job_id` column. Scanning GitLab's pipeline list instead can miss the job on a busy project. + This reads GitLab (read-only) with the token in `GITLAB_TOKEN`; without it the + script prints an error and exits 2. Timing comes from GitLab, not Datadog, + because Datadog job records have no pipeline number. Needs absent from a + pipeline (optional needs) are skipped. Host and project default to + `https://gitlab.rivianvw.io` and `core/firmware` (`--gitlab-host`, `--project`). +1. Build the query and its link: `python3 /scripts/ci_job_timing.py queries --job "" --days 3` 2. Run the printed steps query with the Datadog `ddsql_run_query` tool (from `now-d`). 3. Save the TSV it returns to a file and render the table: diff --git a/product/skills/ci-job-timing/scripts/ci_job_timing.py b/product/skills/ci-job-timing/scripts/ci_job_timing.py index 33e6256f..c942c224 100644 --- a/product/skills/ci-job-timing/scripts/ci_job_timing.py +++ b/product/skills/ci-job-timing/scripts/ci_job_timing.py @@ -3,18 +3,28 @@ The Datadog connector is only reachable from the agent, so this script does the two parts that must be identical on every run: write the query text (`queries`, -`timeline`) and turn the query result into a table (`report`). +`timeline`) and turn the query result into a table (`report`). `context` reads +GitLab (read-only) to show what a job waits for and which prerequisite finished last. """ from __future__ import annotations import argparse import csv import io +import json +import os import re +import statistics import sys +import urllib.error +import urllib.request +from datetime import datetime from urllib.parse import quote DEFAULT_HOST = "rivianvw.datadoghq.com" +DEFAULT_GITLAB = "https://gitlab.rivianvw.io" +DEFAULT_PROJECT = "core/firmware" +TOKEN_ENV = "GITLAB_TOKEN" _JOB_RE = re.compile(r"^[A-Za-z0-9 ()._:/\-]+$") @@ -104,6 +114,179 @@ def report(job: str, days: int, rows: list[dict], url: str) -> str: ] return "\n".join(out) +NEEDS_QUERY = ( + "query($path: ID!, $iid: ID!) { project(fullPath: $path) { pipeline(iid: $iid) { " + "stages { nodes { groups { nodes { jobs { nodes { name needs { nodes { name } } } } } } } } } } }" +) + + +def parse_needs(graphql_json: dict, job: str) -> list[str]: + """Names the job needs, from the stages/groups/jobs GraphQL shape. Job absent -> error.""" + if graphql_json.get("errors"): + raise ValueError(f"GitLab GraphQL returned errors: {graphql_json['errors']}") + pipeline = ((graphql_json.get("data") or {}).get("project") or {}).get("pipeline") + if not pipeline: + raise ValueError("GraphQL reply has no project.pipeline (wrong project or pipeline number)") + for stage in (pipeline.get("stages") or {}).get("nodes") or []: + for group in (stage.get("groups") or {}).get("nodes") or []: + for j in (group.get("jobs") or {}).get("nodes") or []: + if j.get("name") == job: + names = [n["name"] for n in (j.get("needs") or {}).get("nodes") or []] + return list(dict.fromkeys(names)) + raise ValueError(f"job {job!r} not found in the pipeline's stages") + + +def _ts(value: str | None) -> datetime | None: + if not value: + return None + return datetime.fromisoformat(value.replace("Z", "+00:00")) + + +def _latest(jobs: list[dict], name: str, key: str) -> dict | None: + """Latest attempt of one job name by `key` (retries share a name).""" + best, best_t = None, None + for j in jobs: + t = _ts(j.get(key)) if j.get("name") == name else None + if t is not None and (best_t is None or t > best_t): + best, best_t = j, t + return best + + +def summarize(job: str, needs: list[str], jobs_by_pipeline: dict) -> dict: + """Per pipeline: which needed job finished last and how long the job then waited. + + A needed job that is absent or unfinished in a pipeline is skipped (optional needs). + A pipeline where the job never started, or no need finished, yields no row. + """ + rows, last_counts, skipped = [], {}, {} + for pid, jobs in jobs_by_pipeline.items(): + me = _latest(jobs, job, "started_at") + if me is None: + continue + finishers = [] + for need in needs: + n = _latest(jobs, need, "finished_at") + if n is None: + skipped[need] = skipped.get(need, 0) + 1 + else: + finishers.append((_ts(n["finished_at"]), need)) + if not finishers: + continue + fin_t, fin_name = max(finishers) + gap = (_ts(me["started_at"]) - fin_t).total_seconds() + rows.append({ + "pipeline": pid, + "last_finisher": fin_name, + "gap_s": gap, + "queued_s": me.get("queued_duration"), + }) + last_counts[fin_name] = last_counts.get(fin_name, 0) + 1 + gaps = [r["gap_s"] for r in rows] + return { + "job": job, + "needs": list(needs), + "rows": rows, + "last_counts": dict(sorted(last_counts.items(), key=lambda kv: (-kv[1], kv[0]))), + "skipped": skipped, + "median_gap_s": statistics.median(gaps) if gaps else None, + "max_gap_s": max(gaps) if gaps else None, + } + + +def render_context(summary: dict, base_url: str = DEFAULT_GITLAB, project: str = DEFAULT_PROJECT) -> str: + if not summary["rows"]: + raise ValueError("no pipeline had both the job started and a needed job finished") + out = [f"## {summary['job']}: what it waits for", ""] + out.append("Needs: " + (", ".join(f"`{n}`" for n in summary["needs"]) or "none listed")) + n = len(summary["rows"]) + out += ["", f"Last needed job to finish, over {n} pipeline(s):", "", + "| Needed job | Finished last | Share |", "|---|---|---|"] + for name, cnt in summary["last_counts"].items(): + out.append(f"| {name} | {cnt} | {cnt / n:.0%} |") + out += ["", "Gap = last needed job finished -> this job started.", + f"- Median gap: {summary['median_gap_s']:.0f} s", + f"- Maximum gap: {summary['max_gap_s']:.0f} s", "", + "| Pipeline | Last finisher | Gap (s) | Queued (s) |", "|---|---|---|---|"] + for r in summary["rows"]: + url = f"{base_url}/{project}/-/pipelines/{r['pipeline']}" + q = "" if r["queued_s"] is None else f"{float(r['queued_s']):.0f}" + out.append(f"| [{r['pipeline']}]({url}) | {r['last_finisher']} | {r['gap_s']:.0f} | {q} |") + out += ["", "Needs listed but absent (or unfinished) in a pipeline are skipped; they may be optional needs."] + if summary["skipped"]: + out.append("Skipped: " + ", ".join(f"{k} ({v}x)" for k, v in sorted(summary["skipped"].items()))) + return "\n".join(out) + + + + +def http_get_json(url: str, token: str, body: dict | None = None): + """GET (or POST for GraphQL, which is a read-only query). Errors never echo the token.""" + headers = {"PRIVATE-TOKEN": token, "Content-Type": "application/json"} + data = json.dumps(body).encode() if body is not None else None + req = urllib.request.Request(url, data=data, headers=headers) + try: + with urllib.request.urlopen(req, timeout=60) as resp: + return json.load(resp) + except urllib.error.HTTPError as e: + raise OSError(f"GitLab request failed: HTTP {e.code} for {url}") from None + except urllib.error.URLError as e: + raise OSError(f"GitLab request failed: {e.reason} for {url}") from None + + +def _paged(url: str, token: str, get=http_get_json): + page = 1 + while True: + batch = get(f"{url}{'&' if '?' in url else '?'}per_page=100&page={page}", token) + yield from batch + if len(batch) < 100: + return + page += 1 + + +def _pipelines_from_job_ids(api, job, job_ids, token, get): + """Resolve known job numbers to their pipelines. A job number whose name differs is an error.""" + found = {} + for jid in job_ids: + j = get(f"{api}/jobs/{int(jid)}", token) + if j.get("name") != job: + raise ValueError(f"job {int(jid)} is named {j.get('name')!r}, not {job!r}") + pl = j["pipeline"] + found[pl["id"]] = pl["iid"] + return found + + +def fetch_context_inputs(host, project, job, n, token, scan_limit=200, get=http_get_json, job_ids=None): + """Return (needs, jobs_by_pipeline) for pipelines containing `job`. + + With job_ids, those jobs' pipelines are used directly. Without, the newest finished pipelines + are scanned, which can miss the job on a busy project.""" + api = f"{host}/api/v4/projects/{quote(project, safe='')}" + if job_ids: + pipes = _pipelines_from_job_ids(api, job, job_ids, token, get) + jobs_by_pipeline = {pid: list(_paged(f"{api}/pipelines/{pid}/jobs", token, get)) for pid in pipes} + newest_iid = max(pipes.values()) + reply = get(f"{host}/api/graphql", token, + {"query": NEEDS_QUERY, "variables": {"path": project, "iid": str(newest_iid)}}) + return parse_needs(reply, job), jobs_by_pipeline + jobs_by_pipeline, newest_iid, scanned = {}, None, 0 + for p in _paged(f"{api}/pipelines?scope=finished&order_by=id&sort=desc", token, get): + scanned += 1 + if scanned > scan_limit: + break + jobs = list(_paged(f"{api}/pipelines/{p['id']}/jobs", token, get)) + if not any(j.get("name") == job for j in jobs): + continue + jobs_by_pipeline[p["id"]] = jobs + if newest_iid is None: + newest_iid = p["iid"] + if len(jobs_by_pipeline) >= n: + break + if not jobs_by_pipeline: + raise ValueError(f"no job named {job!r} in the {scanned} newest finished pipelines; pass --job-ids with recent job numbers from Datadog") + reply = get(f"{host}/api/graphql", token, + {"query": NEEDS_QUERY, "variables": {"path": project, "iid": str(newest_iid)}}) + return parse_needs(reply, job), jobs_by_pipeline + def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description=__doc__) @@ -117,7 +300,30 @@ def main(argv: list[str] | None = None) -> int: p.add_argument("--job-id", type=int, required=True) if name == "report": p.add_argument("--tsv", required=True, help="file holding the steps query result") + pc = sub.add_parser("context", help="what the job waits for and which need finished last (GitLab, read-only)") + pc.add_argument("--job", required=True) + pc.add_argument("--pipelines", type=int, default=10) + pc.add_argument("--gitlab-host", default=DEFAULT_GITLAB) + pc.add_argument("--project", default=DEFAULT_PROJECT) + pc.add_argument("--job-ids", default="", help="comma-separated recent job numbers for this job (from Datadog); preferred") a = ap.parse_args(argv) + if a.cmd == "context": + token = os.environ.get(TOKEN_ENV) + if not token: + print(f"error: {TOKEN_ENV} is not set; export a GitLab read_api token", file=sys.stderr) + return 2 + try: + check_job(a.job) + if a.pipelines < 1: + raise ValueError("--pipelines must be at least 1") + host = a.gitlab_host.rstrip("/") + ids = [int(x) for x in a.job_ids.split(",") if x.strip()] + needs, by_pipeline = fetch_context_inputs(host, a.project, a.job, a.pipelines, token, job_ids=ids) + print(render_context(summarize(a.job, needs, by_pipeline), host, a.project)) + except (ValueError, OSError) as e: + print(f"error: {e}", file=sys.stderr) + return 2 + return 0 try: if a.cmd == "queries": sql = steps_sql(a.job, a.days) diff --git a/product/skills/ci-job-timing/tests/test_ci_job_timing.py b/product/skills/ci-job-timing/tests/test_ci_job_timing.py index 348fd368..fac3e6ac 100644 --- a/product/skills/ci-job-timing/tests/test_ci_job_timing.py +++ b/product/skills/ci-job-timing/tests/test_ci_job_timing.py @@ -1,7 +1,11 @@ from __future__ import annotations +import contextlib +import io +import os import sys import unittest +from unittest import mock from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) @@ -53,5 +57,115 @@ def test_timeline_uses_job_id(self): self.assertIn("[job_id=253261575]", c.timeline_sql("Build LAGVM (dev)", 253261575, 3)) +def _gql(jobs): + return {"data": {"project": {"pipeline": {"stages": {"nodes": [ + {"groups": {"nodes": [{"jobs": {"nodes": [ + {"name": n, "needs": {"nodes": [{"name": x} for x in needs]}}]}} for n, needs in jobs]}}]}}}}} + + +def _j(name, start, end, queued=5): + return {"name": name, "status": "success", "started_at": start, "finished_at": end, "queued_duration": queued} + + +class Needs(unittest.TestCase): + def test_parse_needs_reads_stages_groups_jobs(self): + g = _gql([("a", []), ("pkg", ["a", "b", "a"])]) + self.assertEqual(c.parse_needs(g, "pkg"), ["a", "b"]) + + def test_missing_job_is_an_error(self): + with self.assertRaises(ValueError): + c.parse_needs(_gql([("a", [])]), "zzz") + + +class Summarize(unittest.TestCase): + P = { + 1: [_j("a", "2026-01-01T00:00:00Z", "2026-01-01T00:10:00Z"), + _j("b", "2026-01-01T00:00:00Z", "2026-01-01T00:20:00Z"), + _j("pkg", "2026-01-01T00:21:00Z", "2026-01-01T00:30:00Z", 7)], + 2: [_j("a", "2026-01-02T00:00:00Z", "2026-01-02T00:30:00Z"), + _j("b", "2026-01-02T00:00:00Z", "2026-01-02T00:20:00Z"), + _j("pkg", "2026-01-02T00:30:10Z", "2026-01-02T00:40:00Z")], + } + + def test_last_finisher_and_gap(self): + s = c.summarize("pkg", ["a", "b"], self.P) + self.assertEqual([r["last_finisher"] for r in s["rows"]], ["b", "a"]) + self.assertEqual([r["gap_s"] for r in s["rows"]], [60.0, 10.0]) + self.assertEqual(s["last_counts"], {"a": 1, "b": 1}) + self.assertEqual(s["median_gap_s"], 35.0) + self.assertEqual(s["max_gap_s"], 60.0) + + def test_absent_need_is_skipped_not_an_error(self): + s = c.summarize("pkg", ["a", "b", "optional"], self.P) + self.assertEqual(s["skipped"], {"optional": 2}) + self.assertEqual(len(s["rows"]), 2) + self.assertIn("optional needs", c.render_context(s)) + + +class ContextCommand(unittest.TestCase): + def _run(self, argv, env): + out, err = io.StringIO(), io.StringIO() + with mock.patch.dict(os.environ, env, clear=False), \ + contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + rc = c.main(argv) + return rc, out.getvalue(), err.getvalue() + + def test_missing_token_exits_2_with_no_report(self): + env = dict(os.environ) + env.pop("GITLAB_TOKEN", None) + out, err = io.StringIO(), io.StringIO() + with mock.patch.dict(os.environ, env, clear=True), \ + contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + rc = c.main(["context", "--job", "pkg", "--pipelines", "2"]) + self.assertEqual(rc, 2) + self.assertEqual(out.getvalue(), "") + self.assertIn("GITLAB_TOKEN", err.getvalue()) + + def test_job_ids_resolve_pipelines_without_scanning(self): + calls = [] + + def fake(url, token, body=None): + calls.append(url) + if url.endswith("/graphql"): + return _gql([("pkg", ["a", "b"])]) + if "/jobs/" in url and "/pipelines/" not in url: + jid = int(url.rsplit("/", 1)[1]) + return {"name": "pkg", "pipeline": {"id": {101: 1, 102: 2}[jid], "iid": {101: 11, 102: 12}[jid]}} + if "/pipelines/" in url: + return Summarize.P[int(url.split("/pipelines/")[1].split("/")[0])] + raise AssertionError(f"unexpected scan call: {url}") + + needs, by_p = c.fetch_context_inputs("https://h", "core/firmware", "pkg", 2, "tok", get=fake, job_ids=[101, 102]) + self.assertEqual(sorted(by_p), [1, 2]) + self.assertEqual(needs, ["a", "b"]) + self.assertFalse(any("order_by" in u for u in calls)) + + def test_job_id_with_a_different_name_is_an_error(self): + def fake(url, token, body=None): + return {"name": "other", "pipeline": {"id": 1, "iid": 11}} + + with self.assertRaises(ValueError): + c.fetch_context_inputs("https://h", "core/firmware", "pkg", 1, "tok", get=fake, job_ids=[101]) + + def test_token_value_never_appears_in_output(self): + secret = "glpat-SECRET-VALUE-123" + + def fake(url, token, body=None): + self.assertEqual(token, secret) + if url.endswith("/graphql"): + return _gql([("pkg", ["a", "b"])]) + if "/pipelines/" in url: + return Summarize.P[int(url.split("/pipelines/")[1].split("/")[0])] + return [{"id": 1, "iid": 11}, {"id": 2, "iid": 12}] + + needs, by_p = c.fetch_context_inputs("https://h", "core/firmware", "pkg", 2, secret, get=fake) + text = c.render_context(c.summarize("pkg", needs, by_p)) + self.assertNotIn(secret, text) + with mock.patch.object(c, "fetch_context_inputs", side_effect=OSError("boom")): + rc, out, err = self._run(["context", "--job", "pkg"], {"GITLAB_TOKEN": secret}) + self.assertEqual(rc, 2) + self.assertNotIn(secret, out + err) + + if __name__ == "__main__": unittest.main()