---
title: "Custom Drains"
description: "Build a drain for any backend without a built-in adapter. Config resolution, retries, timeouts, and identity headers are handled for you."
canonical_url: "https://www.evlog.dev/extend/custom-drains"
---
# Custom Drains

> Build a drain for any backend without a built-in adapter. Config resolution, retries, timeouts, and identity headers are handled for you.

A **drain** is the terminal step of evlog's pipeline: a function that receives wide events and ships them somewhere: an HTTP API, a message queue, a database, a webhook, a local file. evlog ships built-in drains for popular providers ([Adapters overview](https://www.evlog.dev/integrate/adapters/overview)). When you need a destination that isn't covered, you write your own.

Two factories cover every case:

<table>
<thead>
  <tr>
    <th>
      You have…
    </th>
    
    <th>
      Use
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      An HTTP backend (REST, JSON ingest, vendor <code>
        /v1/logs
      </code>
      
       endpoint)
    </td>
    
    <td>
      <a href="#definehttpdrain-the-http-recipe">
        <code>
          defineHttpDrain
        </code>
      </a>
    </td>
  </tr>
  
  <tr>
    <td>
      A non-HTTP transport (gRPC, WebSocket, vendor SDK, queue, raw socket)
    </td>
    
    <td>
      <a href="#definedrain-non-http-transports">
        <code>
          defineDrain
        </code>
      </a>
    </td>
  </tr>
</tbody>
</table>

Both come from `evlog/toolkit` and are the exact factories every built-in adapter uses.

<prompt :actions="["copy","cursor","claude"]" description="Build a custom evlog drain" icon="i-lucide-code-2">

Build a custom evlog drain that ships wide events to a backend without a built-in adapter.

- For HTTP backends, use `defineHttpDrain({ name, resolve, encode })` from `evlog/toolkit`, and never call `fetch` directly
- For non-HTTP transports (queue, DB, native SDK, raw socket), use `defineDrain({ name, send })` and implement `send(events)` myself
- Resolve config lazily inside `resolve()` via `resolveAdapterConfig(namespace, fields, overrides)` so users get the standard precedence (overrides → `runtimeConfig.evlog.<ns>` → env)
- Use the standardized field names: `apiKey` for bearer secrets, `endpoint` for the base URL, `serviceName`, `timeout`
- Encode batched events into the destination's wire format inside `encode(events, config)`, return `{ url, headers, body }` (or `null` to opt out of the batch)
- `defineHttpDrain` handles retries, timeouts, error isolation, batching, and identity headers. Don't reimplement them
- Wire the drain via `defineEvlog({ drain: createMyDrain() })` or my framework's middleware `drain` option
- For production, wrap the result in `createDrainPipeline` for batching + retries

