Kafka Spend Log (SpendLog → Kafka → ClickHouse)

Auto AI Router can publish an extended copy of every spend event to Kafka, in addition to (or instead of) writing it to the LiteLLM PostgreSQL database. From Kafka the event flows into ClickHouse for analytics via a standard ClickHouse Kafka table engine and a materialized view — no custom consumer service is required. PostgreSQL remains the source of truth for auth (ValidateToken), budgets, and the LiteLLM UI; the Kafka/ClickHouse path is a separate, independently-enabled write path for analytics.

For the full design rationale (gap analysis against real LiteLLM data, config decisions, open-question resolutions), see the design document: auto_ai_router_kafka_spend_log_tz.md in the repository root.

Configuration

kafka:
  enabled: os.environ/KAFKA_ENABLED
  brokers:
    - "os.environ/KAFKA_BROKERS"      # "kafka1:9092,kafka2:9092"
  topic: "air.spend_logs"
  client_id: "auto_ai_router"

  log_queue_size: 5000
  log_batch_size: 100
  log_flush_interval: 5s

  tls_enabled: false
  sasl_mechanism: ""                   # "" | "PLAIN" | "SCRAM-SHA-256" | "SCRAM-SHA-512"
  sasl_username: "os.environ/KAFKA_SASL_USERNAME"
  sasl_password: "os.environ/KAFKA_SASL_PASSWORD"

  # Separate, independently-toggleable write-path: raw request/response
  # bodies, published to their own topic. See "Raw bodies" below for why
  # this isn't just a field on the spend event.
  raw_bodies:
    enabled: os.environ/KAFKA_RAW_BODIES_ENABLED   # default: false
    topic: "raw-bodies"
    store_raw_body: os.environ/KAFKA_RAW_BODIES_STORE_RAW_BODY       # default: false
    store_only_errors: os.environ/KAFKA_RAW_BODIES_STORE_ONLY_ERRORS # default: true
    redact_sensitive_fields: os.environ/KAFKA_RAW_BODIES_REDACT_SENSITIVE_FIELDS # default: true

litellm_db:
  enabled: true
  database_url: "os.environ/LITELLM_DATABASE_URL"
  disable_spend_logs_write: os.environ/DISABLE_PG_SPEND_LOGS   # default: false

kafka.enabled and litellm_db.disable_spend_logs_write are independent flags:

Kafka availability is not treated as critical for production traffic: there is no is_required-style flag, and an unreachable Kafka cluster does not block startup or request handling. Producer health is reflected via the manager's IsHealthy() state (surfaced in health/metrics), while the manager retries delivery in the background.

Event schema

The event is a flat JSON document (no nested objects), published to the air.spend_logs topic keyed by request_id. It builds on the existing SpendLogEntry used for Postgres, but expands the single Metadata JSON blob into typed, flat fields and adds credential/server/timing information that Postgres currently discards.

See section 4 of the design document for the complete field-by-field JSON example.

ClickHouse schema

A reference DDL — a Kafka-engine table reading air.spend_logs, a MergeTree table, and a MATERIALIZED VIEW connecting the two — is provided at clickhouse/init/01_spend_logs.sql. It matches the design document's section 8 schema (columns line up 1:1 with the flat JSON event above).

This DDL is a reference/example for local development and for DBAs setting up a production ClickHouse cluster — Auto AI Router itself does not create, own, or administer this schema. Retention (TTL) and Kafka topic partitioning are likewise left to whoever operates the target cluster; the router's only responsibility is producing well-formed events to the topic.

Upgrading an existing ClickHouse pipeline

The init DDL runs only for an empty ClickHouse data directory. Before deploying a router that publishes the cache and Web Search fields, pause AIR Kafka publishing and apply the migrations your schema is missing, in order:

1. clickhouse/migrations/002_cache_web_search_columns.sql — cache and Web Search fields;
2. clickhouse/migrations/003_upstream_send_ms.sql — upstream_send_ms;
3. clickhouse/migrations/004_explicit_cache_columns.sql — cache_type and explicit_cache_read_cost (Alibaba/Qwen explicit cache). Without it ClickHouse silently drops both keys, and explicit-cache requests show a cost breakdown that does not add up to total_cost.
4. clickhouse/migrations/005_tool_usage_columns.sql — per-tool usage and cost of built-in server-side tools (xAI server_side_tool_usage_details), tool_usage_cost and provider_reported_cost. Without it ClickHouse silently drops these keys; total_cost stays correct.
5. clickhouse/migrations/006_video_input_columns.sql — video_input_tokens and video_input_cost (input video priced apart from images, Gemini Embedding 2). Without it ClickHouse silently drops both keys, and requests with video input show a cost breakdown that does not add up to total_cost.

