【系统架构设计百科】全链路压测：大规模系统的性能验证

# 录制生产环境 80 端口的 HTTP 流量，保存到文件
sudo gor --input-raw :80 --output-file requests.gor

# 将录制的流量回放到测试环境，2 倍速
gor --input-file "requests.gor|200%" --output-http "http://staging:80"

# 实时镜像：将生产流量实时转发到测试环境
sudo gor --input-raw :80 --output-http "http://staging:80"

# 带过滤的录制：只录制 API 请求，排除健康检查
sudo gor --input-raw :80 \
  --http-allow-url "/api/" \
  --http-disallow-url "/health" \
  --output-file requests.gor

# 修改请求头，添加压测标记
sudo gor --input-raw :80 \
  --output-http "http://staging:80" \
  --http-set-header "X-Test-Flag: shadow"

# 按比例采样，只回放 10% 的流量
sudo gor --input-raw :80 \
  --output-http "http://staging:80|10%"

// 流量回放预处理器示例：替换会话令牌
package replay

import (
    "net/http"
    "strings"
    "sync"
)

type SessionManager struct {
    mu       sync.RWMutex
    sessions map[string]string // 原始 sessionID -> 新 sessionID
    loginURL string
    client   *http.Client
}

func NewSessionManager(loginURL string) *SessionManager {
    return &SessionManager{
        sessions: make(map[string]string),
        loginURL: loginURL,
        client:   &http.Client{},
    }
}

// ReplaceSession 替换请求中的会话令牌
func (sm *SessionManager) ReplaceSession(req *http.Request) error {
    oldToken := extractToken(req)
    if oldToken == "" {
        return nil
    }

    sm.mu.RLock()
    newToken, exists := sm.sessions[oldToken]
    sm.mu.RUnlock()

    if !exists {
        var err error
        newToken, err = sm.reLogin(oldToken)
        if err != nil {
            return err
        }
        sm.mu.Lock()
        sm.sessions[oldToken] = newToken
        sm.mu.Unlock()
    }

    req.Header.Set("Authorization", "Bearer "+newToken)
    return nil
}

func extractToken(req *http.Request) string {
    auth := req.Header.Get("Authorization")
    if strings.HasPrefix(auth, "Bearer ") {
        return strings.TrimPrefix(auth, "Bearer ")
    }
    return ""
}

func (sm *SessionManager) reLogin(originalToken string) (string, error) {
    // 根据原始令牌关联的用户信息重新登录
    // 返回新的访问令牌
    resp, err := sm.client.Post(sm.loginURL, "application/json", nil)
    if err != nil {
        return "", err
    }
    defer resp.Body.Close()
    // 解析响应获取新令牌（此处省略具体解析逻辑）
    return resp.Header.Get("X-New-Token"), nil
}

# Istio 流量镜像配置
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: order-service
spec:
  hosts:
    - order-service
  http:
    - route:
        - destination:
            host: order-service
            subset: production
          weight: 100
      mirror:
        host: order-service
        subset: shadow
      mirrorPercentage:
        value: 100.0  # 镜像 100% 的流量

// Java 应用层流量复制示例
@Component
public class TrafficMirrorFilter implements Filter {

    private final ExecutorService mirrorExecutor;
    private final HttpClient mirrorClient;
    private final String shadowHost;
    private final MeterRegistry meterRegistry;

