---
title: "自定义 Drain"
description: "为任何没有内置适配器的后端构建 Drain。配置解析、重试、超时和身份标头都由你处理。"
canonical_url: "https://evlog.zhcndoc.com/extend/custom-drains"
---
# 自定义 Drain

> 为任何没有内置适配器的后端构建 Drain。配置解析、重试、超时和身份标头都由你处理。

**drain** 是 evlog 流水线的终端步骤：一个接收宽事件并将其发送到某处的函数：HTTP API、消息队列、数据库、webhook、本地文件。evlog 为常见供应商提供了内置 drain（[适配器概览](https://evlog.zhcndoc.com/integrate/adapters/overview)）。当你需要发送到尚未覆盖的目标时，可以自行编写。

两个工厂覆盖所有情况：

<table>
<thead>
  <tr>
    <th>
      你有……
    </th>
    
    <th>
      使用
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      一个 HTTP 后端（REST、JSON ingest、供应商 <code>
        /v1/logs
      </code>
      
       端点）
    </td>
    
    <td>
      <a href="#definehttpdrain-http-%E9%85%8D%E6%96%B9">
        <code>
          defineHttpDrain
        </code>
      </a>
    </td>
  </tr>
  
  <tr>
    <td>
      一个非 HTTP 传输（gRPC、WebSocket、供应商 SDK、队列、原始套接字）
    </td>
    
    <td>
      <a href="#definedrain-%E9%9D%9E-http-%E4%BC%A0%E8%BE%93">
        <code>
          defineDrain
        </code>
      </a>
    </td>
  </tr>
</tbody>
</table>

两者都来自 `evlog/toolkit`，并且正是每个内置适配器所使用的工厂。

<prompt :actions="["copy","cursor","claude"]" description="构建一个自定义 evlog drain" icon="i-lucide-code-2">

构建一个自定义 evlog drain，将宽事件发送到没有内置适配器的后端。

- 对于 HTTP 后端，请使用 `evlog/toolkit` 中的 `defineHttpDrain({ name, resolve, encode })`，绝不要直接调用 `fetch`
- 对于非 HTTP 传输（队列、数据库、原生 SDK、原始套接字），请使用 `defineDrain({ name, send })`，并自行实现 `send(events)`
- 通过 `resolveAdapterConfig(namespace, fields, overrides)` 在 `resolve()` 内部延迟解析配置，让用户获得标准的优先级（overrides → `runtimeConfig.evlog.<ns>` → env）
- 使用标准化字段名称：承载令牌使用 `apiKey`，基础 URL 使用 `endpoint`，以及 `serviceName`、`timeout`
- 在 `encode(events, config)` 内将批量事件编码为目标端的传输格式，返回 `{ url, headers, body }`（或返回 `null` 以选择退出该批次）
- `defineHttpDrain` 会处理重试、超时、错误隔离、批处理和身份标头。不要重新实现这些功能
- 通过 `defineEvlog({ drain: createMyDrain() })` 或你的框架中间件的 `drain` 选项接入 drain
- 对于生产环境，请将结果包装在 `createDrainPipeline` 中，以实现批处理 + 重试

文档：[https://www.evlog.dev/extend/custom-drains](https://www.evlog.dev/extend/custom-drains)
流水线：[https://www.evlog.dev/extend/drain-pipeline](https://www.evlog.dev/extend/drain-pipeline)

<prompt>

## `defineHttpDrain`（HTTP 配方）

每个内置适配器都遵循的配方。两个纯函数：`resolve()` 返回配置（或 `null` 以跳过），`encode()` 返回 HTTP 请求负载。

<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] 缺少 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>
</prompt>
</prompt>

就是这样。`defineHttpDrain` 会处理重试（默认 2 次）、超时（默认 5000ms）、错误隔离以及身份标头（`User-Agent: evlog/<version>` + `X-Evlog-Source: <name>`）。即使目标服务宕机，你的应用流水线仍会继续运行；而当你将 drain 包装在 [`createDrainPipeline`](https://evlog.zhcndoc.com/extend/drain-pipeline) 中时，流水线会通过 drain 的 `raw` 变体接管重试。

### 一个 5 分钟示例：内部 Loki drain

一个完整可工作的 drain，只有 25 行，并且不需要外部配置助手：

```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)]],
        })),
      }),
    }),
  })
}
```

## 了解哪个配置优先

`resolveAdapterConfig(namespace, fields, overrides)` 会沿着标准链路查找，因此用户获得的配置体验与内置适配器相同：

1. 显式传给你的工厂的 `overrides`
2. `runtimeConfig.evlog.<namespace>`（Nuxt/Nitro）
3. `runtimeConfig.<namespace>`（旧版 Nuxt/Nitro）
4. `<NS>_<FIELD>` 环境变量（在 `ConfigField.env` 中列出 `NUXT_<NS>_<FIELD>`，以实现静默 Nuxt 兼容；在错误消息中仅通过 `formatPublicEnvKeys` 显示 `<NS>_<FIELD>`）

字段名称应遵循项目约定：`apiKey`、`endpoint`、`serviceName`、`timeout`。如果你要重命名现有字段（例如 `token` → `apiKey`），请在一个 major 版本中将两者都保留为 `ConfigField` 条目。有关弃用模式，请参见 `axiom.ts` 和 `better-stack.ts`。

## 将 drain 接入你的框架

一旦 `createMyServiceDrain()` 返回该 drain，就像其他任何东西一样接入它：

<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>

对于生产环境，将其包装一次在 [`createDrainPipeline`](https://evlog.zhcndoc.com/extend/drain-pipeline) 中，这样事件就会被批处理并重试。

## 发送前进行过滤或转换

`encode()` 接收完整的 `WideEvent[]` 批次以及解析后的配置。可以内联进行过滤或转换，返回 `null` 是选择退出该批次的简洁方式：

```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`（非 HTTP 传输）

如果你的目标需要 gRPC、供应商 SDK、队列客户端、WebSocket 或原始套接字，可以用 `defineDrain` 再往下一层。传输由你负责；工具包仍然会为你提供配置解析、错误隔离和一致的形状。

```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 })
    },
  })
```

让 `send` 在失败时抛出异常。`defineDrain` 会处理隔离：直接调用会吞掉错误并使用你的 drain 名称记录日志，而返回函数上的 `raw` 变体会拒绝，这正是[流水线](https://evlog.zhcndoc.com/extend/drain-pipeline)的重试和 `onDropped` 观察失败的方式。自行吞掉错误的 `send` 会使你的 drain 无法重试。

## DrainContext 携带的内容

当 evlog 通过 `evlog:drain` 调用你的 drain 时，它会为每个事件传递一个 `DrainContext`：

```typescript [types.ts]
interface DrainContext {
  /** 包含所有已累积上下文的完整宽事件 */
  event: WideEvent

  /** 请求元数据 */
  request?: {
    method: string
    path: string
    requestId: string
  }

  /** 安全的 HTTP 标头（已过滤敏感标头） */
  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
  // ... 以及通过 log.set() 添加的所有字段
  [key: string]: unknown
}
```

在 `encode()` / `send()` 接收的批量形式中，你会直接得到 `WideEvent[]`（工具包会将每个上下文中的 `event` 解包）。

## 使用工具包辅助函数

`evlog/toolkit` 暴露了每个内置适配器都会用到的相同辅助函数。与 drains 相关的有：

<table>
<thead>
  <tr>
    <th>
      导出
    </th>
    
    <th>
      用途
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        defineHttpDrain(spec)
      </code>
    </td>
    
    <td>
      HTTP 配方——自动重试、超时、身份标头、错误隔离
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        defineDrain(spec)
      </code>
    </td>
    
    <td>
      非 HTTP 传输的相同契约
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        resolveAdapterConfig(ns, fields, overrides)
      </code>
    </td>
    
    <td>
      标准配置优先级链（overrides → <code>
        runtimeConfig.evlog.<ns>
      </code>
      
       → env）
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        httpPost(opts)
      </code>
    </td>
    
    <td>
      每个内置 HTTP 适配器使用的重试 POST 辅助函数——处理超时、重试、脱敏错误消息
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        composeDrains(drains)
      </code>
    </td>
    
    <td>
      将多个 drain 合并为一个（错误隔离，并通过 <code>
        Promise.allSettled
      </code>
      
       并发运行）
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        toTypedAttributeValue(value)
      </code>
    </td>
    
    <td>
      将任意值转换为 Axiom / Sentry 使用的 typed attribute 形状
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        toOtlpAttributeValue(value)
      </code>
    </td>
    
    <td>
      将任意值转换为 OTLP <code>
        AnyValue
      </code>
      
       形状（用于 OTLP / HyperDX / PostHog 日志）
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        OTEL_SEVERITY_NUMBER
      </code>
      
      , <code>
        OTEL_SEVERITY_TEXT
      </code>
    </td>
    
    <td>
      OTEL 日志严重级别表
    </td>
  </tr>
</tbody>
</table>

## 向接收方标识你的流量

`defineHttpDrain` 会自动为每个请求添加两个头部，方便接收方识别流量：

<table>
<thead>
  <tr>
    <th>
      标头
    </th>
    
    <th>
      值
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        User-Agent
      </code>
    </td>
    
    <td>
      <code>
        evlog/<version>
      </code>
      
      （仅限 Node / 服务器运行时——浏览器会移除此头部）
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        X-Evlog-Source
      </code>
    </td>
    
    <td>
      你提供的 drain <code>
        name
      </code>
    </td>
  </tr>
</tbody>
</table>

如果你直接基于 `httpPost` 构建 drain，则可以覆盖或抑制这些标头。参见[身份标头](https://evlog.zhcndoc.com/extend/identity-headers)。

## 错误已经由你处理

`defineHttpDrain` 会自动强制执行所有最佳实践：

1. **绝不抛出异常**：捕获失败并使用 `[evlog/<name>]` 前缀记录日志。
2. **重试**：对于临时错误，默认尝试 2 次（可通过 `retries` 配置）。
3. **超时**：默认 5000ms（可通过 `timeout` 配置）。
4. **优雅降级**：`resolve()` 返回 `null` 会使 drain 成为无操作。

如果你回退使用 `defineDrain`，请手动遵循相同规则。

## 将其发布为社区软件包

社区 drain 的推荐结构：

```text
my-evlog-drain/
├─ src/
│  ├─ drain.ts        # 通过 defineHttpDrain 创建 createMyDrain
│  └─ index.ts        # 重新导出
├─ test/              # vitest，mock fetch
├─ package.json       # peerDependency: "evlog"
└─ README.md
```

将 `evlog` 添加为 `peerDependency`（而不是 `dependency`），这样你的软件包在安装时不会拉取一份 evlog 副本。

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

构建了很棒的东西？[提交 PR](https://github.com/evloghq/evlog/pulls)，为适配器表格添加一行。社区会感谢你。

</callout>

## 下一步

- [Drain 流水线](https://evlog.zhcndoc.com/extend/drain-pipeline)：在生产环境中将你的 drain 包装在批处理 + 重试 + 扇出中
- [适配器概览](https://evlog.zhcndoc.com/integrate/adapters/overview)：了解内置适配器如何使用 `defineHttpDrain`
- [自定义 Enricher](https://evlog.zhcndoc.com/extend/custom-enrichers)：用于派生事件字段的相同工具包形状
- [自定义框架集成](https://evlog.zhcndoc.com/extend/custom-framework)：用于 HTTP 框架的相同工具包形状
- [最佳实践](https://evlog.zhcndoc.com/reference/best-practices)：安全和生产环境技巧。

---

- [Drain 流水线](https://evlog.zhcndoc.com/extend/drain-pipeline)
- [适配器概览](https://evlog.zhcndoc.com/integrate/adapters/overview)
- [最佳实践](https://evlog.zhcndoc.com/reference/best-practices)


## Sitemap

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