3장: 분산 추적(Distributed Tracing)

from opentelemetry import trace
 
tracer = trace.get_tracer("order-service", "1.0.0")
 
with tracer.start_as_current_span("process-order") as span:
    # 속성 추가
    span.set_attribute("order.id", "ORD-2026-001")
    span.set_attribute("order.total", 45000)
    span.set_attribute("customer.tier", "premium")
    
    # 이벤트 추가 (특정 시점의 기록)
    span.add_event("payment-validated", {
        "payment.method": "credit-card",
        "payment.amount": 45000,
    })
    
    try:
        process_payment()
    except Exception as e:
        # 에러 상태 기록
        span.set_status(trace.StatusCode.ERROR, str(e))
        span.record_exception(e)
        raise

from opentelemetry import trace
 
# 현재 활성 스팬의 컨텍스트 가져오기
current_span = trace.get_current_span()
ctx = current_span.get_span_context()
 
print(f"Trace ID: {format(ctx.trace_id, '032x')}")
print(f"Span ID: {format(ctx.span_id, '016x')}")
print(f"Trace Flags: {ctx.trace_flags}")
print(f"Is Remote: {ctx.is_remote}")

from opentelemetry import trace
 
tracer = trace.get_tracer("consumer-service")
 
# 메시지 큐에서 받은 여러 메시지의 원본 스팬과 링크 생성
links = [
    trace.Link(
        context=producer_span_context_1,
        attributes={"messaging.message.id": "msg-001"}
    ),
    trace.Link(
        context=producer_span_context_2,
        attributes={"messaging.message.id": "msg-002"}
    ),
]
 
# 배치 처리 스팬에 링크 추가
with tracer.start_as_current_span("process-batch", links=links) as span:
    span.set_attribute("messaging.batch.message_count", 2)
    process_messages()

from opentelemetry.trace import SpanKind
 
# SERVER 스팬 -- 들어오는 HTTP 요청 처리
with tracer.start_as_current_span("GET /api/orders", kind=SpanKind.SERVER):
    
    # INTERNAL 스팬 -- 비즈니스 로직
    with tracer.start_as_current_span("validate-order", kind=SpanKind.INTERNAL):
        validate()
    
    # CLIENT 스팬 -- 외부 서비스 호출
    with tracer.start_as_current_span("payment-api-call", kind=SpanKind.CLIENT):
        call_payment_service()
    
    # PRODUCER 스팬 -- 메시지 발행
    with tracer.start_as_current_span("send-notification", kind=SpanKind.PRODUCER):
        publish_to_queue()

from opentelemetry.sdk.trace.sampling import (
    TraceIdRatioBased,
    ParentBased,
)
 
# 10% 확률로 샘플링 (새로운 트레이스에만 적용)
sampler = ParentBased(
    root=TraceIdRatioBased(0.1),  # 루트 스팬: 10% 샘플링
    # 부모가 샘플링된 경우 자식도 항상 샘플링
)

processors:
  tail_sampling:
    decision_wait: 10s
    num_traces: 100000
    policies:
      # 에러가 있는 트레이스는 100% 보존
      - name: errors-policy
        type: status_code
        status_code:
          status_codes: [ERROR]
      
      # 지연 시간이 500ms 이상인 트레이스 보존
      - name: latency-policy
        type: latency
        latency:
          threshold_ms: 500
      
      # 나머지는 5% 확률 샘플링
      - name: probabilistic-policy
        type: probabilistic
        probabilistic:
          sampling_percentage: 5

pip install opentelemetry-api \
    opentelemetry-sdk \
    opentelemetry-exporter-otlp-proto-grpc \
    opentelemetry-instrumentation-flask \
    opentelemetry-instrumentation-requests \
    flask requests

from flask import Flask, jsonify
import 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
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor
 
# SDK 초기화
resource = Resource.create({
    "service.name": "order-service",
    "service.version": "1.0.0",
    "deployment.environment": "development",
})
 
provider = TracerProvider(resource=resource)
provider.add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4317"))
)
trace.set_tracer_provider(provider)
 
# 자동 계측 활성화
app = Flask(__name__)
FlaskInstrumentor().instrument_app(app)
RequestsInstrumentor().instrument()
 
# 수동 계측을 위한 트레이서
tracer = trace.get_tracer("order-service.handlers", "1.0.0")
 
 
@app.route("/orders/<order_id>")
def get_order(order_id):
    # 자동 계측: Flask가 SERVER 스팬을 생성
    
    # 수동 계측: 비즈니스 로직에 커스텀 스팬 추가
    with tracer.start_as_current_span("validate-order") as span:
        span.set_attribute("order.id", order_id)
        validate_order(order_id)
    
    with tracer.start_as_current_span("fetch-order-details") as span:
        span.set_attribute("order.id", order_id)
        # 자동 계측: requests가 CLIENT 스팬을 생성
        details = requests.get(
            f"http://inventory-service:8081/items?order={order_id}"
        )
        span.set_attribute("inventory.item_count", len(details.json()))
    
    return jsonify({"order_id": order_id, "status": "confirmed"})
 
 
def validate_order(order_id):
    with tracer.start_as_current_span("check-inventory") as span:
        span.set_attribute("order.id", order_id)
        # 재고 확인 로직
        span.add_event("inventory-checked", {"available": True})
 
 
if __name__ == "__main__":
    app.run(port=8080)

services:
  order-service:
    build: .
    ports:
      - "8080:8080"
    environment:
      - OTEL_SERVICE_NAME=order-service
      - OTEL_EXPORTER_OTLP_ENDPOINT=http://collector:4317
    depends_on:
      - collector
 
  collector:
    image: otel/opentelemetry-collector-contrib:0.100.0
    volumes:
      - ./otel-collector-config.yaml:/etc/otelcol/config.yaml
    ports:
      - "4317:4317"   # OTLP gRPC
      - "4318:4318"   # OTLP HTTP
    depends_on:
      - jaeger
 
  jaeger:
    image: jaegertracing/all-in-one:1.55
    ports:
      - "16686:16686"  # Jaeger UI
      - "4317"         # OTLP gRPC (Jaeger 네이티브)

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: "0.0.0.0:4317"
 
processors:
  batch:
    timeout: 1s
    send_batch_size: 1024
 
exporters:
  otlp/jaeger:
    endpoint: "jaeger:4317"
    tls:
      insecure: true
 
service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [batch]
      exporters: [otlp/jaeger]

# 나쁜 예 -- 카디널리티가 높아 집계 불가
tracer.start_as_current_span(f"GET /users/{user_id}")  # user_id마다 다른 스팬 이름
 
# 좋은 예 -- 패턴화된 이름 + 속성으로 상세 정보 분리
with tracer.start_as_current_span("GET /users/:id") as span:
    span.set_attribute("user.id", user_id)

with tracer.start_as_current_span("risky-operation") as span:
    try:
        result = perform_operation()
        span.set_status(trace.StatusCode.OK)
    except ValueError as e:
        # 비즈니스 에러 -- 경고 수준
        span.set_status(trace.StatusCode.ERROR, f"Validation failed: {e}")
        span.record_exception(e)
        return error_response(400, str(e))
    except Exception as e:
        # 시스템 에러 -- 심각
        span.set_status(trace.StatusCode.ERROR, f"Internal error: {e}")
        span.record_exception(e, attributes={
            "exception.escaped": True,
        })
        raise