アプリケーションにOpenTelemetryを付ける — 自動計測から手動スパンまで、そしてコレクタを置く理由

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はどのスパンにも属さない)

# 最小集合 — この三つがないとデータがどこから来たか分からない
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"}'

# スパンがこのtrace IDの下に付かなければ、下の四つのうちどれかです

# 途切れる — ワーカースレッドにはコンテキストがないので新しいトレースが始まる
from concurrent.futures import ThreadPoolExecutor

pool = ThreadPoolExecutor(max_workers=8)

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

# 生き残る — 現在のコンテキストをキャプチャしてワーカーの中で再度有効化する
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)

# バックグラウンド作業は新しいルートトレースとして始め、原因となったリクエストはリンクとして残す
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 — アプリと同じノードまたはサイドカーに置くエージェント層
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  # 必ず最初。メモリ上限に達すると受信を拒否し、コレクタ自体が死ぬのを防ぐ
  memory_limiter:
    check_interval: 1s
    limit_percentage: 80
    spike_limit_percentage: 20

  # Kubernetesのメタデータをリソース属性として付ける
  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()