31 장

logging과 관측성

표준 logging의 함정, structlog 등 구조화 로깅, OpenTelemetry로 분산 추적, Sentry로 에러 트래킹까지.

30장에서 코드의 정적 정확성을 자동화했습니다. 이번 챕터는 한 단계 더 나아가 운영 중 무슨 일이 일어나는지 보이게 만드는 도구들을 다룹니다 — 표준 logging을 안전하게 쓰는 법, 구조화 로깅, OpenTelemetry로 분산 추적, Sentry로 에러 트래킹까지.

“보이지 않는 것은 고칠 수 없다"가 본 챕터의 한 줄 동기입니다. 28장 테스트와 배포의 다음 단계가 본 챕터입니다.

print가 아닌 logging을 써야 하는 이유 #

작은 스크립트에서 print는 괜찮습니다. 하지만 운영에 올라가면 다음이 필요해집니다.

레벨 — 모든 메시지가 같은 중요도가 아닙니다.
타임스탬프와 위치 — 언제 어디서 발생했는가.
목적지 분기 — stdout / 파일 / 외부 시스템 (예: Datadog, CloudWatch).
포맷 통일 — 사람이 읽거나 기계가 파싱하거나.
켜고 끄기 — 운영에서는 DEBUG 끄고, 디버깅할 때는 켜기.

표준 logging 모듈이 이 모든 걸 줍니다. 다만 멘탈 모델이 복잡해서 함정도 많습니다.

표준 logging의 멘탈 모델 #

logging은 네 가지 객체로 구성됩니다.

객체	역할
Logger	호출 지점. `logger.info(...)`의 그 logger
Handler	출력 목적지. StreamHandler / FileHandler / RotatingFileHandler
Formatter	메시지 모양. 시간, 레벨, 메시지를 한 줄로
Filter	조건부 통과. 특정 모듈만, 특정 레벨만

흐름은 이렇습니다.

logger.info("hello") 호출
  → Logger 가 자기 핸들러들에게 LogRecord 전달
  → 각 Handler 가 자기 Formatter 로 포맷
  → Handler 가 최종 출력 (stdout / 파일 / 네트워크)

첫 사용 #

log_hello.py

import logging

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

handler = logging.StreamHandler()
handler.setFormatter(
    logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s")
)
logger.addHandler(handler)

logger.info("hello")
logger.warning("something off")

2026-05-17 12:00:00,123 [INFO] __main__: hello
2026-05-17 12:00:00,124 [WARNING] __main__: something off

함정 1 — propagate와 중복 출력 #

logger는 계층 구조입니다. app.services.user는 app.services와 app의 자식이고, 메시지를 부모에게도 전파 (propagate) 합니다.

# main.py
import logging
logging.basicConfig(level=logging.INFO)

logger = logging.getLogger("app.services.user")
logger.addHandler(logging.StreamHandler())  # 핸들러 추가

logger.info("hello")

INFO:app.services.user:hello   ← root logger 가 출력
hello                          ← 우리가 추가한 handler 가 출력

같은 메시지가 두 번 나옵니다. 부모 root logger의 핸들러와 우리 핸들러가 둘 다 출력했기 때문입니다.

해결: 자식 logger의 propagate = False 또는 핸들러를 한 곳 (root)에만 둡니다.

logger.propagate = False

원칙: 애플리케이션 코드에서는 logger = logging.getLogger(__name__)만 호출하고 핸들러 설정은 앱 시작 지점 한 곳에서만 합니다.

함정 2 — % 포맷팅의 비용 #

logger.info("user %s logged in", user_id)와 logger.info(f"user {user_id} logged in")의 차이가 있습니다.

전자는 로깅 레벨이 INFO 이상일 때만 포맷팅이 실행 됩니다. 후자는 호출 시점에 항상 f-string이 평가됩니다. DEBUG 메시지에 무거운 객체를 넣으면 운영에서 (DEBUG off)도 계속 비용을 냅니다.

# 권장
logger.debug("user info: %s", expensive_serialize(user))

# 비권장 (DEBUG off 에서도 expensive_serialize 가 호출됨)
logger.debug(f"user info: {expensive_serialize(user)}")

함정 3 — 예외와 traceback #

예외를 로깅할 때는 logger.exception 또는 exc_info=True.