    public TrafficMirrorFilter(
            @Value("${shadow.host}") String shadowHost,
            MeterRegistry meterRegistry) {
        this.shadowHost = shadowHost;
        this.meterRegistry = meterRegistry;
        // 独立线程池，避免影子请求影响主请求
        this.mirrorExecutor = new ThreadPoolExecutor(
            10, 50, 60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1000),
            new ThreadPoolExecutor.DiscardPolicy()  // 队列满时丢弃，保护主链路
        );
        this.mirrorClient = HttpClient.newBuilder()
            .connectTimeout(Duration.ofSeconds(2))
            .build();
    }

    @Override
    public void doFilter(ServletRequest request, ServletResponse response,
                         FilterChain chain) throws IOException, ServletException {
        HttpServletRequest httpReq = (HttpServletRequest) request;

        // 异步发送影子请求
        if (shouldMirror(httpReq)) {
            mirrorExecutor.submit(() -> sendShadowRequest(httpReq));
        }

        // 主请求正常处理，不受影子请求影响
        chain.doFilter(request, response);
    }

    private boolean shouldMirror(HttpServletRequest req) {
        // 只镜像读请求，写请求需要特殊处理
        String method = req.getMethod();
        return "GET".equals(method) || "POST".equals(method);
    }

    private void sendShadowRequest(HttpServletRequest originalReq) {
        try {
            String shadowUrl = shadowHost + originalReq.getRequestURI();
            HttpRequest.Builder builder = HttpRequest.newBuilder()
                .uri(URI.create(shadowUrl))
                .header("X-Shadow", "true")
                .header("X-Original-Host", originalReq.getServerName())
                .timeout(Duration.ofSeconds(5));

            HttpRequest shadowReq = builder
                .method(originalReq.getMethod(), HttpRequest.BodyPublishers.noBody())
                .build();

            HttpResponse<Void> resp = mirrorClient.send(
                shadowReq, HttpResponse.BodyHandlers.discarding());

            meterRegistry.counter("shadow.request",
                "status", String.valueOf(resp.statusCode())).increment();
        } catch (Exception e) {
            // 影子请求失败不影响主链路，仅记录指标
            meterRegistry.counter("shadow.request.error",
                "type", e.getClass().getSimpleName()).increment();
        }
    }
}

// 基于线程上下文的数据源路由
public class ShadowRoutingDataSource extends AbstractRoutingDataSource {

    @Override
    protected Object determineCurrentLookupKey() {
        return TrafficContext.isShadow() ? "shadow" : "production";
    }
}

// 流量上下文，通过 ThreadLocal 在调用链中传递
public class TrafficContext {

    private static final ThreadLocal<Boolean> SHADOW_FLAG = new ThreadLocal<>();

    public static void markShadow() {
        SHADOW_FLAG.set(Boolean.TRUE);
    }

    public static boolean isShadow() {
        return Boolean.TRUE.equals(SHADOW_FLAG.get());
    }

    public static void clear() {
        SHADOW_FLAG.remove();
    }
}

// 拦截器：从请求头中识别压测标记
@Component
public class ShadowTrafficInterceptor implements HandlerInterceptor {

    @Override
    public boolean preHandle(HttpServletRequest request,
                             HttpServletResponse response,
                             Object handler) {
        String shadowHeader = request.getHeader("X-Shadow");
        if ("true".equals(shadowHeader)) {
            TrafficContext.markShadow();
        }
        return true;
    }

    @Override
    public void afterCompletion(HttpServletRequest request,
                                HttpServletResponse response,
                                Object handler, Exception ex) {
        TrafficContext.clear();
    }
}

// Go 实现的压测标记传播中间件
package shadow

import (
    "context"
    "net/http"

    "google.golang.org/grpc/metadata"
)

const (
    ShadowHeaderKey  = "X-Shadow-Traffic"
    ShadowContextKey = "shadow_traffic"
)

type contextKey string

const shadowKey contextKey = ShadowContextKey

// IsShadow 判断当前请求是否为压测流量
func IsShadow(ctx context.Context) bool {
    v, ok := ctx.Value(shadowKey).(bool)
    return ok && v
}

// WithShadow 在 context 中标记压测流量
func WithShadow(ctx context.Context) context.Context {
    return context.WithValue(ctx, shadowKey, true)
}

// HTTPMiddleware 从 HTTP 请求头中提取压测标记
func HTTPMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        ctx := r.Context()
        if r.Header.Get(ShadowHeaderKey) == "true" {
            ctx = WithShadow(ctx)
        }
        next.ServeHTTP(w, r.WithContext(ctx))
    })
}

