Overview
This build logs AI agent metrics in near real time, scales to millions of events/day, and keeps query latency under 200 ms. It uses:

– OpenTelemetry SDKs to emit spans/metrics
– OpenTelemetry Collector to batch and forward
– Kafka for durable buffering
– ClickHouse for fast analytics
– Metabase for dashboards and alerts

Use cases
– Track request latency, token usage, cost per endpoint
– Error rates by model/provider
– Agent step timings and tool success
– Customer-level SLOs and anomaly alerts

Architecture
– Services/agents emit OTLP spans and metrics.
– Otel Collector scrubs PII, batches, retries.
– Kafka provides backpressure and replay.
– ClickHouse ingests via Kafka Engine to a MergeTree.
– Metabase connects directly to ClickHouse.

Data model
Two core tables keep it simple and fast:

– agent_events: per request/step event (span-like)
– ts (DateTime64), trace_id, span_id, parent_span_id
– service, env, agent, model, provider
– route, user_id_hash, status, error_type
– latency_ms, tokens_prompt, tokens_completion, cost_usd
– attrs (JSON)

– agent_metrics_1m: rollups by minute
– ts_min, service, env, agent, model, route
– calls, p50_ms, p95_ms, error_rate, tokens_total, cost_usd

ClickHouse DDL
— Raw Kafka topic table
CREATE TABLE kafka_agent_events (
ts DateTime64(3),
trace_id String,
span_id String,
parent_span_id String,
service LowCardinality(String),
env LowCardinality(String),
agent LowCardinality(String),
model LowCardinality(String),
provider LowCardinality(String),
route LowCardinality(String),
user_id_hash FixedString(32),
status LowCardinality(String),
error_type LowCardinality(String),
latency_ms UInt32,
tokens_prompt UInt32,
tokens_completion UInt32,
cost_usd Decimal(12,6),
attrs JSON
) ENGINE = Kafka
SETTINGS
kafka_broker_list = ‘kafka:9092’,
kafka_topic_list = ‘agent_events_v1’,
kafka_group_name = ‘ch_agent_events_ingest’,
kafka_format = ‘JSONEachRow’,
kafka_num_consumers = 4;

