shahar1 commented on code in PR #75455: URL: https://github.com/apache/airflow/pull/75455#discussion_r4359130144
########## dev/breeze/src/airflow_breeze/utils/workflow_run_approval.py: ########## @@ -0,0 +1,590 @@ +# 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. +""" +Automatic approval of workflow runs from external contributors' fork PRs. + +GitHub holds the workflow runs of a fork PR in ``action_required`` until someone with write access +approves them. This module approves the runs of PRs to ``main`` that do not touch anything that +defines CI or chooses runners, do not need the full set of tests or a dependency upgrade, and whose +jobs, as selected by selective checks, are estimated to take a limited number of job-minutes. +Everything else gets the ``need workflow approval`` label and waits for a maintainer. + +Only API metadata and git objects of the PR are read. The PR's code is never checked out or run: +selective checks come from the trusted ``main`` checkout and only see the list of changed files +and the PR's ``pyproject.toml`` contents through ``git show``. +""" + +from __future__ import annotations + +import os +import re +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +import requests +from rich.markup import escape + +from airflow_breeze.global_constants import GithubEvents +from airflow_breeze.utils.ci_minutes_estimate import CiMinutesEstimate, estimate_ci_minutes +from airflow_breeze.utils.console import console_print +from airflow_breeze.utils.path_utils import AIRFLOW_ROOT_PATH +from airflow_breeze.utils.run_utils import run_command + +NEED_WORKFLOW_APPROVAL_LABEL = "need workflow approval" + +# Only PRs to main are approved automatically. PRs to release branches (``v3-4-test``, +# ``airflow-ctl/v0-1-test``, ...) run those branches' own, older workflows and always need a maintainer. +AUTO_APPROVABLE_BASE_BRANCH = "main" + +# Changes here can change which runner a job uses (``runs-on:``, ``runner-type`` computed by +# selective checks, the scripts build-info runs) or what the workflows do, so they always need a +# maintainer's look. +CI_DEFINITION_PATH_PREFIXES = (".github/", "dev/breeze/", "scripts/ci/") + +# Dependency manifests and lock files. Selective checks only switch to dependency-upgrade mode for +# pyproject.toml, but a changed lock file or JS / Go / Java manifest also changes what CI installs. +DEPENDENCY_FILE_PATTERN = re.compile( + r"(^|/)(pyproject\.toml|uv\.lock|requirements[^/]*\.txt|package\.json|pnpm-lock\.yaml" + r"|package-lock\.json|yarn\.lock|go\.mod|go\.sum|build\.gradle(\.kts)?|settings\.gradle(\.kts)?" + r"|gradle\.lockfile|libs\.versions\.toml|pom\.xml" + r"|gradle-wrapper\.(properties|jar)|maven-wrapper\.(properties|jar)|gradlew(\.bat)?|mvnw(\.cmd)?)$" +) + +# About 23% of a full-tests run (~1330 job-minutes on public runners), just above the cluster of +# single core areas and the Helm chart (274-301). Changes to large providers (amazon 364, google 374) +# and anything bigger wait for a maintainer, to keep the load on the shared public runners down. +MAX_AUTO_APPROVED_JOB_MINUTES = 310 + +# Larger changes to non-test code are features or behaviour changes rather than fixes, and get a +# maintainer's look even when their CI is cheap. 100 lines is the 90th percentile of the PRs that +# pass the other rules; above it the distribution thins out into feature-sized changes. +MAX_AUTO_APPROVED_SOURCE_LINES = 100 + +# Files that are not counted as source: tests (including e2e tests), docs, newsfragments, +# translations, lock files and images. +NON_SOURCE_FILE_PATTERN = re.compile( + r"(^|/)(tests?|test_utils|testing|[^/]*e2e[^/]*)/" + r"|(^|/)test_[^/]*$|_test\.(py|go)$|\.test\.(ts|tsx|js)$|\.spec\.(ts|tsx)$|Test\.java$" + r"|\.(md|rst|txt)$|(^|/)newsfragments/|(^|/)docs?/" + r"|/i18n/locales/|uv\.lock$|pnpm-lock\.yaml$|package-lock\.json$|go\.sum$" + r"|\.(svg|png|jpg|snap)$" +) + +GITHUB_API_URL = "https://api.github.com"; + + +@dataclass +class ChangedFile: + path: str + changed_lines: int + + +@dataclass +class ApprovalDecision: + approve: bool + reasons: list[str] = field(default_factory=list) + estimate: CiMinutesEstimate | None = None + + +def count_source_lines(changed_files: list[ChangedFile]) -> int: + return sum(file.changed_lines for file in changed_files if not NON_SOURCE_FILE_PATTERN.search(file.path)) + + +def find_ci_definition_files(files: list[str]) -> list[str]: + return [file for file in files if file.startswith(CI_DEFINITION_PATH_PREFIXES)] + + +def find_dependency_files(files: list[str]) -> list[str]: + return [file for file in files if DEPENDENCY_FILE_PATTERN.search(file)] + + +def find_new_provider_files(files: list[str]) -> list[str]: + """ + Return provider.yaml files that the trusted checkout does not have. + + Selective checks run from main's tree and do not know a provider the PR adds, so they cannot + estimate its tests. + """ + return [ + file + for file in files + if file.startswith("providers/") + and file.endswith("/provider.yaml") + and not (AIRFLOW_ROOT_PATH / file).exists() + ] + + +def decide_approval( + *, + base_branch: str, + changed_files: list[ChangedFile], + selective_checks: Any, +) -> ApprovalDecision: + """Decide whether the workflow runs of a fork PR commit can be approved without a maintainer.""" + try: + full_tests_needed = selective_checks.full_tests_needed + all_versions = selective_checks.all_versions + is_canary_run = selective_checks._is_canary_run() + upgrade_to_newer_dependencies = selective_checks.upgrade_to_newer_dependencies + estimate = estimate_ci_minutes(selective_checks) + except (SystemExit, Exception): + # Selective checks exit when a PR changes a suspended provider without the label allowing it, + # and raise on malformed PR content such as an invalid pyproject.toml. Either way, fail closed. + return ApprovalDecision(approve=False, reasons=["selective checks fail for this change"]) + if base_branch != AUTO_APPROVABLE_BASE_BRANCH: + return ApprovalDecision( + approve=False, + reasons=[f"it targets {base_branch}, not {AUTO_APPROVABLE_BASE_BRANCH}"], + estimate=estimate, + ) + paths = [file.path for file in changed_files] + ci_files = find_ci_definition_files(paths) + if ci_files: + return ApprovalDecision( + approve=False, + reasons=[f"it changes CI definition files: {', '.join(ci_files)}"], + estimate=estimate, + ) + dependency_files = find_dependency_files(paths) + if dependency_files: + return ApprovalDecision( + approve=False, + reasons=[f"it changes dependency files: {', '.join(dependency_files)}"], + estimate=estimate, + ) + new_provider_files = find_new_provider_files(paths) + if new_provider_files: + return ApprovalDecision( + approve=False, + reasons=[f"it adds a new provider: {', '.join(new_provider_files)}"], + estimate=estimate, + ) + if full_tests_needed: + return ApprovalDecision(approve=False, reasons=["it needs the full set of tests"], estimate=estimate) + if is_canary_run: + # The canary label adds special tests and integration jobs that the estimate does not model. + return ApprovalDecision(approve=False, reasons=["it runs as a canary build"], estimate=estimate) + if all_versions: + # The estimate models the default Python and database versions only. + return ApprovalDecision( + approve=False, reasons=["it runs all Python and database versions"], estimate=estimate + ) + if upgrade_to_newer_dependencies: + return ApprovalDecision(approve=False, reasons=["it changes dependencies"], estimate=estimate) + if estimate.job_minutes > MAX_AUTO_APPROVED_JOB_MINUTES: + return ApprovalDecision( + approve=False, + reasons=[ + f"its jobs take an estimated {estimate.job_minutes} job-minutes " + f"(more than {MAX_AUTO_APPROVED_JOB_MINUTES})" + ], + estimate=estimate, + ) + source_lines = count_source_lines(changed_files) + if source_lines > MAX_AUTO_APPROVED_SOURCE_LINES: + return ApprovalDecision( + approve=False, + reasons=[ + f"it changes {source_lines} lines of source code " + f"(more than {MAX_AUTO_APPROVED_SOURCE_LINES}, tests and docs not counted)" + ], + estimate=estimate, + ) + return ApprovalDecision( + approve=True, + reasons=[ + f"its jobs take an estimated {estimate.job_minutes} job-minutes " + f"(at most {MAX_AUTO_APPROVED_JOB_MINUTES}) and it changes {source_lines} lines of source code" + ], + estimate=estimate, + ) + + +def parse_github_time(value: str) -> datetime: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + + +class GitHubApi: + def __init__(self, *, github_repository: str, github_token: str, report_only: bool): + self.github_repository = github_repository + self.report_only = report_only + self._session = requests.Session() + self._session.headers.update( + { + "Accept": "application/vnd.github+json", + "Authorization": f"Bearer {github_token}", + "X-GitHub-Api-Version": "2022-11-28", + } + ) + + def _get_paginated(self, path: str, *, item_key: str | None = None) -> list[dict[str, Any]]: + items: list[dict[str, Any]] = [] + url: str | None = f"{GITHUB_API_URL}/repos/{self.github_repository}/{path}" + while url: + response = self._session.get(url, timeout=60) + response.raise_for_status() + payload = response.json() + items.extend(payload[item_key] if item_key else payload) + url = response.links.get("next", {}).get("url") + return items + + def get_runs_waiting_for_approval(self, created_since: datetime) -> list[dict[str, Any]]: + created = created_since.strftime("%Y-%m-%dT%H:%M:%SZ") + return self._get_paginated( + f"actions/runs?status=action_required&event=pull_request&per_page=100&created=%3E%3D{created}", + item_key="workflow_runs", + ) + + def get_open_pull_requests(self) -> list[dict[str, Any]]: + return self._get_paginated("pulls?state=open&per_page=100") + + def approve_run(self, run_id: int) -> None: + if self.report_only: + console_print(f"[info]Report only: would approve workflow run {run_id}[/]") + return + response = self._session.post( + f"{GITHUB_API_URL}/repos/{self.github_repository}/actions/runs/{run_id}/approve", timeout=60 + ) + response.raise_for_status() + + def add_label(self, pr_number: int, label: str) -> None: + if self.report_only: + console_print(f"[info]Report only: would add label '{label}' to PR #{pr_number}[/]") + return + response = self._session.post( + f"{GITHUB_API_URL}/repos/{self.github_repository}/issues/{pr_number}/labels", + json={"labels": [label]}, + timeout=60, + ) + response.raise_for_status() + + def get_last_base_change(self, pr_number: int) -> datetime | None: + events = self._get_paginated(f"issues/{pr_number}/timeline?per_page=100") + changes = [ + parse_github_time(event["created_at"]) + for event in events + if event.get("event") == "base_ref_changed" + ] + return max(changes, default=None) + + def remove_label(self, pr_number: int, label: str) -> None: + if self.report_only: + console_print(f"[info]Report only: would remove label '{label}' from PR #{pr_number}[/]") + return + response = self._session.delete( + f"{GITHUB_API_URL}/repos/{self.github_repository}/issues/{pr_number}/labels/{label}", + timeout=60, + ) + if response.status_code != 404: + response.raise_for_status() + + +def is_shallow_repository() -> bool: + result = run_command( + ["git", "rev-parse", "--is-shallow-repository"], + check=False, + capture_output=True, + text=True, + dry_run_override=False, + ) + return result.stdout.strip() == "true" + + +def fetch_merge_commit(*, github_repository: str, pr_number: int, head_sha: str) -> str | None: + """ + Fetch the PR's merge commit (what the CI run tests) and return its sha. + + Return None when there is no merge commit or it was not built from ``head_sha``, so the result + is always tied to the commit whose runs get approved. + """ + ref = f"refs/remotes/auto-approve/pull/{pr_number}/merge" + # Only the merge commit and its parents are needed. Limit the depth only in an already shallow + # checkout (CI): in a full clone, --depth would record shallow boundaries and truncate history. + depth = ["--depth=2"] if is_shallow_repository() else [] + result = run_command( + [ + "git", + "fetch", + "--no-tags", + *depth, + f"https://github.com/{github_repository}.git";, + f"+refs/pull/{pr_number}/merge:{ref}", + ], + check=False, + capture_output=True, + text=True, + dry_run_override=False, + ) + if result.returncode != 0: + return None + parents = run_command( + ["git", "rev-list", "--parents", "-n", "1", ref], + check=False, + capture_output=True, + text=True, + dry_run_override=False, + ) + if parents.returncode != 0: + return None + merge_sha, *parent_shas = parents.stdout.split() + if len(parent_shas) != 2 or parent_shas[1] != head_sha: + return None + return merge_sha + + +NUMSTAT_RECORD = re.compile(r"(\d+|-)\t(\d+|-)\t(.+)", re.DOTALL) + + +def get_changed_files(merge_sha: str) -> list[ChangedFile]: + """ + List the files the merge commit changes, with their changed line counts. + + Paths must be literal for the CI-definition and source checks: without ``-z`` git C-quotes unusual + file names, and rename detection reports ``old => new`` instead of both paths. + """ + result = run_command( + [ + "git", + "-c", + "core.quotePath=false", + "diff", + "--no-renames", + "--numstat", + "-z", + f"{merge_sha}^1", + merge_sha, + ], + check=True, + capture_output=True, + text=True, + dry_run_override=False, + ) + changed_files = [] + for record in result.stdout.split("\0"): + if not record: + continue + match = NUMSTAT_RECORD.fullmatch(record) + if match is None: + raise ValueError(f"Unexpected git numstat record: {record!r}") + added, deleted, path = match.groups() + # Binary files report "-" for both counts; count them over the limit so they never pass. + changed_lines = int(added) + int(deleted) if added != "-" else MAX_AUTO_APPROVED_SOURCE_LINES + 1 + changed_files.append(ChangedFile(path=path, changed_lines=changed_lines)) + return changed_files + + +def build_selective_checks( + *, merge_sha: str, changed_files: list[ChangedFile], pr_labels: tuple[str, ...], base_branch: str +): + from airflow_breeze.utils.selective_checks import SelectiveChecks + + # Release-branch PRs select different jobs (no provider tests, for example), so selective checks + # must see the branch the PR targets. + return SelectiveChecks( + files=tuple(file.path for file in changed_files), + default_branch=base_branch, + commit_ref=merge_sha, + pr_labels=pr_labels, + github_event=GithubEvents.PULL_REQUEST, + ) + + +@dataclass +class PullRequestResult: + pr_number: int + head_sha: str + run_ids: list[int] + decision: ApprovalDecision + + +def process_waiting_runs(api: GitHubApi, *, created_since: datetime) -> list[PullRequestResult]: + """ + Decide on the runs created since ``created_since``. + + Older waiting runs are left alone, so the workflow never approves a backlog of stale runs at once. + """ + runs = api.get_runs_waiting_for_approval(created_since) + if not runs: + console_print( + f"[info]No workflow runs created since {created_since:%Y-%m-%d %H:%M} UTC wait for approval.[/]" + ) + return [] + runs_by_head: dict[tuple[str, str, str], list[dict[str, Any]]] = {} + for run in runs: + head_repository = (run.get("head_repository") or {}).get("full_name", "") + if head_repository == api.github_repository: + continue + runs_by_head.setdefault((head_repository, run["head_branch"], run["head_sha"]), []).append(run) + # The runs of fork PRs do not list their pull request, so match them by head repository, branch + # and commit. + pull_requests: dict[tuple[str, str, str], list[dict[str, Any]]] = {} + for pr in api.get_open_pull_requests(): + head_repository = pr["head"]["repo"]["full_name"] if pr["head"].get("repo") else "" + pull_requests.setdefault((head_repository, pr["head"]["ref"], pr["head"]["sha"]), []).append(pr) + results = [] + for (head_repository, head_branch, head_sha), head_runs in runs_by_head.items(): + run_ids = [run["id"] for run in head_runs] + matching_prs = pull_requests.get((head_repository, head_branch, head_sha), []) + if not matching_prs: + console_print( + f"[warning]No open PR has head {escape(head_repository)}:{escape(head_branch)}@{head_sha}; " + f"leaving runs {run_ids} alone[/]" + ) + continue + for pr in matching_prs: + if len(matching_prs) > 1: + # The runs cannot be told apart by PR, so one PR's decision must not approve them all. + decision = ApprovalDecision( + approve=False, reasons=["its head commit is the head of several open PRs"] + ) + else: + decision = decide_for_pull_request(api=api, pr=pr) + if decision.approve: + run_ids = find_runs_started_for_current_pr(api=api, pr=pr, runs=head_runs) + if not run_ids: + decision = ApprovalDecision( + approve=False, + reasons=["its waiting runs started before the PR was opened or retargeted"], + estimate=decision.estimate, + ) + run_ids = [run["id"] for run in head_runs] + results.append( + PullRequestResult( + pr_number=pr["number"], head_sha=head_sha, run_ids=run_ids, decision=decision + ) + ) + apply_decision(api=api, pr=pr, run_ids=run_ids, decision=decision) Review Comment: **nit** (robustness, no exact rule) — every GitHub call here raises on a non-2xx status (`approve_run`, `add_label`, the timeline read), and `get_changed_files` runs git with `check=True`, but nothing guards a single PR. A run that a maintainer approves or cancels by hand between the listing and the POST, or a transient 5xx/secondary rate limit, aborts the decisions for every later PR in that tick and skips `write_step_summary`, so the summary is lost for the PRs already approved. It fails closed and the next tick retries, so this is not a safety problem. Catching `requests.HTTPError` / `subprocess.CalledProcessError` per PR, recording it as a result, and writing the summary in a `finally` would keep one bad PR from blanking a tick. ########## dev/calibrate_ci_minutes_estimates.py: ########## @@ -0,0 +1,251 @@ +#!/usr/bin/env python3 +# 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. +""" Review Comment: **minor** — `dev/AGENTS.md` → Scripts: *"Each script must include inline script metadata placed **after** the Apache License header, so that `uv run` can execute it without any prior installation […] If the script only uses the standard library, omit the `dependencies` key but keep the `requires-python` line. […] Document `uv run` (not `python`) as the invocation method in READMEs and instructions."* This script imports only the standard library but has no `# /// script` block, and both its docstring and step 1 of `.github/AGENTS.md` say `uv run --project dev/breeze python dev/calibrate_ci_minutes_estimates.py`, which syncs the whole Breeze environment for nothing. Suggest adding ```python # /// script # requires-python = ">=3.11" # /// ``` after the licence header and documenting `uv run dev/calibrate_ci_minutes_estimates.py --since <date>` in both places. -- 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]
