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

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 skips silently rather than failing requests.

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