8.6 流量回放与金丝雀发布

# 从 PostgreSQL 导出真实的 ID 访问分布
psql -tAc "
SELECT user_id FROM request_log
WHERE created_at > now() - interval '1 day'
ORDER BY random() LIMIT 100000;" > sampled-ids.txt

# 在 k6 里按这个列表取样（保持真实分布）
# const ids = open('./sampled-ids.txt').split('\n');
# const id = ids[Math.floor(Math.random() * ids.length)];

对于大多数场景：① 只回放读 + 对写路径用【合成流量】单独压测
对于核心写路径：② 影子库
对于安全的写路径（如"更新最后访问时间"）：③ 空转

// 用一个明确的开关控制所有副作用
object SideEffectGuard {
    @Volatile var enabled = System.getenv("ENABLE_SIDE_EFFECTS")?.toBoolean() ?: true

    fun <T> guarded(name: String, block: () -> T): T? {
        if (!enabled) {
            log.warn("副作用已被阻断（影子环境）：{}", name)
            metrics.counter("side_effect.blocked", "name", name).increment()
            return null
        }
        return block()
    }
}

// 使用
SideEffectGuard.guarded("send-sms") { smsClient.send(phone, msg) }

# tools/sanitize_logs.py
"""
脱敏生产日志，准备用于流量回放。
"""
import hashlib
import json
import re
import sys


class Sanitizer:
    def __init__(self, salt="replay-salt"):
        self.salt = salt
        self.id_map = {}

    def stable_id(self, original: str) -> str:
        """保持关联的 ID 映射：同一输入 → 同一输出"""
        if original not in self.id_map:
            h = hashlib.sha256((self.salt + original).encode()).hexdigest()[:12]
            self.id_map[original] = f"replay-{h}"
        return self.id_map[original]

    def phone(self, value: str) -> str:
        """保持格式：138****1234 → 13900001234（格式相同、内容假）"""
        if len(value) == 11 and value.isdigit():
            return "1" + "39" + "0" * 5 + value[-3:]
        return value

    def sanitize_line(self, line: str) -> str:
        obj = json.loads(line)
        for key in list(obj.keys()):
            if key in ("user_id", "order_id", "session_id"):
                obj[key] = self.stable_id(str(obj[key]))
            elif key in ("phone", "mobile"):
                obj[key] = self.phone(str(obj[key]))
            elif key in ("token", "authorization", "api_key"):
                obj[key] = "replay-token-placeholder"
            elif key in ("email",):
                obj[key] = "user@example.com"
        return json.dumps(obj, ensure_ascii=False)


def main(in_path, out_path):
    s = Sanitizer()
    with open(in_path) as fin, open(out_path, "w") as fout:
        for line in fin:
            if line.strip():
                fout.write(s.sanitize_line(line) + "\n")
    print(f"✅ 脱敏完成：{in_path} → {out_path}")
    print(f"   映射了 {len(s.id_map)} 个 ID（保持关联关系）")


if __name__ == "__main__":
    main(sys.argv[1], sys.argv[2] if len(sys.argv) > 2 else "sanitized.jsonl")

              ┌─→ 生产版本 → 返回给用户
用户请求 ──→ 网关
              └─→ 影子版本 → 只记录，不返回

# Nginx 流量镜像
location /api/ {
    mirror /shadow;                    # 复制一份到 /shadow
    mirror_request_body on;
    proxy_pass http://production;
}

location = /shadow {
    internal;
    proxy_pass http://canary-version$request_uri;
    # 影子版本的响应被丢弃
}

① 部署金丝雀实例（1 个，占总量的 5%）
② 把 5% 流量切给金丝雀
③ 持续对比金丝雀组与基线组的关键指标
④ 达标 → 逐步扩大比例（5% → 25% → 50% → 100%）
⑤ 不达标 → 自动回滚

#!/usr/bin/env bash
# tools/canary-monitor.sh <BASELINE_LABEL> <CANARY_LABEL> [DURATION_MIN]
#
# 持续对比金丝雀与基线，超阈值则触发回滚。
set -uo pipefail

BASELINE="${1:?usage: canary-monitor.sh <baseline> <canary> [duration]}"
CANARY="${2:?}"
DURATION="${3:-10}"
PROM="${PROM_URL:-http://localhost:9090}"

# 回滚阈值（在发布前定义）
P99_DELTA_THRESHOLD=0.20       # P99 恶化 > 20% 回滚
ERROR_DELTA_THRESHOLD=0.005    # 错误率上升 > 0.5pp 回滚
PENDING_THRESHOLD=5            # pending > 5 持续 3 分钟回滚

