JYP Garden

FastAPI 구조화된 로깅 (JSON Logging & Middleware)

Properties1
tagssource, logging, monitoring, fastapi, mlops, module-6, knowledge

6 min read

FastAPI 구조화된 로깅 (JSON Logging & Middleware)

Source: raw/articles/2026-04-26-structured-logging-fastapi-json-middleware.md Type: article By: FastAPI + python-json-logger + structlog Valid as of: 2026-04-26

핵심 Takeaway

  • 구조화된 로깅: 텍스트 로그 → JSON 형식으로 자동 파싱·필터링·집계
  • 왜: ELK/Datadog 같은 로그 수집 도구가 JSON을 자동 분석 가능
  • FastAPI 미들웨어: 모든 요청·응답 자동 추적 (코드 수정 불필요)
  • 핵심 필드: request_id, method, path, status_code, response_time_ms
  • 에러 추적: 예외 발생 시 traceback + request_id로 분산 추적
  • 성능: 비동기 로깅으로 메인 요청 처리 성능 보장
  • 모니터링: Prometheus 메트릭 수출 → Grafana 시각화
  • 알림: 5xx 에러 급증 감지 → 자동 알림

학습 목표 연계

Lecture Plan 2026 — 6-3: 구조화된 로깅 (Bloom L3-Apply)

학습목표증거상태
Behavior: JSON 형식 로그 출력python-json-logger + structlog 구현✅
Condition: 미들웨어로 성능·에러 추적RequestLoggingMiddleware + exception handler✅
Degree: 모든 요청 응답시간·상태코드 로깅실전 예제 + 체크리스트✅

내용 구성

1. 구조화된 로깅의 필요성

  • 텍스트 로그의 한계: 파싱·검색·집계 어려움
  • JSON 구조화 로그의 장점: 자동 파싱, SQL 쿼리처럼 필터링
  • 로그 수집 파이프라인: Docker → ELK/Datadog → 시각화·알림

2. Python 로깅 라이브러리 2가지

  • python-json-logger: 간단, 기본 JSON 포매팅
  • structlog: 강력, 컨텍스트 관리, 중첩 필드

3. FastAPI 미들웨어 구현

  • 요청 시작 시간 기록 → 핸들러 호출 → 응답 시간 계산
  • 요청/응답 로그 자동 생성 (핸들러 수정 불필요)
  • response_time_ms, status_code, user_agent 자동 추적

4. 에러 처리 & 로깅

  • HTTPException 핸들러 + 에러 로깅
  • traceback 포함으로 장애 진단 용이
  • request_id로 분산 추적 (다중 서비스 환경)

5. 성능 메트릭 로깅

  • 응답 시간 기반 경고 레벨 (fast/normal/slow)
  • DB 쿼리 시간 추적 (SQLAlchemy 이벤트 리스너)
  • 느린 쿼리 자동 감지 (> 0.5초)

6. 로그 수집 파이프라인 통합

  • Docker 환경: stdout → Docker 데몬 → 수집 도구
  • ELK Stack (Elasticsearch·Logstash·Kibana): 완전 예제
  • Datadog 통합: API 키 설정 → 자동 수집

7. 완전한 FastAPI 로깅 설정

  • setup_logging() 함수로 JSON 포매터 초기화
  • RequestLoggingMiddleware: 요청/응답 자동 추적
  • HTTPException 핸들러: 예외 자동 로깅
  • /health, /predict 예제 엔드포인트

8. 모범 사례

  • 로그 레벨 구분 (DEBUG/INFO/WARNING/ERROR/CRITICAL)
  • 민감 정보 제외 (암호, API 키 → 존재 여부만)
  • 일관된 필드명 (response_time_ms, elapsed_ms)
  • Request ID로 분산 추적

9. 성능 최적화

  • 비동기 로깅 (QueueHandler + QueueListener)
  • 로그 샘플링 (대규모 트래픽 시 로그량 제어)

10. 모니터링 & 알림

  • Prometheus 메트릭 수출 (fastapi_requests_total, request_duration_seconds)
  • Datadog/Splunk 로그 기반 알림 (5xx > threshold)

관련 개념

모듈 6 다른 학습목표와의 연계

실전 구현 요점

로깅 초기화

import logging
from pythonjsonlogger import jsonlogger
import sys
 
handler = logging.StreamHandler(sys.stdout)
formatter = jsonlogger.JsonFormatter()
handler.setFormatter(formatter)
logger = logging.getLogger()
logger.addHandler(handler)
logger.setLevel(logging.INFO)

미들웨어

from starlette.middleware.base import BaseHTTPMiddleware
import time
 
class RequestLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.perf_counter()
        response = await call_next(request)
        elapsed = time.perf_counter() - start_time
        
        logger.info("request_completed", extra={
            "request_id": request.headers.get("X-Request-ID"),
            "method": request.method,
            "path": request.url.path,
            "status_code": response.status_code,
            "response_time_ms": int(elapsed * 1000)
        })
        return response
 
app.add_middleware(RequestLoggingMiddleware)

에러 핸들러

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    logger.error("http_exception", extra={
        "status_code": exc.status_code,
        "detail": exc.detail,
        "request_id": request.headers.get("X-Request-ID")
    })
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail}
    )

로그 분석 쿼리 예제 (Kibana/Datadog)

// 5xx 에러 > 5개/분
status_code: [500, 501, 502, 503] AND timestamp: last_minute
// Count > 5 → 알림
 
// 응답 시간 > 2초인 요청
response_time_ms > 2000 AND path: "/predict"
 
// 특정 사용자의 요청 추적
user_id: "user_123"
 
// DB 쿼리 시간 급증
elapsed_ms > 1000 AND service: "database"

성능 & 안정성 체크리스트

항목확인 방법목표
로그 형식docker logs 실행 → JSON 출력✅ JSON 파싱 가능
요청 추적request_id 일관성✅ 분산 추적 가능
응답 시간로그 response_time_ms 기록✅ < 1ms 오버헤드
에러 로깅5xx 상황 → traceback 포함✅ 디버깅 정보 충분
로그 수집stdout → ELK/Datadog 도착✅ 실시간 수집

마지막 체크

프로덕션 배포 전 반드시 확인:

  • JSON 형식 로그 확인?
  • 모든 요청에 request_id 할당?
  • 응답 시간(response_time_ms) 로깅?
  • 에러 발생 시 traceback 포함?
  • 민감 정보(암호, API 키) 제외?
  • 로그 수집 도구 연동 완료? (ELK/Datadog)
  • 로그 기반 알림 규칙 설정?
  • 비동기 로깅으로 성능 영향 검증?
  • 대규모 트래픽 시 샘플링 전략?
  • 로그 보관 정책 정의? (30/90일 등)

그래프 뷰