try:
    do_something()
except Exception:
    logger.exception("do_something failed")  # traceback 자동 포함

logger.error(str(e))는 traceback을 잃어버립니다.

구조화 로깅 — JSON 출력의 가치 #

사람이 한 줄씩 읽는 로그는 검색·집계·알람이 어렵습니다. JSON으로 출력 하면 로그 수집기 (Datadog, Loki, CloudWatch Logs Insights)가 필드 단위로 인덱싱하고 쿼리할 수 있습니다.

{"ts": "2026-05-17T12:00:00Z", "level": "info", "logger": "app.services.user", "msg": "user logged in", "user_id": 42, "ip": "1.2.3.4"}

이걸 표준 logging으로 직접 만들려면 Formatter를 커스텀해야 해서 번거롭습니다. structlog이 표준 답입니다.

structlog 설정 #

uv add structlog

app/core/logging.py

import logging
import sys

import structlog


def configure_logging(level: str = "INFO") -> None:
    """앱 시작 시 한 번 호출."""
    timestamper = structlog.processors.TimeStamper(fmt="iso")

    shared_processors = [
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        timestamper,
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
    ]

    structlog.configure(
        processors=[
            *shared_processors,
            structlog.processors.JSONRenderer(),
        ],
        wrapper_class=structlog.make_filtering_bound_logger(
            getattr(logging, level)
        ),
        logger_factory=structlog.PrintLoggerFactory(file=sys.stdout),
        cache_logger_on_first_use=True,
    )

사용 #

app/services/user.py

import structlog

logger = structlog.get_logger(__name__)


async def login(user_id: int, ip: str) -> None:
    logger.info("user logged in", user_id=user_id, ip=ip)

출력:

{"event": "user logged in", "user_id": 42, "ip": "1.2.3.4", "level": "info", "logger": "app.services.user", "timestamp": "2026-05-17T12:00:00.123Z"}

핵심: 메시지는 이벤트 이름 (변하지 않는 짧은 라벨)이고, 변동 데이터는 kwargs로 분리합니다. f"user {id} logged in"처럼 변수를 본문에 박아 넣으면 같은 이벤트의 인스턴스를 집계할 수 없습니다.

로그 레벨 — 실전 기준 #

레벨	언제
DEBUG	개발 중에만 본다. 운영에서는 끈다. 변수 dump, 트레이스 정보
INFO	정상 흐름 중 의미 있는 사건. 로그인, 결제, 작업 완료
WARNING	비정상이지만 자동 복구. 재시도 성공, 외부 API 지연, 차선 경로 사용
ERROR	실패. 처리 중인 요청 / 작업이 실패. 알람 대상
CRITICAL	시스템 전체 위험. DB 다운, 디스크 풀. 즉시 호출 대상

기준선: 운영은 INFO부터 켜고 ERROR부터 알람을 받습니다. WARNING은 모아 보면서 패턴을 보겠습니다.

correlation ID — 요청 단위 추적 #

한 사용자의 한 요청이 여러 모듈·여러 서비스를 거칩니다. 모든 로그에 같은 ID를 붙이면 그 요청에 무슨 일이 있었는지 한꺼번에 볼 수 있습니다.

structlog의 contextvars가 이걸 해결합니다.

app/api/middleware.py

import uuid

import structlog
from starlette.middleware.base import BaseHTTPMiddleware


class RequestIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        structlog.contextvars.clear_contextvars()
        structlog.contextvars.bind_contextvars(request_id=request_id)
        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        return response

이제 같은 요청 처리 중 호출된 모든 logger.info(...) 출력에 request_id가 자동으로 포함됩니다.

app/main.py

from fastapi import FastAPI

from app.api.middleware import RequestIdMiddleware
from app.core.logging import configure_logging

configure_logging()
app = FastAPI()
app.add_middleware(RequestIdMiddleware)

OpenTelemetry — 분산 추적의 표준 #

여러 서비스가 한 요청을 처리할 때 (예: web → API → DB → 외부 API) 각 단계가 얼마나 걸렸는지, 어디서 실패했는지를 보려면 trace가 필요합니다.

OpenTelemetry (이하 OTel)가 이 분야의 표준입니다. 한 번 instrument 하면 Jaeger, Tempo, Datadog, Honeycomb 등 어디든 보낼 수 있습니다.

