beautifulsoups.dev
Published on

Python 데코레이터+ContextVar로 async 로그 추적

Authors

서버가 asyncio 기반으로 바뀌면 로그가 갑자기 “흩어져” 보이기 시작합니다. 동시 요청이 한 프로세스 안에서 섞여 실행되기 때문에, 단순히 전역 변수에 request_id를 넣거나 스레드 로컬을 쓰는 방식은 쉽게 깨집니다. 그 결과는 익숙합니다.

  • 같은 요청의 로그가 서로 다른 요청 로그 사이에 끼어들어 추적이 어려움
  • 예외가 터졌는데 어떤 사용자/요청에서 시작됐는지 연결이 안 됨
  • 외부 API 호출, DB 쿼리, 재시도 로직이 섞이면서 “원인-결과”가 흐려짐

이 글에서는 Python의 ContextVar로 “요청 컨텍스트(예: trace_id)”를 안전하게 전파하고, 데코레이터로 로깅을 표준화해 async 로그 추적을 안정적으로 만드는 방법을 다룹니다.

또한 운영 환경에서 로그만으로 장애를 좁혀가는 과정은 결국 “상관관계(correlation)” 싸움입니다. 예를 들어 EKS에서 파드가 재시작되거나 로그가 끊길 때는 더더욱 요청 단위 추적이 중요합니다. 관련해서는 EKS Pod CrashLoopBackOff 로그 없을 때 7단계 진단도 함께 참고하면 좋습니다.

왜 ContextVar인가: async에서 전역/스레드 로컬이 깨지는 이유

전역 변수의 문제

전역 변수는 동시 요청이 들어오면 덮어씌워집니다. await로 실행이 양보되는 순간 다른 코루틴이 실행되며 값이 바뀌기 때문입니다.

threading.local()의 한계

threading.local()은 “스레드” 단위 격리입니다. 하지만 asyncio는 보통 단일 스레드 이벤트 루프에서 다수 코루틴을 돌립니다. 즉, 스레드 로컬은 격리 단위가 맞지 않습니다.

ContextVar의 장점

ContextVar는 코루틴 컨텍스트 단위로 값을 유지합니다.

  • 같은 Task 흐름에서 await가 여러 번 일어나도 값이 유지
  • 동시에 실행되는 다른 Task에는 영향이 없음
  • contextvars.copy_context()로 명시적 전파도 가능

즉, “요청 단위”의 식별자(예: trace_id, user_id, tenant_id)를 안전하게 들고 다닐 수 있습니다.

목표: 로그에 trace_id를 자동으로 붙이고, 함수 경계에서 표준 로깅

여기서 만들고 싶은 결과는 간단합니다.

  1. 요청이 시작될 때 trace_id를 생성하거나 헤더에서 읽어 ContextVar에 저장
  2. 이후 어느 함수에서든 logger.info(...)만 호출해도 trace_id가 자동으로 로그에 포함
  3. 데코레이터로 다음을 표준화
    • 호출 시작/종료 로그
    • 실행 시간(ms)
    • 예외 발생 시 stack trace + trace_id

1) ContextVar로 trace 컨텍스트 정의

먼저 컨텍스트 변수를 정의합니다.

# tracing_context.py
from __future__ import annotations

from contextvars import ContextVar
from dataclasses import dataclass
from typing import Optional


@dataclass(frozen=True)
class TraceContext:
    trace_id: str
    user_id: Optional[str] = None


trace_ctx_var: ContextVar[Optional[TraceContext]] = ContextVar(
    "trace_ctx_var",
    default=None,
)


def get_trace_context() -> Optional[TraceContext]:
    return trace_ctx_var.get()


def set_trace_context(ctx: TraceContext):
    # token을 반환해두면 요청 종료 시 원복(reset)할 수 있음
    return trace_ctx_var.set(ctx)


def reset_trace_context(token):
    trace_ctx_var.reset(token)

핵심은 ContextVar가 “현재 실행 컨텍스트”에 묶여 있다는 점입니다. 이제 같은 요청 흐름에서는 어디서든 get_trace_context()로 값을 읽을 수 있습니다.

2) 로거에 trace_id를 자동 주입: logging.Filter 패턴

Python 표준 logging만으로도 충분히 구조를 만들 수 있습니다. 가장 깔끔한 방식 중 하나는 Filter를 붙여 LogRecord에 필드를 주입하는 것입니다.

# logging_setup.py
from __future__ import annotations

import logging

from tracing_context import get_trace_context


class TraceContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        ctx = get_trace_context()
        record.trace_id = ctx.trace_id if ctx else "-"
        record.user_id = ctx.user_id if ctx and ctx.user_id else "-"
        return True


def setup_logging() -> None:
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)

    handler = logging.StreamHandler()
    formatter = logging.Formatter(
        fmt=(
            "%(asctime)s %(levelname)s "
            "trace_id=%(trace_id)s user_id=%(user_id)s "
            "%(name)s - %(message)s"
        )
    )
    handler.setFormatter(formatter)
    handler.addFilter(TraceContextFilter())

    logger.handlers.clear()
    logger.addHandler(handler)

