[AI] logging Mixin 패턴으로 구현하는 구조화된 로깅 시스템과 FastAPI 요청 추적

요청별 추적과 구조화된 로깅이 필수적입니다. 이번 글에서는 FastAPI와 Mixin 패턴을 활용하여 효과적인 로깅 시스템을 구축하는 방법을 소개하겠습니다.

 

1. 구조화된 JSON 로깅 시스템

1.1 핵심 구성 요소

먼저 JSON 형식의 구조화된 로그를 생성하는 시스템을 만들어보겠습니다.

비동기 환경에서의 요청 추적: FastAPI나 다른 비동기 웹 프레임워크에서는 여러 요청이 동시에 처리됩니다
요청별 로그 분리: 각 HTTP 요청마다 고유한 request_id를 부여하여, 하나의 요청과 관련된 모든 로그를 쉽게 추적할 수 있습니다

 

자동 로그 태깅: 모든 로그 레코드에 현재 컨텍스트의 request_id를 자동으로 추가합니다
일관성 보장: 개발자가 매번 수동으로 request_id를 로그에 추가할 필요가 없습니다
로그 분석 용이성: JSON 로그에서 request_id로 필터링하여 특정 요청의 전체 흐름을 추적할 수 있습니다

{"timestamp": "2024-01-15 10:30:45", "level": "ERROR", "logger": "program", "request_id": "req-123", "message": "사용자 인증 실패"}

# logger/logger.py
import logging
import os
from logging.handlers import TimedRotatingFileHandler
import contextvars
import json

# 요청별 식별자를 위한 ContextVar
request_id_ctx_var = contextvars.ContextVar("request_id", default=None)

class RequestIdFilter(logging.Filter):
    """모든 로그 레코드에 request_id를 추가하는 필터"""
    def filter(self, record):
        record.request_id = request_id_ctx_var.get() or "-"
        return True

class JSONFormatter(logging.Formatter):
    """구조화된 JSON 로그 형식"""
    def format(self, record):
        log_record = {
            "timestamp": self.formatTime(record, self.datefmt),
            "level": record.levelname,
            "logger": record.name,
            "request_id": getattr(record, "request_id", "-"),
            "message": record.getMessage(),
        }
        return json.dumps(log_record, ensure_ascii=False)

1.2 로그 관리자 클래스

logging생성

로그 관리에는 .jsonl이 효과적

class LoggerManager:
    def __init__(self):
        # 로그 디렉토리 설정
        BASE_DIR = os.path.dirname(os.path.abspath(__file__))
        self.ALL_LOG_DIR = os.path.abspath(os.path.join(BASE_DIR, "../../logs/program_log"))
        self.INSTRUCTION_LOG_DIR = os.path.abspath(os.path.join(BASE_DIR, "../../logs/instruction_log"))
      
        # 디렉토리 생성
        os.makedirs(self.ALL_LOG_DIR, exist_ok=True)
        os.makedirs(self.INSTRUCTION_LOG_DIR, exist_ok=True)

        # JSON Formatter & Filter 설정
        json_formatter = JSONFormatter()
        reqid_filter = RequestIdFilter()

        # 용도별 로그 핸들러 설정
        self._setup_program_logger(json_formatter, reqid_filter)
        self._setup_instruction_logger(json_formatter, reqid_filter)
        self._setup_console_handler(json_formatter, reqid_filter)

    def _setup_program_logger(self, formatter, filter):
        """전체 프로그램 로그 설정"""
        program_log_file = os.path.join(self.ALL_LOG_DIR, "program.jsonl")
        handler = TimedRotatingFileHandler(
            program_log_file, when="midnight", interval=1, backupCount=0, encoding="utf-8"
        )
        handler.suffix = "%Y-%m-%d"
        handler.setFormatter(formatter)
        handler.addFilter(filter)
        
        # 루트 로거 설정
        logging.basicConfig(
            level=logging.INFO,
            handlers=[handler, logging.StreamHandler()]
        )

    def _setup_instruction_logger(self, formatter, filter):
        """명령어 추적용 로거 설정"""
        instruction_log_file = os.path.join(self.INSTRUCTION_LOG_DIR, "instruction.jsonl")
        handler = TimedRotatingFileHandler(
            instruction_log_file, when="midnight", interval=1, backupCount=0, encoding="utf-8"
        )
        handler.suffix = "%Y-%m-%d"
        handler.setFormatter(formatter)
        handler.addFilter(filter)
        
        self.instruction_logger = logging.getLogger("instruction")
        self.instruction_logger.setLevel(logging.INFO)
        self.instruction_logger.addHandler(handler)
        self.instruction_logger.propagate = True

    def get_logger(self, name=None):
        """로거 인스턴스 반환"""
        if name == "instruction":
            return self.instruction_logger
        return logging.getLogger(name)

# 전역 인스턴스
logger_manager = LoggerManager()
get_logger = logger_manager.get_logger

2. Mixin 패턴으로 로거 주입

2.1 HasLogger Mixin 클래스

Mixin 패턴을 활용하여 클래스에 로거를 쉽게 주입할 수 있으며 다중상속 가능

# logger/base_logger.py
from logger.logger import logger_manager

class HasLogger(object):
    """로거를 자동으로 주입하는 Mixin 클래스"""
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # 클래스별 로거 생성
        self.logger = logger_manager.get_logger(
            self.__class__.__module__ + '.' + self.__class__.__name__
        )
        # 특수 목적 로거들
        self.instruction_logger = logger_manager.get_logger("instruction")

2.2 Mixin 패턴 활용 예제

# 사용 예제
class UserService(HasLogger):
    def create_user(self, user_data):
        self.logger.info(f"사용자 생성 시작: {user_data['email']}")
        self.instruction_logger.info(f"CREATE_USER 명령 실행: {user_data}")
        
        # 비즈니스 로직...
        
        self.logger.info("사용자 생성 완료")
        return user_id

3. FastAPI 미들웨어로 요청 추적

3.1 요청 ID 미들웨어

FastAPI 미들웨어를 활용하여 모든 요청에 고유한 ID를 부여하고 로그 추적을 가능하게 합니다.

어느 .py에서 로그가 남던지 api를 통해 동작하는 모든 .py에는 request_id를 통해서 쉽게 찾을 수 있습니다.

# interfaces/api_gateway.py
import uuid
from fastapi import FastAPI, Request
from logger.logger import get_logger, request_id_ctx_var

app = FastAPI()

logger = get_logger(__name__)
instruction_logger = get_logger("instruction")

@app.middleware("http")
async def add_request_id_middleware(request: Request, call_next):
    """모든 HTTP 요청에 고유 ID를 부여하는 미들웨어"""
    request_id = str(uuid.uuid4())
    request_id_ctx_var.set(request_id)
    
    logger.info(f"요청 시작: {request.method} {request.url.path}")
    
    response = await call_next(request)
    
    logger.info(f"요청 완료: {request.method} {request.url.path}")
    return response

3.2 스트리밍 응답에서의 로그 추적

api 로그 추적을 유지하는 방법입니다.

@app.post("/generateStream")
async def stream_response(request: QueryReq):
    logger.info(f"[API][REQ][generateStream] 요청 수신: request={request}")
    
    try:
        instruction_logger.info(f"[API][REQ][generateStream] payload={request.model_dump()}")
    except Exception as e:
        instruction_logger.error(f"[API][REQ][generateStream] 로깅 실패: error={e}")

    return 0