[FastAPI] 미들웨어 - 작동 원리부터 ELK 연동을 위한 JSON 구조화 로깅까지

[FastAPI] 미들웨어 -  작동 원리부터 ELK 연동을 위한 JSON 구조화 로깅까지

FastAPI로 백엔드를 개발하다 보면 "모든 요청의 처리 시간을 재고 싶다"거나 "들어오는 모든 요청의 Body 데이터를 로그로 남기고 싶다"는 욕심이 생깁니다.

 

 오늘은 FastAPI의 숨은 조력자 미들웨어(Middleware)의 모든 것을 파헤쳐 보고, 실무에서 ELK Stack이나 Sentry와 연동 가능한 'JSON 구조화 로깅' 미들웨어를 직접 구현해 보겠습니다.

---

1. 미들웨어 아키텍처

 

클라이언트의 요청(Request)은 핵심 로직(API 엔드포인트)에 도달하기 위해 여러 겹의 미들웨어를 뚫고 들어갑니다. 처리가 끝나면 응답(Response)은 다시 그 미들웨어들을 거꾸로 통과해서 나갑니다.

주의: 등록 순서와 실행 순서

FastAPI에서 add_middleware를 여러 번 사용할 때 순서가 매우 중요합니다.
"나중에 등록한 미들웨어가 가장 먼저 실행됩니다."

# 코드 작성 순서
app.add_middleware(Middleware_A) # 가장 안쪽 (마지막 실행)
app.add_middleware(Middleware_B)
app.add_middleware(Middleware_C) # 가장 바깥쪽 (제일 먼저 실행!)

 

즉, Middleware_C가 요청을 제일 먼저 가로채게 됩니다. 따라서 인증이나 CORS 처럼 가장 먼저 검사해야 하는 기능은 코드의 하단(또는 별도 설정)에 위치해야 합니다.

 

 

---

 

2. 구조화된 로깅(Structured Logging) 설정

실무 서버 로그를 print()로 찍으면, 텍스트가 뒤죽박죽 섞여서 ELK나 CloudWatch에서 검색하기가 불가능합니다.
JSON 포맷으로 로그를 남겨야 합니다. 먼저 로거 설정을 해보겠습니다.

 

# logger_config.py
import logging
import json
import sys

class JsonFormatter(logging.Formatter):
    """
    로그를 JSON 형태로 출력하는 커스텀 Formatter
    """
    def format(self, record):
        log_data = {
            "level": record.levelname,
            "message": record.getMessage(),
            "logger_name": record.name,
            "module": record.module,
            "func_name": record.funcName,
            "timestamp": self.formatTime(record, self.datefmt),
        }
        # extra 필드로 들어온 데이터가 있다면 병합
        if hasattr(record, "extra_data"):
            log_data.update(record.extra_data)

        return json.dumps(log_data, ensure_ascii=False)

def setup_logger():
    logger = logging.getLogger("fastapi_logger")
    logger.setLevel(logging.INFO)

    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JsonFormatter())

    logger.addHandler(handler)
    return logger

# 전역 로거 생성
api_logger = setup_logger()

 

이제 api_logger.info("안녕")을 호출하면 {"level": "INFO", "message": "안녕", ...} 형태의 JSON이 출력됩니다.

---

3. Request Body 로깅 미들웨어 (Class 기반)

 

이 미들웨어는 다음 기능을 수행합니다.

1. 성능 측정: 요청 처리에 걸린 시간(Duration) 측정
2. Body 캡처: Request Body(JSON) 내용을 로그에 포함 (스트림 소모 문제 해결)
3. 구조화된 로그: 모든 정보를 하나의 JSON 객체로 출력

 

전체 코드 (middlewares.py)

import time
import json
from starlette.middleware.base import BaseHTTPMiddleware
from fastapi import Request
from logger_config import api_logger # 위에서 만든 로거 임포트

class AccessLogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()

        # [1] Request Body 읽기 (스트림 이슈 해결)
        # body를 읽어버리면 뒤에서 못 쓰므로, 읽은 후 다시 채워넣어야 함
        body_bytes = await request.body()

        async def receive_wrapper():
            return {"type": "http.request", "body": body_bytes, "more_body": False}

        request._receive = receive_wrapper  # 복구 완료!

        # [2] Body 디코딩 (로깅용)
        request_body = {}
        try:
            if body_bytes:
                request_body = json.loads(body_bytes.decode("utf-8"))
        except (json.JSONDecodeError, UnicodeDecodeError):
            request_body = {"raw": "Not a JSON body"}

        # [3] 실제 요청 처리 (다음 미들웨어/엔드포인트로 전달)
        try:
            response = await call_next(request)
        except Exception as e:
            # 예외 발생 시에도 로그를 남기기 위해 처리
            process_time = time.time() - start_time
            api_logger.error(f"Internal Server Error", extra={
                "extra_data": {
                    "method": request.method,
                    "url": str(request.url.path),
                    "error": str(e),
                    "duration": process_time
                }
            })
            raise e # 에러는 다시 던져서 FastAPI의 ExceptionHandler가 처리하게 함

        # [4] 후처리 및 JSON 로깅
        process_time = time.time() - start_time

        log_data = {
            "method": request.method,
            "url": str(request.url.path),
            "status_code": response.status_code,
            "client_ip": request.client.host,
            "duration_sec": round(process_time, 4),
            "request_body": request_body  # 여기에 Body 내용이 들어감!
        }

        # 헬스체크 등 불필요한 로그 제외
        if request.url.path != "/health":
            api_logger.info("API Request Details", extra={"extra_data": log_data})

        return response

main.py에 적용하기

from fastapi import FastAPI
from middlewares import AccessLogMiddleware

app = FastAPI()

# 미들웨어 등록
app.add_middleware(AccessLogMiddleware)

@app.post("/users")
async def create_user(user_data: dict):
    return {"status": "ok", "received": user_data}

 

 

---

 

4. Deep Dive: 왜 이렇게 복잡하게 짜나요?

 

(1) request.body()와 스트림의 비밀

 미들웨어에서 await request.body() 를 사용하면  정작 뒤에 있는 API 함수에는  "데이터가 없다(Missing)"는 에러가 발생합니다. 그래서 읽은 데이터를 다시 request._receive에 채워 넣는 복구 과정이 필수적입니다.

(2) JSON 로깅

 만든 서버의 로그가 Datadog, Kibana(ELK), Sentry 등으로 넘어갔을 때, 텍스트 형태라면 검색이 불가능합니다.
그렇기 떄문에 request_body, status_code, duration 등을 JSON 필드로 나누어 전송하면, 나중에 "처리 시간이 1초 이상 걸린 요청만 검색해" 또는 "특정 사용자 ID가 포함된 Body만 찾아" 같은 쿼리가 가능해집니다.

---

5. 주의사항 

1. 민감 정보 필터링: request_body 안에 비밀번호(password) 나 개인정보가 평문으로 들어있다면, 로깅 전에 반드시 마스킹(***) 처리 
2. 대용량 파일 업로드: 파일 업로드 요청까지 Body를 읽으려고 하면 메모리가 폭발할 수 있습니다. content-type 헤더를 확인해서 multipart/form-data인 경우 로깅을 건너뛰는 로직이 필요함
3. Health Check 제외: /health, /metrics 처럼 1초에 수십 번 호출되는 로드밸런서의 요청은 로그를 남기지 않도록 제외

---