Python Logging — 12-Hour Outage from a Missing Flush

Three hours into a Sev-1 incident, I was staring at an empty log file. The payment service was returning 500s. Users were double-charged. The CEO was in the Slack thread. And the logs showed nothing — just a clean file with the last entry from 2 AM. The service had been crash-looping since 6 AM.

We spent four hours rebuilding the stack, adding metrics, chasing phantom DB connection leaks. Eventually, a junior engineer noticed the RotatingFileHandler constructor had delay=True. The default. Every crash during log rotation was silently dropping the buffer. The logs existed. They were just sitting in a 256KB kernel buffer that the OOM killer never flushed.

This is not a tutorial about logging levels. This is a postmortem of every logging failure I've seen in 8 years of running Python in production: the flush that never happened, the JSON formatter that threw on unicode, the async app that blocked on logging.info(), and the Kubernetes pod that lost all its logs on restart because someone used a file handler in a container.

By the end, you will have a logging config that survives process kills, rotation races, and traffic spikes. And you will never trust a log file again.

The Config That Won't Lie to You

Most logging tutorials show you logging.basicConfig() and call it done. That config lies. It writes to stderr by default. It uses a format string that gives you zero context — no file name, no line number, no thread. And it's a one-shot call: any import that calls basicConfig() before your main() silently consumes the first call, and yours becomes a no-op.

Here's the config that has survived three production incidents I've debugged:

Rule 1: Never call basicConfig(). Use dictConfig() from the start. It gives you explicit control over every handler, formatter, and logger hierarchy. If you import a library that calls basicConfig(), your dictConfig() overrides it.

Rule 2: Every FileHandler must set delay=False. The default delay=True defers file creation until the first log write. That means if your process crashes between startup and the first log line, no file is created and no error is raised. You think your logging is working. It's not.

Rule 3: Send ERROR and above to a separate handler. In production, you want a high-volume INFO stream for debugging and a low-noise ERROR stream for alerting. If you parse the same file for both, your alerting pipeline reads through megabytes of INFO noise to find ERROR needles.

Rule 4: Never rotate logs with Python's RotatingFileHandler. It renames the file under an open file descriptor — writes continue to the renamed file until the next rotation. After a crash and restart, your logs are split across app.log, app.log.1, and app.log.2 with no way to tell which is current. Use WatchedFileHandler + system logrotate.

import logging.config

LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "json": {
            "()": "app.logging.JsonFormatter",
            "fmt": "%(asctime)s %(name)s %(levelname)s %(message)s",
        },
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "json",
            "stream": "ext://sys.stdout",
        },
        "error_file": {
            "class": "logging.handlers.WatchedFileHandler",
            "level": "ERROR",
            "formatter": "json",
            "filename": "/var/log/app/error.log",
            "delay": False,
        },
    },
    "root": {
        "level": "INFO",
        "handlers": ["console", "error_file"],
    },
    "loggers": {
        "healthcheck": {
            "level": "WARNING",
            "propagate": False,
        },
    },
}

logging.config.dictConfig(LOGGING)

# At shutdown: force flush
import atexit

@atexit.register
def _flush_logs():
    for h in logging.getLogger().handlers:
        h.flush()

Structured JSON Logging: The 3 AM Difference

Plain text logs are grep-hell when you have 12 services, each with its own format, correlating a single user request across 4 hops. Structured JSON logging means every log line is a key-value dictionary that your logging platform can index, filter, and aggregate.

The critical detail most tutorials miss: your formatter must handle every Python type without throwing. A datetime object in a log message, a bytes value, a Decimal from your payment library — any of these will crash the standard json.dumps(). When the formatter throws, the logging module silently catches the exception and calls the default formatter. You get a raw string instead of JSON, and your log aggregation pipeline rejects it silently because it's not valid JSON. You have gaps. You don't know you have gaps.

Here's the fix: default=str in json.dumps(). It converts every non-serializable type to its string representation. The log line is valid JSON. The Decimal shows up as "19.99" instead of causing an empty line. Test this with every type your app touches.

Second thing: always include request_id, user_id, and trace_id in every log line. In a FastAPI app, inject these via middleware. In a Celery task, extract from the task context. Without these, you cannot correlate an ERROR log in the payment service with the WARNING log in the order service that preceded it.

import logging
import json
from datetime import datetime


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        obj = {
            "timestamp": datetime.fromtimestamp(record.created).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "module": record.module,
            "line": record.lineno,
            "message": record.getMessage(),
        }
        if record.exc_info and record.exc_info[0]:
            obj["exception"] = {
                "type": record.exc_info[0].__name__,
                "message": str(record.exc_info[1]),
            }
        if hasattr(record, "request_id"):
            obj["request_id"] = record.request_id
        return json.dumps(obj, default=str)


# Usage in dictConfig:
# "formatters": {
#     "json": {
#         "()": "app.logging.JsonFormatter",
#     },
# }

