《Python编程实战》17.3 灰度发布、回滚与故障演练

import hashlib
from dataclasses import dataclass

@dataclass
class Request:
    user_id: str | None = None
    headers: dict | None = None

def stable_bucket(key: str, buckets: int = 100) -> int:
    """把任意键稳定映射到 [0, buckets)：同一用户永远落同一桶"""
    h = hashlib.sha256(key.encode()).hexdigest()
    return int(h[:8], 16) % buckets

def route(req: Request, canary_pct: int, canary_header: str = "x-canary") -> str:
    if req.headers and req.headers.get(canary_header) == "1":
        return "canary"                                  # 1) 显式头优先
    if req.user_id is not None and stable_bucket(req.user_id) < canary_pct:
        return "canary"                                  # 2) 按用户稳定分流
    return "stable"

hits = sum(route(Request(user_id=f"u{i}"), 20) == "canary" for i in range(10000))
print(f"20% 灰度下命中: {hits}/10000 ({hits/10000:.1%})")
print("u42 连续 5 次:", [route(Request(user_id="u42"), 20) for _ in range(5)])
print("带头强制:", route(Request(user_id="u0", headers={"x-canary": "1"}), 20))

20% 灰度下命中: 1978/10000 (19.8%)
u42 连续 5 次: ['stable', 'stable', 'stable', 'stable', 'stable']
带头强制: canary

from dataclasses import dataclass

@dataclass
class Metrics:
    requests: int
    errors: int
    @property
    def error_rate(self) -> float:
        return self.errors / self.requests if self.requests else 0.0

def decide(canary: Metrics, stable: Metrics, *, max_abs: float = 0.01,
           max_rel: float = 1.5, min_requests: int = 100) -> str:
    """灰度 vs 稳定版错误率 → 推进 / 回滚 / 继续观察"""
    if canary.requests < min_requests:
        return "observe"                       # 样本不足，不下结论
    if canary.error_rate > max_abs and canary.error_rate > stable.error_rate * max_rel:
        return "rollback"
    return "promote"

cases = {
    "样本不足":     (Metrics(50, 5),    Metrics(1000, 8)),
    "错误率翻倍":   (Metrics(1000, 30), Metrics(1000, 8)),
    "绝对错误率低": (Metrics(1000, 3),  Metrics(1000, 8)),
    "与稳定版持平": (Metrics(1000, 9),  Metrics(1000, 8)),
}
for name, (c, s) in cases.items():
    print(f"{name:8s} canary={c.error_rate:.1%} stable={s.error_rate:.1%} -> {decide(c, s)}")

样本不足     canary=10.0% stable=0.8% -> observe
错误率翻倍   canary=3.0% stable=0.8% -> rollback
绝对错误率低 canary=0.3% stable=0.8% -> promote
与稳定版持平 canary=0.9% stable=0.8% -> promote

apiVersion: apps/v1
kind: Deployment
metadata:
  name: orders-api-canary
spec:
  replicas: 1
  selector:
    matchLabels: {app: orders-api, track: canary}
  template:
    metadata:
      labels: {app: orders-api, track: canary}
    spec:
      containers:
        - name: api
          image: registry.example.com/orders-api:1.4.0   # 不可变 tag
          ports: [{containerPort: 8000}]

# Alembic 迁移（expand 阶段）：只加可空列，不改动既有列
def upgrade() -> None:
    op.add_column("orders", sa.Column("channel", sa.String(32), nullable=True))

def downgrade() -> None:
    op.drop_column("orders", "channel")     # 仅在确认无代码读取时才可执行

【准备】
- [ ] 明确演练目标与假设（如「杀掉 1/3 实例，服务不中断」）
- [ ] 圈定影响范围（生产/预发、哪些用户、哪个时间窗）
- [ ] 设定停止条件（错误率 > X% 或业务指标跌破 Y 立即终止）
- [ ] 通知相关方（值班、客服、业务）
- [ ] 确认回滚手段可用（旧镜像在、数据库可前向兼容）
- [ ] 备好观测入口（指标、日志、追踪三个面板就位）

【执行】
- [ ] 从最小影响开始注入故障
- [ ] 全程盯观测面板，记录关键时间点与现象
- [ ] 触发停止条件立即终止并执行回滚
- [ ] 记录「预期行为 vs 实际行为」的差异

【收尾】
- [ ] 恢复环境，确认服务回到正常
- [ ] 24 小时内完成复盘
- [ ] 把发现的缺陷转成有 owner、有 deadline 的改进项

1. 发生了什么？
   - 时间线：从 T0 故障出现到 T4 完全恢复的关键时刻
   - 影响面：多少用户、多少请求、多长时间

2. 为什么发生？
   - 直接原因（触发点）
   - 根本原因（系统为何允许这个错误发生）
   - 为什么没被提前发现（监控/测试的盲区）

3. 我们怎么应对的？
   - 哪些做得好（值得保持）
   - 哪些拖慢了恢复（工具缺失？权限卡点？信息不通？）

4. 怎么防止复发？
   - 改进项清单（每项有 owner 与 deadline）
   - 需要新增的监控/告警/演练
   - 需要更新的 runbook