Log Shipping and Collection for Python Services

A log record that never reaches a searchable store is an expensive way to slow down a request. This guide covers the path from logger.info() to a query result: which sink to write to, what reads it, and the handful of failure modes that quietly destroy records in between. It is part of the Python telemetry pipelines and delivery section and assumes the records themselves are already structured — if they are not, start with structured logging with the Python standard library, because almost every problem below is easier when each record is one JSON object. The focused articles in this topic are Handling Multi-Line Tracebacks in Log Shippers, Sending Python Logs to Loki, Shipping JSON Logs to Elasticsearch from Python, Shipping Python Logs with Fluent Bit and Standard Output vs File Logging in Containers.

Prerequisites

pip install "python-json-logger>=2.0.7,<4.0.0" \
            "structlog>=24.1.0,<26.0.0"

No shipping library is installed in the application. That is the point: the application writes, and something else collects.

Concept and architecture

A shipping pipeline has exactly three responsibilities, and confusion about which component owns which is the source of most of its problems.

The application owns the record's content. Message, level, timestamp, logger name, and every field that only the application knows — the tenant, the order identifier, the trace identifier. Nothing downstream can reconstruct these, so anything the application omits is gone.

The collector owns the record's context. Host, pod, namespace, container, image tag, region, environment. These are facts about where the process is running, not about what it did, and a service that stamps them itself is duplicating information the platform already has and will be wrong the first time it is deployed somewhere new.

The store owns the record's shape. Field types, index mappings, retention. This is where a schema disagreement becomes a rejected document, which is why designing a log schema for a service fleet is worth doing before the second service ships.

The handoff between the first two is a stream of bytes, and the format of those bytes decides how much interpretation the collector has to perform. One JSON object per line requires none. Anything else requires a parser, and a parser is a piece of configuration that has to be correct in every environment, forever, and that fails silently by producing a record with one field called message containing everything.

Step-by-step implementation

Step 1 — Produce one JSON object per record, on one line. The formatter is the only place this can be guaranteed, so guarantee it there rather than hoping no message contains a newline.

# logging_setup.py
import logging
import sys
from pythonjsonlogger import jsonlogger


class SingleLineJsonFormatter(jsonlogger.JsonFormatter):
    """JSON on exactly one line, with the traceback carried as a field."""

    def format(self, record: logging.LogRecord) -> str:
        line = super().format(record)
        # 1. Defensive: nothing reaches the stream with an embedded newline.
        return line.replace("\n", "\\n").replace("\r", "")

    def add_fields(self, target, record, message_dict):
        super().add_fields(target, record, message_dict)
        target["level"] = record.levelname
        target["logger"] = record.name
        # 2. The traceback becomes a string field, never extra physical lines.
        if record.exc_info:
            target["exception"] = self.formatException(record.exc_info)
            target.pop("exc_info", None)


handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(SingleLineJsonFormatter("%(asctime)s %(message)s"))
logging.basicConfig(level=logging.INFO, handlers=[handler], force=True)

Expected Output: an exception produces one line, with the traceback inside it.

{"asctime": "2026-09-18T09:14:02.881Z", "message": "charge failed", "level": "ERROR", "logger": "billing", "exception": "Traceback (most recent call last):\n  File \"billing.py\", line 88, in charge\n    gateway.post(payload)\nTimeoutError: read timed out"}

Step 2 — Put the write behind a queue if the sink can be slow. A StreamHandler writing to a full pipe blocks the calling thread, which under load means request latency caused by the log collector. The stdlib remedy is covered fully in non-blocking logging with QueueHandler; the shipping-relevant part is that the queue must be bounded and its drops counted.

import queue
from logging.handlers import QueueHandler, QueueListener

_q: "queue.Queue[logging.LogRecord]" = queue.Queue(maxsize=10_000)
listener = QueueListener(_q, handler, respect_handler_level=True)
listener.start()

root = logging.getLogger()
root.handlers = [QueueHandler(_q)]

Step 3 — Cap the record size before the runtime does. Several container runtimes split a log line above a fixed size, typically sixteen kilobytes, into two physical lines. Each fragment is invalid JSON, so the store rejects both and the record is gone entirely — a failure mode that only appears for the largest and most interesting records.

MAX_FIELD = 4_000       # characters, not bytes; leave headroom for escaping

def truncate(value: str) -> str:
    if len(value) <= MAX_FIELD:
        return value
    return value[:MAX_FIELD] + f"…[truncated {len(value) - MAX_FIELD} chars]"

Step 4 — Let the platform collect. The agent tails the runtime's log directory, adds the environment metadata, and forwards. Nothing about this is application configuration.