— Final storage
CREATE TABLE agent_events (
ts DateTime64(3),
trace_id String,
span_id String,
parent_span_id String,
service LowCardinality(String),
env LowCardinality(String),
agent LowCardinality(String),
model LowCardinality(String),
provider LowCardinality(String),
route LowCardinality(String),
user_id_hash FixedString(32),
status LowCardinality(String),
error_type LowCardinality(String),
latency_ms UInt32,
tokens_prompt UInt32,
tokens_completion UInt32,
cost_usd Decimal(12,6),
attrs JSON
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(ts)
ORDER BY (ts, service, env, agent, route)
TTL ts + INTERVAL 90 DAY DELETE;

— Streaming materialization
CREATE MATERIALIZED VIEW mv_agent_events TO agent_events AS
SELECT
ts, trace_id, span_id, parent_span_id,
service, env, agent, model, provider, route,
user_id_hash, status, error_type,
latency_ms, tokens_prompt, tokens_completion, cost_usd, attrs
FROM kafka_agent_events;

— Rollup by minute
CREATE MATERIALIZED VIEW mv_agent_metrics_1m
ENGINE = SummingMergeTree
PARTITION BY toYYYYMM(ts_min)
ORDER BY (ts_min, service, env, agent, model, route)
AS
SELECT
toStartOfMinute(ts) AS ts_min,
service, env, agent, model, route,
count() AS calls,
quantileExact(0.5)(latency_ms) AS p50_ms,
quantileExact(0.95)(latency_ms) AS p95_ms,
sumIf(1, status != ‘ok’) / count() AS error_rate,
sum(tokens_prompt + tokens_completion) AS tokens_total,
sum(cost_usd) AS cost_usd
FROM agent_events
GROUP BY ts_min, service, env, agent, model, route;

OpenTelemetry Collector config
Receives spans/metrics over OTLP, drops PII, batches, pushes to Kafka.

receivers:
otlp:
protocols:
http:
grpc:

processors:
attributes:
actions:
– key: user_email
action: delete
– key: user_id
action: hash
batch:
timeout: 2s
send_batch_size: 5000
memory_limiter:
check_interval: 5s
limit_mib: 512
spike_limit_mib: 256
transform:
error_mode: ignore
traces:
– set(name, Concat(attributes[“agent”], “:”, attributes[“route”])) where name == “”

exporters:
kafka:
brokers: [ “kafka:9092” ]
topic: agent_events_v1
encoding: json
balancer: round_robin
retry_on_failure:
enabled: true
max_elapsed_time: 120s

service:
pipelines:
traces:
receivers: [otlp]
processors: [memory_limiter, attributes, transform, batch]
exporters: [kafka]
metrics:
receivers: [otlp]
processors: [memory_limiter, batch]
exporters: [kafka]

Client emission example (Python)
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace.export import BatchSpanProcessor

provider = TracerProvider(resource=Resource.create({
“service.name”: “api”,
“deployment.environment”: “prod”,
“agent”: “order_bot”,
“model”: “gpt-4.1”,
“provider”: “openai”,
“route”: “POST /v1/answer”
}))
trace.set_tracer_provider(provider)
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint=”http://otel-collector:4318/v1/traces”))
provider.add_span_processor(processor)

with trace.get_tracer(“api”).start_as_current_span(“agent_step”) as span:
# do work…
span.set_attribute(“tokens_prompt”, 512)
span.set_attribute(“tokens_completion”, 128)
span.set_attribute(“cost_usd”, 0.0032)
span.set_attribute(“status”, “ok”)
span.set_attribute(“user_id”, “12345”) # will be hashed by Collector

Metabase setup
– Add ClickHouse as a database (native driver).
– Set sync/scan to hourly for new fields.
– Create questions using native SQL.

Useful queries
– Errors by model (last 24h):
SELECT model, provider, count() AS errors
FROM agent_events
WHERE ts > now() – INTERVAL 1 DAY AND status != ‘ok’
GROUP BY model, provider
ORDER BY errors DESC
LIMIT 20;

– P95 latency by route (last 7d):
SELECT route, round(quantile(0.95)(latency_ms)) AS p95_ms, count() AS calls
FROM agent_events
WHERE ts > now() – INTERVAL 7 DAY
GROUP BY route
ORDER BY p95_ms DESC;

– Cost by customer (requires user_id map):
SELECT user_id_hash, sum(cost_usd) AS cost
FROM agent_events
WHERE ts > now() – INTERVAL 30 DAY
GROUP BY user_id_hash
ORDER BY cost DESC
LIMIT 50;

– SLO burn (errors > 2% in 15m windows):
SELECT
toStartOfInterval(ts, INTERVAL 15 minute) AS win,
round(avg(status != ‘ok’) * 100, 2) AS error_pct
FROM agent_events
WHERE ts > now() – INTERVAL 1 DAY
GROUP BY win
ORDER BY win DESC;

Alerting pattern
– Use a saved question with a threshold and Metabase email/Slack alerts.
– For low-latency alerts, add a lightweight worker polling ClickHouse every minute and posting to Slack webhook if error_pct > threshold.

Performance notes
– ClickHouse tuning:
– Use LowCardinality for strings; keep ORDER BY narrow and aligned with filters.
– Prefer DateTime64 for precise latency windows.
– Use TTL to control storage; hot data on fast disks.
– Kafka:
– Start with 3 partitions; increase to match ingest QPS.
– Retain 24–48h for replay.
– Otel Collector:
– Batch size 5k–10k; enable compression if cross-AZ.
– Cost:
– ClickHouse on a single NVMe host handles 50–100k events/sec; scale via shards + replicas.

Security and compliance
– Hash or drop user PII at the Collector.
– Restrict Metabase to read-only users.
– Use network policies or private subnets; encrypt Kafka and ClickHouse at rest and in transit.
– Rotate Kafka credentials and ClickHouse users; audit queries.

Deployment tips
– Docker Compose for a single-node pilot; Terraform + Kubernetes for prod.
– Health checks: Kafka consumer lag, Otel Collector queue size, ClickHouse insert delays.
– Backfill by producing historical JSON to the Kafka topic; ClickHouse will materialize.

What to build next
– Enrich events with billing account or feature flags.
– Add model drift metrics (output length, refusal rate).
– Expose a public status dashboard per-customer via Metabase embeds.

Goal
– Track agent runs, tool calls, latency, errors, tokens, and costs with second-level freshness.
– Keep ingestion cheap, queries fast, and schema stable as features evolve.

Reference architecture (text)
– Clients (agents, workers, webhooks) → OpenTelemetry SDK → OTLP HTTP → OpenTelemetry Collector → ClickHouse (native) → Metabase dashboards and alerts.
– Optional: Kafka between Collector and ClickHouse for burst smoothing.

Data model (minimally sufficient, append-only)
– events
– event_id (UUID)
– ts (DateTime64)
– app (String)
– env (Enum: prod, staging)
– run_id (UUID)
– type (Enum: run_started, run_finished, tool_call, error, token_usage)
– agent (String)
– user_id (String or UUID nullable)
– status (Enum: ok, error, timeout nullable)
– latency_ms (UInt32 nullable)
– tokens_prompt, tokens_completion (UInt32 nullable)
– cost_usd (Decimal(12,6) nullable)
– meta (JSON)

ClickHouse DDL
CREATE TABLE events
(
event_id UUID,
ts DateTime64(3, ‘UTC’),
app LowCardinality(String),
env Enum8(‘prod’ = 1, ‘staging’ = 2),
run_id UUID,
type LowCardinality(String),
agent LowCardinality(String),
user_id String,
status LowCardinality(String),
latency_ms UInt32,
tokens_prompt UInt32,
tokens_completion UInt32,
cost_usd Decimal(12,6),
meta JSON
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(ts)
ORDER BY (app, env, agent, ts)
TTL ts + INTERVAL 90 DAY DELETE
SETTINGS index_granularity = 8192;

Materialized views (rollups for fast dashboards)
CREATE MATERIALIZED VIEW mv_runs_hour
ENGINE = SummingMergeTree
PARTITION BY toYYYYMM(dt)
ORDER BY (app, env, agent, dt)
AS
SELECT
app, env, agent,
toStartOfHour(ts) AS dt,
countIf(type = ‘run_finished’) AS runs,
countIf(status = ‘error’) AS errors,
sumIf(latency_ms, type = ‘run_finished’) AS p50_proxy, — use quantiles below for real pXX
sum(tokens_prompt) AS tokens_prompt,
sum(tokens_completion) AS tokens_completion,
sum(cost_usd) AS cost_usd
FROM events
GROUP BY app, env, agent, dt;

For real latency percentiles, use AggregatingMergeTree with quantileState/quantileMerge:
– Store quantileState(0.5), (0.9), (0.99) and merge in queries.

Ingestion via OpenTelemetry Collector
otel-collector.yaml (core)
receivers:
otlp:
protocols:
http:
endpoint: 0.0.0.0:4318
exporters:
clickhouse:
endpoint: tcp://clickhouse:9000
database: default
logs_table_name: events
ttl: 90d
processors:
batch:
timeout: 1s
send_batch_size: 4096
service:
pipelines:
logs:
receivers: [otlp]
processors: [batch]
exporters: [clickhouse]

Note: We treat all telemetry as “logs” with structured bodies to keep one table. If you need traces, add traces pipeline and derive events via processors.

Python instrumentation (FastAPI worker)
from opentelemetry import trace, metrics
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
import logging, time, uuid, os, json

OTLP_ENDPOINT = os.getenv(“OTLP_HTTP”, “http://otel-collector:4318”)

# Logs as events
lp = LoggerProvider(resource=Resource.create({“service.name”:”agent-service”,”env”:”prod”}))
log_exporter = OTLPLogExporter(endpoint=f”{OTLP_ENDPOINT}/v1/logs”)
lp.add_log_record_processor(BatchSpanProcessor(log_exporter)) # SDK versions may use BatchLogRecordProcessor
logger = logging.getLogger(“ai-agent”)
logger.setLevel(logging.INFO)
logger.addHandler(LoggingHandler(level=logging.INFO, logger_provider=lp))

def emit_event(payload: dict):
logger.info(json.dumps(payload))

def run_agent(agent, user_id, prompt_tokens, completion_tokens, cost):
run_id = str(uuid.uuid4())
t0 = time.time()
emit_event({
“type”:”run_started”,”run_id”:run_id,”agent”:agent,
“app”:”myapp”,”env”:”prod”,”ts”:int(time.time()*1000)
})
try:
# … do work …
latency = int((time.time()-t0)*1000)
emit_event({
“type”:”run_finished”,”run_id”:run_id,”agent”:agent,
“status”:”ok”,”latency_ms”:latency,
“tokens_prompt”:prompt_tokens,
“tokens_completion”:completion_tokens,
“cost_usd”:float(cost),
“app”:”myapp”,”env”:”prod”,”ts”:int(time.time()*1000)
})
except Exception as e:
emit_event({
“type”:”error”,”run_id”:run_id,”agent”:agent,
“status”:”error”,”error_msg”:str(e),
“app”:”myapp”,”env”:”prod”,”ts”:int(time.time()*1000)
})

Node.js instrumentation (workers or Next.js routes)
import pino from ‘pino’
import { logs } from ‘@opentelemetry/api-logs’
import { LoggerProvider } from ‘@opentelemetry/sdk-logs’
import { OTLPLogExporter } from ‘@opentelemetry/exporter-logs-otlp-http’

const provider = new LoggerProvider()
provider.addLogRecordProcessor(new (require(‘@opentelemetry/sdk-logs’).BatchLogRecordProcessor)(
new OTLPLogExporter({ url: process.env.OTLP_HTTP + ‘/v1/logs’ })
))
logs.setGlobalLoggerProvider(provider)
const log = pino()

function emit(payload) { log.info(payload) }

emit({ type:’tool_call’, agent:’router’, run_id:’…’, latency_ms:32, app:’myapp’, env:’prod’, ts: Date.now() })

Metabase setup
– Connect to ClickHouse via the official driver.
– Mark events.ts as Time field. Create model views for:
– Runs by agent, environment
– Error rate by hour (countIf/status)
– P50/P90/P99 latency using quantile functions
– Token and cost per run, per user, per agent
– Parameterize app, env, agent to reuse dashboards across services.

Example SQL (Metabase card: error rate, last 24h)
SELECT
toStartOfMinute(ts) AS minute,
countIf(type = ‘run_finished’) AS runs,
countIf(status = ‘error’) AS errors,
(errors / nullIf(runs,0)) AS error_rate
FROM events
WHERE app = {{app}} AND env = {{env}} AND ts >= now() – INTERVAL 24 HOUR
GROUP BY minute
ORDER BY minute;

Alerts (Metabase → webhook/Slack)
– Threshold: error_rate > 0.05 for 5+ minutes
– Latency SLO breach: quantile(0.9)(latency_ms) > 2000 over last 15 minutes
– Cost guardrail: sum(cost_usd) > {{budget}} per day per agent

Backfill and idempotency
– If upstream replays, use event_id to dedupe:
CREATE TABLE events_dedup AS events ENGINE = ReplacingMergeTree(event_id)
…then insert into events_dedup and read from a view.
– For batch imports, use clickhouse-client with CSV/JSONEachRow. Preserve event_id.

Performance notes
– Aim for ≤1s collector batch timeout, 4–8k batch size.
– Keep meta JSON small and shallow. Move hot keys to top-level columns.
– Partition monthly, order by (app, env, agent, ts) for common filters.
– Prefer materialized aggregates for PXX latency and daily costs to keep dashboard queries <200ms.

Cost control
– TTL 90d raw, 365d for hourly aggregates.
– Compress with ZSTD; default is fine. Expect low storage/GB for numeric columns.
– Use Metabase cache 5–15 minutes on heavy cards.

Security and compliance
– Do not log PII. If needed, hash user_id and store mapping in a separate service.
– Enforce network policies: Collector and ClickHouse behind private network; Metabase via SSO.
– Rotate ClickHouse users; use readonly for BI.

Deployment tips
– Docker compose three services minimum: clickhouse, otel-collector, metabase.
– Health checks: Collector /healthz, ClickHouse system.metrics, Metabase /api/health.
– Sizing starter: 2 vCPU, 8GB RAM ClickHouse; 1 vCPU, 1GB Collector; 2 vCPU, 4GB Metabase. Scale storage IOPS first.

What to track next
– Tool-level success rate and latency
– Vendor breakdown (OpenAI, Anthropic, local) for spend and tokens
– Guardrails: rejection reasons, safe-output hits
– Queue metrics (depth, lag) for throughput planning

Summary
This stack ships real-time, production-grade observability for AI agents with minimal overhead. Start with OTLP → Collector → ClickHouse → Metabase, add materialized rollups for speed, and wire alerts to keep latency, reliability, and cost within SLOs.

This post shows how to deploy a lean, production-ready metrics pipeline for AI agents and automation workflows. The goal: near real-time visibility into latency, token/cost, error rates, tool call outcomes, and user/business impact.

Stack
– Ingestion API: FastAPI
– Queue: Kafka (preferred) or Redis Streams
– Storage: Postgres with TimescaleDB (hypertables) or vanilla Postgres with partitioning
– Transform: dbt (metrics, aggregations, SLOs)
– Dashboard: Metabase (fast to stand up), optional Grafana for alerts
– Orchestration: Docker Compose or Kubernetes
– Auth: API key with HMAC signature
– Optional: Celery/Prefect for scheduled jobs

Event model (core)
Capture minimal, consistent events to avoid schema drift.

Event types:
– agent_run_started
– agent_run_finished
– tool_call
– llm_call
– error
– business_event (e.g., qualified_lead_created)

Shared envelope:
– event_id (uuid)
– event_type (text)
– occurred_at (timestamptz)
– workspace_id (text)
– session_id (text) // user/session/job
– actor_id (text) // user_id, service_id
– source (text) // web, worker, cron
– context (jsonb) // free-form

LLM fields (on llm_call, agent_run_finished):
– model (text)
– input_tokens (int)
– output_tokens (int)
– prompt_hash (text)
– latency_ms (int)
– success (bool)
– error_type (text null)

Tool fields (on tool_call):
– tool_name (text)
– status (text) // success, failure, timeout
– latency_ms (int)

Cost fields (optional):
– provider (text)
– unit_cost_input (numeric)
– unit_cost_output (numeric)

Postgres schema
Use TimescaleDB if available. Otherwise, native partitions by day.

SQL (core events):
CREATE TABLE IF NOT EXISTS ai_events (
event_id uuid PRIMARY KEY,
event_type text NOT NULL,
occurred_at timestamptz NOT NULL,
workspace_id text NOT NULL,
session_id text,
actor_id text,
source text,
model text,
input_tokens int,
output_tokens int,
prompt_hash text,
latency_ms int,
success boolean,
error_type text,
tool_name text,
status text,
provider text,
unit_cost_input numeric(10,6),
unit_cost_output numeric(10,6),
context jsonb
);

Indexes:
– CREATE INDEX ON ai_events (occurred_at DESC);
– CREATE INDEX ON ai_events (workspace_id, occurred_at DESC);
– CREATE INDEX ON ai_events (event_type, occurred_at DESC);
– CREATE INDEX ON ai_events (model);
– CREATE INDEX ON ai_events USING GIN (context);

Timescale hypertable:
SELECT create_hypertable(‘ai_events’, by_range(‘occurred_at’), if_not_exists => true);

Partitioning fallback:
– Daily partitions via trigger or pg_partman.
– Retention policy: keep raw 30-90 days; aggregate forever.

Ingestion API (FastAPI)
– Accept batched events (up to 100).
– Verify HMAC signature.
– Validate via Pydantic.
– Push to Kafka topic ai_events or Redis stream ai_events.

Example (simplified):

from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel, Field
import hmac, hashlib, json, os
from datetime import datetime
from aiokafka import AIOKafkaProducer

SECRET = os.getenv(“INGEST_SECRET”)
KAFKA_BOOTSTRAP = os.getenv(“KAFKA_BOOTSTRAP”)

class Event(BaseModel):
event_id: str
event_type: str
occurred_at: datetime
workspace_id: str
session_id: str | None = None
actor_id: str | None = None
source: str | None = None
model: str | None = None
input_tokens: int | None = None
output_tokens: int | None = None
prompt_hash: str | None = None
latency_ms: int | None = None
success: bool | None = None
error_type: str | None = None
tool_name: str | None = None
status: str | None = None
provider: str | None = None
unit_cost_input: float | None = None
unit_cost_output: float | None = None
context: dict | None = None

class Batch(BaseModel):
events: list[Event] = Field(…, min_items=1, max_items=100)

app = FastAPI()
producer = None

@app.on_event(“startup”)
async def start():
global producer
producer = AIOKafkaProducer(bootstrap_servers=KAFKA_BOOTSTRAP)
await producer.start()

@app.on_event(“shutdown”)
async def stop():
await producer.stop()

def verify_sig(raw_body: bytes, sig: str):
mac = hmac.new(SECRET.encode(), raw_body, hashlib.sha256).hexdigest()
return hmac.compare_digest(mac, sig)

@app.post(“/ingest”)
async def ingest(request: Request):
raw = await request.body()
sig = request.headers.get(“X-Signature”) or “”
if not verify_sig(raw, sig):
raise HTTPException(401, “invalid signature”)
payload = Batch.model_validate_json(raw)
for e in payload.events:
await producer.send_and_wait(“ai_events”, json.dumps(e.model_dump()).encode())
return {“ok”: True, “count”: len(payload.events)}

Consumer (Kafka -> Postgres)
– Use a single writer per partition.
– Idempotent upsert on event_id.

Pseudo:

INSERT INTO ai_events (…) VALUES (…)
ON CONFLICT (event_id) DO NOTHING;

DBT models
Create clean aggregates for dashboards and SLOs.

models/marts/metrics/llm_daily.sql:
select
date_trunc(‘day’, occurred_at) as day,
workspace_id,
model,
count(*) filter (where event_type=’llm_call’) as calls,
percentile_cont(0.5) within group (order by latency_ms) as p50_latency_ms,
percentile_cont(0.9) within group (order by latency_ms) as p90_latency_ms,
sum(coalesce(input_tokens,0)) as input_tokens,
sum(coalesce(output_tokens,0)) as output_tokens,
sum(
coalesce(input_tokens,0)*coalesce(unit_cost_input,0)
+ coalesce(output_tokens,0)*coalesce(unit_cost_output,0)
) as cost_usd,
1.0*sum(case when success then 1 else 0 end)/nullif(count(*),0) as success_rate
from {{ ref(‘stg_ai_events’) }}
where event_type in (‘llm_call’,’agent_run_finished’)
group by 1,2,3;

models/marts/metrics/tool_reliability.sql:
select
date_trunc(‘hour’, occurred_at) as hour,
workspace_id,
tool_name,
count(*) as calls,
1.0*sum(case when status=’success’ then 1 else 0 end)/nullif(count(*),0) as success_rate,
avg(latency_ms) as avg_latency_ms
from {{ ref(‘stg_ai_events’) }}
where event_type=’tool_call’
group by 1,2,3;

SLOs and alerts
– Error budget: 99% success_rate per day on agent_run_finished.
– Latency SLO: p90_latency_ms < target per model.
– Cost guardrail: cost_usd per workspace per day threshold.

Metabase setup
Dashboards (suggested cards):
– LLM Overview
– Calls by model (daily)
– p50/p90 latency by model
– Token in/out trend
– Cost by model/provider
– Success rate over time
– Tool Reliability
– Success rate by tool (hourly)
– Failures by error_type
– Timeouts trend
– Agent Health
– Agent run success rate
– Average steps per run (from context.step_count)
– Top failing prompts (prompt_hash)
– Workspace Billing
– Daily cost per workspace
– Anomalies (z-score on daily spend)

Metabase tips
– Use SQL-native queries on dbt models.
– Add filters for workspace_id, model, date range.
– Cache at 5–15 minutes for near real-time.
– Create pulses (email/Slack) for SLO breaches.

Performance considerations
– Prefer Kafka with 3 partitions for steady ingestion; scale consumers horizontally.
– Batch inserts of 500–5k rows with COPY when backfilling.
– For Timescale: compress chunks older than 3 days; set retention to 90 days on raw.
– For vanilla Postgres: daily partitions + BRIN index on occurred_at.
– Keep jsonb context small; move hot keys to top-level columns when used in WHERE/JOIN.
– Use prompt_hash, not raw prompts, to avoid PII and save space.

Security and compliance
– HMAC signatures on ingest; rotate keys.
– Encrypt at rest (cloud-managed).
– Avoid logging raw user content; hash or redact.
– Separate writer and reader roles; Metabase gets read-only.

Deployment notes
– Docker Compose: kafka, zookeeper, postgres+timescale, fastapi, consumer, metabase.
– Health checks for producer/consumer.
– Use DLQ (dead-letter) topic for schema errors.
– Add OpenTelemetry tracing IDs to correlate events with app logs.

Quick wins
– If you don’t need Kafka, use Redis Streams; swap AIOKafka for aioredis and run a single consumer.
– If you need only dashboards, push events directly to Postgres via an async queue in the API and scale DB vertically.

What you get
– Real-time visibility into agent/LLM performance and cost.
– Traceable tool failures with impact by workspace.
– Guardrails with alerts to prevent runaway spend.
– A path to scale with minimal rework.

This post walks through a practical, low-ops real-time analytics pipeline for SaaS product metrics. It’s built for high write throughput, sub-second query speed, and simple ops:

– Ingest: FastAPI event endpoint
– Stream: Redpanda (Kafka-compatible)
– Storage/OLAP: ClickHouse
– Transform: dbt (on ClickHouse)
– Dashboards/alerts: Metabase
– Observability: OpenTelemetry + Prometheus + Grafana (pipeline health)

Target outcomes: DAU/WAU/MAU, activation funnels, retention cohorts, feature adoption, latency/error rates, and per-tenant views.

1) Event schema and contracts
Define a minimal, versioned schema so producers and consumers stay stable.

– Topic: product_events (partition by user_id or tenant_id)
– JSON payload:
{
“event”: “button_clicked”,
“user_id”: “u_123”,
“tenant_id”: “t_9”,
“session_id”: “s_abc”,
“ts”: “2026-03-14T18:25:43.511Z”,
“properties”: {“button_id”: “save”, “plan”: “pro”, “screen”: “editor”},
“context”: {“app_version”: “2.4.1”, “os”: “macOS”, “locale”: “en-US”},
“event_version”: 1
}

Rules:
– ISO-8601 UTC timestamps
– event_version increments on breaking changes
– Limit properties/context to primitive types
– Drop PII or hash it client-side

2) Ingestion API (FastAPI)
Expose a POST /events that validates payloads and writes to Redpanda.

– Use pydantic for validation
– Rate-limit per IP/tenant
– Buffer and batch send to Redpanda using librdkafka

Python snippet (core idea):
from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel, Field
from confluent_kafka import Producer
import json, time

p = Producer({“bootstrap.servers”: “redpanda:9092”, “queue.buffering.max.messages”: 100000})

class Event(BaseModel):
event: str
user_id: str
tenant_id: str
session_id: str
ts: str
properties: dict = Field(default_factory=dict)
context: dict = Field(default_factory=dict)
event_version: int = 1

app = FastAPI()

@app.post(“/events”)
async def ingest(ev: Event):
try:
p.produce(“product_events”, json.dumps(ev.dict()).encode(“utf-8”), key=ev.tenant_id)
p.poll(0)
return {“ok”: True}
except Exception:
raise HTTPException(status_code=503, detail=”Backpressure”)

3) Streaming into ClickHouse
ClickHouse consumes directly from Kafka-compatible topics via the Kafka engine and writes to a MergeTree table via a materialized view.

