xiaoxia-saas/scripts/ci/ci_transient_retry.sh

#!/bin/bash
# ============================================================================
# CI 瞬态故障自动重试脚本
# ============================================================================
#
# 解决什么问题：
#   当某个 runner 出现瞬态故障（磁盘满、docker 挂掉、网络抖动等），分配到该
#   runner 的 job 会在 "Set up job" / "Checkout code" 阶段就失败。这种失败
#   与代码无关，换到其他 runner 重跑就能通过，但 CI 没有内置重试机制，只能
#   人工干预。
#
# 检测逻辑：
#   1. 查询最近 30 分钟内完成的 workflow runs
#   2. 找到 conclusion=failure 的 run
#   3. 检查失败 run 中的 jobs：
#      a. 失败的 job 的第一个 step 必须是 "Checkout code" 或 "Set up job"
#         且 conclusion=failure（说明 runner 环境有问题，不是代码问题）
#      b. 同一 run 中至少有 2 个成功的 job（排除代码本身全挂的情况）
#      c. 失败 job 数量 < 总 job 数量的 50%（多数成功=瞬态故障）
#   4. 满足以上条件 → 判定为瞬态故障 → 自动 re-run 整个 workflow
#
# 安全措施：
#   - 每个 run 最多自动重试 1 次（通过状态文件跟踪，避免死循环）
#   - 状态文件 24 小时后自动清理
#   - 所有操作记录日志，便于审计
#
# 用法：
#   ./scripts/ci/ci_transient_retry.sh              # 正常运行
#   ./scripts/ci/ci_transient_retry.sh --dry-run     # 只检查不重试
#   ./scripts/ci/ci_transient_retry.sh --verbose     # 详细日志输出
#
# 环境变量：
#   GITEA_API_TOKEN  - Gitea API token（必须设置）
#   GITEA_API_URL    - Gitea API 基础地址（默认 https://git.xiaoxiajianji.com/api/v1）
#   GITEA_REPO       - 仓库路径（默认 xiaoxia/xiaoxia-saas）
#   RETRY_STATE_DIR  - 重试状态目录（默认 /opt/act-runner-docker/ci-retry-state）
#   LOG_FILE         - 日志文件（默认 /var/log/ci-auto-retry.log）
#
# 部署方式：
#   1. 将本脚本复制到 CI 服务器（如 /opt/act-runner-docker/ci-retry-state/）
#   2. 创建 env 文件: echo 'GITEA_API_TOKEN=xxx' > /opt/act-runner-docker/ci-retry-state/env
#   3. 安装 systemd timer:
#      cp scripts/ci/ci_transient_retry.{service,timer} /etc/systemd/system/
#      systemctl daemon-reload
#      systemctl enable --now ci_transient_retry.timer
#   或用 crontab:
#      */5 * * * * GITEA_API_TOKEN=xxx bash /opt/act-runner-docker/ci-retry-state/ci_transient_retry.sh
# ============================================================================
set -eu

# ---- 配置 ----
GITEA_API_URL="${GITEA_API_URL:-https://git.xiaoxiajianji.com/api/v1}"
GITEA_REPO="${GITEA_REPO:-xiaoxia/xiaoxia-saas}"
RETRY_STATE_DIR="${RETRY_STATE_DIR:-/opt/act-runner-docker/ci-retry-state}"
LOG_FILE="${LOG_FILE:-/var/log/ci-auto-retry.log}"
WINDOW_MINUTES=30          # 检查最近 N 分钟内的 run
MAX_RETRY_PER_RUN=1        # 每个 run 最多重试次数
STATE_TTL_HOURS=24         # 状态文件过期时间（小时）
MIN_SUCCESS_JOBS=2         # 至少 N 个 job 成功才算瞬态故障
MAX_FAIL_RATIO=50          # 失败 job 占比上限（%）

# ---- 参数解析 ----
DRY_RUN=false
VERBOSE=false
for arg in "$@"; do
  case "$arg" in
    --dry-run)  DRY_RUN=true ;;
    --verbose)  VERBOSE=true ;;
  esac
done

