Uvicorn: JsonLogging format

[rspadim]

Hi folks, could be possible a Json format logger? i'm using elasticsearch to log data, and json is very intuitive
there's a logger but i don't know how to use it with uvicorn:

from pythonjsonlogger.jsonlogger import JsonFormatter
JsonFormatter("(asctime) (levelname) (name) (module) (funcName) (lineno) (message)")

[erewok]

This is something that I do in all of my projects. Currently, I am running gunicorn with the uvicorn worker and passing a custom logfile like so:

$ gunicorn example:app -w 2 -k uvicorn.workers.UvicornWorker --log-config logging_config.conf

The logging_config.conf looks like this:

[loggers]
keys=root, uvicorn, gunicorn

[handlers]
keys=access_handler

[formatters]
keys=json

[logger_root]
level=INFO
handlers=access_handler
propagate=1

[logger_gunicorn]
level=INFO
handlers=access_handler
propagate=0
qualname=gunicorn

[logger_uvicorn]
level=INFO
handlers=access_handler
propagate=0
qualname=uvicorn

[handler_access_handler]
class=logging.StreamHandler
formatter=json
args=()

[formatter_json]
class=pythonjsonlogger.jsonlogger.JsonFormatter

Note that for high throughput, the uvicorn docs mention turning off access logging, which I have not done here.

[rspadim]

thanks a lot @erewok!

[rspadim]

@erewok about the last comment: "he uvicorn docs mention turning off access logging"
could you provide a example, just to anyone that read this know what it should be?

thanks again!

[pawamoy]

I myself am running uvicorn programatically, and serializing in JSON with loguru. Everything is contained in one Python script:

import os
import logging
import sys

from uvicorn import Config, Server
from loguru import logger

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


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())


def setup_logging():
    # intercept everything at the root logger
    logging.root.handlers = [InterceptHandler()]
    logging.root.setLevel(LOG_LEVEL)

    # remove every other logger's handlers
    # and propagate to root logger
    for name in logging.root.manager.loggerDict.keys():
        logging.getLogger(name).handlers = []
        logging.getLogger(name).propagate = True

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


if __name__ == '__main__':
    server = Server(Config("my_project.main:app", log_level=LOG_LEVEL))

    # setup logging last, to make sure no library overwrites it
    # (they shouldn't, but it happens)
    setup_logging()

    server.run()

Export a JSON_LOGS=1 environment variable to have the logs serialized in JSON.

It's working well in my Linux VM, unfortunately logs are still duplicated (one line as JSON, one line as text) on OpenShift... Still not sure why.

EDIT: ah, I just needed to upgrade my uvicorn version. It requires at least 0.11.6.
Thanks uvicorn developers!

[florimondmanca]

Hello all, since it looks like this is mostly a logging configuration question and that some solutions were provided above, I'm going to close this off for housekeeping purposes. Thanks!