I recently started playing with FastAPI
and HTTPX,
and I am deploying my app with Gunicorn and
Uvicorn workers.
But when serving, the logs from each component looks quite different
from the others. I want them to all look the same, so I can easily read them
or exploit them in something like Kibana.
After a lot of hours trying to understand
how Python logging works,
and how to override libraries' logging settings,
here is what I have...
A single run.py file!
I didn't want to split logging configuration, Gunicorn configuration,
and the rest of the code into multiple files, as it was harder to wrap
my head around it.
Everything is contained in this single file:
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()
If you are in a hurry, copy-paste it, change the Gunicorn options at the end,
and try it!
If you're not, I will explain each part below.
import os
import logging
import sys
from gunicorn.app.base import BaseApplication
from gunicorn.glogging import Logger
from loguru import logger
This part is easy, we simply import the things we need.
The Gunicorn BaseApplication so we can run Gunicorn directly from this script,
and its Logger that we will override a bit.
We are using Loguru later in the code,
to have a pretty log format, or to serialize them.
from my_app.app import app
In my project, I have a my_app package with an app module.
My FastAPI application is declared in this module, something like
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"))
We setup some values from environment variables, useful for development vs.
production setups. JSON_LOGS tells if we should serialize the logs to JSON,
and WORKERS tells how many workers we want to have.
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())
This code is copy-pasted from Loguru's documentation!
This handler will be used to intercept the logs emitted by libraries
and re-emit them through 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)
This code was copied from this
GitHub comment
by @dcosson. Thanks!
It will allow us to override Gunicorn's own logging configuration
so its logs can be formatted like the rest.
I'm not sure about the last two lines, as removing them doesn't change anything.
There are still mysteries about Python logging that I couldn't resolve...
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
This code is taken from
Gunicorn's documentation.
We declare a simple Gunicorn application that we will be able to run.
It accepts all Gunicorn's options.
if __name__ == '__main__':
intercept_handler = InterceptHandler()
# logging.basicConfig(handlers=[intercept_handler], level=LOG_LEVEL)
# logging.root.handlers = [intercept_handler]
logging.root.setLevel(LOG_LEVEL)
We simply instantiate our interception handler,
and set the log level on the root logger.
Once again, I fail to understand how this works exactly, as the two commented
lines have no impact on the result. I did a lot of trial and error and ended
up with something working, but I cannot entirely explain why.
The idea here was to set the handler on the root logger so it intercepts
everything, but it was not enough (logs were not all intercepted).
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]
Here we iterate on all the possible loggers declared by libraries
to override their handlers with our interception handler.
This is where we actually configure every logger to behave the same.
For a reason that I fail to understand, Gunicorn and Uvicorn do not appear
in the root logger manager, so we have to hardcode them in the list.
We also use a set to avoid setting the interception handler on the parent
of a logger that is already configured, because otherwise logs would be
emitted twice or more. I'm not sure this code can handle levels of
nested loggers deeper than two.
logger.configure(handlers=[{"sink": sys.stdout, "serialize": JSON_LOGS}])
Here we configure Loguru to write on the standard output,
and to serialize logs if needed.
At some point I was also using activation=[("", True)]
(see Loguru's docs),
but it seems it's not required either.
options = {
"bind": "0.0.0.0",
"workers": WORKERS,
"accesslog": "-",
"errorlog": "-",
"worker_class": "uvicorn.workers.UvicornWorker",
"logger_class": StubbedGunicornLogger
}
StandaloneApplication(app, options).run()
Finally, we set our Gunicorn options, wiring things up,
and run our application!
Well, I'm not really proud of this code, but it works!
Top comments (0)