当前位置：首页 > news >正文

全链路压测的可观测性设计：压测流量和真实流量的染色区分

news 2026/7/18 22:11:38

全链路压测的可观测性设计：压测流量和真实流量的染色区分

一、压测数据污染生产指标的噩梦

技术团队在凌晨两点做全链路压测，目标是模拟双 11 级别的流量。压测顺利结束了，但第二天早上发现一个严重问题：压测的流量被混入了生产监控，导致 QPS 翻倍、错误率飙升、数据库连接数告警误触发了 20 多条。

更严重的是，部分压测请求落入了生产数据库——虽然没有造成数据污染，但压测生成的测试订单和真实订单混在一起，数据分析团队花了整整一天来清洗。

根本原因：压测流量和生产流量的区分没有做到链路透传。

二、流量染色与全链路透传架构

flowchart TB subgraph Ingress["入口层"] GW["API 网关"] --> Tagger["流量染色器\nHeader: X-Stress-Test: true"] end subgraph Service["服务层"] Tagger --> SVC1["服务 A"] SVC1 -->|"透传 Header"| SVC2["服务 B"] SVC1 -->|"透传 Header"| SVC3["服务 C"] SVC2 -->|"透传 Header"| SVC4["服务 D"] end subgraph Infra["基础设施层"] SVC1 --> DB["数据库\n→ 压测影子表"] SVC1 --> MQ["消息队列\n→ 压测 Topic"] SVC1 --> Cache["缓存\n→ 压测 Key 前缀"] end subgraph Observability["可观测性"] SVC1 --> Metrics["指标\n标签: stress_test=true"] SVC1 --> Logs["日志\n字段: stress_test"] SVC1 --> Traces["链路追踪\nBaggage: stress_test"] Metrics --> Grafana["Grafana\n双面板展示"] Logs --> ES["ES\n索引分离"] Traces --> Jaeger["Jaeger\n采样率区分"] end

流量染色的三个关键环节：

入口打标——在网关层给请求打上染色标签
全链路透传——所有服务间调用透传染色标记
基础设施分离——数据库、消息队列、缓存使用影子资源

三、生产级 Go 实现

流量染色传播

package tracing import ( "context" "net/http" ) // StressTestBaggage 压测标记——在 OpenTelemetry Baggage 中传递 const ( // HeaderStressTest 压测流量请求头 HeaderStressTest = "X-Stress-Test" // BaggageKeyStressTest 链路传播的 Key BaggageKeyStressTest = "stress_test" ) // IsStressTest 判断当前上下文是否为压测流量 func IsStressTest(ctx context.Context) bool { if v, ok := ctx.Value(BaggageKeyStressTest).(bool); ok { return v } // 从 OpenTelemetry Baggage 中读取 // 生产环境建议使用 otel baggage API return false } // WithStressTest 在上下文中标记为压测流量 func WithStressTest(ctx context.Context) context.Context { return context.WithValue(ctx, BaggageKeyStressTest, true) } // StressTestMiddleware Gin/HTTP 中间件——在入口处提取压测标记 func StressTestMiddleware() func(http.Handler) http.Handler { return func(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { ctx := r.Context() // 从请求头中提取压测标记 if r.Header.Get(HeaderStressTest) == "true" { ctx = WithStressTest(ctx) // 在响应头中也标记（方便下游识别） w.Header().Set(HeaderStressTest, "true") } next.ServeHTTP(w, r.WithContext(ctx)) }) } }

数据库影子方案

package database import ( "context" "database/sql" "fmt" ) // ShadowDBManager 影子数据库管理器 type ShadowDBManager struct { primary *sql.DB // 生产数据库连接 shadow *sql.DB // 压测影子数据库连接 } // GetDB 根据压测标记返回对应的数据库连接 func (sm *ShadowDBManager) GetDB(ctx context.Context) *sql.DB { if IsStressTest(ctx) { return sm.shadow } return sm.primary } // ShadowTableRewriter SQL 表名改写器 // 将压测 SQL 中的表名自动添加 _stress 后缀 type ShadowTableRewriter struct { // 需要创建影子表的表名列表 tables []string } func (str *ShadowTableRewriter) RewriteSQL(ctx context.Context, sql string) string { if !IsStressTest(ctx) { return sql } rewritten := sql for _, table := range str.tables { // 简单替换：将表名替换为影子表名 // 生产环境建议使用 SQL parser 做更精确的替换 rewritten = replaceTableName(rewritten, table, table+"_stress") } return rewritten } func replaceTableName(sql, from, to string) string { // 生产环境建议使用 vitess/go-sqlparser 等库 return sql // 简化实现 }

可观测性集成

package observability import ( "context" "fmt" "time" ) // StressTestAwareMetrics 压测感知的指标收集器 type StressTestAwareMetrics struct { // Prometheus Counter Vec — 按 stress_test 标签区分 } // RecordQPS 记录 QPS，自动区分压测和生产 func (m *StressTestAwareMetrics) RecordQPS(ctx context.Context, endpoint string) { isStress := IsStressTest(ctx) labelValue := "false" if isStress { labelValue = "true" } // prometheus.Counter.WithLabelValues(endpoint, labelValue).Inc() fmt.Printf("[METRIC] QPS endpoint=%s stress_test=%s\n", endpoint, labelValue) } // StressTestAwareLogger 压测感知的日志记录器 type StressTestAwareLogger struct{} func (l *StressTestAwareLogger) Info(ctx context.Context, format string, args ...interface{}) { isStress := IsStressTest(ctx) prefix := "[PROD]" if isStress { prefix = "[STRESS]" } msg := fmt.Sprintf(format, args...) fmt.Printf("%s %s %s\n", time.Now().Format(time.RFC3339), prefix, msg) }

网关层流量染色配置

# Nginx / API Gateway 配置示例 # 根据请求来源决定是否染色 # 方案一：通过特定 Header 标记 # 压测工具发送时带上 X-Stress-Test: true # 方案二：根据请求来源 IP # 如果请求来自压测机器的 IP 段，自动染色 # 方案三：根据流量比例染色（A/B 测试） # 对 1% 的流量染色，用于影子流量验证

// 在应用层统一管理压测标识 const ( // 压测 Key 前缀（Redis 缓存） StressKeyPrefix = "stress:" // 压测 Topic 前缀（消息队列） StressTopicPrefix = "stress_" ) // BuildCacheKey 根据压测标记构造缓存 Key func BuildCacheKey(ctx context.Context, baseKey string) string { if IsStressTest(ctx) { return StressKeyPrefix + baseKey } return baseKey } // BuildTopicName 根据压测标记构造消息队列 Topic func BuildTopicName(ctx context.Context, baseTopic string) string { if IsStressTest(ctx) { return StressTopicPrefix + baseTopic } return baseTopic }