merge queue: checking #1354 on main (2f93d6c), stacked on #1332

engine/hooks/_runner/tests/test_wrap_installed.py

@@ -24,6 +24,9 @@ def setUp(self):
         self.old_home = os.environ.get("HOME")
         os.environ["HOME"] = str(self.home)
         self.addCleanup(self._restore_home)
+        dispatcher_patch = mock.patch.dict(os.environ, {"CATSTACK_HOOK_DISPATCHER": "0"})
+        dispatcher_patch.start()
+        self.addCleanup(dispatcher_patch.stop)
         self.claude_path = self.home / ".claude" / "settings.json"
         self.cursor_path = self.home / ".cursor" / "hooks.json"
         self.codex_path = self.home / ".codex" / "hooks.json"

product/skills/ci-job-timing/SKILL.md

@@ -0,0 +1,43 @@
+---
+name: ci-job-timing
+description: >-
+  Repeatable per-step timing report for any GitLab CI job, from Datadog. Use
+  when asked how long a CI job's steps take, where a slow job spends its time,
+  or to compare the same job across dev/prod or across different jobs
+  (for example "Build LAGVM (dev)", "xmm-image-packager (dev)").
+---
+
+# ci-job-timing
+
+One recipe, any job. The job name and the number of days are the only inputs.
+
+## Why it reads log text
+
+Datadog CI Visibility has job-level events (`@ci.job.name`, `@duration` in
+nanoseconds) but no step-level events. Step times come from `[CI_STEP]` log
+lines written by the CI step wrapper. Lines from side-by-side tracks carry a
+prefix (for example `[gvm]`, `[qssi]`) and Datadog does not split them into
+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):
+   `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:
+   `python3 <skill>/scripts/ci_job_timing.py report --job "<job>" --days 3 --tsv <file>`
+4. For side-by-side steps, pick one `job_id` from a raw `[CI_STEP]` line and run
+   `ci_job_timing.py timeline --job "<job>" --job-id <id>`. Read the start and end
+   times there; never add step averages into a job time.
+5. Optional job-level cross-check: `aggregate_datadog_ci_pipeline_events` with
+   `ci_level=job`, query `@ci.job.name:"<job>"`, metric `@duration`.
+
+## Read the table correctly
+
+- `Runs` counts successful step lines only. A run that stopped early does not
+  count in its later steps, so counts differ between steps.
+- Averages with few runs (n under 10) are weak. Say the n.
+- Steps can overlap, so the column is not a share of the job.
+- Zero rows means a wrong job name or an empty window: the script errors, it
+  never prints an empty table.
+- A job whose steps do not log `[CI_STEP]` lines returns nothing; say that.

product/skills/ci-job-timing/scripts/ci_job_timing.py

@@ -0,0 +1,139 @@
+#!/usr/bin/env python3
+"""Build the Datadog queries and the report for one CI job's step timing.
+
+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`).
+"""
+from __future__ import annotations
+
+import argparse
+import csv
+import io
+import re
+import sys
+from urllib.parse import quote
+
+DEFAULT_HOST = "rivianvw.datadoghq.com"
+# GitLab job names look like "Build LAGVM (dev)": letters, digits, space, ( ) . _ : / -
+_JOB_RE = re.compile(r"^[A-Za-z0-9 ()._:/\-]+$")
+
+
+def check_job(job: str) -> str:
+    if not job or not _JOB_RE.match(job):
+        raise ValueError(f"job name has characters the query cannot carry safely: {job!r}")
+    return job
+
+
+def check_days(days: int) -> int:
+    if not 1 <= days <= 30:
+        raise ValueError(f"days must be 1-30, got {days}")
+    return days
+
+
+def _seconds_expr() -> str:
+    return "CAST(SPLIT_PART(SPLIT_PART(message, 'took ', 2), ' seconds', 1) AS DECIMAL)"
+
+
+def steps_sql(job: str, days: int) -> str:
+    """One row per step: runs, average, min, max seconds. Reads message text, so
+    lines prefixed [gvm]/[qssi] (which Datadog does not split into fields) count too."""
+    check_job(job)
+    check_days(days)
+    flt = f'"[CI_STEP]" STEP_SUCCESS "[job={job}]"'
+    sec = _seconds_expr()
+    return (
+        "SELECT SPLIT_PART(SPLIT_PART(message, 'Step: <', 2), '>', 1) AS step, "
+        f"COUNT(*) AS runs, AVG({sec}) AS avg_s, MIN({sec}) AS min_s, MAX({sec}) AS max_s "
+        f"FROM dd.logs(columns => ARRAY['message'], filter => '{flt}', "
+        f"from_timestamp => NOW() - INTERVAL '{days} day', to_timestamp => NOW()) "
+        "AS (message VARCHAR) GROUP BY SPLIT_PART(SPLIT_PART(message, 'Step: <', 2), '>', 1) "
+        "ORDER BY avg_s DESC LIMIT 60"
+    )
+
+
+def timeline_sql(job: str, job_id: int, days: int) -> str:
+    """Every step line of one run, oldest first, so side-by-side steps are visible."""
+    check_job(job)
+    check_days(days)
+    flt = f'"[CI_STEP]" "[job_id={int(job_id)}]" "[job={job}]"'
+    return (
+        "SELECT timestamp, message "
+        f"FROM dd.logs(columns => ARRAY['timestamp', 'message'], filter => '{flt}', "
+        f"from_timestamp => NOW() - INTERVAL '{days} day', to_timestamp => NOW()) "
+        "AS (timestamp TIMESTAMP, message VARCHAR) ORDER BY timestamp LIMIT 200"
+    )
+
+
+def link(sql: str, host: str = DEFAULT_HOST) -> str:
+    return f"https://{host}/ddsql/editor?q={quote(sql, safe='')}"
+
+
+def parse_tsv(text: str) -> list[dict]:
+    """Parse the TSV block ddsql_run_query returns. Missing header or no rows -> error."""
+    body = text
+    m = re.search(r"<TSV_DATA>(.*?)</TSV_DATA>", text, re.S)
+    if m:
+        body = m.group(1)
+    body = body.strip("\n")
+    if not body.strip():
+        raise ValueError("no rows: the query returned nothing (wrong job name, or no runs in the window)")
+    rows = list(csv.DictReader(io.StringIO(body), delimiter="\t"))
+    need = {"step", "runs", "avg_s", "min_s", "max_s"}
+    if not rows or not need <= set(rows[0]):
+        raise ValueError(f"unexpected columns, need {sorted(need)}; got {list(rows[0]) if rows else 'none'}")
+    return rows
+
+
+def report(job: str, days: int, rows: list[dict], url: str) -> str:
+    out = [
+        f"## {job}: step times, last {days} day(s)",
+        "",
+        "| Step | Runs | Avg (s) | Avg (min) | Fastest (s) | Slowest (s) |",
+        "|---|---|---|---|---|---|",
+    ]
+    for r in rows:
+        avg = float(r["avg_s"])
+        out.append(
+            f"| {r['step']} | {int(float(r['runs']))} | {avg:.0f} | {avg / 60:.1f} "
+            f"| {float(r['min_s']):.0f} | {float(r['max_s']):.0f} |"
+        )
+    out += [
+        "",
+        "Steps can run side by side, so do not add these up to a job time.",
+        f"Query: {url}",
+    ]
+    return "\n".join(out)
+
+
+def main(argv: list[str] | None = None) -> int:
+    ap = argparse.ArgumentParser(description=__doc__)
+    sub = ap.add_subparsers(dest="cmd", required=True)
+    for name in ("queries", "report", "timeline"):
+        p = sub.add_parser(name)
+        p.add_argument("--job", required=True)
+        p.add_argument("--days", type=int, default=3)
+        p.add_argument("--host", default=DEFAULT_HOST)
+        if name == "timeline":
+            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")
+    a = ap.parse_args(argv)
+    try:
+        if a.cmd == "queries":
+            sql = steps_sql(a.job, a.days)
+            print(f"-- steps query\n{sql}\n-- link\n{link(sql, a.host)}")
+        elif a.cmd == "timeline":
+            sql = timeline_sql(a.job, a.job_id, a.days)
+            print(f"-- timeline query\n{sql}\n-- link\n{link(sql, a.host)}")
+        else:
+            rows = parse_tsv(open(a.tsv, encoding="utf-8").read())
+            print(report(a.job, a.days, rows, link(steps_sql(a.job, a.days), a.host)))
+    except (ValueError, OSError) as e:
+        print(f"error: {e}", file=sys.stderr)
+        return 2
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

product/skills/ci-job-timing/tests/test_ci_job_timing.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
+
+import ci_job_timing as c  # noqa: E402
+
+TSV = """<METADATA><total_rows>2</total_rows></METADATA>
+<TSV_DATA>
+step\truns\tavg_s\tmin_s\tmax_s
+Build LAGVM System\t4\t1563.0\t1500\t1600
+Build LAGVM Kernel\t16\t295.0\t250\t446
+</TSV_DATA>"""
+
+
+class StepsSql(unittest.TestCase):
+    def test_job_and_days_are_in_the_query(self):
+        sql = c.steps_sql("Build LAGVM (dev)", 3)
+        self.assertIn('"[job=Build LAGVM (dev)]"', sql)
+        self.assertIn("INTERVAL '3 day'", sql)
+
+    def test_same_template_works_for_another_job(self):
+        sql = c.steps_sql("xmm-image-packager (dev)", 7)
+        self.assertIn("[job=xmm-image-packager (dev)]", sql)
+
+    def test_quote_in_job_name_is_refused(self):
+        with self.assertRaises(ValueError):
+            c.steps_sql("x' OR 1=1 --", 3)
+
+    def test_days_out_of_range_is_refused(self):
+        with self.assertRaises(ValueError):
+            c.steps_sql("Build LAGVM (dev)", 0)
+
+
+class Report(unittest.TestCase):
+    def test_parses_and_renders(self):
+        rows = c.parse_tsv(TSV)
+        out = c.report("Build LAGVM (dev)", 3, rows, "https://example/x")
+        self.assertIn("| Build LAGVM System | 4 | 1563 | 26.1 | 1500 | 1600 |", out)
+
+    def test_empty_result_is_an_error_not_an_empty_table(self):
+        with self.assertRaises(ValueError):
+            c.parse_tsv("<TSV_DATA>\n</TSV_DATA>")
+
+    def test_wrong_columns_is_an_error(self):
+        with self.assertRaises(ValueError):
+            c.parse_tsv("<TSV_DATA>\nfoo\tbar\n1\t2\n</TSV_DATA>")
+
+    def test_timeline_uses_job_id(self):
+        self.assertIn("[job_id=253261575]", c.timeline_sql("Build LAGVM (dev)", 253261575, 3))
+
+
+if __name__ == "__main__":
+    unittest.main()