8.2 配套代码:CI 微基准门禁
对应小节:8.2 CI 微基准门禁 四个部分:① CI workflow;② 比较脚本;③ 噪声分析;④ 误报日志。
一、CI workflow(完整版)
# .github/workflows/perf-gate.yml
name: perf-gate
on:
push:
branches: [main] # ⚠️ 只在主干跑
workflow_dispatch:
jobs:
micro-benchmark:
# ⚠️ 生产项目应使用自建、规格固定的 runner
runs-on: ubuntu-latest
timeout-minutes: 30
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0
- uses: actions/setup-java@v4
with:
distribution: temurin
java-version: '21'
- uses: actions/cache@v4
with:
path: ~/.gradle
key: gradle-${{ hashFiles('**/*.gradle.kts') }}
# ① 记录 CI 环境(用于判断"环境是否变化")
- name: Record CI environment
run: |
{
echo "runner_os=$(uname -s -r -m)"
echo "cpus=$(nproc)"
echo "mem=$(free -h | awk '/Mem:/{print $2}')"
echo "java=$(java -version 2>&1 | head -1)"
echo "commit=${{ github.sha }}"
} | tee perf-ci-env.txt
# ② 跑基准
- name: Run benchmarks
run: ./gradlew benchmark -q
# ③ 与基线对比
# ⚠️ 第一阶段用 --report-only(只报告不失败),观察两周确认零误报
# 第二阶段去掉 --report-only 启用失败
- name: Compare against baseline
id: compare
continue-on-error: true
run: |
python3 tools/compare_benchmark.py \
build/reports/benchmarks/main.json \
perf/baseline.json \
--threshold 0.25 \
--report-only \
--output perf-result.md
# ④ 把结果写到 CI 摘要(开发者不用点开日志就能看到)
- name: Publish summary
if: always()
run: |
cat perf-result.md >> $GITHUB_STEP_SUMMARY
echo "" >> $GITHUB_STEP_SUMMARY
echo "### CI 环境" >> $GITHUB_STEP_SUMMARY
cat perf-ci-env.txt >> $GITHUB_STEP_SUMMARY
# ⑤ 上传原始数据(便于事后分析)
- uses: actions/upload-artifact@v4
if: always()
with:
name: benchmark-results-${{ github.sha }}
path: |
build/reports/benchmarks/
perf-result.md
perf-ci-env.txt
# ⑥ 启用失败(第二阶段才打开)
# - name: Fail on regression
# if: steps.compare.outcome == 'failure'
# run: exit 1
注意第 ③ 步的注释:先用 --report-only 观察两周——这是避免门禁被过早讨厌的关键(第 8.2 节)。
二、比较脚本(核心)
# tools/compare_benchmark.py
"""
比较当前基准结果与基线,超过阈值则失败(或只报告)。
用法:
python3 tools/compare_benchmark.py <current.json> <baseline.json> \
[--threshold 0.25] [--report-only] [--output report.md]
"""
import argparse
import json
import pathlib
import sys
# 某些基准天然波动大,可以单独设更宽的阈值
PER_BENCHMARK_THRESHOLD = {
# "bench.SomeFlakyBenchmark.method": 0.40,
}
def load_scores(path):
data = json.loads(pathlib.Path(path).read_text())
scores = {}
for b in data.get("benchmarks", []):
scores[b["name"]] = {
"score": b["score"],
"unit": b.get("scoreUnit", ""),
}
return scores
def main():
ap = argparse.ArgumentParser()
ap.add_argument("current")
ap.add_argument("baseline")
ap.add_argument("--threshold", type=float, default=0.25)
ap.add_argument("--report-only", action="store_true")
ap.add_argument("--output")
args = ap.parse_args()
current = load_scores(args.current)
baseline_path = pathlib.Path(args.baseline)
if not baseline_path.exists():
msg = (f"⚠️ 基线不存在({baseline_path})\n"
" 首次运行:请人工确认结果无误后,把当前结果提交为基线\n"
" cp build/reports/benchmarks/main.json perf/baseline.json")
print(msg)
if args.output:
pathlib.Path(args.output).write_text(msg, encoding="utf-8")
sys.exit(0)
baseline = load_scores(baseline_path)
lines = [
"## 性能门禁结果",
"",
f"- 阈值:退化超过 **{args.threshold:.0%}** 视为失败",
f"- 模式:**{'仅报告(未启用失败)' if args.report_only else '启用失败'}**",
f"- 基线条目:{len(baseline)},当前条目:{len(current)}",
"",
"| 基准 | 基线 | 当前 | 变化 | 判定 |",
"| --- | --- | --- | --- | --- |",
]
failures = []
warnings = []
for name, cur in sorted(current.items()):
short = name.split(".")[-1] if "." in name else name
threshold = PER_BENCHMARK_THRESHOLD.get(name, args.threshold)
if name not in baseline:
lines.append(f"| {short} | — | {cur['score']:.2f} | 新基准 | ℹ️ 首次 |")
continue
base_score = baseline[name]["score"]
cur_score = cur["score"]
delta = (cur_score - base_score) / base_score if base_score else 0
if delta < -threshold:
verdict = "❌ 退化"
failures.append((name, delta, threshold))
elif delta < -threshold * 0.7:
verdict = "⚠️ 接近阈值"
warnings.append((name, delta))
else:
verdict = "✅"
lines.append(
f"| {short} | {base_score:.2f} | {cur_score:.2f} | {delta:+.1%} | {verdict} |")
# 消失的基准(代码里删掉了?还是改名了?)
removed = [n for n in baseline if n not in current]
if removed:
lines += ["", f"⚠️ **{len(removed)} 个基准在基线中存在但本次未运行**(改名或删除了?)"]
for n in removed[:5]:
lines.append(f"- `{n}`")
lines.append("")
lines.append("如果是改名,请同步更新基线;如果是删除,请说明原因。")
lines += ["", "---", ""]
if failures:
lines += [
f"### ❌ 检出 {len(failures)} 项性能退化",
"",
]
for name, delta, threshold in failures:
lines.append(f"- `{name}`: {delta:+.1%}(阈值 -{threshold:.0%})")
lines += [
"",
"**处理方式**:",
"",
"1. ⚠️ **不要直接更新基线**",
"2. 先确认是真退化还是噪声:",
" - 本地复现(用同样的基准)",
" - 检查这次提交是否碰了热路径(`git diff`)",
" - 重跑 3 次看是否稳定复现",
"3. 真退化 → 修复代码",
"4. 预期变化(如换了序列化器)→ **单独提交**更新基线,写明理由",
"5. 噪声 → 记录到 `perf/false-positives.md`,考虑调整阈值",
]
elif warnings:
lines += [
f"### ⚠️ 通过,但有 {len(warnings)} 项接近阈值",
"",
]
for name, delta in warnings:
lines.append(f"- `{name}`: {delta:+.1%}(可能是下次退化的前兆)")
else:
lines.append("### ✅ 通过,无性能退化")
report = "\n".join(lines)
print(report)
if args.output:
pathlib.Path(args.output).write_text(report, encoding="utf-8")
if failures and not args.report_only:
sys.exit(1)
if __name__ == "__main__":
main()
三、CI 环境噪声分析
# tools/analyze-ci-noise.py <NOISE_DIR>
"""
分析 CI 环境的噪声,并给出阈值建议。
用法:
1. 先跑 tools/ci-noise-floor.sh 采集 10 轮数据
2. 用本脚本分析
"""
import json
import pathlib
import statistics as st
import sys
from collections import defaultdict
def load_rounds(d):
rounds = []
for f in sorted(pathlib.Path(d).glob("round-*.json")):
try:
rounds.append(json.loads(f.read_text()))
except Exception:
pass
return rounds
def main(d):
rounds = load_rounds(d)
if len(rounds) < 3:
print(f"❌ 数据不足({len(rounds)} 轮),至少需要 3 轮(推荐 10 轮)")
return
# 收集每个基准的结果
by_bench = defaultdict(list)
for data in rounds:
for b in data.get("benchmarks", []):
by_bench[b["name"]].append(b["score"])
print("═" * 92)
print(f"CI 环境噪声分析({len(rounds)} 轮)")
print("═" * 92)
print()
print(f"{'基准':<44}{'中位数':>10}{'最小':>10}{'最大':>10}{'波动':>8}{'CV':>8}{'建议阈值':>10}")
print("-" * 92)
suggestions = {}
for name, vals in sorted(by_bench.items()):
if len(vals) < 3:
continue
short = name.split(".")[-1] if "." in name else name
med = st.median(vals)
lo, hi = min(vals), max(vals)
# 波动范围:取偏离中位数较大的一侧
swing = max(abs(hi - med), abs(med - lo)) / med * 100 if med else 0
cv = st.stdev(vals) / med * 100 if len(vals) > 1 and med else 0
# 建议阈值 = 波动范围 × 1.5,向上取整到 5% 的倍数
raw = swing * 1.5
suggested = int((raw + 4.99) // 5 * 5)
suggested = max(suggested, 10) # 下限 10%
suggestions[name] = suggested / 100
flag = ""
if cv > 15:
flag = " ⚠️噪声过大"
elif cv > 8:
flag = " ⚠️"
print(f"{short[:42]:<44}{med:>10.2f}{lo:>10.2f}{hi:>10.2f}"
f"{swing:>7.1f}%{cv:>7.1f}%{suggested:>9}%{flag}")
print()
print("═" * 92)
print("阈值建议(= 波动范围 × 1.5,向上取整到 5% 的倍数)")
print()
print("```python")
print("PER_BENCHMARK_THRESHOLD = {")
for name, t in suggestions.items():
print(f' "{name}": {t:.2f},')
print("}")
print("```")
print()
# 环境质量评估
max_cv = max((st.stdev(v) / st.median(v) * 100) for v in by_bench.values() if len(v) > 1 and st.median(v))
print("环境质量评估:")
if max_cv < 5:
print(f" ✅ 最大 CV = {max_cv:.1f}% —— CI 环境很干净,可以做精细对比")
elif max_cv < 10:
print(f" ✅ 最大 CV = {max_cv:.1f}% —— 环境可接受")
elif max_cv < 20:
print(f" ⚠️ 最大 CV = {max_cv:.1f}% —— 噪声偏大,只能做粗粒度对比")
print(" 建议:换自建 runner、确保机器规格固定、关闭同 runner 的并行 job")
else:
print(f" ❌ 最大 CV = {max_cv:.1f}% —— 噪声过大")
print(" → 不建议做自动失败门禁;改成「只报告 + 人工看趋势」")
print()
print("关键提醒:")
print(" ① 阈值必须高于噪声(这里是 × 1.5)—— 否则会频繁误报")
print(" ② 先用 --report-only 观察两周,确认零误报后再启用失败")
print(" ③ 采样轮数越多,波动范围估计越准(推荐 10 轮以上)")
print("═" * 92)
if __name__ == "__main__":
main(sys.argv[1] if len(sys.argv) > 1 else "perf/ci-noise")
预期输出:
基准 中位数 最小 最大 波动 CV 建议阈值
----------------------------------------------------------------------------------------------
SerializationBenchmark.encode 1250.30 1180.20 1389.70 11.2% 4.1% 20%
SerializationBenchmark.decode 1840.50 1720.10 2010.30 15.8% 5.2% 25%
AlgorithmBenchmark.sort 820.40 780.90 901.20 14.7% 4.8% 25%
环境质量评估:
✅ 最大 CV = 5.2% —— 环境可接受
四、误报日志
<!-- perf/false-positives.md —— 记录每次门禁触发的处理结果 -->
# 门禁误报记录
> 用途:累计误报率是调整阈值的依据。
> 「重跑一次过了」不等于「这次是噪声」——必须记录并分析。
| 日期 | commit | 基准 | 变化 | 阈值 | 是否复现 | 真退化? | 处理 |
| --- | --- | --- | --- | --- | --- | --- | --- |
| 2025-01-15 | abc1234 | encode | +26% | 20% | 否(重跑 3 次均通过) | 否 | 记录 |
| 2025-01-18 | def5678 | decode | +31% | 25% | 否 | 否 | 记录 |
| 2025-01-22 | ghi9012 | sort | +28% | 25% | **是**(本地复现) | **是** | 已修复 |
## 统计(2025-01)
- 触发次数:3
- 真实退化:1
- 误报:2
- **误报率:67%**
## 结论与调整
1 月的误报率 67% 偏高(理想应 < 20%)。
**原因分析**:
- 查看 `perf-ci-env.txt`,1 月 15 日与 18 日的 runner 型号与平时不同(`ubuntu-latest` 升级)
- 说明:**外部环境变化导致的误报,不是阈值问题**
**调整方案**:
1. 改用自建、规格固定的 runner(长期方案)
2. 短期内阈值从 20%/25% 提高到 30%(临时方案)
3. 在 `compare_benchmark.py` 里记录 runner 型号,型号变化时**自动跳过对比**(避免环境变化导致误报)
**下一轮复核**:2025-02
注意最后一条方案:「型号变化时自动跳过对比」——这是一个很实用的技巧,可以在脚本里实现。
# 在 compare_benchmark.py 里加上环境一致性检查
import pathlib, json
def check_env_consistency(current_env_file, baseline_env_file):
"""如果 CI runner 型号变化,跳过对比(避免误报)"""
if not (pathlib.Path(current_env_file).exists() and pathlib.Path(baseline_env_file).exists()):
return True # 无法检查,继续
cur = pathlib.Path(current_env_file).read_text()
base = pathlib.Path(baseline_env_file).read_text()
def extract(env_text, key):
for line in env_text.splitlines():
if line.startswith(f"{key}="):
return line.split("=", 1)[1]
return ""
if extract(cur, "runner_os") != extract(base, "runner_os"):
print("⚠️ CI runner 型号变化 —— 跳过对比(避免误报)")
print(f" 基线: {extract(base, 'runner_os')}")
print(f" 当前: {extract(cur, 'runner_os')}")
return False
return True
五、动手改造
| 改动 | 观察什么 |
|---|---|
用 analyze-ci-noise.py 分析你的 CI 数据 |
得到你的阈值建议 |
| 把阈值从建议值调低一半 | 误报率会飙升——验证"阈值必须高于噪声" |
在 workflow 里去掉 --report-only |
门禁立即开始失败/通过——体会"先报告后失败"的价值 |
| 故意改一处热路径代码 | 门禁应该能检出——验证门禁真的有效 |
| 用误报日志的统计调整阈值 | 体会「用数据校准阈值」而非拍脑袋 |
| 加上 runner 型号检查 | 观察"环境变化导致的误报"是否被消除 |
六、这段代码的局限
--threshold 0.25是示例值:必须用analyze-ci-noise.py在你自己的 CI 环境测出来。PER_BENCHMARK_THRESHOLD需要逐个基准调:某些基准天然波动大(比如涉及 GC 的),可以单独放宽。- 误报日志需要人维护:如果没人记录,就无法校准阈值——这是门禁长期存活的关键,但最容易被忽略。
ubuntu-latest会变化:这是 GitHub Actions 的已知问题,生产项目应该用固定的版本号或自建 runner。- 脚本假设 kotlinx-benchmark 的 JSON 格式:如果用 JMH 的原生输出,需要调整
load_scores的解析。