OpenTelemetry 开发实战

# docker-compose 简化示例
services:
  otel-collector:
    image: otel/opentelemetry-collector-contrib:latest
    volumes:
      - ./otel-collector-config.yaml:/etc/otelcol-contrib/config.yaml
    ports:
      - "4317:4317"
      - "4318:4318"
      - "8889:8889"
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - "9090:9090"
  jaeger:
    image: jaegertracing/all-in-one:latest
    ports:
      - "16686:16686"
      - "14250:14250"

// pkg/otel/setup.go
package otel

import (
    "context"
    "errors"
    "fmt"
    "time"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
    "go.opentelemetry.io/otel/propagation"
    sdkmetric "go.opentelemetry.io/otel/sdk/metric"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.21.0"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
)

// Config OTel 初始化配置
type Config struct {
    ServiceName    string  // 必填，所有数据都会带这个标签
    ServiceVersion string  // 建议填，版本信息方便排查
    Environment    string  // production/staging/dev
    OTLPEndpoint   string  // Collector 地址，比如 localhost:4317
    SampleRatio    float64 // 采样率 0~1，生产建议 0.1~0.3
}

// Providers 把初始化结果打包返回，方便调用方在退出时 shutdown
type Providers struct {
    TracerProvider *sdktrace.TracerProvider
    MeterProvider  *sdkmetric.MeterProvider
}

// Setup 一把初始化 TracerProvider + MeterProvider
// 调用方必须在退出时调用 Shutdown，否则 buffer 里的数据会丢
func Setup(ctx context.Context, cfg Config) (*Providers, error) {
    if cfg.ServiceName == "" {
        return nil, errors.New("service name is required")
    }
    if cfg.OTLPEndpoint == "" {
        cfg.OTLPEndpoint = "localhost:4317"
    }
    if cfg.SampleRatio <= 0 {
        cfg.SampleRatio = 1.0 // 默认全采样，开发环境用
    }

    // 资源属性：每个 span/metric/log 都会带上这些
    // service.name 必填，其他都是强烈建议
    res, err := resource.New(ctx,
        resource.WithAttributes(
            semconv.ServiceName(cfg.ServiceName),
            semconv.ServiceVersion(cfg.ServiceVersion),
            semconv.DeploymentEnvironment(cfg.Environment),
        ),
    )
    if err != nil {
        return nil, fmt.Errorf("create resource failed: %w", err)
    }

    // ---- Trace 初始化 ----
    // 关键：dial 一次 gRPC 连接复用给 trace 和 metric，别 new 两次
    // 每次都 dial 会在高 QPS 下耗光连接
    conn, err := grpc.DialContext(ctx, cfg.OTLPEndpoint,
        grpc.WithTransportCredentials(insecure.NewCredentials()),
        // 关键：block=true 等连接建立，否则 exporter 第一次发送会失败
        grpc.WithBlock(),
    )
    if err != nil {
        return nil, fmt.Errorf("dial collector failed: %w", err)
    }

    traceExp, err := otlptracegrpc.New(ctx,
        otlptracegrpc.WithGRPCConn(conn),
    )
    if err != nil {
        return nil, fmt.Errorf("create trace exporter failed: %w", err)
    }

    // 采样器：ParentBased 让子 span 跟随父 span 决策
    // TraceIDRatioBased 按比例采样，简单实用
    sampler := sdktrace.ParentBased(
        sdktrace.TraceIDRatioBased(cfg.SampleRatio),
    )

    tp := sdktrace.NewTracerProvider(
        sdktrace.WithResource(res),
        sdktrace.WithSampler(sampler),
        // BatchSpanProcessor：批量发送，生产必用
        // 关键参数：BatchTimeout 越小延迟越低但吞吐越差
        sdktrace.WithBatcher(traceExp,
            sdktrace.WithBatchTimeout(2*time.Second),
            sdktrace.WithMaxExportBatchSize(512),
        ),
        // 关键：OnExport 失败时记日志，否则 exporter 静默丢数据你都不知道
        // sdktrace.WithRawExporter 之类的高级用法这里不展开
    )
    otel.SetTracerProvider(tp)

    // ---- Metric 初始化 ----
    metricExp, err := otlpmetricgrpc.New(ctx,
        otlpmetricgrpc.WithGRPCConn(conn),
    )
    if err != nil {
        return nil, fmt.Errorf("create metric exporter failed: %w", err)
    }

    // PeriodicReader：每隔一段时间把累积的 metric 推出去
    // 关键：间隔太短会增加 Collector 压力，太长则告警延迟
    reader := sdkmetric.NewPeriodicReader(metricExp,
        sdkmetric.WithInterval(15*time.Second),
        sdkmetric.WithTimeout(5*time.Second),
    )

    mp := sdkmetric.NewMeterProvider(
        sdkmetric.WithResource(res),
        sdkmetric.WithReader(reader),
    )
    otel.SetMeterProvider(mp)

    // 全局 propagator：让 trace context 能跨进程传递
    // baggage 是用来传业务自定义键值对的，跟 trace 上下文一起走
    otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
        propagation.TraceContext{}, // W3C traceparent
        propagation.Baggage{},      // W3C baggage
    ))

    return &Providers{TracerProvider: tp, MeterProvider: mp}, nil
}

