feat(ci): add runner timing distributions

docs/ci/self-hosted-runners.md

@@ -33,7 +33,9 @@ Pour chaque job disposant de timestamps cohérents :
 - **minutes-runner par heure murale** = somme du travail des jobs / durée de la fenêtre ;
 - **équivalents runners moyens** = somme du travail / durée de la fenêtre, les deux exprimées en minutes.
 
-`run_started_at` n'est pas utilisé pour l'attente : l'API peut le rendre égal au `created_at` du run alors que ses jobs attendent encore. Un job annulé après avoir démarré a consommé un runner et compte dans le travail. Un job `skipped` ou encore en file ne devient jamais une durée zéro : il apparaît dans `incomplete_or_untimed_jobs` et réduit `timing_coverage`. GitHub peut aussi inverser deux timestamps adjacents d'exactement une seconde à cause de leur précision : ces jobs sont exclus du calcul et comptés dans `timestamp_skew_jobs`; une inversion supérieure à une seconde casse la mesure (`exit 2`).
+L'analyse publie aussi les distributions d'attente et de travail (`p50`, `p90`, `max`) dans `by_label` et `by_runner`. Un job qui porte plusieurs labels contribue une fois à chacun d'eux ; les groupes sont triés lexicalement pour rendre les replays déterministes. Les identités absentes restent visibles sous `<unlabelled>` et `<unassigned>` au lieu d'être supprimées. Chaque groupe expose `jobs`, `timed_jobs`, `incomplete_or_untimed_jobs`, `timestamp_skew_jobs` et `timing_coverage` : un percentile sans job temporisable vaut `null`, jamais zéro.
+
+`run_started_at` n'est pas utilisé pour l'attente : l'API peut le rendre égal au `created_at` du run alors que ses jobs attendent encore. Un job annulé après avoir démarré a consommé un runner et compte dans le travail. Un job `skipped` ou encore en file ne devient jamais une durée zéro : il apparaît dans `incomplete_or_untimed_jobs` et réduit `timing_coverage`. GitHub peut aussi inverser des timestamps adjacents à cause de leur précision : ces jobs sont exclus du calcul et comptés dans `timestamp_skew_jobs`, sans transformer l'anomalie en durée négative ou nulle.
 
 La provenance est classée en trois catégories :
 
@@ -45,7 +47,7 @@ Un résultat avec `unknown > 0` ne prouve pas « 100 % same-repo ».
 
 ## Exhaustivité et zéros
 
-L'API Actions plafonne certaines recherches filtrées à 1 000 runs. L'instrument bissecte automatiquement la fenêtre temporelle dès que `total_count >= 1000`, déduplique les runs aux frontières, puis pagine tous les jobs de chaque run. Il refuse la mesure (`exit 2`) si une sous-fenêtre d'une seconde reste plafonnée, si une page disparaît avant le dénominateur annoncé ou si des timestamps donnent une durée négative.
+L'API Actions plafonne certaines recherches filtrées à 1 000 runs. L'instrument bissecte automatiquement la fenêtre temporelle dès que `total_count >= 1000`, déduplique les runs aux frontières, puis pagine tous les jobs de chaque run. Il refuse la mesure (`exit 2`) si une sous-fenêtre d'une seconde reste plafonnée ou si une page disparaît avant le dénominateur annoncé. Un job dont les timestamps donnent une durée négative est exclu des distributions et compté dans `timestamp_skew_jobs`.
 
 Une fenêtre réellement vide est valide et imprime explicitement `runs: 0`, `jobs: 0` et `timing_coverage: null`. Elle est donc distincte d'un instrument cassé. Ne jamais citer un zéro sans son dénominateur et son code retour.
 
@@ -58,10 +60,11 @@ Avant toute bascule, relever au minimum :
 3. la couverture temporelle ;
 4. les minutes-runner/heure ;
 5. le détail par workflow et par conclusion ;
-6. les comptes `same_repo`, `fork` et `unknown` ;
-7. les rafales `runs_created_per_minute`.
+6. les p50/p90/max d'attente et de travail par label et par runner, avec leurs dénominateurs ;
+7. les comptes `same_repo`, `fork` et `unknown` ;
+8. les rafales `runs_created_per_minute`.
 
