Gunicorn/Uvicorn/FastAPI 애플리케이션을 위한 통합 Python 로깅

나는 최근에 FastAPI
및 HTTPX,
Gunicorn으로 내 앱을 배포하고 있습니다.
Uvicorn 노동자.

하지만 서빙할 때 각 구성 요소의 로그는 상당히 다르게 보입니다.
다른 사람들로부터. 나는 그것들이 모두 똑같이 보이기를 원하므로 쉽게 읽을 수 있습니다.
또는 Kibana 과 같이 악용하십시오.

이해하려고 많은 시간을 보낸 후
Python logging 작동 방식,
라이브러리의 로깅 설정을 재정의하는 방법,
여기 내가 가진 것이 있습니다 ...

단일run.py 파일!
로깅 구성, Gunicorn 구성,
나머지 코드는 래핑하기가 더 어렵기 때문에 여러 파일로 나뉩니다.
주위에 내 머리.

모든 것이 이 단일 파일에 포함되어 있습니다.

import os
import logging
import sys

from gunicorn.app.base import BaseApplication
from gunicorn.glogging import Logger
from loguru import logger

from my_app.app import app


LOG_LEVEL = logging.getLevelName(os.environ.get("LOG_LEVEL", "DEBUG"))
JSON_LOGS = True if os.environ.get("JSON_LOGS", "0") == "1" else False
WORKERS = int(os.environ.get("GUNICORN_WORKERS", "5"))


class InterceptHandler(logging.Handler):
    def emit(self, record):
        # Get corresponding Loguru level if it exists
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        # Find caller from where originated the logged message
        frame, depth = logging.currentframe(), 2
        while frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1

        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())


class StubbedGunicornLogger(Logger):
    def setup(self, cfg):
        handler = logging.NullHandler()
        self.error_logger = logging.getLogger("gunicorn.error")
        self.error_logger.addHandler(handler)
        self.access_logger = logging.getLogger("gunicorn.access")
        self.access_logger.addHandler(handler)
        self.error_log.setLevel(LOG_LEVEL)
        self.access_log.setLevel(LOG_LEVEL)


class StandaloneApplication(BaseApplication):
    """Our Gunicorn application."""

    def __init__(self, app, options=None):
        self.options = options or {}
        self.application = app
        super().__init__()

    def load_config(self):
        config = {
            key: value for key, value in self.options.items()
            if key in self.cfg.settings and value is not None
        }
        for key, value in config.items():
            self.cfg.set(key.lower(), value)

    def load(self):
        return self.application


if __name__ == '__main__':
    intercept_handler = InterceptHandler()
    # logging.basicConfig(handlers=[intercept_handler], level=LOG_LEVEL)
    # logging.root.handlers = [intercept_handler]
    logging.root.setLevel(LOG_LEVEL)

    seen = set()
    for name in [
        *logging.root.manager.loggerDict.keys(),
        "gunicorn",
        "gunicorn.access",
        "gunicorn.error",
        "uvicorn",
        "uvicorn.access",
        "uvicorn.error",
    ]:
        if name not in seen:
            seen.add(name.split(".")[0])
            logging.getLogger(name).handlers = [intercept_handler]

    logger.configure(handlers=[{"sink": sys.stdout, "serialize": JSON_LOGS}])

    options = {
        "bind": "0.0.0.0",
        "workers": WORKERS,
        "accesslog": "-",
        "errorlog": "-",
        "worker_class": "uvicorn.workers.UvicornWorker",
        "logger_class": StubbedGunicornLogger
    }

    StandaloneApplication(app, options).run()

급하신 분들은 복사해서 붙여넣기 하시고 마지막에 Gunicorn 옵션을 변경하시고,
그리고 그것을 시도!

그렇지 않은 경우 아래에서 각 부분에 대해 설명하겠습니다.

import os
import logging
import sys

from gunicorn.app.base import BaseApplication
from gunicorn.glogging import Logger
from loguru import logger

이 부분은 간단합니다. 필요한 것을 가져오기만 하면 됩니다.
GunicornBaseApplication을 사용하여 이 스크립트에서 직접 Gunicorn을 실행할 수 있습니다.
그리고 그 Logger는 우리가 약간 재정의할 것입니다.
나중에 코드에서 Loguru을 사용하고 있습니다.
예쁜 로그 형식을 갖거나 직렬화할 수 있습니다.

from my_app.app import app

내 프로젝트에는 my_app 모듈이 있는 app 패키지가 있습니다.
내 FastAPI 애플리케이션은 다음과 같이 이 모듈에서 선언됩니다.app = FastAPI() .

LOG_LEVEL = logging.getLevelName(os.environ.get("LOG_LEVEL", "DEBUG"))
JSON_LOGS = True if os.environ.get("JSON_LOGS", "0") == "1" else False
WORKERS = int(os.environ.get("GUNICORN_WORKERS", "5"))

개발 대 개발에 유용한 환경 변수에서 일부 값을 설정합니다.
생산 설정. JSON_LOGS는 로그를 JSON으로 직렬화해야 하는지 여부를 알려줍니다.
그리고 WORKERS는 우리가 갖고 싶은 일꾼의 수를 알려줍니다.

