API弹性设计与混沌工程：构建高可用微服务系统

package circuitbreaker

import (
    "errors"
    "sync"
    "time"
)

type State int

const (
    StateClosed State = iota
    StateOpen
    StateHalfOpen
)

type CircuitBreaker struct {
    mu                sync.RWMutex
    state             State
    failureCount      int
    successCount      int
    lastFailureTime   time.Time
    
    // 配置
    failureThreshold  int           // 触发断路器的失败次数
    successThreshold  int           // 半开状态需要的成功次数
    timeout           time.Duration // 从Open到HalfOpen的等待时间
}

func NewCircuitBreaker(failureThreshold, successThreshold int, timeout time.Duration) *CircuitBreaker {
    return &CircuitBreaker{
        state:            StateClosed,
        failureThreshold: failureThreshold,
        successThreshold: successThreshold,
        timeout:          timeout,
    }
}

func (cb *CircuitBreaker) Execute(operation func() error) error {
    if !cb.canExecute() {
        return errors.New("circuit breaker is open")
    }
    
    err := operation()
    cb.record(err)
    return err
}

func (cb *CircuitBreaker) canExecute() bool {
    cb.mu.RLock()
    defer cb.mu.RUnlock()
    
    switch cb.state {
    case StateClosed:
        return true
    case StateOpen:
        // 检查是否超时
        if time.Since(cb.lastFailureTime) > cb.timeout {
            cb.mu.RUnlock()
            cb.mu.Lock()
            cb.state = StateHalfOpen
            cb.mu.Unlock()
            cb.mu.RLock()
            return true
        }
        return false
    case StateHalfOpen:
        return true
    }
    return false
}

func (cb *CircuitBreaker) record(err error) {
    cb.mu.Lock()
    defer cb.mu.Unlock()
    
    if err != nil {
        cb.failureCount++
        cb.lastFailureTime = time.Now()
        
        if cb.failureCount >= cb.failureThreshold {
            cb.state = StateOpen
        }
    } else {
        if cb.state == StateHalfOpen {
            cb.successCount++
            if cb.successCount >= cb.successThreshold {
                // 恢复
                cb.state = StateClosed
                cb.failureCount = 0
                cb.successCount = 0
            }
        }
    }
}

func (cb *CircuitBreaker) GetState() State {
    cb.mu.RLock()
    defer cb.mu.RUnlock()
    return cb.state
}

package bulkhead

import (
    "context"
    "sync"
)

type Bulkhead struct {
    maxConcurrent int
    maxQueue      int
    semaphore     chan struct{}
    queue         chan func() error
    wg            sync.WaitGroup
}

func NewBulkhead(maxConcurrent, maxQueue int) *Bulkhead {
    bh := &Bulkhead{
        maxConcurrent: maxConcurrent,
        maxQueue:      maxQueue,
        semaphore:     make(chan struct{}, maxConcurrent),
        queue:         make(chan func() error, maxQueue),
    }
    
    // 启动工作协程
    for i := 0; i < maxConcurrent; i++ {
        bh.wg.Add(1)
        go bh.worker()
    }
    
    return bh
}

func (bh *Bulkhead) worker() {
    defer bh.wg.Done()
    for operation := range bh.queue {
        bh.semaphore <- struct{}{}
        operation()
        <-bh.semaphore
    }
}

func (bh *Bulkhead) Execute(ctx context.Context, operation func() error) error {
    select {
    case bh.queue <- operation:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    default:
        return errors.New("bulkhead queue is full")
    }
}

func (bh *Bulkhead) Shutdown() {
    close(bh.queue)
    bh.wg.Wait()
}

package retry

import (
    "context"
    "math"
    "math/rand"
    "time"
)

type RetryConfig struct {
    MaxAttempts     int
    InitialInterval time.Duration
    MaxInterval     time.Duration
    Multiplier      float64
    RandomizationFactor float64
}

