You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by po...@apache.org on 2026/09/22 13:27:42 UTC
(airflow) branch ci-duration-monitor-sustained-alerts created (now f74edbc36b4)
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a change to branch ci-duration-monitor-sustained-alerts
in repository https://gitbox.apache.org/repos/asf/airflow.git
at f74edbc36b4 Alert on CI job slowdowns only when every recent run is slow
This branch includes the following new commits:
new f74edbc36b4 Alert on CI job slowdowns only when every recent run is slow
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(airflow) 01/01: Alert on CI job slowdowns only when every recent run is slow
Posted by po...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch ci-duration-monitor-sustained-alerts
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit f74edbc36b44f2ea9af0340f4b8c37538336c5fd
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Tue Sep 22 14:05:20 2026 +0200
Alert on CI job slowdowns only when every recent run is slow
The per-job duration alert compares the median of the last three canary
runs against the baseline median. A job that was slow twice and has since
recovered still clears that comparison, so the nightly report announced
blips as regressions: on 2026-09-22 it flagged the MySQL 8.4:3.13
serialization job at 25m -> 33m, from the runs 25m, 33m, 34m, where the
most recent run was already back at the baseline.
Replayed over the 17 successful canaries of 2026-09-12..21, requiring
every run in the latest window to clear the threshold cuts the per-job
alerts from 16 to 5 while keeping the sustained climbs - the
constraints-version-check jobs whose work grows with every day of
dependency drift - and dropping the DB-test and image-build one-nighters.
The report now also carries the evidence behind each alert, so a reader
can judge it without opening the API: the flagged job's last runs, the
band it usually lands in, and the recent runs with their start times and
both image-excluded and wall-clock durations.
Generated-by: Claude Opus 5
---
.github/workflows/ci-duration-monitor.yml | 6 +
scripts/ci/analyze_ci_job_durations.py | 155 +++++++++++++++++++---
scripts/tests/ci/test_analyze_ci_job_durations.py | 124 +++++++++++++++++
3 files changed, 270 insertions(+), 15 deletions(-)
diff --git a/.github/workflows/ci-duration-monitor.yml b/.github/workflows/ci-duration-monitor.yml
index 52f2d06c2fb..65de09dcc58 100644
--- a/.github/workflows/ci-duration-monitor.yml
+++ b/.github/workflows/ci-duration-monitor.yml
@@ -63,6 +63,12 @@ jobs:
# Network-bound jobs (constraint resolution, provider installs) legitimately swing
# tens of minutes run-to-run; require a larger sustained jump before flagging them.
JOB_MIN_ABS_INCREASE_MINUTES: "6"
+ # The median of the latest window still reports a job that was slow twice and has
+ # since recovered: the 2026-09-22 MySQL 8.4:3.13 alert came from 25m, 33m, 34m,
+ # where the most recent run was already back at the 25m baseline. Requiring every
+ # run in the window to be slow drops those blips and keeps the sustained climbs
+ # (replayed over the 2026-09-12..21 canaries).
+ JOB_REQUIRE_SUSTAINED: "true"
OUTPUT_FILE: "slack-message.json"
- name: "Post duration alert to Slack"
diff --git a/scripts/ci/analyze_ci_job_durations.py b/scripts/ci/analyze_ci_job_durations.py
index 202dfb449e1..446bddb31b2 100644
--- a/scripts/ci/analyze_ci_job_durations.py
+++ b/scripts/ci/analyze_ci_job_durations.py
@@ -35,6 +35,12 @@ both a relative margin (``REL_THRESHOLD``) and an absolute floor
(``MIN_ABS_INCREASE_MINUTES`` / ``JOB_MIN_ABS_INCREASE_MINUTES``) so short jobs
with noisy timings do not trigger spurious alerts.
+Per-job alerts have a third gate, because a median over the latest window still
+reports a job that had two slow nights and then recovered. With
+``JOB_REQUIRE_SUSTAINED`` (the default) *every* run in the latest window must clear
+the relative threshold on its own, not just their median, so a blip that has already
+passed is not announced as a regression (:func:`is_sustained`).
+
Steps that build, pull or push images (``IMAGE_WORK_STEP_PREFIXES`` — preparing the
CI/PROD image in a test job, and the build+push of the image-cache jobs) occasionally
balloon on a one-off cache miss, so their time is *excluded* from the run and per-job
@@ -57,6 +63,9 @@ Environment variables (optional):
REL_THRESHOLD - Relative increase over baseline to flag, e.g. 0.25 = 25% (default: 0.25)
MIN_ABS_INCREASE_MINUTES - Absolute floor for the overall-run alert (default: 5)
JOB_MIN_ABS_INCREASE_MINUTES - Absolute floor for per-job alerts (default: 3)
+ JOB_REQUIRE_SUSTAINED - Require every run in the latest window to be above the
+ relative threshold, not just their median ("true"/"false",
+ default: true)
IMAGE_BUILD_PERSISTENCE_DAYS - Only report a slow image build once it has stayed
elevated for at least this many days (default: 2)
ANALYZE_JOBS - Whether to fetch per-job durations ("true"/"false", default: true)
@@ -189,6 +198,37 @@ def median(values: list[float]) -> float:
return (ordered[mid - 1] + ordered[mid]) / 2
+# How many recent runs the report shows per flagged job, and in its recent-runs list.
+# Enough to see the swing a job normally has; more turns the Slack message into a table.
+RECENT_VALUES_SHOWN = 8
+
+
+def usual_range(values: list[float]) -> tuple[float, float]:
+ """Return the middle half (25th-75th percentile, nearest rank) of ``values``.
+
+ Reported as the band a job usually lands in. The middle half rather than min-max
+ because a window that happens to contain one cache-cold 58-minute night would
+ otherwise claim that night as "usual" and make the alert unreadable.
+ """
+ ordered = sorted(values)
+ last = len(ordered) - 1
+ return ordered[min(int(0.25 * len(ordered)), last)], ordered[min(int(0.75 * len(ordered)), last)]
+
+
+def is_sustained(latest_values: list[float], baseline: float, rel_threshold: float) -> bool:
+ """Whether *every* run in the latest window is itself above the relative threshold.
+
+ The median of the latest window is not enough on its own. A job that ran 25m, 33m,
+ 34m has a latest median of 33m against a 25m baseline — but its most recent run was
+ back at baseline, so the two slow nights were a blip, not a trend. Requiring each run
+ in the window to clear the threshold is what separates that from a real regression,
+ where every run after the change is slow (verified against 17 canary runs of
+ 2026-09-12..21: it keeps the constraints-check regressions and drops the DB-test and
+ image-build blips).
+ """
+ return all(value >= baseline * (1 + rel_threshold) for value in latest_values)
+
+
def format_duration(seconds: float) -> str:
"""Format a duration in seconds as e.g. ``29m 41s``."""
total = int(round(seconds))
@@ -198,6 +238,46 @@ def format_duration(seconds: float) -> str:
return f"{minutes}m {secs:02d}s"
+def format_duration_compact(seconds: float) -> str:
+ """Format a duration for a series of many values, e.g. ``33m``.
+
+ Whole minutes: a reader scanning a job's last runs is judging a swing of minutes,
+ and the seconds turn the series into a wall of digits.
+ """
+ if seconds < 60:
+ return f"{int(round(seconds))}s"
+ return f"{int(round(seconds / 60))}m"
+
+
+def format_duration_series(values: list[float]) -> str:
+ """Render newest-first durations as ``33m · 31m · 42m``."""
+ return " · ".join(format_duration_compact(value) for value in values)
+
+
+def format_usual_range(band: tuple[float, float] | None) -> str:
+ """Render a baseline band as ``, usually 16m 00s–23m 00s``, or nothing when absent."""
+ if not band:
+ return ""
+ low, high = band
+ return f", usually {format_duration(low)}–{format_duration(high)}"
+
+
+def format_run_timestamp(created_at: str | None) -> str:
+ """Format a run's start as ``Sep 21 13:58`` (UTC), or empty when unparsable."""
+ started = parse_iso(created_at)
+ return started.strftime("%b %d %H:%M") if started else ""
+
+
+def adjusted_run_duration(run: dict) -> float:
+ """A run's wall-clock with its image-build time removed, as the trend measures it.
+
+ Runs carry the adjusted value once :func:`main` has their jobs; falling back to the
+ raw wall-clock keeps the report readable when jobs were not fetched at all
+ (``ANALYZE_JOBS=false``).
+ """
+ return run.get("adjusted_duration", run["duration"])
+
+
def format_duration_delta(seconds: float) -> str:
"""Format a duration delta with an explicit sign."""
if seconds < 0:
@@ -348,11 +428,14 @@ def detect_regression(
baseline_values: list[float],
rel_threshold: float,
min_abs_increase_seconds: float,
+ require_sustained: bool = False,
) -> dict | None:
"""Compare latest durations against a baseline window.
Returns a dict describing the regression when the latest median is above the
- baseline median by both the relative threshold and the absolute floor, else None.
+ baseline median by the relative threshold and the absolute floor — and, when
+ ``require_sustained`` is set, when every run in the latest window clears the
+ relative threshold too — else None.
"""
if not latest_values or not baseline_values:
return None
@@ -360,12 +443,17 @@ def detect_regression(
baseline = median(baseline_values)
increase = latest - baseline
rel_increase = increase / baseline if baseline > 0 else 0.0
- if increase >= min_abs_increase_seconds and rel_increase >= rel_threshold:
+ if (
+ increase >= min_abs_increase_seconds
+ and rel_increase >= rel_threshold
+ and (not require_sustained or is_sustained(latest_values, baseline, rel_threshold))
+ ):
return {
"latest": latest,
"baseline": baseline,
"increase": increase,
"rel_increase": rel_increase,
+ "usual_range": usual_range(baseline_values),
}
return None
@@ -386,12 +474,15 @@ def analyze_jobs(
min_baseline_runs: int,
rel_threshold: float,
min_abs_increase_seconds: float,
+ require_sustained: bool,
) -> list[dict]:
"""Return the jobs whose latest duration regressed, image-build time excluded.
The comparison uses :func:`calculate_work_duration` (wall-clock minus the image-build
step) on both sides, so an occasional image rebuild spike does not flag a job
- that did not actually get slower.
+ that did not actually get slower. Each job must additionally have been slow on every
+ run of the latest window (:func:`is_sustained`), so a job that has already recovered
+ does not fill the report.
"""
latest_job_durations: dict[str, list[float]] = {}
for run in latest_runs:
@@ -409,10 +500,13 @@ def analyze_jobs(
if len(baseline_values) < min_baseline_runs:
continue
regression = detect_regression(
- latest_values, baseline_values, rel_threshold, min_abs_increase_seconds
+ latest_values, baseline_values, rel_threshold, min_abs_increase_seconds, require_sustained
)
if regression:
regression["job"] = name
+ # Newest first, latest window then baseline: the reader sees the step the alert
+ # is about with the run-to-run swing it stands out from underneath it.
+ regression["recent_values"] = (latest_values + baseline_values)[:RECENT_VALUES_SHOWN]
regressions.append(regression)
regressions.sort(key=lambda r: r["rel_increase"], reverse=True)
@@ -549,13 +643,18 @@ def format_slack_message(
)
if job_regressions:
- lines = ["*Jobs that got slower (image build excluded):*"]
+ lines = ["*Jobs that got slower (image build excluded, slow on every recent run):*"]
for reg in job_regressions[:15]:
lines.append(
f"• *{escape_slack_mrkdwn(reg['job'])}* — "
f"{format_duration(reg['baseline'])} → *{format_duration(reg['latest'])}* "
- f"(+{round(reg['rel_increase'] * 100, 1)}%)"
+ f"(+{round(reg['rel_increase'] * 100, 1)}%"
+ f"{format_usual_range(reg.get('usual_range'))})"
)
+ if reg.get("recent_values"):
+ lines.append(
+ f" _last runs (newest first): {format_duration_series(reg['recent_values'])}_"
+ )
text = "\n".join(lines)
if len(text) > 2900:
text = text[:2900] + "\n_...truncated_"
@@ -574,8 +673,22 @@ def format_slack_message(
)
if recent_runs:
+ run_values = [adjusted_run_duration(run) for run in recent_runs]
+ run_lines = [
+ f"*Last {min(len(recent_runs), RECENT_VALUES_SHOWN)} runs "
+ f"(image build excluded; median {format_duration(median(run_values))}"
+ f"{format_usual_range(usual_range(run_values))}):*"
+ ]
+ for run in recent_runs[:RECENT_VALUES_SHOWN]:
+ run_lines.append(
+ f"• <{run['html_url']}|#{run['run_number']}> "
+ f"{format_run_timestamp(run.get('created_at'))} — "
+ f"{format_duration(adjusted_run_duration(run))} "
+ f"(wall-clock {format_duration(run['duration'])})"
+ )
latest_run = recent_runs[0]
blocks.append({"type": "divider"})
+ blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": "\n".join(run_lines)}})
blocks.append(
{
"type": "context",
@@ -627,7 +740,8 @@ def write_step_summary(
"## ⏱️ CI Duration Trend",
"",
f"Workflow `{workflow}` on `{branch}` — baseline from {baseline_count} preceding runs. "
- "Image-build time is excluded from run/job durations and tracked separately.",
+ "Image-build time is excluded from run/job durations and tracked separately. "
+ "A job is only listed when every run in the latest window was slow, not just their median.",
"",
]
@@ -660,13 +774,16 @@ def write_step_summary(
lines += [
"### ⚠️ Slower jobs",
"",
- "| Job | Baseline | Latest | Increase |",
- "|-----|----------|--------|----------|",
+ "| Job | Baseline | Latest | Increase | Usually | Last runs (newest first) |",
+ "|-----|----------|--------|----------|---------|--------------------------|",
]
for reg in job_regressions[:25]:
+ low, high = reg.get("usual_range") or (0.0, 0.0)
lines.append(
f"| {reg['job']} | {format_duration(reg['baseline'])} | "
- f"{format_duration(reg['latest'])} | +{round(reg['rel_increase'] * 100, 1)}% |"
+ f"{format_duration(reg['latest'])} | +{round(reg['rel_increase'] * 100, 1)}% | "
+ f"{format_duration(low)}–{format_duration(high)} | "
+ f"{format_duration_series(reg.get('recent_values', []))} |"
)
lines.append("")
else:
@@ -676,12 +793,14 @@ def write_step_summary(
lines += [
"### Recent run durations",
"",
- "| Run | Event | Duration |",
- "|-----|-------|----------|",
+ "| Run | Started (UTC) | Image build excluded | Wall-clock |",
+ "|-----|---------------|----------------------|------------|",
]
for run in recent_runs[:15]:
lines.append(
- f"| [#{run['run_number']}]({run['html_url']}) | {run['event']} | "
+ f"| [#{run['run_number']}]({run['html_url']}) | "
+ f"{format_run_timestamp(run.get('created_at'))} | "
+ f"{format_duration(adjusted_run_duration(run))} | "
f"{format_duration(run['duration'])} |"
)
lines.append("")
@@ -701,6 +820,7 @@ def main() -> None:
rel_threshold = env_float("REL_THRESHOLD", 0.25)
min_abs_increase_seconds = env_float("MIN_ABS_INCREASE_MINUTES", 5.0) * 60
job_min_abs_increase_seconds = env_float("JOB_MIN_ABS_INCREASE_MINUTES", 3.0) * 60
+ job_require_sustained = env_bool("JOB_REQUIRE_SUSTAINED", True)
image_build_persistence_days = env_float("IMAGE_BUILD_PERSISTENCE_DAYS", 2.0)
do_analyze_jobs = env_bool("ANALYZE_JOBS", True)
only_successful = env_bool("ONLY_SUCCESSFUL", True)
@@ -733,9 +853,13 @@ def main() -> None:
image_build = calculate_image_build_seconds(jobs_by_run_id.get(run["id"], {})) or 0.0
return max(run["duration"] - image_build, 0.0)
+ # Stored on the run so the report can show the same figure the trend is measured on.
+ for run in runs:
+ run["adjusted_duration"] = calculate_adjusted_run_duration(run)
+
overall_regression = detect_regression(
- [calculate_adjusted_run_duration(r) for r in latest_runs],
- [calculate_adjusted_run_duration(r) for r in baseline_runs],
+ [adjusted_run_duration(r) for r in latest_runs],
+ [adjusted_run_duration(r) for r in baseline_runs],
rel_threshold,
min_abs_increase_seconds,
)
@@ -759,6 +883,7 @@ def main() -> None:
min_baseline_runs,
rel_threshold,
job_min_abs_increase_seconds,
+ job_require_sustained,
)
print(f"Jobs that regressed: {len(job_regressions)}")
diff --git a/scripts/tests/ci/test_analyze_ci_job_durations.py b/scripts/tests/ci/test_analyze_ci_job_durations.py
index be5b3cbc2e1..d82c0c3eb8d 100644
--- a/scripts/tests/ci/test_analyze_ci_job_durations.py
+++ b/scripts/tests/ci/test_analyze_ci_job_durations.py
@@ -156,6 +156,77 @@ class TestDetectRegression:
assert durations_module.detect_regression([], [1, 2], 0.25, 300) is None
assert durations_module.detect_regression([1], [], 0.25, 300) is None
+ def test_a_recovered_blip_is_not_flagged(self, durations_module):
+ """The 2026-09-22 MySQL 8.4:3.13 alert: 25m, 33m, 34m over a 25m baseline.
+
+ The latest median is 33m, but the most recent run was already back at baseline —
+ two slow nights, not a regression.
+ """
+ regression = durations_module.detect_regression(
+ latest_values=[1500, 1980, 2040],
+ baseline_values=[1440, 1500, 1560, 1440, 1500, 1500, 1620],
+ rel_threshold=0.25,
+ min_abs_increase_seconds=360,
+ require_sustained=True,
+ )
+ assert regression is None
+
+ def test_a_sustained_climb_is_flagged(self, durations_module):
+ """The same window's Deps 3.10 alert: 31m, 33m, 27m over an 18m baseline."""
+ regression = durations_module.detect_regression(
+ latest_values=[1860, 1980, 1620],
+ baseline_values=[960, 1020, 1080, 1080, 1380, 1020, 960],
+ rel_threshold=0.25,
+ min_abs_increase_seconds=360,
+ require_sustained=True,
+ )
+ assert regression is not None
+ assert regression["latest"] == 1860
+
+ def test_the_same_blip_is_flagged_without_the_sustained_gate(self, durations_module):
+ """Guards the gate itself: the blip clears the margin and the floor on its own."""
+ regression = durations_module.detect_regression(
+ latest_values=[1500, 1980, 2040],
+ baseline_values=[1440, 1500, 1560, 1440, 1500, 1500, 1620],
+ rel_threshold=0.25,
+ min_abs_increase_seconds=360,
+ )
+ assert regression is not None
+
+ def test_reports_the_usual_range(self, durations_module):
+ regression = durations_module.detect_regression(
+ latest_values=[2700],
+ baseline_values=[1500, 1700, 1800, 1900, 2000],
+ rel_threshold=0.25,
+ min_abs_increase_seconds=300,
+ )
+ assert regression is not None
+ assert regression["usual_range"] == (1700, 1900)
+
+
+class TestUsualRange:
+ def test_middle_half_of_the_values(self, durations_module):
+ assert durations_module.usual_range([10, 20, 30, 40]) == (20, 40)
+
+ def test_excludes_a_single_extreme(self, durations_module):
+ """A cache-cold 58-minute night must not be reported as part of the usual band."""
+ low, high = durations_module.usual_range([16, 17, 18, 18, 17, 16, 58])
+ assert (low, high) == (16, 18)
+
+ def test_single_value(self, durations_module):
+ assert durations_module.usual_range([42]) == (42, 42)
+
+
+class TestIsSustained:
+ def test_true_when_every_run_is_above_the_threshold(self, durations_module):
+ assert durations_module.is_sustained([1860, 1980, 1620], 1080, 0.25) is True
+
+ def test_false_when_one_run_is_back_at_baseline(self, durations_module):
+ assert durations_module.is_sustained([1500, 1980, 2040], 1500, 0.25) is False
+
+ def test_true_for_a_single_latest_run(self, durations_module):
+ assert durations_module.is_sustained([2000], 1000, 0.25) is True
+
class TestGetRecentRuns:
def _runs_payload(self):
@@ -429,6 +500,7 @@ class TestAnalyzeJobs:
min_baseline_runs=5,
rel_threshold=0.25,
min_abs_increase_seconds=180,
+ require_sustained=True,
)
names = [r["job"] for r in regressions]
# slow-job regressed; stable-job did not; new-job lacks baseline samples
@@ -452,6 +524,7 @@ class TestAnalyzeJobs:
min_baseline_runs=5,
rel_threshold=0.25,
min_abs_increase_seconds=60,
+ require_sustained=True,
)
assert regressions == []
@@ -487,6 +560,7 @@ class TestAnalyzeJobs:
min_baseline_runs=5,
rel_threshold=0.25,
min_abs_increase_seconds=360,
+ require_sustained=True,
)
assert regressions == []
@@ -644,6 +718,56 @@ class TestFormatSlackMessage:
assert "18m 00s" in json.dumps(msg) # 1080s latest
assert "image build slow" in msg["text"].lower()
+ def test_reports_the_last_runs_and_the_spread(self, durations_module):
+ """The alert has to carry the evidence: what the job did on each recent run."""
+ 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,
+ "usual_range": (960, 1080),
+ "recent_values": [1500, 1020, 960, 1080],
+ }
+ ],
+ image_build_regression=None,
+ recent_runs=[
+ {
+ "run_number": 102,
+ "html_url": "https://example/2",
+ "duration": 2700,
+ "adjusted_duration": 2400,
+ "created_at": "2026-09-21T13:58:37Z",
+ },
+ {
+ "run_number": 101,
+ "html_url": "https://example/1",
+ "duration": 2500,
+ "adjusted_duration": 2200,
+ "created_at": "2026-09-21T01:58:37Z",
+ },
+ ],
+ rel_threshold=0.25,
+ channel="internal-airflow-ci-cd",
+ )
+ # Read the block text directly: json.dumps escapes the non-ASCII separators away.
+ blob = "\n".join(
+ block["text"]["text"] for block in msg["blocks"] if block.get("text", {}).get("text")
+ )
+ assert "usually 16m 00s–18m 00s" in blob
+ assert "last runs (newest first): 25m · 17m · 16m · 18m" in blob
+ assert "Last 2 runs" in blob
+ assert "median 38m 20s, usually 36m 40s–40m 00s" in blob
+ # Both the figure the trend is measured on and the raw wall-clock, per run.
+ assert "Sep 21 13:58 — 40m 00s (wall-clock 45m 00s)" in blob
+ assert "Sep 21 01:58 — 36m 40s (wall-clock 41m 40s)" in blob
+
def test_omits_image_build_section_when_none(self, durations_module):
msg = durations_module.format_slack_message(
repo="apache/airflow",