posts/unify-logging-for-a-gunicorn-uvicorn-app/

[utterances-bot]

Unify Python logging for a Gunicorn/Uvicorn/FastAPI application - pawamoy's website

Findings, thoughts, tutorials, work. Pieces of my mind!

https://pawamoy.github.io/posts/unify-logging-for-a-gunicorn-uvicorn-app/

[ashokc]

Very useful write up. I am using pretty much your set up to generate a consistent looking logs. Thank you.

Couple of comments.

'error_log' should be 'error_logger' and the 'LOG_LEVEL' will stick.

self.error_logger.setLevel(GUNICORN_LOG_LEVEL) self.access_logger.setLevel(GUNICORN_LOG_LEVEL)

And we can have different levels (APP_LOG_LEVEL & GUNICORN_LOG_LEVEL) in order to get better control... say 'debug' for app vs 'info' for gunicorn.

First we comment out (in 'main')

# logging.root.setLevel(APP_LOG_LEVEL)

just so we do not get unwanted extra logs from 3rd party packages...

app_logger = logging.getLogger('app') app_logger.setLevel(APP_LOG_LEVEL)

[pawamoy]

Hi @ashokc, thanks!

'error_log' should be 'error_logger' and the 'LOG_LEVEL' will stick.

Ha, thank you very much!

just so we do not get unwanted extra logs from 3rd party packages...

Of course. In my case I did want the extra logs from all 3rd party packages :smile:

[trallnag]

Your blogpost helped me a lot, thanks. Though it's interesting that for me the Uvicorn handlers all show up in logging.root.manager.loggerDict.keys(). I don't have to declare them manually. Here are my dependencies:

[tool.poetry.dependencies]
python = "^3.9"
loguru = "^0.5.3"
prettyprinter = "^0.18.0"
fastapi = "^0.61.1"
requests = "^2.24.0"
dynaconf = {extras = ["yaml"], version = "^3.1.2"}
uvicorn = "^0.12.2"

[trallnag]

*And I don't use Gunicorn workers

[pawamoy]

Oh, uvicorn 0.12.2, the latest version, I guess they have fixed their logging config so the loggers are children of the root one (I was following such an issue on their GitHub repo).

Also yeah, I got rid of gunicorn myself, and the config is much lighter then :slightly_smiling_face:

I think I'll update my post to reflect this. Maybe I'll also add the version without gunicorn!

Anyway, glad it helped you!

[trallnag]

Btw what did the trick for me in the end was a combination of setting the interception handler from your blog post as the only Händler in logging.basicConfig and passing / overriding the default LOGGING_CONFIG Uvicorn is using with a dict that contains all the loggers Uvicorn relies on (uvicorn.error and so on) while removing the handlers from it. So it's logs get propagated up to the root logger and the interception handler.

Without the second part everything by Uvicorn was logged twice for me (once in Uvicorn default format and then again through the interception handler) even after looping through all loggers and setting the handlers to the interception handler like in your blog post.

If of interest, here are the two relevant py files:

https://github.com/trallnag/prometheus-webhook-proxy/blob/master/prometheus_webhook_proxy/main.py

https://github.com/trallnag/prometheus-webhook-proxy/blob/master/prometheus_webhook_proxy/config/logger.py

[ghost]

To support more than 2 levels of nesting for loggers, I defined the following:

def get_ancestors(logger_name):
    segments = logger_name.split(".")
    return [".".join(segments[0:n+1]) for n in range(len(segments))]

and changed the handler setting to this:

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

Note that "gunicorn" and "uvicorn" are no longer in the list since their children are already there.

[pawamoy]

Thanks for sharing your solution @syk0saje!

To everyone: I've added a Uvicorn-only version to the post.

[ketanpatil3]

Tiny bit optimisation in while loop use substring for comparison[11:15] than using whole string. subStr = logging.file[11:15] amazing if we could remove the while loop :)

[pawamoy]

Hi @ketanpatil3, thanks! I'm not sure to get it though. Do you mean the while loop can be removed in favor of your snippet? In any case, you might want to chat with @Delgan as this code comes directly from his loguru project's documentation :slightly_smiling_face:

[ketanpatil3]

Tiny change in the code. You don't have to compare the whole string. Tiny portion is enough:

substr = logging.__file__[11:15]
while frame.f_code.co_filename[11:15] == substr

[pawamoy]

Arf, wanted to edit your comment to properly format the code, but email replies do not support MarkDown... sorry :sweat_smile: And thanks for the explanation.

[ketanpatil3]