– Kafka table:
CREATE TABLE kafka_product_events (
event String,
user_id String,
tenant_id String,
session_id String,
ts DateTime64(3, ‘UTC’),
properties JSON,
context JSON,
event_version UInt8
) ENGINE = Kafka
SETTINGS kafka_broker_list = ‘redpanda:9092’,
kafka_topic_list = ‘product_events’,
kafka_group_name = ‘ch_consumer_v1’,
kafka_format = ‘JSONEachRow’,
kafka_num_consumers = 3;

– Target table:
CREATE TABLE product_events (
event LowCardinality(String),
user_id String,
tenant_id String,
session_id String,
ts DateTime64(3, ‘UTC’),
properties JSON,
context JSON,
event_version UInt8
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(ts)
ORDER BY (tenant_id, ts, event, user_id)
TTL ts + INTERVAL 365 DAY DELETE
SETTINGS index_granularity = 8192;

– Materialized view:
CREATE MATERIALIZED VIEW mv_kafka_to_events
TO product_events AS
SELECT
event, user_id, tenant_id, session_id,
parseDateTime64BestEffortOrNull(ts) AS ts,
properties, context, event_version
FROM kafka_product_events;

Notes:
– Use LowCardinality for string dims
– Partition by month for prune speed and cost
– TTL for retention

4) Aggregations for speed and cost
Pre-aggregate common metrics into rollup tables updated continuously.

