Logging

시작 하기 전

이전 블로깅 내용에서는 token_validator모듈의 AccessControl 클래스를 이용해서 비즈니스 로직을 구성했는데요.

클래스를 이용해서 객체화와 추상화는 분명한 이점은 이점은 있지만 집약된 OOP는 추후 이게 뭐지?라고 본인도 짜놓고 이해 못하는 상황이 꽤 날수 있어서, 우선 함수로 표현하는 것이 일차적으로 유지보수 측면에서 더 효율 적이에요.

아예 함수로직으로 더 직관적이고 가독성있게 변경했어요.

import time
import typing
import re

import jwt

from fastapi.params import Header
from jwt.exceptions import ExpiredSignatureError, DecodeError
from pydantic import BaseModel
from starlette.requests import Request
from starlette.datastructures import URL, Headers
from starlette.responses import JSONResponse, Response

from app.common.consts import EXCEPT_PATH_LIST, EXCEPT_PATH_REGEX
from app.errors import exceptions as ex
from starlette.types import ASGIApp, Receive, Scope, Send

from app.common import config, consts
from app.common.config import conf
from app.errors.exceptions import APIException
from app.models import UserToken

from app.utils.date_utils import D
from app.utils.logger import api_logger


async def access_control(request: Request, call_next):
    request.state.req_time = D.datetime()
    request.state.start = time.time()
    request.state.inspect = None
    request.state.user = None
    ip = request.headers["x-forwarded-for"] if "x-forwarded-for" in request.headers.keys() else request.client.host
    request.state.ip = ip.split(",")[0] if "," in ip else ip
    headers = request.headers
    cookies = request.cookies
    url = request.url.path
    if await url_pattern_check(url, EXCEPT_PATH_REGEX) or url in EXCEPT_PATH_LIST:
        response = await call_next(request)
        if url != "/":
            await api_logger(request=request, response=response)
        return response

    try:
        if url.startswith("/api"):
            # api 인경우 헤더로 토큰 검사
            if "authorization" in headers.keys():
                token_info = await token_decode(access_token=headers.get("Authorization"))
                request.state.user = UserToken(**token_info)
                # 토큰 없음
            else:
                if "Authorization" not in headers.keys():
                    raise ex.NotAuthorized()
        else:
            # 템플릿 렌더링인 경우 쿠키에서 토큰 검사
            cookies["Authorization"] = "Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpZCI6MTQsImVtYWlsIjoia29hbGFAZGluZ3JyLmNvbSIsIm5hbWUiOm51bGwsInBob25lX251bWJlciI6bnVsbCwicHJvZmlsZV9pbWciOm51bGwsInNuc190eXBlIjpudWxsfQ.4vgrFvxgH8odoXMvV70BBqyqXOFa2NDQtzYkGywhV48"

            if "Authorization" not in cookies.keys():
                raise ex.NotAuthorized()

            token_info = await token_decode(access_token=cookies.get("Authorization"))
            request.state.user = UserToken(**token_info)

        response = await call_next(request)
        await api_logger(request=request, response=response)
    except Exception as e:
        error = await exception_handler(e)
        error_dict = dict(status=error.status_code, msg=error.msg, detail=error.detail, code=error.code)
        response = JSONResponse(status_code=error.status_code, content=error_dict)
        await api_logger(request=request, error=error)

    return response


async def url_pattern_check(path, pattern):
    result = re.match(pattern, path)
    if result:
        return True
    return False


async def token_decode(access_token):
    """
    :param access_token:
    :return:
    """
    try:
        access_token = access_token.replace("Bearer ", "")
        payload = jwt.decode(access_token, key=consts.JWT_SECRET, algorithms=[consts.JWT_ALGORITHM])
    except ExpiredSignatureError:
        raise ex.TokenExpiredEx()
    except DecodeError:
        raise ex.TokenDecodeEx()
    return payload


async def exception_handler(error: Exception):
    if not isinstance(error, APIException):
        error = APIException(ex=error, detail=str(error))
    return error

