From a0c27e3132e1572388763ac385c8a919137e62e7 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Wed, 22 Jul 2026 10:56:59 +0800 Subject: [PATCH 1/6] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0CI=E9=87=8D?= =?UTF-8?q?=E5=A4=8D=E5=A4=B1=E8=B4=A5=E6=A3=80=E6=B5=8B=E8=84=9A=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/ci/ci_repeated_failure_detector.py | 409 +++++++++++++++++++++ 1 file changed, 409 insertions(+) create mode 100644 scripts/ci/ci_repeated_failure_detector.py diff --git a/scripts/ci/ci_repeated_failure_detector.py b/scripts/ci/ci_repeated_failure_detector.py new file mode 100644 index 000000000..2982ec85e --- /dev/null +++ b/scripts/ci/ci_repeated_failure_detector.py @@ -0,0 +1,409 @@ +#!/usr/bin/env python3 +""" +CI重复失败检测脚本 +- 扫描最近N天的CI失败 +- 按job名称分组统计失败率 +- 识别高失败率job（系统性故障） +- 飞书通知告警 +""" + +import os +import sys +import json +import time +import urllib.request +import urllib.error +from datetime import datetime, timedelta, timezone +from collections import defaultdict + + +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 + + # 检查时间范围 + oldest = None + for r in runs: + created = r.get("created_at", "") + if created and created >= since: + all_runs.append(r) + else: + oldest = created + + 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("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_created = 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_created, + "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() \ No newline at end of file -- 2.54.0 From 246406eeb01c7332ab6f698764b7a4ed8ccc39d1 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Wed, 22 Jul 2026 10:57:01 +0800 Subject: [PATCH 2/6] =?UTF-8?q?ci:=20=E6=B7=BB=E5=8A=A0CI=E9=87=8D?= =?UTF-8?q?=E5=A4=8D=E5=A4=B1=E8=B4=A5=E6=A3=80=E6=B5=8B=E5=AE=9A=E6=97=B6?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitea/workflows/ci-failure-monitor.yml | 78 +++++++++++++++++++++++++ 1 file changed, 78 insertions(+) create mode 100644 .gitea/workflows/ci-failure-monitor.yml diff --git a/.gitea/workflows/ci-failure-monitor.yml b/.gitea/workflows/ci-failure-monitor.yml new file mode 100644 index 000000000..659769723 --- /dev/null +++ b/.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 \ No newline at end of file -- 2.54.0 From 8c1566cec3979b215e5a61f658a891365806b804 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Wed, 22 Jul 2026 11:00:34 +0800 Subject: [PATCH 3/6] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E6=97=B6=E9=97=B4?= =?UTF-8?q?=E5=AD=97=E6=AE=B5=E5=90=8D(created=5Fat->started=5Fat)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/ci/ci_repeated_failure_detector.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/scripts/ci/ci_repeated_failure_detector.py b/scripts/ci/ci_repeated_failure_detector.py index 2982ec85e..e12d78536 100644 --- a/scripts/ci/ci_repeated_failure_detector.py +++ b/scripts/ci/ci_repeated_failure_detector.py @@ -67,14 +67,14 @@ def fetch_recent_runs(days=7, per_page=50, max_pages=10): if not runs: break - # 检查时间范围 + # 检查时间范围（Gitea用started_at，格式2026-07-22T10:58:10+08:00） oldest = None for r in runs: - created = r.get("created_at", "") - if created and created >= since: + started = r.get("started_at", r.get("created_at", "")) + if started and started >= since: all_runs.append(r) else: - oldest = created + oldest = started if oldest and oldest < since: break @@ -108,7 +108,7 @@ def analyze_failures(runs): }) # 按时间正序排列（旧→新）用于连续失败计算 - sorted_runs = sorted(runs, key=lambda r: r.get("created_at", "")) + 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}) @@ -117,7 +117,7 @@ def analyze_failures(runs): run_id = run.get("id") run_status = run.get("status", "") run_conclusion = run.get("conclusion", "") - run_created = run.get("created_at", "") + run_started = run.get("started_at", run.get("created_at", "")) event = run.get("event", "") # 只统计pull_request和push事件的CI @@ -146,7 +146,7 @@ def analyze_failures(runs): stats["failure"] += 1 stats["failures"].append({ "run_id": run_id, - "time": run_created, + "time": run_started, "event": event, }) consecutive[name]["current"] += 1 -- 2.54.0 From 2cb273b4720cecf98f1591dcc52548312a9aaaee Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Wed, 22 Jul 2026 11:04:31 +0800 Subject: [PATCH 4/6] style: black format ci_repeated_failure_detector.py --- scripts/ci/ci_repeated_failure_detector.py | 194 ++++++++++++--------- 1 file changed, 113 insertions(+), 81 deletions(-) diff --git a/scripts/ci/ci_repeated_failure_detector.py b/scripts/ci/ci_repeated_failure_detector.py index e12d78536..3d6e02631 100644 --- a/scripts/ci/ci_repeated_failure_detector.py +++ b/scripts/ci/ci_repeated_failure_detector.py @@ -31,7 +31,9 @@ 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")) # 连续失败阈值 +CONSECUTIVE_FAIL_THRESHOLD = int( + get_env("CONSECUTIVE_FAIL_THRESHOLD", "3") +) # 连续失败阈值 WEBHOOK = get_env("CI_NOTIFY_WEBHOOK", "") @@ -56,17 +58,17 @@ 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: @@ -75,13 +77,13 @@ def fetch_recent_runs(days=7, per_page=50, max_pages=10): all_runs.append(r) else: oldest = started - + if oldest and oldest < since: break - + if len(runs) < per_page: break - + return all_runs @@ -97,58 +99,69 @@ def fetch_run_jobs(run_id): 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": [] - }) - + 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", ""))) - + 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, - }) + 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"] @@ -165,7 +178,7 @@ def analyze_failures(runs): elif conclusion == "cancelled": stats["cancelled"] += 1 # cancelled不算失败也不打断 - + # 计算失败率 for name, stats in job_stats.items(): total_actual = stats["total"] - stats["skipped"] - stats["cancelled"] @@ -175,14 +188,14 @@ def analyze_failures(runs): ) 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 @@ -191,13 +204,13 @@ def find_high_failures(job_stats, consecutive): 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, @@ -207,19 +220,19 @@ def find_high_failures(job_stats, consecutive): "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 @@ -230,51 +243,67 @@ def generate_report(critical, warning, info, days, total_runs): 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( + 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"| 🔴 严重 (连续失败≥{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['fail_count']}** / {item['total']} 次运行" + ) lines.append(f"- 失败率: **{item['failure_rate']}%**") - lines.append(f"- 当前连续失败: **{item['current_streak']}** 次 (历史最高: {item['max_streak']} 次)") + lines.append( + f"- 当前连续失败: **{item['current_streak']}** 次 (历史最高: {item['max_streak']} 次)" + ) lines.append("") - if item['recent_failures']: + 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']})") + 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( + 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( + f"| {item['name']} | {item['fail_count']} | {item['total']} | {item['failure_rate']}% | {item['current_streak']} |" + ) lines.append("") - + return "\n".join(lines) @@ -283,17 +312,17 @@ 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]: @@ -302,7 +331,7 @@ def send_feishu_notification(critical, warning, info, days): if len(critical) > 5: text += f" ...还有{len(critical)-5}个\n" text += "\n" - + if warning: text += "【警告】\n" for item in warning[:5]: @@ -310,25 +339,22 @@ def send_feishu_notification(critical, warning, info, days): 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}" - } - } - + + 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"}) - + 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()) @@ -348,56 +374,62 @@ def main(): 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") + 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']}次") + 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( + 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) @@ -406,4 +438,4 @@ def main(): if __name__ == "__main__": - main() \ No newline at end of file + main() -- 2.54.0 From 52e866b750d17baf1277eb8011710f30f250846a Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Wed, 22 Jul 2026 11:11:39 +0800 Subject: [PATCH 5/6] style: fix isort import ordering --- scripts/ci/ci_repeated_failure_detector.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/scripts/ci/ci_repeated_failure_detector.py b/scripts/ci/ci_repeated_failure_detector.py index 3d6e02631..220359173 100644 --- a/scripts/ci/ci_repeated_failure_detector.py +++ b/scripts/ci/ci_repeated_failure_detector.py @@ -7,14 +7,14 @@ CI重复失败检测脚本 - 飞书通知告警 """ +import json import os import sys -import json import time -import urllib.request import urllib.error -from datetime import datetime, timedelta, timezone +import urllib.request from collections import defaultdict +from datetime import datetime, timedelta, timezone def get_env(name, default=None, required=False): -- 2.54.0 From 6d8656b3f6ca46b376a6280f7a6383724857c430 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Wed, 22 Jul 2026 11:18:27 +0800 Subject: [PATCH 6/6] style: fix black/isort formatting (line-length=120) --- scripts/ci/ci_repeated_failure_detector.py | 48 ++++++---------------- 1 file changed, 12 insertions(+), 36 deletions(-) diff --git a/scripts/ci/ci_repeated_failure_detector.py b/scripts/ci/ci_repeated_failure_detector.py index 220359173..5a4c8b556 100644 --- a/scripts/ci/ci_repeated_failure_detector.py +++ b/scripts/ci/ci_repeated_failure_detector.py @@ -31,9 +31,7 @@ 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") -) # 连续失败阈值 +CONSECUTIVE_FAIL_THRESHOLD = int(get_env("CONSECUTIVE_FAIL_THRESHOLD", "3")) # 连续失败阈值 WEBHOOK = get_env("CI_NOTIFY_WEBHOOK", "") @@ -117,9 +115,7 @@ def analyze_failures(runs): ) # 按时间正序排列（旧→新）用于连续失败计算 - sorted_runs = sorted( - runs, key=lambda r: r.get("started_at", r.get("created_at", "")) - ) + 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}) @@ -183,9 +179,7 @@ def analyze_failures(runs): 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 - ) + stats["failure_rate"] = round((stats["failure"] + stats["error"]) / total_actual * 100, 1) else: stats["failure_rate"] = 0.0 @@ -243,21 +237,15 @@ def generate_report(critical, warning, info, days, total_runs): 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(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"| 🔴 严重 (连续失败≥{CONSECUTIVE_FAIL_THRESHOLD}次 或 失败率≥50%) | {len(critical)} |") + lines.append(f"| 🟡 警告 (失败率≥{FAIL_RATE_THRESHOLD}% 且 失败≥{FAIL_THRESHOLD}次) | {len(warning)} |") lines.append(f"| 🔵 关注 (失败≥2次) | {len(info)} |") lines.append("") @@ -267,21 +255,15 @@ def generate_report(critical, warning, info, days, total_runs): for item in critical: lines.append(f"### {item['name']}") lines.append("") - lines.append( - f"- 失败次数: **{item['fail_count']}** / {item['total']} 次运行" - ) + 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(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(f"- [{f['time'][:16]}] run #{f['run_id']} ({f['event']})") lines.append("") if warning: @@ -351,9 +333,7 @@ def send_feishu_notification(critical, warning, info, days): 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"} - ) + req = urllib.request.Request(WEBHOOK, data=data, headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=10) as resp: @@ -400,9 +380,7 @@ def main(): 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" - ) + 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}") @@ -419,9 +397,7 @@ def main(): if warning: print("🟡 警告:") for item in warning[:5]: - print( - f" {item['name']}: {item['fail_count']}次失败, {item['failure_rate']}%" - ) + print(f" {item['name']}: {item['fail_count']}次失败, {item['failure_rate']}%") print() print("5. 发送飞书通知...") -- 2.54.0