# ---- 日志 ----
log() {
  local level="$1"; shift
  local ts
  ts=$(date -u +%Y-%m-%dT%H:%M:%SZ)
  local msg="[$ts] [$level] $*"
  echo "$msg" >> "$LOG_FILE" 2>/dev/null || true
  if [ "$level" = "ERROR" ] || [ "$level" = "WARN" ] || $VERBOSE || $DRY_RUN; then
    echo "$msg"
  fi
}
log_info()  { log INFO "$@"; }
log_warn()  { log WARN "$@"; }
log_error() { log ERROR "$@"; }
log_debug() { $VERBOSE && log DEBUG "$@" || true; }

# ---- 前置检查 ----
if [ -z "${GITEA_API_TOKEN:-}" ]; then
  log_error "GITEA_API_TOKEN 未设置，退出"
  exit 1
fi

mkdir -p "$RETRY_STATE_DIR" 2>/dev/null || true
mkdir -p "$(dirname "$LOG_FILE")" 2>/dev/null || true

API_BASE="${GITEA_API_URL}/repos/${GITEA_REPO}/actions"

log_info "========== CI 瞬态故障检测开始 =========="
$DRY_RUN && log_info "[DRY-RUN 模式] 不会实际触发重试"

# ---- 清理过期状态文件 ----
find "$RETRY_STATE_DIR" -name "*.retry" -mmin "+$((STATE_TTL_HOURS * 60))" -delete 2>/dev/null || true

# ---- 临时文件（用 mktemp 避免引号/转义问题） ----
TMPDIR_WORK=$(mktemp -d)
trap "rm -rf $TMPDIR_WORK" EXIT

# ---- 查询最近完成的 runs ----
curl -sf -H "Authorization: token ${GITEA_API_TOKEN}" \
  "${API_BASE}/runs?status=completed&limit=20" > "${TMPDIR_WORK}/runs.json" 2>/dev/null || echo '[]' > "${TMPDIR_WORK}/runs.json"

# ---- 主分析逻辑（全部用 python3，避免 bash JSON 处理陷阱） ----
python3 - "$TMPDIR_WORK" "$API_BASE" "$GITEA_API_TOKEN" "$RETRY_STATE_DIR" \
  "$WINDOW_MINUTES" "$MIN_SUCCESS_JOBS" "$MAX_FAIL_RATIO" "$MAX_RETRY_PER_RUN" \
  "$DRY_RUN" "$VERBOSE" "$LOG_FILE" << 'PYEOF'
import json, sys, os, subprocess, urllib.request
from datetime import datetime, timezone, timedelta

tmpdir = sys.argv[1]
api_base = sys.argv[2]
token = sys.argv[3]
state_dir = sys.argv[4]
window_min = int(sys.argv[5])
min_success = int(sys.argv[6])
max_fail_pct = int(sys.argv[7])
max_retry = int(sys.argv[8])
dry_run = sys.argv[9] == "true"
verbose = sys.argv[10] == "true"
log_file = sys.argv[11]

def log(level, msg):
    ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    line = f"[{ts}] [{level}] {msg}"
    try:
        with open(log_file, "a") as f:
            f.write(line + "\n")
    except:
        pass
    if level in ("ERROR", "WARN") or verbose or dry_run:
        print(line)

# 加载 runs
with open(os.path.join(tmpdir, "runs.json")) as f:
    runs_data = json.load(f)
if isinstance(runs_data, dict):
    runs_data = runs_data.get("workflow_runs", runs_data.get("runs", []))

cutoff = datetime.now(timezone.utc) - timedelta(minutes=window_min)

# 筛选最近 N 分钟内完成的失败 runs
candidates = []
for r in runs_data:
    completed_str = r.get("completed_at", "")
    if not completed_str or completed_str.startswith("1970"):
        continue
    try:
        completed = datetime.fromisoformat(completed_str.replace("Z", "+00:00"))
        if completed < cutoff:
            continue
    except:
        continue
    if r.get("conclusion") != "failure":
        continue
    candidates.append(r)

if not candidates:
    log("INFO", f"没有发现最近 {window_min} 分钟内失败的 runs")
    log("INFO", "========== 检测结束 ==========")
    sys.exit(0)

log("INFO", f"发现 {len(candidates)} 个失败的 runs 需要分析")

retry_count = 0
headers = {"Authorization": f"token {token}"}

