Logs

Logs are a very important part of any application, both for the developer who needs to understand which parts of the application have errors, and for the client and the business side for accountability purposes.

The best way to log an application is to use structured logs, preferably in JSON format, which allows for easy integration with a wide range of tools. This then allows for a series of automated analyses to be performed retrospectively:

• Performance analysis: identifying bottlenecks, execution time, number of API calls, etc.
• Error analysis: number of errors and their types.
• User accountability.
• Cost analysis: calculating computation costs based on the number of API calls and the type of API called. An API that triggers a data processing flow will have a higher cost impact than a simple database query triggered by a GET call.

At the same time, however, we can’t just print structured JSON logs to the console. A programmer running the program locally needs a brief, concise output on stdout that makes it immediately clear what’s happening in the application and guides them in resolving bugs.

So, to summarize, we would need:

• Structured, verbose logs that are easy to analyze automatically for retrospective statistics (perhaps to be sent to an aggregator with a dashboard: Grafana, Datadog, etc.).
• Concise, short, but clear logs for the console, to be used during development and debugging.

We can achieve both of these solutions using a Python package called “structlog”.

This package allows you to specify the logger’s behavior depending on whether it needs to write to stdout or to a file. In the first case, we’ll specify a simpler and more concise logging format; in the second case, we’ll specify a structured JSON format containing more information.

Installing structlog

pip install structlog python-json-logger

python-json-logger is the package needed to manage logs in Python. Structlog, in fact, is agnostic about the chosen structure for managing logs; it can work with any format you want. Personally, I prefer using the JSON format as it has broader support in third-party tools and is also easier for humans to read compared to other formats like XML, for example.

Logger Configuration

Let’s create a function to set up the logger.

Structlog’s logger works by applying a series of processors (transformations) in cascade to the input log string. At the end of this transformation pipeline, the resulting log is saved to a file or written to the console according to the given specifications. Let’s look at it step by step:

import logging
import sys
import structlog
import os
from pythonjsonlogger import jsonlogger

# structlog Processor Configuration
# These prepare data for both outputs
shared_processors = [
    structlog.contextvars.merge_contextvars,  # Should be the first processor. Adds the logger's local context to the log string.
    structlog.stdlib.add_logger_name,  # Adds the logger's name to the log string.
    structlog.stdlib.add_log_level,  # Adds the level (INFO, ERROR, ...).
    structlog.processors.TimeStamper(fmt="iso"), # Adds a timestamp in ISO format.
    structlog.processors.StackInfoRenderer(),  # Adds stack information. Useful for debugging errors.
    structlog.processors.format_exc_info,  # Adds messages generated by exceptions.
    structlog.stdlib.ProcessorFormatter.wrap_for_formatter # This should be added last. It builds the (logger, name, event_dict) tuple which will then be passed to the logging module's formatter.
]

structlog.configure(
    processors=shared_processors,
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

# Standard Logging Module Configuration
# Console Output
console_formatter = logging.Formatter(
	fmt="%(asctime)s - %(levelname)-8s - %(name)s - %(message)s",
	datefmt="%Y-%m-%d %H:%M:%S"
)
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(console_formatter)
console_handler.setLevel(logging.INFO) # We'll pass this later as input to the log setup function.

# File Output (Structured JSON)
# This formatter will serialize the entire LogRecord (enriched by structlog)
json_log_path = '/var/log/my-app/app1.log' # Example path, make sure the directory exists or is created
json_formatter = jsonlogger.JsonFormatter()
os.makedirs(os.path.dirname(json_log_path), exist_ok=True) # Create the log directory if it doesn't exist
json_file_handler = logging.FileHandler(json_log_path, mode='a') # Append mode
json_file_handler.setFormatter(json_formatter)
json_file_handler.setLevel(logging.INFO) # We'll pass this later as input to the log setup function.

Now let’s add the two loggers we just created to the root logger, which is the one we will call. It will, in turn, automatically call the logger for the console and the one in JSON format:

# Main Logger
root_logger = logging.getLogger()
root_logger.handlers.clear() # Clear any existing handlers
root_logger.addHandler(console_handler) # Add the console logger
root_logger.addHandler(json_file_handler) # Add the JSON format logger
root_logger.setLevel(logging.INFO) # Set the root logger level

Now, we might consider putting these code blocks into a simple function like the one shown below, to be called as the first instruction in our code, to simplify things each time:

def setup_logging(log_level: int = logging.INFO, json_log_path: str = "logs/structured_log.json"): # log_level should be an int
	# ... (all the setup code from above would go in here)
    pass # Placeholder for the actual setup code

In our code, it will then be sufficient to write:

setup_logging() # Setup the structlog package

# Get a logger
logger = structlog.get_logger(__name__)

# Logging examples
logger.debug("Debug message (not shown if level=INFO)")
logger.info("Application started", app_version="1.0.1", environment="production")
logger.warning("Warning: slow operation", duration_ms=1500)

ATTENTION: For applications deployed on Docker or Kubernetes, logs are generally written directly to standard output (stdout and stderr) rather than to files. This is because the Kubernetes or Docker daemon will then take the various logs from stdout and send them, if requested, to an external aggregator such as Splunk, Datadog, etc.

In these cases, therefore, using a dedicated log file is not necessary and, indeed, could only make things more complex.

Consequently, in these cases, the code is simply:

import logging
import sys
import structlog
# import os # os might not be needed if not writing to a file explicitly
from pythonjsonlogger import jsonlogger

# structlog Processor Configuration
# These prepare data for the output
shared_processors = [
    structlog.contextvars.merge_contextvars,  # Should be the first processor. Adds the logger's local context.
    structlog.stdlib.add_logger_name,  # Adds the logger's name.
    structlog.stdlib.add_log_level,  # Adds the log level.
    structlog.processors.TimeStamper(fmt="iso"), # ISO timestamp.
    structlog.processors.StackInfoRenderer(),  # Stack information for debugging.
    structlog.processors.format_exc_info,  # Exception messages.
    structlog.stdlib.ProcessorFormatter.wrap_for_formatter # Prepares for standard logging formatter.
]

structlog.configure(
    processors=shared_processors,
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

# Structured JSON Output to stdout
# This formatter will serialize the entire LogRecord (enriched by structlog)
json_formatter = jsonlogger.JsonFormatter()
stdout_json_handler = **logging.StreamHandler(sys.stdout)** # Changed to StreamHandler for stdout
stdout_json_handler.setFormatter(json_formatter)
stdout_json_handler.setLevel(logging.INFO) # We'll pass this later as input to the log setup function.

# Main Logger
root_logger = logging.getLogger()
root_logger.handlers.clear() # Clear any existing handlers
root_logger.addHandler(stdout_json_handler) # Add the JSON formatter for stdout
root_logger.setLevel(logging.INFO) # Set the root logger level