ClickHouse Adapter

The ClickHouse adapter inserts wide events over the HTTP interface in JSONEachRow format. It works with a local instance, a self-managed cluster, and ClickHouse Cloud.

The default schema gives you typed columns for the fields you filter and aggregate on, plus the complete wide event as JSON in data — so no field is ever lost, and adding a field to your events never requires a migration.

Installation

The ClickHouse adapter comes bundled with evlog:

import { createClickHouseDrain } from 'evlog/clickhouse'

Create the table

Run this once before wiring the drain:

CREATE TABLE IF NOT EXISTS evlog_events
(
  timestamp     DateTime64(3, 'UTC'),
  level         LowCardinality(String),
  service       LowCardinality(String),
  environment   LowCardinality(String),
  request_id    String,
  trace_id      String,
  span_id       String,
  method        LowCardinality(String),
  path          String,
  status        Nullable(UInt16),
  duration      String,
  error_name    String,
  error_message String,
  data          String
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(timestamp)
ORDER BY (service, environment, timestamp)
TTL toDateTime(timestamp) + INTERVAL 30 DAY;

Quick Start

// server/plugins/evlog.ts
import { createClickHouseDrain } from 'evlog/clickhouse'

export default defineNitroPlugin((nitroApp) => {
  nitroApp.hooks.hook('evlog:drain', createClickHouseDrain())
})

import { Hono } from 'hono'
import { evlog } from 'evlog/hono'
import { createClickHouseDrain } from 'evlog/clickhouse'

const app = new Hono()
app.use(evlog({ drain: createClickHouseDrain() }))

import express from 'express'
import { evlog } from 'evlog/express'
import { createClickHouseDrain } from 'evlog/clickhouse'

const app = express()
app.use(evlog({ drain: createClickHouseDrain() }))

import Fastify from 'fastify'
import { evlog } from 'evlog/fastify'
import { createClickHouseDrain } from 'evlog/clickhouse'

const app = Fastify()
await app.register(evlog, { drain: createClickHouseDrain() })

import { Elysia } from 'elysia'
import { evlog } from 'evlog/elysia'
import { createClickHouseDrain } from 'evlog/clickhouse'

const app = new Elysia().use(evlog({ drain: createClickHouseDrain() }))

import { Module } from '@nestjs/common'
import { EvlogModule } from 'evlog/nestjs'
import { createClickHouseDrain } from 'evlog/clickhouse'

@Module({
  imports: [EvlogModule.forRoot({ drain: createClickHouseDrain() })],
})
export class AppModule {}

import { initLogger } from 'evlog'
import { createClickHouseDrain } from 'evlog/clickhouse'

initLogger({
  env: { service: 'my-app' },
  drain: createClickHouseDrain(),
})

Configuration

Environment variables

Options

Deployment

The adapter is identical either way — only the endpoint and credentials change.

Self-hosted

A local or self-managed instance, with or without auth:

createClickHouseDrain({ endpoint: 'http://localhost:8123' })

CLICKHOUSE_ENDPOINT=http://localhost:8123
# Only if your instance requires auth:
CLICKHOUSE_USER=evlog
CLICKHOUSE_PASSWORD=your-password

A fresh container runs as default with no password, which is why username defaults to default and password is optional.

ClickHouse Cloud

Cloud services always require credentials and speak HTTPS on port 8443:

createClickHouseDrain({
  endpoint: 'https://abc123.eu-west-1.aws.clickhouse.cloud:8443',
  password: process.env.CLICKHOUSE_PASSWORD,
  database: 'logs',
})

CLICKHOUSE_ENDPOINT=https://abc123.eu-west-1.aws.clickhouse.cloud:8443
CLICKHOUSE_PASSWORD=your-service-password
CLICKHOUSE_DATABASE=logs

Verify it locally

Spin up ClickHouse in Docker, create the table and push a real event:

docker compose -f packages/evlog/test/e2e/docker-compose.yml up -d
CLICKHOUSE_ENDPOINT=http://localhost:8123 pnpm run test:e2e
docker compose -f packages/evlog/test/e2e/docker-compose.yml down -v

The suite is a round-trip: it inserts events, reads them back with SELECT, and asserts the typed columns and the data JSON survived. The compose file creates evlog_events from the schema above on first boot. Without CLICKHOUSE_ENDPOINT it skips itself with a visible label.

To see the events rather than assert on them, seed the sandbox and open the provisioned dashboard — the stack ships a Grafana with the ClickHouse datasource already wired up:

pnpm run sandbox:up
pnpm run sandbox:seed
# → http://localhost:3001/d/evlog-wide-events

ClickHouse's built-in /play page at http://localhost:8123/play also works for raw SQL, but it is deliberately minimal — the dashboard is the better starting point.

Async inserts

Log ingestion means many small inserts, and one MergeTree part per request would quickly degrade a table. The adapter therefore enables asynchronous inserts by default, and does not wait for the flush:

async_insert=1&wait_for_async_insert=0

ClickHouse buffers rows server-side and writes them in batches. Draining never blocks a request on disk writes.

Set waitForAsyncInsert: true if you need the insert acknowledged as durable before the drain resolves — at the cost of latency. Set asyncInsert: false to insert synchronously, which only makes sense when you already batch client-side with evlog/pipeline.

Querying

Typed columns are indexed by the sort key; everything else lives in data and is reachable with ClickHouse's JSON functions:

-- Error rate per service over the last hour
SELECT service, countIf(level = 'error') / count() AS error_rate
FROM evlog_events
WHERE timestamp > now() - INTERVAL 1 HOUR
GROUP BY service;

-- Slowest routes
SELECT path, count() AS hits, avg(JSONExtractFloat(data, 'durationMs')) AS avg_ms
FROM evlog_events
WHERE timestamp > now() - INTERVAL 1 DAY AND method = 'GET'
GROUP BY path
ORDER BY avg_ms DESC
LIMIT 20;

-- One request end to end
SELECT timestamp, level, path, status, data
FROM evlog_events
WHERE request_id = '4a8ff3a8-...'
ORDER BY timestamp;

-- Reach into a custom field kept only in data
SELECT JSONExtractString(data, 'user', 'plan') AS plan, count()
FROM evlog_events
WHERE timestamp > now() - INTERVAL 7 DAY
GROUP BY plan;

Custom schema

Pass transform to target a table of your own shape:

import { createClickHouseDrain } from 'evlog/clickhouse'

createClickHouseDrain({
  table: 'app_logs',
  transform: event => ({
    ts: event.timestamp,
    lvl: event.level,
    svc: event.service,
    payload: JSON.stringify(event),
  }),
})

Keys must match your column names — ClickHouse rejects a JSONEachRow insert containing unknown columns.

Batching

Pair with a pipeline to insert in larger batches:

import { createDrainPipeline } from 'evlog/pipeline'
import { createClickHouseDrain } from 'evlog/clickhouse'
import type { DrainContext } from 'evlog'

const pipeline = createDrainPipeline<DrainContext>({
  batch: { size: 500, intervalMs: 5000 },
})

export default defineNitroPlugin((nitroApp) => {
  nitroApp.hooks.hook('evlog:drain', pipeline(createClickHouseDrain()))
})

Troubleshooting

Missing endpoint — CLICKHOUSE_ENDPOINT is unset and no endpoint was passed. The drain logs [evlog/clickhouse] Missing endpoint once and then does nothing, so requests are never blocked or failed.

ClickHouse API error: 401 — Check CLICKHOUSE_USER / CLICKHOUSE_PASSWORD. Credentials are sent as X-ClickHouse-User / X-ClickHouse-Key headers, never in the query string, so they never reach system.query_log.

ClickHouse API error: 400 mentioning unknown column — Your table does not match the row shape. Either create the table from the DDL above, or supply a transform matching your columns.

Cannot parse input — The table exists but a column type disagrees. status is Nullable(UInt16) because it is absent on non-HTTP events.

Direct API usage

import { sendBatchToClickHouse, sendToClickHouse } from 'evlog/clickhouse'

await sendToClickHouse(event, { endpoint: 'http://localhost:8123' })
await sendBatchToClickHouse(events, { endpoint: 'http://localhost:8123' })

toClickHouseRow(), toJSONEachRow(), and resolveClickHouseUrl() are also exported.

Next steps

• Sampling — control volume before it reaches ClickHouse
• Enrichers — add derived fields to every event
• Pipeline — batching, retries, and fan-out
• Adapters Overview — all available destinations