– Daily user activity:
CREATE MATERIALIZED VIEW mv_events_to_daily
TO daily_user_activity AS
SELECT
tenant_id,
toDate(ts) AS d,
event,
uniqExact(user_id) AS uu
FROM product_events
GROUP BY tenant_id, d, event;

CREATE TABLE daily_user_activity (
tenant_id String,
d Date,
event LowCardinality(String),
uu UInt64
) ENGINE = ReplacingMergeTree()
ORDER BY (tenant_id, d, event);

– Sessionized funnels (optional): build session_start/session_end materialized views using window functions or handle in dbt.

5) Transformations with dbt (ClickHouse)
Use dbt-clickhouse for semantic models, derived metrics, and SCD dimensions.

– Models:
– stg_product_events: cast types, filter bad rows, standardize event names
– fct_event_counts_daily: DAU/WAU/MAU per tenant
– fct_funnels: configurable funnel steps by event and time window
– fct_retention: N-day retention by cohort_start = first_seen_date

Example dbt SQL idea:
select
tenant_id,
toDate(ts) as d,
countIf(event = ‘signup’) as signups,
uniqExactIf(user_id, event = ‘login’) as dau
from {{ ref(‘stg_product_events’) }}
group by tenant_id, d

6) Dashboards in Metabase
Connect Metabase directly to ClickHouse.

