.gitea/workflows/ci-failure-monitor.yml

@@ -0,0 +1,78 @@
+name: CI Failure Monitor
+
+on:
+  schedule:
+    - cron: '0 */6 * * *'  # 每6小时检查一次
+  workflow_dispatch:
+    inputs:
+      days:
+        description: '统计最近N天的失败'
+        required: false
+        default: '7'
+      fail_threshold:
+        description: '失败次数阈值'
+        required: false
+        default: '3'
+      fail_rate_threshold:
+        description: '失败率阈值(%)'
+        required: false
+        default: '30'
+
+permissions:
+  contents: read
+
+jobs:
+  monitor:
+    name: CI重复失败检测
+    runs-on: ci-l2
+    timeout-minutes: 10
+
+    steps:
+    - name: Checkout code
+      shell: sh
+      env:
+        GITHUB_TOKEN: ${{ github.token }}
+      run: |
+        curl -sH "Authorization: token $GITHUB_TOKEN" \
+          "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/raw/scripts/ci/step_checkout.sh?ref=${GITHUB_SHA}" \
+          | bash
+
+    - name: Record job start time
+      shell: sh
+      run: bash scripts/ci/step_timer_start.sh
+
+    - name: Run failure detection
+      shell: sh
+      env:
+        GITEA_API_TOKEN: ${{ secrets.REVIEW_GITEA_TOKEN }}
+        GITEA_URL: https://git.xiaoxiajianji.com
+        GITEA_REPO: xiaoxia/xiaoxia-saas
+        CI_NOTIFY_WEBHOOK: ${{ secrets.CI_NOTIFY_WEBHOOK }}
+        FAIL_CHECK_DAYS: ${{ inputs.days || 7 }}
+        FAIL_THRESHOLD: ${{ inputs.fail_threshold || 3 }}
+        FAIL_RATE_THRESHOLD: ${{ inputs.fail_rate_threshold || 30 }}
+      run: |
+        set +e
+        python3 scripts/ci/ci_repeated_failure_detector.py
+        EXIT_CODE=$?
+        echo "检测完成，退出码: $EXIT_CODE"
+        # 0=无异常, 1=有警告, 2=有严重问题
+        # 监控脚本永远不fail，避免告警风暴
+        exit 0
+
+    - name: Job duration summary
+      if: always()
+      shell: sh
+      run: bash scripts/ci/step_timer_end.sh
+
+    - name: Report CI trace
+      if: always()
+      shell: sh
+      env:
+        AGENTLOOP_LICENSE_KEY: ${{ secrets.AGENTLOOP_LICENSE_KEY }}
+      run: |
+        STATUS="ok"
+        [ ${{ job.status }} = "success" ] || STATUS="error"
+        START_TIME=""
+        [ -f /tmp/ci_job_start_time ] && START_TIME=$(cat /tmp/ci_job_start_time)
+        python3 scripts/ci/ci_trace_report.py --service xiaoxia-saas-ci --status $STATUS --start-time "$START_TIME" || true

scripts/ci/ci_repeated_failure_detector.py

