灰度发布与A/B测试

# canary-weight-based.yaml
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: webapp-canary
spec:
  hosts:
  - webapp.example.com
  gateways:
  - webapp-gateway
  http:
  - route:
    - destination:
        host: webapp-stable
        port:
          number: 8080
      weight: 90
    - destination:
        host: webapp-canary
        port:
          number: 8080
      weight: 10
    retries:
      attempts: 3
      perTryTimeout: 2s

# canary-header-based.yaml
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: webapp-feature-canary
spec:
  hosts:
  - webapp.example.com
  http:
  # 规则 1: 内部员工路由到灰度版本
  - match:
    - headers:
        x-user-type:
          exact: "internal"
    route:
    - destination:
        host: webapp-canary
        port:
          number: 8080

  # 规则 2: Beta 用户路由到灰度版本
  - match:
    - headers:
        x-user-tier:
          exact: "beta"
    route:
    - destination:
        host: webapp-canary
        port:
          number: 8080

  # 规则 3: 特定地域路由到灰度版本
  - match:
    - headers:
        x-region:
          regex: "^(us-west|ap-east)$"
    route:
    - destination:
        host: webapp-canary
        port:
          number: 8080

  # 规则 4: 基于用户 ID 取模的灰度
  - match:
    - headers:
        x-user-id:
          regex: "^[0-9]*[0-3]$"  # 用户 ID 尾数 0-3（约 10%）
    route:
    - destination:
        host: webapp-canary
        port:
          number: 8080

  # 默认路由到稳定版本
  - route:
    - destination:
        host: webapp-stable
        port:
          number: 8080

# destination-rule.yaml
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: webapp-dest
spec:
  host: webapp
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        h2UpgradePolicy: DEFAULT
        http1MaxPendingRequests: 100
        http2MaxRequests: 100
    outlierDetection:
      consecutive5xxErrors: 5
      interval: 30s
      baseEjectionTime: 30s
      maxEjectionPercent: 50
  subsets:
  - name: stable
    labels:
      version: v1
    trafficPolicy:
      connectionPool:
        http:
          http1MaxPendingRequests: 200
  - name: canary
    labels:
      version: v2
    trafficPolicy:
      connectionPool:
        http:
          http1MaxPendingRequests: 50  # 灰度版本限制并发

# rollout-canary-istio.yaml
apiVersion: argoproj.io/v1alpha1
kind: Rollout
metadata:
  name: webapp
spec:
  replicas: 10
  selector:
    matchLabels:
      app: webapp
  template:
    metadata:
      labels:
        app: webapp
    spec:
      containers:
      - name: webapp
        image: webapp:v2.0
        ports:
        - containerPort: 8080
        readinessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 10

  strategy:
    canary:
      steps:
      - setWeight: 5
      - pause: {duration: 5m}
      - setWeight: 10
      - pause: {duration: 5m}
      - analysis:
          templates:
          - templateName: canary-metrics-check
          args:
          - name: service-name
            value: webapp-canary
      - setWeight: 30
      - pause: {duration: 10m}
      - analysis:
          templates:
          - templateName: canary-metrics-check
      - setWeight: 50
      - pause: {duration: 10m}
      - setWeight: 100

      # Istio 流量管理
      trafficRouting:
        istio:
          virtualService:
            name: webapp-vsvc
            routes:
            - primary
          destinationRule:
            name: webapp-destrule
            canarySubsetName: canary
            stableSubsetName: stable

---
# AnalysisTemplate
apiVersion: argoproj.io/v1alpha1
kind: AnalysisTemplate
metadata:
  name: canary-metrics-check
spec:
  args:
  - name: service-name
  metrics:
  - name: error-rate
    interval: 60s
    count: 5
    successCondition: result[0] < 0.01
    failureLimit: 2
    provider:
      prometheus:
        address: http://prometheus-server:9090
        query: |
          sum(rate(http_requests_total{service="{{args.service-name}}",status=~"5.."}[2m]))
          /
          sum(rate(http_requests_total{service="{{args.service-name}}"}[2m]))

  - name: latency-p99
    interval: 60s
    count: 5
    successCondition: result[0] < 500
    failureLimit: 2
    provider:
      prometheus:
        address: http://prometheus-server:9090
        query: |
          histogram_quantile(0.99,
            sum(rate(http_request_duration_seconds_bucket{service="{{args.service-name}}"}[2m])) by (le)
          ) * 1000

// Feature Flag 灰度发布实现
package main

