Shipping LLM logs to Elasticsearch with structured fields

Shipping LLM logs to Elasticsearch gives you a queryable audit trail of model usage, latency, and failures. This tutorial builds a minimal Python pipeline that captures structured fields from OpenAI-compatible API calls and indexes them into Elasticsearch 8.x without third-party log shippers. You will end up with a mapping you can aggregate on and a buffered writer that survives traffic spikes.

Prerequisites

• Docker for a local Elasticsearch 8.13 node.
• Python 3.11+ and pip.
• openai, elasticsearch, python-dotenv installed in a virtualenv.
• An OpenAI-compatible API key (we use OPENAI_API_KEY). If you route through a gateway such as n4n.ai, the response already includes per-token usage metering and provider fallback headers; map those directly to the schema below.

Why structured fields instead of text logs

A line like INFO request to gpt-4o-mini took 412ms is useless when you need p95 latency per model across a week. Elasticsearch keyword and numeric fields let you aggregate without regex. The cost is upfront schema discipline: decide types before you write the first document.

Design the document schema

Define explicit mappings. Dynamic mapping will guess latency_ms as a float and won’t mark @timestamp as a date, breaking time-series queries.

Create the index

curl -X PUT "localhost:9200/llm-logs" \
  -H 'Content-Type: application/json' \
  -u "elastic:changeme" \
  -d '{
    "mappings": {
      "properties": {
        "@timestamp": { "type": "date" },
        "model": { "type": "keyword" },
        "provider": { "type": "keyword" },
        "prompt_tokens": { "type": "integer" },
        "completion_tokens": { "type": "integer" },
        "latency_ms": { "type": "integer" },
        "status": { "type": "keyword" },
        "error_type": { "type": "keyword" },
        "request_id": { "type": "keyword" },
        "cache_hit": { "type": "boolean" }
      }
    }
  }'

Expected: {"acknowledged":true}.

Stand up Elasticsearch locally

For development, a single node is enough:

docker run -d --name es \
  -p 9200:9200 -p 9300:9300 \
  -e "discovery.type=single-node" \
  -e "xpack.security.enabled=false" \
  -e "ES_JAVA_OPTS=-Xms512m -Xmx512m" \
  docker.elastic.co/elasticsearch/elasticsearch:8.13.4

Wait for startup, then check health:

curl -X GET "localhost:9200/_cluster/health?pretty"

You’ll see "status":"yellow" (single node, replicas unassigned) which is fine locally.

Python project setup

mkdir llm-logs-es && cd llm-logs-es
python -m venv .venv && source .venv/bin/activate
pip install openai elasticsearch python-dotenv

.env file:

OPENAI_API_KEY=sk-...
ES_URL=http://localhost:9200
ES_INDEX=llm-logs

Build the Elasticsearch logger

Batch writes. When shipping LLM logs to Elasticsearch under load, per-call index requests create small segments and waste CPU.

# es_logger.py
import os
from datetime import datetime, timezone
from elasticsearch import Elasticsearch, helpers

class LLMLogger:
    def __init__(self, es_url: str, index: str, batch_size: int = 50):
        self.es = Elasticsearch(es_url)
        self.index = index
        self.batch_size = batch_size
        self._buffer = []

    def record(self, event: dict):
        event.setdefault("@timestamp", datetime.now(timezone.utc).isoformat())
        self._buffer.append({"_index": self.index, "_source": event})
        if len(self._buffer) >= self.batch_size:
            self.flush()

    def flush(self):
        if not self._buffer:
            return
        helpers.bulk(self.es, self._buffer)
        self._buffer.clear()

helpers.bulk takes a list of dicts with _index and _source. It retries on transient errors by default.

Instrument the LLM call

Use the official openai client. Capture timing with perf_counter, and pull token usage from the response. Use with_raw_response so we can read cache headers if the gateway sets them.

# call_llm.py
import time, uuid, os
from openai import OpenAI
from dotenv import load_dotenv
from es_logger import LLMLogger

load_dotenv()
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
logger = LLMLogger(os.environ["ES_URL"], os.environ["ES_INDEX"])

