8.8 配套代码：评审清单、周报与业务语言
对应小节：8.8 性能评审与组织落地 三部分：① PR 模板（关键 5 项）；② 性能周报生成器；③ 业务语言翻译器。
一、PR 模板（只有关键 5 项）
<!-- .github/pull_request_template.md -->
## 变更说明
<!-- 简要描述这个 PR 做了什么 -->
## 性能影响
> ⚠️ 只需回答**与本次改动相关**的项。无关的项请勾选「无性能影响」。
>
> 详细清单见 `docs/performance-review-checklist.md`（按需查阅）。
- [ ] **无性能影响**（纯文案 / 纯样式 / 纯配置注释等，可跳过下面所有项）
### 如果改动涉及数据访问、外部调用或热路径：
- [ ] **IO / RPC 次数**：这个改动增加了多少次数据库查询、外部调用或序列化？
→ 如果是循环里的调用（N+1），请说明为什么不能批处理
→ 增加了 ____ 次
- [ ] **资源边界**：引入了新的缓存 / 队列 / 线程池 / 连接池吗？
→ 如果有，它们**有容量上限吗**？（无界的缓存/队列会导致 OOM）
→ 上限是 ____
- [ ] **超时与失败**：新增的下游调用有超时吗？
→ 超时值必须**小于上游给的预算**（逐层递减）
→ 超时是 ____ ms，上游预算是 ____ ms
→ 失败时会快速失败还是排队？
- [ ] **可观测性**：新增的接口/依赖有指标吗？
→ 至少要有：延迟（直方图）、错误率
→ 出问题时能否回答「这个接口的 P99 里，数据库占多少毫秒」？
- [ ] **性能验证**：关键路径做了 before/after 对比吗？
→ 附实验编号（如 E08-xxx），或说明为什么不需要
→ 实验编号：____
## 需要同步更新的文档
- [ ] 容量表（`docs/capacity.md`）——如果改变了资源用量
- [ ] SLO 定义（`slo/slos.yaml`）——如果新增了接口或改变了目标
- [ ] 告警规则（`observability/alerts.yml`）——如果新增了关键路径
- [ ] 不需要更新
## 其他
- [ ] 已运行单元测试
- [ ] 已运行 lint / format
这个模板的设计原则：
| 设计 | 理由 |
|---|---|
| 只有 5 项 | 长了会被机械全勾（第 8.8 节） |
| 「无性能影响」放在第一项 | 允许快速跳过（避免形式主义） |
| 每项都有追问（“增加了多少次”） | 逼出具体答案，而不是打勾了事 |
| 明确列出连带更新 | 防止漏掉容量表/SLO/告警 |
二、性能周报生成器
# tools/weekly-perf-report.py <PROMETHEUS_URL> [--output report.md]
"""
从 Prometheus 生成性能周报。
设计意图：让性能「可见」——每周在团队频道发一次，
而不是等到出问题时才谈性能。
"""
import argparse
import json
import sys
import urllib.parse
import urllib.request
from datetime import datetime, timedelta
def query(prom, expr):
url = f"{prom}/api/v1/query?" + urllib.parse.urlencode({"query": expr})
try:
with urllib.request.urlopen(url, timeout=10) as r:
data = json.load(r)
result = data["data"]["result"]
return float(result[0]["value"][1]) if result else None
except Exception:
return None
def fmt(v, unit="", precision=1):
if v is None:
return "n/a"
return f"{v:.{precision}f}{unit}"
def main(prom, output):
now = datetime.now()
week_ago = now - timedelta(days=7)
lines = [
f"## 性能周报（{week_ago.strftime('%m-%d')} ~ {now.strftime('%m-%d')}）",
"",
]
# ── 核心指标 ────────────────────────────────────────────
metrics = [
("P99 延迟", 'histogram_quantile(0.99, sum by (le) (rate(http_request_duration_seconds_bucket[7d])))', "ms", 1000),
("P50 延迟", 'histogram_quantile(0.50, sum by (le) (rate(http_request_duration_seconds_bucket[7d])))', "ms", 1000),
("错误率", 'sum(rate(http_requests_total{status=~"5.."}[7d])) / sum(rate(http_requests_total[7d]))', "%", 100),
("峰值 QPS", 'max_over_time(sum(rate(http_requests_total[5m]))[7d:5m])', "", 1),
]
lines += ["| 指标 | 本周 | 上周 | 变化 |", "| --- | --- | --- | --- |"]
for label, expr, unit, scale in metrics:
# 本周
cur = query(prom, expr)
# 上周（用 offset）
last = query(prom, expr.replace("[7d]", "[7d] offset 7d"))
if cur is not None:
cur *= scale
if last is not None:
last *= scale
if cur is not None and last is not None and last != 0:
delta = (cur - last) / last * 100
flag = "⚠️" if abs(delta) > 10 else ""
delta_str = f"{delta:+.1f}% {flag}"
else:
delta_str = "—"
lines.append(f"| {label} | {fmt(cur, unit)} | {fmt(last, unit)} | {delta_str} |")
lines.append("")
# ── 错误预算 ────────────────────────────────────────────
budget = query(prom, "slo:orders_availability:budget_remaining")
if budget is not None:
lines += [
"### 错误预算",
"",
f"- 剩余：**{budget * 100:.0f}%**（30 天窗口）",
]
if budget < 0.3:
lines.append("- ⚠️ **预算不足 30% —— 建议暂停非必要的功能发布**")
elif budget < 0.5:
lines.append("- ⚠️ 预算不足 50% —— 建议谨慎发布")
else:
lines.append("- ✅ 预算充足")
lines.append("")
# ── 饱和度（领先指标）──────────────────────────────────
lines += [
"### 饱和度（领先指标）",
"",
"| 指标 | 本周峰值 | 状态 |",
"| --- | --- | --- |",
]
sat_metrics = [
("连接池 pending", 'max_over_time(db_pool_pending[7d])', 0, "= 0"),
("队列深度峰值", 'max_over_time(executor_queue_depth[7d])', 100, "< 100"),
("CPU 节流峰值", 'max_over_time(rate(container_cpu_cfs_throttled_periods_total[5m]) / rate(container_cpu_cfs_periods_total[5m])[7d:5m])', 0.25, "< 25%"),
]
for label, expr, threshold, good_desc in sat_metrics:
v = query(prom, expr)
if v is None:
lines.append(f"| {label} | n/a | — |")
continue
ok = v <= threshold
lines.append(f"| {label} | {v:.2f} | {'✅' if ok else '❌'} 应 {good_desc} |")
lines.append("")
# ── 需要关注的事 ────────────────────────────────────────
lines += ["### 需要关注", ""]
concerns = []
p99 = query(prom, metrics[0][1])
if p99 and p99 * 1000 > 150:
concerns.append(f"P99 = {p99*1000:.0f}ms，已接近 SLO（200ms）")
pending = query(prom, 'max_over_time(db_pool_pending[7d])')
if pending and pending > 0:
concerns.append(f"本周出现过连接池排队（峰值 {pending:.0f}）")
throttled = query(prom, sat_metrics[2][1])
if throttled and throttled > 0.1:
concerns.append(f"容器 CPU 节流峰值 {throttled*100:.0f}%")
if concerns:
for c in concerns:
lines.append(f"- ⚠️ {c}")
else:
lines.append("- ✅ 本周无异常")
lines.append("")
lines += [
"---",
"",
f"<sub>自动生成于 {now.strftime('%Y-%m-%d %H:%M')} · "
f"[Grafana 看板](http://grafana/d/slo) · [容量表](docs/capacity.md)</sub>",
]
report = "\n".join(lines)
print(report)
if output:
import pathlib
pathlib.Path(output).write_text(report, encoding="utf-8")
print(f"\n✅ 已写入 {output}", file=sys.stderr)
if __name__ == "__main__":
ap = argparse.ArgumentParser()
ap.add_argument("prometheus", nargs="?", default="http://localhost:9090")
ap.add_argument("--output")
args = ap.parse_args()
main(args.prometheus, args.output)
预期输出：
## 性能周报（01-08 ~ 01-15）
| 指标 | 本周 | 上周 | 变化 |
| --- | --- | --- | --- |
| P99 延迟 | 98.3ms | 95.2ms | +3.3% |
| P50 延迟 | 12.4ms | 12.3ms | +0.8% |
| 错误率 | 0.04% | 0.05% | -20.0% ⚠️ |
| 峰值 QPS | 3120 | 2980 | +4.7% |
### 错误预算
- 剩余：**78%**（30 天窗口）
- ✅ 预算充足
### 饱和度（领先指标）
| 指标 | 本周峰值 | 状态 |
| --- | --- | --- |
| 连接池 pending | 0.00 | ✅ 应 = 0 |
| 队列深度峰值 | 12.00 | ✅ 应 < 100 |
| CPU 节流峰值 | 0.08 | ✅ 应 < 25% |
### 需要关注
- ⚠️ P99 = 98ms，已接近 SLO（200ms）
注意「P99 +3.3%」这个变化——它在噪声范围内，所以周报里不加"⚠️"（脚本只在 > 10% 时标记）。但它连续几周上升就是趋势，值得在周报正文里提一句。
三、业务语言翻译器
# tools/translate-for-business.py <INPUT.json>
"""
把技术指标翻译成业务语言。
输入：
{
"service": "订单服务",
"daily_requests": 50000,
"current": {"p50_ms": 12, "p99_ms": 400, "error_rate": 0.008},
"target": {"p50_ms": 12, "p99_ms": 200, "error_rate": 0.005},
"monthly_revenue_per_order": 50,
"avg_order_value": 200
}
"""
import json
import sys
def translate(d):
cur, tgt = d["current"], d["target"]
daily = d["daily_requests"]
# 计算"受影响的人数"
slow_pct = 0.01 # P99 意味着 1% 的请求
slow_users = int(daily * slow_pct)
failed_users = int(daily * cur["error_rate"])
target_failed = int(daily * tgt["error_rate"])
# 计算"感知阈值"（200ms 是人类可感知卡顿的门槛，400ms 是明显卡顿）
p99 = cur["p99_ms"]
if p99 < 200:
experience = "流畅"
elif p99 < 500:
experience = "偶尔有轻微卡顿"
elif p99 < 1000:
experience = "明显卡顿"
else:
experience = "经常卡住"
lines = [
f"# {d['service']} 性能现状（给业务方）",
"",
"## 用户现在的体验",
"",
f"- **典型体验**：{cur['p50_ms']} 毫秒（{daily:,} 个请求中的大多数）",
f"- **最慢的 1% 用户**：要等 **{p99} 毫秒** —— {experience}",
f"- 按每天 {daily:,} 次请求计算，**每天约 {slow_users:,} 个用户**遇到这种情况",
"",
f"- **成功率**：{(1 - cur['error_rate']) * 100:.2f}%"
f"（每天约 **{failed_users:,} 次失败**）",
"",
"## 与我们的承诺相比",
"",
"| 项目 | 承诺 | 现状 | 是否达标 |",
"| --- | --- | --- | --- |",
f"| 典型响应时间 | {tgt['p50_ms']} ms | {cur['p50_ms']} ms | "
f"{'✅' if cur['p50_ms'] <= tgt['p50_ms'] else '❌'} |",
f"| 最慢 1% 的响应时间 | {tgt['p99_ms']} ms | {cur['p99_ms']} ms | "
f"{'✅' if cur['p99_ms'] <= tgt['p99_ms'] else '❌'} |",
f"| 成功率 | {(1 - tgt['error_rate']) * 100:.2f}% | "
f"{(1 - cur['error_rate']) * 100:.2f}% | "
f"{'✅' if cur['error_rate'] <= tgt['error_rate'] else '❌'} |",
"",
]
# 影响估算
if cur["error_rate"] > tgt["error_rate"]:
gap = failed_users - target_failed
revenue_loss = gap * d.get("avg_order_value", 0)
lines += [
"## 可能的影响",
"",
f"- 每天比承诺多 **{gap:,} 次失败**",
f"- 按客单价 {d.get('avg_order_value', 0)} 元估算，"
f"每月约影响 **{revenue_loss * 30:,.0f} 元**营收",
"",
]
# 趋势（如果提供了历史数据）
trend = d.get("trend")
if trend:
lines += [
"## 趋势",
"",
f"- {trend['description']}",
f"- {trend['impact']}",
"",
]
# 建议
lines += [
"## 建议",
"",
]
for i, s in enumerate(d.get("recommendations", []), 1):
lines.append(f"{i}. **{s['action']}**（{s['effort']}）")
lines.append(f" → 预期：{s['expected']}")
lines.append("")
# 不做的后果
lines += [
"## 如果不做",
"",
d.get("consequence", "问题可能继续恶化（需要评估具体影响）"),
"",
]
return "\n".join(lines)
if __name__ == "__main__":
if len(sys.argv) < 2:
demo = {
"service": "订单服务",
"daily_requests": 50000,
"current": {"p50_ms": 12, "p99_ms": 400, "error_rate": 0.008},
"target": {"p50_ms": 15, "p99_ms": 200, "error_rate": 0.005},
"avg_order_value": 200,
"trend": {
"description": "P99 在过去 3 个月从 210ms 上升到 400ms",
"impact": "按当前趋势，2 个月后最慢的 1% 用户将等待超过 1 秒",
},
"recommendations": [
{"action": "修复下单失败的主要原因", "effort": "1 名工程师 × 1 周",
"expected": "成功率回到 99.4%"},
{"action": "优化订单查询的索引与批量查询", "effort": "1 名工程师 × 2 周",
"expected": "P99 从 400ms 降到 120ms"},
],
"consequence": (
"P99 正在以每月约 15% 的速度恶化。如果不处理，两个月后会有约 5% 的用户"
"（每天 2500 人次）遇到明显卡顿，届时修复成本是现在的 3–5 倍。"
),
}
print(translate(demo))
sys.exit(0)
data = json.loads(open(sys.argv[1], encoding="utf-8").read())
print(translate(data))
预期输出：
# 订单服务 性能现状（给业务方）
## 用户现在的体验
- **典型体验**：12 毫秒（50,000 个请求中的大多数）
- **最慢的 1% 用户**：要等 **400 毫秒** —— 偶尔有轻微卡顿
- 按每天 50,000 次请求计算，**每天约 500 个用户**遇到这种情况
- **成功率**：99.20%（每天约 **400 次失败**）
## 与我们的承诺相比
| 项目 | 承诺 | 现状 | 是否达标 |
| --- | --- | --- | --- |
| 典型响应时间 | 15 ms | 12 ms | ✅ |
| 最慢 1% 的响应时间 | 200 ms | 400 ms | ❌ |
| 成功率 | 99.50% | 99.20% | ❌ |
## 可能的影响
- 每天比承诺多 **150 次失败**
- 按客单价 200 元估算，每月约影响 **900,000 元**营收
...
注意最后那个数字：「每月影响 90 万元营收」——即使这个估算很粗糙，它也比「P99 超标 100%」更能推动决策。
四、动手改造
| 改动 | 观察什么 |
|---|---|
用 weekly-perf-report.py 生成一份周报 |
看它能否自动标出需要关注的点 |
| 把周报发到团队频道（每周） | 观察"性能可见"带来的讨论变化 |
用 translate-for-business.py 翻译你项目的指标 |
看业务方能否理解 |
| 把 PR 模板里的 5 项加进你的仓库 | 观察有多少 PR 会认真回答 |
| 故意在 PR 里加一个无界缓存 | 模板会强制你回答"有容量上限吗" |
五、这段代码的局限
weekly-perf-report.py依赖 Prometheus 的指标名：需要按你的项目调整。- 周报的"趋势判断"很粗糙（只看 10% 阈值）：真正的趋势分析需要看多周数据（可以用
predict_linear或简单移动平均）。 translate-for-business.py的营收估算很粗略：它假设"失败 = 损失一单"，实际转化率损失更复杂——但它足以把技术问题变成业务问题。- PR 模板依赖人的诚实：可以被敷衍（全部打勾）——关键是 code review 时由 reviewer 追问（第 8.8 节）。
- 业务语言的翻译需要业务知识：脚本只能提供结构，具体的营收估算、用户影响需要和业务方确认。