feat(report): measure how long each job waits for a runner

report/ci_report.py

@@ -24,7 +24,7 @@
 import subprocess
 import sys
 from collections.abc import Callable
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 
 RUN_FIELDS = "databaseId,conclusion,status,workflowName,createdAt,startedAt,updatedAt,event,headBranch"
@@ -75,6 +75,17 @@ def jobs(self, repo: str, run_id: int) -> list[dict]:
         )
         return json.loads(out or "{}").get("jobs", [])
 
+    def job_timings(self, repo: str, run_id: int) -> list[dict]:
+        """The jobs of a run with `created_at` (when each became ready) and `started_at`: the
+        difference is the time it waited for a runner. `gh run view` does not give the former."""
+        try:
+            out = self._run(
+                ["gh", "api", f"repos/{repo}/actions/runs/{run_id}/jobs?per_page=100"]
+            )
+        except subprocess.CalledProcessError:
+            return []  # the run is gone or unreadable: no figure is better than a wrong one
+        return json.loads(out or "{}").get("jobs", [])
+
     def runners(self, repo: str) -> list[dict]:
         try:
             out = self._run(["gh", "api", f"repos/{repo}/actions/runners"])
@@ -170,6 +181,13 @@ class RepoStats:
     # Only for repositories served by the shared merge train; None elsewhere.
     merged_total: int | None = None
     merged_by_hand: int | None = None
+    # Time each job of the sampled runs waited for a runner (created_at to started_at).
+    jobs_sampled: int = 0
+    job_wait_median_s: float | None = None
+    job_wait_p90_s: float | None = None
+    job_wait_max_s: float | None = None
+    # (job name, median wait in seconds, jobs seen), the three that waited longest.
+    slowest_jobs: list[tuple[str, float, int]] = field(default_factory=list)
 
     @property
     def fail_rate(self) -> float | None:
@@ -201,6 +219,29 @@ def _parse(moment: str) -> dt.datetime:
     return dt.datetime.fromisoformat(moment)
 
 