// Shutdown 优雅退出，必须调用
// 不调的话 BatchSpanProcessor 里没发出去的 span 会丢
// 我踩过坑：以为进程退出 OS 会自动 flush，实际上 OTel 没这逻辑
func (p *Providers) Shutdown(ctx context.Context) error {
    var errs []error
    if p.TracerProvider != nil {
        if err := p.TracerProvider.Shutdown(ctx); err != nil {
            errs = append(errs, fmt.Errorf("tracer shutdown: %w", err))
        }
    }
    if p.MeterProvider != nil {
        if err := p.MeterProvider.Shutdown(ctx); err != nil {
            errs = append(errs, fmt.Errorf("meter shutdown: %w", err))
        }
    }
    return errors.Join(errs...)
}

// cmd/api/main.go
package main

import (
    "context"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    "yourapp/pkg/otel"
)

func main() {
    ctx, cancel := signal.NotifyContext(context.Background(),
        syscall.SIGINT, syscall.SIGTERM)
    defer cancel()

    providers, err := otel.Setup(ctx, otel.Config{
        ServiceName:    "order-service",
        ServiceVersion: "v1.2.3",
        Environment:    "production",
        OTLPEndpoint:   "otel-collector:4317",
        SampleRatio:    0.2,
    })
    if err != nil {
        log.Fatalf("otel setup failed: %v", err)
    }
    // 关键：defer shutdown，但要用独立的 context，因为主 ctx 可能已被 cancel
    defer func() {
        shutdownCtx, c := context.WithTimeout(context.Background(), 5*time.Second)
        defer c()
        if err := providers.Shutdown(shutdownCtx); err != nil {
            log.Printf("otel shutdown failed: %v", err)
        }
    }()

    // 业务代码...
    <-ctx.Done()
    log.Println("service shutting down")
}

// internal/order/service.go
package order

import (
    "context"
    "errors"
    "fmt"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/baggage"
    "go.opentelemetry.io/otel/codes"
    "go.opentelemetry.io/otel/trace"
)

type Service struct {
    tracer trace.Tracer
    repo   Repo
    cache  Cache
}

func NewService(repo Repo, cache Cache) *Service {
    // tracer 名字建议用包名或者服务名，方便筛选
    return &Service{
        tracer: otel.Tracer("order-service"),
        repo:   repo,
        cache:  cache,
    }
}

