5.7 配套代码:噪声底线测量与回归门禁
对应小节:5.7 噪声底线与回归检测 这是本章最重要的一份代码。 它产出你后续所有实验的标尺。
一、噪声底线测量流程
#!/usr/bin/env bash
# tools/noise-floor.sh <EXP_ID> [ROUNDS] [RATE] [DURATION]
#
# 测量噪声底线:代码完全不变,重复 N 轮实验。
#
# 关键要求:
# ① 代码一字不改
# ② 每轮都重启服务(确保同一初始状态)
# ③ 每轮都独立预热(不计入统计)
# ④ 至少 5 轮(推荐 10 轮)
set -euo pipefail
EXP_ID="${1:?usage: noise-floor.sh <EXP_ID> [rounds] [rate] [duration]}"
ROUNDS="${2:-10}"
RATE="${3:-500}"
DURATION="${4:-5m}"
DIR="docs/experiments/${EXP_ID}"
mkdir -p "$DIR/results" "$DIR/scripts"
echo "═══ 噪声底线测量 ═══"
echo " 实验编号 : $EXP_ID"
echo " 轮数 : $ROUNDS"
echo " 负载 : $RATE req/s × $DURATION"
echo " ⚠️ 请确认:代码未改动、环境已按第 5.4 节检查"
echo
# ① 环境检查
tools/check-environment.sh "$EXP_ID" || true
# ② 冻结脚本
cp loadtest/*.js "$DIR/scripts/" 2>/dev/null || true
# ③ 启动环境事件记录(用于解释异常轮)
tools/record-events.sh "$EXP_ID" 10 &
EVENTS_PID=$!
trap 'kill $EVENTS_PID 2>/dev/null || true' EXIT
# ④ 记录代码状态(必须干净)
{
echo "commit=$(git rev-parse HEAD)"
echo "git_dirty_files=$(git status --porcelain | wc -l | tr -d ' ')"
} | tee "$DIR/results/code-state.txt"
DIRTY=$(git status --porcelain | wc -l | tr -d ' ')
if [ "$DIRTY" -gt 0 ]; then
echo "❌ 工作区有未提交改动($DIRTY 个文件)—— 噪声底线必须在干净代码上测"
exit 1
fi
# ⑤ 跑 N 轮
echo
for i in $(seq 1 "$ROUNDS"); do
echo "───────── 第 $i / $ROUNDS 轮 ─────────"
# 重启服务(关键!)
scripts/restart-app.sh "$DIR/results/gc-$i.log"
sleep 5
# 独立预热(不计入统计)
BASE_URL=http://127.0.0.1:8080 k6 run --quiet --vus 20 --duration 60s \
loadtest/profile-constant.js > /dev/null
echo " 预热完成"
# 正式测量
BASE_URL=http://127.0.0.1:8080 RATE="$RATE" DURATION="$DURATION" \
k6 run --summary-export="$DIR/results/round-$i.json" \
loadtest/profile-constant.js > "$DIR/results/round-$i-stdout.txt" 2>&1
# 打印本轮关键值
python3 - "$DIR/results/round-$i.json" <<'PY'
import json, sys
m = json.load(open(sys.argv[1]))["metrics"]
d = m["http_req_duration"]
print(f" QPS={m['http_reqs']['rate']:.1f} P50={d['med']:.1f} P95={d['p(95)']:.1f} "
f"P99={d['p(99)']:.1f} 错误率={m['http_req_failed']['rate']:.2%}")
PY
done
kill $EVENTS_PID 2>/dev/null || true
echo
echo "═══ 测量完成,开始分析 ═══"
python3 tools/noise_floor.py "$DIR/results"
二、噪声底线计算与 MDD 推导
# tools/noise_floor.py <results_dir>
"""
计算噪声底线与最小可检测差异(MDD)。
核心逻辑:
① 从 N 轮数据里算出每个指标的波动范围(最大偏离中位数的百分比)
② 噪声底线 = 波动范围 × 1.5(安全系数)
③ MDD = 噪声底线 × 2(因为两组数据相减,噪声叠加)
④ 检查是否存在趋势漂移(单调上升/下降)
"""
import json
import pathlib
import statistics as st
import sys
def load_rounds(d):
rounds = []
for f in sorted(pathlib.Path(d).glob("round-*.json")):
try:
m = json.loads(f.read_text())["metrics"]
dr = m["http_req_duration"]
rounds.append({
"file": f.name,
"qps": m["http_reqs"]["rate"],
"p50": dr.get("med", 0),
"p95": dr.get("p(95)", 0),
"p99": dr.get("p(99)", 0),
"err": m["http_req_failed"]["rate"],
})
except Exception as e:
print(f"⚠️ 跳过 {f}: {e}")
return rounds
def trend(values):
"""判断是否有单调趋势:返回 (斜率方向, 相关系数绝对值)"""
n = len(values)
if n < 3:
return "样本不足", 0.0
xs = list(range(n))
mx, my = sum(xs) / n, sum(values) / n
num = sum((x - mx) * (y - my) for x, y in zip(xs, values))
den_x = sum((x - mx) ** 2 for x in xs)
den_y = sum((y - my) ** 2 for y in values)
if den_x == 0 or den_y == 0:
return "无趋势", 0.0
slope = num / den_x
r = abs(num / ((den_x * den_y) ** 0.5))
direction = "上升" if slope > 0 else "下降"
return f"{direction}(r={r:.2f})", r
def analyze(rounds):
if len(rounds) < 3:
print("❌ 至少需要 3 轮数据(推荐 5–10 轮)")
return None
print("═" * 76)
print(f"噪声底线分析({len(rounds)} 轮,代码未改动)")
print("═" * 76)
print()
# 逐轮列出
print(f"{'轮次':<5}{'QPS':>10}{'P50':>10}{'P95':>10}{'P99':>10}{'错误率':>10} 备注")
for i, r in enumerate(rounds, 1):
print(f"{i:<5}{r['qps']:>10.1f}{r['p50']:>10.1f}{r['p95']:>10.1f}{r['p99']:>10.1f}{r['err']:>10.2%}")
print()
# 计算波动范围
print(f"{'指标':<8}{'中位数':>10}{'最小':>10}{'最大':>10}{'波动范围':>12}{'CV':>8} 趋势")
print("-" * 76)
results = {}
for key, label in [("qps", "QPS"), ("p50", "P50"), ("p95", "P95"), ("p99", "P99")]:
vals = [r[key] for r in rounds]
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
tr, r = trend(vals)
results[key] = {"median": med, "swing": swing, "cv": cv, "trend_r": r, "values": vals}
print(f"{label:<8}{med:>10.2f}{lo:>10.2f}{hi:>10.2f}{swing:>11.1f}%{cv:>7.1f}% {tr}")
print()
# 噪声底线与 MDD
print("═" * 76)
print("噪声底线(波动范围 × 1.5 安全系数)与最小可检测差异(MDD = 噪声底线 × 2)")
print("═" * 76)
print()
print(f"{'指标':<8}{'波动范围':>12}{'噪声底线':>12}{'MDD':>12} 含义")
print("-" * 76)
for key, label in [("qps", "QPS"), ("p50", "P50"), ("p95", "P95"), ("p99", "P99")]:
r = results[key]
noise = r["swing"] * 1.5
mdd = noise * 2
meaning = f"声称改善 < {mdd:.0f}% 无法证实"
print(f"{label:<8}{r['swing']:>11.1f}%{noise:>11.1f}%{mdd:>11.1f}% {meaning}")
print()
# 质量评估
print("═" * 76)
print("环境质量评估")
print("═" * 76)
print()
max_cv = max(r["cv"] for r in results.values())
p99_cv = results["p99"]["cv"]
if p99_cv < 3:
print(f"✅ P99 的轮间 CV = {p99_cv:.1f}% —— 环境很干净")
print(" → 可以检测 5% 级别的差异(但仍需统计检验确认)")
elif p99_cv < 5:
print(f"✅ P99 的轮间 CV = {p99_cv:.1f}% —— 环境可接受")
print(" → 可以检测 10% 级别的差异")
elif p99_cv < 10:
print(f"⚠️ P99 的轮间 CV = {p99_cv:.1f}% —— 环境噪声偏大")
print(" → 只能判断 20% 以上的差异;建议先改善环境")
else:
print(f"❌ P99 的轮间 CV = {p99_cv:.1f}% —— 环境噪声过大")
print(" → 任何结论都不可信;必须先修环境(第 5.4 节)")
print()
# 趋势漂移检查
print("趋势漂移检查(单调上升/下降说明环境在变,不是噪声):")
drift_found = False
for key, label in [("qps", "QPS"), ("p50", "P50"), ("p95", "P95"), ("p99", "P99")]:
r = results[key]
if r["trend_r"] > 0.7:
drift_found = True
print(f" ❌ {label}: 存在明显趋势(r={r['trend_r']:.2f})")
print(f" 值 = {[round(v, 1) for v in r['values']]}")
print(f" → 可能原因:数据量增长 / 缓存污染 / 内存泄漏 / 机器降频")
print(f" → 必须先解决漂移,再重测噪声底线")
if not drift_found:
print(" ✅ 未检测到趋势漂移")
print()
print("═" * 76)
print("如何使用这个结果:")
print("═" * 76)
print()
print(" ① 写进实验档案(docs/experiments/E0x-noise/README.md)")
print(" ② 后续任何优化实验,都拿它的差异与这里的 MDD 对比")
print(" ③ CI 门禁阈值 = 噪声底线 × 1.5~2(不能直接等于噪声底线)")
print(" ④ 每季度或环境重大变更后重测")
print("═" * 76)
return results
if __name__ == "__main__":
d = sys.argv[1] if len(sys.argv) > 1 else "docs/experiments/E05-noise/results"
analyze(load_rounds(d))
三、预期输出
════════════════════════════════════════════════════════════════════════════
噪声底线分析(10 轮,代码未改动)
════════════════════════════════════════════════════════════════════════════
轮次 QPS P50 P95 P99 错误率 备注
1 498.2 32.1 68.4 93.2 0.00%
2 499.1 33.0 70.2 96.1 0.00%
3 497.5 32.4 67.9 94.0 0.00%
4 498.8 34.2 74.1 103.5 0.00%
5 499.4 32.8 69.3 95.2 0.00%
...
指标 中位数 最小 最大 波动范围 CV 趋势
----------------------------------------------------------------------------
QPS 498.80 497.50 499.70 0.2% 0.1% 无趋势
P50 32.80 32.10 34.20 4.3% 1.8% 无趋势
P95 69.20 67.90 74.10 7.1% 2.7% 无趋势
P99 95.10 93.20 103.50 8.8% 3.1% 无趋势
════════════════════════════════════════════════════════════════════════════
噪声底线(波动范围 × 1.5 安全系数)与最小可检测差异(MDD = 噪声底线 × 2)
════════════════════════════════════════════════════════════════════════════
指标 波动范围 噪声底线 MDD 含义
----------------------------------------------------------------------------
QPS 0.2% 0.3% 0.7% 声称改善 < 1% 无法证实
P50 4.3% 6.5% 13.0% 声称改善 < 13% 无法证实
P95 7.1% 10.7% 21.4% 声称改善 < 21% 无法证实
P99 8.8% 13.2% 26.4% 声称改善 < 26% 无法证实
════════════════════════════════════════════════════════════════════════════
环境质量评估
════════════════════════════════════════════════════════════════════════════
✅ P99 的轮间 CV = 3.1% —— 环境可接受
→ 可以检测 10% 级别的差异
这份输出最重要的两个数字:
| 指标 | 含义 |
|---|---|
| 噪声底线 P99 = ±13.2% | 低于这个幅度的差异,无法区分是噪声还是真实变化 |
| MDD P99 = ±26.4% | 声称"改善了 X%“时,X 必须 > 26% 才有足够说服力 |
这就是本章要你面对的现实:在这个环境里,26% 以内的优化你都只能"相信"而不能"证明”。这不是悲观,而是诚实。
四、回归门禁(基于噪声底线)
# tools/regression_gate.py <current.json> <baseline.json> <noise_floor_json>
"""
CI 性能门禁:对比当前结果与基线,超过阈值则失败退出。
设计原则:
① 相对比较(不清绝对阈值)—— 换 runner 型号不会全线失败
② 阈值 = 噪声底线 × 1.5~2 —— 必须高于噪声,否则天天误报
③ 只对稳定指标设门禁 —— max/P999 波动大,只做趋势观察
"""
import json
import sys
def load(path):
m = json.loads(open(path).read())["metrics"]
return {
"qps": m["http_reqs"]["rate"],
"p50": m["http_req_duration"]["med"],
"p95": m["http_req_duration"]["p(95)"],
"p99": m["http_req_duration"]["p(99)"],
"err": m["http_req_failed"]["rate"],
}
def main():
if len(sys.argv) < 4:
print("usage: regression_gate.py <current.json> <baseline.json> <noise_floor.json>")
print()
print("noise_floor.json 格式:")
print(' {"qps": 0.3, "p50": 6.5, "p95": 10.7, "p99": 13.2} (百分比)')
sys.exit(1)
current = load(sys.argv[1])
baseline = load(sys.argv[2])
noise = json.loads(open(sys.argv[3]).read())
SAFETY = 1.5 # 阈值 = 噪声底线 × SAFETY
failures = []
warnings = []
print("═" * 70)
print("性能回归门禁")
print("═" * 70)
print()
print(f"{'指标':<8}{'基线':>10}{'当前':>10}{'变化':>10}{'阈值':>10} 判定")
print("-" * 70)
# 延迟和错误率:越低越好
for key, label in [("p50", "P50"), ("p95", "P95"), ("p99", "P99")]:
if key not in noise:
continue
b, c = baseline[key], current[key]
delta = (c - b) / b * 100 if b else 0
threshold = noise[key] * SAFETY
if delta > threshold:
verdict = "❌ 退化"
failures.append(f"{label} 退化 {delta:+.1f}%(阈值 +{threshold:.1f}%)")
elif delta > threshold * 0.7:
verdict = "⚠️ 接近阈值"
warnings.append(f"{label} 变化 {delta:+.1f}%")
else:
verdict = "✅"
print(f"{label:<8}{b:>10.1f}{c:>10.1f}{delta:>9.1f}%{threshold:>9.1f}% {verdict}")
# QPS:越高越好(下降是退化)
if "qps" in noise:
b, c = baseline["qps"], current["qps"]
delta = (c - b) / b * 100 if b else 0
threshold = noise["qps"] * SAFETY
if delta < -threshold:
verdict = "❌ 退化"
failures.append(f"QPS 下降 {delta:+.1f}%")
else:
verdict = "✅"
print(f"{'QPS':<8}{b:>10.1f}{c:>10.1f}{delta:>9.1f}%{threshold:>9.1f}% {verdict}")
# 错误率:绝对值判断
if current["err"] > 0.01:
failures.append(f"错误率 {current['err']:.2%} > 1%")
print(f"{'错误率':<8}{baseline['err']:>9.2%}{current['err']:>10.2%}{'—':>10}{'1.0%':>10} ❌ 超标")
print()
print("═" * 70)
if failures:
print("❌ 门禁失败:")
for f in failures:
print(f" - {f}")
print()
print("处理方式:")
print(" ① 确认是否为真实退化(本地复现一次)")
print(" ② 若确认是噪声导致的误报,先提高阈值或改用固定 runner")
print(" ③ 【不要】在未验证的情况下直接更新基线")
sys.exit(1)
elif warnings:
print("⚠️ 门禁通过,但有接近阈值的指标:")
for w in warnings:
print(f" - {w}")
print()
print("建议观察几轮,若持续接近阈值则需排查。")
sys.exit(0)
else:
print("✅ 门禁通过,无性能退化")
sys.exit(0)
if __name__ == "__main__":
main()
五、动手改造
| 改动 | 观察什么 |
|---|---|
把 ROUNDS 从 10 改成 5 |
CV 与波动范围如何变化?(样本少 → 波动范围可能偏小,低估噪声) |
| 故意在某一轮开后台上传任务 | 波动范围变大,噪声底线上升——理解"环境决定精度" |
把 SAFETY 从 1.5 改成 1.0 |
门禁开始误报——验证"阈值必须高于噪声" |
把 SAFETY 改成 3.0 |
几乎不误报,但会漏掉 15% 的真实退化 |
用 noise_floor.py 的输出作为 regression_gate.py 的输入 |
打通「测底线 → 设门禁」的完整流程 |
| 连续跑 3 次噪声测量 | 看三次的噪声底线是否稳定(如果不稳定,说明环境本身在变) |
六、这段代码的局限
trend()用皮尔逊相关系数判断趋势:对非线性趋势(比如先降后升)不敏感。- 波动范围用「最大偏离中位数」:对异常值敏感——如果有一轮异常,波动范围会被拉大。这其实是保守的(宁可高估噪声),但如果异常轮有明确成因(第 5.3 节),剔除后重算会更准。
- MDD = 噪声底线 × 2 是近似:严格的 MDD 计算需要考虑样本量与检验的统计功效(power),但对工程实践来说这个近似够了。
- 门禁只对比单轮结果:更严谨的做法是对比多轮的中位数(避免单轮噪声触发门禁),但那需要 CI 里跑多轮(成本高)。
- 噪声底线会变:换机器、换时间段、换数据量都可能改变它。建议每季度重测。