func WithRetry(ctx context.Context, config RetryConfig, operation func() error) error {
    var lastErr error
    interval := config.InitialInterval
    
    for attempt := 0; attempt < config.MaxAttempts; attempt++ {
        // 检查上下文
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
        }
        
        err := operation()
        if err == nil {
            return nil
        }
        
        lastErr = err
        
        // 最后一次尝试不需要等待
        if attempt == config.MaxAttempts-1 {
            break
        }
        
        // 计算带抖动的等待时间
        delta := config.RandomizationFactor * float64(interval)
        min := float64(interval) - delta
        max := float64(interval) + delta
        jitter := time.Duration(min + (rand.Float64() * (max - min + 1)))
        
        select {
        case <-time.After(jitter):
        case <-ctx.Done():
            return ctx.Err()
        }
        
        // 指数退避
        interval = time.Duration(float64(interval) * config.Multiplier)
        if interval > config.MaxInterval {
            interval = config.MaxInterval
        }
    }
    
    return lastErr
}

// 使用示例
func CallExternalAPI(ctx context.Context) error {
    config := RetryConfig{
        MaxAttempts:         5,
        InitialInterval:     100 * time.Millisecond,
        MaxInterval:         10 * time.Second,
        Multiplier:          2.0,
        RandomizationFactor: 0.5,
    }
    
    return WithRetry(ctx, config, func() error {
        // 执行API调用
        return nil
    })
}

package fallback

import (
    "context"
)

type FallbackStrategy interface {
    Execute(ctx context.Context, primaryError error) (interface{}, error)
}

// 缓存降级：返回缓存数据
type CacheFallback struct {
    cache Cache
    key   string
}

func (f *CacheFallback) Execute(ctx context.Context, primaryError error) (interface{}, error) {
    return f.cache.Get(ctx, f.key)
}

// 默认值降级：返回默认值
type DefaultValueFallback struct {
    value interface{}
}

func (f *DefaultValueFallback) Execute(ctx context.Context, primaryError error) (interface{}, error) {
    return f.value, nil
}

// 功能降级：提供有限功能
type PartialFunctionalityFallback struct {
    limitedFunc func() (interface{}, error)
}

func (f *PartialFunctionalityFallback) Execute(ctx context.Context, primaryError error) (interface{}, error) {
    return f.limitedFunc()
}

// 弹性执行器
type ResilientExecutor struct {
    primary  func(ctx context.Context) (interface{}, error)
    fallbacks []FallbackStrategy
}

func NewResilientExecutor(primary func(ctx context.Context) (interface{}, error), fallbacks ...FallbackStrategy) *ResilientExecutor {
    return &ResilientExecutor{
        primary:   primary,
        fallbacks: fallbacks,
    }
}

func (e *ResilientExecutor) Execute(ctx context.Context) (interface{}, error) {
    result, err := e.primary(ctx)
    if err == nil {
        return result, nil
    }
    
    // 尝试降级策略
    for _, fallback := range e.fallbacks {
        result, fallbackErr := fallback.Execute(ctx, err)
        if fallbackErr == nil {
            return result, nil
        }
    }
    
    // 所有降级策略都失败，返回原始错误
    return nil, err
}

# 安装Chaos Mesh
helm repo add chaos-mesh https://charts.chaos-mesh.org
helm install chaos-mesh chaos-mesh/chaos-mesh \
  --namespace=chaos-testing \
  --create-namespace \
  --set dashboard.create=true

# 访问Dashboard
kubectl port-forward -n chaos-testing svc/chaos-dashboard 2333:2333

# network-delay.yaml
apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: network-delay
  namespace: default
spec:
  action: delay
  mode: all
  selector:
    namespaces:
      - default
    labelSelectors:
      app: payment-service
  delay:
    latency: "200ms"
    correlation: "100"
    jitter: "50ms"
  direction: both
  duration: "5m"
  scheduler:
    cron: "@every 1h"

# network-partition.yaml
apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: network-partition
  namespace: default
spec:
  action: partition
  mode: all
  selector:
    namespaces:
      - default
    labelSelectors:
      app: order-service
  direction: both
  target:
    selector:
      namespaces:
        - default
      labelSelectors:
        app: inventory-service
  duration: "3m"