import (
    "hash/fnv"
    "log"
    "net/http"
)

type CanaryRule struct {
    FlagName     string
    Percentage   int     // 0-100
    UserSegments []string // 允许的用户段
    Regions      []string // 允许的地域
}

type CanaryManager struct {
    rules  map[string]CanaryRule
    client FeatureFlagClient
}

// 一致性 Hash 分流：同一用户始终分配到同一版本
func (cm *CanaryManager) getVariant(userID string, rule CanaryRule) string {
    // 检查用户段
    if len(rule.UserSegments) > 0 {
        userSegment := getUserSegment(userID)
        for _, seg := range rule.UserSegments {
            if seg == userSegment {
                return "canary"
            }
        }
    }

    // 检查地域
    if len(rule.Regions) > 0 {
        userRegion := getUserRegion(userID)
        for _, reg := range rule.Regions {
            if reg == userRegion {
                return "canary"
            }
        }
    }

    // 按比例分流（一致性 Hash）
    h := fnv.New32a()
    h.Write([]byte(userID + ":" + rule.FlagName))
    hashValue := h.Sum32() % 100

    if hashValue < uint32(rule.Percentage) {
        return "canary"
    }
    return "stable"
}

// 灰度发布处理器
func (cm *CanaryManager) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    userID := r.Header.Get("X-User-ID")
    flagName := r.URL.Query().Get("feature")

    rule, exists := cm.rules[flagName]
    if !exists {
        // Flag 不存在，使用稳定版本
        serveStable(w, r)
        return
    }

    variant := cm.getVariant(userID, rule)

    // 记录灰度事件
    log.Printf("canary: user=%s feature=%s variant=%s", userID, flagName, variant)

    switch variant {
    case "canary":
        serveCanary(w, r)
    default:
        serveStable(w, r)
    }
}

#!/usr/bin/env python3
"""
A/B 测试数据分析工具
支持: 统计显著性检验、样本量计算、置信区间估计
"""

import math
from dataclasses import dataclass
from typing import Tuple, Optional

from scipy import stats
import numpy as np


@dataclass
class ABTestResult:
    """A/B 测试结果"""
    control_conversions: int
    control_total: int
    treatment_conversions: int
    treatment_total: int
    control_rate: float
    treatment_rate: float
    lift: float
    p_value: float
    confidence_interval: Tuple[float, float]
    is_significant: bool
    power: float
    min_sample_size: int


def calculate_min_sample_size(
    baseline_rate: float,
    mde: float,
    alpha: float = 0.05,
    power: float = 0.8,
) -> int:
    """
    计算最小样本量

    Args:
        baseline_rate: 基线转化率
        mde: 最小可检测效应（相对变化量）
        alpha: 显著性水平
        power: 统计功效

    Returns:
        每组需要的最小样本量
    """
    p1 = baseline_rate
    p2 = baseline_rate * (1 + mde)

    # Z 值
    z_alpha = stats.norm.ppf(1 - alpha / 2)
    z_beta = stats.norm.ppf(power)

    # 样本量公式（双比例检验）
    p_avg = (p1 + p2) / 2
    n = ((z_alpha * math.sqrt(2 * p_avg * (1 - p_avg)) +
          z_beta * math.sqrt(p1 * (1 - p1) + p2 * (1 - p2))) ** 2) / \
        (p2 - p1) ** 2

    return math.ceil(n)


