分布式链路追踪实战：Jaeger 与 Tempo 选型对比

helm repo add grafana https://grafana.github.io/helm-charts
helm repo update

# values-tempo.yaml
tempo:
  storage:
    trace:
      backend: s3
      s3:
        bucket: my-tempo-traces
        endpoint: s3.us-west-2.amazonaws.com
        region: us-west-2
        # 使用 IRSA（推荐）或 accessKey/secretKey
        # access_key: ""
        # secret_key: ""

  # 保留 7 天数据
  retention: 168h

  # 接收端口配置
  receivers:
    otlp:
      protocols:
        grpc:
          endpoint: "0.0.0.0:4317"
        http:
          endpoint: "0.0.0.0:4318"
    jaeger:
      protocols:
        grpc:
          endpoint: "0.0.0.0:14250"

  # 启用 TraceQL metrics（实验性）
  metricsGenerator:
    enabled: true
    remoteWriteUrl: "http://prometheus-operated:9090/api/v1/write"
    ingestion:
      tenantId: anonymous
    storage:
      path: /var/tempo/wal
      remote_write_flush_deadline: 1m
    traces_storage:
      path: /var/tempo/traces

  # Ingester 配置
  ingester:
    max_block_duration: 5m
    trace_idle_period: 10s
    flush_check_period: 10s

  # 查询配置
  querier:
    max_concurrent_queries: 20
    search:
      prefer_self: 10

  # 全局采样配置（概率采样）
  distributor:
    receivers:
      otlp:
        protocols:
          grpc: {}
          http: {}

# 持久化 WAL
persistence:
  enabled: true
  size: 10Gi
  storageClassName: gp3

# 资源限制
resources:
  requests:
    cpu: 500m
    memory: 1Gi
  limits:
    cpu: 2
    memory: 4Gi

serviceMonitor:
  enabled: true

kubectl create namespace observability

# 如果用 IRSA，需要给 ServiceAccount 打 annotation
helm install tempo grafana/tempo-distributed \
  -n observability \
  -f values-tempo.yaml

kubectl get pods -n observability | grep tempo
# tempo-compactor-xxx         1/1  Running
# tempo-distributor-xxx       1/1  Running
# tempo-ingester-0            1/1  Running
# tempo-querier-xxx           1/1  Running
# tempo-query-frontend-xxx    1/1  Running

# grafana-datasources.yaml（通过 ConfigMap 注入）
apiVersion: v1
kind: ConfigMap
metadata:
  name: grafana-datasources
  namespace: observability
data:
  datasources.yaml: |
    apiVersion: 1
    datasources:
      - name: Tempo
        type: tempo
        url: http://tempo-query-frontend.observability:3100
        jsonData:
          tracesToLogsV2:
            datasourceUid: loki
            spanStartTimeShift: "-1m"
            spanEndTimeShift: "1m"
            filterByTraceID: true
            filterBySpanID: false
            customQuery: true
            query: '{namespace="${__span.tags.k8s.namespace.name}"} |= "${__trace.traceId}"'
          tracesToMetrics:
            datasourceUid: prometheus
            queries:
              - name: "Request Rate"
                query: 'rate(http_requests_total{service="${__span.tags.service.name}"}[5m])'
          serviceMap:
            datasourceUid: prometheus
          nodeGraph:
            enabled: true
          search:
            hide: false
          lokiSearch:
            datasourceUid: loki

package main

import (
    "context"
    "log"
    "net/http"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.21.0"
    "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
)

func initTracer(ctx context.Context) (*sdktrace.TracerProvider, error) {
    // 创建 OTLP gRPC exporter，发送到 OTel Collector 或直接发 Tempo
    exporter, err := otlptracegrpc.New(ctx,
        otlptracegrpc.WithEndpoint("otel-collector:4317"),
        otlptracegrpc.WithInsecure(),
    )
    if err != nil {
        return nil, err
    }

    // 资源属性：标识这个服务
    res, err := resource.New(ctx,
        resource.WithAttributes(
            semconv.ServiceName("order-service"),
            semconv.ServiceVersion("v1.2.3"),
            semconv.DeploymentEnvironment("production"),
        ),
        resource.WithFromEnv(),   // 支持 OTEL_RESOURCE_ATTRIBUTES 环境变量
        resource.WithProcess(),   // 自动添加进程信息
        resource.WithOS(),
        resource.WithContainer(), // 自动添加容器/Pod 信息
        resource.WithHost(),
    )
    if err != nil {
        return nil, err
    }

    // TracerProvider：采样策略用概率采样 10%，错误全量保留
    tp := sdktrace.NewTracerProvider(
        sdktrace.WithBatcher(exporter),
        sdktrace.WithResource(res),
        sdktrace.WithSampler(
            sdktrace.ParentBased(
                sdktrace.TraceIDRatioBased(0.1), // 10% 采样
            ),
        ),
    )

    // 设置全局 TracerProvider 和 Propagator
    otel.SetTracerProvider(tp)
    otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
        propagation.TraceContext{}, // W3C TraceContext
        propagation.Baggage{},
    ))

    return tp, nil
}

