2024-04-09 09:09

FastAPI 애플리케이션 로깅하기

현재 사용중인 FastAPI의 로깅을 설정해보자.

요구사항

user id를 액세스 로그에서 보고싶다.
user ip, exec time, url 를 보고싶다.
요청 실패 시 (dev 는 4xx, 5xx, prod는 5xx) request body도 함께 보고 싶다.
sqlalchemy 로그를 실패시에만 찍거나, prod 에만 찍고 싶다.
외부 api 호출 실패는 debug 레벨 로그로 보고 싶다.

user id, user ip 등 기본적으로 제공하는 데이터 이외에 엑세스 로그를 보고 싶다면 미들웨어를 사용하거나, api router 를 사용할 수 있다.

두 방식의 큰 차이점은 api router 는 특정 경로에 로그레벨을 지정하거나, 특정 로그만 보게 하는 등 추가적인 기능을 제공하지만, 어차피 엑세스 로그는 모든 접근에 대해 로깅하므로 의미가 없어 자료가 많은 미들웨어 방식으로 선택했다.

1. user id 를 액세스 로그에서 보고싶다.

현재 방식

[2024-04-01 03:47:46,735] [uvicorn.access] [1-140157264808712] [INFO] 172.31.48.251:55620 - "GET / HTTP/1.1" 200

현재 로그인 방식은 jwt 토큰 인증 방식을 사용한다. 따라서 user id 를 확인하기 위해서는 request 의 Authentication 헤더에서 토큰 값을 가져와야한다.

class LoggingMiddleware(BaseHTTPMiddleware):

    def __init__(
            self,
            app: FastAPI,
            *,
            logger: logging.Logger = logging.getLogger('access')
    ) -> None:
        self.security = HTTPBearer()
        self.logger = logger
        super().__init__(app)
        
    async def dispatch(self,
                       request: Request,
                       call_next: Callable,
                       ) -> Response:

        user_id = 'ANONYMOUS'

        try:
            credentials = await self.security(request)
            user_id = decode_jwt(credentials.credentials).get('sub')
        except HTTPException:
            pass

		...

fastapi에서 제공하는 HTTPBearer 를 사용하면 별도의 파싱 작업 없이 토큰 값을 가져와 사용할 수 있다.
credentials(토큰값) 을 jwt 디코딩하여 user_id 값을 얻어낼 수 있다.
HTTPException 발생 시 credentials 값이 없거나 잘못된 경우이므로 ANONYMOUS 유저 그대로 둔다.
위와 같이 미들웨어로 등록할 클래스에 dispatch 메서드를 정의하면 이후에 fastapi 가 알아서 dispatch 메서드를 호출한다.

2. user ip, exec time, url 를 보고싶다.

        ...
        										
        start_time = time.perf_counter()
        response: Response = await call_next(request)
        end_time = time.perf_counter()

        exec_time = end_time - start_time
        
        user_ip = request.client.host
        method = request.method
        url = request.url
        res_code = response.status_code
        
        ...

Request 객체, Response 객체에서 간단하게 꺼내와 사용할 수 있다.
Response 객체를 얻기 위해서는 dispatch 메서드의 call_next 를 호출하여 얻을 수 있다. (미들웨어는 클라이언트와 서버 사이에서 동작하므로 중간에 가로채서 로깅만 하고 이후 로직을 실행시키도록 call_next 같은 함수를 호출해주는 것)

3. 요청 실패 시 (dev 는 4xx, 5xx, prod는 5xx) request body도 함께 보고 싶다.

        ...
        req_body = await request.body()
        await self._set_body(request, req_body)

        start_time = time.perf_counter()
        response: Response = await call_next(request)
        end_time = time.perf_counter()

        exec_time = end_time - start_time

        res_body = b''
        async for chunk in response.body_iterator:
            res_body += chunk

        if response.status_code >= 400:
            task = BackgroundTask(self.log_warning, user_id, exec_time, request, response, req_body)
        else:
            task = BackgroundTask(self.log_info, user_id, exec_time, request, response)

        return Response(content=res_body, status_code=response.status_code,
                    headers=dict(response.headers), media_type=response.media_type, background=task)
                    
                    
    def log_info(self, user_id, exec_time, request, response):
        self.logger.info(f"[user {user_id}] [{request.client.host}] [excuted in {exec_time:0.5f}] {request.method} {request.url} {response.status_code}")

    def log_warning(self, user_id, exec_time, request, response, req_body):
        self.logger.warning(f"[user {user_id}] [{request.client.host}] [excuted in {exec_time:0.5f}] {request.method} {request.url} {response.status_code} {req_body}")
        
    async def _set_body(self, request: Request, body: bytes):
        async def receive() -> Message:
            return {'type': 'http.request', 'body': body}

        request._receive = receive

request.body() 를 호출만 하면 읽어올 수 있다. 하지만, 그냥 request.body() 만 호출하고 끝내버리면 내부적으로 request, response 는 이터레이터를 통해 값을 가져오기 때문에 실제 서버 애플리케이션에서 request의 body 를 사용할 수 없게되는 문제가 발생한다.
따라서, 꺼낸 뒤 다시 넣어주는 작업(_set_body 메서드)이 필요하다.
request 값이 달라져 response를 그대로 반환 시 에러가 발생하기도 하고, 로깅 작업을 백그라운드로 실행시키기 위해 새로운 Response 객체를 리턴한다.

4. sqlalchemy 로그를 실패시에만 찍거나, prod 에만 찍고 싶다.

sqlalchemy.engine 로그는 INFO 레벨로 쿼리와 파라미터를, DEBUG 레벨로 쿼리 결과를 볼 수 있다. 이 외의 로그 레벨에 해당하는 로그는 없다.
쿼리 런타임 에러는 내부에서 발생하여 그대로 서버 예외로 던져지고, 500 에러가 반환된다.
따라서 dev 서버는 DEBUG(or INFO) 레벨로 찍고 prod 서버에서는 sqlalchemy 로그를 따로 찍지 않으면 실패시에만 에러를 로깅한다.

5. 외부 api 호출 실패는 debug 레벨 로그로 보고 싶다.

직접 raise 로 처리하거나 print 로 처리되어 있는 부분을 logging.debug() 를 통해 로깅하도록 바꾼다

참고.

https://stackoverflow.com/questions/69670125/how-to-log-raw-http-request-response-in-python-fastapi

FastAPI

sqlalchemy