[v3-3-test] Show prepare breeze timing in CI duration alerts (#69732)

scripts/ci/analyze_ci_job_durations.py

@@ -66,8 +66,15 @@
 import sys
 from datetime import datetime
 from pathlib import Path
+from typing import TypedDict
 
 ISO_SUFFIX_Z = "Z"
+PREPARE_BREEZE_STEP_PREFIX = "Prepare breeze & CI image"
+
+
+class JobDuration(TypedDict):
+    duration: float
+    prepare_breeze_duration: float | None
 
 
 def env_float(name: str, default: float) -> float:
@@ -171,6 +178,25 @@ def format_duration(seconds: float) -> str:
     return f"{minutes}m {secs:02d}s"
 
 
+def format_duration_delta(seconds: float) -> str:
+    """Format a duration delta with an explicit sign."""
+    if seconds < 0:
+        return f"-{format_duration(abs(seconds))}"
+    return f"+{format_duration(seconds)}"
+
+
+def format_prepare_breeze_timing(regression: dict) -> str | None:
+    """Format prepare breeze timing details for a job regression."""
+    if prepare_breeze := regression.get("prepare_breeze"):
+        return (
+            f"{PREPARE_BREEZE_STEP_PREFIX}: "
+            f"{format_duration(prepare_breeze['baseline'])} → "
+            f"{format_duration(prepare_breeze['latest'])} "
+            f"({format_duration_delta(prepare_breeze['increase'])})"
+        )
+    return None
+
+
 # Wall-clock shorter than this almost always means a run that was cancelled,
 # skipped by selective checks, or never really executed the test matrix — not a
 # representative "main build". Such runs would corrupt the duration baseline.
@@ -227,8 +253,18 @@ def get_recent_runs(
     return runs
 
 
-def get_run_jobs(repo: str, run_id: int) -> dict[str, float]:
-    """Return a mapping of job name -> duration in seconds for a single run.
+def get_prepare_breeze_step_duration(job: dict) -> float | None:
+    """Return the prepare breeze step duration for a job, when the step exists."""
+    for step in job.get("steps", []):
+        name = step.get("name", "")
+        if not name.startswith(PREPARE_BREEZE_STEP_PREFIX):
+            continue
+        return duration_seconds(step.get("startedAt"), step.get("completedAt"))
+    return None
+
+
+def get_run_jobs(repo: str, run_id: int) -> dict[str, JobDuration]:
+    """Return a mapping of job name -> duration details for a single run.
 
     Only jobs that completed successfully are included, so that a job which was
     cancelled or skipped on a particular run does not pollute its duration trend.
@@ -247,7 +283,7 @@ def get_run_jobs(repo: str, run_id: int) -> dict[str, float]:
     except json.JSONDecodeError:
         return {}
 
-    durations: dict[str, float] = {}
+    durations: dict[str, JobDuration] = {}
     for job in data.get("jobs", []):
         if job.get("conclusion") != "success":
             continue
@@ -256,7 +292,12 @@ def get_run_jobs(repo: str, run_id: int) -> dict[str, float]:
             continue
         name = job.get("name", "unknown")
         # A matrix can surface the same job name more than once per run; keep the longest.
-        durations[name] = max(durations.get(name, 0.0), seconds)
+        existing = durations.get(name)
+        if existing is None or seconds > existing["duration"]:
+            durations[name] = {
+                "duration": seconds,
+                "prepare_breeze_duration": get_prepare_breeze_step_duration(job),
+            }
     return durations
 
 
@@ -297,14 +338,22 @@ def analyze_jobs(
 ) -> list[dict]:
     """Fetch per-job durations and return the jobs whose latest duration regressed."""
     latest_job_durations: dict[str, list[float]] = {}
+    latest_prepare_breeze_durations: dict[str, list[float]] = {}
     for run in latest_runs:
-        for name, seconds in get_run_jobs(repo, run["id"]).items():
-            latest_job_durations.setdefault(name, []).append(seconds)
+        for name, job_data in get_run_jobs(repo, run["id"]).items():
+            latest_job_durations.setdefault(name, []).append(job_data["duration"])
+            prepare_breeze_duration = job_data["prepare_breeze_duration"]
+            if prepare_breeze_duration is not None:
+                latest_prepare_breeze_durations.setdefault(name, []).append(prepare_breeze_duration)
 
     baseline_job_durations: dict[str, list[float]] = {}
+    baseline_prepare_breeze_durations: dict[str, list[float]] = {}
     for run in baseline_runs:
-        for name, seconds in get_run_jobs(repo, run["id"]).items():
-            baseline_job_durations.setdefault(name, []).append(seconds)
+        for name, job_data in get_run_jobs(repo, run["id"]).items():
+            baseline_job_durations.setdefault(name, []).append(job_data["duration"])
+            prepare_breeze_duration = job_data["prepare_breeze_duration"]
+            if prepare_breeze_duration is not None:
+                baseline_prepare_breeze_durations.setdefault(name, []).append(prepare_breeze_duration)
 
     regressions: list[dict] = []
     for name, latest_values in latest_job_durations.items():
@@ -315,6 +364,16 @@ def analyze_jobs(
             latest_values, baseline_values, rel_threshold, min_abs_increase_seconds
         )
         if regression:
+            latest_prepare_breeze_values = latest_prepare_breeze_durations.get(name, [])
+            baseline_prepare_breeze_values = baseline_prepare_breeze_durations.get(name, [])
+            if latest_prepare_breeze_values and len(baseline_prepare_breeze_values) >= min_baseline_runs:
+                latest_prepare_breeze = median(latest_prepare_breeze_values)
+                baseline_prepare_breeze = median(baseline_prepare_breeze_values)
+                regression["prepare_breeze"] = {
+                    "latest": latest_prepare_breeze,
+                    "baseline": baseline_prepare_breeze,
+                    "increase": latest_prepare_breeze - baseline_prepare_breeze,
+                }
             regression["job"] = name
             regressions.append(regression)
 
@@ -373,11 +432,14 @@ def format_slack_message(
     if job_regressions:
         lines = ["*Jobs that got slower:*"]
         for reg in job_regressions[:15]:
-            lines.append(
+            line = (
                 f"• *{escape_slack_mrkdwn(reg['job'])}* — "
                 f"{format_duration(reg['baseline'])} → *{format_duration(reg['latest'])}* "
                 f"(+{round(reg['rel_increase'] * 100, 1)}%)"
             )
+            if prepare_breeze_timing := format_prepare_breeze_timing(reg):
+                line += f"\n  {escape_slack_mrkdwn(prepare_breeze_timing)}"
+            lines.append(line)
         text = "\n".join(lines)
         if len(text) > 2900:
             text = text[:2900] + "\n_...truncated_"
@@ -470,8 +532,11 @@ def write_step_summary(
             "|-----|----------|--------|----------|",
         ]
         for reg in job_regressions[:25]:
+            job = reg["job"]
+            if prepare_breeze_timing := format_prepare_breeze_timing(reg):
+                job += f"<br>{prepare_breeze_timing}"
             lines.append(
-                f"| {reg['job']} | {format_duration(reg['baseline'])} | "
+                f"| {job} | {format_duration(reg['baseline'])} | "
                 f"{format_duration(reg['latest'])} | +{round(reg['rel_increase'] * 100, 1)}% |"
             )
         lines.append("")

scripts/tests/ci/test_analyze_ci_job_durations.py

@@ -101,6 +101,14 @@ def test_zero_pads_seconds(self, durations_module):
         assert durations_module.format_duration(60 + 5) == "1m 05s"
 
 
+class TestFormatDurationDelta:
+    def test_positive(self, durations_module):
+        assert durations_module.format_duration_delta(60 + 5) == "+1m 05s"
+
+    def test_negative(self, durations_module):
+        assert durations_module.format_duration_delta(-(60 + 5)) == "-1m 05s"
+
+
 class TestDetectRegression:
     def test_flags_regression_above_both_thresholds(self, durations_module):
         # baseline median ~1800s (30m), latest 2700s (45m) -> +50%, +15m
@@ -238,6 +246,13 @@ def test_parses_successful_jobs(self, durations_module):
                         "conclusion": "success",
                         "startedAt": "2026-06-10T13:00:00Z",
                         "completedAt": "2026-06-10T13:20:00Z",
+                        "steps": [
+                            {
+                                "name": "Prepare breeze & CI image: 3.10",
+                                "startedAt": "2026-06-10T13:00:00Z",
+                                "completedAt": "2026-06-10T13:05:00Z",
+                            }
+                        ],
                     },
                     {
                         "name": "Skipped job",
@@ -251,7 +266,58 @@ def test_parses_successful_jobs(self, durations_module):
         completed = subprocess.CompletedProcess(args=[], returncode=0, stdout=payload, stderr="")
         with patch.object(subprocess, "run", return_value=completed):
             jobs = durations_module.get_run_jobs("apache/airflow", 2)
-        assert jobs == {"Tests": 20 * 60}
+        assert jobs == {"Tests": {"duration": 20 * 60, "prepare_breeze_duration": 5 * 60}}
+
+    def test_omits_prepare_breeze_duration_when_step_missing(self, durations_module):
+        payload = json.dumps(
+            {
+                "jobs": [
+                    {
+                        "name": "Tests",
+                        "conclusion": "success",
+                        "startedAt": "2026-06-10T13:00:00Z",
+                        "completedAt": "2026-06-10T13:20:00Z",
+                        "steps": [],
+                    }
+                ]
+            }
+        )
+        completed = subprocess.CompletedProcess(args=[], returncode=0, stdout=payload, stderr="")
+        with patch.object(subprocess, "run", return_value=completed):
+            jobs = durations_module.get_run_jobs("apache/airflow", 2)
+        assert jobs == {"Tests": {"duration": 20 * 60, "prepare_breeze_duration": None}}
+
+    def test_keeps_longest_duplicate_job_name(self, durations_module):
+        payload = json.dumps(
+            {
+                "jobs": [
+                    {
+                        "name": "Tests",
+                        "conclusion": "success",
+                        "startedAt": "2026-06-10T13:00:00Z",
+                        "completedAt": "2026-06-10T13:10:00Z",
+                        "steps": [],
+                    },
+                    {
+                        "name": "Tests",
+                        "conclusion": "success",
+                        "startedAt": "2026-06-10T13:00:00Z",
+                        "completedAt": "2026-06-10T13:20:00Z",
+                        "steps": [
+                            {
+                                "name": "Prepare breeze & CI image: 3.10",
+                                "startedAt": "2026-06-10T13:00:00Z",
+                                "completedAt": "2026-06-10T13:05:00Z",
+                            }
+                        ],
+                    },
+                ]
+            }
+        )
+        completed = subprocess.CompletedProcess(args=[], returncode=0, stdout=payload, stderr="")
+        with patch.object(subprocess, "run", return_value=completed):
+            jobs = durations_module.get_run_jobs("apache/airflow", 2)
+        assert jobs == {"Tests": {"duration": 20 * 60, "prepare_breeze_duration": 5 * 60}}
 
     def test_empty_on_command_failure(self, durations_module):
         completed = subprocess.CompletedProcess(args=[], returncode=1, stdout="", stderr="boom")
@@ -266,9 +332,16 @@ def test_reports_only_regressed_jobs_with_enough_baseline(self, durations_module
 
         def fake_jobs(_repo, run_id):
             if run_id == 100:
-                return {"slow-job": 2700, "stable-job": 600, "new-job": 999}
+                return {
+                    "slow-job": {"duration": 2700, "prepare_breeze_duration": 900},
+                    "stable-job": {"duration": 600, "prepare_breeze_duration": None},
+                    "new-job": {"duration": 999, "prepare_breeze_duration": 300},
+                }
             # baseline runs
-            return {"slow-job": 1800, "stable-job": 590}
+            return {
+                "slow-job": {"duration": 1800, "prepare_breeze_duration": 300},
+                "stable-job": {"duration": 590, "prepare_breeze_duration": None},
+            }
 
         with patch.object(durations_module, "get_run_jobs", side_effect=fake_jobs):
             regressions = durations_module.analyze_jobs(
@@ -282,6 +355,7 @@ def fake_jobs(_repo, run_id):
         names = [r["job"] for r in regressions]
         # slow-job regressed; stable-job did not; new-job lacks baseline samples
         assert names == ["slow-job"]
+        assert regressions[0]["prepare_breeze"] == {"latest": 900, "baseline": 300, "increase": 600}
 
 
 class TestFormatSlackMessage:
@@ -308,3 +382,42 @@ def test_includes_channel_and_blocks(self, durations_module):
         text_blob = json.dumps(msg)
         assert "Tests" in text_blob
         assert "main" in msg["text"]
+
+    def test_includes_prepare_breeze_timing_when_available(self, durations_module):
+        msg = durations_module.format_slack_message(
+            repo="apache/airflow",
+            workflow="ci-amd.yml",
+            branch="main",
+            overall_regression=None,
+            job_regressions=[
+                {
+                    "job": "Tests",
+                    "latest": 1500,
+                    "baseline": 1000,
+                    "increase": 500,
+                    "rel_increase": 0.5,
+                    "prepare_breeze": {"latest": 600, "baseline": 300, "increase": 300},
+                }
+            ],
+            recent_runs=[{"run_number": 102, "html_url": "https://example/2", "duration": 2700}],
+            rel_threshold=0.25,
+            channel="internal-airflow-ci-cd",
+        )
+        text_blob = json.dumps(msg)
+        assert "Prepare breeze & CI image: 5m 00s" in text_blob
+        assert "10m 00s" in text_blob
+
+    def test_omits_prepare_breeze_timing_when_unavailable(self, durations_module):
+        msg = durations_module.format_slack_message(
+            repo="apache/airflow",
+            workflow="ci-amd.yml",
+            branch="main",
+            overall_regression=None,
+            job_regressions=[
+                {"job": "Tests", "latest": 1500, "baseline": 1000, "increase": 500, "rel_increase": 0.5}
+            ],
+            recent_runs=[{"run_number": 102, "html_url": "https://example/2", "duration": 2700}],
+            rel_threshold=0.25,
+            channel="internal-airflow-ci-cd",
+        )
+        assert "Prepare breeze" not in json.dumps(msg)