Shipping JSON Logs to Elasticsearch from Python

A Python service emitting clean JSON can still end up with half its records missing from the search index, and the reason is almost never the shipper. This page covers the part of the path that the application and the platform share: field types, index templates, and the mapping conflicts that reject documents while every component upstream reports success. It is a task article under log shipping and collection, part of the Python telemetry pipelines and delivery section, and it assumes records are produced as described in structured logging with the Python standard library.

Prerequisites

The application installs nothing Elasticsearch-specific; the shipper handles transport. What the application does own is the shape of each record, so a formatter that produces consistent types is the real prerequisite.

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

Implementation

Step 1 — Declare the field types before the first document. An index without a template gets its mapping from whatever arrives first, which means the type of every field in your fleet is decided by the first service to start after an index rolls over. That is not a stable basis for anything, and the fix is a template applied ahead of time.

PUT _index_template/python-logs
{
  "index_patterns": ["logs-python-*"],
  "data_stream": {},
  "template": {
    "mappings": {
      "dynamic": "strict",
      "properties": {
        "@timestamp":  { "type": "date" },
        "message":     { "type": "text" },
        "level":       { "type": "keyword" },
        "logger":      { "type": "keyword" },
        "service":     { "type": "keyword" },
        "trace_id":    { "type": "keyword" },
        "span_id":     { "type": "keyword" },
        "duration_ms": { "type": "float" },
        "http":        { "properties": {
            "method": { "type": "keyword" },
            "route":  { "type": "keyword" },
            "status": { "type": "short" }
        }},
        "exception":   { "type": "text" },
        "context":     { "type": "flattened" }
      }
    }
  }
}

Two decisions in that template carry most of the weight. dynamic: strict means a document containing an undeclared field is rejected loudly rather than silently creating a new mapping — painful for a week, and the only way to keep a cluster's field count bounded. And context is mapped as flattened, which gives services somewhere to put arbitrary key–value data without each distinct key becoming a permanent field in the cluster state.

Step 2 — Make the application's types match, permanently. A Python formatter that stringifies values is the usual source of conflict, because str(42) and 42 are indistinguishable to the author and completely different to the index.

import logging
from pythonjsonlogger import jsonlogger


class TypedFormatter(jsonlogger.JsonFormatter):
    """Keep native types; never stringify a number or a boolean."""

    NUMERIC = {"duration_ms", "retry_count", "bytes"}

    def add_fields(self, target, record, message_dict):
        super().add_fields(target, record, message_dict)
        target["level"] = record.levelname
        target["logger"] = record.name
        for key in self.NUMERIC & target.keys():
            # 1. Coerce once, here, rather than trusting every call site.
            try:
                target[key] = float(target[key])
            except (TypeError, ValueError):
                target.pop(key)          # better absent than the wrong type

Step 3 — Map identifiers as keywords. The distinction between text and keyword is the one mapping decision that engineers coming from a relational background get wrong most often, because both of them hold strings. A text field is analysed: it is lowercased, split on word boundaries, and stored as a list of tokens, which is exactly right for a human-readable message and exactly wrong for anything whose value is an opaque identifier. A trace identifier stored as text is analysed into tokens, so an exact-match query on it behaves unpredictably and an aggregation over it is meaningless. It is also stored twice — once analysed, once not — for no benefit, and it makes a terms aggregation over that field return fragments rather than values. This matters most for the trace identifiers added to log records, which exist specifically to be matched exactly.

Step 4 — Route each service to its own data stream. One stream per service lets retention, shard count and mapping differ where the services differ, and confines a mapping conflict to the service that caused it rather than to everything writing into a shared index. It also makes the cost attributable: a service producing ten times the volume of its neighbours is visible as a stream, not as a line item somebody has to go looking for. The cost is more indices to manage, which index lifecycle policies handle without human attention once they are written.

[OUTPUT]
    Name             es
    Match            kube.*
    Host             logs.observability.svc
    Suppress_Type_Name On
    Index            logs-python-${SERVICE_NAME}
    Trace_Error      On      # log the per-document rejection reason

