Build cost tracking pipeline from gateway metrics

Introduction

Engineers often build gateway middleware that logs every model request, but those logs don't arrive with dollar amounts attached. Calculating per-team LLM spend requires a dedicated pipeline that reads raw events from a stream, attaches pricing data from a model catalog, and writes attributed cost records to a database where dashboards and chargeback reports can query them. Infrastructure expenses — GPU compute hours and persistent storage — add a second cost dimension that must be collected from Kubernetes metrics and merged into the same store. By the end of this lesson, you'll be able to build a three-stage cost tracking pipeline that ingests gateway log events, enriches them with per-model pricing, and writes attributed cost records to PostgreSQL alongside Kubernetes infrastructure costs.

Key Terminology

• CostEvent — the enriched data record produced by the pipeline for each gateway log event, carrying raw fields like input_tokens, output_tokens, model_id, and team_id alongside the calculated cost_usd value ready for database writes and chargeback queries.
• cost enrichment — the pipeline stage that calls the price catalog with a model identifier and token counts to compute a dollar amount, transforming a structurally complete but price-free gateway log event into a CostEvent ready for persistence.
• batch acknowledgment — the pattern in CostPipelineWorker where the stream consumer calls source.acknowledge() only after store.write_batch() succeeds, guaranteeing at-least-once delivery so a mid-batch crash triggers re-delivery rather than silent data loss.
• InfrastructureCostCollector — the component that queries the Kubernetes metrics API for per-tenant GPU hours via get_gpu_hours and PVC storage via get_pvc_usage_gb, converting raw resource readings into cost figures using configurable price rates.
• storage proration — the calculation that converts a monthly per-GB price into a daily cost figure ((storage_gb * price_per_gb_month) / 30) so infrastructure storage costs can be summed in the same daily aggregation window as per-request token costs.
• cost attribution — the process of tagging each CostEvent with tenant_id and team_id sourced from the original gateway log event, enabling downstream SQL queries and dashboards to aggregate spend by organizational unit for chargeback reporting.

Concepts

The Three-Stage Pipeline Architecture

Raw gateway logs carry everything needed to identify a request — model, token counts, team, tenant — but they carry no dollar amounts. The pipeline's job is to close that gap in three discrete stages: ingestion reads batches of events from the stream (Redis Streams or Kafka), enrichment calls the price catalog to compute a cost for each event, and the store writer persists the resulting CostEvent records to PostgreSQL where dashboards and chargeback reports can query them. Keeping the stages separate makes each one independently testable and replaceable: you can swap the stream source without touching enrichment logic, or change how the price catalog is structured without touching the store schema.

The CostPipelineWorker wires these three stages into a single processing loop (see Code Walkthrough). Events that fail enrichment are logged and skipped rather than crashing the batch, so a misconfigured model ID in one event doesn't block the remaining records from being written.

At-Least-Once Delivery via Batch Acknowledgment

The order of operations inside CostPipelineWorker.run() is not arbitrary. The worker calls store.write_batch() first, and only after that call succeeds does it call source.acknowledge(). This sequence implements an at-least-once delivery guarantee: if the process crashes between writing and acknowledging, the stream broker re-delivers the same batch on the next startup. The price of this guarantee is that the cost store must tolerate duplicate writes — using event_id (mapped from request_id) as a deduplication key at the database layer prevents double-counting when re-delivery occurs.

This is the correct tradeoff for financial data. Silently dropping a cost event means under-reporting spend with no signal that anything went wrong. Processing a duplicate event and deduplicating at write time is a far cheaper failure mode to detect and handle.

Infrastructure Costs as a Parallel Dimension

Gateway logs only capture what flows through the model API — prompt tokens, completions, model routing decisions. They are completely silent about the shared compute that runs the cluster: GPU pods, persistent volume claims, node allocation. These infrastructure costs must be collected from a different source — the Kubernetes metrics API — and expressed in different units (GPU-hours, gigabyte-months) before they can join token costs in the same aggregation.

InfrastructureCostCollector handles this by querying k8s_metrics_client for per-tenant GPU usage and PVC storage, then applying configurable price rates to produce figures in the same currency as CostEvent.cost_usd. Storage is prorated from its natural billing unit (monthly per-GB) to a daily figure so it can be summed alongside per-request costs within a standard reporting window. The result is a unified cost store where a single query can aggregate both token spend and infrastructure overhead per tenant — the foundation for a chargeback report that reflects actual organizational cost rather than just API spend.

Code Walkthrough

Building on the three-stage architecture — Event Ingestion, Cost Enrichment, and Cost Store Writer — the code below wires all three stages into a running pipeline worker and adds a separate collector for infrastructure costs from Kubernetes.

The CostPipelineWorker reads batches of gateway log events from the configured source (a Redis Streams or Kafka consumer), calls the price catalog to calculate the cost for each event, and writes enriched CostEvent records to the cost store. Processing is batched: acknowledgment to the stream happens only after the entire batch is successfully written to PostgreSQL, so a crash mid-batch causes events to be re-delivered rather than silently dropped.

