Structured Logging for Distributed Systems: Schema, IDs and Levels

What structured logging is and why free-text logs fail across services

Structured logging means emitting each event as a record of named fields instead of a sentence. The usual form is one JSON object per line: a timestamp, a level, the service name, a message, and whatever fields describe the event, such as order_id or duration_ms. A machine can read every field directly, so nothing has to be pulled back out of prose with a regex.

Free-text logs are fine for a single process that one person tails. They fail once a request crosses service boundaries. Each team picks its own timestamp format and its own wording for the same failure, and parsers written against one service's lines break when someone rewords a message. Worse, the lines carry nothing that ties them together. You cannot reliably collect every event produced by one checkout request when it touched an API gateway, a payment service and a queue consumer.

So treat the log format as a contract between teams, not as a private choice inside one codebase. Field names, types and meanings have to agree across producers so that one query works everywhere. The teams that gain the most are those running many services owned by different groups, where nobody can read every codebase to work out what a log line means.

Copyimport json
import logging
import sys
from datetime import datetime, timezone

SERVICE = "payment-service"  # every producer sets its own name

class JsonFormatter(logging.Formatter):
    """Render each record as one JSON object per line."""
    def format(self, record):
        event = {
            # Shared contract fields: same names and types in every service
            "ts": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(),
            "level": record.levelname.lower(),
            "service": SERVICE,
            "msg": record.getMessage(),
        }
        # Event-specific fields arrive as a dict, never baked into the message
        event.update(getattr(record, "fields", {}))
        return json.dumps(event)

handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
log = logging.getLogger(SERVICE)
log.addHandler(handler)
log.setLevel(logging.INFO)

# Free-text version: readable, but order_id and duration need a regex to extract
# log.info("Charged order ord_8812 in 182ms for request req_3f9a")

# Structured version: the same facts as queryable fields
log.info("payment captured", extra={"fields": {
    "request_id": "req_3f9a",  # same ID the gateway and queue consumer log
    "order_id": "ord_8812",
    "duration_ms": 182,
}})
# -> {"ts": "...", "level": "info", "service": "payment-service",
#     "msg": "payment captured", "request_id": "req_3f9a",
#     "order_id": "ord_8812", "duration_ms": 182}

A minimal Python standard-library JSON formatter that turns a free-text log line into a one-object-per-line record with shared contract fields and a request_id that other services can join on.

Defining a minimal shared schema for JSON logs

Keep the required set small enough that every service can emit it from day one. Six fields cover most cross-service queries:

• timestamp: UTC, RFC 3339 with a trailing Z and fixed millisecond precision (2025-03-14T09:26:53.120Z). Local offsets and mixed precision make sorting events from different hosts unreliable.
• level: one lowercase value from a closed list, so filters do not have to match WARN, Warning and warn.
• service and env: they tell you which deployment produced the line without relying on hostnames or collector metadata.
• message: a short, constant description of the event.
• trace_id and span_id (or correlation_id): the keys that join events across process boundaries.

Extra fields need two rules. First, pick one naming style, such as snake_case with dotted namespaces (http.status_code, user_id), so that userId and user_id never both exist and split every query in two. Second, record each field's type in a shared registry. If order_id is a string in one service and an integer in another, many log backends will reject the event, coerce the value or create a conflicting mapping. Put units in the name (duration_ms, size_bytes) so nobody has to guess.

The difference is clearest in a single event:

With a constant message, the event can be grouped exactly. With typed fields, you can filter on any value.

Copyimport json
import logging
import sys
from datetime import datetime, timezone

SERVICE = "payments-api"
ENV = "production"
# Closed, lowercase level vocabulary shared by every service
LEVELS = {"DEBUG": "debug", "INFO": "info", "WARNING": "warn",
          "ERROR": "error", "CRITICAL": "fatal"}

class JsonFormatter(logging.Formatter):
    def format(self, record):
        # Extra fields first, so they can never overwrite the required ones
        event = dict(getattr(record, "fields", {}))
        event.update({
            # UTC, RFC 3339, fixed millisecond precision, trailing Z
            "timestamp": datetime.fromtimestamp(record.created, tz=timezone.utc)
                .isoformat(timespec="milliseconds").replace("+00:00", "Z"),
            "level": LEVELS.get(record.levelname, "info"),
            "service": SERVICE,
            "env": ENV,
            "message": record.getMessage(),
        })
        return json.dumps(event, separators=(",", ":"))

handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
log = logging.getLogger("payments")
log.addHandler(handler)
log.setLevel(logging.INFO)

# Constant message; variables live in typed snake_case fields with units in names
log.error("payment authorization failed", extra={"fields": {
    "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
    "span_id": "00f067aa0ba902b7",
    "order_id": "8812",   # always a string, per the shared field registry
    "duration_ms": 340,    # always an integer
    "currency": "EUR",
}})

A standard-library Python JSON formatter that emits the six required fields in a fixed format and keeps the message constant, with variables passed as typed snake_case fields.