– Collections:
– Product KPIs: DAU/WAU/MAU, activation rate, feature adoption
– Growth: signups, conversion by channel (if captured)
– Reliability: event ingestion lag, API error rates
– Segmentation:
– Filters: tenant_id, plan, app_version, region
– Drill-through: user-level trails (guard with RBAC)
– Alerts:
– Pulses to Slack/Email when DAU drops > X% day-over-day
– Alert if ingestion lag > N minutes (query kafka table offsets vs. latest ts)

7) Health monitoring and SLIs
Track pipeline reliability separately from business metrics.

– Ingestion SLIs:
– HTTP 2xx rate and p95 latency for POST /events (Prometheus + FastAPI metrics)
– Backpressure count (producer queue size)
– Stream SLIs:
– Kafka consumer lag per partition (Redpanda metrics)
– Storage SLIs:
– ClickHouse insert errors, merges backlog, disk usage, parts count
– Dashboards:
– Grafana board combining FastAPI, Redpanda, ClickHouse exporters
– Alerts:
– Pager on consumer lag > threshold
– Disk usage > 80%
– Insert failures > 0.1% 5-min rate

8) Security and governance
– API: JWT per tenant, HMAC request signing, rate limits
– Data: avoid PII or hash client-side; encrypt at rest and in transit
– ClickHouse RBAC: read-only roles for BI, tenant row-level security using allow_read_policies
– Retention: enforce TTL; archive cold data to S3 via S3 table engine if needed