Hi, Do you know how to contextualize information? extra = {'app':'hello','name':'myname'} logger.configure(extra=extra) #using loguru logger mylog = logging.getLogger("gunicorn.error") mylog.info('testing') mylog.error('testing error')

I get the below output 2021-02-06 21:25:03.086 | INFO | main:run_app:125 - testing 2021-02-06 21:25:03.086 | ERROR | main:run_app:126 - testing error How to make it extra information in the output? I need to change the format for the above and add {extra}

On Sat, Feb 6, 2021 at 6:22 PM Timothée Mazzucotelli < notifications@github.com> wrote:

Arf, wanted to edit your comment to properly format the code, but email replies do not support MarkDown... sorry 😅 And thanks for the explanation.

— You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub https://github.com/pawamoy/pawamoy.github.io/issues/17#issuecomment-774473028, or unsubscribe https://github.com/notifications/unsubscribe-auth/AEH6WJ2JKZ2KBZWJQXFBXKDS5U3PBANCNFSM4RKYCJQQ .

[Delgan]

Hi @pawamoy and @ketanpatil3. :)

I'm not sure of what the substr = logging.__file__[11:15] is mean to represent. The idea behind the while loop is to find the calling frame which happens to be the frame just before internal functions of builtin logging module. This is very similar to what is done in CPython actually.

It's safer to compare the whole path rather than trying to optimize comparison with a substring, because others unrelated files may actually have the same name, technically. Also, I think the slicing operation may actually be slower than comparing the whole string.

[pawamoy]

Also, I think the slicing operation may actually be slower than comparing the whole string.

I was thinking the same thing when trying to understand @ketanpatil3's suggestion. Thanks for chiming in @Delgan :slightly_smiling_face:

@ketanpatil3 about your question on how to contextualize information, you'll have better luck asking on loguru's issue tracker, I'm not an expert here :slightly_smiling_face: @Delgan do you have a gitter or something for your project? Maybe you could enable discussions as well.

[ketanpatil3]

@pawamoy Thank you for your wonderful code. I never thought replying to mail would end up posting comments here. Apologies . It is up to individuals to pick and choose what works(coz with my approach it might introduce new bug). I tried timeit and found comparing 64 characters is slower than slicing and comparing 4 characters(difference is negligible 6.5%). Posting this for info to others: For contextualising you have to add format parameter to logger.config and add metadata(dict) to extra parameter in logger.config.

[Delgan]

I don't have any gitter, but anyone should feel free to open a new regular issue if they need help or have suggestions. :+1:

[sandys]

this is very cool! how do I add log format ? for example, I want the log format to come as "{'remote_ip':'%(h)s','request_id':'%({X-Request-Id}i)s','response_code':'%(s)s','request_method':'%(m)s','request_path':'%(U)s','request_querystring':'%(q)s','request_timetaken':'%(D)s','response_length':'%(B)s'}"

[gooalkhan]

hello, I found that your code for uvicorn only version works only when the auto-reloading and workers options are disabled. When the mentioned 2 options are enabled, they quite don't work well as you suggest. how did you solve this problem with multiple workers?

https://replit.com/@gooalkhan/logger-uvicorn

[pawamoy]

Hey @gooalkhan! Unfortunately I'm guilty of always using only one worker, because I deploy my apps in a Kubernetes cluster which handles scaling, therefore I did not need more than one worker per app yet. In fact, I didn't know this was an issue! I'll try to find some time to replicate and see if I can find a solution :confused: If you find one, please share!

[jaytimbadia]

Hey, While configuring logs to file, I created a new handler config, but it seems that file is not getting created.

logger.configure(handlers=[{"sink": sys.stdout, "serialize": True}, {"sink": 'runtime.log', "serialize": True, "level": logging.DEBUG}])

runtime.log is not getting created only, but logs are getting populated on console.

Can you please let me know the issue?

[pawamoy]

You'll have more luck asking on the loguru issue tracker directly! Make sure you have the permissions to write a file in the current working directory.

[jaytimbadia]

Thanks for the update. I was able to resolve it. Now logs are being populated in files in json format. Just that I want request time i.e. how long reauest takes to complete, too in the logs, but loguru do not have that in their format, gunicorn have that, but not sure how to integrate.

Can you please help?

[pawamoy]

Not sure, you'd have to check how gunicorn is logging, and if there's a way for you to add the request handling time into your log format.

[mahenzon]

Hi! Thanks, it helped me a lot. My solution to run it with workers / start reloader:

# loguru_intercept_handler.py

import logging
import sys

from loguru import logger

