필사 모드: 애플리케이션에 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

  # 쿠버네티스 메타데이터를 리소스 속성으로 붙인다
  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()