How correlation IDs flow through HTTP, queues and background jobs

The correlation ID should be created once, at the first component you control: the load balancer, API gateway or edge service. If an inbound request already carries an ID from a trusted caller, reuse it. Otherwise, generate a random one. Every downstream hop copies it forward:

• HTTP: a request header, such as the conventional X-Request-ID or the W3C traceparent header.
• Queues: message attributes or headers. Don't put it in the payload body, because consumers shouldn't have to parse business data to find it.
• Background jobs: a field in the job's stored arguments, so it survives until a worker picks the job up.

Developers shouldn't pass the ID through function signatures by hand. Middleware should read it at the boundary and put it in request-scoped or context-local storage. The logger then reads from that storage when it writes each line. This way every log line gets correlation_id automatically, including lines from library code that has never heard of it.

If you run distributed tracing, the trace ID is effectively your correlation ID, and the span ID identifies one unit of work within it. Log both. With trace_id on every line, an engineer can open a slow trace and jump to its exact log lines, or start from an error log and open the full trace.

Propagation tends to break without any error at these boundaries:

• Async queues: the producer doesn't copy context into the message metadata.
• Thread pools and executors: tasks run on threads that don't inherit the caller's context-local storage.
• Scheduled jobs: there is no inbound request, so the job has to create its own ID.
• Third-party calls: external services drop your headers, so log the ID on the outgoing call and on the response.

Copyimport contextvars
import logging
import uuid
from concurrent.futures import ThreadPoolExecutor

# Context-local storage for the current request's correlation ID.
correlation_id = contextvars.ContextVar("correlation_id", default="-")

class CorrelationFilter(logging.Filter):
    # Stamp every record, including records from library loggers, with the current ID.
    def filter(self, record):
        record.correlation_id = correlation_id.get()
        return True

handler = logging.StreamHandler()
handler.addFilter(CorrelationFilter())
handler.setFormatter(logging.Formatter("%(levelname)s correlation_id=%(correlation_id)s %(name)s %(message)s"))
logging.basicConfig(level=logging.INFO, handlers=[handler])
log = logging.getLogger("orders")

def charge_card():
    # No ID parameter: the logger finds it in context-local storage.
    log.info("charging card")

def handle_request(headers):
    # Boundary middleware: reuse a trusted inbound ID, otherwise generate one.
    correlation_id.set(headers.get("X-Request-ID") or uuid.uuid4().hex)
    log.info("request received")
    with ThreadPoolExecutor(max_workers=2) as pool:
        # Pool threads do not inherit contextvars. Without copy_context(),
        # charge_card would log correlation_id=- and the link would be lost.
        ctx = contextvars.copy_context()
        pool.submit(ctx.run, charge_card).result()
    # Outgoing hop: copy the ID into request headers or queue message attributes.
    return {"X-Request-ID": correlation_id.get()}

if __name__ == "__main__":
    handle_request({"X-Request-ID": "req-7f3a9c"})
    handle_request({})  # no inbound ID, so a new one is generated

A Python sketch that stores the correlation ID in a contextvar at the request boundary, uses a logging filter to add it to every log line, and copies the context into a thread pool so the ID is not silently lost there.

Log levels with operational meaning and error handling

Log levels only help if every team means the same thing by them. If one service uses ERROR for a 404 and another uses it for data loss, alerting on ERROR becomes noise. Tie each level to an action:

Read the level from configuration that can change at runtime, through a config service, an admin endpoint, or a watched file. Ideally you can scope the change to one service or tenant. A redeploy restarts the process and can clear the conditions you were trying to observe.

Sampling or dropping DEBUG and INFO to cut volume carries a risk. Rare failures are the cases where you most need the events that came before them. When you sample, decide per correlation ID rather than per line, and keep every event for traces that end in an error.

Log errors as structured fields: error.type, error.message and error.stack_trace. Do not interpolate them into the message string, because fields can be grouped and counted and free text cannot. Log each error once, at the boundary where it is handled. That is usually the request handler or the job runner. Inner layers should wrap the error with context and rethrow it. If every layer logs on the way up, a single failure appears as five ERROR events, which inflates counts and hides the root cause.

Copyimport logging
import traceback

logger = logging.getLogger("orders")

class OrderError(Exception):
    pass

def charge_card(order_id):
    raise TimeoutError("payment gateway timed out after 5s")

def place_order(order_id):
    # Inner layer: add context and rethrow. Do not log here.
    try:
        charge_card(order_id)
    except TimeoutError as exc:
        raise OrderError(f"charge failed for order {order_id}") from exc

def handle_request(order_id, correlation_id):
    # Boundary: the only place this failure is logged, once, as ERROR.
    try:
        place_order(order_id)
    except OrderError as exc:
        logger.error(
            "order placement failed",  # stable message; details go in fields
            extra={
                "correlation_id": correlation_id,
                "order_id": order_id,
                "error.type": type(exc).__name__,
                "error.message": str(exc),
                # Python 3.10+: includes the chained TimeoutError cause
                "error.stack_trace": "".join(traceback.format_exception(exc)),
            },
        )
        return 500
    return 200

