Structured Logs for an Aggregator

One JSON object per line, a correlation id on every record, and nothing secret in the payload.

Syntaxlogger.info("charge accepted", extra={"amount_cents": 1999, "currency": "GBP"})

A log aggregator indexes fields, not sentences. If every line is a single JSON object you can ask for every record where status is 500 and tenant is a given id. If the line is prose, you are left with substring search over a terabyte.

One object per line

Subclass logging.Formatter and override format(record) to return json.dumps(...). record.getMessage() applies the % arguments for you. Per-event fields travel in the extra dictionary, whose keys are written straight onto LogRecord.__dict__.

A correlation id

Hold the request id in a contextvars.ContextVar (3.7) and copy it onto every record from a logging.Filter. Each asyncio task gets its own copy of the context, so concurrent requests cannot read each other's id — a module-level global makes no such promise.

Short form, long form

logger.exception(msg) is exactly logger.error(msg, exc_info=True). The module-level logging.info(...) is the root logger plus an implicit basicConfig() — acceptable in a script, wrong in a library, which should call logging.getLogger(__name__) and configure nothing. And logger.info("charged %s", amount) defers the formatting until a handler actually wants the line, which an f-string cannot do.

What never goes in

No tokens, no Authorization headers, no card numbers, no whole request bodies. Log the identifiers you can join on and nothing you would have to rotate.

print() loses because it carries no level, no logger name and no exception support, and because stdout is block-buffered when it is not a terminal: the lines explaining a crash are still in the buffer when the process dies.

Example

Try it yourself
Loading editor…
Press Run to execute the code.

When to use it

  • Search for every 500 from one tenant across fifty replicas without grepping raw text.
  • Follow a single customer request through the gateway, the service and the worker by one id.
  • Alert on a count of error-level records carrying a specific error code rather than on a text pattern.
  • Keep tokens and request bodies out of a log store that a wider team can read.

More examples

Wiring it to stdout in a real service

basicConfig with handlers= and force= is the whole configuration for a containerised service: one handler, stdout, no log files to rotate. Set PYTHONUNBUFFERED=1 in the image so the last lines before a crash are actually flushed.

Example · python
# logging_setup.py - imported once, before anything logs
import logging
import sys

from .json_formatter import JsonFormatter   # the formatter from the example above

def configure(level="INFO"):
    handler = logging.StreamHandler(sys.stdout)   # stdout: the container runtime collects it
    handler.setFormatter(JsonFormatter())
    logging.basicConfig(level=level, handlers=[handler], force=True)
    # force=True (3.8+) removes handlers a library installed behind your back

    # Third-party loggers are noisy at INFO and useless below WARNING.
    logging.getLogger("botocore").setLevel(logging.WARNING)
    logging.getLogger("urllib3").setLevel(logging.WARNING)

    # setLevel accepts the string straight from the environment (3.2+).
    logging.getLogger("myapp.sql").setLevel("DEBUG")

Threading the id through a request

Every log line emitted anywhere beneath this middleware, including inside a library, picks the id up through the filter without being passed it. The reset in finally matters on a worker that serves requests in sequence; a leaked id silently attributes one customer's errors to another.

Example · python
import uuid

from .logging_setup import request_id   # the ContextVar

async def correlation_middleware(request, call_next):
    # Trust an inbound id so one trace spans every service, but validate it:
    # an unbounded header value becomes an unbounded log field.
    incoming = request.headers.get("X-Request-Id", "")
    rid = incoming if incoming.isascii() and len(incoming) <= 64 else uuid.uuid4().hex

    token = request_id.set(rid)           # set() returns a Token
    try:
        response = await call_next(request)
        response.headers["X-Request-Id"] = rid   # hand it back for the client's logs
        return response
    finally:
        request_id.reset(token)           # reset in finally, or the value leaks to the
                                         # next request served by this worker

Redacting before the line is written

This is a backstop, not a policy: the real rule is never to pass a secret to a logging call. Attach it to the handler rather than a logger so records escalated from third-party libraries pass through it too, and keep the key list in one place that a reviewer can check.

Example · python
import logging
import re

SENSITIVE_KEYS = {"password", "token", "authorization", "api_key", "secret", "card_number"}
BEARER = re.compile(r"Bearer\s+[A-Za-z0-9._\-]+")

class Redact(logging.Filter):
    """Last line of defence. A filter on the handler sees every record from
    every logger, including the ones in dependencies you do not control."""

    def filter(self, record):
        for key in list(record.__dict__):
            if key.lower() in SENSITIVE_KEYS:
                record.__dict__[key] = "[redacted]"
        if isinstance(record.msg, str):
            record.msg = BEARER.sub("Bearer [redacted]", record.msg)
        return True

handler = logging.StreamHandler()
handler.addFilter(Redact())   # on the handler, not on one logger

Related

py quality loggingpy cloud healthpy cloud sdkpy cloud envpy cloud signals

Discussion

  • Be the first to comment on this lesson.