JSON Logging Inside a Flask Application: configuration for producing JSON logs under a Flask app running under gunicorn

flask_app.py

"""
flask application
"""
import logging
import logging.config
import os

from flask import Flask
import structlog

from . import constants
from . import loggers


__version__ = "0.0.1"
ENVIRONMENT = os.getenv("ENVIRONMENT")


structlog.configure(
    processors=[
        loggers.add_app_name,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt=constants.LOGGING_TS_FORMAT),
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
        structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
        # structlog.processors.JSONRenderer()  # Not necessary because it gets formatted at a higher level
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

if ENVIRONMENT is not None and ENVIRONMENT == constants.LOCAL:  # pragma: no cover
    # This is completely unnecessary: just fancy coloring for local development logging.
    # It also fails to properly format one of the Flask default messages so you'll see
    # a Traceback for that message on startup!
    logging.config.dictConfig({
        "version": 1,
        "disable_existing_loggers": True,
        "formatters": {
            "colors": {
                "()": structlog.stdlib.ProcessorFormatter,
                "processor": structlog.dev.ConsoleRenderer(colors=True),
            }
        },
        "handlers": {
            "default": {
                "level": "INFO",
                "class": "logging.StreamHandler",
                "formatter": "colors",
            }
        },
        "loggers": {
            "": {
                "handlers": ["default"],
                "level": "INFO",
                "propagate": True,
            },
        }
    })

def create_app(package_name, 
               config_path=None, start_msg="API started", use_log_handlers="gunicorn.error", 
               settings_override=None, env=None):
    app = Flask(package_name, instance_relative_config=True)
    
    app.config.from_object(config_path)
    app.config.from_object(settings_override)  # useful for testing
    
    requested_logger = logging.getLogger(use_log_handlers)
    app.logger.handlers = requested_logger.handlers[:]
    
    # here's how you use the `structlog` logger for this application
    # first, import `logger` from this module.
    # Then, call `log = logger.new()` to create a new log instance
    log = logger.new()
    # after that, you can pass messages and arbitrary kwargs
    log.info(start_msg, arbitrary_key="some value that will appear in the JSON")
    return app

gunicorn_logging.conf

[loggers]
keys=root, gunicorn.error, gunicorn.access, scripts

[handlers]
keys=error_file, access_file, scripts_file

[formatters]
keys=json

[logger_root]
level=INFO
handlers=access_file

[logger_gunicorn.access]
level=INFO
handlers=access_file
propagate=0
qualname=gunicorn.access

[logger_gunicorn.error]
level=ERROR
handlers=error_file
propagate=0
qualname=gunicorn.error

[logger_scripts]
level=INFO
handlers=scripts_file
qualname=scripts

# Change Location if running this locally
[handler_access_file]
class=logging.handlers.WatchedFileHandler
formatter=json
args=('/some_server_location/my_flask_app_gunicorn_access_log.json',)

# Change Location if running this locally
[handler_error_file]
class=logging.handlers.WatchedFileHandler
formatter=json
args=('/some_server_location/my_flask_app_gunicorn_error_log.json',)

# Change Location if running this locally
[handler_scripts_file]
class=logging.handlers.WatchedFileHandler
formatter=json
args=('/some_server_location/my_flask_app_scripts.json',)

[formatter_json]
class=my_app.loggers.JsonLogFormatter

loggers.py

import datetime

from pythonjsonlogger import jsonlogger

from . import constants


def add_app_name(logger, log_method, event_dict):  # pragma: no cover
    event_dict["application"] = constants.APPLICATION_NAME
    return event_dict


class JsonLogFormatter(jsonlogger.JsonFormatter):  # pragma: no cover
    def add_fields(self, log_record, record, message_dict):
        """
        This method allows us to inject custom data into resulting log messages
        """
        for field in self._required_fields:
            log_record[field] = record.__dict__.get(field)
        log_record.update(message_dict)

        # Add timestamp and application name if not present
        if "timestamp" not in log_record:
            now = datetime.datetime.utcnow()
            log_record["timestamp"] = datetime.datetime.strftime(now, format=constants.LOGGING_TS_FORMAT)
        if "application" not in log_record:
            log_record["application"] = constants.APPLICATION_NAME

        jsonlogger.merge_record_extra(record, log_record, reserved=self._skip_fields)

requirements.txt

flask
gunicorn
json-logging
structlog

run.sh

gunicorn -b 0.0.0.0:8000 -w 4 --log-config gunicorn_logging.conf runner:my_wsgi_flask_app

thanks! please consider adding example constants.py and a main.py to let the people run and check your solution in seconds