// CreateOrder 创建订单，演示一个完整的 span 用法
func (s *Service) CreateOrder(ctx context.Context, userID string, items []string) (*Order, error) {
    // 1. 创建 span。ctx 一定要传进来，否则跟父 span 断了
    // span 名建议用动词+名词，比如 "order.create"，别用 "CreateOrder" 函数名
    ctx, span := s.tracer.Start(ctx, "order.create",
        // 关键：业务属性在这里塞，排查问题就靠这些
        trace.WithAttributes(
            attribute.String("user.id", userID),
            attribute.Int("items.count", len(items)),
        ),
        // span kind：server/internal/client/producer/consumer
        // 内部业务逻辑用 Internal，接收外部请求用 Server
        trace.WithSpanKind(trace.SpanKindInternal),
    )
    defer span.End()

    // 2. baggage：跨 span 跨进程传业务键值对
    // 跟 attribute 的区别：baggage 会跟着 trace context 传到下游服务
    // 比如把 user.id 一直传到 payment-service
    if m, err := baggage.NewMember("user.id", userID); err == nil {
        if b, err := baggage.New(m); err == nil {
            ctx = baggage.ContextWithBaggage(ctx, b)
        }
    }

    // 3. 缓存查询子 span
    cacheHit, err := s.checkCache(ctx, userID)
    if err != nil {
        // 关键：错误不要直接 span.SetStatus，先记 event 再标记
        // event 比 status 信息更丰富，带时间戳和属性
        span.AddEvent("cache.check.failed", trace.WithAttributes(
            attribute.String("error.type", "redis_error"),
            attribute.String("error.message", err.Error()),
        ))
        // 缓存失败不阻断主流程，继续走 DB
    }

    if cacheHit != nil {
        span.SetAttributes(attribute.Bool("cache.hit", true))
        return cacheHit, nil
    }
    span.SetAttributes(attribute.Bool("cache.hit", false))

    // 4. 写 DB
    order, err := s.repo.Create(ctx, userID, items)
    if err != nil {
        // 错误处理标准三件套：record error + set status + return
        return s.recordError(ctx, span, "db.create.failed", err)
    }

    // 5. 发事件，记录关键业务节点
    span.AddEvent("order.created", trace.WithAttributes(
        attribute.String("order.id", order.ID),
        attribute.Float64("order.amount", order.Amount),
    ))

    return order, nil
}

// checkCache 演示一个子 span
func (s *Service) checkCache(ctx context.Context, userID string) (*Order, error) {
    // 关键：子 span 用 ctx 启动，自动建立 parent 关系
    ctx, span := s.tracer.Start(ctx, "cache.get")
    defer span.End()

    span.SetAttributes(attribute.String("cache.key", "user:"+userID))

    order, err := s.cache.Get(ctx, "user:"+userID)
    if err != nil {
        // 这里别 span.SetStatus(Error)，缓存失败不是业务错误
        // 加个 attribute 标记一下就行，主 span 自己决定要不要降级
        span.SetAttributes(attribute.String("cache.error", err.Error()))
        return nil, err
    }
    return order, nil
}

// recordError 统一错误处理，避免每个地方都写三行
func (s *Service) recordError(ctx context.Context, span trace.Span, event string, err error) (*Order, error) {
    // 1. RecordError：标准 OTel API，会自动加 exception.type/message/stacktrace 属性
    span.RecordError(err)
    // 2. SetStatus：把 span 标记为 ERROR，Jaeger 里会显示红色
    span.SetStatus(codes.Error, err.Error())
    // 3. AddEvent：再加个自定义事件，方便按事件名筛选
    span.AddEvent(event, trace.WithAttributes(
        attribute.String("error.class", fmt.Sprintf("%T", err)),
    ))
    // 4. 返回包装后的错误，业务层照常处理
    return nil, fmt.Errorf("create order failed: %w", err)
}

// Order 订单结构
type Order struct {
    ID     string  `json:"id"`
    UserID string  `json:"user_id"`
    Amount float64 `json:"amount"`
    Items  []string `json:"items"`
}

// Repo 接口抽象
type Repo interface {
    Create(ctx context.Context, userID string, items []string) (*Order, error)
}

// Cache 接口抽象
type Cache interface {
    Get(ctx context.Context, key string) (*Order, error)
}

var _ = errors.New  // 防 import 报错

// internal/order/metrics.go
package order

import (
    "context"
    "time"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/metric"
)

// Metrics 把所有指标集中起来，方便管理
type Metrics struct {
    ordersCreated  metric.Int64Counter
    orderDuration  metric.Float64Histogram
    ordersInFlight metric.Int64UpDownCounter
    cacheSize      metric.Int64ObservableGauge
    cache          Cache
}