# pod-kill.yaml
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
  name: pod-kill
  namespace: default
spec:
  action: pod-kill
  mode: one
  selector:
    namespaces:
      - default
    labelSelectors:
      app: user-service
  gracePeriod: 0
  duration: "1m"
  scheduler:
    cron: "@every 30m"

# pod-failure.yaml
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
  name: pod-failure
  namespace: default
spec:
  action: pod-failure
  mode: fixed-percent
  value: "30"
  selector:
    namespaces:
      - default
    labelSelectors:
      app: recommendation-service
  duration: "5m"

# stress-cpu.yaml
apiVersion: chaos-mesh.org/v1alpha1
kind: StressChaos
metadata:
  name: stress-cpu
  namespace: default
spec:
  mode: all
  selector:
    namespaces:
      - default
    labelSelectors:
      app: api-gateway
  stressors:
    cpu:
      workers: 2
      load: 80
  duration: "10m"

# stress-memory.yaml
apiVersion: chaos-mesh.org/v1alpha1
kind: StressChaos
metadata:
  name: stress-memory
  namespace: default
spec:
  mode: one
  selector:
    namespaces:
      - default
    labelSelectors:
      app: cache-service
  stressors:
    memory:
      workers: 1
      size: "512MB"
  duration: "5m"

package resilience_test

import (
    "context"
    "testing"
    "time"
)

type ResilienceTestSuite struct {
    chaosClient ChaosClient
    monitor     MetricsClient
}

func (suite *ResilienceTestSuite) TestCircuitBreaker(t *testing.T) {
    ctx := context.Background()
    
    // 注入故障：让payment-service 50%请求失败
    suite.chaosClient.InjectFault(ctx, NetworkChaos{
        Target: "payment-service",
        Type:   "http-error",
        Rate:   0.5,
    })
    defer suite.chaosClient.RemoveFault(ctx)
    
    // 等待断路器打开
    time.Sleep(30 * time.Second)
    
    // 验证断路器状态
    state := suite.monitor.GetCircuitBreakerState("payment-service")
    assert.Equal(t, StateOpen, state)
    
    // 验证降级策略生效
    response := suite.callAPI("/api/orders")
    assert.Equal(t, 200, response.StatusCode)
    assert.Contains(t, response.Body, "payment_pending")
}

func (suite *ResilienceTestSuite) TestBulkheadIsolation(t *testing.T) {
    ctx := context.Background()
    
    // 注入故障：让recommendation-service响应变慢
    suite.chaosClient.InjectFault(ctx, NetworkChaos{
        Target:  "recommendation-service",
        Type:    "delay",
        Latency: "5s",
    })
    defer suite.chaosClient.RemoveFault(ctx)
    
    // 发送大量请求
    start := time.Now()
    var wg sync.WaitGroup
    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            suite.callAPI("/api/products")
        }()
    }
    wg.Wait()
    duration := time.Since(start)
    
    // 验证其他服务不受影响（应该在2秒内完成）
    assert.Less(t, duration, 2*time.Second)
}

func (suite *ResilienceTestSuite) TestTimeoutAndRetry(t *testing.T) {
    ctx := context.Background()
    
    // 注入故障：让external-api间歇性失败
    suite.chaosClient.InjectFault(ctx, NetworkChaos{
        Target: "external-api",
        Type:   "http-error",
        Rate:   0.3,
    })
    defer suite.chaosClient.RemoveFault(ctx)
    
    // 验证重试机制
    successCount := 0
    for i := 0; i < 100; i++ {
        response := suite.callAPI("/api/sync-external")
        if response.StatusCode == 200 {
            successCount++
        }
    }
    
    // 重试应该让成功率接近100%
    assert.Greater(t, successCount, 95)
}

#!/bin/bash
# resilience-test.sh

echo "=== 弹性验证测试 ==="

