In high-scale, asynchronous Python applications, traditional flat logs quickly become unmanageable. Debugging distributed systems, correlating events across microservices, and identifying performance bottlenecks require machine-readable, contextualized logs. This post guides developers and data scientists building production Python services on how to implement advanced logging patterns that transform raw log lines into actionable insights.
Key Takeaways
- Adopt structured (JSON) logging for machine readability and easier integration with centralized logging systems.
- Utilize `ContextVar` in `asyncio` applications to inject dynamic, request-specific context (e.g., `correlation_id`) into every log record.
- Implement asynchronous logging using `logging.handlers.QueueHandler` to prevent log I/O from blocking the main event loop.
The Problem
Traditional logging methods in Python can lead to unstructured and unreadable logs, making it difficult to debug and monitor applications. This is especially true for asynchronous applications, where the main event loop can be blocked by logging operations.
Data and Sources
This post uses the Slack Engineering RSS Feed (`https://slack.engineering/feed/`) as a data source to demonstrate the logging techniques. The `feedparser` library is used to parse the RSS feed. Data accessed on 2024-09-16.
Direct links to the dataset, API, official docs, or reference material:
- Slack Engineering RSS Feed: `https://slack.engineering/feed/`
- `feedparser` documentation: `https://feedparser.readthedocs.io/en/latest/`
- Python `logging` module documentation: `https://docs.python.org/3/library/logging.html`
- Python `ContextVar` documentation: `https://docs.python.org/3/library/contextvars.html`
Loading the Data
To load the data, we use the `feedparser` library to parse the RSS feed.
import feedparser
feed = feedparser.parse('https://slack.engineering/feed/')
for entry in feed.entries[:5]:
print(entry.title, entry.link)
Step 1 — Architecting Structured Logging with `dictConfig` and JSON
This step introduces a declarative way to configure logging and output logs in a consistent, machine-readable JSON format.
import logging.config
import json
logging_config = {
'version': 1,
'formatters': {
'json': {
'()': 'pythonjsonlogger.jsonlogger.JsonFormatter',
'format': '%(asctime)s %(levelname)s %(message)s'
}
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'json'
}
},
'root': {
'level': 'INFO',
'handlers': ['console']
}
}
logging.config.dictConfig(logging_config)
logger = logging.getLogger()
logger.info('This is an info message', extra={'correlation_id': '12345'})
Step 2 — Injecting Asynchronous Context with `ContextVar` and Custom Filters
This step shows how to propagate contextual data like `correlation_id` without explicit parameter passing.
import asyncio
from contextvars import ContextVar
correlation_id = ContextVar('correlation_id')
class CorrelationIdFilter(logging.Filter):
def filter(self, record):
record.correlation_id = correlation_id.get()
return True
logger.addFilter(CorrelationIdFilter())
async def main():
correlation_id.set('67890')
logger.info('This is an info message')
asyncio.run(main())
Step 3 — Preventing Blockage: Asynchronous Log Handling with `QueueHandler`
This step offloads log processing to a separate thread.
import logging.handlers
import queue
log_queue = queue.Queue()
logging_config = {
'version': 1,
'formatters': {
'json': {
'()': 'pythonjsonlogger.jsonlogger.JsonFormatter',
'format': '%(asctime)s %(levelname)s %(message)s'
}
},
'handlers': {
'queue': {
'class': 'logging.handlers.QueueHandler',
'queue': log_queue
}
},
'root': {
'level': 'INFO',
'handlers': ['queue']
}
}
logging.config.dictConfig(logging_config)
def log_listener():
while True:
record = log_queue.get()
if record is None:
break
logger = logging.getLogger(record.name)
logger.handle(record)
log_queue.task_done()
async def main():
logger.info('This is an info message')
asyncio.run(main())
Step 4 — Robust API Interaction Logging and Error Handling
This step demonstrates how to log structured messages before fetching, after successful parsing, and upon failure.
import feedparser
import logging
async def fetch_feed(feed_url):
try:
feed = feedparser.parse(feed_url)
logger.info('Feed fetched successfully', extra={'feed_url': feed_url, 'num_entries': len(feed.entries)})
return feed
except Exception as e:
logger.error('Error fetching feed', exc_info=True, extra={'feed_url': feed_url, 'error': str(e)})
async def main():
feed_url = 'https://slack.engineering/feed/'
feed = await fetch_feed(feed_url)
asyncio.run(main())
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
import asyncio
import feedparser
import json
import logging
import logging.config
import logging.handlers
import queue
from contextvars import ContextVar
from pythonjsonlogger import jsonlogger
# Define the logging configuration
logging_config = {
'version': 1,
'formatters': {
'json': {
'()': 'pythonjsonlogger.jsonlogger.JsonFormatter',
'format': '%(asctime)s %(levelname)s %(message)s'
}
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'json'
},
'queue': {
'class': 'logging.handlers.QueueHandler',
'queue': queue.Queue()
}
},
'root': {
'level': 'INFO',
'handlers': ['queue']
}
}
# Define the correlation ID context variable
correlation_id = ContextVar('correlation_id')
# Define the custom filter for correlation ID
class CorrelationIdFilter(logging.Filter):
def filter(self, record):
record.correlation_id = correlation_id.get()
return True
# Configure the logging
logging.config.dictConfig(logging_config)
logger = logging.getLogger()
logger.addFilter(CorrelationIdFilter())
# Define the log listener
def log_listener():
while True:
record = queue.Queue().get()
if record is None:
break
logger = logging.getLogger(record.name)
logger.handle(record)
queue.Queue().task_done()
# Define the asynchronous main function
async def main():
# Set the correlation ID
correlation_id.set('12345')
# Log an info message
logger.info('This is an info message')
# Fetch the feed
feed_url = 'https://slack.engineering/feed/'
try:
feed = feedparser.parse(feed_url)
logger.info('Feed fetched successfully', extra={'feed_url': feed_url, 'num_entries': len(feed.entries)})
except Exception as e:
logger.error('Error fetching feed', exc_info=True, extra={'feed_url': feed_url, 'error': str(e)})
# Run the main function
asyncio.run(main())
Expected Output
The expected output will be a JSON-formatted log message with the correlation ID, feed URL, and number of entries.
Limitations and Tradeoffs
The approach presented in this post has several limitations and tradeoffs:
- Increased complexity in initial setup compared to basic logging.
- Potential for over-logging if not managed, leading to increased storage costs and performance overhead.
- Security implications of logging sensitive data (PII, credentials) and the need for careful redaction or exclusion.
Frequently Asked Questions
What's the performance impact of structured and asynchronous logging?
The performance impact of structured and asynchronous logging is minimal, as the logging operations are offloaded to a separate thread. However, the overhead of serializing log messages to JSON can be significant for very high-volume applications.
How do I ensure sensitive data isn't logged in production?
To ensure sensitive data isn't logged in production, you can use a combination of techniques such as log filtering, redaction, and exclusion. You can also use a logging framework that supports sensitive data handling, such as the `pythonjsonlogger` library.
Can I use this approach with non-async Python applications (e.g., Flask/Django)?
Yes, you can use this approach with non-async Python applications. However, you will need to use a different logging handler, such as the `logging.handlers.RotatingFileHandler`, to handle the logging operations.
What I'd Change
In a production environment, I would consider using a more robust logging framework, such as the `structlog` library, to handle the logging operations. I would also implement a more comprehensive logging strategy, including log rotation, retention, and monitoring, to ensure that the logging system is scalable and reliable.