def analyze_ab_test(
    control_conversions: int,
    control_total: int,
    treatment_conversions: int,
    treatment_total: int,
    alpha: float = 0.05,
) -> ABTestResult:
    """
    分析 A/B 测试结果

    Args:
        control_conversions: 对照组转化数
        control_total: 对照组总人数
        treatment_conversions: 实验组转化数
        treatment_total: 实验组总人数
        alpha: 显著性水平

    Returns:
        ABTestResult 包含完整分析结果
    """
    # 转化率
    control_rate = control_conversions / control_total
    treatment_rate = treatment_conversions / treatment_total

    # 提升幅度
    lift = (treatment_rate - control_rate) / control_rate

    # Z 检验（双比例检验）
    p_pool = (control_conversions + treatment_conversions) / \
             (control_total + treatment_total)
    se = math.sqrt(p_pool * (1 - p_pool) * (1/control_total + 1/treatment_total))
    z_score = (treatment_rate - control_rate) / se
    p_value = 2 * (1 - stats.norm.cdf(abs(z_score)))

    # 置信区间（差异的置信区间）
    se_diff = math.sqrt(
        control_rate * (1 - control_rate) / control_total +
        treatment_rate * (1 - treatment_rate) / treatment_total
    )
    diff = treatment_rate - control_rate
    z_crit = stats.norm.ppf(1 - alpha / 2)
    ci_lower = diff - z_crit * se_diff
    ci_upper = diff + z_crit * se_diff

    # 统计功效（事后）
    p1, p2 = control_rate, treatment_rate
    z_alpha = stats.norm.ppf(1 - alpha / 2)
    h = 2 * math.asin(math.sqrt(p2)) - 2 * math.asin(math.sqrt(p1))
    n_per_group = min(control_total, treatment_total)
    power = stats.norm.cdf(math.sqrt(n_per_group) * abs(h) - z_alpha)

    # 最小样本量
    min_sample = calculate_min_sample_size(control_rate, abs(lift), alpha, 0.8)

    return ABTestResult(
        control_conversions=control_conversions,
        control_total=control_total,
        treatment_conversions=treatment_conversions,
        treatment_total=treatment_total,
        control_rate=control_rate,
        treatment_rate=treatment_rate,
        lift=lift,
        p_value=p_value,
        confidence_interval=(ci_lower, ci_upper),
        is_significant=p_value < alpha,
        power=power,
        min_sample_size=min_sample,
    )


def sequential_testing_check(
    daily_results: list,
    alpha: float = 0.05,
    num_peeks: int = 10,
) -> dict:
    """
    序贯检验（Sequential Testing）
    解决多次查看数据导致的 p-value 膨胀问题

    使用 Pocock 边界法调整显著性水平

    Args:
        daily_results: 每日数据列表 [(control_conv, control_total, treat_conv, treat_total)]
        alpha: 总体显著性水平
        num_peeks: 预期查看次数

    Returns:
        检验结果
    """
    # Pocock 边界调整
    alpha_spending = alpha / num_peeks * 2  # 简化的 Pocock 调整

    cumulative_control_conv = 0
    cumulative_control_total = 0
    cumulative_treat_conv = 0
    cumulative_treat_total = 0

    for day_idx, (cc, ct, tc, tt) in enumerate(daily_results):
        cumulative_control_conv += cc
        cumulative_control_total += ct
        cumulative_treat_conv += tc
        cumulative_treat_total += tt

        # 最小样本量检查
        if cumulative_control_total < 100 or cumulative_treat_total < 100:
            continue

        result = analyze_ab_test(
            cumulative_control_conv,
            cumulative_control_total,
            cumulative_treat_conv,
            cumulative_treat_total,
            alpha=alpha_spending,
        )

        if result.is_significant:
            return {
                "stop": True,
                "day": day_idx + 1,
                "reason": f"统计显著 (adjusted p={result.p_value:.4f} < {alpha_spending:.4f})",
                "result": result,
            }

    return {
        "stop": False,
        "day": len(daily_results),
        "reason": "未达到统计显著性，继续实验",
        "result": None,
    }


# 使用示例
if __name__ == "__main__":
    # 示例 1: 计算 A/B 测试所需样本量
    baseline = 0.05  # 基线转化率 5%
    mde = 0.20       # 希望检测 20% 的相对提升
    sample_size = calculate_min_sample_size(baseline, mde)
    print(f"基线转化率: {baseline*100}%")
    print(f"最小可检测效应: {mde*100}% 相对提升")
    print(f"每组最小样本量: {sample_size}")
    print()

    # 示例 2: 分析 A/B 测试结果
    result = analyze_ab_test(
        control_conversions=250,
        control_total=5000,
        treatment_conversions=310,
        treatment_total=5000,
    )
    print("A/B 测试分析结果:")
    print(f"  对照组转化率: {result.control_rate*100:.2f}%")
    print(f"  实验组转化率: {result.treatment_rate*100:.2f}%")
    print(f"  提升幅度: {result.lift*100:.2f}%")
    print(f"  P-value: {result.p_value:.4f}")
    print(f"  置信区间: ({result.confidence_interval[0]*100:.2f}%, {result.confidence_interval[1]*100:.2f}%)")
    print(f"  统计显著: {result.is_significant}")
    print(f"  统计功效: {result.power*100:.1f}%")
    print(f"  最小样本量: {result.min_sample_size}")
    print()

    # 示例 3: 序贯检验
    daily_data = [
        (50, 1000, 55, 1000),   # Day 1
        (100, 2000, 120, 2000),  # Day 2
        (150, 3000, 180, 3000),  # Day 3
        (200, 4000, 245, 4000),  # Day 4
        (250, 5000, 310, 5000),  # Day 5
    ]
    seq_result = sequential_testing_check(daily_data)
    print(f"序贯检验结果: stop={seq_result['stop']}, day={seq_result['day']}")
    if seq_result['result']:
        print(f"  原因: {seq_result['reason']}")