# 测试1: 服务重启恢复
echo "测试1: Pod重启恢复"
kubectl delete pod -l app=order-service
kubectl rollout status deployment/order-service --timeout=60s
if [ $? -eq 0 ]; then
    echo "✓ Pod重启成功"
else
    echo "✗ Pod重启失败"
    exit 1
fi

# 测试2: 数据库连接池恢复
echo "测试2: 数据库连接恢复"
kubectl exec -it $(kubectl get pod -l app=db-proxy -o name) -- pkill -HUP pgpool
sleep 10
curl -f http://api-gateway/health
if [ $? -eq 0 ]; then
    echo "✓ 数据库连接恢复成功"
else
    echo "✗ 数据库连接恢复失败"
    exit 1
fi

# 测试3: 缓存失效恢复
echo "测试3: 缓存失效恢复"
kubectl exec -it $(kubectl get pod -l app=redis -o name) -- redis-cli FLUSHALL
sleep 5
curl -f http://api-gateway/api/products
if [ $? -eq 0 ]; then
    echo "✓ 缓存失效后服务正常"
else
    echo "✗ 缓存失效后服务异常"
    exit 1
fi

echo "=== 所有测试通过 ==="

package queue

import (
    "context"
    "errors"
    "sync"
    "time"
)

// LoadSheddingQueue 基于权重的请求丢弃队列
type LoadSheddingQueue struct {
    capacity     int
    queue        chan Request
    semaphore    chan struct{}
    dropping     bool
    dropRate     float64
    queueTimeMax time.Duration
    mu           sync.RWMutex
}

func NewLoadSheddingQueue(capacity, maxConcurrency int, queueTimeMax time.Duration) *LoadSheddingQueue {
    return &LoadSheddingQueue{
        capacity:     capacity,
        queue:        make(chan Request, capacity),
        semaphore:    make(chan struct{}, maxConcurrency),
        queueTimeMax: queueTimeMax,
    }
}

func (q *LoadSheddingQueue) Submit(ctx context.Context, req Request) error {
    // 1. 检查是否处于丢弃模式
    if q.shouldDrop() {
        return errors.New("load shedding: request dropped")
    }
    
    // 2. 检查队列等待时间
    select {
    case q.queue <- req:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    default:
        // 队列已满，开启丢弃模式
        q.enableLoadShedding()
        return errors.New("queue full: request rejected")
    }
}

func (q *LoadSheddingQueue) shouldDrop() bool {
    q.mu.RLock()
    defer q.mu.RUnlock()
    if !q.dropping {
        return false
    }
    // 按丢弃率随机丢弃
    return rand.Float64() < q.dropRate
}

func (q *LoadSheddingQueue) enableLoadShedding() {
    q.mu.Lock()
    defer q.mu.Unlock()
    q.dropping = true
    q.dropRate = 0.1  // 从丢弃 10% 开始
    
    // 自适应调整丢弃率
    go func() {
        ticker := time.NewTicker(5 * time.Second)
        defer ticker.Stop()
        for range ticker.C {
            q.mu.Lock()
            if len(q.queue) < q.capacity/2 {
                q.dropping = false
                q.mu.Unlock()
                return
            }
            q.dropRate = min(q.dropRate+0.1, 0.8)  // 最高丢弃 80%
            q.mu.Unlock()
        }
    }()
}

func (q *LoadSheddingQueue) Process() {
    for req := range q.queue {
        q.semaphore <- struct{}{}
        go func(r Request) {
            defer func() { <-q.semaphore }()
            r.Handler(r.Context)
        }(req)
    }
}

# litmus-experiment.yaml
apiVersion: litmuschaos.io/v1alpha1
kind: ChaosEngine
metadata:
  name: order-service-chaos
  namespace: litmus
spec:
  appinfo:
    appns: 'production'
    applabel: 'app=order-service'
    appkind: 'deployment'
  chaosServiceAccount: litmus-admin
  experiments:
    - name: pod-cpu-hog
      spec:
        components:
          env:
            - name: CPU_CORES
              value: "2"
            - name: TOTAL_CHAOS_DURATION
              value: "120"
    - name: pod-memory-hog
      spec:
        components:
          env:
            - name: MEMORY_CONSUMPTION
              value: "500"
            - name: TOTAL_CHAOS_DURATION
              value: "120"