func main() {
    ctx := context.Background()

    tp, err := initTracer(ctx)
    if err != nil {
        log.Fatal(err)
    }
    defer tp.Shutdown(ctx)

    // otelhttp 自动为每个请求创建 Span，传播上下文
    mux := http.NewServeMux()
    mux.HandleFunc("/api/orders", handleOrders)

    handler := otelhttp.NewHandler(mux, "order-service",
        otelhttp.WithMessageEvents(otelhttp.ReadEvents, otelhttp.WriteEvents),
    )

    log.Fatal(http.ListenAndServe(":8080", handler))
}

package service

import (
    "context"
    "fmt"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/codes"
    semconv "go.opentelemetry.io/otel/semconv/v1.21.0"
)

var tracer = otel.Tracer("order-service/service")

func (s *OrderService) CreateOrder(ctx context.Context, userID int64, items []Item) (*Order, error) {
    // 创建子 Span
    ctx, span := tracer.Start(ctx, "OrderService.CreateOrder")
    defer span.End()

    // 添加业务属性
    span.SetAttributes(
        attribute.Int64("user.id", userID),
        attribute.Int("order.items_count", len(items)),
    )

    // 校验库存
    if err := s.checkInventory(ctx, items); err != nil {
        // 记录错误并设置 Span 状态
        span.RecordError(err)
        span.SetStatus(codes.Error, "inventory check failed")
        return nil, fmt.Errorf("inventory check: %w", err)
    }

    // 写数据库（数据库 SDK 通常自动插桩，这里演示手动）
    ctx, dbSpan := tracer.Start(ctx, "db.insert_order")
    dbSpan.SetAttributes(
        semconv.DBSystem("mysql"),
        semconv.DBName("orders"),
        semconv.DBOperation("INSERT"),
        semconv.DBStatement("INSERT INTO orders (user_id, status) VALUES (?, ?)"),
    )

    order, err := s.repo.Insert(ctx, userID, items)
    if err != nil {
        dbSpan.RecordError(err)
        dbSpan.SetStatus(codes.Error, err.Error())
        dbSpan.End()
        span.SetStatus(codes.Error, "db insert failed")
        return nil, err
    }
    dbSpan.SetAttributes(attribute.Int64("order.id", order.ID))
    dbSpan.End()

    // 发送 MQ 消息（添加关键事件）
    span.AddEvent("order_created", trace.WithAttributes(
        attribute.Int64("order.id", order.ID),
    ))

    return order, nil
}

# requirements.txt
# opentelemetry-sdk
# opentelemetry-exporter-otlp-proto-grpc
# opentelemetry-instrumentation-flask
# opentelemetry-instrumentation-sqlalchemy
# opentelemetry-instrumentation-requests

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource, SERVICE_NAME, SERVICE_VERSION
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor
from opentelemetry.propagate import set_global_textmap
from opentelemetry.propagators.composite import CompositePropagator
from opentelemetry.propagators.b3 import B3Format
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator

def init_tracing(app, db_engine):
    resource = Resource.create({
        SERVICE_NAME: "payment-service",
        SERVICE_VERSION: "2.1.0",
        "deployment.environment": "production",
    })

    exporter = OTLPSpanExporter(
        endpoint="http://otel-collector:4317",
        insecure=True,
    )

    provider = TracerProvider(
        resource=resource,
        # 尾采样由 OTel Collector 处理，这里 100% 发送到 Collector
        # sampler=TraceIdRatioBased(0.1),  # 如果在应用层头采样
    )
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)

    set_global_textmap(TraceContextTextMapPropagator())

    # 自动插桩 Flask
    FlaskInstrumentor().instrument_app(app)

    # 自动插桩 SQLAlchemy（自动记录所有 SQL 语句）
    SQLAlchemyInstrumentor().instrument(engine=db_engine)

    # 自动插桩 requests 库（HTTP 出口请求）
    RequestsInstrumentor().instrument()

from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

tracer = trace.get_tracer("payment-service.payment")