Docs: [https://www.evlog.dev/extend/custom-drains](https://www.evlog.dev/extend/custom-drains)
Pipeline: [https://www.evlog.dev/extend/drain-pipeline](https://www.evlog.dev/extend/drain-pipeline)

</prompt>

## `defineHttpDrain` (the HTTP recipe)

The recipe every built-in adapter follows. Two pure functions: `resolve()` returns the config (or `null` to skip), `encode()` returns the HTTP request payload.

<code-collapse>

```typescript [lib/my-drain.ts]
import {
  defineHttpDrain,
  resolveAdapterConfig,
  type ConfigField,
} from 'evlog/toolkit'

interface MyServiceConfig {
  apiKey: string
  endpoint?: string
  timeout?: number
}

const FIELDS: ConfigField<MyServiceConfig>[] = [
  { key: 'apiKey', env: ['MYSERVICE_API_KEY'] },
  { key: 'endpoint', env: ['MYSERVICE_ENDPOINT'] },
  { key: 'timeout' },
]

export function createMyServiceDrain(overrides?: Partial<MyServiceConfig>) {
  return defineHttpDrain<MyServiceConfig>({
    name: 'myservice',
    resolve: async () => {
      const cfg = await resolveAdapterConfig<MyServiceConfig>('myservice', FIELDS, overrides)
      if (!cfg.apiKey) {
        console.error('[evlog/myservice] Missing apiKey')
        return null
      }
      return cfg as MyServiceConfig
    },
    encode: (events, cfg) => ({
      url: `${cfg.endpoint ?? 'https://api.myservice.com'}/v1/ingest`,
      headers: {
        'Content-Type': 'application/json',
        Authorization: `Bearer ${cfg.apiKey}`,
      },
      body: JSON.stringify(events),
    }),
  })
}
```

</code-collapse>

That's it. `defineHttpDrain` handles retries (default 2), timeouts (default 5000ms), error isolation, and the identity headers (`User-Agent: evlog/<version>` + `X-Evlog-Source: <name>`). Your app pipeline keeps running even if your destination is down, and when you wrap the drain in [`createDrainPipeline`](https://www.evlog.dev/extend/drain-pipeline), the pipeline takes over retries through the drain's `raw` variant.

### A 5-minute example: internal Loki drain

A complete working drain in 25 lines, with no external config helper:

```ts [lib/loki-drain.ts]
import { defineHttpDrain } from 'evlog/toolkit'

export function createLokiDrain(overrides?: { url?: string, token?: string }) {
  return defineHttpDrain<{ url: string, token: string }>({
    name: 'loki',
    resolve: () => ({
      url: overrides?.url ?? process.env.LOKI_URL!,
      token: overrides?.token ?? process.env.LOKI_TOKEN!,
    }),
    encode: (events, config) => ({
      url: `${config.url}/loki/api/v1/push`,
      headers: {
        'Content-Type': 'application/json',
        Authorization: `Bearer ${config.token}`,
      },
      body: JSON.stringify({
        streams: events.map(e => ({
          stream: { service: e.service, level: e.level },
          values: [[String(Date.parse(e.timestamp) * 1e6), JSON.stringify(e)]],
        })),
      }),
    }),
  })
}
```

## Know which config wins

`resolveAdapterConfig(namespace, fields, overrides)` walks the standard chain so users get the same configuration UX as built-in adapters:

1. Explicit `overrides` passed to your factory
2. `runtimeConfig.evlog.<namespace>` (Nuxt/Nitro)
3. `runtimeConfig.<namespace>` (legacy Nuxt/Nitro)
4. `<NS>_<FIELD>` env vars (list `NUXT_<NS>_<FIELD>` in `ConfigField.env` for silent Nuxt compat; show only `<NS>_<FIELD>` in error messages via `formatPublicEnvKeys`)

Field names should follow the project conventions: `apiKey`, `endpoint`, `serviceName`, `timeout`. If you're renaming an existing field (e.g. `token` → `apiKey`), keep both as `ConfigField` entries for one major version. See `axiom.ts` and `better-stack.ts` for the deprecation pattern.

## Wire the drain into your framework

Once `createMyServiceDrain()` returns the drain, wire it like any other:

<framework-tabs>

```typescript [Nuxt]server/plugins/evlog-drain.ts
import { createMyServiceDrain } from '~/server/utils/my-drain'

const drain = createMyServiceDrain()

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

```typescript [Nitro]server/plugins/evlog-drain.ts
import { definePlugin } from 'nitro'
import { createMyServiceDrain } from '~/server/utils/my-drain'

const drain = createMyServiceDrain()

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

```typescript [Next.js]lib/evlog.ts
import { createEvlog } from 'evlog/next'
import { createMyServiceDrain } from './my-drain'

export const { withEvlog, useLogger, log, createError } = createEvlog({
  service: 'my-app',
  drain: createMyServiceDrain(),
})
```

```typescript [SvelteKit]src/hooks.server.ts
import { createEvlogHooks } from 'evlog/sveltekit'
import { createMyServiceDrain } from './my-drain'

export const { handle, handleError } = createEvlogHooks({ drain: createMyServiceDrain() })
```

```typescript [TanStack Start]server/plugins/evlog-drain.ts
import { definePlugin } from 'nitro'
import { createMyServiceDrain } from '~/server/utils/my-drain'

const drain = createMyServiceDrain()

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

```typescript [React Router]app/root.tsx
import { evlog } from 'evlog/react-router'
import { createMyServiceDrain } from './my-drain'

export const middleware: Route.MiddlewareFunction[] = [
  evlog({ drain: createMyServiceDrain() }),
]
```

```typescript [Hono]src/index.ts
import { evlog } from 'evlog/hono'
import { createMyServiceDrain } from './my-drain'

app.use(evlog({ drain: createMyServiceDrain() }))
```

```typescript [Express]src/index.ts
import { evlog } from 'evlog/express'
import { createMyServiceDrain } from './my-drain'

app.use(evlog({ drain: createMyServiceDrain() }))
```

```typescript [Fastify]src/index.ts
await app.register(evlog, { drain: createMyServiceDrain() })
```

```typescript [Elysia]src/index.ts
import { evlog } from 'evlog/elysia'
import { createMyServiceDrain } from './my-drain'

app.use(evlog({ drain: createMyServiceDrain() }))
```

```typescript [NestJS]src/app.module.ts
EvlogModule.forRoot({ drain: createMyServiceDrain() })
```

```typescript [oRPC]server/orpc.ts
import { withEvlog } from 'evlog/orpc'
import { createMyServiceDrain } from './my-drain'

const handler = withEvlog(new RPCHandler(router), { drain: createMyServiceDrain() })
```

```typescript [Cloudflare Workers]src/worker.ts
import { withEvlog } from 'evlog/workers'
import { createMyServiceDrain } from './my-drain'

export default withEvlog(handler, { drain: createMyServiceDrain() })
```

```typescript [Standalone]
import { initLogger } from 'evlog'
import { createMyServiceDrain } from './my-drain'

initLogger({ drain: createMyServiceDrain() })
```

</framework-tabs>

For production, wrap it once in [`createDrainPipeline`](https://www.evlog.dev/extend/drain-pipeline) so events are batched and retried.

## Filter or transform before sending

`encode()` receives the full batch of `WideEvent[]` plus the resolved config. Filter or transform inline, and returning `null` is a clean opt-out for that batch:

```typescript
encode: (events, cfg) => {
  const filtered = events.filter(e => e.level === 'error' && e.path !== '/health')
  if (filtered.length === 0) return null

  const payload = filtered.map(e => ({
    ts: new Date(e.timestamp).getTime(),
    severity: e.level.toUpperCase(),
    attributes: { method: e.method, path: e.path, status: e.status, durationMs: e.durationMs },
  }))

  return {
    url: `${cfg.endpoint}/v1/push`,
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(payload),
  }
}
```

## `defineDrain` (non-HTTP transports)

If your destination requires gRPC, a vendor SDK, a queue client, a WebSocket, or a raw socket, drop one level lower with `defineDrain`. You own the transport; the toolkit still gives you config resolution, error isolation, and a consistent shape.

```typescript
import { defineDrain } from 'evlog/toolkit'

export const createCustomTransportDrain = () =>
  defineDrain<{ apiKey: string }>({
    name: 'custom',
    resolve: async () => ({ apiKey: process.env.MY_KEY! }),
    send: async (events, cfg) => {
      await myVendorSdk.publish(events, { token: cfg.apiKey })
    },
  })
```

Let `send` throw on failure. `defineDrain` handles the isolation: a direct call swallows the error and logs it with your drain's name, while the `raw` variant on the returned function rejects, which is how [the pipeline](https://www.evlog.dev/extend/drain-pipeline)'s retry and `onDropped` observe failures. A `send` that swallows its own errors makes your drain unretryable.

## What a DrainContext carries

When evlog calls your drain through `evlog:drain`, it passes a `DrainContext` per event:

```typescript [types.ts]
interface DrainContext {
  /** The complete wide event with all accumulated context */
  event: WideEvent

  /** Request metadata */
  request?: {
    method: string
    path: string
    requestId: string
  }

  /** Safe HTTP headers (sensitive headers filtered) */
  headers?: Record<string, string>
}

interface WideEvent {
  timestamp: string
  level: 'debug' | 'info' | 'warn' | 'error'
  service: string
  environment?: string
  version?: string
  region?: string
  commitHash?: string
  requestId?: string
  // ... plus all fields added via log.set()
  [key: string]: unknown
}
```

In the batched form your `encode()` / `send()` receives, you get `WideEvent[]` directly (the toolkit unwraps `event` from each context).

## Reach for a toolkit helper

`evlog/toolkit` exposes the same helpers every built-in adapter uses. The ones relevant to drains:

<table>
<thead>
  <tr>
    <th>
      Export
    </th>
    
    <th>
      Purpose
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        defineHttpDrain(spec)
      </code>
    </td>
    
    <td>
      The HTTP recipe — auto retries, timeouts, identity headers, error isolation
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        defineDrain(spec)
      </code>
    </td>
    
    <td>
      Same contract for non-HTTP transports
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        resolveAdapterConfig(ns, fields, overrides)
      </code>
    </td>
    
    <td>
      Standard config priority chain (overrides → <code>
        runtimeConfig.evlog.<ns>
      </code>
      
       → env)
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        httpPost(opts)
      </code>
    </td>
    
    <td>
      The retried POST helper used by every built-in HTTP adapter — handles timeout, retries, redacted error messages
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        composeDrains(drains)
      </code>
    </td>
    
    <td>
      Combine multiple drains into one (errors isolated, runs concurrently with <code>
        Promise.allSettled
      </code>
      
      )
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        toTypedAttributeValue(value)
      </code>
    </td>
    
    <td>
      Convert any value to the typed attribute shape used by Axiom / Sentry
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        toOtlpAttributeValue(value)
      </code>
    </td>
    
    <td>
      Convert any value to the OTLP <code>
        AnyValue
      </code>
      
       shape (used by OTLP / HyperDX / Sazabi / PostHog logs)
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        OTEL_SEVERITY_NUMBER
      </code>
      
      , <code>
        OTEL_SEVERITY_TEXT
      </code>
    </td>
    
    <td>
      OTEL log severity tables
    </td>
  </tr>
</tbody>
</table>

## Identify your traffic to the receiver

`defineHttpDrain` automatically tags every request with two headers so receivers can identify the traffic:

<table>
<thead>
  <tr>
    <th>
      Header
    </th>
    
    <th>
      Value
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        User-Agent
      </code>
    </td>
    
    <td>
      <code>
        evlog/<version>
      </code>
      
       (Node / server runtimes only — browsers strip this header)
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        X-Evlog-Source
      </code>
    </td>
    
    <td>
      The drain <code>
        name
      </code>
      
       you provided
    </td>
  </tr>
</tbody>
</table>

If you build a drain on top of `httpPost` directly, you can override or suppress them. See [Identity headers](https://www.evlog.dev/extend/identity-headers).

## Errors are already handled for you

`defineHttpDrain` enforces every best practice automatically:

1. **Never throws**: failures are caught and logged with the `[evlog/<name>]` prefix.
2. **Retries**: defaults to 2 attempts on transient errors (configurable via `retries`).
3. **Timeouts**: defaults to 5000ms (configurable via `timeout`).
4. **Graceful degradation**: `resolve()` returning `null` makes the drain a no-op.

If you fall back to `defineDrain`, follow the same rules manually.

## Publish it as a community package

Recommended structure for a community drain:

```text
my-evlog-drain/
├─ src/
│  ├─ drain.ts        # createMyDrain via defineHttpDrain
│  └─ index.ts        # re-exports
├─ test/              # vitest, mock fetch
├─ package.json       # peerDependency: "evlog"
└─ README.md
```

Add `evlog` as a `peerDependency` (not a `dependency`), so your package will not pull in a copy of evlog at install time.

<callout color="neutral" icon="i-lucide-heart">

Built something great? [Open a PR](https://github.com/evloghq/evlog/pulls) to add a row to the Adapters table. The community will thank you.

</callout>

## Next steps

- [Drain Pipeline](https://www.evlog.dev/extend/drain-pipeline): wrap your drain in batch + retry + fanout for production
- [Adapters Overview](https://www.evlog.dev/integrate/adapters/overview): see how the built-in adapters use `defineHttpDrain`
- [Custom Enrichers](https://www.evlog.dev/extend/custom-enrichers): same toolkit shape for derived event fields
- [Custom Framework Integration](https://www.evlog.dev/extend/custom-framework): same toolkit shape for HTTP frameworks
- [Best Practices](https://www.evlog.dev/reference/best-practices): security and production tips

---

- [Drain Pipeline](https://www.evlog.dev/extend/drain-pipeline)
- [Adapters Overview](https://www.evlog.dev/integrate/adapters/overview)
- [Best Practices](https://www.evlog.dev/reference/best-practices)


## Sitemap

See the full [sitemap](https://www.evlog.dev/sitemap.md) for all pages.