Every migration is safe to re-run on its own — but only forward, never out of order. Each one rebuilds air.spend_logs_kafka from its own column set alone, so re-running an earlier migration after a later one has already been applied narrows the Kafka table back down, below what the later migration added. air.spend_logs (the MergeTree table) only ever gets ADD COLUMN, so it keeps every column regardless — the two tables then disagree on column count, and every message the Kafka table hands to the materialized view fails with NUMBER_OF_COLUMNS_DOESNT_MATCH until the later migration(s) are re-applied to rebuild it with the full set again.

The Kafka-engine table doesn't support ALTER ... ADD COLUMN (ClickHouse rejects it with NOT_IMPLEMENTED), so each migration drops and recreates air.spend_logs_kafka and the materialized view with the new columns already in place, while adding the fields to the MergeTree table (air.spend_logs) with a plain ALTER ... ADD COLUMN IF NOT EXISTS instead — that table is never dropped. No events are lost: Kafka tracks consumer progress under kafka_group_name, which each migration keeps unchanged, so the recreated table resumes exactly where the old one stopped. For replicated production tables, add the cluster-specific ON CLUSTER clause required by your deployment.

After re-enabling publishing, send one synthetic event with non-zero cached_audio_input_tokens, cache_creation_5m_tokens, cache_creation_1h_tokens, web_search_requests, and web_search_cost. Verify that the row appears in air.spend_logs and that system.kafka_consumers reports no parse exceptions before completing the rollout.

PostgreSQL intentionally retains the upstream LiteLLM schema. Detailed cache and Web Search values live in LiteLLM_SpendLogs.metadata; the typed fields are available in Kafka and ClickHouse.

Replication: the air.spend_logs table uses plain MergeTree because docker-compose.kafka.yml runs a single, unreplicated ClickHouse node. For a production cluster with more than one replica, swap it for ReplicatedMergeTree (or a Replicated database engine) once Keeper/ZooKeeper and {shard}/{replica} macros are set up — otherwise a node failure loses spend/billing data. See the comment above the CREATE TABLE air.spend_logs statement in clickhouse/init/01_spend_logs.sql for the exact engine syntax.

Running locally

A self-contained Kafka (KRaft, single node) + ClickHouse stack for local development is provided in docker-compose.kafka.yml at the repository root. It auto-applies the ClickHouse schema above on first start via the standard docker-entrypoint-initdb.d mechanism.

docker compose -f docker-compose.kafka.yml up -d

This exposes Kafka on localhost:9092 and ClickHouse on localhost:8123 (HTTP) / localhost:9000 (native). Point KAFKA_BROKERS=localhost:9092 (from the host) or kafka:9092/kafka:29092 (from another container on the same compose network) at it to test the pipeline end to end.

Out of scope: request/response bodies for successful requests

Capturing and storing request/response bodies for every request (success included) is still out of scope — it would make air.spend_logs both heavier and much longer-retained than it needs to be, and is a distinct problem from "help me debug this one failure" (see "Raw bodies" below, which covers that narrower case). The event fields body_captured, body_request_bytes, and body_response_bytes remain reserved placeholders: body_captured: false, byte counts 0, no delivery/storage/PII-handling logic behind them.

Raw bodies (separate write-path)

The raw provider response body for a failed request is published separately from the spend event, to its own Kafka topic (kafka.raw_bodies, default topic raw-bodies, staged in ClickHouse as air.raw_bodies_kafka/air.raw_bodies_read) as a flat kafkalog.RawBodyEvent:

response_body and client_response_body are usually different values, on purpose. maskedUpstreamErrorBody (internal/proxy/errors.go) replaces the provider's own error text with a short, pre-vetted message for essentially every 4xx/5xx response — unconditionally, not gated by credential type — specifically so provider internals are never echoed back to the client. response_body is what the provider actually said; client_response_body is what the client was told instead. They're identical only when a mid-stream error is detected after the response has already committed and streamed those exact bytes to the client live — at that point there's nothing left to mask in hindsight.

