8.3 配套代码：全链路性能门禁
对应小节：8.3 全链路性能门禁 一份完整的门禁脚本，含饱和度检查（最容易被忽略但最有价值的一项）。
一、完整的全链路门禁脚本
#!/usr/bin/env bash
# tools/e2e-perf-gate.sh <VERSION_TAG>
#
# 全链路性能门禁：发布候选阶段运行。
#
# 前置条件（缺一不可，否则结果不可信）：
# ① 专用环境（固定规格、独占、压测客户端独立）
# ② 固定的数据集（每次重置到同一状态）
# ③ 固定的 JVM 参数与容器配置
set -euo pipefail
VERSION="${1:?usage: e2e-perf-gate.sh <version_tag>}"
EXP_ID="gate-${VERSION}"
DIR="docs/experiments/${EXP_ID}"
mkdir -p "$DIR/results"
echo "═══════════════════════════════════════════════════════════════"
echo "全链路性能门禁：$VERSION"
echo "═══════════════════════════════════════════════════════════════"
echo
# ═══ 1/7 环境检查（不通过就终止）═══
echo "【1/7】环境检查"
tools/check-environment.sh "$EXP_ID" > /dev/null 2>&1 || true
if grep -q "❌" "$DIR/results/env-check.txt" 2>/dev/null; then
echo "❌ 环境检查未通过 —— 本次结果不可信，终止"
grep "❌" "$DIR/results/env-check.txt" | sed 's/^/ /'
exit 2
fi
tools/collect-env.sh "$EXP_ID" > /dev/null 2>&1 || true
echo " ✅ 环境检查通过"
echo
# ═══ 2/7 记录代码版本 ═══
echo "【2/7】记录代码版本"
{
echo "commit=$(git rev-parse HEAD)"
echo "tag=$VERSION"
echo "dirty=$(git status --porcelain | wc -l | tr -d ' ')"
} | tee "$DIR/results/code-state.txt"
echo
# ═══ 3/7 重置数据集并重启 ═══
echo "【3/7】重置数据集"
scripts/reset-dataset.sh
scripts/restart-app.sh "$DIR/results/gc.log"
sleep 10
echo " ✅ 服务已重启"
echo
# ═══ 4/7 预热（不计入统计）═══
echo "【4/7】预热 60 秒"
BASE_URL="$TARGET_URL" k6 run --quiet --vus 20 --duration 60s \
loadtest/profile-constant.js > /dev/null 2>&1
echo " ✅ 预热完成"
echo
# ═══ 5/7 测量 3 轮 ═══
echo "【5/7】测量 3 轮（每轮 3 分钟）"
for i in 1 2 3; do
echo " ── 第 $i 轮 ──"
BASE_URL="$TARGET_URL" RATE="${TARGET_QPS:-500}" DURATION=3m \
k6 run --summary-export="$DIR/results/round-$i.json" \
loadtest/profile-constant.js > "$DIR/results/round-$i-stdout.txt" 2>&1
# ⭐ 关键：同时采集饱和度指标（不只是延迟）
curl -s --max-time 5 "${TARGET_URL%/}/metrics" > "$DIR/results/metrics-$i.txt" 2>/dev/null || true
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} "
f"P95={d['p(95)']:.1f} P99={d['p(99)']:.1f} 错误率={m['http_req_failed']['rate']:.2%}")
PY
done
echo
# ═══ 6/7 检查（四项）═══
echo "【6/7】门禁检查"
python3 tools/e2e_gate_check.py "$DIR/results" perf/e2e-baseline.json
GATE_RESULT=$?
echo
# ═══ 7/7 结论 ═══
echo "【7/7】结论"
if [ $GATE_RESULT -eq 0 ]; then
echo "✅ 性能门禁通过"
echo
echo "后续："
echo " ① 把结果归档：cp -r $DIR perf/gate-history/$VERSION"
echo " ② 继续发布流程（进入金丝雀阶段）"
echo " ③ 发布完成后，把本次结果更新为新的候选基线"
else
echo "❌ 性能门禁失败"
echo
echo "后续："
echo " ① 查看 $DIR/results/ 下的详细数据"
echo " ② 按第 8.3 节的决策树处理："
echo " 查环境 → 判幅度 → 定位（第 6 章）→ 决策"
echo " ③ ⚠️ 不要直接更新基线"
exit 1
fi
二、四项检查的实现（含饱和度）
# tools/e2e_gate_check.py <RESULTS_DIR> <BASELINE_JSON>
"""
全链路门禁的检查逻辑。
四项检查：
① SLO 达标（延迟 + 错误率）
② 拐点未后移（如果有拐点数据）
③ 饱和度正常 ⭐（最容易被忽略）
④ 资源无异常（内存、连接）
"""
import json
import pathlib
import re
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({
"qps": m["http_reqs"]["rate"],
"p50": dr["med"], "p95": dr["p(95)"], "p99": dr["p(99)"],
"err": m["http_req_failed"]["rate"],
})
except Exception:
pass
return rounds
def load_metrics_snapshot(d, round_no):
"""从 /metrics 快照里提取饱和度指标"""
p = pathlib.Path(d) / f"metrics-{round_no}.txt"
if not p.exists():
return {}
text = p.read_text(errors="ignore")
out = {}
for key, pattern in [
("pool_pending", r"^db_pool_pending\s+([\d.]+)"),
("pool_active", r"^db_pool_active\s+([\d.]+)"),
("pool_total", r"^db_pool_total\s+([\d.]+)"),
("queue_depth", r"^executor_queue_depth\s+([\d.]+)"),
("heap_used_mb", r"^jvm_memory_used_bytes\{.*heap.*\}\s+([\d.]+)"),
]:
m = re.search(pattern, text, re.MULTILINE)
if m:
v = float(m.group(1))
out[key] = v / 1048576 if key == "heap_used_mb" else v
return out
def main(results_dir, baseline_path):
rounds = load_rounds(results_dir)
if not rounds:
print("❌ 没有找到测量数据")
sys.exit(2)
baseline = {}
bp = pathlib.Path(baseline_path)
if bp.exists():
baseline = json.loads(bp.read_text())
problems = []
warnings = []
passed = []
print("═" * 84)
print("全链路门禁检查（四项）")
print("═" * 84)
print()
# ── ① SLO 达标 ──────────────────────────────────────────
print("【① SLO 达标】")
p99 = st.median([r["p99"] for r in rounds])
err = st.median([r["err"] for r in rounds])
p99_slo = baseline.get("slo_p99_ms", 200)
err_slo = baseline.get("slo_error_rate", 0.001)
print(f" P99 (中位数): {p99:.1f} ms （SLO: < {p99_slo} ms）")
print(f" 错误率 : {err:.2%} （SLO: < {err_slo:.2%}）")
if p99 > p99_slo:
problems.append(f"SLO 违约：P99 {p99:.1f}ms > {p99_slo}ms")
print(" ❌ SLO 违约")
elif p99 > p99_slo * 0.9:
warnings.append(f"P99 接近 SLO（{p99:.1f}/{p99_slo}ms）")
print(" ⚠️ 接近 SLO 上限")
else:
passed.append("SLO 达标")
print(" ✅ 达标")
if err > err_slo:
problems.append(f"错误率超标：{err:.2%} > {err_slo:.2%}")
print(" ❌ 错误率超标")
print()
# ── ② 基线对比（延迟与吞吐）──────────────────────────────
print("【② 与基线对比】")
if baseline.get("p99"):
base_p99 = baseline["p99"]
delta = (p99 - base_p99) / base_p99 * 100
threshold = baseline.get("p99_threshold_pct", 20)
print(f" P99: 基线 {base_p99:.1f}ms → 当前 {p99:.1f}ms （{delta:+.1f}%，阈值 ±{threshold}%）")
if delta > threshold:
problems.append(f"P99 退化 {delta:+.1f}%")
print(" ❌ 退化超阈值")
else:
passed.append("P99 无退化")
print(" ✅")
else:
print(" （基线中无 P99 数据，跳过）")
print()
# ── ③ 饱和度 ⭐（最容易被忽略）──────────────────────────
print("【③ 饱和度检查】⭐ 领先指标")
pending_vals, queue_vals = [], []
for i in range(1, len(rounds) + 1):
snap = load_metrics_snapshot(results_dir, i)
if "pool_pending" in snap:
pending_vals.append(snap["pool_pending"])
if "queue_depth" in snap:
queue_vals.append(snap["queue_depth"])
if pending_vals:
max_pending = max(pending_vals)
print(f" 连接池 pending 最大值: {max_pending:.0f}")
if max_pending > 0:
problems.append(f"连接池出现排队（pending 最大 {max_pending:.0f}）")
print(" ❌ 存在排队 —— 系统余量已被耗尽")
print(" 即使延迟达标，也应拒绝通过（第 8.3 节）")
else:
passed.append("连接池无排队")
print(" ✅ 无排队")
else:
print(" （未采集到 pending 指标 —— 建议在 metrics 快照里加上）")
if queue_vals:
max_queue = max(queue_vals)
print(f" 线程池队列最大深度: {max_queue:.0f}")
if max_queue > 100:
problems.append(f"线程池队列积压（{max_queue:.0f}）")
print(" ❌ 队列积压")
else:
passed.append("队列无积压")
print(" ✅ 队列正常")
print()
# ── ④ 资源趋势（泄漏检测）──────────────────────────────
print("【④ 资源趋势】")
heaps = []
for i in range(1, len(rounds) + 1):
snap = load_metrics_snapshot(results_dir, i)
if "heap_used_mb" in snap:
heaps.append(snap["heap_used_mb"])
if len(heaps) >= 3:
print(f" 堆使用: {[f'{h:.0f}' for h in heaps]} MB")
if heaps[-1] > heaps[0] * 1.5:
warnings.append("堆使用有上升趋势（建议做浸泡测试）")
print(" ⚠️ 有上升趋势 —— 建议单独做浸泡测试确认（第 3 章 3.5）")
else:
passed.append("资源稳定")
print(" ✅ 稳定")
else:
print(" （轮数不足或未采集，跳过）")
print()
# ── 汇总 ────────────────────────────────────────────────
print("═" * 84)
if problems:
print(f"❌ 门禁失败（{len(problems)} 项问题）")
for p in problems:
print(f" - {p}")
print()
print("处理方式：")
print(" ① 检查环境（env-check.txt 有没有 ❌）")
print(" ② 判断幅度（是噪声还是真退化）")
print(" ③ 定位（用第 6 章的延迟分解与火焰图）")
print(" ④ 不要直接更新基线")
sys.exit(1)
elif warnings:
print(f"⚠️ 门禁通过，但有 {len(warnings)} 项警告")
for w in warnings:
print(f" - {w}")
sys.exit(0)
else:
print("✅ 门禁通过（四项检查全部通过）")
sys.exit(0)
if __name__ == "__main__":
main(sys.argv[1], sys.argv[2] if len(sys.argv) > 2 else "perf/e2e-baseline.json")
三、基线文件格式
{
"version": "v2.3.0",
"date": "2025-01-10",
"environment": {
"instance": "4C8G",
"replicas": 3,
"jvm_args": "-Xms2g -Xmx2g -XX:+UseG1GC",
"container_cpu_limit": "4",
"container_mem_limit": "8Gi"
},
"slo_p99_ms": 200,
"slo_error_rate": 0.001,
"p99": 95.2,
"p95": 42.1,
"p50": 12.3,
"qps": 500,
"p99_threshold_pct": 20,
"saturation": {
"pool_pending_max": 0,
"queue_depth_max": 0
},
"notes": "拐点约 900 RPS（E02 实验）；数据量 100 万行"
}
注意 saturation 这一栏：把基线时的饱和度也记下来——这样当前值不为 0 时就能立刻发现"余量在减少"。
四、拐点复测（可选的高阶检查）
#!/usr/bin/env bash
# tools/retest-knee.sh <EXP_ID>
#
# 复测拐点（比常规门禁更耗时，用于发布候选或季度复测）
set -uo pipefail
EXP_ID="${1:?usage: retest-knee.sh <exp_id>}"
DIR="docs/experiments/${EXP_ID}"
mkdir -p "$DIR/results"
echo "═══ 拐点复测 ═══"
echo " ⚠️ 这将运行完整的阶梯加压（约 30–60 分钟）"
echo
scripts/restart-app.sh "$DIR/results/gc.log"
sleep 5
BASE_URL="$TARGET_URL" k6 run --quiet --vus 20 --duration 60s loadtest/profile-constant.js > /dev/null
k6 run --out json="$DIR/results/k6-staircase.json" \
--summary-export="$DIR/results/k6-staircase-summary.json" \
loadtest/profile-staircase.js
echo
python3 tools/find-knee.py "$DIR/results/k6-staircase.json" "100,200,400,700,1000,1500,2000"
echo
echo "与基线拐点对比："
python3 - <<'PY'
import json, pathlib
base = pathlib.Path("perf/e2e-baseline.json")
if base.exists():
b = json.loads(base.read_text())
print(f" 基线拐点: {b.get('knee_qps', '未记录')} RPS")
print()
print(" 如果当前拐点低于基线的 90%，说明容量退化了 ——")
print(" 即使当前 SLO 达标，也应该评估是否需要扩容。")
else:
print(" （基线中未记录拐点）")
PY
为什么拐点复测重要：
❌ 只看 SLO：P99 = 190ms < 200ms ✅ 通过
但拐点从 900 RPS 降到 700 RPS
→ 虽然现在达标，但容量余量少了很多
→ 下次流量上涨就会出问题
五、动手改造
| 改动 | 观察什么 |
|---|---|
| 在 Lab 6 的埋雷服务上跑门禁 | 它会检出 P99 超标 + 连接池排队 |
| 故意把服务配置成"延迟达标但 pending > 0" | 看门禁是否会拒绝通过——验证饱和度检查的价值 |
| 去掉第 ③ 项饱和度检查 | 你会漏掉"余量耗尽"的信号——理解为什么它最容易被忽略却最有价值 |
把 p99_threshold_pct 从 20 改成 5 |
误报率上升——验证"阈值要高于噪声" |
| 加上拐点复测 | 观察"延迟达标但容量退化"的情况 |
六、这段代码的局限
- 需要专用环境：在共享环境上跑这些检查只会得到噪声（第 8.3 节的三个必要条件）。
e2e_gate_check.py的饱和度提取依赖指标名：如果你的指标名不同（如hikaricp_connections_pending），需要调整正则。- 拐点复测成本高（30–60 分钟）：不建议每次发布都做，用在发布候选与季度复测。
- 基线里的
knee_qps需要人工维护：它来自第 1 章 1.6 节的容量曲线实验。 - 样本量问题：3 轮 × 3 分钟 = 9 分钟，对 P99 的估计仍然不够稳；如果噪声大，需要更长时间或更多轮。