func NewMetrics(cache Cache) (*Metrics, error) {
    meter := otel.Meter("order-service")

    m := &Metrics{cache: cache}

    var err error
    // 1. Counter：只能加不能减，用于累计计数
    // 关键：description 和 unit 别省，Grafana 配置时要用
    m.ordersCreated, err = meter.Int64Counter(
        "orders_created_total",
        metric.WithDescription("Total number of orders created"),
        metric.WithUnit("{order}"),
    )
    if err != nil {
        return nil, err
    }

    // 2. Histogram：分桶统计，用于延迟分布
    // 关键：boundaries 要根据业务设，默认的边界偏 Web 通用
    // 订单创建场景 5ms~10s 比较合理，别用默认的 0~1000s
    m.orderDuration, err = meter.Float64Histogram(
        "order_duration_seconds",
        metric.WithDescription("Order creation duration"),
        metric.WithUnit("s"),
        metric.WithExplicitBucketBoundaries(0.005, 0.01, 0.05, 0.1, 0.3, 0.5, 1, 3, 5, 10),
    )
    if err != nil {
        return nil, err
    }

    // 3. UpDownCounter：可加可减，用于当前值（队列长度、活跃请求数）
    m.ordersInFlight, err = meter.Int64UpDownCounter(
        "orders_in_flight",
        metric.WithDescription("Number of orders being processed"),
        metric.WithUnit("{order}"),
    )
    if err != nil {
        return nil, err
    }

    // 4. ObservableGauge：异步指标，SDK 采集时回调你提供的函数拿值
    // 适合那些"已经有现成数据源"的场景，比如 Redis 连接池当前大小
    // 关键：回调函数会被 SDK 周期性调用，别在里面做重活
    m.cacheSize, err = meter.Int64ObservableGauge(
        "cache_size_bytes",
        metric.WithDescription("Current cache size in bytes"),
        metric.WithUnit("By"),
    )
    if err != nil {
        return nil, err
    }
    // RegisterCallback 注册采集回调
    // 关键：回调返回的 Observation 必须用 observable 个体的 ObserveInt64 方法
    _, err = meter.RegisterCallback(func(ctx context.Context, observer metric.Observer) error {
        size := m.cache.Size(ctx) // 业务侧自己实现
        observer.ObserveInt64(m.cacheSize, size)
        return nil
    }, m.cacheSize)
    if err != nil {
        return nil, err
    }

    return m, nil
}

// RecordOrder 在创建订单的业务函数里调用
func (m *Metrics) RecordOrder(ctx context.Context, duration time.Duration, success bool) {
    // Counter：按结果加标签，方便分别看成功率
    attrs := metric.WithAttributes()
    if !success {
        // 关键：标签值不要用高基数字段，比如 order_id
        // 高基数会把 Prometheus 内存撑爆
        attrs = metric.WithAttributes() // 实际用 attribute.String("result", "error")
    }
    m.ordersCreated.Add(ctx, 1, attrs)
    m.orderDuration.Record(ctx, duration.Seconds())
}

// StartInFlight 标记订单开始处理
func (m *Metrics) StartInFlight(ctx context.Context) {
    m.ordersInFlight.Add(ctx, 1)
}

// EndInFlight 标记订单处理结束
func (m *Metrics) EndInFlight(ctx context.Context) {
    m.ordersInFlight.Add(ctx, -1)
}

# otel-collector-config.yaml
# 用 otel/opentelemetry-collector-contrib 镜像跑，基础镜像不带这些组件

# ============ Receivers：数据入口 ============
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
        # 关键：max_recv_msg_size 默认 4MB，大 trace 会超限
        max_recv_msg_size: 16MB
      http:
        endpoint: 0.0.0.0:4318

  # 顺手收 hostmetrics，监控 Collector 自己
  hostmetrics:
    collection_interval: 30s
    scrapers:
      cpu: {}
      memory: {}
      disk: {}
      network: {}

# ============ Processors：数据处理 ============
processors:
  # 1. memory_limiter 必须放最前面，防 OOM
  memory_limiter:
    check_interval: 1s
    limit_mib: 1500
    spike_limit_mib: 300

  # 2. resource 补全资源属性
  resource:
    attributes:
      - key: deployment.environment
        value: production
        action: upsert
      - key: k8s.cluster.name
        value: prod-bj
        action: upsert
      # 关键：删掉应用没用的属性，避免后端索引膨胀
      - key: process.pid
        action: delete

  # 3. filter 过滤无用数据
  filter:
    traces:
      span:
        # 健康检查 span 不要，纯噪音
        - 'attributes["http.route"] == "/healthz"'
        - 'attributes["http.route"] == "/metrics"'
        - 'attributes["http.route"] == "/ping"'
    metrics:
      metric:
        # 系统指标里一些没用的也过滤掉
        - 'name == "process.runtime.go.gc.heap_frees"'

  # 4. attributes 脱敏
  attributes:
    actions:
      # 删掉敏感请求头
      - key: http.request.header.authorization
        action: delete
      - key: http.request.header.cookie
        action: delete
      # SQL 文本 hash 掉，不存明文
      - key: db.statement
        action: hash

  # 5. tail_sampling 尾部采样
  # 注意：用 tail_sampling 后 batch 必须放在它后面
  tail_sampling:
    decision_wait: 10s
    num_traces: 50000
    expected_new_traces_per_sec: 1000
    policies:
      # 错误 trace 100% 保留
      - name: errors
        type: status_code
        status_code:
          status_codes: [ERROR]
      # 慢 trace 100% 保留
      - name: slow
        type: latency
        latency:
          threshold_ms: 500
      # 正常 trace 1% 采样
      - name: random_low
        type: probabilistic
        probabilistic:
          sampling_percentage: 1

  # 6. batch 批处理，放最后
  batch:
    timeout: 5s
    send_batch_size: 1024
    send_batch_max_size: 2048