# fluent-bit.conf equivalent, expressed as the pieces that matter
[INPUT]
    Name              tail
    Path              /var/log/containers/*.log
    Parser            cri
    DB                /var/log/flb_state.db      # offset survives a restart
    Mem_Buf_Limit     32MB
    Skip_Long_Lines   Off                         # count them, do not discard silently

[FILTER]
    Name    kubernetes
    Match   kube.*
    Merge_Log           On                        # parse the JSON the app emitted
    Keep_Log            Off

Step 5 — Verify the whole path with one token. Testing hops individually proves nothing about the path. Log a value nothing else will ever contain and search for it.

import uuid
probe = uuid.uuid4().hex
logging.getLogger("startup").info("collection probe", extra={"probe": probe})
print(f"search the store for probe={probe}")

Configuration reference

Async and concurrency considerations

Writing to standard output from asyncio has the same hazard as any other synchronous call on the event loop: StreamHandler.emit performs a write syscall, and if the pipe buffer is full that syscall blocks the single loop thread, stalling every coroutine in the process. Under normal conditions the write returns in microseconds and this never matters. Under a slow or stopped collector it matters enormously, and the symptom — every endpoint slow at once, CPU flat — is indistinguishable from the other event-loop stalls described in logging from asyncio tasks without blocking.

The queue handler fixes this by moving the write to a separate thread, and the fix is complete only if the queue is bounded: an unbounded queue converts a blocked write into unbounded memory growth, which fails later and worse.

Under a prefork server every worker writes to the same file descriptor. Writes below the pipe atomicity limit — four kilobytes on Linux — are atomic, so records do not interleave; above it they can, which is a second reason to cap record size. This is also why a file sink shared by multiple processes needs the treatment in thread-safe logging in multiprocessing rather than a shared FileHandler.

Production code examples

A complete startup module: JSON on one line, a bounded queue, a size cap, and a sequence number that makes loss detectable.

# observability/logsetup.py
import atexit
import itertools
import logging
import queue
import sys
from logging.handlers import QueueHandler, QueueListener

from pythonjsonlogger import jsonlogger

_seq = itertools.count()


class ShippingFormatter(jsonlogger.JsonFormatter):
    MAX = 4_000

    def add_fields(self, target, record, message_dict):
        super().add_fields(target, record, message_dict)
        target["level"] = record.levelname
        target["logger"] = record.name
        # 1. A per-process sequence number turns silent loss into a gap you can see.
        target["seq"] = next(_seq)
        if record.exc_info:
            target["exception"] = self.formatException(record.exc_info)[: self.MAX]
            target.pop("exc_info", None)
        # 2. Cap every string field so no line can be split by the runtime.
        for key, value in list(target.items()):
            if isinstance(value, str) and len(value) > self.MAX:
                target[key] = value[: self.MAX] + "…[truncated]"

    def format(self, record):
        return super().format(record).replace("\n", "\\n").replace("\r", "")


def configure(level: str = "INFO") -> None:
    stream = logging.StreamHandler(sys.stdout)
    stream.setFormatter(ShippingFormatter("%(asctime)s %(message)s"))

    q: "queue.Queue[logging.LogRecord]" = queue.Queue(maxsize=10_000)
    listener = QueueListener(q, stream, respect_handler_level=True)
    listener.start()

    root = logging.getLogger()
    root.setLevel(level)
    root.handlers = [QueueHandler(q)]

    # 3. Without this, records in the queue at exit never reach the stream.
    atexit.register(listener.stop)

Expected Output: two consecutive records, with the sequence number the gap detector reads.

{"asctime": "2026-09-18T09:20:11.004Z", "message": "order accepted", "level": "INFO", "logger": "orders", "seq": 48213}
{"asctime": "2026-09-18T09:20:11.009Z", "message": "payment authorised", "level": "INFO", "logger": "billing", "seq": 48214}

A gap check that runs against the store and reports what the pipeline lost:

# gapcheck.py — run against a window of records from one process.
def missing_sequences(seqs: list[int]) -> list[tuple[int, int]]:
    """Return (start, end) ranges absent from an otherwise contiguous run."""
    ordered = sorted(set(seqs))
    gaps = []
    for prev, nxt in zip(ordered, ordered[1:]):
        if nxt != prev + 1:
            gaps.append((prev + 1, nxt - 1))
    return gaps


print(missing_sequences([48211, 48212, 48213, 48219, 48220]))

Expected Output:

[(48214, 48218)]

Five records produced and never stored. That is an answer no dashboard provides, obtained from one integer field.

Choosing a collection path

Four paths are in common use, and each is right somewhere.

Standard output, collected by an agent. The default for containers. The application writes to a stream it always has, the runtime persists it, and an agent reads it. It survives almost any degradation of the process, needs no network from the application, and costs one syscall per record. Its weaknesses are the runtime's line limit and the fact that everything transits a file on the node, so a node with a full disk stops collecting.

A file, tailed by an agent. The right choice on long-lived hosts where an agent is already tailing a directory, and for records that must outlive the collection path — an audit log that has to exist on disk even if the shipper never runs. It costs the application a rotation policy, covered in best practices for log rotation in Python, and it introduces the possibility of the file being rotated out from under a slow reader.

A network handler in the application. A SysLogHandler, an HTTP handler, or a vendor SDK. This is the only path that gets records out of a host whose disk is full, and it is the only one that puts the collector's availability inside your request path. It is defensible for a small number of high-value records and a poor default for everything, for the reasons set out in HTTP and webhook logging handlers.

The OpenTelemetry logs pipeline. The application emits log records through the same SDK and the same OTLP connection as its spans, so logs arrive already carrying trace_id and resource attributes, and one collector configuration covers all three signals. It costs an SDK dependency and a provider lifecycle, and it is described in exporting Python logs through OpenTelemetry.

The decision is rarely exclusive. A common and sensible arrangement is standard output for everything, plus a separate file sink for audit records that must not depend on the collector at all.

Verifying and monitoring the pipeline

A log pipeline fails by producing fewer records, and fewer records look exactly like a quieter service. Three checks separate them.

Per-service record rate. The number of records arriving in the store, per service, per minute. This is the single most valuable pipeline alert, because it catches an agent that died, a parser that started rejecting, a node whose disk filled and a deploy that accidentally raised the log level — all of which present identically from anywhere else. Alert on a drop relative to the same service an hour ago rather than on an absolute floor, since services legitimately differ by orders of magnitude.

Parse failure count. Every collector counts records it could not parse. This should be zero, and when it is not, it is usually one service emitting one malformed field. A non-zero and growing count is the early warning for a schema change that will shortly start rejecting documents.

Sequence gaps. The seq field from the example above, checked over a window. This is the only check that distinguishes "the service logged less" from "the pipeline delivered less", because the numbers are assigned at the point of production. It is worth the one integer.

# the record rate for a service, compared against its own recent past
sum by (service) (rate(logs_ingested_total[5m]))
  <
sum by (service) (rate(logs_ingested_total[5m] offset 1h)) * 0.4

# anything the collector could not parse
rate(fluentbit_output_retries_failed_total[5m]) > 0

Expected Output: during a rollout that changed a field's type, the parse failure alert fires several minutes before the ingest rate drops, which is the ordering that makes it useful.

LogParseFailures   firing   service=checkout   for 3m
LogVolumeDrop      pending  service=checkout

Verification at deploy time is cheaper than either. The probe token from step 5, emitted at startup and checked from a smoke test, proves the whole path for that service in that environment before anybody depends on it.

One further note on ordering. Nothing in this path preserves it. Records from one process arrive in write order, but records from several processes are interleaved by the collector, batched by the shipper and indexed concurrently by the store, so two records written a millisecond apart on different pods can be stored in either order. Any reasoning that depends on sequence — which of two events happened first, whether a retry preceded a failure — has to come from a timestamp with sub-millisecond resolution recorded at the point of production, or from a trace, where causality is explicit rather than inferred. Sorting a log view by ingestion time and reading it as a sequence of events is one of the most reliable ways to reach a confident wrong conclusion during an incident.

Common mistakes

json.JSONDecodeError: Extra data in the collector, repeatedly. A record contained a raw newline, so one logical record arrived as two fragments. Root cause: a message or field value with embedded newlines reaching the stream unescaped. Remediation: escape in the formatter rather than trusting callers, as in step 1.

Half the tracebacks are missing their exception type. The runtime split a long line. Root cause: a record above the runtime's per-line limit. Remediation: cap field lengths in the formatter and count the truncations so the cap's effect is visible.

Logs stop arriving from one pod and nothing alerts. The agent lost its offset after a restart and resumed at the end of the file, or the file was rotated while the agent was behind. Root cause: no persisted position, or an agent that cannot keep up with the write rate. Remediation: persist the offset to a volume, and alert on per-service record rate rather than on agent health.

Every field is a string in the store. The application emitted numbers as strings, or the collector's parser produced a single text field. Root cause: formatting numbers with str() or leaving JSON parsing off in the collector. Remediation: keep native types in the formatter and enable structured parsing at the collection step.

The store rejects documents and the collector reports success. Most collectors count a batch as delivered once the backend accepts the request, and a bulk indexing API can accept a request while rejecting individual documents inside it. Root cause: treating transport success as indexing success. Remediation: read the backend's per-document rejection counter, not only the collector's send counter, and alert on it separately.

Two services disagree about a field name and neither is wrong. One writes user, the other user_id, and a query that joins them silently returns half the data. Root cause: no shared vocabulary. Remediation: agree the field names once and enforce them in a shared logging package, so a new service inherits the schema instead of inventing one.

A field's type changed between releases and documents are silently rejected. One release emitted user_id as an integer, the next as a string. Root cause: no schema discipline. Remediation: fix the type at the schema level and treat a change to it as a breaking change, as covered in adopting Elastic Common Schema fields for Python logs.

Frequently Asked Questions