shahar1 commented on code in PR #74822: URL: https://github.com/apache/airflow/pull/74822#discussion_r4367561222
########## dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py: ########## @@ -0,0 +1,1251 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +""" +Scheduled reconciler for pull requests gated to fork CI (AIP-120). + +``classify`` is a pure function from what GitHub recorded about one pull request to the mutations +that bring it to the desired state; ``run_reconcile`` gathers that data and applies the result. +See "Reconciler" in dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md. +""" + +from __future__ import annotations + +import base64 +import functools +import json +from collections.abc import Callable, Iterable, Sequence +from dataclasses import dataclass, field, replace +from datetime import datetime, timedelta +from enum import Enum +from pathlib import Path +from typing import Any +from urllib.parse import quote + +import yaml + +from airflow_breeze.utils.console import console_print +from airflow_breeze.utils.fork_ci_gate import ( + GATED_REASONS, + LABEL_FORK_CI_REQUIRED, + LABEL_NEEDS_PROJECT_CI, + LABEL_READY_FOR_REVIEW, + LABEL_USE_FORK_CI, + LABEL_USE_PROJECT_CI, + TAMPER_CHECKED_REASONS, + GateConfig, + GateEvent, + decide, + parse_config, + parse_github_datetime, +) +from airflow_breeze.utils.fork_ci_github import GitHubClient, GitHubError +from airflow_breeze.utils.fork_ci_messages import ( + CI_PASSED_CONTEXT, + CI_PASSED_DESCRIPTION, + CI_PASSED_REVOKED_DESCRIPTION, + STATUS_CONTEXT, + STATUS_STATES, + Marker, + fork_actions_url, + inactivity_close_comment, + nudge_comment, + parse_marker, + released_comment, + state_comment, + status_description, + tamper_close_comment, + ungated_comment, +) + +CI_WORKFLOW_FILE = "ci-amd.yml" +# The files the gate job of ci-amd.yml reads from a pull request's base branch. +GATE_CONFIG_PATH = ".github/fork-ci-gate.toml" +GATE_SCRIPT_PATH = "dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py" +GATE_JOB_NAME = "Fork CI gate" +# Jobs that also run, without needing the gate, in a run the gate left to fork CI: main's +# check-run-needed (no steps outside schedule and push runs) and the "CI passed" gate job of the +# merge ruleset, which reports under this name when project CI did not run. Like the gate, each +# must finish within GATE_MAX_DURATION, so project CI work moved under one of these names counts. +GATED_RUN_JOB_NAMES = frozenset({"Check if the run is needed", "CI passed (project CI not run)"}) +BOT_LOGINS = frozenset({"github-actions", "github-actions[bot]"}) +DISCOVERY_WINDOW = timedelta(hours=24) +PASS_TWO_PAGE_SIZE = 25 +FORK_BATCH_SIZE = 50 +# Project CI runs created this long ago or later are checked for gate tampering. Several hours +# overlap the 15-minute cadence, so a tampered run is seen even after the pull request was +# pushed again with the gate restored. +TAMPER_WINDOW = timedelta(hours=6) +# A ci-amd.yml run in which only the gate executed finishes in about a minute; only longer runs +# can have done project CI work and are worth listing jobs for. +SUSPICIOUS_RUN_DURATION = timedelta(minutes=5) +GATE_MAX_DURATION = timedelta(minutes=5) +# Paths whose content defines what fork CI runs. A pull request changing them controls its own +# fork CI result, so a green fork run cannot be trusted for it. +CI_DEFINING_PREFIXES = ( + ".github/", + "dev/breeze/", + "scripts/ci/", + "scripts/docker/", + "scripts/in_container/", + "Dockerfile", +) +CI_DEFINING_SUFFIXES = (".pre-commit-config.yaml",) +PULL_REQUEST_FILES_CAP = 3000 +COMPARE_FILES_CAP = 300 +SLACK_SECTION_LIMIT = 3000 +SLACK_MAX_BLOCKS = 50 +SLACK_ITEM_LIMIT = 500 +# Events whose runs use the workflow files of the pull request's merge commit, so a pull request +# can change what they run. pull_request_target and issue_comment use the base branch's files. +MERGE_REF_EVENTS = ("pull_request", "pull_request_review", "pull_request_review_comment") + +RUNNING_STATUSES = frozenset({"QUEUED", "IN_PROGRESS", "PENDING", "WAITING", "REQUESTED"}) +FAILED_CONCLUSIONS = frozenset({"FAILURE", "CANCELLED", "TIMED_OUT", "STARTUP_FAILURE", "ACTION_REQUIRED"}) +SETUP_WORKFLOW_STATES = frozenset({"disabled_fork", "disabled_manually", "missing"}) + + +class ForkState(str, Enum): + SETUP_REQUIRED = "setup_required" + AWAITING_PUSH = "awaiting_push" + RUNNING = "running" + FAILED = "failed" + GREEN = "green" + GREEN_UNVERIFIED = "green_unverified" + GATE_BYPASSED = "gate_bypassed" + CLOSED = "closed" + RELEASED = "released" + UNGATED = "ungated" + + +CLOSABLE_STATES = frozenset({ForkState.SETUP_REQUIRED, ForkState.AWAITING_PUSH}) + + +@dataclass(frozen=True) +class WorkflowRun: + """A run of a workflow in apache/airflow triggered by a pull_request event.""" + + id: int + path: str + html_url: str + head_sha: str + head_repo: str + head_branch: str + created_at: datetime + run_started_at: datetime + updated_at: datetime + status: str + event: str = "pull_request" + + @property + def workflow_file(self) -> str: + return self.path.rsplit("/", 1)[-1] + + def is_long_running(self, now: datetime) -> bool: + # Time spent waiting for a runner is not work, so a run that has not started is never + # long. A running one is judged as it goes: one tampered to outlast the tamper window + # would otherwise never be judged at all. + if self.status == "completed": + return self.updated_at - self.run_started_at >= SUSPICIOUS_RUN_DURATION + return self.status == "in_progress" and now - self.run_started_at >= SUSPICIOUS_RUN_DURATION + + +@dataclass(frozen=True) +class PullRequestSnapshot: + number: int + node_id: str + is_draft: bool + labels: tuple[str, ...] + author_login: str + author_association: str + created_at: datetime + head_sha: str + head_repo: str | None + base_ref: str + head_ref: str = "" + last_draft_actor: str | None = None + last_project_ci_unlabel_at: datetime | None = None + last_reopened_at: datetime | None = None + last_reopened_by_anyone_at: datetime | None = None + last_base_change_at: datetime | None = None + current_status: tuple[str, str, str] | None = None + status_created_at: datetime | None = None + ci_passed_state: str | None = None + marker: Marker | None = None + marker_comment_id: int | None = None + marker_comment_body: str | None = None + marker_error: str | None = None + # Set when the state comment was not among the comments fetched with the pull request and + # more comments follow; the reconciler pages from here before deciding anything. + comments_cursor: str | None = None + + +@dataclass(frozen=True) +class ForkObservation: + suite_status: str | None = None + suite_conclusion: str | None = None + run_url: str | None = None + workflow_state: str | None = None + error: str | None = None + + +@dataclass +class Plan: + state: ForkState | None = None + add_labels: list[str] = field(default_factory=list) + remove_labels: list[str] = field(default_factory=list) + draft: bool | None = None + status: tuple[str, str, str] | None = None + # (state, target URL, description) of the `CI passed` status. + ci_passed: tuple[str, str, str] | None = None + marker_comment: str | None = None + new_comments: list[str] = field(default_factory=list) + close: bool = False + attention: list[str] = field(default_factory=list) + unknown: list[str] = field(default_factory=list) + + +def _set_label(plan: Plan, labels: set[str], label: str, *, present: bool) -> None: + if present and label not in labels: + plan.add_labels.append(label) + elif not present and label in labels: + plan.remove_labels.append(label) + + +def _workflow_file(resource_path: str | None) -> str: + return (resource_path or "").rsplit("/", 1)[-1] + + +def _is_bot(login: str | None) -> bool: + return login in BOT_LOGINS + + +def observe_fork_state(fork: ForkObservation, marker: Marker | None) -> tuple[ForkState | None, str | None]: + """Map what the fork shows for the head commit to a state, or return the reason it cannot.""" + if fork.error: + return None, fork.error + if fork.suite_status is not None: + if fork.suite_status in RUNNING_STATUSES: + return ForkState.RUNNING, None + if fork.suite_status != "COMPLETED": + return None, f"unexpected fork check suite status {fork.suite_status!r}" + if fork.suite_conclusion == "SUCCESS": + return ForkState.GREEN, None + if fork.suite_conclusion in FAILED_CONCLUSIONS: + return ForkState.FAILED, None + return None, f"unexpected fork check suite conclusion {fork.suite_conclusion!r}" + if fork.workflow_state is None: + if marker is not None and marker.state == ForkState.AWAITING_PUSH.value: + return ForkState.AWAITING_PUSH, None + return None, "no fork run and the fork workflow state was not read" + if fork.workflow_state in SETUP_WORKFLOW_STATES: + return ForkState.SETUP_REQUIRED, None + if fork.workflow_state == "active": + return ForkState.AWAITING_PUSH, None + return None, f"unexpected fork workflow state {fork.workflow_state!r}" + + +def needs_workflow_state(marker: Marker | None) -> bool: + return marker is None or marker.state != ForkState.AWAITING_PUSH.value + + +STARTED_JOB_STATUSES = frozenset({"in_progress", "completed"}) + + +def jobs_show_project_ci(jobs: Iterable[dict[str, Any]], now: datetime) -> bool: + """ + Whether the jobs of a ci-amd.yml run show more than the gate deciding to skip. + + Any started job other than the gate and the ``GATED_RUN_JOB_NAMES`` jobs that was not skipped + counts, and so does a gate job that is missing or duplicated, or one of those jobs running for + longer than they ever need, which is how a renamed job doing project CI work under their name + shows up. Jobs that have not started do not count: a queued job reports its creation as + ``started_at``, so waiting for a runner would look like work. + """ + jobs = list(jobs) + gate_jobs = [job for job in jobs if job["name"] == GATE_JOB_NAME] + if len(gate_jobs) != 1: + return True + expected_names = GATED_RUN_JOB_NAMES | {GATE_JOB_NAME} + for job in jobs: + if job["name"] not in expected_names or job.get("status") not in STARTED_JOB_STATUSES: + continue + started = parse_github_datetime(job.get("started_at") or "") + completed = parse_github_datetime(job.get("completed_at") or "") or now + if started is not None and completed - started > GATE_MAX_DURATION: + return True + return any( + job["name"] not in expected_names + and job.get("status") in STARTED_JOB_STATUSES + and job.get("conclusion") != "skipped" + for job in jobs + ) + + +def touches(files: Iterable[str], prefixes: tuple[str, ...]) -> bool: + return any(path.startswith(prefixes) for path in files) + + +def changed_paths(entries: Iterable[dict[str, Any]]) -> list[str]: + """ + Every path a changed-file listing touches. + + A rename lists only its new path as ``filename``; moving a file out of a CI-defining directory + changes that directory too, so the old path counts as well. + """ + paths = [] + for entry in entries: + paths.append(entry["filename"]) + if entry.get("previous_filename"): + paths.append(entry["previous_filename"]) + return paths + + +def changes_ci_definition(files: Sequence[str], *, file_count: int | None = None) -> bool: + """ + Whether a pull request's changed files can alter what its fork CI runs. + + :param file_count: entries in the listing, when ``files`` also holds the old paths of renames. + """ + # GitHub lists at most 3000 files of a pull request; a longer one is not trusted. + return ( + (len(files) if file_count is None else file_count) >= PULL_REQUEST_FILES_CAP + or touches(files, CI_DEFINING_PREFIXES) + or any(path.endswith(CI_DEFINING_SUFFIXES) for path in files) + ) + + +def _gate_event(pr: PullRequestSnapshot, repo: str) -> GateEvent: + return GateEvent( + event_name="pull_request", + repository=repo, + ref="", + head_repository=pr.head_repo or "", + author_login=pr.author_login, + author_association=pr.author_association, + created_at=pr.created_at, + labels=pr.labels, + ) + + +def status_state(status: tuple[str, str, str] | None) -> str | None: + """The state the reconciler recorded in a commit status it posted, if it posted it.""" + if status is None: + return None + return next( + ( + state + for state, github_state in STATUS_STATES.items() + if (github_state, status_description(state)) == (status[0], status[2]) + ), + None, + ) + + +def _set_status( + plan: Plan, pr: PullRequestSnapshot, state: str, target_url: str, *, force: bool = False +) -> None: + desired = (STATUS_STATES[state], target_url, status_description(state)) + if force or desired != pr.current_status: + plan.status = desired + + +def _set_ci_passed(plan: Plan, pr: PullRequestSnapshot, *, passed: bool, target_url: str) -> None: + """ + Report fork CI to the `CI passed` check the merge ruleset requires. + + Only success is ever posted: a gated pull request without the status cannot merge, which is + what a pending or failing one would say. A success that no longer holds for the head commit, + such as a failed re-run in the fork, is overwritten so the pull request cannot merge on it. + """ + if passed and pr.ci_passed_state != "success": + plan.ci_passed = ("success", target_url, CI_PASSED_DESCRIPTION) + elif not passed and pr.ci_passed_state == "success": + plan.ci_passed = ("pending", target_url, CI_PASSED_REVOKED_DESCRIPTION) + + +def _set_marker_comment(plan: Plan, pr: PullRequestSnapshot, body: str) -> None: + if body != pr.marker_comment_body: + plan.marker_comment = body + + +def classify( + pr: PullRequestSnapshot, + fork: ForkObservation, + *, + config: GateConfig | None, + now: datetime, + repo: str, + find_bypass_run: Callable[[], str | None], + changes_ci_files: Callable[[], bool | None], + count_behind: Callable[[], int | None] = lambda: None, +) -> Plan: + """ + Decide the mutations for one gated pull request. + + :param config: the gate configuration of the pull request's base branch; None when the base + branch has no gate, so its gate job runs project CI. + :param find_bypass_run: returns the project CI run that bypassed the gate, if any; only + called for pull requests that the gate tampering check applies to. + :param changes_ci_files: whether the pull request changes CI-defining paths, None when that + cannot be told for the snapshot's head commit; only called when a new head commit turns + green in the fork. On None nothing is changed, since the decision would be recorded on a + commit the answer may not belong to. + :param count_behind: how many commits the head commit is behind the base branch, None when + unknown; called at the same time as ``changes_ci_files``. + """ + plan = Plan() + labels = set(pr.labels) + marker = pr.marker + if pr.marker_error: + plan.unknown.append(pr.marker_error) + marker = None + + def release(state: ForkState, comment: Callable[[Marker], str]) -> Plan: + plan.state = state + plan.remove_labels.append(LABEL_FORK_CI_REQUIRED) + _set_label(plan, labels, LABEL_NEEDS_PROJECT_CI, present=False) + if pr.is_draft and _is_bot(pr.last_draft_actor): + plan.draft = False + since = marker.since if marker and marker.state == state.value else now + _set_marker_comment(plan, pr, comment(Marker(state.value, pr.head_sha, since))) + _set_status(plan, pr, state.value, f"https://github.com/{repo}/pull/{pr.number}/checks";) + return plan + + if LABEL_USE_PROJECT_CI in labels: + if LABEL_USE_FORK_CI in labels: + plan.attention.append("carries both `use project ci` and `use fork ci`; project CI wins") + return release(ForkState.RELEASED, released_comment) + + if config is None: + return release(ForkState.UNGATED, functools.partial(ungated_comment, base_ref=pr.base_ref)) + + if pr.head_repo is None: + plan.unknown.append("head repository no longer exists") + return plan + + reason = decide(_gate_event(pr, repo), config).reason + if reason in TAMPER_CHECKED_REASONS: + bypass_run = find_bypass_run() + if bypass_run is not None: + plan.state = ForkState.GATE_BYPASSED + if not pr.is_draft: + plan.draft = True + if LABEL_READY_FOR_REVIEW in labels: + plan.remove_labels.append(LABEL_READY_FOR_REVIEW) + _set_label(plan, labels, LABEL_NEEDS_PROJECT_CI, present=False) + _set_status(plan, pr, ForkState.GATE_BYPASSED.value, bypass_run) + _set_marker_comment( + plan, + pr, + state_comment( + Marker(ForkState.GATE_BYPASSED.value, pr.head_sha, now), + fork=pr.head_repo, + run_url=bypass_run, + ), + ) + plan.new_comments.append(tamper_close_comment(run_url=bypass_run)) + plan.close = True + plan.attention.append(f"closed for gate tampering (author @{pr.author_login}, run {bypass_run})") + return plan + + # A closing state in the marker of an open pull request that was not reopened since means that + # posting the explanation or closing failed on an earlier run: finish the closure. Only a + # maintainer's reopen lifts a closure for gate tampering, as for the tampering check itself. + reopened_since_closing = { + ForkState.CLOSED.value: pr.last_reopened_by_anyone_at, + ForkState.GATE_BYPASSED.value: pr.last_reopened_at, + } + if marker and marker.state in reopened_since_closing: + reopened = reopened_since_closing[marker.state] + if reopened is None or reopened <= marker.since: + plan.state = ForkState(marker.state) + _set_label(plan, labels, LABEL_NEEDS_PROJECT_CI, present=False) + if plan.state == ForkState.CLOSED: + comment = inactivity_close_comment(days=config.close_after_days, fork=pr.head_repo) + else: + # The status written with the marker links the run that bypassed the gate. + bypass_url = ( + pr.current_status[1] + if pr.current_status and status_state(pr.current_status) == ForkState.GATE_BYPASSED.value + else f"https://github.com/{repo}/pull/{pr.number}/checks"; + ) + comment = tamper_close_comment(run_url=bypass_url) + plan.new_comments.append(comment) + plan.close = True + return plan + + state, unknown_reason = observe_fork_state(fork, marker) + if state is None: + plan.unknown.append(unknown_reason or "unknown fork state") + return plan + refresh_status = False + behind = marker.behind if marker and marker.sha == pr.head_sha else None + if state == ForkState.GREEN: + # The decision is read back from the commit status on the head commit, which only + # apache/airflow's own tokens can write, never from a comment. The changed files depend + # on the base too, so a status written before the pull request was retargeted is stale. + # The status is shared by every pull request with this head commit; scoping it to one is + # tracked at https://github.com/apache/airflow/issues/74907 + previous = status_state(pr.current_status) + decided = previous in (ForkState.GREEN.value, ForkState.GREEN_UNVERIFIED.value) + stale = pr.last_base_change_at is not None and ( + pr.status_created_at is None or pr.status_created_at <= pr.last_base_change_at + ) + if decided and not stale: + unverified = previous == ForkState.GREEN_UNVERIFIED.value + else: + # Writing the status again, even unchanged, records when this decision was made. + refresh_status = decided + # The `needs project ci` label lists these pull requests for maintainers. + changes = changes_ci_files() + if changes is None: + return Plan() + unverified = changes + behind = count_behind() + if unverified: + state = ForkState.GREEN_UNVERIFIED + plan.state = state + since = marker.since if marker and marker.state == state.value else now + + # The marker records awaiting_push before the nudge is posted, so a failed post is not retried; + # tracked at https://github.com/apache/airflow/issues/74908 + if marker and marker.state == ForkState.SETUP_REQUIRED.value and state == ForkState.AWAITING_PUSH: + plan.new_comments.append(nudge_comment(fork=pr.head_repo)) + + if state in CLOSABLE_STATES and now - since >= timedelta(days=config.close_after_days): + plan.state = ForkState.CLOSED + _set_label(plan, labels, LABEL_NEEDS_PROJECT_CI, present=False) + _set_marker_comment( + plan, + pr, + state_comment(Marker(ForkState.CLOSED.value, pr.head_sha, now), fork=pr.head_repo, run_url=None), + ) + plan.new_comments.append(inactivity_close_comment(days=config.close_after_days, fork=pr.head_repo)) + plan.close = True + return plan + + if state == ForkState.GREEN: + if pr.is_draft and _is_bot(pr.last_draft_actor): + plan.draft = False + ready = not pr.is_draft or plan.draft is False + if ready and LABEL_READY_FOR_REVIEW not in labels: + plan.add_labels.append(LABEL_READY_FOR_REVIEW) + else: + if not pr.is_draft: + plan.draft = True + if LABEL_READY_FOR_REVIEW in labels: + plan.remove_labels.append(LABEL_READY_FOR_REVIEW) + # Lets maintainers list the pull requests that only wait for them to run project CI, which + # otherwise look like any other draft with a pending fork CI status. + _set_label(plan, labels, LABEL_NEEDS_PROJECT_CI, present=state == ForkState.GREEN_UNVERIFIED) + + target_url = fork.run_url or fork_actions_url(pr.head_repo) + _set_status(plan, pr, state.value, target_url, force=refresh_status) + _set_ci_passed(plan, pr, passed=state == ForkState.GREEN, target_url=target_url) + green = state in (ForkState.GREEN, ForkState.GREEN_UNVERIFIED) + _set_marker_comment( + plan, + pr, + state_comment( + Marker(state.value, pr.head_sha, since, behind=behind if green else None), + fork=pr.head_repo, + run_url=fork.run_url, + base_ref=pr.base_ref, + ), + ) + return plan + + +# --------------------------------------------------------------------------------------------- +# GitHub data gathering +# --------------------------------------------------------------------------------------------- + +DISCOVERY_QUERY = """ +query($q: String!, $cursor: String) { + search(type: ISSUE, first: 100, after: $cursor, query: $q) { + pageInfo { hasNextPage endCursor } + nodes { ... on PullRequest { + number createdAt authorAssociation baseRefName + author { login } + headRepository { nameWithOwner } + labels(first: 50) { nodes { name } } + } } + } +} +""" + +PASS_TWO_QUERY = """ +query($q: String!, $cursor: String) { + search(type: ISSUE, first: %(page)d, after: $cursor, query: $q) { + pageInfo { hasNextPage endCursor } + nodes { ... on PullRequest { + number id isDraft createdAt authorAssociation baseRefName headRefName headRefOid + author { login } + headRepository { nameWithOwner } + labels(first: 50) { nodes { name } } + timelineItems( + last: 100, + itemTypes: [CONVERT_TO_DRAFT_EVENT, UNLABELED_EVENT, REOPENED_EVENT, BASE_REF_CHANGED_EVENT] + ) { + nodes { + __typename + ... on ConvertToDraftEvent { createdAt actor { login } } + ... on UnlabeledEvent { createdAt label { name } } + ... on ReopenedEvent { createdAt actor { login } } + ... on BaseRefChangedEvent { createdAt } + } + } + commits(last: 1) { nodes { commit { Review Comment: `current_status`, `status_created_at` and `ci_passed_state` are read from `commits(last: 1)`, but the mutations go to `headRefOid`. As far as I know, GitHub orders a PR's commit list by date rather than by topology. After a rebase that keeps old dates, the last listed commit can then be something other than the head. In that case the state the reconciler reads belongs to a different commit than the one it writes. The likely result is a status reposted every run. The one that matters: a `CI passed: success` on the head is not revoked when a fork re-run of the same commit fails. I have not verified the ordering, so treat this as a question. **Suggestion:** Add `oid` to the selection and fall back to unknown (or skip the status shortcut) when it differs from `headRefOid`, or read the status through `headRef { target { ... on Commit { status { ... } } } }`. ########## dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py: ########## @@ -0,0 +1,1251 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +""" +Scheduled reconciler for pull requests gated to fork CI (AIP-120). + +``classify`` is a pure function from what GitHub recorded about one pull request to the mutations +that bring it to the desired state; ``run_reconcile`` gathers that data and applies the result. +See "Reconciler" in dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md. +""" + +from __future__ import annotations + +import base64 +import functools +import json +from collections.abc import Callable, Iterable, Sequence +from dataclasses import dataclass, field, replace +from datetime import datetime, timedelta +from enum import Enum +from pathlib import Path +from typing import Any +from urllib.parse import quote + +import yaml + +from airflow_breeze.utils.console import console_print +from airflow_breeze.utils.fork_ci_gate import ( + GATED_REASONS, + LABEL_FORK_CI_REQUIRED, + LABEL_NEEDS_PROJECT_CI, + LABEL_READY_FOR_REVIEW, + LABEL_USE_FORK_CI, + LABEL_USE_PROJECT_CI, + TAMPER_CHECKED_REASONS, + GateConfig, + GateEvent, + decide, + parse_config, + parse_github_datetime, +) +from airflow_breeze.utils.fork_ci_github import GitHubClient, GitHubError +from airflow_breeze.utils.fork_ci_messages import ( + CI_PASSED_CONTEXT, + CI_PASSED_DESCRIPTION, + CI_PASSED_REVOKED_DESCRIPTION, + STATUS_CONTEXT, + STATUS_STATES, + Marker, + fork_actions_url, + inactivity_close_comment, + nudge_comment, + parse_marker, + released_comment, + state_comment, + status_description, + tamper_close_comment, + ungated_comment, +) + +CI_WORKFLOW_FILE = "ci-amd.yml" +# The files the gate job of ci-amd.yml reads from a pull request's base branch. +GATE_CONFIG_PATH = ".github/fork-ci-gate.toml" +GATE_SCRIPT_PATH = "dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py" +GATE_JOB_NAME = "Fork CI gate" +# Jobs that also run, without needing the gate, in a run the gate left to fork CI: main's +# check-run-needed (no steps outside schedule and push runs) and the "CI passed" gate job of the +# merge ruleset, which reports under this name when project CI did not run. Like the gate, each +# must finish within GATE_MAX_DURATION, so project CI work moved under one of these names counts. +GATED_RUN_JOB_NAMES = frozenset({"Check if the run is needed", "CI passed (project CI not run)"}) +BOT_LOGINS = frozenset({"github-actions", "github-actions[bot]"}) +DISCOVERY_WINDOW = timedelta(hours=24) +PASS_TWO_PAGE_SIZE = 25 +FORK_BATCH_SIZE = 50 +# Project CI runs created this long ago or later are checked for gate tampering. Several hours +# overlap the 15-minute cadence, so a tampered run is seen even after the pull request was +# pushed again with the gate restored. +TAMPER_WINDOW = timedelta(hours=6) +# A ci-amd.yml run in which only the gate executed finishes in about a minute; only longer runs +# can have done project CI work and are worth listing jobs for. +SUSPICIOUS_RUN_DURATION = timedelta(minutes=5) +GATE_MAX_DURATION = timedelta(minutes=5) +# Paths whose content defines what fork CI runs. A pull request changing them controls its own +# fork CI result, so a green fork run cannot be trusted for it. +CI_DEFINING_PREFIXES = ( + ".github/", + "dev/breeze/", + "scripts/ci/", + "scripts/docker/", + "scripts/in_container/", + "Dockerfile", +) +CI_DEFINING_SUFFIXES = (".pre-commit-config.yaml",) +PULL_REQUEST_FILES_CAP = 3000 +COMPARE_FILES_CAP = 300 +SLACK_SECTION_LIMIT = 3000 +SLACK_MAX_BLOCKS = 50 +SLACK_ITEM_LIMIT = 500 +# Events whose runs use the workflow files of the pull request's merge commit, so a pull request +# can change what they run. pull_request_target and issue_comment use the base branch's files. +MERGE_REF_EVENTS = ("pull_request", "pull_request_review", "pull_request_review_comment") + +RUNNING_STATUSES = frozenset({"QUEUED", "IN_PROGRESS", "PENDING", "WAITING", "REQUESTED"}) +FAILED_CONCLUSIONS = frozenset({"FAILURE", "CANCELLED", "TIMED_OUT", "STARTUP_FAILURE", "ACTION_REQUIRED"}) +SETUP_WORKFLOW_STATES = frozenset({"disabled_fork", "disabled_manually", "missing"}) + + +class ForkState(str, Enum): + SETUP_REQUIRED = "setup_required" + AWAITING_PUSH = "awaiting_push" + RUNNING = "running" + FAILED = "failed" + GREEN = "green" + GREEN_UNVERIFIED = "green_unverified" + GATE_BYPASSED = "gate_bypassed" + CLOSED = "closed" + RELEASED = "released" + UNGATED = "ungated" + + +CLOSABLE_STATES = frozenset({ForkState.SETUP_REQUIRED, ForkState.AWAITING_PUSH}) + + +@dataclass(frozen=True) +class WorkflowRun: + """A run of a workflow in apache/airflow triggered by a pull_request event.""" + + id: int + path: str + html_url: str + head_sha: str + head_repo: str + head_branch: str + created_at: datetime + run_started_at: datetime + updated_at: datetime + status: str + event: str = "pull_request" + + @property + def workflow_file(self) -> str: + return self.path.rsplit("/", 1)[-1] + + def is_long_running(self, now: datetime) -> bool: + # Time spent waiting for a runner is not work, so a run that has not started is never + # long. A running one is judged as it goes: one tampered to outlast the tamper window + # would otherwise never be judged at all. + if self.status == "completed": + return self.updated_at - self.run_started_at >= SUSPICIOUS_RUN_DURATION + return self.status == "in_progress" and now - self.run_started_at >= SUSPICIOUS_RUN_DURATION + + +@dataclass(frozen=True) +class PullRequestSnapshot: + number: int + node_id: str + is_draft: bool + labels: tuple[str, ...] + author_login: str + author_association: str + created_at: datetime + head_sha: str + head_repo: str | None + base_ref: str + head_ref: str = "" + last_draft_actor: str | None = None + last_project_ci_unlabel_at: datetime | None = None + last_reopened_at: datetime | None = None + last_reopened_by_anyone_at: datetime | None = None + last_base_change_at: datetime | None = None + current_status: tuple[str, str, str] | None = None + status_created_at: datetime | None = None + ci_passed_state: str | None = None + marker: Marker | None = None + marker_comment_id: int | None = None + marker_comment_body: str | None = None + marker_error: str | None = None + # Set when the state comment was not among the comments fetched with the pull request and + # more comments follow; the reconciler pages from here before deciding anything. + comments_cursor: str | None = None + + +@dataclass(frozen=True) +class ForkObservation: + suite_status: str | None = None + suite_conclusion: str | None = None + run_url: str | None = None + workflow_state: str | None = None + error: str | None = None + + +@dataclass +class Plan: + state: ForkState | None = None + add_labels: list[str] = field(default_factory=list) + remove_labels: list[str] = field(default_factory=list) + draft: bool | None = None + status: tuple[str, str, str] | None = None + # (state, target URL, description) of the `CI passed` status. + ci_passed: tuple[str, str, str] | None = None + marker_comment: str | None = None + new_comments: list[str] = field(default_factory=list) + close: bool = False + attention: list[str] = field(default_factory=list) + unknown: list[str] = field(default_factory=list) + + +def _set_label(plan: Plan, labels: set[str], label: str, *, present: bool) -> None: + if present and label not in labels: + plan.add_labels.append(label) + elif not present and label in labels: + plan.remove_labels.append(label) + + +def _workflow_file(resource_path: str | None) -> str: + return (resource_path or "").rsplit("/", 1)[-1] + + +def _is_bot(login: str | None) -> bool: + return login in BOT_LOGINS + + +def observe_fork_state(fork: ForkObservation, marker: Marker | None) -> tuple[ForkState | None, str | None]: + """Map what the fork shows for the head commit to a state, or return the reason it cannot.""" + if fork.error: + return None, fork.error + if fork.suite_status is not None: + if fork.suite_status in RUNNING_STATUSES: + return ForkState.RUNNING, None + if fork.suite_status != "COMPLETED": + return None, f"unexpected fork check suite status {fork.suite_status!r}" + if fork.suite_conclusion == "SUCCESS": + return ForkState.GREEN, None + if fork.suite_conclusion in FAILED_CONCLUSIONS: + return ForkState.FAILED, None + return None, f"unexpected fork check suite conclusion {fork.suite_conclusion!r}" + if fork.workflow_state is None: + if marker is not None and marker.state == ForkState.AWAITING_PUSH.value: + return ForkState.AWAITING_PUSH, None + return None, "no fork run and the fork workflow state was not read" + if fork.workflow_state in SETUP_WORKFLOW_STATES: + return ForkState.SETUP_REQUIRED, None + if fork.workflow_state == "active": + return ForkState.AWAITING_PUSH, None + return None, f"unexpected fork workflow state {fork.workflow_state!r}" + + +def needs_workflow_state(marker: Marker | None) -> bool: + return marker is None or marker.state != ForkState.AWAITING_PUSH.value + + +STARTED_JOB_STATUSES = frozenset({"in_progress", "completed"}) + + +def jobs_show_project_ci(jobs: Iterable[dict[str, Any]], now: datetime) -> bool: + """ + Whether the jobs of a ci-amd.yml run show more than the gate deciding to skip. + + Any started job other than the gate and the ``GATED_RUN_JOB_NAMES`` jobs that was not skipped + counts, and so does a gate job that is missing or duplicated, or one of those jobs running for + longer than they ever need, which is how a renamed job doing project CI work under their name + shows up. Jobs that have not started do not count: a queued job reports its creation as + ``started_at``, so waiting for a runner would look like work. + """ + jobs = list(jobs) + gate_jobs = [job for job in jobs if job["name"] == GATE_JOB_NAME] + if len(gate_jobs) != 1: + return True + expected_names = GATED_RUN_JOB_NAMES | {GATE_JOB_NAME} + for job in jobs: + if job["name"] not in expected_names or job.get("status") not in STARTED_JOB_STATUSES: + continue + started = parse_github_datetime(job.get("started_at") or "") + completed = parse_github_datetime(job.get("completed_at") or "") or now + if started is not None and completed - started > GATE_MAX_DURATION: + return True + return any( + job["name"] not in expected_names + and job.get("status") in STARTED_JOB_STATUSES + and job.get("conclusion") != "skipped" + for job in jobs + ) + + +def touches(files: Iterable[str], prefixes: tuple[str, ...]) -> bool: + return any(path.startswith(prefixes) for path in files) + + +def changed_paths(entries: Iterable[dict[str, Any]]) -> list[str]: + """ + Every path a changed-file listing touches. + + A rename lists only its new path as ``filename``; moving a file out of a CI-defining directory + changes that directory too, so the old path counts as well. + """ + paths = [] + for entry in entries: + paths.append(entry["filename"]) + if entry.get("previous_filename"): + paths.append(entry["previous_filename"]) + return paths + + +def changes_ci_definition(files: Sequence[str], *, file_count: int | None = None) -> bool: + """ + Whether a pull request's changed files can alter what its fork CI runs. + + :param file_count: entries in the listing, when ``files`` also holds the old paths of renames. + """ + # GitHub lists at most 3000 files of a pull request; a longer one is not trusted. + return ( + (len(files) if file_count is None else file_count) >= PULL_REQUEST_FILES_CAP + or touches(files, CI_DEFINING_PREFIXES) + or any(path.endswith(CI_DEFINING_SUFFIXES) for path in files) + ) + + +def _gate_event(pr: PullRequestSnapshot, repo: str) -> GateEvent: + return GateEvent( + event_name="pull_request", + repository=repo, + ref="", + head_repository=pr.head_repo or "", + author_login=pr.author_login, + author_association=pr.author_association, + created_at=pr.created_at, + labels=pr.labels, + ) + + +def status_state(status: tuple[str, str, str] | None) -> str | None: + """The state the reconciler recorded in a commit status it posted, if it posted it.""" + if status is None: + return None + return next( + ( + state + for state, github_state in STATUS_STATES.items() + if (github_state, status_description(state)) == (status[0], status[2]) + ), + None, + ) + + +def _set_status( + plan: Plan, pr: PullRequestSnapshot, state: str, target_url: str, *, force: bool = False +) -> None: + desired = (STATUS_STATES[state], target_url, status_description(state)) + if force or desired != pr.current_status: + plan.status = desired + + +def _set_ci_passed(plan: Plan, pr: PullRequestSnapshot, *, passed: bool, target_url: str) -> None: + """ + Report fork CI to the `CI passed` check the merge ruleset requires. + + Only success is ever posted: a gated pull request without the status cannot merge, which is + what a pending or failing one would say. A success that no longer holds for the head commit, + such as a failed re-run in the fork, is overwritten so the pull request cannot merge on it. + """ + if passed and pr.ci_passed_state != "success": + plan.ci_passed = ("success", target_url, CI_PASSED_DESCRIPTION) + elif not passed and pr.ci_passed_state == "success": + plan.ci_passed = ("pending", target_url, CI_PASSED_REVOKED_DESCRIPTION) + + +def _set_marker_comment(plan: Plan, pr: PullRequestSnapshot, body: str) -> None: + if body != pr.marker_comment_body: + plan.marker_comment = body + + +def classify( + pr: PullRequestSnapshot, + fork: ForkObservation, + *, + config: GateConfig | None, + now: datetime, + repo: str, + find_bypass_run: Callable[[], str | None], + changes_ci_files: Callable[[], bool | None], + count_behind: Callable[[], int | None] = lambda: None, +) -> Plan: + """ + Decide the mutations for one gated pull request. + + :param config: the gate configuration of the pull request's base branch; None when the base + branch has no gate, so its gate job runs project CI. + :param find_bypass_run: returns the project CI run that bypassed the gate, if any; only + called for pull requests that the gate tampering check applies to. + :param changes_ci_files: whether the pull request changes CI-defining paths, None when that + cannot be told for the snapshot's head commit; only called when a new head commit turns + green in the fork. On None nothing is changed, since the decision would be recorded on a + commit the answer may not belong to. + :param count_behind: how many commits the head commit is behind the base branch, None when + unknown; called at the same time as ``changes_ci_files``. + """ + plan = Plan() + labels = set(pr.labels) + marker = pr.marker + if pr.marker_error: + plan.unknown.append(pr.marker_error) + marker = None + + def release(state: ForkState, comment: Callable[[Marker], str]) -> Plan: + plan.state = state + plan.remove_labels.append(LABEL_FORK_CI_REQUIRED) + _set_label(plan, labels, LABEL_NEEDS_PROJECT_CI, present=False) + if pr.is_draft and _is_bot(pr.last_draft_actor): + plan.draft = False + since = marker.since if marker and marker.state == state.value else now + _set_marker_comment(plan, pr, comment(Marker(state.value, pr.head_sha, since))) + _set_status(plan, pr, state.value, f"https://github.com/{repo}/pull/{pr.number}/checks";) + return plan + + if LABEL_USE_PROJECT_CI in labels: + if LABEL_USE_FORK_CI in labels: + plan.attention.append("carries both `use project ci` and `use fork ci`; project CI wins") + return release(ForkState.RELEASED, released_comment) + + if config is None: + return release(ForkState.UNGATED, functools.partial(ungated_comment, base_ref=pr.base_ref)) + + if pr.head_repo is None: + plan.unknown.append("head repository no longer exists") + return plan + + reason = decide(_gate_event(pr, repo), config).reason + if reason in TAMPER_CHECKED_REASONS: + bypass_run = find_bypass_run() + if bypass_run is not None: + plan.state = ForkState.GATE_BYPASSED + if not pr.is_draft: + plan.draft = True + if LABEL_READY_FOR_REVIEW in labels: + plan.remove_labels.append(LABEL_READY_FOR_REVIEW) + _set_label(plan, labels, LABEL_NEEDS_PROJECT_CI, present=False) + _set_status(plan, pr, ForkState.GATE_BYPASSED.value, bypass_run) + _set_marker_comment( + plan, + pr, + state_comment( + Marker(ForkState.GATE_BYPASSED.value, pr.head_sha, now), + fork=pr.head_repo, + run_url=bypass_run, + ), + ) + plan.new_comments.append(tamper_close_comment(run_url=bypass_run)) + plan.close = True + plan.attention.append(f"closed for gate tampering (author @{pr.author_login}, run {bypass_run})") + return plan + + # A closing state in the marker of an open pull request that was not reopened since means that + # posting the explanation or closing failed on an earlier run: finish the closure. Only a + # maintainer's reopen lifts a closure for gate tampering, as for the tampering check itself. + reopened_since_closing = { + ForkState.CLOSED.value: pr.last_reopened_by_anyone_at, + ForkState.GATE_BYPASSED.value: pr.last_reopened_at, + } + if marker and marker.state in reopened_since_closing: + reopened = reopened_since_closing[marker.state] + if reopened is None or reopened <= marker.since: + plan.state = ForkState(marker.state) + _set_label(plan, labels, LABEL_NEEDS_PROJECT_CI, present=False) + if plan.state == ForkState.CLOSED: + comment = inactivity_close_comment(days=config.close_after_days, fork=pr.head_repo) + else: + # The status written with the marker links the run that bypassed the gate. + bypass_url = ( + pr.current_status[1] + if pr.current_status and status_state(pr.current_status) == ForkState.GATE_BYPASSED.value + else f"https://github.com/{repo}/pull/{pr.number}/checks"; + ) + comment = tamper_close_comment(run_url=bypass_url) + plan.new_comments.append(comment) + plan.close = True + return plan + + state, unknown_reason = observe_fork_state(fork, marker) + if state is None: + plan.unknown.append(unknown_reason or "unknown fork state") + return plan + refresh_status = False + behind = marker.behind if marker and marker.sha == pr.head_sha else None + if state == ForkState.GREEN: + # The decision is read back from the commit status on the head commit, which only + # apache/airflow's own tokens can write, never from a comment. The changed files depend + # on the base too, so a status written before the pull request was retargeted is stale. + # The status is shared by every pull request with this head commit; scoping it to one is + # tracked at https://github.com/apache/airflow/issues/74907 + previous = status_state(pr.current_status) + decided = previous in (ForkState.GREEN.value, ForkState.GREEN_UNVERIFIED.value) + stale = pr.last_base_change_at is not None and ( + pr.status_created_at is None or pr.status_created_at <= pr.last_base_change_at + ) + if decided and not stale: + unverified = previous == ForkState.GREEN_UNVERIFIED.value + else: + # Writing the status again, even unchanged, records when this decision was made. + refresh_status = decided + # The `needs project ci` label lists these pull requests for maintainers. + changes = changes_ci_files() + if changes is None: + return Plan() + unverified = changes + behind = count_behind() + if unverified: + state = ForkState.GREEN_UNVERIFIED + plan.state = state + since = marker.since if marker and marker.state == state.value else now + + # The marker records awaiting_push before the nudge is posted, so a failed post is not retried; + # tracked at https://github.com/apache/airflow/issues/74908 + if marker and marker.state == ForkState.SETUP_REQUIRED.value and state == ForkState.AWAITING_PUSH: + plan.new_comments.append(nudge_comment(fork=pr.head_repo)) + + if state in CLOSABLE_STATES and now - since >= timedelta(days=config.close_after_days): + plan.state = ForkState.CLOSED + _set_label(plan, labels, LABEL_NEEDS_PROJECT_CI, present=False) + _set_marker_comment( + plan, + pr, + state_comment(Marker(ForkState.CLOSED.value, pr.head_sha, now), fork=pr.head_repo, run_url=None), + ) + plan.new_comments.append(inactivity_close_comment(days=config.close_after_days, fork=pr.head_repo)) + plan.close = True + return plan + + if state == ForkState.GREEN: + if pr.is_draft and _is_bot(pr.last_draft_actor): + plan.draft = False + ready = not pr.is_draft or plan.draft is False + if ready and LABEL_READY_FOR_REVIEW not in labels: + plan.add_labels.append(LABEL_READY_FOR_REVIEW) + else: + if not pr.is_draft: + plan.draft = True + if LABEL_READY_FOR_REVIEW in labels: + plan.remove_labels.append(LABEL_READY_FOR_REVIEW) + # Lets maintainers list the pull requests that only wait for them to run project CI, which + # otherwise look like any other draft with a pending fork CI status. + _set_label(plan, labels, LABEL_NEEDS_PROJECT_CI, present=state == ForkState.GREEN_UNVERIFIED) + + target_url = fork.run_url or fork_actions_url(pr.head_repo) + _set_status(plan, pr, state.value, target_url, force=refresh_status) + _set_ci_passed(plan, pr, passed=state == ForkState.GREEN, target_url=target_url) + green = state in (ForkState.GREEN, ForkState.GREEN_UNVERIFIED) + _set_marker_comment( + plan, + pr, + state_comment( + Marker(state.value, pr.head_sha, since, behind=behind if green else None), + fork=pr.head_repo, + run_url=fork.run_url, + base_ref=pr.base_ref, + ), + ) + return plan + + +# --------------------------------------------------------------------------------------------- +# GitHub data gathering +# --------------------------------------------------------------------------------------------- + +DISCOVERY_QUERY = """ +query($q: String!, $cursor: String) { + search(type: ISSUE, first: 100, after: $cursor, query: $q) { + pageInfo { hasNextPage endCursor } + nodes { ... on PullRequest { + number createdAt authorAssociation baseRefName + author { login } + headRepository { nameWithOwner } + labels(first: 50) { nodes { name } } + } } + } +} +""" + +PASS_TWO_QUERY = """ +query($q: String!, $cursor: String) { + search(type: ISSUE, first: %(page)d, after: $cursor, query: $q) { + pageInfo { hasNextPage endCursor } + nodes { ... on PullRequest { + number id isDraft createdAt authorAssociation baseRefName headRefName headRefOid + author { login } + headRepository { nameWithOwner } + labels(first: 50) { nodes { name } } + timelineItems( + last: 100, + itemTypes: [CONVERT_TO_DRAFT_EVENT, UNLABELED_EVENT, REOPENED_EVENT, BASE_REF_CHANGED_EVENT] + ) { + nodes { + __typename + ... on ConvertToDraftEvent { createdAt actor { login } } + ... on UnlabeledEvent { createdAt label { name } } + ... on ReopenedEvent { createdAt actor { login } } + ... on BaseRefChangedEvent { createdAt } + } + } + commits(last: 1) { nodes { commit { + status { + context(name: "%(context)s") { state targetUrl description createdAt } + ciPassed: context(name: "%(ci_passed)s") { state } + } + } } } + comments(first: 100) { + pageInfo { hasNextPage endCursor } + nodes { databaseId body author { login } } + } + } } + } +} +""" + + +def _labels(node: dict[str, Any]) -> tuple[str, ...]: + return tuple(label["name"] for label in (node.get("labels") or {}).get("nodes") or []) + + +def _login(node: dict[str, Any] | None) -> str: + return (node or {}).get("login") or "" + + +def find_marker_comment( + comments: Iterable[dict[str, Any]], +) -> tuple[Marker | None, int, str, str | None] | None: + """ + Return ``(marker, comment id, body, parse error)`` of the state comment, None if not among these. + + The state comment is the earliest bot comment with a marker: the reconciler creates it on + first contact and only edits it afterwards. Later ones may come from other workflows that echo + pull-request-controlled text, so they are never trusted. + """ + for comment in comments: + if not _is_bot(_login(comment.get("author"))): + continue + try: + parsed = parse_marker(comment["body"]) + except ValueError as e: + return None, comment["databaseId"], comment["body"], str(e) + if parsed is not None: + return parsed, comment["databaseId"], comment["body"], None + return None + + +def parse_pull_request(node: dict[str, Any]) -> PullRequestSnapshot: + last_draft_actor = None + last_unlabel = None + last_reopened = None + last_reopened_by_anyone = None + last_base_change = None + for item in (node.get("timelineItems") or {}).get("nodes") or []: + if item["__typename"] == "ConvertToDraftEvent": + last_draft_actor = _login(item.get("actor")) + elif item["__typename"] == "BaseRefChangedEvent": + last_base_change = parse_github_datetime(item["createdAt"]) + elif item["__typename"] == "ReopenedEvent": + last_reopened_by_anyone = parse_github_datetime(item["createdAt"]) + # Only a reopen by someone else counts for the gate tampering check: an author closing + # and reopening their own pull request must not hide a tampered run from it. + if _login(item.get("actor")).casefold() != _login(node.get("author")).casefold(): + last_reopened = last_reopened_by_anyone + elif ( + item["__typename"] == "UnlabeledEvent" + and (item.get("label") or {}).get("name") == LABEL_USE_PROJECT_CI + ): + last_unlabel = parse_github_datetime(item["createdAt"]) + + current_status = None + status_created_at = None + ci_passed_state = None + commits = (node.get("commits") or {}).get("nodes") or [] + if commits: + status = commits[0]["commit"].get("status") or {} + if status.get("ciPassed"): + ci_passed_state = status["ciPassed"]["state"].lower() + context = status.get("context") + if context: + current_status = ( + context["state"].lower(), + context["targetUrl"] or "", + context["description"] or "", + ) + status_created_at = parse_github_datetime(context.get("createdAt") or "") + + comments = node.get("comments") or {} + found = find_marker_comment(comments.get("nodes") or []) + page_info = comments.get("pageInfo") or {} + comments_cursor = page_info.get("endCursor") if found is None and page_info.get("hasNextPage") else None + marker = marker_id = marker_body = marker_error = None + if found is not None: + marker, marker_id, marker_body, marker_error = found + + head_repo = (node.get("headRepository") or {}).get("nameWithOwner") + return PullRequestSnapshot( + number=node["number"], + node_id=node["id"], + is_draft=node["isDraft"], + labels=_labels(node), + author_login=_login(node.get("author")), + author_association=node["authorAssociation"], + created_at=parse_github_datetime(node["createdAt"]), # type: ignore[arg-type] + head_sha=node["headRefOid"], + head_repo=head_repo, + base_ref=node["baseRefName"], + head_ref=node["headRefName"], + last_draft_actor=last_draft_actor, + last_project_ci_unlabel_at=last_unlabel, + last_reopened_at=last_reopened, + last_reopened_by_anyone_at=last_reopened_by_anyone, + last_base_change_at=last_base_change, + current_status=current_status, + status_created_at=status_created_at, + ci_passed_state=ci_passed_state, + marker=marker, + marker_comment_id=marker_id, + marker_comment_body=marker_body, + marker_error=marker_error, + comments_cursor=comments_cursor, + ) + + +def _search(client: GitHubClient, query: str, search_query: str) -> Iterable[dict[str, Any]]: + cursor = None + while True: + result = client.graphql(query, {"q": search_query, "cursor": cursor})["search"] + yield from (node for node in result["nodes"] if node) + if not result["pageInfo"]["hasNextPage"]: + return + cursor = result["pageInfo"]["endCursor"] + + +def _is_not_found(error: GitHubError) -> bool: + return "404" in str(error) or "Not Found" in str(error) + + +class BaseGateConfigs: + """ + The gate configuration of each base branch, as the gate job of a pull request against it reads it. + + The default branch's configuration comes from the reconciler's own checkout; any other base + branch's is read from GitHub once per run. + """ + + def __init__(self, client: GitHubClient, *, repo: str, default_branch: str, default_config: GateConfig): + self._client = client + self._repo = repo + self._configs: dict[str, GateConfig | None] = {default_branch: default_config} + + def _read(self, path: str, base_ref: str) -> str | None: + try: + content = self._client.rest_get( + f"repos/{self._repo}/contents/{path}?ref={quote(base_ref, safe='')}" + ) + except GitHubError as e: + if _is_not_found(e): + return None + raise + return base64.b64decode(content["content"]).decode() + + def get(self, base_ref: str) -> GateConfig | None: + """Return the base branch's configuration, or None when the base branch has no gate.""" + if base_ref not in self._configs: + config_text = self._read(GATE_CONFIG_PATH, base_ref) + has_script = self._read(GATE_SCRIPT_PATH, base_ref) is not None + if config_text is None and not has_script: + self._configs[base_ref] = None + elif config_text is None or not has_script: + # The gate job fails for these pull requests instead of deciding; so does this. + raise ValueError( + f"base branch {base_ref} has only one of {GATE_CONFIG_PATH} and {GATE_SCRIPT_PATH}" + ) + else: + self._configs[base_ref] = parse_config(config_text) + return self._configs[base_ref] + + +def discover( + client: GitHubClient, + configs: BaseGateConfigs, + *, + repo: str, + now: datetime, + unknown: list[tuple[int, str]] | None = None, +) -> list[int]: + """ + Label newly gated pull requests with `fork ci required` and return their numbers. + + :param unknown: collects the pull requests whose base branch configuration could not be read. + """ + since = (now - DISCOVERY_WINDOW).strftime("%Y-%m-%dT%H:%M:%SZ") + queries = ( + f'repo:{repo} is:pr is:open updated:>={since} -label:"{LABEL_FORK_CI_REQUIRED}"', + f'repo:{repo} is:pr is:open label:"{LABEL_USE_FORK_CI}" -label:"{LABEL_FORK_CI_REQUIRED}"', + ) + labelled: list[int] = [] + for search_query in queries: + for node in _search(client, DISCOVERY_QUERY, search_query): + if node["number"] in labelled: + continue + try: + config = configs.get(node["baseRefName"]) + except (GitHubError, ValueError) as e: + if unknown is not None: + unknown.append((node["number"], f"{type(e).__name__}: {e}")) + continue + if config is None: + continue + event = GateEvent( + event_name="pull_request", + repository=repo, + ref="", + head_repository=(node.get("headRepository") or {}).get("nameWithOwner") or "", + author_login=_login(node.get("author")), + author_association=node["authorAssociation"], + created_at=parse_github_datetime(node["createdAt"]), + labels=_labels(node), + ) + if decide(event, config).reason in GATED_REASONS: + client.add_labels(node["number"], [LABEL_FORK_CI_REQUIRED]) + labelled.append(node["number"]) + return labelled + + +def fetch_fork_observations( + client: GitHubClient, prs: list[PullRequestSnapshot] +) -> dict[int, ForkObservation]: + observations: dict[int, ForkObservation] = {} + with_fork = [pr for pr in prs if pr.head_repo] + for start in range(0, len(with_fork), FORK_BATCH_SIZE): + batch = with_fork[start : start + FORK_BATCH_SIZE] + declarations, fields, variables = [], [], {} + for index, pr in enumerate(batch): + owner, name = pr.head_repo.split("/", 1) # type: ignore[union-attr] + declarations.append(f"$o{index}: String!, $n{index}: String!, $s{index}: GitObjectID!") + fields.append( + f"pr{index}: repository(owner: $o{index}, name: $n{index}) {{ object(oid: $s{index}) {{ " + "... on Commit { checkSuites(first: 30) { nodes { createdAt status conclusion " + "workflowRun { url event workflow { resourcePath } } } } } } }" + ) + variables.update({f"o{index}": owner, f"n{index}": name, f"s{index}": pr.head_sha}) + query = f"query({', '.join(declarations)}) {{ {' '.join(fields)} }}" + data = client.graphql(query, variables, allow_partial=True) + for index, pr in enumerate(batch): + repository = data.get(f"pr{index}") + if repository is None: + observations[pr.number] = ForkObservation(error=f"fork {pr.head_repo} is not readable") + continue + commit = repository.get("object") or {} + # Only a push run executes the workflow from the head commit itself. A pull_request + # run inside the fork (into the fork's own main) runs a merge the contributor controls + # and could be made to pass trivially. + suites = [ + suite + for suite in (commit.get("checkSuites") or {}).get("nodes") or [] + if (run := suite.get("workflowRun") or {}).get("event") == "push" + and _workflow_file((run.get("workflow") or {}).get("resourcePath")) == CI_WORKFLOW_FILE + ] + if not suites: + observations[pr.number] = ForkObservation() + continue + latest = max(suites, key=lambda suite: suite["createdAt"]) + observations[pr.number] = ForkObservation( + suite_status=latest["status"], + suite_conclusion=latest["conclusion"], + run_url=latest["workflowRun"]["url"], + ) + return observations + + +def fetch_workflow_state(client: GitHubClient, fork: str) -> str: + try: + return client.rest_get(f"repos/{fork}/actions/workflows/{CI_WORKFLOW_FILE}")["state"] + except GitHubError as e: + if "404" in str(e) or "Not Found" in str(e): + return "missing" + raise + + +def parse_workflow_run(run: dict[str, Any]) -> WorkflowRun: + return WorkflowRun( + id=run["id"], + path=run["path"], + html_url=run["html_url"], + head_sha=run["head_sha"], + head_repo=(run.get("head_repository") or {}).get("full_name") or "", + head_branch=run.get("head_branch") or "", + created_at=parse_github_datetime(run["created_at"]), # type: ignore[arg-type] + run_started_at=parse_github_datetime(run.get("run_started_at") or run["created_at"]), # type: ignore[arg-type] + updated_at=parse_github_datetime(run["updated_at"]), # type: ignore[arg-type] + status=run["status"], + event=run["event"], + ) + + +def get_workflow_triggers(workflow: dict[Any, Any]) -> frozenset[str]: + """The events a parsed workflow file is triggered by.""" + # PyYAML reads the bare `on:` key as the boolean True. + on = workflow.get("on", workflow.get(True)) + if isinstance(on, str): + return frozenset({on}) + if isinstance(on, list): + return frozenset(on) + if isinstance(on, dict): + return frozenset(on) + return frozenset() + + +def load_workflow_triggers(workflows_dir: Path) -> dict[str, frozenset[str]]: + """Map each workflow file name of a checkout to the events that trigger it.""" + triggers = {} + for path in sorted(workflows_dir.glob("*.y*ml")): + workflow = yaml.safe_load(path.read_text()) or {} + triggers[path.name] = get_workflow_triggers(workflow if isinstance(workflow, dict) else {}) + return triggers + + +class BypassFinder: + """ + Finds project CI runs that bypassed the gate, from the runs apache/airflow recorded. + + It looks at every recent run of a pull request's branch for the events that run workflow + files from the pull request's merge commit, not only the runs of its head commit, so pushing a + tampered commit and then one that restores the gate does not hide the tampered run. Both + signals of the design must hold for a run: it did project CI work, and its own commit changes + `.github/workflows/`. A run did project CI work when the base branch's version of its + workflow is not triggered by the run's event (an injected workflow, or a trigger the pull + request added to an existing one, such as `pull_request` on ci-arm.yml), or when it is a long + ci-amd.yml run whose jobs show more than the gate. + """ + + def __init__( + self, client: GitHubClient, *, repo: str, now: datetime, base_triggers: dict[str, frozenset[str]] + ): + self._client = client + self._repo = repo + self._now = now + self._base_triggers = base_triggers + since = (now - TAMPER_WINDOW).strftime("%Y-%m-%dT%H:%M:%SZ") + self._runs_by_branch: dict[tuple[str, str], list[WorkflowRun]] = {} + incomplete = [] + for event in MERGE_REF_EVENTS: + pages = client.rest_get_pages( + f"repos/{repo}/actions/runs?event={event}&per_page=100&created=%3E%3D{since}" + ) + runs = [run for page in pages for run in page["workflow_runs"]] + total = pages[0]["total_count"] if pages else 0 + # GitHub lists at most 1000 runs for a filtered query; anything beyond is not checked. + if total > len(runs): + incomplete.append(f"only {len(runs)} of {total} {event} runs") + for run in map(parse_workflow_run, runs): + self._runs_by_branch.setdefault((run.head_repo.casefold(), run.head_branch), []).append(run) + self.incomplete = ( + f"{', '.join(incomplete)} of the last {TAMPER_WINDOW} could be listed, so the gate tampering " + "check did not see the rest" + if incomplete + else None + ) + self._did_project_ci: dict[int, bool] = {} + self._changes_workflows: dict[tuple[str, str], bool | None] = {} + + def _did_run_project_ci(self, run: WorkflowRun) -> bool: + if run.id not in self._did_project_ci: + if run.event not in self._base_triggers.get(run.workflow_file, frozenset()): + # Counts even when every job skipped; tracked at + # https://github.com/apache/airflow/issues/74905 + result = True + elif run.workflow_file != CI_WORKFLOW_FILE or not run.is_long_running(self._now): + result = False + else: + jobs = self._client.rest_get( + f"repos/{self._repo}/actions/runs/{run.id}/jobs?filter=latest&per_page=100", + paginate=True, + items_key="jobs", + ) + result = jobs_show_project_ci(jobs, self._now) + self._did_project_ci[run.id] = result + return self._did_project_ci[run.id] + + def _commit_changes_workflows(self, run: WorkflowRun, base_ref: str) -> bool | None: + """Whether the run's commit changes workflows against the base; None when not knowable.""" + key = (run.head_sha, base_ref) + if key not in self._changes_workflows: + owner = run.head_repo.split("/", 1)[0] + # The compare API returns `files` only on its first page; later pages list commits. + files = self._client.rest_get( + f"repos/{self._repo}/compare/{base_ref}...{owner}:{run.head_sha}?per_page=1" + ).get("files", []) + if touches(changed_paths(files), (".github/workflows/",)): + self._changes_workflows[key] = True + else: + # The compare API lists at most 300 files; a workflow change could be beyond them. + self._changes_workflows[key] = None if len(files) >= COMPARE_FILES_CAP else False + return self._changes_workflows[key] + + def find(self, pr: PullRequestSnapshot) -> str | None: + if not pr.head_repo: + return None + # Matching by fork branch alone also picks up runs of another pull request from that branch + # or from before a retarget; tracked at https://github.com/apache/airflow/issues/74906 + runs = self._runs_by_branch.get((pr.head_repo.casefold(), pr.head_ref), []) + # Runs before `use project ci` was removed were legitimate. Runs before a maintainer + # reopened the pull request were already acted on; reopening without restoring the gate + # triggers a new run, which is caught again. + not_before = max( Review Comment: Only the removal of `use project ci` and a reopen by someone other than the author bound which runs count. A PR that becomes tamper-checked for any other reason keeps its legitimate full runs from the previous 6 hours in scope. That covers `enabled_since` moved back (the backlog reset), an allowlist entry removed, or a collaborator losing their association. If that PR also edits `.github/workflows/`, both signals hold and it is closed for tampering. The ADR says the backlog reset never closes anything ("legitimate project CI runs (backlog reset, ...) ... are never closed"), but that is only true for PRs that don't touch workflows. The design doc already calls out the same trap for a `[pilot]` instant earlier than its merge. The backlog reset is a non-goal of this PR, so the cheapest fix is to narrow the claim and add the caveat to rollout step 8 (or to the allowlist-removal procedure), not to change code. **Suggestion:** Narrow the ADR and design-doc wording to "PRs that don't change `.github/workflows/`", and note in rollout step 8 that a reset closes recently tested workflow-editing PRs unless they carry `use project ci` first. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
