A logging handler can take down your application. Most homegrown ones eventually do.
Python’s logging module is well designed and sits in the standard library, and almost nobody uses more than logging.info(). This guide wires it up to a central server and covers the three things that matter in production: structured output, batching, and surviving an unreachable log server.
Make the logs worth shipping first. The default format gives you lines like this:
INFO:root:user 42 checked out cart 981 for $34.99
Now find every checkout over $30. You need a regex. Emit JSON instead and those numbers become fields.
Python 3.12 and later ship a JSON-capable formatter through logging.config. A small custom formatter works on any version and gives you more control:
import json
import logging
from datetime import datetime, timezone
class JSONFormatter(logging.Formatter):
"""Render log records as a single line of JSON."""
# Attributes LogRecord always has; anything else was passed via `extra`.
RESERVED = frozenset(logging.LogRecord("", 0, "", 0, "", (), None).__dict__)
def format(self, record: logging.LogRecord) -> str:
payload = {
"timestamp": datetime.fromtimestamp(
record.created, tz=timezone.utc
).isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
"module": record.module,
"line": record.lineno,
}
if record.exc_info:
payload["exception"] = self.formatException(record.exc_info)
# Merge anything passed as extra={...}
for key, value in record.__dict__.items():
if key not in self.RESERVED and not key.startswith("_"):
payload[key] = value
return json.dumps(payload, default=str)
Now extra becomes real structured data:
logger.info(
"checkout completed",
extra={"user_id": 42, "cart_id": 981, "amount_cents": 3499},
)
produces
{"timestamp":"2026-08-07T16:40:02.114+00:00","level":"INFO","logger":"shop.checkout","message":"checkout completed","module":"checkout","line":88,"user_id":42,"cart_id":981,"amount_cents":3499}
Keep default=str on json.dumps. Drop it and one stray datetime or Decimal in extra raises TypeError inside your logging call. Crashing a request handler on a log line is an annoying way to spend an afternoon.
Does your application log to stdout under systemd or Docker? Stop here. Let the platform collect stdout and forward it with the CL Agent or the journald driver. Less code, fewer failure modes than anything below.
import logging, sys
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler])
The rest of this guide covers shipping from inside the application. You need that for serverless, for desktop apps, and anywhere you do not control the host.
The naive handler makes one blocking HTTPS request per log line, on whatever thread called logger.info(). Do not write that. A slow endpoint becomes slow request handling. A hung endpoint becomes an outage.
Buffer in memory and flush from a background thread instead:
import atexit
import json
import logging
import queue
import threading
import urllib.request
import urllib.error
class CentralLoggingHandler(logging.Handler):
"""Buffer log records and POST them to Central Logging in batches.
Records are handed to a background thread, so logging never blocks the
caller. If the endpoint is unreachable, records are dropped rather than
accumulated without bound.
"""
def __init__(self, endpoint, token, batch_size=100,
flush_interval=5.0, max_queue=10_000, level=logging.NOTSET):
super().__init__(level)
self.url = f"{endpoint.rstrip('/')}/api/v1/ingest_logs/{token}"
self.batch_size = batch_size
self.flush_interval = flush_interval
self._queue = queue.Queue(maxsize=max_queue)
self._stopping = threading.Event()
self._worker = threading.Thread(
target=self._run, name="central-logging", daemon=True
)
self._worker.start()
atexit.register(self.close)
def emit(self, record):
try:
self._queue.put_nowait(self.format(record))
except queue.Full:
# The endpoint is down or we are producing faster than we can
# ship. Dropping is the right call: the alternative is unbounded
# memory growth in the application process.
pass
except Exception:
self.handleError(record)
def _run(self):
batch = []
while not self._stopping.is_set():
try:
batch.append(self._queue.get(timeout=self.flush_interval))
except queue.Empty:
pass
if len(batch) >= self.batch_size or (
batch and self._queue.empty()
):
self._send(batch)
batch = []
# Drain whatever is left at shutdown.
while True:
try:
batch.append(self._queue.get_nowait())
except queue.Empty:
break
if batch:
self._send(batch)
def _send(self, batch):
body = "\n".join(batch).encode("utf-8")
req = urllib.request.Request(
self.url, data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
try:
urllib.request.urlopen(req, timeout=10).close()
except Exception:
# Never raise out of the logging path. Losing logs is bad;
# crashing the application because logging failed is worse.
pass
def close(self):
if self._stopping.is_set():
return
self._stopping.set()
self._worker.join(timeout=self.flush_interval + 5)
super().close()
Wire it up:
import logging
handler = CentralLoggingHandler(
endpoint="https://logs.example.com",
token="YOUR-SOURCE-TOKEN",
)
handler.setFormatter(JSONFormatter())
logging.basicConfig(level=logging.INFO, handlers=[
handler,
logging.StreamHandler(), # keep local output too
])
Three decisions in that handler are the ones people get wrong.
Bounded queue. max_queue=10_000 makes the application drop records when the log server goes away. The alternative is a heap that grows until the OOM killer arrives. Your observability tool must never kill the process it observes.
Swallowed exceptions in _send. Logging must not raise. An endpoint returning 500 is a problem for later. It is not an exception for a request handler that only wanted to log a line.
Daemon thread plus atexit. The daemon flag keeps the thread from blocking interpreter shutdown. The atexit hook lets it drain the queue first. Skip the hook and every clean exit loses its last few seconds of logs — often the ones explaining why you exited.
In settings.py:
LOGGING = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"json": {"()": "myapp.logging.JSONFormatter"},
},
"handlers": {
"central": {
"()": "myapp.logging.CentralLoggingHandler",
"endpoint": "https://logs.example.com",
"token": os.environ["CL_SOURCE_TOKEN"],
"formatter": "json",
"level": "INFO",
},
"console": {
"class": "logging.StreamHandler",
"formatter": "json",
},
},
"root": {"handlers": ["central", "console"], "level": "INFO"},
"loggers": {
"django.request": {"level": "WARNING", "propagate": True},
},
}
Read the token from the environment. It is a credential, and settings.py lives in version control.
import logging
from flask import Flask, request, g
app = Flask(__name__)
handler = CentralLoggingHandler(endpoint=..., token=...)
handler.setFormatter(JSONFormatter())
app.logger.addHandler(handler)
@app.after_request
def log_request(response):
app.logger.info("request", extra={
"method": request.method,
"path": request.path,
"status": response.status_code,
"remote_addr": request.remote_addr,
})
return response
The logs are JSON, so every field you attached with extra answers to SQL:
SELECT
"timestamp",
json_extract(msg, '$.level') AS level,
json_extract(msg, '$.logger') AS logger,
json_extract(msg, '$.message') AS message,
json_extract(msg, '$.user_id') AS user_id
FROM logs
WHERE json_valid(msg)
AND json_extract(msg, '$.level') IN ('ERROR', 'CRITICAL')
ORDER BY "timestamp" DESC
LIMIT 200;
Which exceptions are most common:
SELECT
substr(json_extract(msg, '$.exception'), 1, 120) AS exc,
COUNT(*) AS occurrences
FROM logs
WHERE json_valid(msg)
AND json_extract(msg, '$.exception') IS NOT NULL
GROUP BY exc
ORDER BY occurrences DESC
LIMIT 25;
Per-user error counts. This is how you find the one customer hitting a broken code path:
SELECT
json_extract(msg, '$.user_id') AS user_id,
COUNT(*) AS errors
FROM logs
WHERE json_valid(msg)
AND json_extract(msg, '$.level') = 'ERROR'
AND json_extract(msg, '$.user_id') IS NOT NULL
GROUP BY user_id
ORDER BY errors DESC
LIMIT 20;
Structured logging makes it easy to dump a whole object into your log store. Log storage is a bad home for personal data and a worse one for credentials. Three habits:
extra. Logging request headers? Filter Authorization and Cookie first.user_id: 42 helps you. The full user row with an email and a home address is a liability.logger.exception() in code that touches payment or auth data. Some configurations capture local variables in the traceback.A retention window on the log source limits how long any of it survives. See Configuration.
Emit JSON so extra turns into queryable fields.
Does your platform already collect stdout? Use it and skip the custom handler.
Shipping directly? Buffer in a background thread, bound the queue, and never let a logging failure raise.
Central Logging is a single binary that takes these logs over HTTP and queries them with SQL. See Deploy.
💌 Get notified on new features and updates