9) Deployment blueprint (Docker Compose)
Services:
– fastapi, redpanda, clickhouse, metabase, grafana, prometheus
– Optional: dbt runner (scheduled via cron/k8s Job)

Compose tips:
– Pin versions; mount persistent volumes for Redpanda/ClickHouse
– Separate networks for public API vs. internal data plane
– Set ClickHouse max_memory_usage, max_concurrent_queries
– Metabase env for ClickHouse driver; increase MB_JETTY_MAXTHREADS for load

10) Cost and performance notes
– Redpanda + ClickHouse can run on a single VM for MVP; scale to 3 nodes each later
– Expect 50k–200k events/sec on modest hardware with proper batching
– Keep event payloads < 2 KB; move large blobs to object storage
– Use materialized rollups for dashboards; avoid heavy ad-hoc scans during peak

11) Validation checklist
– Backfill: load historical CSV/Parquet to product_events via clickhouse-client
– Data quality: dbt tests for not_null, accepted_values, freshness
– Latency: end-to-end under 3s for most events
– Access: tenant-level filters enforced in Metabase

What you get
– Real-time dashboards that stay fast as volume grows
– Clear SLIs and actionable alerts
– A straightforward upgrade path from single node to HA

If you want a reference repo or a Compose file with sane defaults, reach out—we use this pattern in production for product analytics, ops telemetry, and feature adoption tracking.