request.state.start = time.time(), request.state.inspect = None, request.state.user = None를 추후 어떻게 활용하는지 살펴 볼게요.

ip = request.headers["x-forwarded-for"] if "x-forwarded-for" in request.headers.keys() else request.client.host에서 x-forwarded-for의 키를 가지고 처리하는데요. 만약 loadbalance를 거치지 않는 경우 해당 키가 없기 때문에 else문에서 request.clinet.host를 가져오게 되요.

이후 ip변수를 정제 처리가 필요한데요. split()메소드와 if else를 통해서 깔끔하게 처리한 이후 request.state.ip에 값을 할당하게되요.

header, cookies, url 변수를 각각 변수로 선언해주게 되요.
아래 if await url_pattern_check(url, EXCEPT_PATH_REGEX) OR url in EXCEPT_PATH_LIST: 조건에서 참인 경우 call_next()를 통해서 바로 다음 함수를 호출하게 되요.

아래 코드는 만약 루트 url이 아닌 경우에는 logging을 시작하라는 의미에요.

if url != "/":
	await api_logger(request=request, response=response)
return response

try:
        if url.startswith("/api"):
            # api 인경우 헤더로 토큰 검사
            if "authorization" in headers.keys():
                token_info = await token_decode(access_token=headers.get("Authorization"))
                request.state.user = UserToken(**token_info)
                # 토큰 없음
            else:
                if "Authorization" not in headers.keys():
                    raise ex.NotAuthorized()
        else:
            # 템플릿 렌더링인 경우 쿠키에서 토큰 검사
            cookies["Authorization"] = "Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpZCI6MTQsImVtYWlsIjoia29hbGFAZGluZ3JyLmNvbSIsIm5hbWUiOm51bGwsInBob25lX251bWJlciI6bnVsbCwicHJvZmlsZV9pbWciOm51bGwsInNuc190eXBlIjpudWxsfQ.4vgrFvxgH8odoXMvV70BBqyqXOFa2NDQtzYkGywhV48"

            if "Authorization" not in cookies.keys():
                raise ex.NotAuthorized()

            token_info = await token_decode(access_token=cookies.get("Authorization"))
            request.state.user = UserToken(**token_info)

        response = await call_next(request)
        await api_logger(request=request, response=response)

위 소스코드들 대부분이 토큰 검사를 본격적으로 하게되고요 끝난 이후(마지막 2번째줄) call_next()를 호출하게되요.
그리고 await키워드가 붙어있어서 기다린다는 걸 알겠는데요. 무엇을 언제까지 기다릴까요?

바로 routes패키지의 users모듈 내의 get_user가 return할 때까지 기다리게 되요. 그리고 return한 값은 response 변수로 할당 받게되요.
또한 아래 api_logger()메서드를 호출하여서 오류 여부를 다시 logging하게 되요.

참고로 이전 블로깅 내용과 조금 달라 졌어요.

async def exception_handler(error: Exception):
    if not isinstance(error, APIException):
        error = APIException(ex=error, detail=str(error))
    return error

만약 오류가 서버에서 정의한 오류가 아닐 경우 APIExcetion으로 바꿔버리게 되요. 근데 왜? Exception을 APIException으로 바꾸냐고요?
HttpResponse로 오류 응답을 하기 위해서에요.
그냥 Exception에는 status_code, code, msg가 없기 때문이조. 그래서 별도로 우리가 정의해서 오류 역시 HttpResponse로 넘겨주게 되는거고요.

<참고로 # 템플릿 렌더링인 경우 쿠키에서 토큰 검사를 하게 되는되요. 이때 token_decode()를 한다는 점 유의하세요.>

로거

logger.py

import json
import logging
from datetime import timedelta, datetime
from time import time
from fastapi.requests import Request

from fastapi.logger import logger

logger.setLevel(logging.INFO)


