Python 계측

Python 애플리케이션에 OpenTelemetry를 적용하는 방법을 안내합니다.

개요

Python은 OpenTelemetry에서 잘 지원되는 언어 중 하나입니다.

계측 옵션

자동 계측 (권장)

OTel Operator 사용

가장 쉬운 방법은 OTel Operator를 통한 자동 주입입니다.

Deployment 설정

apiVersion: apps/v1
kind: Deployment
metadata:
  name: my-python-app
  namespace: production
spec:
  template:
    metadata:
      annotations:
        instrumentation.opentelemetry.io/inject-python: "skuber-observability/otel-instrumentation"
    spec:
      containers:
        - name: app
          image: my-python-app:1.0.0
          ports:
            - containerPort: 8000

지원 프레임워크

자동으로 계측되는 프레임워크:

수동 Agent 설정

Kubernetes 외부 환경에서:

# 패키지 설치
pip install opentelemetry-distro opentelemetry-exporter-otlp

# 자동 계측 패키지 설치
opentelemetry-bootstrap -a install

# 애플리케이션 실행 (자동 계측 활성화)
OTEL_SERVICE_NAME=my-python-app \
OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4317 \
opentelemetry-instrument python app.py

Dockerfile 예시

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# OTel 패키지 설치
RUN pip install opentelemetry-distro opentelemetry-exporter-otlp && \
    opentelemetry-bootstrap -a install

COPY . .

ENV OTEL_SERVICE_NAME=my-python-app
ENV OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4317

# 자동 계측으로 실행
CMD ["opentelemetry-instrument", "python", "app.py"]

수동 계측

비즈니스 로직에 커스텀 스팬을 추가하려면 수동 계측을 사용합니다.

의존성 설치

pip install opentelemetry-api \
            opentelemetry-sdk \
            opentelemetry-exporter-otlp

SDK 초기화

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, SERVICE_NAME

def init_tracer():
    """OpenTelemetry 트레이서 초기화"""

    # 리소스 정의
    resource = Resource.create({
        SERVICE_NAME: "order-service",
        "service.version": "1.0.0",
    })

    # TracerProvider 생성
    provider = TracerProvider(resource=resource)

    # OTLP Exporter 설정
    exporter = OTLPSpanExporter(
        endpoint="http://otel-collector:4317",
        insecure=True
    )

    # BatchSpanProcessor 추가
    provider.add_span_processor(BatchSpanProcessor(exporter))

    # 전역 TracerProvider 설정
    trace.set_tracer_provider(provider)

    return trace.get_tracer(__name__)

트레이스 생성

기본 스팬 생성

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

tracer = trace.get_tracer(__name__)

def create_order(customer_id: str, items: list) -> dict:
    """주문 생성"""

    with tracer.start_as_current_span("create_order") as span:
        # 속성 추가
        span.set_attribute("customer.id", customer_id)
        span.set_attribute("items.count", len(items))

        try:
            # 비즈니스 로직
            order = process_order(customer_id, items)

            # 결과 속성 추가
            span.set_attribute("order.id", order["id"])
            span.set_attribute("order.total", order["total"])

            return order

        except Exception as e:
            # 에러 기록
            span.record_exception(e)
            span.set_status(Status(StatusCode.ERROR, str(e)))
            raise

중첩 스팬

def process_order(customer_id: str, items: list) -> dict:
    """주문 처리"""

    with tracer.start_as_current_span("process_order") as span:
        # 하위 스팬은 자동으로 부모와 연결됨
        validate_inventory(items)
        pricing = calculate_pricing(items)
        order = save_order(customer_id, items, pricing)

        return order


def validate_inventory(items: list):
    """재고 확인"""

    with tracer.start_as_current_span("validate_inventory") as span:
        span.set_attribute("items.count", len(items))
        # 재고 확인 로직
        for item in items:
            check_stock(item)

데코레이터 사용

from functools import wraps
from opentelemetry import trace

def traced(name: str = None):
    """트레이싱 데코레이터"""

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            span_name = name or func.__name__
            tracer = trace.get_tracer(__name__)

            with tracer.start_as_current_span(span_name) as span:
                try:
                    result = func(*args, **kwargs)
                    return result
                except Exception as e:
                    span.record_exception(e)
                    span.set_status(Status(StatusCode.ERROR))
                    raise

        return wrapper
    return decorator