class InterceptHandler(logging.Handler):
    def emit(self, record):
        # Get corresponding Loguru level if it exists
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        # Find caller from where originated the logged message
        frame, depth = logging.currentframe(), 2
        while frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1

        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())

이 코드는 Loguru의 문서에서 복사하여 붙여넣었습니다!
이 핸들러는 라이브러리에서 내보낸 로그를 가로채는 데 사용됩니다.
Loguru를 통해 재방출합니다.

class StubbedGunicornLogger(Logger):
    def setup(self, cfg):
        handler = logging.NullHandler()
        self.error_logger = logging.getLogger("gunicorn.error")
        self.error_logger.addHandler(handler)
        self.access_logger = logging.getLogger("gunicorn.access")
        self.access_logger.addHandler(handler)
        self.error_log.setLevel(LOG_LEVEL)
        self.access_log.setLevel(LOG_LEVEL)

이 코드는 여기에서 복사되었습니다.
GitHub comment
@dcosson . 감사!
Gunicorn의 자체 로깅 구성을 재정의할 수 있습니다.
따라서 로그를 나머지와 같이 형식화할 수 있습니다.

마지막 두 줄은 제거해도 아무 변화가 없기 때문에 확실하지 않습니다.
아직 풀리지 않은 파이썬 로깅에 대한 미스터리가 있습니다...

class StandaloneApplication(BaseApplication):
    """Our Gunicorn application."""

    def __init__(self, app, options=None):
        self.options = options or {}
        self.application = app
        super().__init__()

    def load_config(self):
        config = {
            key: value for key, value in self.options.items()
            if key in self.cfg.settings and value is not None
        }
        for key, value in config.items():
            self.cfg.set(key.lower(), value)

    def load(self):
        return self.application

이 코드는
Gunicorn's documentation .
실행할 수 있는 간단한 Gunicorn 애플리케이션을 선언합니다.
Gunicorn의 모든 옵션을 수락합니다.

if __name__ == '__main__':
    intercept_handler = InterceptHandler()
    # logging.basicConfig(handlers=[intercept_handler], level=LOG_LEVEL)
    # logging.root.handlers = [intercept_handler]
    logging.root.setLevel(LOG_LEVEL)

우리는 단순히 가로채기 핸들러를 인스턴스화합니다.
루트 로거에서 로그 수준을 설정합니다.

다시 한 번, 두 사람이 논평한 것처럼 이것이 정확히 어떻게 작동하는지 이해하지 못합니다.
라인은 결과에 영향을 미치지 않습니다. 시행착오를 많이 하고 끝내고
작동하는 무언가가 있지만 그 이유를 완전히 설명할 수는 없습니다.
여기서 아이디어는 루트 로거에 핸들러를 설정하여 가로채도록 하는 것입니다.
모든 것이지만 충분하지 않았습니다(로그가 모두 차단되지는 않음).

    seen = set()
    for name in [
        *logging.root.manager.loggerDict.keys(),
        "gunicorn",
        "gunicorn.access",
        "gunicorn.error",
        "uvicorn",
        "uvicorn.access",
        "uvicorn.error",
    ]:
        if name not in seen:
            seen.add(name.split(".")[0])
            logging.getLogger(name).handlers = [intercept_handler]

여기에서 우리는 라이브러리에 의해 선언된 모든 가능한 로거를 반복합니다.
차단 핸들러로 핸들러를 재정의합니다.
여기에서 실제로 모든 로거가 동일하게 작동하도록 구성합니다.

이해가 안되는 이유로 Gunicorn과 Uvicorn은 나타나지 않습니다.
루트 로거 관리자에 있으므로 목록에 하드코딩해야 합니다.

우리는 또한 부모에 대한 가로채기 핸들러 설정을 피하기 위해 세트를 사용합니다.
이미 구성된 로거의 경우, 그렇지 않으면 로그가
2회 이상 방출됩니다. 이 코드가 다음 수준을 처리할 수 있는지 잘 모르겠습니다.
2보다 깊은 중첩 로거.

    logger.configure(handlers=[{"sink": sys.stdout, "serialize": JSON_LOGS}])

여기에서 표준 출력에 기록하도록 Loguru를 구성합니다.
필요한 경우 로그를 직렬화합니다.

어느 시점에서 나는 또한 사용하고있었습니다 activation=[("", True)](see Loguru's docs ),
그러나 그것은 필요하지도 않은 것 같습니다.

    options = {
        "bind": "0.0.0.0",
        "workers": WORKERS,
        "accesslog": "-",
        "errorlog": "-",
        "worker_class": "uvicorn.workers.UvicornWorker",
        "logger_class": StubbedGunicornLogger
    }

    StandaloneApplication(app, options).run()

마지막으로 Gunicorn 옵션을 설정하고 배선을 하고,
그리고 우리의 응용 프로그램을 실행하십시오!


글쎄요, 저는 이 코드가 자랑스럽진 않지만 작동합니다!