request_body is a separate, explicit opt-in — off by default — and even then never carries prompt content by default. kafka.raw_bodies.store_raw_body (default false) must be turned on deliberately for request_body to ever be non-empty; leaving it off (the default) preserves the original failure-response-only design exactly. When it is on, redactRequestBodyForLogging (internal/proxy/proxy_helpers.go) still strips the actual conversation content before it ever reaches logCtx: messages, system, prompt, input, contents, and instructions are replaced with a shape-preserving placeholder (turn count and role kept, content replaced with "[REDACTED]"). Every other field keeps its structure — object keys stay as they are — but every string, number and boolean in it is masked to "[REDACTED]" unless its key is on a short allowlist: structural and enumerated strings (model, type, format, required, tool_choice, reasoning_effort, service_tier, size, …; see loggableStringFields), sampling/length limits and JSON-Schema constraints (temperature, top_p, max_tokens, minimum, maxLength, …; see loggableNumberFields) and request flags (stream, strict, …; see loggableBoolFields). The allowlists apply only at the top level, inside known parameter objects (reasoning, thinking, generationConfig, …) and inside tool, function, MCP server and structured-output definitions, which also keep name and server_label. Inside any other object — extra_body, the client's own data — every value is masked, so {"customer": {"name": "…", "phone": 79161234567}} keeps only its keys. So tool and response_format/text.format schema descriptions, enum/default values, user, seed, safety_identifier, prompt_cache_key, prediction, a FIM suffix, web search user_location and MCP server URLs/tokens never reach the log, while model, tool names, parameter names and types do. metadata, headers, user_location, latLng, logit_bias and JSON-Schema literals carry client-chosen keys as well as values, so only their size is kept: an object becomes "[REDACTED: N keys]", an array keeps its length. If the body isn't valid JSON (multipart, binary, malformed), nothing is captured at all rather than risk shipping something unredacted.

kafka.raw_bodies.redact_sensitive_fields (default true) is the toggle that controls this: set it to false and request_body captures the request verbatim, prompt included. This is a deliberate escape hatch, not a recommended default — it exists for a short-lived, access-controlled debugging session where the actual prompt is genuinely needed, at the cost of reintroducing exactly the privacy exposure the redaction exists to avoid. If you need to reproduce a specific failure's actual prompt without turning this off, correlate request_id with your own request logging outside AIR under whatever consent/retention rules already govern that data.

Three independent toggles control scope, all under kafka.raw_bodies:

With all three left at their defaults, behavior is unchanged from the original design: failure-only, provider/client response bodies only, no request content.

This is deliberately not a field on SpendEvent/air.spend_logs:

• Spend/billing rows are kept for a long time (retention measured in months/years) and are meant to stay light; raw bodies are bulky and only useful for a short debugging window, so they need their own, independently configurable ClickHouse retention (TTL) — a separate table gives you that for free, a shared one doesn't.
• It's independently toggleable (kafka.raw_bodies.enabled) precisely so it can be turned on temporarily while debugging without touching the always-on spend-log path, and turned back off (or left permanently off, the default) without affecting spend/billing at all.
• By default it only fires for failures (status == "failure") — no additional exposure of completion content for successful traffic beyond what already exists today, unless store_only_errors is explicitly disabled.

Join back to the spend event / air.errors on (request_id, server_router_id), not request_id alone. In a chained deployment (one AIR instance proxying to another as an upstream credential), request_id is derived from the upstream provider's own response id and is echoed back through every hop unchanged — two different hops logging the same logical request in the same millisecond can share a request_id. server_router_id (the hostname/pod of the specific instance that logged the row) disambiguates them; see the ORDER BY on air.logs in your ClickHouse schema for the same reasoning applied to the spend table.

A reference join view for a ClickHouse deployment with the matching air.errors/air.raw_bodies tables:

CREATE VIEW air.errors_with_raw AS
SELECT e.*, b.response_body, b.client_response_body, b.request_body
FROM air.errors AS e
LEFT JOIN air.raw_bodies AS b
    ON e.request_id = b.request_id AND e.server_router_id = b.server_router_id;

Implementation notes

The Kafka producer/event-builder lives in internal/kafkalog (manager, event, config, async logger). Both write-paths share one generic Logger[T] engine (T is a pointer type implementing Keyed, i.e. *SpendEvent or *RawBodyEvent) — same queue/batch/retry/DLQ/health-check machinery, just parameterized over the event type and pointed at a different topic (Manager/DefaultManager for spend events, RawBodyManager/DefaultRawBodyManager for error bodies).

Both are wired into the existing spend-logging call site in internal/proxy/proxy_log.go (logSpendToLiteLLMDB), which already runs from every place a request's log is finalized — no changes to individual call sites are needed to add either Kafka publish. The raw-body publish additionally requires kafka.raw_bodies.enabled, plus either status == "failure" or store_only_errors: false.