-Le détail par workflow sépare la capacité réellement consommée de l'auto-contention. En particulier, le `PR gate` peut occuper un runner pendant qu'il sonde des checks eux-mêmes en file : dimensionner sur la demande brute financerait ce temps d'attente au lieu de le corriger.
+Le détail par workflow sépare la capacité réellement consommée de l'auto-contention. En particulier, le `PR gate` peut occuper un runner pendant qu'il sonde des checks eux-mêmes en file : dimensionner sur la demande brute financerait ce temps d'attente au lieu de le corriger. Les distributions par label et runner localisent une saturation observée ; elles ne révèlent pas à elles seules combien de runners partagent un hôte physique, son plafond de concurrence, ni la politique de capacité à retenir. Ces décisions exigent une mesure de topologie distincte.
 
 ## Topologie retenue
 

scripts/ci/measure_runner_demand.py

@@ -248,6 +248,80 @@ def _duration_minutes(start: str, end: str, label: str) -> float | None:
     return (b - a).total_seconds() / 60.0
 
 
+def _percentile(values: list[float], quantile: float) -> float | None:
+    if not values:
+        return None
+    ordered = sorted(values)
+    position = (len(ordered) - 1) * quantile
+    lower = int(position)
+    upper = min(lower + 1, len(ordered) - 1)
+    fraction = position - lower
+    return ordered[lower] + (ordered[upper] - ordered[lower]) * fraction
+
+
+def _timing_group() -> dict:
+    return {
+        "jobs": 0,
+        "timed_jobs": 0,
+        "incomplete_or_untimed_jobs": 0,
+        "timestamp_skew_jobs": 0,
+        "queue_waits": [],
+        "runtimes": [],
+    }
+
+
+def _record_job(group: dict, state: str, wait: float | None, work: float | None) -> None:
+    group["jobs"] += 1
+    if state == "incomplete":
+        group["incomplete_or_untimed_jobs"] += 1
+    elif state == "skew":
+        group["timestamp_skew_jobs"] += 1
+    else:
+        group["timed_jobs"] += 1
+        group["queue_waits"].append(wait)
+        group["runtimes"].append(work)
+
+
+def _render_timing_groups(groups: dict[str, dict], key: str) -> list[dict]:
+    rendered = []
+    for name in sorted(groups):
+        group = groups[name]
+        waits = group["queue_waits"]
+        runtimes = group["runtimes"]
+        rendered.append({
+            key: name,
+            "jobs": group["jobs"],
+            "timed_jobs": group["timed_jobs"],
+            "incomplete_or_untimed_jobs": group["incomplete_or_untimed_jobs"],
+            "timestamp_skew_jobs": group["timestamp_skew_jobs"],
+            "timing_coverage": (
+                round(group["timed_jobs"] / group["jobs"], 6)
+                if group["jobs"] else None
+            ),
+            "queue_wait_minutes": {
+                percentile: (
+                    round(value, 3) if value is not None else None
+                )
+                for percentile, value in (
+                    ("p50", _percentile(waits, 0.5)),
+                    ("p90", _percentile(waits, 0.9)),
+                    ("max", max(waits) if waits else None),
+                )
+            },
+            "runtime_minutes": {
+                percentile: (
+                    round(value, 3) if value is not None else None
+                )
+                for percentile, value in (
+                    ("p50", _percentile(runtimes, 0.5)),
+                    ("p90", _percentile(runtimes, 0.9)),
+                    ("max", max(runtimes) if runtimes else None),
+                )
+            },
+        })
+    return rendered
+
+
 def analyze(snapshot: dict) -> dict:
     if snapshot.get("schema_version") != 1:
         raise MeasurementError("unsupported or missing snapshot schema_version")
@@ -267,7 +341,8 @@ def analyze(snapshot: dict) -> dict:
         lambda: {"runs": 0, "jobs": 0, "timed_jobs": 0, "runner_minutes": 0.0, "queue_minutes": 0.0}
     )
     workflow_conclusions: dict[str, Counter] = defaultdict(Counter)