멘탈 모델 #

Trace — 한 요청의 전체 흐름. 고유 ID.
Span — Trace 내 한 단위 작업. 시작 시간, 끝 시간, 속성을 가짐.
Span의 계층 — 부모-자식 관계로 호출 트리를 형성.
Metric — 시간에 따른 수치 (요청 수, 지연 백분위, CPU).
Log — 위에서 본 그 로그. trace와 연결돼야 가치가 큼.

FastAPI 자동 instrumentation #

uv add opentelemetry-distro opentelemetry-exporter-otlp \
       opentelemetry-instrumentation-fastapi \
       opentelemetry-instrumentation-sqlalchemy \
       opentelemetry-instrumentation-httpx

app/core/telemetry.py

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
    OTLPSpanExporter,
)
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor


def configure_telemetry(service_name: str, app, engine) -> None:
    resource = Resource.create({"service.name": service_name})
    provider = TracerProvider(resource=resource)
    provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
    trace.set_tracer_provider(provider)

    FastAPIInstrumentor.instrument_app(app)
    SQLAlchemyInstrumentor().instrument(engine=engine)
    HTTPXClientInstrumentor().instrument()

OTEL_EXPORTER_OTLP_ENDPOINT 환경변수로 수집기 주소를 지정합니다 (http://localhost:4317 같은 형태). 로컬에서 Jaeger 띄워보는 것이 시작점입니다.

docker run --rm -p 16686:16686 -p 4317:4317 jaegertracing/all-in-one

브라우저에서 http://localhost:16686으로 trace를 보겠습니다.

수동 span #

자동 instrumentation이 못 잡는 비즈니스 로직은 직접 감쌉니다.

from opentelemetry import trace

tracer = trace.get_tracer(__name__)


async def calculate_invoice(user_id: int) -> Invoice:
    with tracer.start_as_current_span("calculate_invoice") as span:
        span.set_attribute("user.id", user_id)
        # ... 계산
        span.set_attribute("invoice.total", total)
        return invoice

Sentry — 에러 트래킹 #

OTel이 “전반 흐름과 성능"이라면 Sentry는 “발생한 에러” 자체에 깊이 집중 합니다.

같은 에러를 자동 그룹핑 (스택트레이스 시그니처 기반).
빈도, 첫 발생, 영향받은 사용자 수 자동 집계.
릴리스 단위 추적 — “이 에러는 v1.2.3에 새로 들어왔다”.
breadcrumb — 에러 직전 무슨 일이 있었는지.

설치 #

uv add sentry-sdk

app/core/sentry.py

import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration


def configure_sentry(dsn: str, environment: str, release: str) -> None:
    sentry_sdk.init(
        dsn=dsn,
        environment=environment,
        release=release,
        traces_sample_rate=0.1,         # 10% 만 trace
        profiles_sample_rate=0.1,       # 10% 만 profile
        integrations=[
            FastApiIntegration(),
            SqlalchemyIntegration(),
        ],
        send_default_pii=False,         # 기본 PII 미전송
    )

release를 Git SHA나 버전 태그로 넘기면 “이 커밋 이후 새 에러"를 자동으로 보여줍니다. CI의 환경변수에서 받아오는 것이 일반적입니다.

export SENTRY_RELEASE=$(git rev-parse HEAD)

절대 로깅하면 안 되는 것 #

이건 보안·법적 책임에 직결되는 영역이라 자동화된 차단이 필수입니다.

PII (개인 식별 정보) #

이메일 전체, 주민번호, 전화번호, 카드번호
주소, 생년월일, 위치 좌표
사진 / 음성 / 영상 원본

내부 ID (DB primary key)는 보통 PII가 아니라 괜찮습니다. 외부 식별자 (이메일 등)가 위험합니다.

Secret #

API 토큰, 비밀번호 (해시조차도)
TLS 개인키
데이터베이스 연결 문자열 (비밀번호 포함)
AWS access key

큰 페이로드 #

요청/응답 본문 전체를 로깅하면 두 가지 문제가 같이 발생합니다.

로그 비용이 급증.
PII / secret이 본문에 섞여 있으면 함께 새어 나옴.

자동 차단 패턴 #

구조화 로깅의 장점이 여기서 살아납니다. 민감 키를 자동 마스킹 하는 processor를 등록합니다.

app/core/logging.py 확장

SENSITIVE_KEYS = {"password", "token", "authorization", "secret", "card_number"}


def mask_sensitive(logger, method_name, event_dict):
    for key in list(event_dict.keys()):
        if key.lower() in SENSITIVE_KEYS:
            event_dict[key] = "***"
    return event_dict

structlog.configure(
    processors=[
        *shared_processors,
        mask_sensitive,
        structlog.processors.JSONRenderer(),
    ],
    ...
)

이로써 logger.info("login", password="hunter2") 같은 실수가 마스킹돼 출력됩니다.

로컬 개발 vs 운영의 설정 차이 #

항목	로컬	운영
출력 포맷	사람이 읽기 좋은 컬러 텍스트	JSON
레벨	DEBUG	INFO (또는 WARNING)
목적지	stdout	stdout (컨테이너 로그 → 수집기)
Sentry	dev DSN 또는 비활성	운영 DSN, release 태그
OTel sample rate	100%	1~10%

ENV 환경변수로 분기합니다.

app/core/logging.py

import os

def configure_logging() -> None:
    env = os.getenv("ENV", "local")
    renderer = (
        structlog.dev.ConsoleRenderer(colors=True)
        if env == "local"
        else structlog.processors.JSONRenderer()
    )
    level = "DEBUG" if env == "local" else "INFO"
    # ... structlog.configure(processors=[..., renderer], ...)

29장 종합 실습에 적용하기 #

29장 종합 실습 — TODO API 완성하기의 헬스체크와 main에서 본 챕터 도구를 통합합니다.

app/main.py 확장

from contextlib import asynccontextmanager
from fastapi import FastAPI

from app.core.config import settings
from app.core.logging import configure_logging
from app.core.telemetry import configure_telemetry
from app.core.sentry import configure_sentry
from app.api.middleware import RequestIdMiddleware
from app.db import engine


@asynccontextmanager
async def lifespan(app: FastAPI):
    configure_logging()
    if settings.sentry_dsn:
        configure_sentry(settings.sentry_dsn, settings.env, settings.release)
    yield


app = FastAPI(lifespan=lifespan)
app.add_middleware(RequestIdMiddleware)

if settings.otel_enabled:
    configure_telemetry("todo-api", app, engine.sync_engine)

이제 모든 요청에 request_id가 붙고, 에러는 Sentry에, trace는 Jaeger / Tempo에 자동으로 흘러갑니다.

연습문제 #

structlog 마스킹 강화 — 본 챕터의 mask_sensitive가 최상위 키만 검사합니다. user = {"email": "...", "password": "..."} 같이 중첩된 dict의 키도 마스킹하려면 어떻게 재귀 처리하시겠나요? 그리고 그 처리가 큰 dict에서 성능 문제가 될 만한 임계는 어디라고 보시나요?
OTel 자동 instrumentation의 한계 — FastAPI 자동 instrumentation은 라우트 단위 span을 만들지만 라우트 내부 비즈니스 로직은 잡지 않습니다. 본인 프로젝트의 가장 느린 라우트 하나를 골라 수동 span을 3~4 곳에 추가하고, Jaeger에서 그 span 들의 시간 비중을 확인해 보세요. 어느 단계가 예상보다 오래 걸렸나요?
Sentry release와 회귀 — Sentry에 release를 등록하면 “이 release에서 새로 등장한 에러"를 보여줍니다. GitHub Actions의 release job에 SENTRY_RELEASE=$(git rev-parse HEAD)를 넘기고, 일부러 한 버전에만 들어가는 에러를 만들어 보세요. Sentry가 그 에러를 “new in release X"로 정확히 분류하나요?

노트

한 줄 요약 — print가 아니라 structlog로 구조화 JSON 로그를 내고, request_id로 한 요청을 묶고, OpenTelemetry로 흐름과 성능을, Sentry로 에러를 보겠습니다. 그리고 PII · secret · 큰 페이로드는 마스킹 processor로 자동 차단합니다.

다음 챕터는 uv로 라이브러리 배포입니다. 운영 도구를 정비했으니 이번엔 만든 코드를 다른 사람이 쓸 수 있게 PyPI에 출시하는 과정을 다룹니다.