// GRPCUnaryInterceptor gRPC 一元调用拦截器，传播压测标记
func GRPCUnaryInterceptor(
    ctx context.Context,
    req interface{},
    info *grpc.UnaryServerInfo,
    handler grpc.UnaryHandler,
) (interface{}, error) {
    md, ok := metadata.FromIncomingContext(ctx)
    if ok {
        values := md.Get(ShadowHeaderKey)
        if len(values) > 0 && values[0] == "true" {
            ctx = WithShadow(ctx)
        }
    }
    return handler(ctx, req)
}

// 根据压测标记动态选择表名
func TableName(ctx context.Context, baseTable string) string {
    if IsShadow(ctx) {
        return baseTable + "_shadow"
    }
    return baseTable
}

// 使用示例
func (r *OrderRepository) Create(ctx context.Context, order *Order) error {
    table := TableName(ctx, "orders")
    query := fmt.Sprintf("INSERT INTO %s (id, user_id, amount, created_at) VALUES (?, ?, ?, ?)", table)
    _, err := r.db.ExecContext(ctx, query, order.ID, order.UserID, order.Amount, order.CreatedAt)
    return err
}

// Redis 键名隔离
func CacheKey(ctx context.Context, baseKey string) string {
    if IsShadow(ctx) {
        return "shadow:" + baseKey
    }
    return baseKey
}

// 使用示例
func (c *CacheService) GetUserProfile(ctx context.Context, userID string) (*Profile, error) {
    key := CacheKey(ctx, "user:profile:"+userID)
    data, err := c.redis.Get(ctx, key).Bytes()
    if err != nil {
        return nil, err
    }
    var profile Profile
    if err := json.Unmarshal(data, &profile); err != nil {
        return nil, err
    }
    return &profile, nil
}

// Kafka Topic 隔离
func TopicName(ctx context.Context, baseTopic string) string {
    if IsShadow(ctx) {
        return baseTopic + "_shadow"
    }
    return baseTopic
}

// 生产者发送消息时自动选择 Topic
func (p *Producer) Send(ctx context.Context, baseTopic string, msg []byte) error {
    topic := TopicName(ctx, baseTopic)
    return p.kafka.Produce(topic, msg)
}

// 影子数据清理任务
func (c *Cleaner) CleanShadowData(ctx context.Context) error {
    tables := []string{"orders_shadow", "payments_shadow", "inventory_shadow"}
    for _, table := range tables {
        query := fmt.Sprintf(
            "DELETE FROM %s WHERE created_at < NOW() - INTERVAL 24 HOUR",
            table,
        )
        result, err := c.db.ExecContext(ctx, query)
        if err != nil {
            return fmt.Errorf("清理 %s 失败: %w", table, err)
        }
        rows, _ := result.RowsAffected()
        log.Printf("已清理 %s 中 %d 条影子数据", table, rows)
    }
    return nil
}

// k6 全链路压测脚本示例
import http from 'k6/http';
import { check, sleep, group } from 'k6';
import { Rate, Trend } from 'k6/metrics';

// 自定义指标
const orderSuccessRate = new Rate('order_success_rate');
const orderDuration = new Trend('order_duration', true);

export const options = {
  scenarios: {
    // 阶梯式负载：逐步增加并发用户数
    ramp_up: {
      executor: 'ramping-arrival-rate',  // 开环模型
      startRate: 10,
      timeUnit: '1s',
      preAllocatedVUs: 500,
      maxVUs: 2000,
      stages: [
        { duration: '2m', target: 100 },   // 2 分钟内增加到 100 RPS
        { duration: '5m', target: 500 },   // 5 分钟内增加到 500 RPS
        { duration: '10m', target: 500 },  // 维持 500 RPS 10 分钟
        { duration: '5m', target: 1000 },  // 增加到 1000 RPS
        { duration: '10m', target: 1000 }, // 维持 1000 RPS 10 分钟
        { duration: '3m', target: 0 },     // 3 分钟内降到 0
      ],
    },
  },
  thresholds: {
    http_req_duration: ['p(95)<500', 'p(99)<1000'],  // P95 < 500ms, P99 < 1s
    http_req_failed: ['rate<0.01'],                   // 错误率 < 1%
    order_success_rate: ['rate>0.99'],                 // 下单成功率 > 99%
  },
};

