OpenTelemetry Observability Blueprint: Integrating Metrics, Logs, and Traces

                    [Application Pods]
                    ┌──────────────────┐
                    │  OTel SDK        │
                    │  (Auto + Manual) │
                    │  ┌─────────────┐ │
                    │  │ Traces      │ │
                    │  │ Metrics     │ │
                    │  │ Logs        │ │
                    │  └──────┬──────┘ │
                    └─────────┼────────┘
                              │ OTLP (gRPC)
                              ▼
                    ┌──────────────────┐
                    │  OTel Collector  │
                    │  (DaemonSet)     │
                    │  ┌────────────┐  │
                    │  │ Receivers  │  │
                    │  │ Processors │  │
                    │  │ Exporters  │  │
                    │  └────────────┘  │
                    └────────┬─────────┘
                             │ OTLP
                             ▼
                    ┌──────────────────┐
                    │  OTel Collector  │
                    │  (Gateway)       │
                    │  - Sampling      │
                    │  - Enrichment    │
                    │  - Routing       │
                    └──┬─────┬──────┬──┘
                       │     │      │
                ┌──────┘     │      └──────┐
                ▼            ▼             ▼
          ┌──────────┐ ┌──────────┐ ┌──────────┐
          │ Tempo    │ │ Mimir    │ │ Loki     │
          │ (Traces) │ │ (Metrics)│ │ (Logs)   │
          └──────────┘ └──────────┘ └──────────┘
                └──────────┼──────────┘
                           ▼
                    ┌──────────────┐
                    │   Grafana    │
                    │  (Unified)   │
                    └──────────────┘

# otel-collector-agent.yaml
# DaemonSet으로 배포. 각 노드에서 로컬 수집 담당.
apiVersion: v1
kind: ConfigMap
metadata:
  name: otel-collector-agent
data:
  config.yaml: |
    receivers:
      otlp:
        protocols:
          grpc:
            endpoint: 0.0.0.0:4317
          http:
            endpoint: 0.0.0.0:4318

      # Kubernetes 노드/파드 메트릭 수집
      kubeletstats:
        collection_interval: 30s
        auth_type: serviceAccount
        endpoint: "https://${env:NODE_IP}:10250"
        insecure_skip_verify: true
        metric_groups:
          - node
          - pod
          - container

      # 호스트 메트릭 (CPU, 메모리, 디스크)
      hostmetrics:
        collection_interval: 30s
        scrapers:
          cpu:
            metrics:
              system.cpu.utilization:
                enabled: true
          memory:
            metrics:
              system.memory.utilization:
                enabled: true
          disk: {}
          network: {}

    processors:
      # 메모리 사용량 제한 (Agent는 가볍게)
      memory_limiter:
        check_interval: 5s
        limit_mib: 512
        spike_limit_mib: 128

      # Kubernetes 메타데이터 자동 추가
      k8sattributes:
        auth_type: serviceAccount
        extract:
          metadata:
            - k8s.pod.name
            - k8s.pod.uid
            - k8s.namespace.name
            - k8s.node.name
            - k8s.deployment.name
          labels:
            - tag_name: app
              key: app.kubernetes.io/name
            - tag_name: version
              key: app.kubernetes.io/version

      # 리소스 속성 추가 (모든 신호에 공통 적용)
      resource:
        attributes:
          - key: deployment.environment
            value: "${env:DEPLOY_ENV}"
            action: upsert
          - key: cloud.region
            value: "${env:CLOUD_REGION}"
            action: upsert

      batch:
        send_batch_size: 1024
        timeout: 5s

    exporters:
      otlp:
        endpoint: otel-collector-gateway:4317
        tls:
          insecure: false
          ca_file: /etc/ssl/certs/ca.crt

    service:
      pipelines:
        traces:
          receivers: [otlp]
          processors: [memory_limiter, k8sattributes, resource, batch]
          exporters: [otlp]
        metrics:
          receivers: [otlp, kubeletstats, hostmetrics]
          processors: [memory_limiter, k8sattributes, resource, batch]
          exporters: [otlp]
        logs:
          receivers: [otlp]
          processors: [memory_limiter, k8sattributes, resource, batch]
          exporters: [otlp]