// 灰度用户选择策略
package canary

import (
    "hash/fnv"
    "sort"
)

// 策略 1: 一致性 Hash（推荐）
// 同一用户始终路由到同一版本，避免体验不一致
func ConsistentHash(userID string, flagName string, percentage int) bool {
    h := fnv.New32a()
    h.Write([]byte(userID + ":" + flagName))
    return h.Sum32()%100 < uint32(percentage)
}

// 策略 2: 分层实验
// 不同实验使用不同的 Hash 键，避免实验间干扰
func LayeredHash(userID string, experimentID string, flagName string, percentage int) bool {
    h := fnv.New32a()
    h.Write([]byte(userID + ":" + experimentID + ":" + flagName))
    return h.Sum32()%100 < uint32(percentage)
}

// 策略 3: 渐进式扩大
// 保持已灰度用户不变，新增用户进入灰度
type ProgressiveCanary struct {
    currentPercentage int
    includedUsers     map[string]bool
    excludedUsers     map[string]bool
}

func (pc *ProgressiveCanary) Expand(newPercentage int, allUserIDs []string) {
    // 计算需要新增的灰度用户数
    totalUsers := len(allUserIDs)
    targetCount := totalUsers * newPercentage / 100
    currentCount := len(pc.includedUsers)

    if targetCount <= currentCount {
        return // 不缩小灰度范围
    }

    // 找出未分配的用户
    var unassigned []string
    for _, uid := range allUserIDs {
        if !pc.includedUsers[uid] && !pc.excludedUsers[uid] {
            unassigned = append(unassigned, uid)
        }
    }

    // 随机选择新增用户
    sort.Slice(unassigned, func(i, j int) bool {
        hi := fnv.New32a()
        hi.Write([]byte(unassigned[i]))
        hj := fnv.New32a()
        hj.Write([]byte(unassigned[j]))
        return hi.Sum32() < hj.Sum32()
    })

    addCount := targetCount - currentCount
    for i := 0; i < addCount && i < len(unassigned); i++ {
        pc.includedUsers[unassigned[i]] = true
    }

    // 剩余用户标记为排除
    for _, uid := range unassigned[addCount:] {
        pc.excludedUsers[uid] = true
    }

    pc.currentPercentage = newPercentage
}

# 灰度监控仪表板
apiVersion: v1
kind: ConfigMap
metadata:
  name: canary-dashboard
data:
  dashboard.json: |
    {
      "dashboard": {
        "title": "Canary Release Metrics",
        "panels": [
          {
            "title": "Error Rate Comparison",
            "type": "graph",
            "targets": [
              {
                "expr": "sum(rate(http_requests_total{version='stable',status=~'5..'}[5m])) / sum(rate(http_requests_total{version='stable'}[5m]))",
                "legendFormat": "Stable Error Rate"
              },
              {
                "expr": "sum(rate(http_requests_total{version='canary',status=~'5..'}[5m])) / sum(rate(http_requests_total{version='canary'}[5m]))",
                "legendFormat": "Canary Error Rate"
              }
            ]
          },
          {
            "title": "P99 Latency Comparison",
            "type": "graph",
            "targets": [
              {
                "expr": "histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket{version='stable'}[5m])) by (le)) * 1000",
                "legendFormat": "Stable P99"
              },
              {
                "expr": "histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket{version='canary'}[5m])) by (le)) * 1000",
                "legendFormat": "Canary P99"
              }
            ]
          },
          {
            "title": "Traffic Split",
            "type": "pie",
            "targets": [
              {
                "expr": "sum(rate(http_requests_total{version='stable'}[5m]))",
                "legendFormat": "Stable"
              },
              {
                "expr": "sum(rate(http_requests_total{version='canary'}[5m]))",
                "legendFormat": "Canary"
              }
            ]
          }
        ]
      }
    }

手工灰度 → 自动化灰度 → A/B 测试 → 实验平台
   ↓          ↓           ↓          ↓
 降低风险   提升效率    数据驱动    规模化实验