const BASE_URL = __ENV.BASE_URL || 'http://staging:8080';
const SHADOW_HEADER = { 'X-Shadow': 'true' };

export default function () {
  group('浏览商品', function () {
    // 搜索商品
    const searchRes = http.get(
      `${BASE_URL}/api/products/search?q=手机&page=1`,
      { headers: SHADOW_HEADER }
    );
    check(searchRes, {
      '搜索返回 200': (r) => r.status === 200,
      '搜索结果非空': (r) => JSON.parse(r.body).items.length > 0,
    });

    sleep(Math.random() * 2 + 1);  // 模拟用户思考时间

    // 查看商品详情
    const products = JSON.parse(searchRes.body).items;
    if (products.length > 0) {
      const productId = products[Math.floor(Math.random() * products.length)].id;
      const detailRes = http.get(
        `${BASE_URL}/api/products/${productId}`,
        { headers: SHADOW_HEADER }
      );
      check(detailRes, {
        '详情返回 200': (r) => r.status === 200,
      });
    }
  });

  sleep(Math.random() * 3 + 1);

  group('下单流程', function () {
    const startTime = Date.now();

    // 添加购物车
    const cartRes = http.post(
      `${BASE_URL}/api/cart/items`,
      JSON.stringify({ productId: 'P10001', quantity: 1 }),
      { headers: { ...SHADOW_HEADER, 'Content-Type': 'application/json' } }
    );

    // 创建订单
    if (cartRes.status === 200) {
      const orderRes = http.post(
        `${BASE_URL}/api/orders`,
        JSON.stringify({ cartId: JSON.parse(cartRes.body).cartId }),
        { headers: { ...SHADOW_HEADER, 'Content-Type': 'application/json' } }
      );

      const success = orderRes.status === 201;
      orderSuccessRate.add(success);
      orderDuration.add(Date.now() - startTime);

      check(orderRes, {
        '下单返回 201': (r) => r.status === 201,
      });
    }
  });

  sleep(Math.random() * 5 + 2);
}

// 压测平台核心数据结构
package loadtest

import "time"

// Scenario 定义一个压测场景
type Scenario struct {
    ID          string        `json:"id"`
    Name        string        `json:"name"`
    TargetURL   string        `json:"target_url"`
    TrafficSpec TrafficSpec   `json:"traffic_spec"`
    Duration    time.Duration `json:"duration"`
    Thresholds  Thresholds    `json:"thresholds"`
    DataPolicy  DataPolicy    `json:"data_policy"`
}

// TrafficSpec 定义流量规格
type TrafficSpec struct {
    Model     LoadModel `json:"model"`      // open-loop 或 closed-loop
    InitialRPS int      `json:"initial_rps"`
    TargetRPS  int      `json:"target_rps"`
    RampUpTime time.Duration `json:"ramp_up_time"`
    Stages     []Stage  `json:"stages"`
}

type LoadModel string

const (
    OpenLoop  LoadModel = "open-loop"
    ClosedLoop LoadModel = "closed-loop"
)

type Stage struct {
    Duration  time.Duration `json:"duration"`
    TargetRPS int           `json:"target_rps"`
}

// Thresholds 定义压测通过的条件
type Thresholds struct {
    MaxP95Latency  time.Duration `json:"max_p95_latency"`
    MaxP99Latency  time.Duration `json:"max_p99_latency"`
    MaxErrorRate   float64       `json:"max_error_rate"`
    MinThroughput  int           `json:"min_throughput"`
}

