This is an automated email from the ASF dual-hosted git repository. potiuk pushed a commit to branch fork-ci-status-latency-and-detail in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 9e39ebbcf699e1e7f832e7a5030ab989b447cd7e Author: Jarek Potiuk <[email protected]> AuthorDate: Tue Nov 10 05:42:51 2026 -0700 List the jobs that did not pass in the fork CI state comment When the fork run of a gated pull request failed, the state comment only said "failed" and linked the run, so the author and reviewers had to open the fork's run to see what broke, and with job statuses turned off or deferred by the write budget the pull request showed nothing more. The state comment of a failed run now lists the jobs that failed, timed out or were cancelled, linking each job in the fork (at most 10). The jobs are reused from the job statuses when they were read in the same run, and a failed read leaves the comment for the next run instead of dropping the list. Generated-by: Claude Opus 5 --- .../doc/ci/08_fork_ci_for_external_contributors.md | 12 ++- .../src/airflow_breeze/utils/fork_ci_messages.py | 41 ++++++- .../src/airflow_breeze/utils/fork_ci_reconcile.py | 73 ++++++++++++- dev/breeze/tests/test_fork_ci_messages.py | 31 ++++++ dev/breeze/tests/test_fork_ci_reconcile.py | 119 ++++++++++++++++++++- 5 files changed, 270 insertions(+), 6 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 6ede7917e34..3014ab1c345 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 @@ -608,7 +608,7 @@ since the draft-state job runs after the close and GitHub refuses to draft a clo | **setup_required**: workflow state is `disabled_fork`, `disabled_manually` or `disabled_inactivity`, or the fork has no `ci-amd.yml` | `error` (red), link to the fork's Actions page for `ci-amd.yml` | create or update: how to enable, with the exact `gh repo sync <fork> --branch main` and `gh workflow enable ci-amd.yml -R <fork>` lines, and `breeze ci audit`; on transition from any other state but `closed` (the workflow stopped running after it worked): also post a **new** comment sayin [...] | **awaiting_push**: enabled, no run for the head SHA | `pending` (yellow), link to the fork's Actions page | on transition from setup_required: post a **new** comment asking to push again (an edit would not notify) | | **running**: run in progress | `pending` (yellow), link to the run | update state and run link | -| **failed**: conclusion is failure, cancelled, timed out or startup failure | `failure` (red), link to the run; for a startup failure the description says the run could not start | update with the run link; for a startup failure the text says no job ran and points to the fork's Actions settings | +| **failed**: conclusion is failure, cancelled, timed out or startup failure | `failure` (red), link to the run; for a startup failure the description says the run could not start | update with the run link and the jobs that did not pass; for a startup failure the text says no job ran and points to the fork's Actions settings | | **green**: conclusion success and the pull request does not change CI-defining files | `success` (green), link to the run | update: ready for review, run link; on the author's draft: mark it ready for review when it is | | **green_unverified**: conclusion success, but the pull request changes CI-defining files (see "Trusting a green fork run") | `pending` (yellow), link to the run | update: a maintainer must run project CI; on the author's draft: mark it ready for review, then a maintainer runs project CI | | **closed**: setup_required or awaiting_push for `close_after_days` | unchanged | close with a new comment; reopening restarts the clock | @@ -873,6 +873,16 @@ are no job statuses either, and GitHub's error is only on the run page. It is ha but `Fork CI / Tests (AMD)` says "Fork CI run could not start - check your fork's Actions settings", and the state comment explains what to check instead of asking to fix failures. +For a failed run that started, the state comment lists the jobs that did not pass, each linking +to the job in the fork, at most 10 of them; the job statuses list them all. The jobs come from +`GET /repos/<fork>/actions/runs/<id>/jobs?filter=latest`, reused from the job statuses when they +read them in the same run, so a failed pull request costs one jobs read per reconciler run. Job +names come from the contributor's workflow, so each is shown in a code span with backticks and +line breaks replaced. When the jobs cannot be read the state comment is left as it is until the +next run, rather than dropping the list. A read that failed after its retries is not repeated in +the same run, whether the job statuses or the state comment made it, so persistent server errors +cost each failed pull request one sequence of retries a run, not two. + ### Commit status The reconciler posts a commit status with context `Fork CI / Tests (AMD)` on the pull diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py b/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py index c2682485f13..20212ad8fd7 100644 --- a/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py +++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py @@ -30,8 +30,10 @@ from __future__ import annotations import json import re +from collections.abc import Sequence from dataclasses import dataclass from datetime import datetime +from typing import NamedTuple STATUS_CONTEXT = "Fork CI / Tests (AMD)" # Each job of the fork run is reported under this prefix. Job names come from the contributor's @@ -189,6 +191,37 @@ STARTUP_FAILURE_TEXT = ( ) +class FailedJob(NamedTuple): + """A job of a failed fork run that did not pass, as listed in the state comment.""" + + name: str + url: str + conclusion: str + + +# A full-matrix run can fail in dozens of jobs; the job statuses list them all. +MAX_LISTED_FAILED_JOBS = 10 +_FAILED_JOB_OUTCOMES = { + "failure": "failed", + "timed_out": "timed out", + "cancelled": "cancelled", + "startup_failure": "could not start", +} + + +def build_failed_job_list(failed_jobs: Sequence[FailedJob]) -> str: + listed = [ + # Job names come from the contributor's workflow; a code span keeps them from posting + # Markdown, including mentions, as the bot. + f"- [`{_sanitize_path(job.name[:JOB_NAME_MAX_LENGTH])}`]({job.url}): " + f"{_FAILED_JOB_OUTCOMES.get(job.conclusion, 'failed')}" + for job in failed_jobs[:MAX_LISTED_FAILED_JOBS] + ] + if len(failed_jobs) > MAX_LISTED_FAILED_JOBS: + listed.append(f"- and {len(failed_jobs) - MAX_LISTED_FAILED_JOBS} more, see the run") + return "Jobs that did not pass:\n\n" + "\n".join(listed) + + def behind_warning(behind: int, *, base_ref: str) -> str: return ( f"**This branch is {behind} commits behind `{base_ref}`.** Fork CI tests the branch as it is, " @@ -205,8 +238,12 @@ def state_comment( base_ref: str = "main", author_draft: bool = False, startup_failure: bool = False, + failed_jobs: Sequence[FailedJob] = (), ) -> str: - """:param startup_failure: the fork run of a failed state did not start, see ``STARTUP_FAILURE_TEXT``.""" + """ + :param startup_failure: the fork run of a failed state did not start, see ``STARTUP_FAILURE_TEXT``. + :param failed_jobs: the jobs of a failed state's fork run that did not pass. + """ text = _AUTHOR_DRAFT_TEXTS.get(marker.state) if author_draft else None if startup_failure: text = STARTUP_FAILURE_TEXT @@ -219,6 +256,8 @@ def state_comment( parts.append(behind_warning(marker.behind, base_ref=base_ref)) if run_url: parts.append(f"Latest run: {run_url}") + if marker.state == "failed" and failed_jobs: + parts.append(build_failed_job_list(failed_jobs)) if marker.state == "setup_required": parts.append(_setup_instructions(fork)) parts.append(f"How this works: [CI in your fork]({CONTRIBUTING_URL}).") 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 7cc8419e022..1e6c9e0e211 100644 --- a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py +++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py @@ -69,6 +69,7 @@ from airflow_breeze.utils.fork_ci_messages import ( STALE_JOB_STATUS, STATUS_CONTEXT, STATUS_STATES, + FailedJob, Marker, fork_actions_url, gate_modified_close_comment, @@ -618,6 +619,7 @@ def classify( jobs_reported: str | None = None, head_shared_across_bases: bool = False, find_gate_change: Callable[[], str | None] = lambda: None, + find_failed_jobs: Callable[[], list[FailedJob] | None] = lambda: [], ) -> Plan: """ Decide the mutations for one gated pull request. @@ -639,6 +641,9 @@ def classify( :param find_gate_change: describes how the pull request changes the fork CI gate, or returns None; only called for pull requests that the gate tampering check applies to, once per head commit. + :param find_failed_jobs: the jobs of the fork run that did not pass, None when they cannot be + read; only called for a failed run that started. On None the state comment is left for the + next run, so that a failed read does not drop the list from it. """ plan = Plan() labels = set(pr.labels) @@ -828,6 +833,9 @@ def classify( _set_ci_passed(plan, pr, passed=state == ForkState.GREEN, target_url=target_url) _set_ci_run_started(plan, pr, fork) green = state in (ForkState.GREEN, ForkState.GREEN_UNVERIFIED) + failed_jobs = find_failed_jobs() if state == ForkState.FAILED and not startup_failure else [] + if failed_jobs is None: + return plan _set_marker_comment( plan, pr, @@ -845,6 +853,7 @@ def classify( base_ref=pr.base_ref, author_draft=_is_author_draft(pr), startup_failure=startup_failure, + failed_jobs=failed_jobs, ), ) return plan @@ -1445,6 +1454,48 @@ def build_job_statuses(jobs: Iterable[dict[str, Any]]) -> dict[str, tuple[str, s return statuses +def find_failed_jobs(jobs: Iterable[dict[str, Any]]) -> list[FailedJob]: + """The finished jobs of a fork run that did not pass, by name, leaving out the jobs every run has.""" + failed: dict[str, FailedJob] = {} + for job in jobs: + if job["name"] in UNREPORTED_JOB_NAMES or job.get("status") in RUNNING_JOB_STATUSES: + continue + conclusion = job.get("conclusion") or "" + if JOB_STATUSES.get(conclusion, ("",))[0] == "failure": + failed.setdefault(job["name"], FailedJob(job["name"], job["html_url"], conclusion)) + return sorted(failed.values()) + + +# The jobs of fork runs read in this reconciler run, by run id. None records a read that failed after +# its retries, so the next reader gives up at once instead of repeating them; the next run reads again. +ReadJobs = dict[int, list[dict[str, Any]] | None] + + +def read_failed_jobs( + client: GitHubClient, + pr: PullRequestSnapshot, + fork: ForkObservation, + read_jobs: ReadJobs, +) -> list[FailedJob] | None: + """ + ``find_failed_jobs`` of the pull request's fork run; None when its jobs cannot be read. + + :param read_jobs: the jobs already read in this reconciler run; a read made here is recorded too. + """ + if fork.run_id is None or not pr.head_repo: + return [] + if fork.run_id in read_jobs: + jobs = read_jobs[fork.run_id] + else: + try: + jobs = client.get_run_jobs(pr.head_repo, fork.run_id) + except GitHubError as e: + console_print(f"[warning]#{pr.number}: could not read the jobs of {fork.run_url}: {e}[/]") + jobs = None + read_jobs[fork.run_id] = jobs + return None if jobs is None else find_failed_jobs(jobs) + + class JobStatusReporter: """Reports each job of a pull request's fork run as a commit status, within a per-run budget.""" @@ -1455,12 +1506,17 @@ class JobStatusReporter: repo: str, budget: int = MAX_JOB_STATUSES_PER_RUN, now: datetime | None = None, + read_jobs: ReadJobs | None = None, ): - """:param now: when given, deferring the statuses of a run that finished long ago is reported.""" + """ + :param now: when given, deferring the statuses of a run that finished long ago is reported. + :param read_jobs: where the jobs read, or a failed read, are kept, for the state comment to reuse. + """ self._client = client self._repo = repo self._now = now self.remaining = budget + self.read_jobs = {} if read_jobs is None else read_jobs self.deferred: list[int] = [] # (pull request, attention item) for runs whose statuses are deferred long after they finished. self.delayed: list[tuple[int, str]] = [] @@ -1505,7 +1561,12 @@ class JobStatusReporter: if self.remaining == 0: self._defer(pr, fork) return None - jobs = self._client.get_run_jobs(pr.head_repo, fork.run_id) + try: + jobs = self._client.get_run_jobs(pr.head_repo, fork.run_id) + except Exception: + self.read_jobs[fork.run_id] = None + raise + self.read_jobs[fork.run_id] = jobs # By the case-folded context, as GitHub compares them. current = { status["context"].casefold(): ( @@ -2360,8 +2421,13 @@ def run_reconcile( ci_run_started_note = ( f"reporting `{CI_RUN_STARTED_CONTEXT}` for project CI failed: {type(e).__name__}" ) + read_jobs: ReadJobs = {} job_reporter = JobStatusReporter( - client, repo=repo, budget=max(MAX_JOB_STATUSES_PER_RUN - ci_run_started_writes, 0), now=now + client, + repo=repo, + budget=max(MAX_JOB_STATUSES_PER_RUN - ci_run_started_writes, 0), + now=now, + read_jobs=read_jobs, ) previous_job_status_failure_runs = read_job_status_failure_runs(reported_file) @@ -2411,6 +2477,7 @@ def run_reconcile( jobs_reported=jobs_reported, head_shared_across_bases=pr.head_sha in shared_heads, find_gate_change=functools.partial(gate_change_finder.find, pr), + find_failed_jobs=functools.partial(read_failed_jobs, client, pr, fork, read_jobs), ) draft_change = apply_plan(client, pr, plan) if draft_change is not None: diff --git a/dev/breeze/tests/test_fork_ci_messages.py b/dev/breeze/tests/test_fork_ci_messages.py index ccc12a48ce3..bbd3b03c568 100644 --- a/dev/breeze/tests/test_fork_ci_messages.py +++ b/dev/breeze/tests/test_fork_ci_messages.py @@ -36,6 +36,8 @@ from airflow_breeze.utils.fork_ci_messages import ( JOB_NAME_MAX_LENGTH, JOB_STATUS_CONTEXT_PREFIX, MAX_LISTED_CI_FILES, + MAX_LISTED_FAILED_JOBS, + FailedJob, Marker, ci_change_approval_comment, ci_modification_note_comment, @@ -120,6 +122,35 @@ def test_state_comment_links_the_contributor_documentation(startup_failure): assert text.endswith(FOOTER) +def test_failed_job_list_is_safe_markdown_and_capped(): + jobs = [ + FailedJob("Tests `evil`\n\n@someone", "https://job/0";, "timed_out"), + *( + FailedJob(f"Job {i}", f"https://job/{i + 1}", "failure") + for i in range(MAX_LISTED_FAILED_JOBS + 1) + ), + ] + + text = state_comment( + Marker("failed", "a", MARKER.since), fork="u/airflow", run_url=None, failed_jobs=jobs + ) + + assert "- [`Tests ?evil???@someone`](https://job/0): timed out" in text + assert "\n@someone" not in text + assert "- [`Job 0`](https://job/1): failed" in text + assert "- and 2 more, see the run" in text + + +def test_failed_jobs_are_listed_only_for_a_failed_state(): + jobs = [FailedJob("Static checks", "https://job/1";, "failure")] + + text = state_comment( + Marker("running", "a", MARKER.since), fork="u/airflow", run_url=None, failed_jobs=jobs + ) + + assert "Static checks" not in text + + def test_parse_marker_absent(): assert parse_marker("just a comment") is None diff --git a/dev/breeze/tests/test_fork_ci_reconcile.py b/dev/breeze/tests/test_fork_ci_reconcile.py index 441901e4feb..140078f5dda 100644 --- a/dev/breeze/tests/test_fork_ci_reconcile.py +++ b/dev/breeze/tests/test_fork_ci_reconcile.py @@ -36,6 +36,7 @@ from airflow_breeze.utils.fork_ci_messages import ( JOB_STATUS_CONTEXT_PREFIX, STALE_JOB_STATUS, STATUS_CONTEXT, + FailedJob, Marker, parse_marker, render_marker, @@ -68,6 +69,7 @@ from airflow_breeze.utils.fork_ci_reconcile import ( fetch_open_pull_requests, fetch_workflow_state, find_ci_run_since, + find_failed_jobs, find_heads_shared_across_bases, find_project_ci_run_url, find_state_comment, @@ -81,6 +83,7 @@ from airflow_breeze.utils.fork_ci_reconcile import ( parse_pull_request, pull_request_changes_ci_files, read_ci_run_at, + read_failed_jobs, read_head_status, read_job_status_failure_runs, read_project_ci_run_url, @@ -2709,6 +2712,9 @@ def test_run_reconcile_reports_job_statuses_before_classifying( (pr, fork), _ = mock_reporter.return_value.report.call_args assert pr.number == 1 assert mock_classify.call_args.kwargs["jobs_reported"] == "7:SUCCESS" + find_failed = mock_classify.call_args.kwargs["find_failed_jobs"] + assert find_failed.func is read_failed_jobs + assert find_failed.args == (client, pr, fork, mock_reporter.call_args.kwargs["read_jobs"]) @pytest.mark.parametrize("status", ["queued", "in_progress", "waiting"]) @@ -3562,7 +3568,7 @@ def test_run_reconcile_marks_older_pilot_pull_requests_and_reports_project_ci_ru assert mock_classify.call_args.args[0].ci_run_at == datetime(2026, 11, 2, tzinfo=UTC) mock_report.assert_called_once_with(client, repo=REPO, skip={1}) - mock_reporter.assert_called_once_with(client, repo=REPO, budget=job_status_budget, now=NOW) + mock_reporter.assert_called_once_with(client, repo=REPO, budget=job_status_budget, now=NOW, read_jobs={}) assert has_alerts is fails if fails: assert "reporting `CI run started` for project CI failed: GitHubError" in output.read_text() @@ -3699,3 +3705,114 @@ def test_run_reconcile_reports_delayed_job_statuses_as_attention_items( assert _reconcile_with_reporter(client, tmp_path) is True assert "job statuses of fork run R are still not all posted" in (tmp_path / "s.json").read_text() + + +def test_find_failed_jobs_lists_finished_jobs_that_did_not_pass(): + jobs = [ + _fork_job("Static checks"), + _fork_job("Postgres tests", conclusion="failure"), + _fork_job("Helm tests", status="in_progress", conclusion=None), + _fork_job("Docs", conclusion="timed_out"), + _fork_job("Java SDK tests", conclusion="skipped"), + _fork_job("Platform: AMD", conclusion="failure"), + _fork_job("Postgres tests", conclusion="cancelled", url="https://job/second";), + ] + + assert find_failed_jobs(jobs) == [ + FailedJob("Docs", "https://job/Docs";, "timed_out"), + FailedJob("Postgres tests", "https://job/Postgres tests", "failure"), + ] + + [email protected]( + ("read_jobs", "side_effect", "expected", "reads"), + [ + pytest.param({1: [_fork_job("A", conclusion="failure")]}, None, ["A"], False, id="read-this-run"), + pytest.param({}, [[_fork_job("B", conclusion="failure")]], ["B"], True, id="not-read-yet"), + pytest.param({}, GitHubError("HTTP 502"), None, True, id="unreadable"), + pytest.param({1: None}, None, None, False, id="read-failed-this-run"), + ], +) +def test_read_failed_jobs(client, read_jobs, side_effect, expected, reads): + client.get_run_jobs.side_effect = side_effect + + failed = read_failed_jobs(client, _pr(), COMPLETED_RUN, read_jobs) + + assert (None if failed is None else [job.name for job in failed]) == expected + assert client.get_run_jobs.called is reads + assert 1 in read_jobs + + +def test_a_failed_jobs_read_is_not_repeated_for_the_state_comment(client): + client.rest_get.return_value = [] + client.get_run_jobs.side_effect = GitHubError("HTTP 502") + read_jobs: dict = {} + reporter = JobStatusReporter(client, repo=REPO, budget=10, read_jobs=read_jobs) + + with pytest.raises(GitHubError): + reporter.report(_pr(), COMPLETED_RUN) + + assert read_failed_jobs(client, _pr(), COMPLETED_RUN, read_jobs) is None + client.get_run_jobs.assert_called_once_with("someone/airflow", 1) + + +def test_job_status_reporter_keeps_the_jobs_it_read(client): + jobs = [_fork_job("Static checks", conclusion="failure")] + reporter = JobStatusReporter(_reporter_client(client, jobs), repo=REPO, budget=10) + + reporter.report(_pr(), COMPLETED_RUN) + + assert reporter.read_jobs == {1: jobs} + + +FAILED_JOB = FailedJob("Static checks", "https://job/1";, "failure") + + [email protected]( + ("fork", "failed_jobs", "called", "listed"), + [ + pytest.param(COMPLETED_RUN, [FAILED_JOB], True, True, id="failed"), + pytest.param( + dataclasses.replace(COMPLETED_RUN, suite_conclusion="STARTUP_FAILURE"), + [FAILED_JOB], + False, + False, + id="startup-failure", + ), + pytest.param(RUNNING_RUN, [FAILED_JOB], False, False, id="running"), + ], +) +def test_state_comment_lists_the_jobs_of_a_failed_run(fork, failed_jobs, called, listed): + find_failed = mock.Mock(return_value=failed_jobs) + + plan = classify( + _pr(), + fork, + config=CONFIG, + now=NOW, + repo=REPO, + find_bypass_run=lambda: None, + changes_ci_files=lambda: False, + find_failed_jobs=find_failed, + ) + + assert find_failed.called is called + assert ("- [`Static checks`](https://job/1): failed" in plan.marker_comment) is listed + + +def test_state_comment_is_left_for_the_next_run_when_failed_jobs_cannot_be_read(): + plan = classify( + _pr(is_draft=False, labels=("fork ci required", "ready for maintainer review")), + COMPLETED_RUN, + config=CONFIG, + now=NOW, + repo=REPO, + find_bypass_run=lambda: None, + changes_ci_files=lambda: False, + find_failed_jobs=lambda: None, + ) + + assert plan.state == ForkState.FAILED + assert plan.status == ("failure", RUN_URL, "Fork CI failed - see Details") + assert plan.remove_labels == ["ready for maintainer review"] + assert plan.marker_comment is None
