This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 0b03b14cada [v3-3-test] Show prepare breeze timing in CI duration
alerts (#69732) (#69753)
0b03b14cada is described below
commit 0b03b14cada1d59c69926516b1a4b1d4864ae68c
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sat Jul 11 15:45:16 2026 +0200
[v3-3-test] Show prepare breeze timing in CI duration alerts (#69732)
(#69753)
(cherry picked from commit 2e5962245f1e33d5efc8ca880643754009380c5e)
Co-authored-by: GPK <[email protected]>
---
scripts/ci/analyze_ci_job_durations.py | 85 ++++++++++++++--
scripts/tests/ci/test_analyze_ci_job_durations.py | 119 +++++++++++++++++++++-
2 files changed, 191 insertions(+), 13 deletions(-)
diff --git a/scripts/ci/analyze_ci_job_durations.py
b/scripts/ci/analyze_ci_job_durations.py
index 6cc512f645b..a03ddf7a57e 100644
--- a/scripts/ci/analyze_ci_job_durations.py
+++ b/scripts/ci/analyze_ci_job_durations.py
@@ -66,8 +66,15 @@ import subprocess
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("")
diff --git a/scripts/tests/ci/test_analyze_ci_job_durations.py
b/scripts/tests/ci/test_analyze_ci_job_durations.py
index 432fefe44a5..6bac585b60c 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 @@ class TestFormatDuration:
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 @@ class TestGetRunJobs:
"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 @@ class TestGetRunJobs:
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 @@ class TestAnalyzeJobs:
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 @@ class TestAnalyzeJobs:
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 @@ class TestFormatSlackMessage:
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)
