diff --git a/scripts/ci/analyze_ci_job_durations.py b/scripts/ci/analyze_ci_job_durations.py index 6cc512f645ba3..a03ddf7a57e88 100644 --- a/scripts/ci/analyze_ci_job_durations.py +++ b/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"
{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("") diff --git a/scripts/tests/ci/test_analyze_ci_job_durations.py b/scripts/tests/ci/test_analyze_ci_job_durations.py index 432fefe44a576..6bac585b60c1b 100644 --- a/scripts/tests/ci/test_analyze_ci_job_durations.py +++ b/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)