给应用接上 OpenTelemetry —— 从自动埋点到手动 span，以及为什么要放一个 collector

pip install \
  'opentelemetry-distro[otlp]' \
  opentelemetry-instrumentation-fastapi \
  opentelemetry-instrumentation-sqlalchemy \
  opentelemetry-instrumentation-requests \
  opentelemetry-instrumentation-redis \
  opentelemetry-instrumentation-logging

# 自动探测已安装的埋点包并接上
opentelemetry-bootstrap --action=install

export OTEL_SERVICE_NAME=checkout-api
export OTEL_RESOURCE_ATTRIBUTES=service.version=2.7.1,deployment.environment.name=prod,service.namespace=commerce
export OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector.observability.svc:4317
export OTEL_EXPORTER_OTLP_PROTOCOL=grpc
export OTEL_TRACES_SAMPLER=parentbased_always_on
export OTEL_PYTHON_LOG_CORRELATION=true

opentelemetry-instrument uvicorn app.main:app --host 0.0.0.0 --port 8000

SERVER    checkout-api   POST /v1/orders                       1421ms
├─ CLIENT    GET http://auth.internal/verify                      31ms
├─ CLIENT    SELECT carts WHERE id = ?                             6ms
├─ CLIENT    redis GET promo:rules:t-8871                          2ms
├─ CLIENT    POST http://payment.internal/charge                  74ms
└─ (剩下的 1308ms 不属于任何一个 span)

# 最小集合 —— 没有这三个，就没法知道数据是从哪来的
OTEL_SERVICE_NAME=checkout-api
OTEL_RESOURCE_ATTRIBUTES=service.version=2.7.1,deployment.environment.name=prod

# 实务中额外有用的属性
OTEL_RESOURCE_ATTRIBUTES=service.version=2.7.1,\
deployment.environment.name=prod,\
service.namespace=commerce,\
service.instance.id=checkout-api-7d9f4b-x2k9m

# deployment.yaml
env:
  - name: OTEL_SERVICE_NAME
    value: checkout-api
  - name: POD_NAME
    valueFrom:
      fieldRef:
        fieldPath: metadata.name
  - name: POD_NAMESPACE
    valueFrom:
      fieldRef:
        fieldPath: metadata.namespace
  - name: OTEL_RESOURCE_ATTRIBUTES
    value: >-
      service.version=2.7.1,
      deployment.environment.name=prod,
      service.namespace=commerce,
      service.instance.id=$(POD_NAME),
      k8s.namespace.name=$(POD_NAMESPACE)

# 冒充网关直接注入请求头，再到后端用这个 trace ID 查一遍
curl -sS -o /dev/null -w '%{http_code}\n' \
  http://checkout-api.internal/v1/orders \
  -H 'traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01' \
  -H 'content-type: application/json' \
  -d '{"cart_id":"c-1"}'

# 如果没有 span 挂在这个 trace ID 下面，就是下面四种情况之一

# 会断 —— worker 线程里没有上下文，所以会开启一条新的 trace
from concurrent.futures import ThreadPoolExecutor

pool = ThreadPoolExecutor(max_workers=8)

def enrich_all(items):
    return list(pool.map(fetch_details, items))

# 能存活 —— 先捕获当前上下文，在 worker 内部重新激活它
from concurrent.futures import ThreadPoolExecutor
from opentelemetry import context as otel_context

pool = ThreadPoolExecutor(max_workers=8)

def _with_context(ctx, fn, *args):
    token = otel_context.attach(ctx)
    try:
        return fn(*args)
    finally:
        otel_context.detach(token)

def enrich_all(items):
    ctx = otel_context.get_current()
    futures = [pool.submit(_with_context, ctx, fetch_details, it) for it in items]
    return [f.result() for f in futures]

from opentelemetry import propagate, trace
from opentelemetry.trace import SpanKind

tracer = trace.get_tracer("checkout", "2.7.1")

def publish_order(producer, order):
    with tracer.start_as_current_span(
        "orders publish", kind=SpanKind.PRODUCER
    ) as span:
        span.set_attribute("messaging.system", "kafka")
        span.set_attribute("messaging.destination.name", "orders")
        headers = {}
        propagate.inject(headers)          # 把 traceparent 放进 dict
        producer.send(
            "orders",
            value=order.to_bytes(),
            headers=[(k, v.encode()) for k, v in headers.items()],
        )

from opentelemetry import propagate, trace
from opentelemetry.trace import SpanKind, Link

def consume_batch(messages):
    links = []
    for m in messages:
        headers = {k: v.decode() for k, v in (m.headers or [])}
        ctx = propagate.extract(headers)
        sc = trace.get_current_span(ctx).get_span_context()
        if sc.is_valid:
            links.append(Link(sc))

    with tracer.start_as_current_span(
        "orders process", kind=SpanKind.CONSUMER, links=links
    ) as span:
        span.set_attribute("messaging.batch.message_count", len(messages))
        for m in messages:
            handle(m)

