409 lines
14 KiB
Python
409 lines
14 KiB
Python
#!/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() |