merge queue: checking #1373 + #1310 + #1356 together on main (b399990)

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 (`<skill>` is this skill's folder):
+0. Context first: why does the job start late? What does it wait for, and which
+   prerequisite finished last? (`<skill>` is this skill's folder.)
+   `python3 <skill>/scripts/ci_job_timing.py context --job "<exact job name>" --pipelines 10 --job-ids <ids>`
+   Get `<ids>` (comma-separated) from Datadog first: `search_datadog_ci_pipeline_events`,
+   `ci_level=job`, query `@ci.job.name:"<job>" @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 <skill>/scripts/ci_job_timing.py queries --job "<exact job name>" --days 3`
 2. Run the printed steps query with the Datadog `ddsql_run_query` tool (from `now-<days>d`).
 3. Save the TSV it returns to a file and render the table:

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)

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()