# 后台任务以新的根 trace 开始，导致它发生的那个请求只留一个 link
def schedule_reindex(cart_id):
    origin = trace.get_current_span().get_span_context()

    def run():
        links = [Link(origin)] if origin.is_valid else []
        with tracer.start_as_current_span(
            "cart.reindex", kind=SpanKind.INTERNAL, links=links
        ) as span:
            span.set_attribute("cart.id", cart_id)
            reindex(cart_id)

    background.add_task(run)

# 确认实际到达的请求头最快的办法
kubectl -n commerce exec deploy/checkout-api -- \
  sh -c 'timeout 20 tcpdump -A -s0 -i any "tcp port 8000" 2>/dev/null | grep -i traceparent'

# 或者给应用加一个临时端点，把收到的请求头原样返回
curl -s http://checkout-api.internal/__debug/headers \
  -H 'traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01' | jq .

OTEL_PROPAGATORS=tracecontext,baggage,b3multi

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

tracer = trace.get_tracer("checkout", "2.7.1")

async def apply_promotions(cart, tenant_id):
    with tracer.start_as_current_span("checkout.apply_promotions") as span:
        span.set_attribute("cart.item_count", len(cart.items))
        span.set_attribute("tenant.id", tenant_id)
        span.set_attribute("promotion.engine", "rules-v3")
        try:
            with tracer.start_as_current_span("promotion.load_rules") as load:
                cached = await rules.from_cache(tenant_id)
                load.set_attribute("cache.hit", cached is not None)
                ruleset = cached or await rules.compile(tenant_id)
                load.set_attribute("promotion.rule_count", len(ruleset))

            with tracer.start_as_current_span("promotion.evaluate") as ev:
                result = ruleset.evaluate(cart)
                ev.set_attribute("promotion.evaluated", result.evaluated)
                ev.set_attribute("promotion.matched", len(result.matched))
            return result
        except Exception as exc:
            span.record_exception(exc)
            span.set_status(Status(StatusCode.ERROR, str(exc)))
            raise

SERVER    checkout-api   POST /v1/orders                       1421ms
├─ CLIENT    GET http://auth.internal/verify                      31ms
├─ CLIENT    SELECT carts WHERE id = ?                             6ms
├─ INTERNAL  checkout.apply_promotions                          1298ms
│  ├─ INTERNAL  promotion.load_rules   cache.hit=false           1241ms  <-- 就是这里
│  └─ INTERNAL  promotion.evaluate     evaluated=812               54ms
├─ CLIENT    POST http://payment.internal/charge                  74ms
└─ (self time 12ms)

# otel-collector.yaml —— 部署在和应用同一个节点上、或作为 sidecar 的 agent 层
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  # 必须放第一位。一旦触及内存上限就拒绝接收，防止 collector 自己被拖死
  memory_limiter:
    check_interval: 1s
    limit_percentage: 80
    spike_limit_percentage: 20

  # 把 Kubernetes 元数据附加成 resource 属性
  k8sattributes:
    auth_type: serviceAccount
    extract:
      metadata:
        - k8s.namespace.name
        - k8s.deployment.name
        - k8s.pod.name
        - k8s.node.name

  # 清除敏感属性 —— 不用改应用，在这里挡住
  attributes/redact:
    actions:
      - key: http.request.header.authorization
        action: delete
      - key: user.email
        action: delete
      - key: db.query.text
        action: hash

  # 永远放最后。减少网络往返
  batch:
    timeout: 5s
    send_batch_size: 8192
    send_batch_max_size: 16384

exporters:
  otlp/gateway:
    endpoint: otel-gateway.observability.svc:4317
    tls:
      insecure: true
    sending_queue:
      enabled: true
      num_consumers: 10
      queue_size: 5000
    retry_on_failure:
      enabled: true
      initial_interval: 5s
      max_elapsed_time: 300s

service:
  telemetry:
    metrics:
      level: detailed
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, k8sattributes, attributes/redact, batch]
      exporters: [otlp/gateway]
    metrics:
      receivers: [otlp]
      processors: [memory_limiter, k8sattributes, batch]
      exporters: [otlp/gateway]
    logs:
      receivers: [otlp]
      processors: [memory_limiter, k8sattributes, attributes/redact, batch]
      exporters: [otlp/gateway]

OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT=64
OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT=2048
OTEL_BSP_MAX_QUEUE_SIZE=4096
OTEL_BSP_MAX_EXPORT_BATCH_SIZE=512
OTEL_BSP_SCHEDULE_DELAY=2000

from opentelemetry import trace

def main():
    run_job()
    # 批处理作业必须显式清空队列
    trace.get_tracer_provider().force_flush(timeout_millis=10_000)
    trace.get_tracer_provider().shutdown()