# 사용 예시
@traced("calculate_pricing")
def calculate_pricing(items: list) -> float:
    total = sum(item["price"] * item["quantity"] for item in items)
    return total

메트릭 생성

from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter

def init_metrics():
    """메트릭 초기화"""

    exporter = OTLPMetricExporter(
        endpoint="http://otel-collector:4317",
        insecure=True
    )

    reader = PeriodicExportingMetricReader(exporter)
    provider = MeterProvider(metric_readers=[reader])
    metrics.set_meter_provider(provider)

    return metrics.get_meter(__name__)


# 메트릭 사용
meter = init_metrics()

orders_created = meter.create_counter(
    name="orders.created",
    description="Number of orders created",
    unit="1"
)

order_value = meter.create_histogram(
    name="order.value",
    description="Order value distribution",
    unit="USD"
)

# 메트릭 기록
orders_created.add(1, {"customer_type": "premium"})
order_value.record(150.50, {"currency": "USD"})

프레임워크별 통합

FastAPI

from fastapi import FastAPI
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

app = FastAPI()

# 자동 계측 활성화
FastAPIInstrumentor.instrument_app(app)

tracer = trace.get_tracer(__name__)

@app.post("/orders")
async def create_order(order: OrderRequest):
    with tracer.start_as_current_span("process_order_request") as span:
        span.set_attribute("customer.id", order.customer_id)
        # 비즈니스 로직
        return {"order_id": "12345"}

Django

# settings.py

INSTALLED_APPS = [
    # ...
    'opentelemetry.instrumentation.django',
]

# 초기화 코드 (wsgi.py 또는 manage.py)
from opentelemetry.instrumentation.django import DjangoInstrumentor
DjangoInstrumentor().instrument()

Flask

from flask import Flask
from opentelemetry.instrumentation.flask import FlaskInstrumentor

app = Flask(__name__)

# 자동 계측 활성화
FlaskInstrumentor().instrument_app(app)

@app.route('/orders', methods=['POST'])
def create_order():
    with tracer.start_as_current_span("create_order"):
        # 비즈니스 로직
        return {"status": "created"}

비동기 코드 계측

asyncio

import asyncio
from opentelemetry import trace

tracer = trace.get_tracer(__name__)

async def process_items(items: list):
    """비동기 아이템 처리"""

    with tracer.start_as_current_span("process_items") as span:
        span.set_attribute("items.count", len(items))

        # 병렬 처리
        tasks = [process_item(item) for item in items]
        results = await asyncio.gather(*tasks)

        return results


async def process_item(item: dict):
    """개별 아이템 처리"""

    with tracer.start_as_current_span("process_item") as span:
        span.set_attribute("item.id", item["id"])
        # 처리 로직
        await asyncio.sleep(0.1)
        return {"status": "processed"}

환경 변수 설정

Kubernetes에서 환경 변수로 설정:

env:
  - name: OTEL_SERVICE_NAME
    value: "order-service"
  - name: OTEL_EXPORTER_OTLP_ENDPOINT
    value: "http://otel-collector:4317"
  - name: OTEL_TRACES_SAMPLER
    value: "parentbased_traceidratio"
  - name: OTEL_TRACES_SAMPLER_ARG
    value: "0.1"  # 10% 샘플링
  - name: OTEL_PYTHON_LOG_CORRELATION
    value: "true"

트러블슈팅

트레이스가 수집되지 않음

1. 의존성 확인:

   pip list | grep opentelemetry

2. Exporter 연결 확인:

   import logging
   logging.basicConfig(level=logging.DEBUG)

3. OTLP 엔드포인트 연결 테스트

자동 계측이 작동하지 않음

1. 지원 프레임워크인지 확인
2. 계측 패키지 설치 확인:

   opentelemetry-bootstrap -a requirements

다음 단계

• 자동 계측 - OTel Operator 상세 설정
• 트레이스 - 트레이스 분석 방법
• 서비스 - 서비스 성능 모니터링