class InterceptHandler(logging.Handler):
    """
    Default handler from examples in loguru documentaion.
    See https://loguru.readthedocs.io/en/stable/overview.html#entirely-compatible-with-standard-logging
    """

    def emit(self, record: logging.LogRecord):
        # 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(log_level: int, json_logs: bool):
    # 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
    # noinspection PyUnresolvedReferences
    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}])

# ...
# settings.py (or settings obj)
LOG_LEVEL = "DEBUG"  # (or 10 if `logging.DEBUG`)
# custom handlers removed, we catch logs via loguru
UVICORN_LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "default": {
            "()": "uvicorn.logging.DefaultFormatter",
            "fmt": "%(levelprefix)s %(message)s",
            "use_colors": None,
        },
        "access": {
            "()": "uvicorn.logging.AccessFormatter",
            "fmt": '%(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s',
        },
    },
    "loggers": {
        "uvicorn": {"level": "INFO"},
        "uvicorn.error": {"level": "INFO"},
        "uvicorn.access": {"level": "INFO", "propagate": False},
    },
}

# ...
# main.py

import uvicorn

import settings
from loguru_intercept_handler import setup_logging
from app import app

__all__ = (
    "app",
    "main",
)

# run it on init, not in main, because main is not called when run from str
setup_logging(settings.LOG_LEVEL, settings.JSON_LOGS)

def main():
    uvicorn.run(
        "main:app",
        log_config=settings.UVICORN_LOGGING_CONFIG,
        log_level=settings.LOG_LEVEL,
        debug=settings.DEBUG,
        host=settings.HOST,
        port=settings.PORT,
    )

if __name__ == "__main__":
    main()

[pawamoy]

Hey @mahenzon, sorry for the delay, and thanks for posting your solution! It will be useful to others, and probably myself as well :slightly_smiling_face:

[Shiny21]

Hi @pawamoy, excellent article. Is there a way to capture http request object in the logs with loguru? I am actually using google stackdriver for log storage of different applications and hence want to maintain common pattern.

[pawamoy]

Hi @Shiny21, thank you for the kind words :slightly_smiling_face: Since every log is intercepted by the InterceptHandler, including the uvicorn ones (potentially containing extra fields), I guess you can do additional processing with the record in its emit method? You can also checkout loguru's docs to see how to configure additional "sinks" or behaviors. Otherwise you can maybe add a middleware to your FastAPI application that will do some processing with the requests, without going through the logging system.

[obrua]

Hi! Thanks, it helped me a lot. My solution to run it with workers / start reloader:

# loguru_intercept_handler.py

import logging
import sys

from loguru import logger

class InterceptHandler(logging.Handler):
    """
    Default handler from examples in loguru documentaion.
    See https://loguru.readthedocs.io/en/stable/overview.html#entirely-compatible-with-standard-logging
    """

    def emit(self, record: logging.LogRecord):
        # 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(log_level: int, json_logs: bool):
    # 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
    # noinspection PyUnresolvedReferences
    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}])

# ...
# settings.py (or settings obj)
LOG_LEVEL = "DEBUG"  # (or 10 if `logging.DEBUG`)
# custom handlers removed, we catch logs via loguru
UVICORN_LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "default": {
            "()": "uvicorn.logging.DefaultFormatter",
            "fmt": "%(levelprefix)s %(message)s",
            "use_colors": None,
        },
        "access": {
            "()": "uvicorn.logging.AccessFormatter",
            "fmt": '%(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s',
        },
    },
    "loggers": {
        "uvicorn": {"level": "INFO"},
        "uvicorn.error": {"level": "INFO"},
        "uvicorn.access": {"level": "INFO", "propagate": False},
    },
}

# ...
# main.py

import uvicorn

import settings
from loguru_intercept_handler import setup_logging
from app import app

__all__ = (
    "app",
    "main",
)

# run it on init, not in main, because main is not called when run from str
setup_logging(settings.LOG_LEVEL, settings.JSON_LOGS)

def main():
    uvicorn.run(
        "main:app",
        log_config=settings.UVICORN_LOGGING_CONFIG,
        log_level=settings.LOG_LEVEL,
        debug=settings.DEBUG,
        host=settings.HOST,
        port=settings.PORT,
    )

if __name__ == "__main__":
    main()

Your code helped me a lot, but I encountered a problem. The parameters log_level and access_log of uvicorn do not work in this case!

[mahenzon]

Hi @obrua Yeah, you need to set it in the dict config UVICORN_LOGGING_CONFIG

[Winand]

