shahar1 commented on code in PR #74822: URL: https://github.com/apache/airflow/pull/74822#discussion_r4367561216
########## 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"}) Review Comment: A `ci-amd.yml` run that is cancelled before its gate job starts reads as project CI work here, so a gated PR that edits `.github/workflows/` legitimately can be closed for gate tampering. `ci-amd.yml` uses `cancel-in-progress: true` for `pull_request`. When a push supersedes a run whose `Fork CI gate` is still waiting for an `ubuntu-slim` runner, every job of the old run ends as `status: completed`. `Build info` and the other downstream jobs get `conclusion: cancelled`. The `always()` job is named `CI passed`, not `CI passed (project CI not run)`, because `build-info` was cancelled rather than skipped, and it finishes as `failure`. This PR's own superseded run https://github.com/apache/airflow/actions/runs/40093873055 shows exactly that shape: `Fork CI gate` COMPLETED/CANCELLED, `Build info` COMPLETED/CANCELLED, `CI passed` COMPLETED/FAILURE. Its never-started jobs (e.g. `Build info`, `Java SDK tests`) report `status: completed`, `conclusion: cancelled`, `runner_id: null` and no steps. In a run like that, `jobs_show_project_ci` returns True on `Build info` and on `CI passed`. The gate job also counts on its own: if a never-started job reports its creation time as `started_at`, which is what the docstring says about queued jobs, a gate that queued for more than 5 minutes trips the `GATE_MAX_DURATION` check. The only remaining filter is `is_long_running`, and a run that queued 5 minutes or more and was then cancelled passes it, because `updated_at - run_started_at` includes the queue time. Signal 2 holds for any PR that legitimately changes `.github/workflows/`. Such a PR gets closed with the tampering comment and an `error` status, and the closure is reported to Slack as gate tampering. The queued-job case is excluded on purpose (`queued-job-is-not-counted`, `gate-waiting-for-a-runner`), but the cancelled-while-queued case has no test. **Suggestion:** Count a job as started only when a runner actually picked it up. The jobs API returns `runner_id`/`runner_name` (null for a job that never ran) and `steps` (empty), so it could be `job.get("runner_id") is not None` or a non-empty `steps`, instead of relying on `status`. Apply the same test to the `GATE_MAX_DURATION` loop. Add a `test_jobs_show_project_ci` case for a run cancelled before the gate started. -- 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]
