lesson 236 Observability & Operations · advanced

Write logs a program can read

A log line like `loaded 2025-03-03-00.jsonl: 41 events, 1 rejected` is easy for a person and hard for a program. Structured logs write each record as one JSON object with fixed fields plus context, so the logs themselves can be searched, filtered and counted.

You'll be able to

Explain what structured logging gives you that text log messages don't
Subclass logging.Formatter to emit one JSON object per record, including fields passed with extra=
Record an exception's type in the JSON, and parse log lines back to filter them

The idea

In the Debugging & Logging module you logged messages with a level and a logger name, formatted as text. Text works while a person reads the log. In production, logs from every server and job land in one place, and the questions are asked by programs: *how many lines were rejected per file today? which errors happened for repository X?* Answering that from sentences means fragile pattern matching that breaks when someone rewords a message.

Structured logging keeps the message short and fixed ("hour loaded") and puts the variable details in named fields (file, events, rejected). Each record is written as one JSON object per line (the same JSON Lines you read in the CSV JSON & JSONL module), so any tool can parse it.

In Python, the shape of a log line is decided by a formatter. Subclass logging.Formatter and override format(record) to return the JSON text. Context goes in with extra={...}: logging copies each key onto the record as an attribute, and the formatter picks up every attribute a plain record doesn't have.

Run each example to see what it prints. Running examples doesn't count toward progress.

example-1.py · A JSON formatter

import json
import logging
import sys

STANDARD = set(vars(logging.LogRecord("x", logging.INFO, "x", 1, "x", None, None))) | {"message"}

class JsonFormatter(logging.Formatter):
    def format(self, record):
        entry = {"level": record.levelname, "logger": record.name, "message": record.getMessage()}
        for key, value in vars(record).items():
            if key not in STANDARD:
                entry[key] = value
        return json.dumps(entry)

log = logging.getLogger("ingest")
log.handlers.clear()
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
log.addHandler(handler)
log.setLevel(logging.INFO)
log.propagate = False

log.info("hour loaded", extra={"file": "2025-03-03-00.jsonl", "events": 41, "rejected": 1})
log.warning("line rejected", extra={"file": "2025-03-03-00.jsonl", "line": 3, "reason": "Expecting ',' delimiter"})
log.debug("not shown: below the logger's level")

extra keys become record attributes, so they must not collide with the names logging already uses:

example-2.py · An extra field that clashes with the record

import logging

log = logging.getLogger("ingest")
log.warning("line rejected", extra={"message": "bad JSON", "line": 3})

Failures need structure most. When a record carries an exception (from log.exception(...) inside except, or exc_info=True), put the exception's type in its own field, so errors can be counted by kind. Then the log is data: read it back line by line with json.loads and filter.

example-3.py · Errors as fields, and logs read back as data

import io
import json
import logging

class JsonFormatter(logging.Formatter):
    def format(self, record):
        entry = {"level": record.levelname, "message": record.getMessage()}
        entry["file"] = getattr(record, "file", None)
        if record.exc_info:
            entry["error"] = type(record.exc_info[1]).__name__
        return json.dumps(entry)

stream = io.StringIO()
handler = logging.StreamHandler(stream)
handler.setFormatter(JsonFormatter())
log = logging.getLogger("backfill")
log.handlers.clear()
log.addHandler(handler)
log.propagate = False

for name, text in [("00.jsonl", "[]"), ("01.jsonl", "{bad"), ("02.jsonl", "[]"), ("03.jsonl", "")]:
    try:
        json.loads(text)
        log.warning("file read", extra={"file": name})
    except json.JSONDecodeError:
        log.exception("file unreadable", extra={"file": name})

entries = [json.loads(line) for line in stream.getvalue().splitlines()]
failed = [e["file"] for e in entries if e["level"] == "ERROR"]
print(len(entries), "records;", "failed files:", failed)
print(entries[1])

When to use it: for any program whose logs are read by tools: services, scheduled jobs, pipelines. Mistakes to avoid: variable details baked into the message (f"loaded {file}") instead of fields; building JSON by gluing strings (a quote in a value breaks it; use json.dumps); extra keys that clash with record attributes; and logging secrets in a field (see Secure Python).

Review

Structured logs: a fixed message plus named fields, one JSON object per line.
A logging.Formatter subclass decides the line's shape; extra= adds fields (not named like record attributes).
Put an exception's type in its own field, and read logs back with json.loads to answer questions.
Next: logs record events one by one. Metrics summarise them: counts, rates and percentiles.

Practise this