Andrushika commented on code in PR #74822: URL: https://github.com/apache/airflow/pull/74822#discussion_r4311746303
########## dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py: ########## @@ -0,0 +1,1181 @@ +# 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, + Reason, + decide, + parse_config, + parse_github_datetime, +) +from airflow_breeze.utils.fork_ci_github import GitHubClient, GitHubError +from airflow_breeze.utils.fork_ci_messages import ( + 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" +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: + end = self.updated_at if self.status == "completed" else now + return end - 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 + 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 + 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 + + +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 job other than the gate that was not skipped counts, and so does a gate job that is + missing, duplicated or ran for longer than the gate ever needs, which is how a renamed job + doing project CI work under the gate's name shows up. + """ + jobs = list(jobs) + gate_jobs = [job for job in jobs if job["name"] == GATE_JOB_NAME] + if len(gate_jobs) != 1: + return True + started = parse_github_datetime(gate_jobs[0].get("started_at") or "") + completed = parse_github_datetime(gate_jobs[0].get("completed_at") or "") or now + if started is not None and completed - started > GATE_MAX_DURATION: + return True + return any(job["name"] != GATE_JOB_NAME 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_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], +) -> 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; only called + when a new head commit turns green in the fork. + """ + 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 + 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 + unverified = changes_ci_files() + if unverified: + plan.attention.append( Review Comment: We might not need the Slack post for `green_unverified` now that the `needs project ci` label is there, or at least I think a channel other than `internal-airflow-ci-cd` would be better. That channel gets the urgent CI breakages today (canary, constraints), and this one fires again for every new green commit on such a PR, so it could bury those alerts. --- Drafted-by: Claude Code (Opus 5.5); reviewed by @Andrushika before posting -- 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]
