Structured logging assigns stable meaning to fields instead of asking each query to infer everything from message text. A record might contain an event name, request identifier, numeric duration, severity, and a human-readable message. Free text remains useful and can also be searched or parsed. The practical gain is that routine questions can use a documented schema whose meaning survives changes to explanatory wording.

That schema needs more than curly braces. A duration stored as a number supports numeric comparisons, while a duration stored as a string may be indexed or compared differently. Milliseconds and seconds must not share an ambiguous field name. An event name should describe what happened consistently across versions. A schema version can identify deliberate format changes, but it does not automatically transform old records or keep every consumer compatible.

Put context on the correct record

A request identifier is useful only if it belongs to the operation that emitted the record. Concurrent requests make a process-global mutable variable unsuitable for that purpose. Python context variables can hold task-local state across asynchronous work, provided values are set and reset at the intended scope. This does not mean every thread or custom execution mechanism inherits context in the same way; verify the actual boundary.

Capture context while it is available. This example formats each record synchronously in the emitting task, so its formatter can read the current context variable. A queue-based logging pipeline needs a different placement: attach the required fields before handing the record to another thread or process. Reading the listener’s current request context would associate the record with the listener, rather than reliably recovering the original request.

When it breaks

Hand-concatenating JSON can turn quotes or line breaks from a message into invalid records. A line-oriented collector may then split one application event into several apparent events. Use a JSON serializer and one complete JSON object per physical line. The message can retain a logical newline inside its string value while the serialized line contains an escaped representation. Rendering that message later should still keep its record boundary clear.

Structured output does not automatically create indexes, reduce query bills, or guarantee delivery. Unbounded fields, accidental type changes, missing identifiers, and sensitive values can make a JSON stream difficult or inappropriate to use. Select fields according to real queries, keep the schema stable, and verify the collector’s parser and retention rules. Searchable structure is a contract between producers and consumers, not a property established by the file extension.

Check it quickly

Save this as structured-logging.py and run it with Python; the executed version was 3.14.4. It uses only standard-library modules. Two asynchronous requests each log a start event, wait at a shared gate, then log a completion event containing a deliberately troublesome message. The durations are fixed synthetic values for checking JSON types, not measurements of request performance. Output is captured in memory for assertions.

import asyncio
import io
import json
import logging
from collections import Counter
from contextvars import ContextVar

request_id = ContextVar("request_id", default=None)
stream = io.StringIO()


class JsonFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps({"schema_version": 1, "event": record.event,
                           "request_id": request_id.get(), "worker": record.worker,
                           "duration_ms": record.duration_ms,
                           "level": record.levelname, "message": record.getMessage()},
                          ensure_ascii=False, separators=(",", ":"))


logger = logging.getLogger("structured-log-lab")
logger.setLevel(logging.INFO)
logger.propagate = False
handler = logging.StreamHandler(stream)
handler.setFormatter(JsonFormatter())
logger.addHandler(handler)


async def request(name, ready, release):
    token = request_id.set(name)
    try:
        logger.info("starting %s", name, extra={"event": "request.started",
                    "worker": name, "duration_ms": 0})
        ready.set()
        await release.wait()
        await asyncio.sleep(0)
        logger.info('untrusted line\n{"event":"forged"}',
                    extra={"event": "request.completed", "worker": name, "duration_ms": 4})
    finally:
        request_id.reset(token)


async def main():
    ready = [asyncio.Event(), asyncio.Event()]
    release = asyncio.Event()
    tasks = [asyncio.create_task(request(name, signal, release))
             for name, signal in zip(("req-a", "req-b"), ready)]
    await asyncio.gather(*(signal.wait() for signal in ready))
    release.set()
    await asyncio.gather(*tasks)
    assert request_id.get() is None


try:
    asyncio.run(main())
finally:
    logger.removeHandler(handler)
    handler.close()
lines = stream.getvalue().splitlines()
records = [json.loads(line) for line in lines]
assert len(lines) == 4
assert Counter(record["request_id"] for record in records) == {"req-a": 2, "req-b": 2}
assert all(record["request_id"] == record["worker"] for record in records)
assert all(type(record["duration_ms"]) is int for record in records)
completed = [record for record in records if record["event"] == "request.completed"]
assert len(completed) == 2 and all("\n" in record["message"] for record in completed)
assert not any(record["event"] == "forged" for record in records)
print(json.dumps({"physical_lines": len(lines), "parsed_records": len(records),
                  "requests": 2, "context_mixing": False,
                  "numeric_duration_preserved": True, "forged_records": 0}, indent=2))

The assertions require four physical lines and four parsed records, with two records for each request identifier. Each record’s identifier must match the worker that emitted it, even though the requests overlap. Duration values remain JSON numbers. The embedded newline survives inside each decoded completion message, while no additional record named forged appears. Resetting each token also leaves the enclosing context without a request identifier.

This verifies task isolation, synchronous formatting, and line-safe serialization. It does not exercise a background queue, collector retry, storage indexing, or secret redaction. When integrating it, check the complete path with representative fields and the actual concurrency model. Inspect what the backend stores after parsing, and test the queries that motivated those fields. A successful producer-side serialization check is one useful boundary test, while a dependable logging system also needs evidence that the intended records reach their consumers with their meaning intact.

Further reading

Sources

  • Python logging cookbook, contextual records and logging across concurrent work, checked 2026-09-13.
  • Python contextvars, task-local context and token reset behavior, checked 2026-09-13.
  • Python json, serialization, escaped characters, and JSON value types, checked 2026-09-13.
  • JSON Lines, one complete JSON value per physical line, checked 2026-09-13.