# ============ Exporters：数据出口 ============
exporters:
  # 1. Prometheus Remote Write 推 metrics
  prometheusremotewrite:
    endpoint: http://prometheus:9090/api/v1/write
    # 关键：external labels 帮助区分集群
    external_labels:
      cluster: prod-bj
    # 写失败重试
    retry_on_failure:
      enabled: true
      initial_interval: 1s
      max_interval: 30s
      max_elapsed_time: 300s

  # 2. OTLP 推 traces 到 Jaeger
  # Jaeger 1.35+ 支持 OTLP 直接接收
  otlp/jaeger:
    endpoint: jaeger:4317
    tls:
      insecure: true

  # 3. Loki 推 logs
  # 注意：loki exporter 在 contrib 镜像里
  loki:
    endpoint: http://loki:3100/loki/api/v1/push
    # 关键：default_labels_added 会给所有日志加标签，方便 LogQL 筛选
    default_labels_added:
      - exporter: OTLP
      - environment: production

  # 4. debug exporter，开发时用
  debug:
    verbosity: basic
    sampling_initial: 2
    sampling_thereafter: 1

# ============ Extensions：辅助组件 ============
extensions:
  # 健康检查，K8s liveness probe 用
  health_check:
    endpoint: 0.0.0.0:13133
  # pprof，性能分析用
  pprof:
    endpoint: 0.0.0.0:1777
  # zpages，运行时调试页面
  zpages:
    endpoint: 0.0.0.0:55679

# ============ Service：流水线组装 ============
service:
  extensions: [health_check, pprof, zpages]
  telemetry:
    logs:
      level: info
    metrics:
      address: 0.0.0.0:8888
  pipelines:
    # metrics 流水线
    metrics:
      receivers: [otlp, hostmetrics]
      processors: [memory_limiter, resource, filter, batch]
      exporters: [prometheusremotewrite, debug]

    # traces 流水线
    # 关键：tail_sampling 必须在 batch 之前
    traces:
      receivers: [otlp]
      processors: [memory_limiter, resource, filter, attributes, tail_sampling, batch]
      exporters: [otlp/jaeger, debug]

    # logs 流水线
    logs:
      receivers: [otlp]
      processors: [memory_limiter, resource, batch]
      exporters: [loki, debug]

// api-gateway: 接收外部请求，发起对下游的 HTTP 调用
package main

import (
    "net/http"

    "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/propagation"
)

func main() {
    // 关键：用 otelhttp.NewHandler 包整个 mux，所有入口请求自动建 server span
    // span 名是 HTTP 方法+路径，比如 "GET /api/orders"
    mux := http.NewServeMux()
    mux.HandleFunc("/api/orders", proxyToOrderService)

    handler := otelhttp.NewHandler(mux, "api-gateway",
        otelhttp.WithTracerProvider(otel.GetTracerProvider()),
        // 关键：用 composite propagator 同时支持 traceparent 和 baggage
        otelhttp.WithPropagators(propagation.NewCompositeTextMapPropagator(
            propagation.TraceContext{},
            propagation.Baggage{},
        )),
    )
    _ = http.ListenAndServe(":8080", handler)
}