Expected Output: a rejection, as it appears once Trace_Error is on — which is the difference between a mystery and a one-line fix.

{"error": {"type": "document_parsing_exception",
           "reason": "failed to parse field [duration_ms] of type [float]",
           "caused_by": {"type": "number_format_exception", "reason": "For input string: \"42 ms\""}},
 "status": 400}

The application's side of the contract

Very little of this page is Elasticsearch configuration, which is the point. The cluster enforces a contract; the application either honours it or its records are rejected. Three habits keep a Python service on the right side of that line.

Type stability is a public interface. A field that has ever been a number must stay a number, in every code path, in every release, forever. This is easy to say and easy to violate accidentally — an f-string in one branch, a str() in an error handler, a default of "unknown" where the happy path emits an integer. Coercing in the formatter, as step 2 does, moves the guarantee from every call site to one place. Where a field genuinely has two shapes, it is two fields.

Field names are a shared vocabulary. The cost of a service inventing latency when the fleet uses duration_ms is not one extra mapping entry; it is every dashboard, alert and saved query needing to know about both. A shared logging package that supplies the field names is the cheapest enforcement available, and it is the same argument made in designing a log schema for a service fleet.

Free-form data needs a home. Services always have something to log that does not fit the schema: a third-party response body, a computed diagnostic, a dictionary of feature flags. Forbidding it outright means it gets stuffed into the message string, where it is unqueryable. Giving it a single context field mapped as flattened means it is queryable, costs one mapping entry, and cannot grow the cluster state however creative its keys become. The rule that follows is simple enough to review in a pull request: top-level fields are declared, everything else goes under context.

Configuration options

The refresh interval is worth changing deliberately. The default optimises for documents becoming searchable within a second, which costs a segment flush at that cadence; log search almost never needs it, and raising it measurably reduces indexing cost on a busy cluster.

Verification

The check that matters is not whether documents arrive but whether any are rejected. Compare what the collector says it sent against what the index says it holds.

# documents the index actually contains, for the last hour
curl -s 'localhost:9200/logs-python-checkout/_count' \
  -H 'Content-Type: application/json' -d '{
    "query": {"range": {"@timestamp": {"gte": "now-1h"}}}
  }'

# indexing failures the cluster recorded over the same period
curl -s 'localhost:9200/_stats/indexing?filter_path=**.indexing' | python3 -m json.tool

Expected Output: counts that agree, and a failure count that is exactly zero.

{"count": 1842301}
{"indexing": {"index_total": 1842301, "index_failed": 0, "index_time_in_millis": 91204}}

A non-zero index_failed is the number to alert on. It is the only counter in the whole path that distinguishes "delivered" from "stored", and it is not exposed by the shipper, the runtime or the application.

Common mistakes

Documents disappear after a release. Error signature: document_parsing_exception in the cluster log, nothing anywhere else. Root cause: a field's type changed because a code path started formatting a value before logging it. Remediation: coerce types in the formatter as in step 2, and treat a field's type as part of the service's contract.

The cluster slows down and the cause is the mapping. Error signature: cluster state size in the hundreds of megabytes, slow master operations. Root cause: dynamic mapping over unbounded field names, usually a dictionary keyed by an identifier. Remediation: move free-form data under a flattened field and set dynamic: strict so the next attempt fails immediately instead of silently.

Exact searches on a trace identifier return the wrong documents. Error signature: a query for one identifier matching records from other traces. Root cause: the field is mapped as text and analysed into tokens. Remediation: map it as keyword and reindex, or accept the split until the current index rolls over.

Bulk requests succeed while half the records vanish. Error signature: the shipper reports no errors and the record count in the store is well below the count produced. Root cause: nothing reads the per-item errors in the bulk response. Remediation: enable the shipper's per-document error logging and alert on the cluster's index_failed counter, as described in log shipping and collection.

Frequently Asked Questions