이제 어디서든

import logging
logger = logging.getLogger(__name__)
logger.info("hello")

만 해도 trace_id가 자동으로 찍힙니다.

3) 요청 시작점에서 ContextVar 세팅 (FastAPI 예시)

대부분의 async 웹 서버에서는 “요청 시작점”이 명확합니다. FastAPI라면 미들웨어가 가장 일반적입니다.

# app.py
from __future__ import annotations

import uuid
import logging

from fastapi import FastAPI, Request

from tracing_context import TraceContext, set_trace_context, reset_trace_context
from logging_setup import setup_logging


setup_logging()
logger = logging.getLogger(__name__)

app = FastAPI()


@app.middleware("http")
async def trace_middleware(request: Request, call_next):
    incoming = request.headers.get("x-trace-id")
    trace_id = incoming or uuid.uuid4().hex

    user_id = request.headers.get("x-user-id")

    token = set_trace_context(TraceContext(trace_id=trace_id, user_id=user_id))
    try:
        logger.info("request_start method=%s path=%s", request.method, request.url.path)
        response = await call_next(request)
        response.headers["x-trace-id"] = trace_id
        logger.info("request_end status_code=%s", response.status_code)
        return response
    finally:
        reset_trace_context(token)

여기서 token으로 원복하는 이유는 테스트/백그라운드 작업 등에서 컨텍스트가 “새어 나가는” 문제를 예방하기 위해서입니다.

4) 데코레이터로 함수 경계 로깅 표준화 (async 지원)

이제 본론입니다. “모든 서비스 함수”에 공통적으로 다음을 넣고 싶습니다.

  • 시작/종료
  • 실행 시간
  • 예외 시 에러 로그

이걸 매번 손으로 쓰면 금방 코드가 지저분해지므로 데코레이터로 묶습니다.

# trace_decorators.py
from __future__ import annotations

import functools
import inspect
import logging
import time
from typing import Any, Callable, TypeVar, ParamSpec


P = ParamSpec("P")
R = TypeVar("R")

logger = logging.getLogger("trace")


def traced(operation: str | None = None):
    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        name = operation or func.__name__

        if inspect.iscoroutinefunction(func):
            @functools.wraps(func)
            async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> Any:
                start = time.perf_counter()
                logger.info("start op=%s", name)
                try:
                    result = await func(*args, **kwargs)
                    elapsed_ms = (time.perf_counter() - start) * 1000
                    logger.info("end op=%s elapsed_ms=%.2f", name, elapsed_ms)
                    return result
                except Exception:
                    elapsed_ms = (time.perf_counter() - start) * 1000
                    logger.exception("error op=%s elapsed_ms=%.2f", name, elapsed_ms)
                    raise

            return async_wrapper  # type: ignore[return-value]

        @functools.wraps(func)
        def sync_wrapper(*args: P.args, **kwargs: P.kwargs) -> Any:
            start = time.perf_counter()
            logger.info("start op=%s", name)
            try:
                result = func(*args, **kwargs)
                elapsed_ms = (time.perf_counter() - start) * 1000
                logger.info("end op=%s elapsed_ms=%.2f", name, elapsed_ms)
                return result
            except Exception:
                elapsed_ms = (time.perf_counter() - start) * 1000
                logger.exception("error op=%s elapsed_ms=%.2f", name, elapsed_ms)
                raise

        return sync_wrapper  # type: ignore[return-value]

    return decorator

이제 서비스 코드에서는 다음처럼 사용합니다.

# services.py
from __future__ import annotations

import asyncio
import logging

from trace_decorators import traced

logger = logging.getLogger(__name__)


@traced("fetch_user_profile")
async def fetch_user_profile(user_id: str) -> dict:
    logger.info("calling downstream")
    await asyncio.sleep(0.05)
    return {"user_id": user_id, "name": "Alice"}

이 패턴의 장점은 “추적 포맷”을 한 곳에서 바꿀 수 있다는 점입니다. 예를 들어 elapsed_ms를 p95 분석용으로 남기거나, op 이름을 표준화하는 등의 변경이 쉬워집니다.

5) asyncio.create_task에서 컨텍스트 전파 주의점

대부분의 경우 Python은 Task 생성 시점의 컨텍스트를 복사해 전파합니다(현대 Python에서 일반적으로 기대하는 동작). 하지만 다음 상황에서는 주의가 필요합니다.

  • 외부 라이브러리가 자체 스레드/이벤트 루프를 만들거나
  • executor로 스레드 풀에 넘기거나
  • 컨텍스트를 의도적으로 분리해야 하거나

스레드 풀로 넘길 때

run_in_executor로 넘어가면 스레드 경계가 생깁니다. 이때는 컨텍스트가 자동으로 이어지지 않을 수 있으므로 명시 전파가 안전합니다.

import asyncio
from contextvars import copy_context


def blocking_work(x: int) -> int:
    # 여기서도 trace_id가 필요하다면 컨텍스트 전파가 필요
    return x * 2


async def run_blocking(x: int) -> int:
    loop = asyncio.get_running_loop()
    ctx = copy_context()
    return await loop.run_in_executor(None, ctx.run, blocking_work, x)