// DataPolicy 定义数据隔离策略
type DataPolicy struct {
    Isolation    IsolationType `json:"isolation"`
    ShadowSuffix string        `json:"shadow_suffix"`
    TTL          time.Duration `json:"ttl"`
    CleanupAfter bool          `json:"cleanup_after"`
}

type IsolationType string

const (
    PhysicalIsolation IsolationType = "physical"
    ShadowTable       IsolationType = "shadow-table"
    LogicalIsolation  IsolationType = "logical"
)

// Result 压测结果
type Result struct {
    ScenarioID    string         `json:"scenario_id"`
    StartTime     time.Time      `json:"start_time"`
    EndTime       time.Time      `json:"end_time"`
    TotalRequests int64          `json:"total_requests"`
    SuccessCount  int64          `json:"success_count"`
    ErrorCount    int64          `json:"error_count"`
    Throughput    float64        `json:"throughput_rps"`
    Latency       LatencyStats   `json:"latency"`
    ErrorDetails  map[string]int `json:"error_details"`
    Passed        bool           `json:"passed"`
}

type LatencyStats struct {
    Min    time.Duration `json:"min"`
    Max    time.Duration `json:"max"`
    Mean   time.Duration `json:"mean"`
    Median time.Duration `json:"median"`
    P90    time.Duration `json:"p90"`
    P95    time.Duration `json:"p95"`
    P99    time.Duration `json:"p99"`
    P999   time.Duration `json:"p999"`
}

# GitHub Actions 性能测试工作流示例
name: Performance Test Gate

on:
  pull_request:
    branches: [main]

jobs:
  performance-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Deploy to staging
        run: |
          kubectl apply -f k8s/staging/

      - name: Wait for deployment
        run: |
          kubectl rollout status deployment/app -n staging --timeout=300s

      - name: Run k6 load test
        uses: grafana/k6-action@v0.3.1
        with:
          filename: tests/performance/baseline.js
          flags: --out influxdb=http://influxdb:8086/k6
        env:
          BASE_URL: http://staging-lb:8080

      - name: Check results
        run: |
          # 从 InfluxDB 查询压测结果，验证是否通过阈值
          P99=$(curl -s "http://influxdb:8086/query?db=k6&q=SELECT+percentile(value,99)+FROM+http_req_duration" | jq '.results[0].series[0].values[0][1]')
          ERROR_RATE=$(curl -s "http://influxdb:8086/query?db=k6&q=SELECT+mean(value)+FROM+http_req_failed" | jq '.results[0].series[0].values[0][1]')

          echo "P99 Latency: ${P99}ms"
          echo "Error Rate: ${ERROR_RATE}"

          if (( $(echo "$P99 > 500" | bc -l) )); then
            echo "FAILED: P99 latency exceeds 500ms threshold"
            exit 1
          fi

          if (( $(echo "$ERROR_RATE > 0.01" | bc -l) )); then
            echo "FAILED: Error rate exceeds 1% threshold"
            exit 1
          fi

          echo "PASSED: All performance thresholds met"

// 性能基线对比
func CompareWithBaseline(current, baseline *Result) *ComparisonReport {
    report := &ComparisonReport{
        CurrentVersion:  current.ScenarioID,
        BaselineVersion: baseline.ScenarioID,
    }

    // 延迟对比
    p99Change := float64(current.Latency.P99-baseline.Latency.P99) /
        float64(baseline.Latency.P99) * 100
    report.P99LatencyChange = p99Change

    // 吞吐量对比
    throughputChange := (current.Throughput - baseline.Throughput) /
        baseline.Throughput * 100
    report.ThroughputChange = throughputChange

    // 判定是否通过
    report.Passed = p99Change < 10.0 && throughputChange > -5.0

    return report
}

type ComparisonReport struct {
    CurrentVersion   string  `json:"current_version"`
    BaselineVersion  string  `json:"baseline_version"`
    P99LatencyChange float64 `json:"p99_latency_change_pct"`
    ThroughputChange float64 `json:"throughput_change_pct"`
    Passed           bool    `json:"passed"`
}