+def job_waits(timings: list[dict]) -> list[tuple[str, float]]:
+    """(job name, seconds it waited for a runner) for the jobs that actually started."""
+    waits = []
+    for item in timings:
+        if item.get("conclusion") == "skipped":
+            continue
+        created, started = item.get("created_at"), item.get("started_at")
+        if not created or not started:
+            continue
+        seconds = (_parse(started) - _parse(created)).total_seconds()
+        waits.append((item.get("name") or "unnamed job", max(seconds, 0)))
+    return waits
+
+
+def percentile(values: list[float], pct: int) -> float | None:
+    """Nearest-rank percentile; None for no values."""
+    if not values:
+        return None
+    ordered = sorted(values)
+    rank = max(1, -(-len(ordered) * pct // 100))  # ceil without floats
+    return ordered[min(rank, len(ordered)) - 1]
+
+
 def parse_moment(text: str) -> dt.datetime:
     """A date, or a date and time, read as UTC."""
     moment = dt.datetime.fromisoformat(text)
@@ -262,6 +303,24 @@ def collect(
     if waits:
         stats.median_queue_s = statistics.median(waits)
 
+    by_job: dict[str, list[float]] = {}
+    for r in sorted(conclusive, key=lambda item: item["createdAt"], reverse=True)[
+        :sample
+    ]:
+        for name, seconds in job_waits(gh.job_timings(repo, r["databaseId"])):
+            by_job.setdefault(name, []).append(seconds)
+    every = [seconds for group in by_job.values() for seconds in group]
+    if every:
+        stats.jobs_sampled = len(every)
+        stats.job_wait_median_s = statistics.median(every)
+        stats.job_wait_p90_s = percentile(every, 90)
+        stats.job_wait_max_s = max(every)
+        ranked = sorted(
+            ((n, statistics.median(g), len(g)) for n, g in by_job.items()),
+            key=lambda item: (-item[1], item[0]),
+        )
+        stats.slowest_jobs = ranked[:3]
+
     for runner in gh.runners(repo):
         if runner.get("status") == "online":
             stats.runners_online += 1
@@ -410,6 +469,28 @@ def render(
             f"| {s.repo} | {s.conclusive} | {s.failed} | {_pct(s.fail_rate)} | {_pct(s.infra_share)}"
             f" | {fmt_duration(s.median_duration_s)} | {fmt_duration(s.median_queue_s)} |"
         )
+    if any(s.jobs_sampled for s in stats):
+        lines += [
+            "",
+            "## Queue time per job",
+            "",
+            "Time each job of the most recent runs waited for a runner, from the moment it was "
+            "ready to the moment it started. A job waits when every runner slot is taken, so a "
+            "high figure with few runners online asks for more slots; one with plenty of idle "
+            "runners points at the runner service instead.",
+            "",
+            "| Repository | Jobs | Median wait | p90 | Max | Slowest to start | Runners online |",
+            "| --- | ---: | ---: | ---: | ---: | --- | ---: |",
+        ]
+        for s in stats:
+            slowest = ", ".join(
+                f"{name} ({fmt_duration(wait)})" for name, wait, _ in s.slowest_jobs[:1]
+            )
+            lines.append(
+                f"| {s.repo} | {s.jobs_sampled} | {fmt_duration(s.job_wait_median_s)}"
+                f" | {fmt_duration(s.job_wait_p90_s)} | {fmt_duration(s.job_wait_max_s)}"
+                f" | {slowest or '-'} | {s.runners_online} |"
+            )
     lines += ["", "## Snapshot", ""]
     for s in stats:
         lines.append(f"### {s.repo}")

tests/test_ci_report.py

@@ -68,6 +68,9 @@ def jobs(self, repo, run_id):
             out.append(job(name, conclusion, started, failed_step))
         return out
 
+    def job_timings(self, repo, run_id):
+        return getattr(self, "timings", {}).get(run_id, [])
+
     def runners(self, repo):
         return self._runners
 
@@ -414,6 +417,93 @@ def test_durations_are_formatted_for_people(self):
         self.assertEqual(report.fmt_duration(3720), "1h02m")
 
 
+def timing(name, queued_s, conclusion="success", started=True):
+    """A job from the REST API that waited `queued_s` seconds for a runner."""
+    created = NOW - dt.timedelta(hours=1)
+    began = created + dt.timedelta(seconds=queued_s)
+    stamp = "%Y-%m-%dT%H:%M:%SZ"
+    return {
+        "name": name,
+        "conclusion": conclusion,
+        "created_at": created.strftime(stamp),
+        "started_at": began.strftime(stamp) if started else None,
+    }
+
+
+class JobQueueTime(unittest.TestCase):
+    def test_a_job_waits_from_its_creation_to_its_start(self):
+        waits = report.job_waits([timing("Verify", 45), timing("Seal", 780)])
+        self.assertEqual(waits, [("Verify", 45), ("Seal", 780)])
+
+    def test_skipped_and_never_started_jobs_do_not_count(self):
+        jobs = [timing("A", 0, "skipped"), timing("B", 0, None, started=False)]
+        self.assertEqual(report.job_waits(jobs), [])
+
+    def test_a_clock_skew_never_gives_a_negative_wait(self):
+        self.assertEqual(report.job_waits([timing("A", -3)]), [("A", 0)])
+
+    def test_the_percentile_is_the_nearest_rank(self):
+        values = [10, 20, 30, 40, 50, 60, 70, 80, 90, 100]
+        self.assertEqual(report.percentile(values, 90), 90)
+        self.assertEqual(report.percentile(values, 50), 50)
+        self.assertEqual(report.percentile([7], 90), 7)
+        self.assertIsNone(report.percentile([], 90))
+
+    def test_the_report_gives_median_p90_max_and_the_slowest_jobs(self):
+        gh = sample_github()
+        gh.timings = {
+            1: [timing("Verify", 10), timing("Seal", 780)],
+            2: [timing("Verify", 20), timing("Seal", 600)],
+            3: [timing("Verify", 30), timing("Seal", 900)],
+        }
+        stats = report.collect(gh, REPO, "CI", days=7, now=NOW)
+        self.assertEqual(stats.jobs_sampled, 6)
+        self.assertEqual(stats.job_wait_max_s, 900)
+        self.assertEqual(stats.job_wait_p90_s, 900)
+        self.assertEqual(stats.job_wait_median_s, 315)  # 10, 20, 30, 600, 780, 900
+        self.assertEqual(stats.slowest_jobs[0], ("Seal", 780, 3))
+
+    def test_a_repository_without_timings_has_no_queue_figures(self):
+        stats = report.collect(sample_github(), REPO, "CI", days=7, now=NOW)
+        self.assertEqual(stats.jobs_sampled, 0)
+        self.assertIsNone(stats.job_wait_median_s)
+        self.assertEqual(stats.slowest_jobs, [])
+
+    def test_the_section_names_the_slowest_job_and_the_runner_count(self):
+        gh = sample_github()
+        gh.timings = {1: [timing("Verify", 10), timing("Seal", 780)]}
+        stats = report.collect(gh, REPO, "CI", days=7, now=NOW)
+        text = report.render([stats], days=7, now=NOW)
+        self.assertIn("## Queue time per job", text)
+        self.assertIn(
+            f"| {REPO} | 2 | 6m35s | 13m00s | 13m00s | Seal (13m00s) | 1 |", text
+        )
+
+    def test_the_section_is_left_out_when_no_repository_has_timings(self):
+        stats = report.collect(sample_github(), REPO, "CI", days=7, now=NOW)
+        self.assertNotIn(
+            "## Queue time per job", report.render([stats], days=7, now=NOW)
+        )
+
+    def test_the_real_command_reads_the_jobs_of_the_run_with_their_creation_time(self):
+        calls = []
+
+        def fake(command):
+            calls.append(command)
+            return json.dumps({"jobs": [timing("Seal", 5)]})
+
+        jobs = report.Github(fake).job_timings(REPO, 42)
+        self.assertEqual(jobs[0]["name"], "Seal")
+        self.assertEqual(calls[0][:2], ["gh", "api"])
+        self.assertIn(f"repos/{REPO}/actions/runs/42/jobs", " ".join(calls[0]))
+
+    def test_a_run_whose_jobs_cannot_be_read_gives_no_timings(self):
+        def refuse(command):
+            raise subprocess.CalledProcessError(1, command, stderr="HTTP 404")
+
+        self.assertEqual(report.Github(refuse).job_timings(REPO, 42), [])
+
+
 class RealCommands(unittest.TestCase):
     def setUp(self):
         self.commands = []