def process_payment(user_id: int, amount: float, currency: str) -> dict:
    with tracer.start_as_current_span("PaymentService.process") as span:
        span.set_attribute("user.id", user_id)
        span.set_attribute("payment.amount", amount)
        span.set_attribute("payment.currency", currency)

        try:
            # 调用风控服务
            with tracer.start_as_current_span("risk.check") as risk_span:
                risk_result = risk_client.check(user_id, amount)
                risk_span.set_attribute("risk.score", risk_result.score)
                risk_span.set_attribute("risk.decision", risk_result.decision)

            if risk_result.decision == "REJECT":
                span.set_status(Status(StatusCode.ERROR, "payment rejected by risk"))
                span.set_attribute("payment.status", "rejected")
                return {"status": "rejected", "reason": "risk_control"}

            # 调用支付网关
            with tracer.start_as_current_span("gateway.charge") as gw_span:
                gw_span.set_attribute("gateway.provider", "stripe")
                result = stripe_client.charge(amount, currency)
                gw_span.set_attribute("gateway.transaction_id", result.transaction_id)

            span.add_event("payment_completed", {
                "transaction_id": result.transaction_id,
            })
            return {"status": "success", "transaction_id": result.transaction_id}

        except Exception as e:
            span.record_exception(e)
            span.set_status(Status(StatusCode.ERROR, str(e)))
            raise

# otelcol-config.yaml
processors:
  tail_sampling:
    decision_wait: 30s          # 等待 Trace 完成的最长时间
    num_traces: 100000          # 内存中缓存的最大 Trace 数量
    expected_new_traces_per_sec: 10000
    policies:
      # 策略1：所有错误 Trace 全量保留
      - name: errors-policy
        type: status_code
        status_code:
          status_codes: [ERROR]

      # 策略2：慢请求保留（> 1 秒）
      - name: slow-traces-policy
        type: latency
        latency:
          threshold_ms: 1000

      # 策略3：正常请求 5% 概率采样
      - name: probabilistic-policy
        type: probabilistic
        probabilistic:
          sampling_percentage: 5

      # 策略4：特定服务全量保留（如支付服务）
      - name: payment-service-policy
        type: string_attribute
        string_attribute:
          key: service.name
          values: ["payment-service"]
          enabled_regex_matching: false

    # 组合策略：满足任一条件即保留
    composite:
      max_total_spans_per_second: 50000
      policy_order:
        - errors-policy
        - slow-traces-policy
        - payment-service-policy
        - probabilistic-policy

# tempo values.yaml
tempo:
  metricsGenerator:
    enabled: true
    processor:
      service_graphs:
        enabled: true
        # 只为采样率内的 Trace 生成 service graph
        max_items: 10000
        wait: 10s
      span_metrics:
        enabled: true
        # 生成 traces_spanmetrics_* 指标
        histogram_buckets:
          - 0.002
          - 0.004
          - 0.008
          - 0.016
          - 0.032
          - 0.064
          - 0.128
          - 0.256
          - 0.512
          - 1.024
          - 2.048

import (
    "github.com/prometheus/client_golang/prometheus"
    "go.opentelemetry.io/otel/trace"
)

var httpDuration = prometheus.NewHistogramVec(
    prometheus.HistogramOpts{
        Name:    "http_request_duration_seconds",
        Help:    "HTTP request duration",
        Buckets: prometheus.DefBuckets,
    },
    []string{"method", "path", "status"},
)

func recordWithExemplar(ctx context.Context, method, path, status string, duration float64) {
    spanCtx := trace.SpanFromContext(ctx).SpanContext()
    if spanCtx.IsValid() {
        httpDuration.WithLabelValues(method, path, status).(prometheus.ExemplarObserver).ObserveWithExemplar(
            duration,
            prometheus.Labels{
                "traceID": spanCtx.TraceID().String(),
                "spanID":  spanCtx.SpanID().String(),
            },
        )
    } else {
        httpDuration.WithLabelValues(method, path, status).Observe(duration)
    }
}

# prometheus.yaml
global:
  scrape_interval: 15s

# 启用 Exemplar 存储（需要 Prometheus 2.43+）
storage:
  exemplars:
    max_exemplars: 100000

import (
    "go.opentelemetry.io/otel/trace"
    "go.uber.org/zap"
)

func logWithTrace(ctx context.Context, logger *zap.Logger, msg string, fields ...zap.Field) {
    spanCtx := trace.SpanFromContext(ctx).SpanContext()
    if spanCtx.IsValid() {
        fields = append(fields,
            zap.String("traceID", spanCtx.TraceID().String()),
            zap.String("spanID", spanCtx.SpanID().String()),
        )
    }
    logger.Info(msg, fields...)
}