// SLO 监控指标定义
type SLOMetrics struct {
    // 可用性
    Availability    *prometheus.GaugeVec  // 服务可用百分比
    
    // 延迟
    LatencyP50      *prometheus.Histogram // 中位数延迟
    LatencyP99      *prometheus.Histogram // P99 延迟
    
    // 错误率
    ErrorRate       *prometheus.GaugeVec  // HTTP 5xx 比例
    
    // 吞吐量
    Throughput      *prometheus.CounterVec // 请求 QPS
    
    // 恢复时间
    MTTR            *prometheus.GaugeVec   // 平均恢复时间
    MTBF            *prometheus.GaugeVec   // 平均故障间隔
}

#!/bin/bash
# game-day-checklist.sh

echo "=== 混沌工程演练日检查清单 ==="

# 1. 确认监控告警就绪
echo "[1/5] 检查 Prometheus/Grafana 告警规则"
kubectl get prometheusrules -n monitoring | grep -i resilience

# 2. 确认降级策略生效
echo "[2/5] 验证降级开关"
curl -s http://api-gateway/actuator/features | jq '.fallbacks'

# 3. 确认 on-call 团队在线
echo "[3/5] 通知值班团队"
slack notify "#sre-alerts" "Game Day starting in 5 minutes. Chaos experiments: pod-kill, network-delay, cpu-hog"

# 4. 备份数据
echo "[4/5] 数据快照"
velero backup create game-day-backup

# 5. 确认回滚方案
echo "[5/5] 验证快速回滚能力"
kubectl rollout history deployment/order-service

echo "=== 检查完成，准备注入故障 ==="

# post-mortem-template.yaml
incident:
  id: chaos-game-day-20240915
  date: "2024-09-15T14:00:00Z"
  
  experiments:
    - type: pod-kill
      target: payment-service
      duration: 5m
      result: "successs"
      observations:
        - "断路器在 12s 后打开"
        - "降级到缓存支付状态，用户体验轻微降级"
        - "Pod 重启耗时 8s，期间请求排队"
      
    - type: network-delay
      target: inventory-service
      latency: 2s
      result: "partial"
      observations:
        - "超时 1.5s 后触发重试，库存查询变慢"
        - "用户侧体感延迟增大，但未报错"
  
  action_items:
    - "优化 payment-service 启动时间（目标 < 3s）"
    - "inventory-service 增加本地缓存，减少网络依赖"
    - "完善 SRE 手册：Pod 杀死后快速诊断流程"

# prometheus-rules.yaml
apiVersion: monitoring.coreos.com/v1
kind: PrometheusRule
metadata:
  name: resilience-alerts
spec:
  groups:
    - name: circuit-breaker
      rules:
        - alert: CircuitBreakerOpen
          expr: circuit_breaker_state{state="open"} == 1
          for: 1m
          labels:
            severity: warning
          annotations:
            summary: "Circuit breaker is open for {{ $labels.service }}"
        
        - alert: HighCircuitBreakerFailureRate
          expr: |
            rate(circuit_breaker_failures_total[5m]) /
            rate(circuit_breaker_requests_total[5m]) > 0.5
          for: 5m
          labels:
            severity: critical
          annotations:
            summary: "High failure rate for {{ $labels.service }}"
    
    - name: bulkhead
      rules:
        - alert: BulkheadQueueFull
          expr: bulkhead_queue_size / bulkhead_queue_capacity > 0.9
          for: 1m
          labels:
            severity: warning
          annotations:
            summary: "Bulkhead queue nearly full for {{ $labels.service }}"
    
    - name: retry
      rules:
        - alert: HighRetryRate
          expr: |
            rate(retry_attempts_total[5m]) /
            rate(retry_requests_total[5m]) > 0.3
          for: 5m
          labels:
            severity: warning
          annotations:
            summary: "High retry rate for {{ $labels.service }}"