Emit logs in JSON format from fastapi/gunicorn, for DataDog

import logging

from pythonjsonlogger import jsonlogger

def init() -> None:
    log_in_json()
    map_levelname_to_status()

# Force all loggers to talk JSON rather than text so DataDog can parse the output.
def log_in_json() -> None:
    loggers = [
        logging.getLogger("uvicorn.access"),
        logging.getLogger("uvicorn.error"),
        logging.getLogger("uvicorn"),
        logging.getLogger(),
    ]
    for logger in loggers:
        for handler in logger.handlers:
            logger.removeHandler(handler)
        logger.level = logging.DEBUG
        log_handler = logging.StreamHandler()
        formatter = jsonlogger.JsonFormatter(
            "%(asctime)s %(levelname)s %(name)s %(message)s"
        )
        log_handler.setFormatter(formatter)
        logger.addHandler(log_handler)

# DataDog is expecting 'status' for log level but the python default is 'levelname'.
def map_levelname_to_status() -> None:
    old_factory = logging.getLogRecordFactory()

    def record_factory(*args: str, **kwargs: str) -> logging.LogRecord:
        record = old_factory(*args, **kwargs)
        record.status = record.levelname  # type: ignore
        return record

    logging.setLogRecordFactory(record_factory)

Leave a Reply

Your email address will not be published.