Files
doris/.github/scripts/review_github_api.py
shuke cb41339ffd [fix](ci) Retry transient review GET failures within bounded budgets (#69510)
Related PR: #69469

Problem Summary: A single transient GitHub response currently aborts the
review pipeline. [Input resolution for
#69322](https://github.com/apache/doris/actions/runs/39794013905/job/119876748768)
stopped on a GET 503 despite `Retry-After: 2`; [aggregate resolution for
#69438](https://github.com/apache/doris/actions/runs/39760934956/job/119764632659)
stopped on a paginated GET 504 after emitting partial output.

The shared helper now retries recognized body-free REST GETs on
502/503/504, for at most three attempts. Each attempt has a 30-second
limit; attempts and waits share a 90-second budget or the caller's
shorter timeout. It respects Retry-After, records every failed attempt,
discards failed pagination output, and returns only a successful
attempt's complete output. Exhaustion still fails. Writes, GraphQL,
unknown argument forms, other HTTP errors, transport/timeouts and JSON
parsing errors are not retried.

Input resolution can also fail before publishing its PR-number output.
Failure notices now fall back to the authorized trigger's numeric PR
identity, explicitly select the repository, and name the
input-resolution failure in both the notice and terminal error. This
fallback does not permit status publication for an unresolved head/base.

This change can merge independently. It includes the same explicit
`--repo` argument as #69469; a local combination with #69469 and #69504
merged cleanly and passed the macOS suite.
2026-11-02 14:23:14 +08:00

287 lines
12 KiB
Python

#!/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.
"""Capture gh HTTP failures and retry transient failures of read-only REST GETs."""
import json
import os
import re
import subprocess
import sys
import time
import uuid
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path
from urllib.parse import urlsplit
LIMIT = 65536
HEADERS = {
"content-type", "content-length", "date", "x-github-request-id",
"retry-after", "x-ratelimit-limit", "x-ratelimit-remaining",
"x-ratelimit-reset", "x-ratelimit-resource",
}
def redact(value):
if isinstance(value, bytes):
value = value.decode("utf-8", errors="replace")
value = value or ""
# Artifacts do not inherit Actions' secret masker. Never store the request
# headers or body; also redact credentials echoed by an error response.
for name, secret in os.environ.items():
if re.search(r"(?:TOKEN|SECRET|PASSWORD|(?:^|_)KEY|(?:^|_)AK|(?:^|_)SK)$", name) and secret:
for encoded in (secret, json.dumps(secret)[1:-1]):
value = value.replace(encoded, "[REDACTED]")
value = re.sub(r"\b(?:gh[pousr]_[A-Za-z0-9]+|github_pat_[A-Za-z0-9_]+)\b", "[REDACTED]", value)
value = re.sub(
r'(?i)(["\']?(?:access_token|refresh_token|id_token|token|api_key|secret|'
r'client_secret|password|authorization|cookie|set-cookie)["\']?\s*[:=]\s*)'
r'("(?:\\.|[^"\\])*"|[^\r\n,}]+)', r'\1"[REDACTED]"', value,
)
value = re.sub(r"(https?://)[^/\s:@]+:[^/@\s]+@", r"\1[REDACTED]@", value)
value = re.sub(r"(?i)([?&](?:token|access_token|signature|sig|key)=)[^&\s\"']+", r"\1[REDACTED]", value)
return value
def excerpt(value):
value = redact(value)
return value[:LIMIT] + ("\n[truncated]" if len(value) > LIMIT else "")
def split_trace(stderr):
"""Keep only the final HTTP exchange; discard every outgoing header/body.
GH_DEBUG=api writes HTTP diagnostics to stderr, independently of gh api's
response parser. Its body is a formatted, size-limited debug excerpt, not
a byte-for-byte copy. Unknown debug formatting is discarded, never dumped.
"""
text = stderr.decode("utf-8", errors="replace") if isinstance(stderr, bytes) else (stderr or "")
response = {"method": None, "path": None, "status": None, "headers": {},
"body_excerpt": "", "body_source": "gh_debug", "request_count": 0}
ordinary = []
mode = "ordinary"
body = []
body_chars = 0
for line in text.splitlines():
if line.startswith("* Request at "):
count = response["request_count"] + 1
response = {**response, "method": None, "path": None, "status": None,
"headers": {}, "body_excerpt": "", "request_count": count}
body = []
body_chars = 0
response.pop("body_truncated", None)
mode = "request"
elif line.startswith("> "):
match = re.match(r"> ([A-Z]+) (\S+) HTTP/", line)
if match:
response["method"] = match[1]
response["path"] = redact(urlsplit(match[2]).path)
mode = "request"
elif re.match(r"< HTTP/\S+ \d{3}", line):
response["status"] = int(line.split()[2])
response["headers"] = {}
body = []
body_chars = 0
mode = "headers"
elif mode == "headers" and line.startswith("< "):
name, sep, value = line[2:].partition(":")
if sep and name.lower() in HEADERS:
response["headers"][name.lower()] = excerpt(value.strip())
elif line.startswith("* Request took "):
mode = "ordinary"
elif mode == "headers" and not line:
mode = "body"
elif mode == "body":
# Bound retained response content even if debug output grows.
line = redact(line) # Redact before truncation can split a secret.
if body_chars < LIMIT:
response["body_truncated"] = len(line) + body_chars > LIMIT
body.append(line[:LIMIT - body_chars])
body_chars += len(line) + 1
else:
response["body_truncated"] = True
elif mode == "ordinary" and not line.startswith("* "):
ordinary.append(line)
response["body_excerpt"] = excerpt("\n".join(body).strip())
if mode == "request":
# A transport failure has no response; do not mistake request data for
# its body. subprocess timeout/type still identifies the failure.
response["body_excerpt"] = ""
return response, excerpt("\n".join(ordinary).strip())
def record_failure(command, error, stdout="", stderr="", *, diagnostic=None):
response, ordinary = split_trace(stderr) if diagnostic is None else (diagnostic, excerpt(stderr))
record = {
"time": datetime.now(timezone.utc).isoformat(),
"run_id": os.environ.get("GITHUB_RUN_ID"),
"run_attempt": os.environ.get("GITHUB_RUN_ATTEMPT"),
"job": os.environ.get("GITHUB_JOB"),
"gh_version": os.environ.get("REVIEW_GH_VERSION"),
"error_type": type(error).__name__,
"returncode": getattr(error, "returncode", None),
"timeout_seconds": getattr(error, "timeout", None),
"http": response,
"stderr": ordinary,
"stdout_excerpt": excerpt(stdout),
}
# Do not record argv: -f/--body/--input can contain credentials or entire
# reviews. Endpoint metadata comes only from the outgoing request line.
summary = (
f"GitHub request failed: {response['method'] or 'unknown'} {response['path'] or 'unknown'}; "
f"HTTP={response['status']}; request_id={response['headers'].get('x-github-request-id', 'unknown')}; "
f"{record['error_type']}; exit={record['returncode']}; timeout={record['timeout_seconds']}"
)
record["summary"] = redact(summary)
directory = os.environ.get("REVIEW_GITHUB_DIAGNOSTICS_DIR")
if directory:
try:
path = Path(directory)
path.mkdir(parents=True, exist_ok=True, mode=0o700)
data = json.dumps(record, ensure_ascii=True, indent=2) + "\n"
# Unique files support concurrent helper calls without append races.
filename = path / f"{time.time_ns()}-{uuid.uuid4().hex}.json"
with open(filename, "x", opener=lambda p, f: os.open(p, f, 0o600)) as handle:
handle.write(data)
except OSError as exc:
print(f"Could not persist GitHub diagnostic ({type(exc).__name__})", file=sys.stderr)
print(record["summary"], file=sys.stderr, flush=True)
return ordinary, record
def is_read_only_get(command):
"""Recognize only the body-free gh api forms used by the review pipeline.
Unknown options remain single-attempt. In particular, fields/input imply
POST in gh, and GraphQL uses POST even without an explicit method. Never
infer safety from the last HTTP response: a write can redirect to GET.
"""
if len(command) < 3 or Path(command[0]).name != "gh" or command[1] != "api":
return False
endpoint = None
args = iter(command[2:])
for arg in args:
if arg in ("--paginate", "--slurp", "-XGET", "--method=GET"):
continue
if arg in ("-X", "--method"):
if next(args, None) != "GET":
return False
elif arg in ("-H", "--header"):
if next(args, None) is None:
return False
elif arg.startswith("-") or endpoint is not None:
return False
else:
endpoint = arg
return endpoint is not None and urlsplit(endpoint).path.rstrip("/").split("/")[-1] != "graphql"
def retry_delay(response, attempt):
"""Do not shorten a server-requested wait, or guess if it is malformed."""
delay = 2 ** attempt
value = response["headers"].get("retry-after")
if value is None:
return delay
try:
if value.isdigit():
return max(delay, int(value))
when = parsedate_to_datetime(value)
return max(delay, (when - datetime.now(timezone.utc)).total_seconds())
except (TypeError, ValueError, OverflowError):
return None
def run_gh(command, **kwargs):
"""Buffer each attempt; a caller timeout bounds the entire GET operation."""
env = dict(kwargs.pop("env", os.environ))
env.update(GH_DEBUG="api", NO_COLOR="1")
retry_get = (is_read_only_get(command) and kwargs.get("check") is True
and kwargs.get("capture_output") is True and kwargs.get("text") is True
and kwargs.get("input") is None and kwargs.get("stdin") is None)
if retry_get:
# Fit within the runner's two-minute input-resolution step. Existing
# shorter submission/watchdog budgets include all attempts and sleeps.
budget = min(90, kwargs.get("timeout") if kwargs.get("timeout") is not None else 90)
deadline = time.monotonic() + budget
for attempt in range(1, 4):
options = dict(kwargs)
if retry_get:
options["timeout"] = min(30, max(0, deadline - time.monotonic()))
try:
result = subprocess.run(command, env=env, **options)
except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as error:
ordinary, record = record_failure(command, error, error.stdout, error.stderr)
error.stderr = ordinary + "\n" + record["summary"]
if retry_get:
# gh --paginate may emit incomplete JSON before a later page
# fails. Retain it only in diagnostics, never in caller output.
error.stdout = ""
response = record["http"]
if (isinstance(error, subprocess.CalledProcessError)
and error.returncode > 0 and response["method"] == "GET"
and response["status"] in (502, 503, 504) and attempt < 3):
delay = retry_delay(response, attempt)
if delay is not None and delay < deadline - time.monotonic():
print(f"Retrying GitHub GET after HTTP {response['status']} in {delay:g}s "
f"(attempt {attempt + 1}/3)", file=sys.stderr, flush=True)
time.sleep(delay)
if time.monotonic() < deadline:
continue
raise
except OSError as error:
record_failure(command, error)
raise
response, ordinary = split_trace(result.stderr)
result.stderr = ordinary
result.github_response = response
return result
def decode_json(result, command):
try:
return json.loads(result.stdout)
except json.JSONDecodeError as error:
record_failure(command, error, result.stdout, result.stderr,
diagnostic=result.github_response)
raise
def main():
command = ["gh", *sys.argv[1:]]
try:
result = run_gh(command, capture_output=True, text=True, check=True)
except subprocess.CalledProcessError as error:
sys.stdout.write(error.stdout or "")
print(error.stderr, file=sys.stderr)
return error.returncode if error.returncode >= 0 else 128 - error.returncode
except subprocess.TimeoutExpired as error:
print(error.stderr, file=sys.stderr)
return 124
except OSError as error:
print(f"Could not execute gh ({type(error).__name__})", file=sys.stderr)
return 1
sys.stdout.write(result.stdout)
if result.stderr:
print(result.stderr, file=sys.stderr)
return 0
if __name__ == "__main__":
sys.exit(main())