This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 82456dd938a Report failing fork job statuses only when the failure
persists (#76185)
82456dd938a is described below
commit 82456dd938ad2042f09a4c4c10d1294da84f7c5c
Author: Jarek Potiuk <[email protected]>
AuthorDate: Tue Nov 3 21:49:55 2026 +0100
Report failing fork job statuses only when the failure persists (#76185)
The fork CI monitor posted a Slack item each time reading a fork run's
jobs hit a transient GitHub server error (HTTP 502, a reset HTTP/2
stream). The item carried the error text, so a changed error re-posted
it, and one successful run in between re-posted it too.
Count the consecutive runs whose job statuses failed per pull request,
report only after three in a row, and use fixed wording so the item is
posted once while the failure persists. The run log keeps each error.
Generated-by: Claude Opus 5
---
.../doc/ci/08_fork_ci_for_external_contributors.md | 11 ++--
.../src/airflow_breeze/utils/fork_ci_reconcile.py | 47 ++++++++++++++++-
dev/breeze/tests/test_fork_ci_reconcile.py | 60 +++++++++++++++++++++-
3 files changed, 111 insertions(+), 7 deletions(-)
diff --git a/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
b/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
index 5378363b6ef..f1007448d61 100644
--- a/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
+++ b/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
@@ -861,9 +861,14 @@ suites); its jobs come from `GET
/repos/<fork>/actions/runs/<id>/jobs?filter=lat
`jobs: "<run id>:<attempt>:<conclusion>"` (`runAttempt` is fetched with the
check suites), so
later runs do not read its jobs again. A run in progress is read on every
run, and a re-run in
the fork is read again because its attempt changes, even when it ends with
the same conclusion.
-- Reporting job statuses is caught on its own: when it fails, the failure is
listed with the
- unknown items for Slack, and the pull request is still classified, so for
example a `CI passed`
- that no longer holds is still revoked.
+- Reporting job statuses is caught on its own: when it fails, the pull request
is still
+ classified, so for example a `CI passed` that no longer holds is still
revoked. Reading a fork
+ run's jobs fails now and then on GitHub server errors (HTTP 502, a reset
HTTP/2 stream) that the
+ next run gets past, so each failure is only logged, and the pull request is
listed with the
+ unknown items for Slack once its job statuses failed in three runs in a row.
The per pull request
+ count is kept in the same record as that of deferred job statuses; without a
readable record it
+ starts again. The item's wording is fixed whatever the error, so it is
posted once while the
+ failure persists, and the run log has each run's error.
The job statuses are informational. Merging still depends on `CI passed`, and
the state machine
on `Fork CI / Tests (AMD)`; a job status never changes either.
diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
index d15d3a8da37..c9a6865cd96 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
@@ -185,6 +185,16 @@ DEFERRED_RUNS_BEFORE_ALERT = 3
# GitHub answering with rate limits in this many consecutive runs is reported:
a single burst is
# retried, while a persisting one means the repository's shared limits are
close to exhausted.
RATE_LIMITED_RUNS_BEFORE_ALERT = 2
+# Reading a fork run's jobs fails now and then on GitHub server errors (HTTP
502, a reset HTTP/2
+# stream) that the next run gets past, so a pull request whose job statuses
failed is only reported
+# when it failed in this many consecutive runs.
+JOB_STATUS_FAILURE_RUNS_BEFORE_ALERT = 3
+# Fixed text, so the item is posted to Slack once while the failure persists,
whatever the error;
+# the run log has each run's error.
+JOB_STATUSES_FAILED_REASON = (
+ f"reporting job statuses failed in {JOB_STATUS_FAILURE_RUNS_BEFORE_ALERT}
runs in a row; "
+ "the run log has the errors"
+)
# A pull request whose fork run finished this long ago with job statuses still
not all posted is
# reported by itself: deferring is expected to catch up within a few runs, not
to leave one behind.
MIRROR_DELAY_BEFORE_ALERT = timedelta(hours=1)
@@ -2100,6 +2110,7 @@ def drop_reported(
record: bool = True,
deferred_runs: int = 0,
rate_limited_runs: int = 0,
+ job_status_failure_runs: dict[int, int] | None = None,
) -> tuple[list[tuple[int, str]], list[tuple[int, str]], list[str]]:
"""
Leave out the items the previous run already reported, and record this
run's items.
@@ -2109,6 +2120,8 @@ def drop_reported(
:param deferred_runs: the consecutive runs that deferred job statuses,
recorded for the next run.
:param rate_limited_runs: the consecutive runs GitHub rate-limited,
recorded for the next run.
+ :param job_status_failure_runs: per pull request, the consecutive runs
whose job statuses failed,
+ recorded for the next run.
"""
current: dict[str, Any] = {
"unknown": [list(item) for item in unknown],
@@ -2116,6 +2129,9 @@ def drop_reported(
"notes": notes,
"deferred_runs": deferred_runs,
"rate_limited_runs": rate_limited_runs,
+ "job_status_failure_runs": {
+ str(number): runs for number, runs in (job_status_failure_runs or
{}).items()
+ },
}
try:
previous = json.loads(reported_file.read_text())
@@ -2148,6 +2164,24 @@ def read_run_count(reported_file: Path | None, key: str)
-> int | None:
return count if isinstance(count, int) and not isinstance(count, bool)
else 0
+def read_job_status_failure_runs(reported_file: Path | None) -> dict[int, int]:
+ """Per pull request, the consecutive runs whose job statuses failed, as
the previous run recorded."""
+ if reported_file is None:
+ return {}
+ try:
+ previous = json.loads(reported_file.read_text())
+ except (OSError, ValueError):
+ return {}
+ counts = previous.get("job_status_failure_runs") if isinstance(previous,
dict) else None
+ if not isinstance(counts, dict):
+ return {}
+ return {
+ int(number): runs
+ for number, runs in counts.items()
+ if number.isdigit() and isinstance(runs, int) and not isinstance(runs,
bool)
+ }
+
+
def log_rate_limits(client: GitHubClient) -> None:
"""Log what is left of the token's REST and GraphQL quotas; reading them
does not count against them."""
try:
@@ -2243,6 +2277,8 @@ def run_reconcile(
client, repo=repo, budget=max(MAX_JOB_STATUSES_PER_RUN -
ci_run_started_writes, 0), now=now
)
+ previous_job_status_failure_runs =
read_job_status_failure_runs(reported_file)
+ job_status_failure_runs: dict[int, int] = {}
shared_heads = find_heads_shared_across_bases(prs)
gate_change_finder = GateChangeFinder(client, repo=repo)
for listed in prs:
@@ -2262,8 +2298,14 @@ def run_reconcile(
try:
jobs_reported = job_reporter.report(pr, fork)
except Exception as e:
- console_print(f"[warning]#{pr.number}: reporting job
statuses failed: {e}[/]")
- unknown.append((pr.number, f"reporting job statuses
failed: {type(e).__name__}: {e}"))
+ failure_runs =
previous_job_status_failure_runs.get(pr.number, 0) + 1
+ job_status_failure_runs[pr.number] = failure_runs
+ console_print(
+ f"[warning]#{pr.number}: reporting job statuses failed
({failure_runs} runs in a "
+ f"row): {type(e).__name__}: {e}[/]"
+ )
+ if failure_runs >= JOB_STATUS_FAILURE_RUNS_BEFORE_ALERT:
+ unknown.append((pr.number, JOB_STATUSES_FAILED_REASON))
base_config = configs.get(pr.base_ref)
if base_config is not None:
@@ -2338,6 +2380,7 @@ def run_reconcile(
record=not client.dry_run,
deferred_runs=deferred_runs,
rate_limited_runs=rate_limited_runs,
+ job_status_failure_runs=job_status_failure_runs,
)
has_alerts = bool(unknown or attention or notes)
if has_alerts:
diff --git a/dev/breeze/tests/test_fork_ci_reconcile.py
b/dev/breeze/tests/test_fork_ci_reconcile.py
index 330157e83e0..81962145a43 100644
--- a/dev/breeze/tests/test_fork_ci_reconcile.py
+++ b/dev/breeze/tests/test_fork_ci_reconcile.py
@@ -42,6 +42,7 @@ from airflow_breeze.utils.fork_ci_messages import (
)
from airflow_breeze.utils.fork_ci_reconcile import (
JOB_STATUSES_DEFERRED_NOTE,
+ JOB_STATUSES_FAILED_REASON,
RATE_LIMITED_NOTE,
BaseGateConfigs,
BypassFinder,
@@ -78,6 +79,7 @@ from airflow_breeze.utils.fork_ci_reconcile import (
pull_request_changes_ci_files,
read_ci_run_at,
read_head_status,
+ read_job_status_failure_runs,
read_project_ci_run_url,
report_project_ci_runs_started,
run_reconcile,
@@ -2252,6 +2254,7 @@ def
test_drop_reported_reports_everything_without_a_readable_previous_run(tmp_pa
"notes": [],
"deferred_runs": 0,
"rate_limited_runs": 0,
+ "job_status_failure_runs": {},
}
@@ -2646,10 +2649,63 @@ def
test_run_reconcile_classifies_when_reporting_job_statuses_fails(
mock_reporter.return_value.deferred = []
mock_reporter.return_value.delayed = []
- assert _reconcile_with_reporter(client, tmp_path) is True
+ _reconcile_with_reporter(client, tmp_path)
assert mock_classify.call_args.kwargs["jobs_reported"] is None
- assert "reporting job statuses failed" in (tmp_path / "s.json").read_text()
+
+
[email protected]("airflow_breeze.utils.fork_ci_reconcile.JobStatusReporter",
autospec=True)
[email protected]("airflow_breeze.utils.fork_ci_reconcile.classify", autospec=True)
+def test_run_reconcile_reports_failing_job_statuses_once_when_failing_persists(
+ mock_classify, mock_reporter, client, tmp_path
+):
+ mock_classify.return_value = Plan(state=ForkState.RUNNING)
+ mock_reporter.return_value.deferred = []
+ mock_reporter.return_value.delayed = []
+ client.dry_run = False
+ reported_file = tmp_path / "reported.json"
+ reported_file.write_text("{}")
+ alerts = []
+ errors = [
+ GitHubError("HTTP 502"),
+ GitHubError("HTTP 502"),
+ None,
+ GitHubError("HTTP 502"),
+ GitHubError("stream ID 1; CANCEL"),
+ GitHubError("HTTP 502"),
+ GitHubError("stream ID 1; CANCEL"),
+ ]
+ for error in errors:
+ mock_reporter.return_value.report.side_effect = error
+ mock_reporter.return_value.report.return_value = None
+ alerts.append(_reconcile_with_reporter(client, tmp_path,
reported_file=reported_file))
+
+ assert alerts == [False, False, False, False, False, True, False]
+ assert JOB_STATUSES_FAILED_REASON in (tmp_path / "s.json").read_text()
+ assert json.loads(reported_file.read_text())["job_status_failure_runs"] ==
{"1": 4}
+
+
[email protected](
+ ("content", "expected"),
+ [
+ pytest.param(None, {}, id="missing"),
+ pytest.param("not json", {}, id="not-json"),
+ pytest.param("[]", {}, id="not-a-dict"),
+ pytest.param('{"deferred_runs": 1}', {},
id="recorded-before-the-count"),
+ pytest.param('{"job_status_failure_runs": []}', {},
id="malformed-counts"),
+ pytest.param(
+ '{"job_status_failure_runs": {"1": 2, "x": 3, "4": true, "5":
"6"}}',
+ {1: 2},
+ id="malformed-entries",
+ ),
+ ],
+)
+def test_read_job_status_failure_runs(tmp_path, content, expected):
+ reported_file = tmp_path / "reported.json"
+ if content is not None:
+ reported_file.write_text(content)
+
+ assert read_job_status_failure_runs(reported_file) == expected
@mock.patch("airflow_breeze.utils.fork_ci_reconcile.JobStatusReporter",
autospec=True)