@@ -0,0 +1,417 @@
+#!/usr/bin/env python3
+"""
+CI重复失败检测脚本
+- 扫描最近N天的CI失败
+- 按job名称分组统计失败率
+- 识别高失败率job（系统性故障）
+- 飞书通知告警
+"""
+
+import json
+import os
+import sys
+import time
+import urllib.error
+import urllib.request
+from collections import defaultdict
+from datetime import datetime, timedelta, timezone
+
+
+def get_env(name, default=None, required=False):
+    val = os.environ.get(name, default)
+    if required and not val:
+        print(f"❌ 缺少环境变量: {name}")
+        sys.exit(1)
+    return val
+
+
+GITEA_URL = get_env("GITEA_URL", "https://git.xiaoxiajianji.com")
+GITEA_TOKEN = get_env("GITEA_API_TOKEN", required=False) or get_env("GITHUB_TOKEN", "")
+REPO = get_env("GITEA_REPO", "xiaoxia/xiaoxia-saas")
+DAYS = int(get_env("FAIL_CHECK_DAYS", "7"))
+FAIL_THRESHOLD = int(get_env("FAIL_THRESHOLD", 3))  # 失败次数阈值
+FAIL_RATE_THRESHOLD = float(get_env("FAIL_RATE_THRESHOLD", "30"))  # 失败率阈值%
+CONSECUTIVE_FAIL_THRESHOLD = int(get_env("CONSECUTIVE_FAIL_THRESHOLD", "3"))  # 连续失败阈值
+WEBHOOK = get_env("CI_NOTIFY_WEBHOOK", "")
+
+
+def api_get(path):
+    """调用Gitea API"""
+    url = f"{GITEA_URL}/api/v1{path}"
+    req = urllib.request.Request(url)
+    if GITEA_TOKEN:
+        req.add_header("Authorization", f"token {GITEA_TOKEN}")
+    try:
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            return json.loads(resp.read())
+    except urllib.error.HTTPError as e:
+        print(f"  HTTP {e.code}: {path}")
+        return None
+    except Exception as e:
+        print(f"  错误: {e}")
+        return None
+
+
+def fetch_recent_runs(days=7, per_page=50, max_pages=10):
+    """获取最近N天的runs"""
+    since = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
+    all_runs = []
+
+    for page in range(1, max_pages + 1):
+        path = f"/repos/{REPO}/actions/runs?page={page}&limit={per_page}"
+        data = api_get(path)
+        if not data:
+            break
+
+        runs = data.get("workflow_runs", data.get("runs", []))
+        if not runs:
+            break
+
+        # 检查时间范围（Gitea用started_at，格式2026-07-22T10:58:10+08:00）
+        oldest = None
+        for r in runs:
+            started = r.get("started_at", r.get("created_at", ""))
+            if started and started >= since:
+                all_runs.append(r)
+            else:
+                oldest = started
+
+        if oldest and oldest < since:
+            break
+
+        if len(runs) < per_page:
+            break
+
+    return all_runs
+
+
+def fetch_run_jobs(run_id):
+    """获取run的所有jobs"""
+    path = f"/repos/{REPO}/actions/runs/{run_id}/jobs"
+    data = api_get(path)
+    if not data:
+        return []
+    return data.get("jobs", [])
+
+
+def analyze_failures(runs):
+    """
+    分析失败情况
+
+    返回:
+    - job_stats: {job_name: {total, success, failure, skipped, failure_rate, failures: [...]}}
+    - consecutive_failures: {job_name: current_streak, max_streak, last_status}
+    """
+    job_stats = defaultdict(
+        lambda: {
+            "total": 0,
+            "success": 0,
+            "failure": 0,
+            "error": 0,
+            "skipped": 0,
+            "cancelled": 0,
+            "failures": [],
+        }
+    )
+
+    # 按时间正序排列（旧→新）用于连续失败计算
+    sorted_runs = sorted(runs, key=lambda r: r.get("started_at", r.get("created_at", "")))
+
+    # 连续失败跟踪 {job_name: streak}
+    consecutive = defaultdict(lambda: {"current": 0, "max": 0, "last_run": None})
+
+    for run in sorted_runs:
+        run_id = run.get("id")
+        run_status = run.get("status", "")
+        run_conclusion = run.get("conclusion", "")
+        run_started = run.get("started_at", run.get("created_at", ""))
+        event = run.get("event", "")
+
+        # 只统计pull_request和push事件的CI
+        if event not in ("pull_request", "push"):
+            continue
+
+        jobs = fetch_run_jobs(run_id)
+
+        for job in jobs:
+            name = job.get("name", "")
+            status = job.get("status", "")
+            conclusion = job.get("conclusion", "")
+
+            # 跳过非CI核心job（如AI Code Review、Preview等）
+            skip_prefixes = ("AI Code Review", "Preview", "PR Automation", "Auto")
+            if any(name.startswith(p) for p in skip_prefixes):
+                continue
+
+            stats = job_stats[name]
+            stats["total"] += 1
+
+            if conclusion == "success":
+                stats["success"] += 1
+                consecutive[name]["current"] = 0
+            elif conclusion == "failure":
+                stats["failure"] += 1
+                stats["failures"].append(
+                    {
+                        "run_id": run_id,
+                        "time": run_started,
+                        "event": event,
+                    }
+                )
+                consecutive[name]["current"] += 1
+                if consecutive[name]["current"] > consecutive[name]["max"]:
+                    consecutive[name]["max"] = consecutive[name]["current"]
+                consecutive[name]["last_run"] = run_id
+            elif conclusion == "error":
+                stats["error"] += 1
+                # error也算失败的一种
+                consecutive[name]["current"] += 1
+                if consecutive[name]["current"] > consecutive[name]["max"]:
+                    consecutive[name]["max"] = consecutive[name]["current"]
+            elif conclusion == "skipped":
+                stats["skipped"] += 1
+                # skipped不算也不打断连续失败
+            elif conclusion == "cancelled":
+                stats["cancelled"] += 1
+                # cancelled不算失败也不打断
+
+    # 计算失败率
+    for name, stats in job_stats.items():
+        total_actual = stats["total"] - stats["skipped"] - stats["cancelled"]
+        if total_actual > 0:
+            stats["failure_rate"] = round((stats["failure"] + stats["error"]) / total_actual * 100, 1)
+        else:
+            stats["failure_rate"] = 0.0
+
+    return dict(job_stats), dict(consecutive)
+
+
+def find_high_failures(job_stats, consecutive):
+    """
+    找出高风险job
+
+    告警级别：
+    - critical: 连续失败 >= CONSECUTIVE_FAIL_THRESHOLD，或 失败率>=50%且失败次数>=5
+    - warning: 失败率>=FAIL_RATE_THRESHOLD且失败次数>=FAIL_THRESHOLD
+    - info: 失败次数>=2
+    """
+    critical = []
+    warning = []
+    info = []
+
+    for name, stats in job_stats.items():
+        fail_count = stats["failure"] + stats["error"]
+        rate = stats["failure_rate"]
+        streak = consecutive.get(name, {}).get("current", 0)
+        max_streak = consecutive.get(name, {}).get("max", 0)
+
+        issue = {
+            "name": name,
+            "fail_count": fail_count,
+            "total": stats["total"],
+            "failure_rate": rate,
+            "current_streak": streak,
+            "max_streak": max_streak,
+            "recent_failures": stats["failures"][-5:],  # 最近5次
+        }
+
+        if streak >= CONSECUTIVE_FAIL_THRESHOLD or (rate >= 50 and fail_count >= 5):
+            critical.append(issue)
+        elif rate >= FAIL_RATE_THRESHOLD and fail_count >= FAIL_THRESHOLD:
+            warning.append(issue)
+        elif fail_count >= 2:
+            info.append(issue)
+
+    # 按失败次数倒序
+    critical.sort(key=lambda x: x["fail_count"], reverse=True)
+    warning.sort(key=lambda x: x["fail_count"], reverse=True)
+    info.sort(key=lambda x: x["fail_count"], reverse=True)
+
+    return critical, warning, info
+
+
+def generate_report(critical, warning, info, days, total_runs):
+    """生成Markdown报告"""
+    lines = []
+    lines.append("# CI重复失败检测报告")
+    lines.append("")
+    lines.append(f"**统计周期**: 最近{days}天")
+    lines.append(f"**扫描Runs**: {total_runs}个")
+    lines.append(f"**生成时间**: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}")
+    lines.append("")
+
+    lines.append(f"## 概览")
+    lines.append("")
+    lines.append(f"| 级别 | 数量 |")
+    lines.append(f"|------|------|")
+    lines.append(f"| 🔴 严重 (连续失败≥{CONSECUTIVE_FAIL_THRESHOLD}次 或 失败率≥50%) | {len(critical)} |")
+    lines.append(f"| 🟡 警告 (失败率≥{FAIL_RATE_THRESHOLD}% 且 失败≥{FAIL_THRESHOLD}次) | {len(warning)} |")
+    lines.append(f"| 🔵 关注 (失败≥2次) | {len(info)} |")
+    lines.append("")
+
+    if critical:
+        lines.append("## 🔴 严重问题")
+        lines.append("")
+        for item in critical:
+            lines.append(f"### {item['name']}")
+            lines.append("")
+            lines.append(f"- 失败次数: **{item['fail_count']}** / {item['total']} 次运行")
+            lines.append(f"- 失败率: **{item['failure_rate']}%**")
+            lines.append(f"- 当前连续失败: **{item['current_streak']}** 次 (历史最高: {item['max_streak']} 次)")
+            lines.append("")
+            if item["recent_failures"]:
+                lines.append("最近失败:")
+                lines.append("")
+                for f in item["recent_failures"]:
+                    lines.append(f"- [{f['time'][:16]}] run #{f['run_id']} ({f['event']})")
+                lines.append("")
+
+    if warning:
+        lines.append("## 🟡 警告")
+        lines.append("")
+        for item in warning:
+            lines.append(
+                f"- **{item['name']}**: {item['fail_count']}次失败 / {item['total']}次运行 ({item['failure_rate']}%)"
+            )
+        lines.append("")
+
+    if info:
+        lines.append("## 🔵 关注列表")
+        lines.append("")
+        lines.append("| Job名称 | 失败次数 | 总次数 | 失败率 | 当前连续 |")
+        lines.append("|---------|----------|--------|--------|----------|")
+        for item in info[:20]:  # 最多显示20个
+            lines.append(
+                f"| {item['name']} | {item['fail_count']} | {item['total']} | {item['failure_rate']}% | {item['current_streak']} |"
+            )
+        lines.append("")
+
+    return "\n".join(lines)
+
+
+def send_feishu_notification(critical, warning, info, days):
+    """发送飞书通知"""
+    if not WEBHOOK:
+        print("  ⚠️  未配置WEBHOOK，跳过飞书通知")
+        return False
+
+    total_issues = len(critical) + len(warning) + len(info)
+    if total_issues == 0:
+        print("  ✅ 无异常，不发送通知")
+        return True
+
+    level = "🔴 严重告警" if critical else "🟡 警告" if warning else "🔵 关注"
+
+    title = f"CI重复失败检测 - {level}"
+    text = f"统计周期: 最近{days}天\n\n"
+
+    if critical:
+        text += "【严重问题】\n"
+        for item in critical[:5]:
+            text += f"• {item['name']}\n"
+            text += f"  失败 {item['fail_count']}/{item['total']} ({item['failure_rate']}%) 连续{item['current_streak']}次\n"
+        if len(critical) > 5:
+            text += f"  ...还有{len(critical)-5}个\n"
+        text += "\n"
+
+    if warning:
+        text += "【警告】\n"
+        for item in warning[:5]:
+            text += f"• {item['name']}: {item['fail_count']}次失败 ({item['failure_rate']}%)\n"
+        if len(warning) > 5:
+            text += f"  ...还有{len(warning)-5}个\n"
+        text += "\n"
+
+    if info and not critical and not warning:
+        text += "【关注列表】\n"
+        for item in info[:10]:
+            text += f"• {item['name']}: {item['fail_count']}次失败\n"
+        text += "\n"
+
+    text += f"共发现 {total_issues} 个异常job"
+
+    payload = {"msg_type": "text", "content": {"text": f"{title}\n\n{text}"}}
+
+    data = json.dumps(payload).encode()
+    req = urllib.request.Request(WEBHOOK, data=data, headers={"Content-Type": "application/json"})
+
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            result = json.loads(resp.read())
+            if result.get("code") == 0 or result.get("StatusCode") == 0:
+                print("  ✅ 飞书通知已发送")
+                return True
+            else:
+                print(f"  ⚠️  飞书返回: {result}")
+                return False
+    except Exception as e:
+        print(f"  ❌ 飞书通知失败: {e}")
+        return False
+
+
+def main():
+    print(f"=== CI重复失败检测 ===")
+    print(f"统计周期: 最近{DAYS}天")
+    print(f"仓库: {REPO}")
+    print()
+
+    print("1. 获取最近的Runs...")
+    runs = fetch_recent_runs(days=DAYS)
+    print(f"   找到 {len(runs)} 个runs")
+
+    if not runs:
+        print("⚠️  没有找到runs，退出")
+        return
+
+    print()
+    print("2. 分析job失败情况（可能需要点时间）...")
+    job_stats, consecutive = analyze_failures(runs)
+    print(f"   共统计 {len(job_stats)} 个job")
+
+    print()
+    print("3. 识别高风险job...")
+    critical, warning, info = find_high_failures(job_stats, consecutive)
+    print(f"   🔴 严重: {len(critical)}")
+    print(f"   🟡 警告: {len(warning)}")
+    print(f"   🔵 关注: {len(info)}")
+
+    print()
+    print("4. 生成报告...")
+    report = generate_report(critical, warning, info, DAYS, len(runs))
+
+    # 保存报告
+    report_path = os.environ.get("REPORT_PATH", f"/tmp/ci_failure_report_{int(time.time())}.md")
+    with open(report_path, "w") as f:
+        f.write(report)
+    print(f"   报告已保存: {report_path}")
+
+    # 打印摘要
+    print()
+    print("=== 摘要 ===")
+    if critical:
+        print("🔴 严重问题:")
+        for item in critical[:5]:
+            print(
+                f"  {item['name']}: {item['fail_count']}次失败, {item['failure_rate']}%, 连续{item['current_streak']}次"
+            )
+    if warning:
+        print("🟡 警告:")
+        for item in warning[:5]:
+            print(f"  {item['name']}: {item['fail_count']}次失败, {item['failure_rate']}%")
+
+    print()
+    print("5. 发送飞书通知...")
+    send_feishu_notification(critical, warning, info, DAYS)
+
+    print()
+    print("✅ 检测完成")
+
+    # 有严重问题时退出码非零，方便workflow标记
+    if critical:
+        sys.exit(2)
+    elif warning:
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()