datasources:
  - name: Loki
    type: loki
    url: http://loki:3100
    jsonData:
      derivedFields:
        - datasourceName: Tempo
          datasourceUid: tempo-uid
          matcherRegex: '"traceID":"(\w+)"'
          name: TraceID
          url: "$${__value.raw}"

# tempo 存储配置
tempo:
  storage:
    trace:
      backend: s3
      block:
        bloom_filter_false_positive: 0.01    # 布隆过滤器假阳率，越低占用空间越大
        bloom_filter_shard_size_bytes: 100000 # 每个 shard 大小
        v2_encoding: zstd                    # 压缩算法，zstd 压缩比高
        v2_index_downsample_bytes: 1000000   # 索引采样间隔
      wal:
        encoding: snappy                     # WAL 压缩
      pool:
        max_workers: 100                     # 并发读取 S3 的 worker 数
        queue_depth: 10000

  compactor:
    compaction:
      block_retention: 720h                  # 数据保留 30 天
      compacted_block_retention: 1h          # 已合并的旧块保留 1 小时
      compaction_window: 1h                  # 合并窗口
      max_block_bytes: 107374182400          # 单块最大 100GB
      max_compaction_objects: 6000000        # 单次合并最大对象数
      retention_concurrency: 10

// 错误：把高基数值放到 Span Name
span.SetName(fmt.Sprintf("GET /api/orders/%d", orderID)) // ❌

// 正确：Span Name 用固定模板，高基数值用 Attribute
span.SetName("GET /api/orders/:id")  // ✓
span.SetAttributes(attribute.Int64("order.id", orderID))  // ✓

// 错误：SQL 语句直接插值（高基数 + 安全风险）
span.SetAttributes(attribute.String("db.statement",
    fmt.Sprintf("SELECT * FROM orders WHERE id = %d", orderID))) // ❌

// 正确：使用参数化 SQL
span.SetAttributes(attribute.String("db.statement",
    "SELECT * FROM orders WHERE id = ?"))  // ✓

processors:
  filter/drop_health_checks:
    error_mode: ignore
    traces:
      span:
        # 丢弃健康检查请求
        - 'attributes["http.url"] == "/health"'
        - 'attributes["http.url"] == "/metrics"'
        - 'attributes["http.url"] == "/readyz"'

# prometheus-rules.yaml
groups:
  - name: tracing.rules
    rules:
      # 告警：服务错误率超过 1%
      - alert: ServiceHighErrorRate
        expr: |
          (
            sum by (service_name) (
              rate(traces_spanmetrics_calls_total{status_code="STATUS_CODE_ERROR"}[5m])
            )
            /
            sum by (service_name) (
              rate(traces_spanmetrics_calls_total[5m])
            )
          ) > 0.01
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "服务 {{ $labels.service_name }} 错误率过高"
          description: "过去 5 分钟错误率为 {{ $value | humanizePercentage }}，超过 1% 阈值"

      # 告警：P99 延迟超过 2 秒
      - alert: ServiceHighLatency
        expr: |
          histogram_quantile(0.99,
            sum by (service_name, le) (
              rate(traces_spanmetrics_latency_bucket[5m])
            )
          ) > 2
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "服务 {{ $labels.service_name }} P99 延迟过高"
          description: "P99 延迟为 {{ $value | humanizeDuration }}，超过 2s 阈值"

      # 告警：Tempo Ingester 写入失败
      - alert: TempoIngesterWriteFailures
        expr: |
          rate(tempo_ingester_traces_created_total{err!=""}[5m]) > 0
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "Tempo Ingester 写入错误"
          description: "Ingester 出现写入失败，可能丢失 Trace 数据"

      # 告警：OTel Collector 丢弃 Span
      - alert: OtelCollectorSpanDropped
        expr: |
          rate(otelcol_processor_dropped_spans_total[5m]) > 100
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "OTel Collector 正在丢弃 Span"
          description: "每秒丢弃 {{ $value }} 个 Span，检查 Collector 容量或下游连接"

exporters:
  loadbalancing:
    protocol:
      otlp:
        tls:
          insecure: true
    resolver:
      dns:
        hostname: otelcol-tail-sampler-headless.observability
        port: 4317

otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
    propagation.TraceContext{},
    propagation.Baggage{},
    jaegerprop.Jaeger{}, // 兼容 Jaeger header
))