echo "═══ 金丝雀监控 ═══"
echo "  基线组: $BASELINE"
echo "  金丝雀: $CANARY"
echo "  时长  : ${DURATION} 分钟"
echo

query() {
  curl -s --get "$PROM/api/v1/query" --data-urlencode "query=$1" \
    | python3 -c "import json,sys; d=json.load(sys.stdin); r=d['data']['result']; print(r[0]['value'][1] if r else 'NaN')" 2>/dev/null || echo "NaN"
}

p99() {
  query "histogram_quantile(0.99, sum by (le) (rate(http_request_duration_seconds_bucket{version=\"$1\"}[5m])))"
}

error_rate() {
  query "sum(rate(http_requests_total{version=\"$1\",status=~\"5..\"}[5m])) / sum(rate(http_requests_total{version=\"$1\"}[5m]))"
}

pending() {
  query "max(db_pool_pending{version=\"$1\"})"
}

ROLLBACK=0
PENDING_BREACH_COUNT=0

# 用一个 Python 脚本做判定，避免在 bash 里拼复杂的浮点比较
cat > /tmp/canary_verdict.py <<'PYEOF'
import sys

b_p99, c_p99, b_err, c_err, c_pend = sys.argv[1:6]
p99_threshold = float(sys.argv[6])
err_threshold = float(sys.argv[7])
pend_threshold = float(sys.argv[8])

def num(x):
    try:
        return float(x)
    except (ValueError, TypeError):
        return 0.0

b_p99, c_p99 = num(b_p99), num(c_p99)
b_err, c_err, c_pend = num(b_err), num(c_err), num(c_pend)

delta = (c_p99 - b_p99) / b_p99 if b_p99 else 0.0
err_delta = c_err - b_err

verdict = "OK"
reason = ""
if delta > p99_threshold:
    verdict, reason = "ROLLBACK", f"P99 恶化 {delta*100:.1f}% > {p99_threshold*100:.0f}%"
elif err_delta > err_threshold:
    verdict, reason = "ROLLBACK", f"错误率上升 {err_delta*100:.2f}pp"
elif c_pend > pend_threshold:
    verdict, reason = "PENDING_BREACH", f"pending={c_pend:.0f} > {pend_threshold:.0f}"

print(f"{delta*100:.1f}|{verdict}|{reason}")
PYEOF

for i in $(seq 1 "$DURATION"); do
  B_P99=$(p99 "$BASELINE"); C_P99=$(p99 "$CANARY")
  B_ERR=$(error_rate "$BASELINE"); C_ERR=$(error_rate "$CANARY")
  C_PEND=$(pending "$CANARY")

  VERDICT_LINE=$(python3 /tmp/canary_verdict.py \
    "$B_P99" "$C_P99" "$B_ERR" "$C_ERR" "$C_PEND" \
    "$P99_DELTA_THRESHOLD" "$ERROR_DELTA_THRESHOLD" "$PENDING_THRESHOLD")

  DELTA="${VERDICT_LINE%%|*}"
  REST="${VERDICT_LINE#*|}"
  VERDICT="${REST%%|*}"
  REASON="${REST#*|}"

  echo "[$i/$DURATION] 基线 P99=${B_P99}ms 金丝雀 P99=${C_P99}ms（${DELTA}%）  pending=${C_PEND}"

  case "$VERDICT" in
    ROLLBACK)
      echo "  ❌ $REASON"
      ROLLBACK=1
      break
      ;;
    PENDING_BREACH)
      PENDING_BREACH_COUNT=$((PENDING_BREACH_COUNT + 1))
      echo "  ⚠️  $REASON（第 ${PENDING_BREACH_COUNT} 次，连续 3 次则回滚）"
      if [ "$PENDING_BREACH_COUNT" -ge 3 ]; then
        echo "  ❌ 连接池 pending 持续超标"
        ROLLBACK=1
        break
      fi
      ;;
    *)
      PENDING_BREACH_COUNT=0
      ;;
  esac

  sleep 60
done

echo
if [ "$ROLLBACK" -eq 1 ]; then
  echo "⛔ 触发回滚"
  scripts/rollback-canary.sh
  exit 1
else
  echo "✅ 金丝雀监控通过（${DURATION} 分钟）"
  echo "   可以继续扩大流量比例"
fi

❌ 金丝雀只接 5% 流量，观察 5 分钟
   → 样本太少（假设总 QPS 1000，5 分钟只有 15000 个请求）
   → P99 的估计不稳定，容易误判

✅ 两个选择：
   ① 接更多流量（比如 20%），或
   ② 观察更长时间（比如 30 分钟），或
   ③ 只对「错误率」和「饱和度」这类稳定指标做自动判定，
      P99 做人工观察