이 패턴을 알아두면 “특정 로그만 trace_id가 -로 찍히는” 미스터리를 빨리 끝낼 수 있습니다.

6) 구조화 로그(JSON)로 확장하기

운영에서는 텍스트 로그보다 JSON 로그가 훨씬 다루기 쉽습니다. 예를 들어 trace_id로 필터링하고, op별 지연 시간을 집계하고, 에러율을 대시보드로 만들기 좋습니다.

표준 logging만으로도 JSON을 만들 수 있지만, 보통은 python-json-logger 같은 라이브러리를 씁니다. 여기서는 개념만 간단히 예시로 남깁니다.

import json
import logging

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "trace_id": getattr(record, "trace_id", "-"),
            "user_id": getattr(record, "user_id", "-"),
            "time": self.formatTime(record, self.datefmt),
        }
        return json.dumps(payload, ensure_ascii=False)

JSON으로 바꾸면, 장애 상황에서 “어느 요청이 S3에서 403을 맞았는지” 같은 상관관계 추적이 훨씬 쉬워집니다. S3 권한/정책 이슈를 로그로 좁혀가는 흐름은 S3 AccessDenied 403 진단 - 버킷 정책·SCP·VPCE 글과도 연결됩니다.

7) 실전 팁: 무엇을 컨텍스트에 넣을 것인가

trace_id만 넣어도 효과가 크지만, 운영에서 자주 도움이 되는 필드는 다음과 같습니다.

  • trace_id: 요청 상관관계의 핵심
  • user_id: 사용자 단위 문제 재현/추적
  • tenant_id: 멀티테넌트에서 필수
  • request_path, method: 미들웨어에서 세팅(너무 길면 주의)
  • client_ip: 개인정보/보안 정책 고려

다만 컨텍스트는 “너무 많은 값”을 넣으면 오히려 비용이 됩니다.

  • 로그 크기 증가(저장/전송 비용)
  • 민감 정보 유출 위험
  • 포맷 변경 시 파급

권장 접근은 trace_id를 최소로 시작하고, 필요할 때만 필드를 늘리는 것입니다.

8) 테스트 전략: 컨텍스트 누수와 병렬 실행

ContextVar는 강력하지만, 테스트에서 다음 케이스를 꼭 확인해야 합니다.

  • 요청 종료 후 컨텍스트가 원복되는지(reset)
  • 동시에 두 요청을 흉내 내면 trace_id가 섞이지 않는지

간단한 동시성 테스트 예시는 다음과 같습니다.

import asyncio
from tracing_context import TraceContext, set_trace_context, reset_trace_context, get_trace_context


async def worker(tid: str) -> str:
    token = set_trace_context(TraceContext(trace_id=tid))
    try:
        await asyncio.sleep(0.01)
        ctx = get_trace_context()
        return ctx.trace_id if ctx else "-"
    finally:
        reset_trace_context(token)


async def main():
    a, b = await asyncio.gather(worker("A"), worker("B"))
    assert a == "A"
    assert b == "B"


asyncio.run(main())

이런 테스트가 통과하면 “최소한의 상관관계”는 확보된 것입니다.

9) 어디까지가 로깅이고, 어디부터가 트레이싱인가

ContextVar + 데코레이터는 애플리케이션 내부에서 상관관계를 만들기에 매우 좋습니다. 하지만 다음 니즈가 생기면 분산 트레이싱(OpenTelemetry 등)로 확장하는 것을 고려할 수 있습니다.

  • 서비스가 여러 개로 쪼개져 있고, 서비스 간 호출을 end-to-end로 보고 싶다
  • DB/HTTP 클라이언트 계층까지 span 단위로 보고 싶다
  • 지연 시간 병목을 자동으로 시각화하고 싶다

그럼에도 불구하고, 많은 팀에서 “가장 먼저” 효과를 보는 건 이 글의 패턴입니다. 구현이 단순하고, 기존 로깅 체계를 크게 흔들지 않으며, 즉시 운영 효용이 나오기 때문입니다.

정리

  • async 환경에서는 전역 변수나 스레드 로컬로 요청 단위 로그 추적이 어렵습니다.
  • ContextVar는 코루틴 컨텍스트에 안전하게 값을 묶어 trace_id 같은 상관관계 키를 유지합니다.
  • logging.Filter로 trace_id를 자동 주입하면, 애플리케이션 전역에서 일관된 로그를 얻을 수 있습니다.
  • 데코레이터로 함수 경계 로깅(시작/종료/지연/예외)을 표준화하면, 운영에서 추적 속도가 크게 올라갑니다.
  • 스레드 풀이나 특수 실행 경계에서는 copy_context()로 전파를 명시하면 “trace_id 누락”을 줄일 수 있습니다.

이 패턴을 기본으로 깔아두면, 이후에 OpenTelemetry 같은 분산 트레이싱을 붙이더라도 trace_id 중심의 사고방식이 그대로 이어져 디버깅과 관측 가능성이 훨씬 단단해집니다.

Discuss on Twitter • View on GitHub