for run in candidates:
    run_id = run["id"]
    run_number = run.get("run_number", run_id)
    branch = run.get("head_branch", "")
    event = run.get("event", "")

    log("INFO", f"分析 Run #{run_number} (id={run_id}) branch={branch} event={event}")

    # 检查是否已重试
    state_file = os.path.join(state_dir, f"{run_id}.retry")
    prev_attempts = 0
    if os.path.exists(state_file):
        try:
            with open(state_file) as f:
                prev_attempts = int(f.read().strip())
        except:
            pass
    if prev_attempts >= max_retry:
        log("INFO", f"  Run #{run_number} 已重试过 {prev_attempts} 次，跳过")
        continue

    # 获取 jobs
    try:
        req = urllib.request.Request(
            f"{api_base}/runs/{run_id}/jobs",
            headers=headers
        )
        with urllib.request.urlopen(req, timeout=15) as resp:
            jobs_data = json.loads(resp.read())
    except Exception as e:
        log("ERROR", f"  获取 jobs 失败: {e}")
        continue

    jobs = jobs_data.get("jobs", [])
    total = len(jobs)
    success_count = 0
    fail_count = 0
    transient_jobs = []
    code_fail_jobs = []

    for j in jobs:
        conclusion = j.get("conclusion", "")
        if conclusion == "success":
            success_count += 1
        elif conclusion == "failure":
            fail_count += 1
            steps = j.get("steps", [])
            if steps:
                first = steps[0]
                fname = first.get("name", "").lower()
                fconc = first.get("conclusion", "")
                # 瞬态故障特征：第一个 step（checkout/setup）失败
                if fconc == "failure" and any(kw in fname for kw in ["checkout", "set up", "setup"]):
                    transient_jobs.append({
                        "name": j["name"],
                        "runner": j.get("runner_name", "?"),
                    })
                else:
                    code_fail_jobs.append(j["name"])
            else:
                code_fail_jobs.append(j["name"])

    skip_count = total - success_count - fail_count
    log("INFO", f"  Jobs: total={total} success={success_count} fail={fail_count} skip={skip_count}")

    if transient_jobs:
        log("INFO", f"  瞬态故障 jobs: {', '.join(j['name'] for j in transient_jobs)}")
    if code_fail_jobs:
        log("INFO", f"  代码失败 jobs: {', '.join(code_fail_jobs)}")

    # 判定
    is_transient = False
    reason = ""
    if transient_jobs and not code_fail_jobs:
        if success_count >= min_success:
            is_transient = True
            reason = f"所有失败 job 在 checkout/setup 阶段失败，{success_count} 个 job 成功"
        else:
            reason = f"checkout 失败但成功 job 数不足 ({success_count} < {min_success})"
    elif transient_jobs and code_fail_jobs:
        if fail_count < total * max_fail_pct / 100 and success_count >= min_success:
            is_transient = True
            reason = f"{len(transient_jobs)} 瞬态 + {len(code_fail_jobs)} 代码，但成功 job 占多数"
        else:
            reason = f"混合失败: {len(transient_jobs)} 瞬态 + {len(code_fail_jobs)} 代码"
    elif code_fail_jobs:
        reason = f"纯代码失败: {', '.join(code_fail_jobs[:3])}"
    else:
        reason = "无失败 job"

    log("INFO", f"  判定: {reason}")

    if not is_transient:
        log("INFO", "  → 非瞬态故障，跳过")
        continue

    # 触发重试
    log("WARN", f"  → 检测到瞬态故障！准备重试 Run #{run_number}")

    if dry_run:
        log("INFO", "  [DRY-RUN] 跳过实际重试")
        continue

    # 调用 re-run API
    try:
        req = urllib.request.Request(
            f"{api_base}/runs/{run_id}/rerun",
            headers={**headers, "Content-Type": "application/json"},
            method="POST",
            data=b""
        )
        with urllib.request.urlopen(req, timeout=30) as resp:
            result = json.loads(resp.read())
            new_run_id = result.get("id", "?")
            new_status = result.get("status", "?")

        # 记录重试状态
        with open(state_file, "w") as f:
            f.write(str(prev_attempts + 1))

        log("INFO", f"  ✅ Re-run 成功! 新 Run ID: {new_run_id}, 状态: {new_status}")
        retry_count += 1
    except Exception as e:
        log("ERROR", f"  ❌ Re-run 失败: {e}")

log("INFO", f"========== 检测结束: 检查 {len(candidates)} 个失败 runs，重试 {retry_count} 个 ==========")
PYEOF

exit 0