# otel-collector-gateway.yaml
# Deployment로 배포. 중앙 처리 및 라우팅 담당.
apiVersion: v1
kind: ConfigMap
metadata:
  name: otel-collector-gateway
data:
  config.yaml: |
    receivers:
      otlp:
        protocols:
          grpc:
            endpoint: 0.0.0.0:4317

    processors:
      memory_limiter:
        check_interval: 5s
        limit_mib: 4096
        spike_limit_mib: 1024

      # Tail-based sampling (Gateway에서만 수행)
      # 모든 trace를 저장하면 비용이 폭발하므로 지능적 샘플링 필수
      tail_sampling:
        decision_wait: 10s
        num_traces: 100000
        policies:
          # 에러가 포함된 trace는 100% 보존
          - name: errors-policy
            type: status_code
            status_code:
              status_codes: [ERROR]
          # 느린 요청은 100% 보존 (P95 이상)
          - name: latency-policy
            type: latency
            latency:
              threshold_ms: 1000
          # 정상 요청은 10% 샘플링
          - name: probabilistic-policy
            type: probabilistic
            probabilistic:
              sampling_percentage: 10

      # 불필요한 속성 제거 (비용 절감)
      attributes/remove:
        actions:
          - key: http.request.header.authorization
            action: delete
          - key: http.request.header.cookie
            action: delete
          - key: db.statement
            action: hash  # SQL문은 해싱하여 개인정보 보호

      batch:
        send_batch_size: 2048
        timeout: 10s

    exporters:
      # Traces -> Grafana Tempo
      otlp/tempo:
        endpoint: tempo:4317
        tls:
          insecure: true

      # Metrics -> Grafana Mimir (Prometheus 호환)
      prometheusremotewrite:
        endpoint: http://mimir:9009/api/v1/push
        resource_to_telemetry_conversion:
          enabled: true

      # Logs -> Grafana Loki
      loki:
        endpoint: http://loki:3100/loki/api/v1/push

    service:
      pipelines:
        traces:
          receivers: [otlp]
          processors: [memory_limiter, tail_sampling, attributes/remove, batch]
          exporters: [otlp/tempo]
        metrics:
          receivers: [otlp]
          processors: [memory_limiter, batch]
          exporters: [prometheusremotewrite]
        logs:
          receivers: [otlp]
          processors: [memory_limiter, attributes/remove, batch]
          exporters: [loki]

# app/tracing.py
from opentelemetry import trace, metrics
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.semconv.resource import ResourceAttributes
import logging

def setup_observability(
    service_name: str,
    service_version: str,
    otlp_endpoint: str = "http://otel-collector:4317",
):
    """OpenTelemetry 초기화. 애플리케이션 시작 시 한 번 호출."""

    # 리소스 정의 (모든 신호에 공통 적용되는 속성)
    resource = Resource.create({
        ResourceAttributes.SERVICE_NAME: service_name,
        ResourceAttributes.SERVICE_VERSION: service_version,
        ResourceAttributes.DEPLOYMENT_ENVIRONMENT: "production",
        "team.name": "platform",
    })

    # --- Traces ---
    tracer_provider = TracerProvider(resource=resource)
    tracer_provider.add_span_processor(
        BatchSpanProcessor(
            OTLPSpanExporter(endpoint=otlp_endpoint, insecure=True),
            max_queue_size=2048,
            max_export_batch_size=512,
            schedule_delay_millis=5000,
        )
    )
    trace.set_tracer_provider(tracer_provider)

    # --- Metrics ---
    metric_reader = PeriodicExportingMetricReader(
        OTLPMetricExporter(endpoint=otlp_endpoint, insecure=True),
        export_interval_millis=30000,  # 30초마다 export
    )
    meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader])
    metrics.set_meter_provider(meter_provider)

    # --- Logs (OTel Logs Bridge) ---
    # Python logging을 OTel로 브릿지
    from opentelemetry.sdk._logs import LoggerProvider
    from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
    from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter
    from opentelemetry._logs import set_logger_provider

    logger_provider = LoggerProvider(resource=resource)
    logger_provider.add_log_record_processor(
        BatchLogRecordProcessor(
            OTLPLogExporter(endpoint=otlp_endpoint, insecure=True)
        )
    )
    set_logger_provider(logger_provider)

    # Python logging handler 연결
    from opentelemetry.sdk._logs import LoggingHandler
    handler = LoggingHandler(logger_provider=logger_provider)
    logging.getLogger().addHandler(handler)