def set_log_level(level_name):
    # Called by an admin endpoint or config watcher; no restart required.
    logger.setLevel(level_name)  # e.g. "DEBUG" while investigating

Inner layers wrap and rethrow with context, the request handler logs the failure once with error.type, error.message and error.stack_trace as separate fields for a JSON formatter to emit, and the logger level can be changed at runtime without a redeploy.

Implementing and enforcing the schema across teams

Agreeing on a schema is the easy part. Keeping a dozen teams on it means making the compliant path the default one.

Ship a library, not a document

Publish a thin logging wrapper per language that sets the required fields, enforces level names and serializes one JSON object per line to stdout. Services should not open sockets, manage files or batch logs themselves. The platform collector handles shipping, buffering and retries, and it adds host, pod or container metadata. This keeps backpressure handling and enrichment in one place instead of reimplementing them in every service.

Developers still need readable output. Add a local mode that pretty-prints the same events. Both formats must come from one event model, or the human-readable path will quietly drift from what production emits.

Test the output, not the intent

Add a CI test that runs a representative code path, captures stdout and validates every line against the schema as a JSON Schema check. This catches a renamed field or a stray print before it breaks a dashboard.

Govern changes like an API

Dashboards, alerts and queries depend on the field names. Additive fields are cheap. Renames and type changes need review, a deprecation window and dual-writing. A small owning group prevents several teams from inventing user_id, userId and uid independently.

Keep secrets out

Structured fields make leaked data easy to find, query and export. Have the library drop fields that aren't on an allowlist rather than relying on a blocklist. Add redaction in the pipeline as a second layer.

When to avoid it and what the alternatives are

A shared schema pays for itself when requests cross service and team boundaries. Below that threshold it is mostly overhead. A single service with one owning team can use whatever JSON shape its logging library emits by default, because nobody else has to join against it. A prototype expected to live for weeks doesn't need schema reviews. A team of three who can ask each other what a field means gets little from a formal contract. Emitting JSON at all is still cheap and worth doing. The governance is what to skip.

Choosing the right signal

Why logs make poor metrics

You can build a dashboard by counting log lines that match a query, but every refresh has to scan and parse raw events. Cost then grows with traffic volume, not with the number of series you care about. A counter or histogram is aggregated in the process before it leaves, so each scrape sends a few numbers whether the service handled ten requests or ten million. Alerts built on log queries also inherit ingestion delay and sampling gaps, so they fire later and less reliably than alerts on purpose-built metrics.

Impact on cost and engineering time

Structured logging adds costs in three places: CPU and I/O on the service, ingestion and indexing on the logging platform, and coordination time across teams. Each cost has a specific mitigation.

Illustrative estimate (assumed numbers, not a benchmark)

Daily volume = requests/s × lines per request × bytes per line × 86,400

Assumed inputs: 2,000 req/s, 6 lines per request, 600 bytes per line.

2,000 × 6 × 600 × 86,400 ≈ 622 GB/day.

Now keep every warning and error, and sample successful requests at 10%. Assume this brings the average to 1 line per request: 2,000 × 1 × 600 × 86,400 ≈ 104 GB/day.

Sample by trace ID, not per line. If you drop individual lines, the requests you keep will be missing pieces, and you can no longer join them across services.

Engineer-hours saved per month = incidents × responders × hours saved per incident

Assumed inputs: 4 incidents, 3 responders, 1.5 hours saved per incident = 18 engineer-hours per month.

The saved time comes from filtering one correlation ID across every service and jumping straight to the matching trace. Without that, responders grep individual hosts and line up timestamps by hand.

To compare the two sides, price both in your own storage costs and engineering rates.

How OneEnv can help

Disclosure: OneEnv is the product behind this blog.

OneEnv (oneenv.io) is a workspace-based governance platform for the things services share — environment variables, database structure, API routes and validation rules. A change to a shared value is staged on a branch, reviewed once, and then every affected service explicitly approves or rejects it before it goes live, so you can see who a change touches before it ships instead of finding out in production.

Applied to structured logging, this is where it helps:

Saves time

• Every service that inherits a shared value is listed in one place, so nobody searches repos and chats to find out who is affected.
• One review per branch replaces coordinating each team separately; each service still keeps its own approve/reject.
• Reviewers see the exact diff and the list of affected services up front, so approval is a decision rather than an investigation.

Saves cost

• Fewer production incidents and rollbacks caused by a shared value, route or column changing under a consumer unannounced.
• Less duplicated configuration to maintain: one shared definition inherited by many services.
• Per-service approve/reject means a bad shared value can be refused where it would break something, before it is live.
• OneEnv has a free plan (1 workspace, unlimited services and members, full governance), so trying it costs nothing.

These are qualitative benefits, not measured results — how much you save depends on how many services share the value.

Want to try it? Start at oneenv.io.