If you run your app on Postgres and need sub-second analytics without crushing OLTP, this stack works reliably in production:
– Ingest: Postgres → ClickHouse (MaterializedPostgreSQL or Kafka + Debezium)
– Transform: dbt (dbt-clickhouse)
– Serve: Grafana dashboards + alerts

Use cases
– Funnel and retention analysis
– Feature usage and cohort metrics
– Operational dashboards (latency, errors, throughput)
– Finance-lite rollups (orders, MRR, refunds)

Reference architecture
– Source DB: Postgres 14+ on managed cloud (RDS/Cloud SQL)
– Analytics store: ClickHouse Cloud or self-managed cluster
– Replication: ClickHouse MaterializedPostgreSQL (for simplicity) or Kafka (for multi-sink / heavy transforms)
– Transform: dbt runner (GitHub Actions, Dagster, or Airflow)
– Viz: Grafana with ClickHouse plugin
– Storage/lineage: S3/GCS parquet exports (optional but recommended)
– Secrets: environment or Vault; network via PrivateLink/VPC peering

Table design in Postgres
– Prefer narrow event table + dimension tables
– Use UUID or bigint IDs; avoid wide JSONB for hot paths
– For events:
– event_id (UUID, PK)
– user_id
– occurred_at (timestamptz)
– event_name (text)
– properties (jsonb) — sparse, normalized over time
– For orders/subscriptions:
– immutable facts + status changes as events
– avoid in-place updates on hot rows

ClickHouse schema (denormalized for reads)
– Use MergeTree with time + shard key
– Example:

CREATE TABLE analytics.events
(
event_id UUID,
user_id String,
event_name LowCardinality(String),
occurred_at DateTime64(3, ‘UTC’),
properties_json JSON,
ingest_ts DateTime DEFAULT now()
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(occurred_at)
ORDER BY (event_name, occurred_at, user_id)
SETTINGS index_granularity = 8192;

– Add projections or materialized views for frequent queries (daily rollups, funnels)

Option A: Direct Postgres → ClickHouse via MaterializedPostgreSQL
Good when you control Postgres and schema churn is moderate.

Steps
1) Enable logical replication in Postgres:
– rds.logical_replication = 1 (RDS) or wal_level = logical
– Create a replication user with REPLICATION
2) In ClickHouse:

CREATE DATABASE pg_repl ENGINE = MaterializedPostgreSQL(
‘postgres-host:5432’, ‘app_db’, ‘replicator_user’, ‘REDACTED’,
‘public’, — schema
16 — max threads
);

— Exposed tables appear under pg_repl.

3) Hydrate analytics tables from the replicated ones:
CREATE TABLE analytics.events AS
SELECT
event_id,
user_id::String,
event_name,
occurred_at,
properties as properties_json
FROM pg_repl.events
ENGINE = MergeTree
PARTITION BY toYYYYMM(occurred_at)
ORDER BY (event_name, occurred_at, user_id);

Notes
– MaterializedPostgreSQL tails WAL and applies row-level changes
– For deletes/updates, ensure primary keys exist in Postgres
– For high write volume, separate heavy tables into their own ClickHouse databases to parallelize apply

Option B: Postgres → Kafka (Debezium) → ClickHouse
Use when you need multi-sink, schema registry, or custom buses.

– Debezium captures CDC to Kafka with Avro/JSON schema
– ClickHouse reads via Kafka engine + materialized views
– Handle upserts with ReplacingMergeTree or CollapsingMergeTree keyed by primary_id + version/flag

dbt transforms (dbt-clickhouse)
– Create semantic models and rollups; keep them small and incremental
– Example model: sessions_daily.sql

{{
config(
materialized=’incremental’,
unique_key=’session_day_user’,
incremental_strategy=’delete+insert’
)
}}

WITH sessions AS (
SELECT
user_id,
toStartOfDay(occurred_at) AS session_day,
countIf(event_name = ‘session_start’) AS starts,
countIf(event_name = ‘session_end’) AS ends
FROM {{ source(‘analytics’, ‘events’) }}
{% if is_incremental() %}
WHERE occurred_at >= date_sub(day, 3, today())
{% endif %}
GROUP BY user_id, session_day
)
SELECT
concat(user_id, ‘_’, toString(session_day)) AS session_day_user,
user_id,
session_day,
starts,
ends
FROM sessions;

– Schedule:
– Small models every 5 minutes
– Heavy rollups hourly
– Backfills in off-peak

Grafana dashboards
– Install ClickHouse data source
– Panels:
– Events per minute with 1m/5m rollups
– DAU/WAU/MAU with asof joins
– Feature adoption by cohort week
– p50/p95 latency by endpoint
– Error rate by service/version
– Add alert rules on thresholds (error_rate > 2%, DAU drop > 20% d/d)

Performance and cost tips
– ClickHouse:
– Partition by month, order by (event_name, occurred_at, user_id)
– Use LowCardinality for strings
– Keep JSON for long tail props, but extract top N to typed columns for filters
– Set max_threads per query; use quotas for noisy tenants
– dbt:
– Favor incremental + small windows
– Stage raw to clean, then marts; limit cross-joins
– Postgres:
– Keep CDC slot monitored; alert on replication lag
– Index primary keys; avoid mass vacuum stalls
– Storage:
– TTL old raw events to S3 via ClickHouse S3 TTL or periodic exports
– Costs:
– Use ClickHouse Cloud autoscaling with sane concurrency caps
– Compress JSON and avoid SELECT *

Reliability and ops
– Health checks:
– Replication delay (Postgres WAL LSN vs ClickHouse apply lag)
– Kafka consumer lag (if using Debezium)
– dbt run success rates and duration SLAs
– Schema changes:
– Additive first (new columns), then backfill, then drop
– Use dbt contracts/tests and column-level lineage
– Dedupe:
– For CDC upserts, use ReplacingMergeTree with a version column
– For at-least-once ingestion, dedupe on event_id in dbt staging
– Backfills:
– Snapshot Postgres to S3 (pg_dump or logical snapshot), bulk copy into ClickHouse, then reattach WAL

Security
– Restrict replication role to specific DB and tables
– Private networking between services
– Rotate secrets; limit Grafana viewer vs editor roles
– PII handling: hash/email_tokenize in ingestion; store raw PII in a separate restricted table

Quick start checklist
– Stand up ClickHouse Cloud and Grafana
– Enable Postgres logical replication
– Create MaterializedPostgreSQL database in ClickHouse
– Define analytics.events table and initial projections
– Configure dbt-clickhouse; build staging and marts
– Publish Grafana dashboards and alerts
– Add monitors for lag, error rates, and costs

What this delivers
– Sub-second reads on billions of events
– Minimal load on Postgres OLTP
– Clear, versioned transforms
– Actionable dashboards and alerts for product and ops