# app/services/order_service.py
from opentelemetry import trace, metrics
import logging

tracer = trace.get_tracer("order-service", "1.0.0")
meter = metrics.get_meter("order-service", "1.0.0")
logger = logging.getLogger(__name__)

# 비즈니스 메트릭 정의
order_counter = meter.create_counter(
    name="orders.created",
    description="생성된 주문 수",
    unit="1",
)
order_amount_histogram = meter.create_histogram(
    name="orders.amount",
    description="주문 금액 분포",
    unit="KRW",
)
payment_duration = meter.create_histogram(
    name="payment.duration",
    description="결제 처리 시간",
    unit="s",
)

async def create_order(user_id: str, items: list, payment_method: str):
    """주문 생성 - Traces, Metrics, Logs가 모두 연결된 예시"""

    # 1. 상위 span 생성
    with tracer.start_as_current_span(
        "create_order",
        attributes={
            "user.id": user_id,
            "order.item_count": len(items),
            "payment.method": payment_method,
        },
    ) as span:
        total_amount = sum(item["price"] * item["qty"] for item in items)
        span.set_attribute("order.total_amount", total_amount)

        # 2. 재고 확인 (자식 span)
        with tracer.start_as_current_span("check_inventory") as inv_span:
            for item in items:
                available = await check_stock(item["sku"], item["qty"])
                if not available:
                    inv_span.set_attribute("inventory.out_of_stock_sku", item["sku"])
                    # 로그에도 trace context가 자동 포함됨
                    logger.warning(
                        f"재고 부족: SKU={item['sku']}, 요청={item['qty']}",
                        extra={"sku": item["sku"], "requested_qty": item["qty"]},
                    )
                    span.set_status(trace.StatusCode.ERROR, "재고 부족")
                    raise OutOfStockError(item["sku"])

        # 3. 결제 처리 (자식 span)
        import time
        payment_start = time.monotonic()
        with tracer.start_as_current_span(
            "process_payment",
            attributes={"payment.method": payment_method},
        ) as pay_span:
            try:
                result = await payment_gateway.charge(total_amount, payment_method)
                pay_span.set_attribute("payment.transaction_id", result.tx_id)
                logger.info(
                    f"결제 성공: tx_id={result.tx_id}, amount={total_amount}",
                    extra={"tx_id": result.tx_id, "amount": total_amount},
                )
            except PaymentError as e:
                pay_span.set_status(trace.StatusCode.ERROR, str(e))
                logger.error(f"결제 실패: {e}", exc_info=True)
                raise
            finally:
                elapsed = time.monotonic() - payment_start
                payment_duration.record(elapsed, {"payment.method": payment_method})

        # 4. 메트릭 기록
        order_counter.add(1, {
            "payment.method": payment_method,
            "order.status": "created",
        })
        order_amount_histogram.record(total_amount, {
            "payment.method": payment_method,
        })

        return {"order_id": "ORD-12345", "status": "created"}