def chat(model: str, prompt: str, provider: str = "openai"):
    req_id = str(uuid.uuid4())
    start = time.perf_counter()
    try:
        raw = client.chat.completions.with_raw_response.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.2,
        )
        resp = raw.parse()
        headers = raw.headers
        latency = int((time.perf_counter() - start) * 1000)
        usage = resp.usage
        logger.record({
            "model": model,
            "provider": provider,
            "prompt_tokens": usage.prompt_tokens,
            "completion_tokens": usage.completion_tokens,
            "latency_ms": latency,
            "status": "success",
            "request_id": req_id,
            "cache_hit": headers.get("x-cache") == "HIT",
        })
        return resp.choices[0].message.content
    except Exception as e:
        latency = int((time.perf_counter() - start) * 1000)
        logger.record({
            "model": model,
            "provider": provider,
            "latency_ms": latency,
            "status": "error",
            "error_type": type(e).__name__,
            "request_id": req_id,
        })
        raise

if __name__ == "__main__":
    answer = chat("gpt-4o-mini", "What is 2+2? Answer in one word.")
    print(answer)
    logger.flush()

Run:

python call_llm.py

Console prints Four. That confirms the call and the logging path.

Verify the document in Elasticsearch

curl -X GET "localhost:9200/llm-logs/_search?pretty" \
  -H 'Content-Type: application/json' \
  -d '{
    "size": 1,
    "sort": [{ "@timestamp": "desc" }]
  }'

Expected _source:

{
  "@timestamp": "2024-06-12T08:22:31.123456+00:00",
  "model": "gpt-4o-mini",
  "provider": "openai",
  "prompt_tokens": 12,
  "completion_tokens": 1,
  "latency_ms": 412,
  "status": "success",
  "request_id": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "cache_hit": false
}

If that shape appears, you are shipping LLM logs to Elasticsearch with structured fields correctly.

Capture gateway-level metadata

When a gateway fronts multiple providers, record which backend served the request and whether automatic fallback occurred. The n4n.ai gateway, for example, returns per-token usage metering and forwards provider cache-control hints; read those from response headers (x-provider, x-cache) and add them to the event before logger.record. This keeps documents comparable across backends without custom parsing.

Mapping fallback events

If the gateway retries on provider rate limits, emit a separate event with status: "error" and error_type: "ProviderFallback" so you can count degradation:

logger.record({
    "model": model,
    "provider": headers.get("x-provider", "unknown"),
    "latency_ms": latency,
    "status": "error",
    "error_type": "ProviderFallback",
    "request_id": req_id,
})

Query patterns for ops

Aggregate to spot regressions:

curl -X POST "localhost:9200/llm-logs/_search?pretty" \
  -H 'Content-Type: application/json' \
  -d '{
    "size": 0,
    "aggs": {
      "per_model": {
        "terms": { "field": "model" },
        "aggs": {
          "avg_latency": { "avg": { "field": "latency_ms" } },
          "p95_latency": { "percentiles": { "field": "latency_ms", "percents": [95] } },
          "errors": { "filter": { "term": { "status": "error" } } }
        }
      }
    }
  }'

This returns average, p95, and error counts per model.

Production hardening

• Apply an ILM policy: roll over at 50GB, delete after 30 days.
• Ship from a single worker or queue (Redis/Kafka) rather than giving every app process write access to ES.
• Mask PII. We omitted prompt and completion text from the schema on purpose; storing user input in ES is a compliance risk.
• Use integer for tokens and latency; do not let them become long or float by accident.

When shipping LLM logs to Elasticsearch at scale, batch size and mapping correctness decide whether your cluster survives a traffic spike. A 50-document buffer and explicit mappings are enough for most mid-size deployments.

Extending the schema

Add user_id (keyword) or trace_id (keyword) if you correlate with application traces. Keep enums as keyword, never text, or aggregations will tokenize them. With this foundation, you can build Kibana dashboards that show cost per model by multiplying prompt_tokens and completion_tokens with your price sheet at query time.