@pawamoy you mention that you got rid of gunicorn. But in Uvicorn Deployment they say: "Run uvicorn ... for local development. Run gunicorn ... for production." Do you use uvicorn's workers? I'm trying to write my first web service with fastapi (containerized) and i'm stuck with logging now. It works fine with uvicorn but I fail to intercept gunicorn's logs started from terminal (gunicorn -k uvicorn.workers.UvicornWorker). So it's essential to start gunicorn from Python script?

[pawamoy]

At work we are deploying our apps in a Kubernetes cluster. The scaling is done using pods, therefore I only need 1 uvicorn worker per pod/container. If you are deploying your app on a bare server, then gunicorn might help, but it is not required.

[Winand]

This code works for me for both Uvicorn(0.15) and Gunicorn(20.1)+Uvicorn.

def init(lvl: str="INFO"):
    logger.configure(handlers=[
        {"sink": sys.stderr, "format": log_format, "level": lvl}
    ])
    intercept_handler = InterceptHandler()
    # level=NOTSET - loguru controls actual level
    logging.basicConfig(handlers=[intercept_handler], level=logging.NOTSET)
    for name in logging.root.manager.loggerDict.keys():
        _logger = logging.getLogger(name)
        if _logger.name.startswith("gunicorn"):
            _logger.handlers = [intercept_handler]
        else:
            # By default uvicorn.access has a handler and doesn't propagate
            # (uvicorn.access controls INFO messages on requests)
            _logger.propagate = True
            _logger.handlers = []

But it doesn't catch exceptions during main module loading. I have to use logger.catch in certain places, where an exception may happen during startup.

[esteban-gs]

I'm using the Uvicorn option only. This is a stupid question, but how do I actually log?

[pawamoy]

You can log things with standard logging.

https://docs.python.org/3/howto/logging.html#logging-basic-tutorial

In your library code:

import logging

logger = logging.getLogger(__name__)
logger.debug(msg)
logger.info(msg)
logger.warning(msg)
logger.error(msg)
logger.critical(msg)

Only, ONLY in the command line entry point (the highest possible):

import logging

logging.basicConfig(filename='example.log', encoding='utf-8', level=logging.DEBUG)
# more logging configuration

---

You can also log things with third-party libs, like loguru:

https://loguru.readthedocs.io/en/stable/overview.html#suitable-for-scripts-and-libraries

from loguru import logger

logger.debug(msg)
logger.info(msg)
logger.warning(msg)
logger.error(msg)
logger.critical(msg)

[AlyanQ]

Hey everyone, I was having a lot of problems logging properly with gunicorn + uvicorn workers but this guide helped me out a ton. One problem I'm having currently is that after changing the errorlog and accesslog to direct to logfiles, the logs aren't being written to the logfiles, they just show up in the terminal. I don't mind having the logs in the terminal but having them stored in a logfile would be nice. Anyone have any solutions? I followed the guide above exactly (for gunicorn+uvicorn)

[AlyanQ]

Nevermind, figured it out. Leaving it here for anyone else who faces the same issue. You must change the line logger.configure(handlers=[{"sink": 'sys.stdout', "serialize": JSON_LOGS}]) to logger.configure(handlers=[{"sink": 'YOUR_LOGFILE, "serialize": JSON_LOGS}])

[GroverChouT]

I figured out that I can combine your two solutions for gunicorn and uvicorn into one:

class GunicornLogger(Logger):
    def setup(self, cfg) -> None:
        handler = InterceptHandler()

        # Add log handler to logger and set log level
        self.error_log.addHandler(handler)
        self.error_log.setLevel(settings.LOG_LEVEL)
        self.access_log.addHandler(handler)
        self.access_log.setLevel(settings.LOG_LEVEL)

        # Configure logger before gunicorn starts logging
        logger.configure(handlers=[{"sink": sys.stdout, "level": settings.LOG_LEVEL}])

def configure_logger() -> None:
    logging.root.handlers = [InterceptHandler()]
    logging.root.setLevel(settings.LOG_LEVEL)

    # Remove all log handlers and propagate to root logger
    for name in logging.root.manager.loggerDict.keys():
        logging.getLogger(name).handlers = []
        logging.getLogger(name).propagate = True

    # Configure logger (again) if gunicorn is not used
    logger.configure(handlers=[{"sink": sys.stdout, "level": settings.LOG_LEVEL}])

Call configure_logger() at very top:

configure_logger()
app = FastAPI()

And use following command line to start it up:

# gunicorn
gunicorn some.package.main:app -k uvicorn.workers.UvicornWorker --logger-class some.package.logger.GunicornLogger

# uvicorn
uvicorn some.package.main:app

Thanks!

[pawamoy]

Nice, thanks for sharing!

[Winand]

🤔my example above worked for me without creating additional class and passing it to --logger-class