# grafana/provisioning/datasources/datasources.yaml
apiVersion: 1
datasources:
  - name: Tempo
    type: tempo
    url: http://tempo:3200
    jsonData:
      tracesToLogsV2:
        datasourceUid: loki
        filterByTraceID: true
        filterBySpanID: true
        tags:
          - key: service.name
            value: service_name
      tracesToMetrics:
        datasourceUid: mimir
        tags:
          - key: service.name
            value: service
      serviceMap:
        datasourceUid: mimir

  - name: Loki
    type: loki
    url: http://loki:3100
    jsonData:
      derivedFields:
        - name: TraceID
          datasourceUid: tempo
          matcherRegex: "trace_id=(\\w+)"
          url: '$${__value.raw}'
          matcherType: regex

  - name: Mimir
    type: prometheus
    url: http://mimir:9009/prometheus
    jsonData:
      exemplarTraceIdDestinations:
        - name: trace_id
          datasourceUid: tempo

1. Grafana 대시보드에서 에러율 스파이크 발견
   -> Mimir 쿼리: rate(http_requests_total{status="500"}[5m])

2. 해당 시간대의 에러 요청에 포함된 exemplar 클릭
   -> Exemplar에 포함된 trace_id로 Tempo 이동

3. Tempo에서 전체 trace 확인
   -> 어떤 서비스의 어떤 span에서 에러 발생했는지 시각화
   -> span의 attributes에서 user.id, order.id 등 비즈니스 컨텍스트 확인

4. 해당 span과 같은 trace_id를 가진 로그 확인
   -> Loki 쿼리: {service_name="order-service"} | trace_id="abc123..."
   -> 에러 스택트레이스, 입력 데이터, 중간 상태값 확인

5. 전체 진단 시간: 2-3분 (기존: 15-30분)

# Collector에서 높은 카디널리티 속성 제거
processors:
  # user_id, session_id 같은 high-cardinality label 제거
  # 메트릭에는 사용하지 않아야 함 (trace에서 사용)
  transform/metrics:
    metric_statements:
      - context: datapoint
        statements:
          - delete_key(attributes, "user.id")
          - delete_key(attributes, "session.id")
          - delete_key(attributes, "request.id")
          - delete_key(attributes, "http.url") # path parameter 포함 시 카디널리티 폭발

def estimate_monthly_cost(
    daily_requests: int,
    avg_spans_per_trace: int = 8,
    avg_log_lines_per_request: int = 5,
    trace_sample_rate: float = 0.10,
    log_sample_rate: float = 0.20,
) -> dict:
    """월간 관측성 데이터 비용 추정"""
    monthly_requests = daily_requests * 30

    # Traces
    traces_per_month = monthly_requests * trace_sample_rate
    spans_per_month = traces_per_month * avg_spans_per_trace
    trace_storage_gb = spans_per_month * 0.5 / 1e6  # span당 약 0.5KB

    # Metrics (항상 100%, 카디널리티로 관리)
    # 서비스 10개, 메트릭 50종, 카디널리티 100 = 50,000 시계열
    metric_series = 50_000
    metric_storage_gb = metric_series * 30 * 24 * 2 * 8 / 1e9  # 데이터포인트당 8B

    # Logs
    log_lines_per_month = monthly_requests * avg_log_lines_per_request * log_sample_rate
    log_storage_gb = log_lines_per_month * 0.2 / 1e6  # 로그라인당 약 0.2KB

    # 비용 계산 (Grafana Cloud 기준 대략적 단가)
    trace_cost = trace_storage_gb * 2.0   # $2/GB
    metric_cost = metric_series * 0.008   # $8 per 1000 series
    log_cost = log_storage_gb * 0.50      # $0.50/GB

    return {
        "traces": {
            "sampled_per_month": int(traces_per_month),
            "storage_gb": round(trace_storage_gb, 1),
            "cost_usd": round(trace_cost, 2),
        },
        "metrics": {
            "active_series": metric_series,
            "storage_gb": round(metric_storage_gb, 1),
            "cost_usd": round(metric_cost, 2),
        },
        "logs": {
            "sampled_lines_per_month": int(log_lines_per_month),
            "storage_gb": round(log_storage_gb, 1),
            "cost_usd": round(log_cost, 2),
        },
        "total_monthly_usd": round(trace_cost + metric_cost + log_cost, 2),
    }

# 일 100만 요청 서비스 기준
cost = estimate_monthly_cost(daily_requests=1_000_000)
print(f"월간 총 비용 추정: ${cost['total_monthly_usd']}")