func proxyToOrderService(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context()

    // 关键：用 otelhttp.NewTransport 包 transport，下游调用自动建 client span
    // client span 会把 trace context 注入到 HTTP 头里，下游服务就能续上
    client := &http.Client{
        Transport: otelhttp.NewTransport(http.DefaultTransport),
    }

    req, _ := http.NewRequestWithContext(ctx, "GET", "http://order-service:8080/orders", nil)
    // 注意：不要手动设置 traceparent 头！
    // otelhttp.Transport 会自动从 ctx 提取并注入
    resp, err := client.Do(req)
    if err != nil {
        http.Error(w, err.Error(), 500)
        return
    }
    defer resp.Body.Close()
    // 复制响应...
}

// order-service: 接收 gateway 请求，调 Redis
package main

import (
    "net/http"

    "go.opentelemetry.io/contrib/instrumentation/github.com/redis/go-redis/v9/redisotel"
    "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
    "go.opentelemetry.io/otel"
    "github.com/redis/go-redis/v9"
)

func main() {
    // Redis 客户端：用 redisotel 包装，所有 Redis 命令自动建 span
    rdb := redis.NewClient(&redis.Options{Addr: "redis:6379"})
    // 关键：Hook 必须在客户端初始化时加，运行时不能改
    rdb.AddHook(redisotel.NewTracingHook())

    mux := http.NewServeMux()
    mux.HandleFunc("/orders", func(w http.ResponseWriter, r *http.Request) {
        ctx := r.Context()

        // 这里 ctx 已经被 otelhttp 注入了 trace context
        // 后续所有用这个 ctx 的调用（包括 Redis）都会自动续上 trace
        val, err := rdb.Get(ctx, "order:123").Result()
        if err != nil {
            http.Error(w, err.Error(), 500)
            return
        }
        w.Write([]byte(val))
    })

    handler := otelhttp.NewHandler(mux, "order-service")
    _ = http.ListenAndServe(":8080", handler)
}

// 生产者：发消息时注入 trace context
func produce(ctx context.Context, writer *kafka.Writer, msgValue []byte) error {
    ctx, span := otel.Tracer("producer").Start(ctx, "kafka.produce",
        trace.WithSpanKind(trace.SpanKindProducer),
    )
    defer span.End()

    msg := kafka.Message{Value: msgValue}
    // 关键：把 trace context 注入到 Kafka 消息头
    // 消费者读出来后能续上 trace
    otel.GetTextMapPropagator().Inject(ctx, propagation.NewCarrierWriter(&msg.HeaderCarrier))

    return writer.WriteMessages(ctx, msg)
}

// 消费者：读消息时提取 trace context
func consume(ctx context.Context, reader *kafka.Reader) error {
    msg, err := reader.ReadMessage(ctx)
    if err != nil {
        return err
    }

    // 关键：从消息头提取 trace context，建立消费者 span
    ctx = otel.GetTextMapPropagator().Extract(ctx, propagation.NewCarrierReader(&msg.HeaderCarrier))
    ctx, span := otel.Tracer("consumer").Start(ctx, "kafka.consume",
        trace.WithSpanKind(trace.SpanKindConsumer),
    )
    defer span.End()

    // 后续业务处理就跟普通 HTTP 请求一样了
    return processMessage(ctx, msg.Value)
}

# aiops/anomaly_detect.py
import requests
import numpy as np
from datetime import datetime, timedelta

PROM_BASE = "http://prometheus:9090"

def query_prom(query: str, t: datetime = None) -> list:
    """查 Prometheus，返回 series 列表"""
    t = t or datetime.now()
    params = {"query": query, "time": t.timestamp()}
    r = requests.get(f"{PROM_BASE}/api/v1/query", params=params, timeout=10)
    r.raise_for_status()
    return r.json()["data"]["result"]

def query_range(query: str, start: datetime, end: datetime, step: str = "60s") -> list:
    """范围查询，返回每个 series 的时间序列"""
    params = {
        "query": query,
        "start": start.timestamp(),
        "end": end.timestamp(),
        "step": step,
    }
    r = requests.get(f"{PROM_BASE}/api/v1/query_range", params=params, timeout=30)
    r.raise_for_status()
    return r.json()["data"]["result"]