async def api_logger(request: Request, response=None, error=None):
    time_format = "%Y/%m/%d %H:%M:%S"
    t = time() - request.state.start
    status_code = error.status_code if error else response.status_code
    error_log = None
    user = request.state.user
    body = await request.body()
    if error:
        if request.state.inspect:
            frame = request.state.inspect
            error_file = frame.f_code.co_filename
            error_func = frame.f_code.co_name
            error_line = frame.f_lineno
        else:
            error_func = error_file = error_line = "UNKNOWN"

        error_log = dict(
            errorFunc=error_func,
            location="{} line in {}".format(str(error_line), error_file),
            raised=str(error.__class__.__name__),
            msg=str(error.ex),
        )

    email = user.email.split("@") if user and user.email else None
    user_log = dict(
        client=request.state.ip,
        user=user.id if user and user.id else None,
        email="**" + email[0][2:-1] + "*@" + email[1] if user and user.email else None,
    )

    log_dict = dict(
        url=request.url.hostname + request.url.path,
        method=str(request.method),
        statusCode=status_code,
        errorDetail=error_log,
        client=user_log,
        processedTime=str(round(t * 1000, 5)) + "ms",
        datetimeUTC=datetime.utcnow().strftime(time_format),
        datetimeKR=(datetime.utcnow() + timedelta(hours=9)).strftime(time_format),
    )
    if body:
        log_dict["body"] = body
    if error and error.status_code >= 500:
        logger.error(json.dumps(log_dict))
    else:
        logger.info(json.dumps(log_dict))

현재 위의 로거 로직은 json으로 저장하기 때문에 저장 비용이 꽤 소모되요. text로 쌓는게 더 좋습니다.

참고할 만한 사항이 elastic beanstalk의 경우 AWS Cloudwatch와 쿵짝이 맞아요.

• Elastic Beanstalk에 로그를 적재
• 자동으로 Log Rotation을 지원
• S3에 텍스트 파일로 저장
• Cloudwatch에서 로그 조회 검색
• 텍스트 파일은 DB가 아니기 때문에 Athena(비쌈)라는 서비스를 이용

결론적으로 이 로그는 찍어 내기만 할 것이지만 만약 추후 외부 공개가 된다면 S3에 쌓아서 주기적으로 Dynamo DB, DocumentDB, 작은 EC2또는 MongoDB에 담아주면 되요.

NoSQL DB(MongoDB)에 로그를 쌓고 볼 수 있게끔 하면 되요.
기간은 60일치로 해주면 됩니다. 그 이후에는 모두 S3에 넣어주면 되요.

6개월이 지난 로그들은 glacier라는 아카이브 서비스에 저장해주면 되요.
왜? 이렇게까지 하냐고요? 로그에 어떤내용이 있을지 모르고 결제와 관련된 내용이 포함 될 수 있어서 결국 5년간 보관해야하는 필수불가결한 조치도 필요하기 때문이조.

코드를 뜯어 보도록 할게요.
async def api_logger(request: Request, response=None, error=None): request, response, error를 각각 매개변수로 받아요. 여기서 response는 jsonresponse에 해당해요. error의 경우는 서버에서 만든 객체에 해당되요.

body변수 정의는 추후 debug를 위해서 만들어 둡니다.
그리고 참고할 만한 것이 request.body()는 put, get을 제외하고는 사용 할 수 있다는 거!!

무엇보다 에러 관리의 측면에서 요청시 body에 어떤 값이 었는지 추적하기 위해 필요한 부분이기 떄문이조.

log_dict변수가 실제적인 로그 기록이 있어요.
url, method, statusCode, errorDetail, client, processedTime, datetimeUTC, datetimeKR등이 키값으로 있어요.

if body로 있는 경우 log_dict에 담아 줄게요.
또한 error가 500이상일 경우에는 logger.error()를 호출할게요.
그게 아닌 400이하의 오류에는 info()로 찍어줄게요.

블로깅 내용은 개인의 학습 목적을 위해 작성 된 것입니다.
잘못된 내용이 있으면 언제든지 댓글 남겨주시면 감사하겠습니다.

reference - https://www.youtube.com/watch?v=tFyE7kw-AfA&t=134s