목표: 핵심 서비스 3개에 distributed tracing 활성화

작업:
1. OTel Collector DaemonSet + Gateway 배포
2. 핵심 서비스에 OTel SDK 추가 (auto-instrumentation 우선)
3. Tempo 백엔드 배포
4. Grafana에서 trace 검색 확인

성공 기준:
- 서비스 간 trace가 연결되어 보이는가
- P95 latency가 높은 요청의 병목 구간을 trace에서 식별할 수 있는가

목표: Prometheus 메트릭을 OTel 파이프라인으로 통합

작업:
1. OTel Collector에 Prometheus receiver 추가
2. 기존 Prometheus 서버에서 remote_write로 Mimir 전환
3. Exemplar 연결 설정 (메트릭 -> 트레이스)
4. 기존 Grafana 대시보드 마이그레이션

성공 기준:
- 메트릭 대시보드에서 exemplar 클릭으로 trace 이동 가능한가
- 기존 알림이 동일하게 동작하는가

목표: 구조화 로그를 OTel로 수집하고 trace와 연결

작업:
1. 애플리케이션 로깅에 OTel Logs Bridge 적용
2. 로그에 trace_id, span_id 자동 삽입 확인
3. Loki 백엔드 배포 및 Collector 연결
4. Grafana에서 trace <-> log 양방향 연결 설정

성공 기준:
- trace에서 로그로, 로그에서 trace로 원클릭 이동 가능한가
- 에러 로그의 trace context로 전체 요청 흐름을 추적할 수 있는가

목표: 비용 최적화, 샘플링 튜닝, 팀 온보딩

작업:
1. Tail-based sampling 활성화 및 비용 30% 절감 확인
2. Semantic conventions 표준 문서화
3. 팀별 대시보드 템플릿 배포
4. 온콜 런북에 관측성 워크플로 추가

성공 기준:
- MTTD(Mean Time To Detect) 50% 개선
- MTTR(Mean Time To Resolve) 30% 개선
- 관측성 데이터 비용이 인프라 비용의 5% 이내

# 1. Context propagation 확인 - HTTP 헤더에 traceparent가 전달되는지
curl -v http://service-a/api/test 2>&1 | grep -i traceparent
# traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01

# 2. 서비스 B가 해당 헤더를 수신하고 전파하는지 확인
# 서비스 B의 로그에서 trace_id 검색

# 3. Collector에서 span이 drop되지 않는지 확인
curl -s http://otel-collector:8888/metrics | grep otelcol_exporter_sent_spans

Error: memory usage exceeded limit: 512 MiB

# memory_limiter 설정 확인 및 조정
processors:
  memory_limiter:
    check_interval: 5s
    limit_mib: 512 # Agent는 512MB 이내
    spike_limit_mib: 128
    # limit 도달 시 데이터를 drop하므로
    # 충분한 메모리를 할당하되, 노드에 영향 주지 않을 범위로

  # batch 크기 줄이기
  batch:
    send_batch_size: 512 # 2048에서 512로 줄임
    timeout: 5s

# 카디널리티가 높은 메트릭 찾기
topk(10, count by (__name__)({__name__=~".+"}))

# 특정 메트릭의 label 카디널리티 확인
count(http_requests_total) by (url)
# url에 path parameter가 포함되어 수만 개의 시계열이 생성됨

import logging
from opentelemetry import trace

class TraceContextFilter(logging.Filter):
    def filter(self, record):
        span = trace.get_current_span()
        ctx = span.get_span_context()
        record.trace_id = format(ctx.trace_id, '032x') if ctx.trace_id else ""
        record.span_id = format(ctx.span_id, '016x') if ctx.span_id else ""
        return True

handler = logging.StreamHandler()
handler.addFilter(TraceContextFilter())
handler.setFormatter(logging.Formatter(
    '%(asctime)s %(levelname)s [trace_id=%(trace_id)s span_id=%(span_id)s] %(message)s'
))
logging.getLogger().addHandler(handler)