def detect_metric_anomaly(values: list, threshold_sigma: float = 3.0) -> dict:
    """
    3-sigma 异常检测：最近一个点偏离历史均值超过 N 个标准差就告警
    返回 {is_anomaly, current, mean, std, zscore}
    """
    if len(values) < 30:
        # 关键：数据点太少不可靠，曾经我踩过坑：3 个点算 std 直接 0
        return {"is_anomaly": False, "reason": "insufficient_data"}

    arr = np.array(values, dtype=float)
    history = arr[:-1]  # 前 N-1 个做基线
    current = arr[-1]   # 最后一个判断

    mean = float(np.mean(history))
    std = float(np.std(history))

    # 关键：std 为 0 时不能除，加个 epsilon
    if std < 1e-9:
        return {"is_anomaly": False, "reason": "zero_variance"}

    zscore = abs(current - mean) / std
    return {
        "is_anomaly": zscore > threshold_sigma,
        "current": float(current),
        "mean": mean,
        "std": std,
        "zscore": float(zscore),
    }

def check_service_error_rate(service: str) -> dict:
    """检查某个服务的错误率是否异常"""
    now = datetime.now()
    start = now - timedelta(hours=1)

    # 关键：rate() 的窗口要够长，太短抖动大
    # 5m 窗口是常见的折中
    query = f'''
        sum(rate(http_requests_total{{service="{service}",status=~"5.."}}[5m])) by (service)
        /
        sum(rate(http_requests_total{{service="{service}"}}[5m])) by (service)
    '''
    series = query_range(query, start, now, "60s")
    if not series:
        return {"service": service, "status": "no_data"}

    results = []
    for s in series:
        values = [float(v[1]) for v in s["values"]]
        anomaly = detect_metric_anomaly(values, threshold_sigma=3.0)
        # 关键：错误率除了看突变，还要看绝对值
        # 即使没突变，错误率超 1% 也得告警
        if anomaly["is_anomaly"] or (anomaly.get("current", 0) > 0.01):
            results.append({
                "service": service,
                "anomaly": anomaly,
                "severity": "critical" if anomaly.get("current", 0) > 0.05 else "warning",
            })
    return {"service": service, "anomalies": results}

def check_trace_error_rate(service: str) -> dict:
    """检查 trace 错误率，从 Jaeger 查"""
    # Jaeger 有 HTTP API 可以查 trace 统计
    # 这里简化，实际可以用 Jaeger 的 /api/traces?service=X&tags={"error":true}
    jaeger_url = f"http://jaeger:16686/api/traces"
    params = {
        "service": service,
        "limit": 100,
        "lookback": "1h",
    }
    r = requests.get(jaeger_url, params=params, timeout=10)
    r.raise_for_status()
    data = r.json()

    traces = data.get("data", [])
    if not traces:
        return {"service": service, "status": "no_traces"}

    error_count = 0
    slow_count = 0
    durations = []

    for trace in traces:
        # 关键：trace duration 是所有 span 里最长 end-start
        duration_ms = max(
            (span.get("duration", 0) for span in trace.get("spans", [])),
            default=0,
        )
        durations.append(duration_ms)

        # 错误判断：看是否有 error tag
        for span in trace.get("spans", []):
            for tag in span.get("tags", []):
                if tag.get("key") == "error" and tag.get("value") is True:
                    error_count += 1
                    break

        # 慢 trace 判断
        if duration_ms > 500:
            slow_count += 1

    total = len(traces)
    # 关键：trace 通常采样过，要按采样率反推真实量
    # 假设采样率 10%，错误率 = error_count / total 仍然准确，因为是比例
    return {
        "service": service,
        "total_traces": total,
        "error_rate": error_count / total if total else 0,
        "slow_rate": slow_count / total if total else 0,
        "p99_duration_ms": float(np.percentile(durations, 99)) if durations else 0,
        "is_anomaly": (error_count / total > 0.05) if total else False,
    }

def run_detection(services: list) -> list:
    """对所有服务跑一遍异常检测"""
    alerts = []
    for svc in services:
        # 关键：metric 和 trace 两个维度都看，互相印证
        # 单一维度容易误报，两个维度都异常基本就是真异常
        metric_result = check_service_error_rate(svc)
        trace_result = check_trace_error_rate(svc)

        if metric_result.get("anomalies") or trace_result.get("is_anomaly"):
            alerts.append({
                "service": svc,
                "metric": metric_result,
                "trace": trace_result,
                "timestamp": datetime.now().isoformat(),
            })
    return alerts

if __name__ == "__main__":
    services = ["order-service", "payment-service", "user-service"]
    alerts = run_detection(services)
    for a in alerts:
        print(f"[ALERT] {a['service']}: metric={bool(a['metric'].get('anomalies'))} "
              f"trace={a['trace'].get('is_anomaly')}")