Code snippetpython
1import asyncio
2import logging
3from dataclasses import dataclass
4
5logger = logging.getLogger(__name__)
6
7@dataclass
8class CostEvent:
9    event_id: str
10    request_id: str
11    tenant_id: str
12    team_id: str
13    model_id: str
14    provider: str
15    input_tokens: int
16    output_tokens: int
17    cost_usd: float
18    cached: bool = False
19
20class CostPipelineWorker:
21    def __init__(self, event_source, cost_calculator, cost_store, batch_size: int = 100):
22        self.source = event_source
23        self.calculator = cost_calculator
24        self.store = cost_store
25        self.batch_size = batch_size
26
27    async def run(self):
28        logger.info("Cost pipeline worker started")
29        while True:
30            events = await self.source.read_batch(self.batch_size, timeout_ms=1000)
31            if not events:
32                continue
33
34            cost_events = []
35            for event in events:
36                try:
37                    cost = await self.calculator.calculate(
38                        model_id=event["model"],
39                        input_tokens=event["input_tokens"],
40                        output_tokens=event["output_tokens"],
41                        cached=event.get("cached", False),
42                    )
43                    cost_events.append(CostEvent(
44                        event_id=event["request_id"],
45                        request_id=event["request_id"],
46                        tenant_id=event["tenant_id"],
47                        team_id=event.get("team_id", "unknown"),
48                        model_id=event["model"],
49                        provider=event.get("provider", "unknown"),
50                        input_tokens=event["input_tokens"],
51                        output_tokens=event["output_tokens"],
52                        cost_usd=cost["cost_usd"],
53                        cached=event.get("cached", False),
54                    ))
55                except Exception as e:
56                    logger.error(f"Failed to calculate cost for {event['request_id']}: {e}")
57
58            if cost_events:
59                await self.store.write_batch(cost_events)
60                await self.source.acknowledge(events)
61                logger.info(f"Processed {len(cost_events)} cost events")

Alongside token-level costs, the pipeline must account for shared infrastructure expenses. The InfrastructureCostCollector queries the Kubernetes metrics API to retrieve GPU hours consumed and PVC storage usage per tenant, then converts those readings into cost figures using configurable price rates. Storage cost is prorated from a monthly rate to a daily figure so it can be summed alongside per-request costs in the same aggregation window.

Code snippetpython
1class InfrastructureCostCollector:
2    def __init__(self, k8s_metrics_client, price_per_gpu_hour: float, price_per_gb_month: float):
3        self.k8s = k8s_metrics_client
4        self.gpu_price = price_per_gpu_hour
5        self.storage_price = price_per_gb_month
6
7    async def collect_compute_costs(self, tenant_id: str, period_hours: int = 24) -> dict:
8        gpu_usage = await self.k8s.get_gpu_hours(tenant_id, period_hours)
9        return {
10            "tenant_id": tenant_id,
11            "resource": "gpu_compute",
12            "quantity": gpu_usage,
13            "unit": "gpu-hours",
14            "cost": round(gpu_usage * self.gpu_price, 4),
15            "period_hours": period_hours,
16        }
17
18    async def collect_storage_costs(self, tenant_id: str) -> dict:
19        storage_gb = await self.k8s.get_pvc_usage_gb(tenant_id)
20        daily_cost = (storage_gb * self.storage_price) / 30
21        return {
22            "tenant_id": tenant_id,
23            "resource": "storage",
24            "quantity": storage_gb,
25            "unit": "GB",
26            "cost": round(daily_cost, 4),
27        }

Confirm that running CostPipelineWorker.run() against a seeded event source produces CostEvent records in the PostgreSQL cost table and that InfrastructureCostCollector.collect_compute_costs() returns a non-zero cost dict when your Kubernetes metrics mock reports GPU usage.

Do's and Don'ts

Building on the pipeline architecture and the worker/collector code above, the imperatives below codify the operational habits that keep cost attribution accurate — and the pitfalls that silently corrupt chargeback totals.

Do's

1. ✓Do acknowledge stream events only after the full batch is written to PostgreSQL — CostPipelineWorker deliberately calls self.source.acknowledge(events) after self.store.write_batch(cost_events) succeeds, so a mid-batch crash causes re-delivery rather than silently dropping cost records and under-reporting spend.
2. ✓Do keep team_id and tenant_id as explicit fields on every CostEvent — per-team chargeback queries depend on these identifiers being written to the cost store at ingestion time; if team_id defaults to "unknown" because the gateway log omits it, that spend becomes unattributable and falls out of dashboards.
3. ✓Do prorate infrastructure storage costs to a daily figure by dividing the monthly rate by 30 — InfrastructureCostCollector.collect_storage_costs() uses (storage_gb * self.storage_price) / 30 so PVC costs land in the same per-day aggregation window as per-request token costs and can be summed in a single chargeback report.

Don'ts

1. ✗Don't acknowledge stream events before writing to the cost store — flipping the order of write_batch and acknowledge in CostPipelineWorker.run() means a PostgreSQL write failure causes events to be marked consumed on the stream, permanently losing the cost records with no re-delivery possible.
2. ✗Don't pass raw token counts directly to cost calculations without routing through cost_calculator.calculate() — the price catalog handles per-model pricing differences, cached-token discounts, and provider-specific rates; bypassing it with hard-coded rates produces incorrect cost_usd values that corrupt chargeback totals.
3. ✗Don't report GPU compute costs and storage costs in the same unit without distinguishing the resource field — InfrastructureCostCollector tags each record with "gpu_compute" or "storage" precisely so downstream aggregations can break out infrastructure cost dimensions; collapsing them into a single cost figure makes it impossible to show teams which resource is driving their infrastructure spend.