-    workflow_conclusions: dict[str, Counter] = defaultdict(Counter)
+    label_data: dict[str, dict] = defaultdict(_timing_group)
+    runner_data: dict[str, dict] = defaultdict(_timing_group)
     total_jobs = timed_jobs = incomplete_jobs = skipped_without_start = 0
     timestamp_skew_jobs = 0
     runner_minutes = queue_minutes = 0.0
@@ -295,29 +370,48 @@ def analyze(snapshot: dict) -> dict:
         for job in jobs:
             total_jobs += 1
             workflow_data[workflow]["jobs"] += 1
+            raw_labels = job.get("labels")
+            if not isinstance(raw_labels, list) or any(
+                not isinstance(label, str) for label in raw_labels
+            ):
+                raise MeasurementError(f"job {job.get('id')} labels must be a list of strings")
+            labels = sorted({label for label in raw_labels if label})
+            if not labels:
+                labels = ["<unlabelled>"]
+            runner = str(job.get("runner_name") or "<unassigned>")
+            groups = [label_data[label] for label in labels] + [runner_data[runner]]
+
             started, completed = job.get("started_at"), job.get("completed_at")
             created_job = job.get("created_at")
             if not started:
                 incomplete_jobs += 1
                 if job.get("conclusion") == "skipped":
                     skipped_without_start += 1
+                for group in groups:
+                    _record_job(group, "incomplete", None, None)
                 continue
             if not completed:
                 incomplete_jobs += 1
+                for group in groups:
+                    _record_job(group, "incomplete", None, None)
                 continue
             if not created_job:
                 raise MeasurementError(f"started job {job.get('id')} has no created_at")
             work = _duration_minutes(started, completed, f"job {job.get('id')} runtime")
             wait = _duration_minutes(created_job, started, f"job {job.get('id')} queue wait")
             if work is None or wait is None:
                 timestamp_skew_jobs += 1
+                for group in groups:
+                    _record_job(group, "skew", wait, work)
                 continue
             timed_jobs += 1
             runner_minutes += work
             queue_minutes += wait
             workflow_data[workflow]["timed_jobs"] += 1
             workflow_data[workflow]["runner_minutes"] += work
             workflow_data[workflow]["queue_minutes"] += wait
+            for group in groups:
+                _record_job(group, "timed", wait, work)
 
     by_workflow = []
     for name, data in workflow_data.items():
@@ -357,6 +451,8 @@ def analyze(snapshot: dict) -> dict:
         "run_conclusions": dict(sorted(conclusions.items())),
         "runs_created_per_minute": dict(sorted(burst.items())),
         "by_workflow": by_workflow,
+        "by_label": _render_timing_groups(label_data, "label"),
+        "by_runner": _render_timing_groups(runner_data, "runner_name"),
     }
 
 

scripts/tests/test_measure_runner_demand.py

@@ -35,7 +35,15 @@ def run(run_id, created, repo="jsboige/CoursIA", name="Build", conclusion="succe
     }
 
 
-def job(job_id, created, started, completed, conclusion="success"):
+def job(
+    job_id,
+    created,
+    started,
+    completed,
+    conclusion="success",
+    runner_name=None,
+    labels=None,
+):
     return {
         "id": job_id,
         "name": f"job-{job_id}",
@@ -44,8 +52,8 @@ def job(job_id, created, started, completed, conclusion="success"):
         "created_at": mod.iso_z(created) if created else None,
         "started_at": mod.iso_z(started) if started else None,
         "completed_at": mod.iso_z(completed) if completed else None,
-        "runner_name": "GitHub Actions 1" if started else None,
-        "labels": ["ubuntu-latest"],
+        "runner_name": runner_name or ("GitHub Actions 1" if started else None),
+        "labels": ["ubuntu-latest"] if labels is None else labels,
     }
 
 
@@ -191,16 +199,72 @@ def test_workflow_breakdown_keeps_run_conclusions():
     assert by_name["Build"]["run_conclusions"] == {"cancelled": 1}
 
 