# Adding context via LoggerAdapter
class RequestLogger(logging.LoggerAdapter):
    def process(self, msg, kwargs):
        kwargs.setdefault("extra", {})
        kwargs["extra"].setdefault("request_id", self.extra["request_id"])
        return msg, kwargs

logger = RequestLogger(logging.getLogger(__name__), {"request_id": "req-abc-123"})
logger.info("Payment processed")

Log Rotation Without Losing Data

Python's RotatingFileHandler looks convenient: 4 lines of code and you get automatic rotation. Here's why it's dangerous in production:

When RotatingFileHandler rotates, it renames the current file (e.g., app.log → app.log.1) and creates a new app.log. But Python's file descriptor still points to the old inode — the renamed file. Until doRollover() is called again, all log writes go to app.log.1, not app.log. After a process restart, Python opens app.log fresh. The latest logs are in app.log.1. The current file app.log is hours stale.

The fix: use WatchedFileHandler with system logrotate. WatchedFileHandler periodically stat()s the file's inode. If the inode changes (because logrotate renamed the file), it closes the old fd and opens the new file. This is the correct behavior: the external tool handles rotation, Python just follows the fd.

Configure `logrotate`: `` /var/log/app/*.log { daily rotate 30 compress delaycompress missingok notifempty copytruncate } ` copytruncate` is critical: it copies the file then truncates the original. No rename. No inode change. Python keeps writing to the same fd. Zero log loss.

# /etc/logrotate.d/app
/var/log/app/*.log {
    daily
    rotate 30
    compress
    delaycompress
    missingok
    notifempty
    copytruncate
    postrotate
        # Notify WatchedFileHandler via SIGHUP (optional)
        # kill -HUP $(cat /var/run/app.pid) 2>/dev/null || true
    endscript
}

# Verify rotation worked:
# $ sudo logrotate -d /etc/logrotate.d/app     # dry run
# $ sudo logrotate -f /etc/logrotate.d/app     # force rotate
# $ ls -la /var/log/app/

Async Logging: Don't Block the Event Loop

Here's a bug I've seen in every async Python codebase I've audited: someone calls logging.info() inside a FastAPI route handler. That call triggers a synchronous write() to a file handler. The write blocks the event loop for 0.5-5ms. At 500 req/s, that's 250-2500ms of blocking per second. Your p95 latency doubles.

You have two options:

Option 1: QueueHandler + QueueListener (recommended) A QueueHandler puts log records onto a thread-safe queue. A QueueListener runs in a dedicated thread that drains the queue and writes to handlers. Your async code never blocks on I/O.

Option 2: Write to stdout only (simple, container-native) Inside containers, StreamHandler with stdout is already non-blocking enough for most apps (the kernel buffer handles the write). Only reach for QueueHandler when profiling shows logging.info() in your hot path.

Critical: configure the QueueListener with respect_handler_level=True. Without it, the listener thread processes records regardless of handler level filters, which can cause WARNING-level handlers to process DEBUG records.

import logging
import logging.handlers
import queue
import threading
from typing import Optional


def setup_async_logging(
    *,
    queue_size: int = 10000,
    level: int = logging.INFO,
    json_format: bool = True,
) -> logging.Logger:
    """Configure logging for async apps. Returns the root logger."""
    log_queue: queue.SimpleQueue = queue.SimpleQueue()
    queue_handler = logging.handlers.QueueHandler(log_queue)

    # Console handler — JSON format, goes to stdout
    console = logging.StreamHandler()
    console.setLevel(level)
    if json_format:
        console.setFormatter(
            logging.Formatter(
                '{"timestamp":"%(asctime)s","level":"%(levelname)s",'
                '"message":"%(message)s"}'
            )
        )
    else:
        console.setFormatter(
            logging.Formatter("%(asctime)s %(levelname)s %(message)s")
        )

    # Listener — runs in a daemon thread
    listener = logging.handlers.QueueListener(
        log_queue,
        console,
        respect_handler_level=True,
    )
    listener.start()

    root = logging.getLogger()
    root.addHandler(queue_handler)
    root.setLevel(level)

    # Store listener for graceful shutdown
    root._queue_listener = listener
    return root


# In your FastAPI lifespan:
# @asynccontextmanager
# async def lifespan(app):
#     logger = setup_async_logging()
#     yield
#     logger._queue_listener.stop()

# Usage — same logging API, zero blocking:
# logger = logging.getLogger(__name__)
# logger.info("Request received", extra={"path": "/payments"})

What to Log and What to Burn

The most common logging anti-pattern is logging everything at INFO. Every request, every DB query, every external API call. Your log volume explodes. You can't find anything. Your logging costs (CloudWatch, Datadog, ELK) hit $10k+/month. And worst of all: you miss real errors because they're drowning in INFO noise.

Here's the only strategy that works:

INFO: Log every business-relevant transition. Payment created, order shipped, user registered. Do NOT log every HTTP request body. Do NOT log every SQL query. If you need that for debugging, use DEBUG and enable it per-module.

WARNING: Log conditions that are unexpected but handled. Retry attempts, rate limit hits, stale cache fallback. These are signals for potential problems, not problems themselves.

ERROR: Log every exception that bubbles up to the service boundary. Every 5xx response. Every database connection failure. Every external API timeout. This is your alerting signal. If your ERROR log volume is >10/hour, you have a problem.

CRITICAL: Log conditions that require immediate human intervention. Data corruption detected, license expiry <7 days, security policy violation. These pages someone at 3 AM.

Rule: if an ERROR log doesn't trigger an alert, it's not ERROR-level. Move it to WARNING. Alert fatigue kills more Sev-1 responses than bad code.

import logging

logger = logging.getLogger(__name__)

# GOOD: business transition at INFO
logger.info("Payment captured", extra={"payment_id": "py_abc", "amount": 29.99})

# BAD: HTTP request noise at INFO
logger.info("GET /api/payments 200 OK")  # This should be DEBUG

# GOOD: handled retry at WARNING
logger.warning(
    "Payment gateway timeout, retry 2/3",
    extra={"payment_id": "py_abc", "retry_count": 2, "gateway": "stripe"},
)

# GOOD: real failure at ERROR
logger.error(
    "Payment gateway rejected with 402",
    extra={"payment_id": "py_abc", "gateway_response": "card_declined"},
    exc_info=True,
)

# BAD: catch-all with no context
logger.error("Something went wrong")  # Useless

# CORRECT: always include the context needed to reproduce
logger.exception(
    "Charge failed after 3 retries, sending to dead-letter queue",
    extra={"payment_id": "py_abc", "dlq": "payments.failed", "attempts": 3},
)

Testing Your Logging: The Only Way to Trust It

You test your database queries. You test your API endpoints. You test your business logic. But you don't test your logging. That's insane. Logging is your only window into production — and you're running untested code that writes it.

A logging test suite should cover:

1. Formatter doesn't crash on any type: datetime, Decimal, bytes, None, nested dicts, lists, custom objects. If json.dumps() throws, your log line is silently dropped.
2. Handler writes to the expected destination: File handler writes to the file. Stream handler writes to stdout. Syslog handler sends to the server. Test with tmp_path in pytest.
3. Log level filters work: A DEBUG log at the root logger with level=INFO produces zero output. An ERROR log at a child logger with level=WARNING propagates to root.
4. QueueHandler + QueueListener delivers records: Send a log, stop the listener, verify the record was written. This catches configuration errors where the listener never started.
5. Sensitive data is redacted: If you have a filter that redacts credit card numbers, write a test that proves it. This is a compliance requirement (PCI-DSS, SOC2).

Run these tests in CI. A broken logging config means you're flying blind in production.

import json
import logging
from datetime import datetime, date
from decimal import Decimal
from pathlib import Path

import pytest


class TestJsonFormatter:
    def test_handles_all_types(self):
        """Formatter must produce valid JSON for every Python type."""
        formatter = logging.Formatter(
            '{"msg": "%(message)s"}'
        )

        for value in [datetime.now(), date.today(), Decimal("19.99"),
                       b"bytes", None, {"nested": "dict"}, [1, 2, 3]]:
            record = logging.LogRecord(
                name="test", level=logging.INFO,
                pathname=__file__, lineno=42,
                msg=f"value={value}", args=(),
                exc_info=None,
            )
            result = formatter.format(record)
            parsed = json.loads(result)  # raises if invalid JSON
            assert "msg" in parsed

    def test_handles_exception_info(self):
        formatter = logging.Formatter(
            '{"msg": "%(message)s"}'
        )
        try:
            1 / 0
        except ZeroDivisionError:
            record = logging.LogRecord(
                name="test", level=logging.ERROR,
                pathname=__file__, lineno=42,
                msg="division error", args=(),
                exc_info=sys.exc_info(),
            )
        result = formatter.format(record)
        parsed = json.loads(result)
        assert isinstance(parsed, dict)

    def test_handler_writes_to_file(self, tmp_path: Path):
        log_file = tmp_path / "app.log"
        handler = logging.FileHandler(str(log_file), delay=False)
        logger = logging.getLogger("test_file")
        logger.addHandler(handler)
        logger.setLevel(logging.INFO)

        logger.info("test message")
        handler.flush()
        handler.close()

        content = log_file.read_text()
        assert "test message" in content

    def test_queue_handler_delivers(self, tmp_path: Path):
        """Verify QueueHandler+QueueListener actually writes."""
        import queue
        import logging.handlers

        log_file = tmp_path / "queue_test.log"
        file_handler = logging.FileHandler(str(log_file), delay=False)
        file_handler.setFormatter(
            logging.Formatter("%(message)s")
        )

        log_queue: queue.SimpleQueue = queue.SimpleQueue()
        listener = logging.handlers.QueueListener(
            log_queue, file_handler, respect_handler_level=True
        )
        listener.start()

        logger = logging.getLogger("test_queue")
        logger.addHandler(logging.handlers.QueueHandler(log_queue))
        logger.setLevel(logging.INFO)

        logger.info("queue delivered")
        listener.stop()

        assert "queue delivered" in log_file.read_text()