-def test_workflow_breakdown_keeps_run_conclusions():
-    rows = [
-        with_jobs(run(1, dt(0, 1), name="PR gate", conclusion="success"), []),
-        with_jobs(run(2, dt(0, 2), name="PR gate", conclusion="failure"), []),
-        with_jobs(run(3, dt(0, 3), name="Build", conclusion="cancelled"), []),
+def test_label_and_runner_breakdowns_include_percentiles_and_denominators():
+    jobs = [
+        job(
+            1, dt(), dt(0, 1), dt(0, 3), runner_name="runner-b",
+            labels=["self-hosted", "coursia-linux"],
+        ),
+        job(
+            2, dt(), dt(0, 2), dt(0, 6), runner_name="runner-a",
+            labels=["self-hosted", "coursia-linux"],
+        ),
+        job(
+            3, dt(), dt(0, 3), dt(0, 9), runner_name="runner-b",
+            labels=["self-hosted", "coursia-linux"],
+        ),
+        job(
+            4, dt(), dt(0, 4), dt(0, 12), runner_name="runner-a",
+            labels=["self-hosted", "coursia-linux"],
+        ),
+        job(
+            5, dt(), dt(0, 5), dt(0, 15), runner_name="runner-b",
+            labels=["self-hosted", "coursia-linux"],
+        ),
     ]
-    result = mod.analyze(snapshot(rows))
-    by_name = {row["workflow"]: row for row in result["by_workflow"]}
-    assert by_name["PR gate"]["run_conclusions"] == {"failure": 1, "success": 1}
-    assert by_name["Build"]["run_conclusions"] == {"cancelled": 1}
+    result = mod.analyze(snapshot([with_jobs(run(1, dt()), jobs)]))
+
+    assert [row["label"] for row in result["by_label"]] == [
+        "coursia-linux", "self-hosted",
+    ]
+    label = result["by_label"][0]
+    assert label["jobs"] == 5
+    assert label["timed_jobs"] == 5
+    assert label["timing_coverage"] == 1.0
+    assert label["queue_wait_minutes"] == {"p50": 3.0, "p90": 4.6, "max": 5.0}
+    assert label["runtime_minutes"] == {"p50": 6.0, "p90": 9.2, "max": 10.0}
+
+    assert [row["runner_name"] for row in result["by_runner"]] == [
+        "runner-a", "runner-b",
+    ]
+    runner_a = result["by_runner"][0]
+    assert runner_a["queue_wait_minutes"] == {"p50": 3.0, "p90": 3.8, "max": 4.0}
+    assert runner_a["runtime_minutes"] == {"p50": 6.0, "p90": 7.6, "max": 8.0}
+
+
+def test_group_breakdowns_expose_incomplete_skew_and_missing_identity():
+    incomplete = job(1, dt(), None, None, labels=[])
+    skew = job(
+        2, dt(), dt(0, 2), dt(0, 1), runner_name="runner-z",
+        labels=["self-hosted"],
+    )
+    result = mod.analyze(snapshot([with_jobs(run(1, dt()), [incomplete, skew])]))
+
+    labels = {row["label"]: row for row in result["by_label"]}
+    assert labels["<unlabelled>"]["incomplete_or_untimed_jobs"] == 1
+    assert labels["<unlabelled>"]["queue_wait_minutes"]["p50"] is None
+    assert labels["self-hosted"]["timestamp_skew_jobs"] == 1
+    assert labels["self-hosted"]["timed_jobs"] == 0
+
+    runners = {row["runner_name"]: row for row in result["by_runner"]}
+    assert runners["<unassigned>"]["incomplete_or_untimed_jobs"] == 1
+    assert runners["runner-z"]["timestamp_skew_jobs"] == 1
+
+
+def test_non_string_job_label_is_broken():
+    invalid = job(1, dt(), dt(), dt(0, 1), labels=["self-hosted", 3])
+    with pytest.raises(mod.MeasurementError, match="labels must be a list of strings"):
+        mod.analyze(snapshot([with_jobs(run(1, dt()), [invalid])]))
 
 
 def test_provenance_has_same_repo_fork_and_unknown():