---
title: "Drain 流水线"
description: "该流水线包装每个 drain：它会批量处理事件、在失败时重试、将事件分发到多个目标，并发送浏览器日志。"
canonical_url: "https://evlog.zhcndoc.com/extend/drain-pipeline"
---
# Drain 流水线

> 该流水线包装每个 drain：它会批量处理事件、在失败时重试、将事件分发到多个目标，并发送浏览器日志。

在生产环境中，为每个发出的事件发送一个 HTTP 请求无法扩展。drain pipeline 会缓冲事件并批量发送，在临时失败时重试，缓冲区溢出时丢弃最旧的事件，并允许单个 drain 函数并行分发到多个目标。用于客户端日志的 [HTTP browser drain](#http-drain-browser-to-server) 也由同一条流水线提供支持。

<drain-pipeline-batching>



</drain-pipeline-batching>

<table>
<thead>
  <tr>
    <th>
      你想要…
    </th>
    
    <th>
      查看
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      将任意 drain 包装为批量 + 重试 + 缓冲
    </td>
    
    <td>
      <a href="#quick-start">
        快速开始
      </a>
    </td>
  </tr>
  
  <tr>
    <td>
      将每个事件并行发送到多个目标
    </td>
    
    <td>
      <a href="#fanout">
        Fanout
      </a>
    </td>
  </tr>
  
  <tr>
    <td>
      将浏览器日志发送到你的服务器端点
    </td>
    
    <td>
      <a href="#http-drain-browser-to-server">
        HTTP drain（浏览器到服务器）
      </a>
    </td>
  </tr>
  
  <tr>
    <td>
      调整批量大小、重试策略、缓冲区大小
    </td>
    
    <td>
      <a href="#configuration">
        配置
      </a>
    </td>
  </tr>
</tbody>
</table>

<prompt :actions="["copy","cursor","claude"]" description="添加 drain 流水线（批量 + 重试 + 分发）" icon="i-lucide-workflow">

将我的 evlog drain 包装到共享流水线中，让它能够批量处理、重试，并在临时失败时继续运行。

- 从 `evlog/pipeline` 导入 `createDrainPipeline`
- 包装底层 drain：`const drain = createDrainPipeline<DrainContext>()(createAxiomDrain())`
- 配置 `batch`（大小 + intervalMs）、`retry`（maxAttempts + backoff）和 `maxBufferSize`（溢出时丢弃最旧的事件）。合理的默认值：`{ batch: { size: 50, intervalMs: 5000 }, retry: { maxAttempts: 3 } }`
- 对于多个目标，编写一个在内部进行分发的单一 drain 函数，使用 `Promise.allSettled([drainA(batch), drainB(batch)])`，然后将其传递给 `pipeline(...)`。一个共享缓冲区，一个共享重试预算
- 关闭时调用 `drain.flush()` 以发送缓冲的事件（Nitro 使用 `close` hook，Standalone 在 `process.exit` 前调用）。在 serverless 环境中，将流水线作为 `drain` 选项传递即可：中间件会向运行时的 `waitUntil` 注册 `drain.settled()`
- 在生产环境中始终使用流水线。直接使用 drain 会为每个事件发起一次 HTTP 请求，并且很快就会崩溃

文档：[https://www.evlog.dev/extend/drain-pipeline](https://www.evlog.dev/extend/drain-pipeline)
Adapters：[https://www.evlog.dev/integrate/adapters/overview](https://www.evlog.dev/integrate/adapters/overview)

</prompt>

## 快速开始

该流水线可以包装任意 drain。具体接入方式取决于你的框架，因此请选择与你使用的框架匹配的选项卡；下面的其他示例都使用相同的结构。

<framework-tabs>

```typescript [Nuxt]server/plugins/evlog-drain.ts
import type { DrainContext } from 'evlog'
import { createDrainPipeline } from 'evlog/pipeline'
import { createAxiomDrain } from 'evlog/axiom'

export default defineNitroPlugin((nitroApp) => {
  const pipeline = createDrainPipeline<DrainContext>()
  const drain = pipeline(createAxiomDrain())

  nitroApp.hooks.hook('evlog:drain', drain)
  nitroApp.hooks.hook('close', () => drain.flush())
})
```

```typescript [Nitro]server/plugins/evlog-drain.ts
import { definePlugin } from 'nitro'
import type { DrainContext } from 'evlog'
import { createDrainPipeline } from 'evlog/pipeline'
import { createAxiomDrain } from 'evlog/axiom'

export default definePlugin((nitroApp) => {
  const pipeline = createDrainPipeline<DrainContext>()
  const drain = pipeline(createAxiomDrain())

  nitroApp.hooks.hook('evlog:drain', drain)
  nitroApp.hooks.hook('close', () => drain.flush())
})
```

```typescript [Next.js]lib/evlog.ts
import type { DrainContext } from 'evlog'
import { createEvlog } from 'evlog/next'
import { createDrainPipeline } from 'evlog/pipeline'
import { createAxiomDrain } from 'evlog/axiom'

const pipeline = createDrainPipeline<DrainContext>()
const drain = pipeline(createAxiomDrain())

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

export const flushEvlog = () => drain.flush()
```

```typescript [SvelteKit]src/hooks.server.ts
import { createEvlogHooks } from 'evlog/sveltekit'
import type { DrainContext } from 'evlog'
import { createDrainPipeline } from 'evlog/pipeline'
import { createAxiomDrain } from 'evlog/axiom'

const pipeline = createDrainPipeline<DrainContext>()
const drain = pipeline(createAxiomDrain())

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

process.on('SIGTERM', () => drain.flush())
```

```typescript [TanStack Start]server/plugins/evlog-drain.ts
import { definePlugin } from 'nitro'
import type { DrainContext } from 'evlog'
import { createDrainPipeline } from 'evlog/pipeline'
import { createAxiomDrain } from 'evlog/axiom'

export default definePlugin((nitroApp) => {
  const pipeline = createDrainPipeline<DrainContext>()
  const drain = pipeline(createAxiomDrain())

  nitroApp.hooks.hook('evlog:drain', drain)
  nitroApp.hooks.hook('close', () => drain.flush())
})
```

```typescript [React Router]app/root.tsx
import { evlog } from 'evlog/react-router'
import type { DrainContext } from 'evlog'
import { createDrainPipeline } from 'evlog/pipeline'
import { createAxiomDrain } from 'evlog/axiom'

const pipeline = createDrainPipeline<DrainContext>()
const drain = pipeline(createAxiomDrain())

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

process.on('SIGTERM', () => drain.flush())
```

```typescript [Hono]src/index.ts
import { evlog } from 'evlog/hono'
import type { DrainContext } from 'evlog'
import { createDrainPipeline } from 'evlog/pipeline'
import { createAxiomDrain } from 'evlog/axiom'

const pipeline = createDrainPipeline<DrainContext>()
const drain = pipeline(createAxiomDrain())

app.use(evlog({ drain }))

process.on('SIGTERM', () => drain.flush())
```

```typescript [Express]src/index.ts
import { evlog } from 'evlog/express'
import type { DrainContext } from 'evlog'
import { createDrainPipeline } from 'evlog/pipeline'
import { createAxiomDrain } from 'evlog/axiom'

const pipeline = createDrainPipeline<DrainContext>()
const drain = pipeline(createAxiomDrain())

app.use(evlog({ drain }))

process.on('SIGTERM', () => drain.flush())
```

```typescript [Fastify]src/index.ts
import { evlog } from 'evlog/fastify'
import type { DrainContext } from 'evlog'
import { createDrainPipeline } from 'evlog/pipeline'
import { createAxiomDrain } from 'evlog/axiom'

const pipeline = createDrainPipeline<DrainContext>()
const drain = pipeline(createAxiomDrain())

await app.register(evlog, { drain })

process.on('SIGTERM', () => drain.flush())
```

```typescript [Elysia]src/index.ts
import { evlog } from 'evlog/elysia'
import type { DrainContext } from 'evlog'
import { createDrainPipeline } from 'evlog/pipeline'
import { createAxiomDrain } from 'evlog/axiom'

const pipeline = createDrainPipeline<DrainContext>()
const drain = pipeline(createAxiomDrain())

app.use(evlog({ drain }))

process.on('SIGTERM', () => drain.flush())
```

```typescript [NestJS]src/app.module.ts
import { EvlogModule } from 'evlog/nestjs'
import type { DrainContext } from 'evlog'
import { createDrainPipeline } from 'evlog/pipeline'
import { createAxiomDrain } from 'evlog/axiom'

const pipeline = createDrainPipeline<DrainContext>()
const drain = pipeline(createAxiomDrain())

EvlogModule.forRoot({ drain })

process.on('SIGTERM', () => drain.flush())
```

```typescript [oRPC]server/orpc.ts
import { withEvlog } from 'evlog/orpc'
import type { DrainContext } from 'evlog'
import { createDrainPipeline } from 'evlog/pipeline'
import { createAxiomDrain } from 'evlog/axiom'

const pipeline = createDrainPipeline<DrainContext>()
const drain = pipeline(createAxiomDrain())

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

process.on('SIGTERM', () => drain.flush())
```

```typescript [Cloudflare Workers]src/worker.ts
import { withEvlog } from 'evlog/workers'
import type { DrainContext } from 'evlog'
import { createDrainPipeline } from 'evlog/pipeline'
import { createAxiomDrain } from 'evlog/axiom'

const pipeline = createDrainPipeline<DrainContext>()
const drain = pipeline(createAxiomDrain())

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

```typescript [Standalone]index.ts
// Plain TypeScript / Bun / Node script
import type { DrainContext } from 'evlog'
import { initLogger } from 'evlog'
import { createDrainPipeline } from 'evlog/pipeline'
import { createAxiomDrain } from 'evlog/axiom'

const pipeline = createDrainPipeline<DrainContext>()
const drain = pipeline(createAxiomDrain())

initLogger({ drain })

await drain.flush() // 在退出前
```

</framework-tabs>

<callout color="warning" icon="i-lucide-alert-triangle">

始终在进程退出前刷新流水线（`drain.flush()`）。在 Nitro 中使用 `close` hook；在 standalone 脚本中于 `process.exit` 前调用。在 serverless 运行时中，中间件会为你延长实例的生命周期，参见 [Serverless](#serverless)。

</callout>

## 流水线对事件的处理方式

事件到达 `evlog:drain` 后会被缓冲。当达到 `batch.size` 或 `batch.intervalMs` 到期时（以先发生者为准），批次会被刷新。失败时，相同批次会按照配置的退避策略重试；一旦 `retry.maxAttempts` 耗尽，就会使用丢失的事件调用 `onDropped`（如果没有配置该回调，则记录丢弃日志）。缓冲区大小受 `maxBufferSize` 限制。缓冲区填满后，会丢弃最旧的事件，以保持内存占用稳定。

### 流水线负责重试

直接调用时，Adapter drains（`createAxiomDrain()`、`createDatadogDrain()` 等）会吞掉传输失败，因此失败的后端永远不会中断请求。在流水线内部，这会使 `retry` 和 `onDropped` 成为无效代码，因此流水线会将适配器解包为会抛出错误的 `raw` 版本：失败会进入重试循环，而原始传输默认只进行一次 HTTP 尝试，以避免批次被重复重试。如果你明确希望在每次流水线尝试中额外进行 HTTP 尝试，请仅在适配器配置中设置 `retries`。

## 配置

```typescript [pipeline-config.ts]
import type { DrainContext } from 'evlog'
import { createDrainPipeline } from 'evlog/pipeline'
import { createAxiomDrain } from 'evlog/axiom'

const pipeline = createDrainPipeline<DrainContext>({
  batch: {
    size: 50,          // 每 50 个事件刷新一次
    intervalMs: 5000,  // 或每 5 秒刷新一次，以先到者为准
  },
  retry: {
    maxAttempts: 3,
    backoff: 'exponential',
    initialDelayMs: 1000,
    maxDelayMs: 30000,
  },
  maxBufferSize: 1000,
  onDropped: (events, error) => {
    console.error(`[evlog] 丢弃了 ${events.length} 个事件：`, error?.message)
  },
})

export const drain = pipeline(createAxiomDrain())
```

### 每个选项及其默认值

<table>
<thead>
  <tr>
    <th>
      选项
    </th>
    
    <th>
      默认值
    </th>
    
    <th>
      描述
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        batch.size
      </code>
    </td>
    
    <td>
      <code>
        50
      </code>
    </td>
    
    <td>
      每批最多事件数
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        batch.intervalMs
      </code>
    </td>
    
    <td>
      <code>
        5000
      </code>
    </td>
    
    <td>
      刷新部分批次前的最大时间（ms）
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        retry.maxAttempts
      </code>
    </td>
    
    <td>
      <code>
        3
      </code>
    </td>
    
    <td>
      包括初始尝试在内的总尝试次数
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        retry.backoff
      </code>
    </td>
    
    <td>
      <code>
        'exponential'
      </code>
    </td>
    
    <td>
      <code>
        'exponential'
      </code>
      
       | <code>
        'linear'
      </code>
      
       | <code>
        'fixed'
      </code>
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        retry.initialDelayMs
      </code>
    </td>
    
    <td>
      <code>
        1000
      </code>
    </td>
    
    <td>
      第一次重试的基础延迟
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        retry.maxDelayMs
      </code>
    </td>
    
    <td>
      <code>
        30000
      </code>
    </td>
    
    <td>
      任意重试延迟的上限
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        maxBufferSize
      </code>
    </td>
    
    <td>
      <code>
        1000
      </code>
    </td>
    
    <td>
      丢弃最旧事件前的最大缓冲事件数
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        onDropped
      </code>
    </td>
    
    <td>
      -
    </td>
    
    <td>
      事件被丢弃时的回调（溢出或重试耗尽）
    </td>
  </tr>
</tbody>
</table>

### 选择重试间隔策略

<table>
<thead>
  <tr>
    <th>
      策略
    </th>
    
    <th>
      延迟模式
    </th>
    
    <th>
      使用场景
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        exponential
      </code>
    </td>
    
    <td>
      1s, 2s, 4s, 8s…
    </td>
    
    <td>
      默认值。最适合可能需要时间恢复的临时失败
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        linear
      </code>
    </td>
    
    <td>
      1s, 2s, 3s, 4s…
    </td>
    
    <td>
      可预测的延迟增长
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        fixed
      </code>
    </td>
    
    <td>
      1s, 1s, 1s, 1s…
    </td>
    
    <td>
      每次延迟都相同。适合限流 API
    </td>
  </tr>
</tbody>
</table>

### 包装器返回的内容

`pipeline(drain)` 返回的函数与 hook 兼容，并提供：

<table>
<thead>
  <tr>
    <th>
      属性
    </th>
    
    <th>
      类型
    </th>
    
    <th>
      描述
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        drain(ctx)
      </code>
    </td>
    
    <td>
      <code>
        (ctx: T) => void
      </code>
    </td>
    
    <td>
      将单个事件推入缓冲区
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        drain.flush()
      </code>
    </td>
    
    <td>
      <code>
        () => Promise<void>
      </code>
    </td>
    
    <td>
      强制刷新所有缓冲的事件
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        drain.settled()
      </code>
    </td>
    
    <td>
      <code>
        () => Promise<void>
      </code>
    </td>
    
    <td>
      当目前为止缓冲的所有内容都已发送或丢弃后解析，不会强制刷新
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        drain.pending
      </code>
    </td>
    
    <td>
      <code>
        number
      </code>
    </td>
    
    <td>
      当前缓冲的事件数
    </td>
  </tr>
</tbody>
</table>

## Serverless

`drain(ctx)` 会在事件进入缓冲区后返回，因此在 serverless 运行时中，关键问题是：在批次实际发送完成前，是什么让实例保持运行。当你将流水线作为框架集成的 `drain` 选项传递时，中间件会向运行时的 `waitUntil` 注册 `drain.settled()`（Next 的 `after()`、Workers 的 `ctx.waitUntil`）：实例会保持运行，直到缓冲的批次发送或丢弃。`batch.intervalMs` 会限制首次传输尝试开始的时间；重试退避可能会使生命周期超过该时间，直到传输成功或 `retry.maxAttempts` 耗尽。如果你希望缩短响应后的生命周期，而不是优先进行批处理，请在 serverless 环境中降低 `intervalMs`。

Nitro 是例外：在 `evlog:drain` hook 上注册的 drains 对插件来说是不透明的，因此插件无法为你延长生命周期。在 Node 中，快速开始示例中的 `close` hook 刷新操作可以覆盖关闭流程；在 Cloudflare 中，请在处理请求的地方自行向执行上下文的 `waitUntil` 注册 `drain.settled()`。

## Fanout

<drain-fan-out>



</drain-fan-out>

在生产环境中，同一类宽泛事件通常需要到达多个目标：长期存储（Axiom）、指标工具（Datadog）、错误追踪器（Sentry），以及用于事件回放的本地 fs drain。流水线先批量处理一次，然后你的 drain 通过 `Promise.allSettled` 将该批次并行分发到每个目标，这样单个缓慢/失败的目标就不会阻塞其他目标。

<prompt :actions="["copy","cursor","claude"]" description="将 evlog 事件分发到多个目标" icon="i-lucide-share-2">

通过单个 drain 流水线并行将每个宽泛事件发送到多个目标。

- 在分发函数外包装一个来自 `evlog/pipeline` 的 `createDrainPipeline`，该函数会在 `Promise.allSettled([drainA(batch), drainB(batch), …])` 中调用每个目标 drain。使用 `allSettled`，这样一个失败的 drain 不会导致整个批次被拒绝
- 根据用途选择目标：长期存储（Axiom / Better Stack / Datadog）、错误追踪器（Sentry，通常使用 `{ minLevel: 'error' }`，以免接收所有事件）、本地回放（`createFsDrain`）
- 在流水线级别统一调整 `batch.size`、`batch.intervalMs`、`retry.maxAttempts` 和 `maxBufferSize`，这些配置会应用于所有目标
- 对于需要不同过滤条件的目标，优先使用每个 drain 的 `minLevel` / `filter` 选项，而不是进行包装
- 不要忘记在关闭时调用 `drain.flush()`。为 fanout 缓冲的事件会在突然退出时丢失

文档：[https://www.evlog.dev/extend/drain-pipeline](https://www.evlog.dev/extend/drain-pipeline)

</prompt>

### 接入 fan-out

```ts
import { createDrainPipeline } from 'evlog/pipeline'
import { createAxiomDrain } from 'evlog/axiom'
import { createDatadogDrain } from 'evlog/datadog'
import { createSentryDrain } from 'evlog/sentry'
import { createFsDrain } from 'evlog/fs'
import type { DrainContext } from 'evlog'

const pipeline = createDrainPipeline<DrainContext>({
  batch: { size: 50, intervalMs: 5000 },
  retry: { maxAttempts: 3 },
  maxBufferSize: 1000,
})

const axiom = createAxiomDrain()
const datadog = createDatadogDrain()
const sentry = createSentryDrain({ minLevel: 'error' })
const fs = createFsDrain({ dir: '.evlog/logs', maxFiles: 14 })

export const drain = pipeline(async (batch) => {
  await Promise.allSettled([
    axiom(batch),
    datadog(batch),
    sentry(batch),
    fs(batch),
  ])
})
```

### fan-out 的保证

- **并行分发**：每个目标都会通过 `Promise.allSettled` 并发接收该批次
- **容错 fanout**：如果 Datadog 的 API 抛出错误，Axiom / Sentry / fs 仍会完成；只有包装函数被拒绝时，流水线才会重试整个批次
- **共享背压**：整个流水线只设置一次缓冲区大小；如果包装的 drain 处理落后，所有目标都会一致地丢弃最旧事件

### 将不同事件发送到每个 drain

包装某个目标 drain，让它只看到你关心的事件：

```ts
import type { DrainContext } from 'evlog'

const sentry = createSentryDrain({ dsn: process.env.SENTRY_DSN! })

async function sentryErrorsOnly(batch: DrainContext[]): Promise<void> {
  const errors = batch.filter(c => c.event?.level === 'error')
  if (errors.length > 0) await sentry(errors)
}

export const drain = pipeline(async (batch) => {
  await Promise.allSettled([
    axiom(batch),
    sentryErrorsOnly(batch),
  ])
})
```

大多数内置 drains 都直接暴露 `minLevel`，因此你只在需要非 level 过滤（路径、自定义字段等）时才需要这种模式。

## 自己编写 drain

你不需要适配器。传入任意接受一个 batch 的异步函数：

```typescript [pipeline-custom.ts]
import type { DrainContext } from 'evlog'
import { createDrainPipeline } from 'evlog/pipeline'

const pipeline = createDrainPipeline<DrainContext>({ batch: { size: 100 } })

export const drain = pipeline(async (batch) => {
  await fetch('https://your-service.com/logs', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(batch.map(ctx => ctx.event)),
  })
})
```

对于更复杂的场景（配置解析、重试、身份标头），请改用 [`defineHttpDrain`](https://evlog.zhcndoc.com/extend/custom-drains)，让工具包处理样板代码。

## HTTP drain（浏览器到服务器）

HTTP drain 会将客户端日志发送到你的服务器，无论服务器使用什么框架。它构建于相同的流水线基础能力之上，并使用浏览器专用的默认设置（`fetch keepalive` + 在 `visibilitychange` 时使用 `sendBeacon`）。

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

`evlog/browser` 导入路径已被 **弃用**，它会重新导出与 `evlog/http` 相同的 API。它将在下一个 **主版本** 中移除。新代码请优先使用 `evlog/http`。

</callout>

<prompt :actions="["copy","cursor","claude"]" description="为客户端日志设置 HTTP 传输" icon="i-lucide-globe">

设置 HTTP 传输，以便我的浏览器日志发送到服务器。

- 安装 evlog：pnpm add evlog
- 从 `evlog/http` 导入 `createHttpLogDrain`（不要从已弃用的 `evlog/browser` 导入）
- 使用 `{ endpoint: 'https://logs.example.com/v1/ingest' }` 创建 drain，并可选地配置流水线（`{ batch: { size, intervalMs } }`）
- 在客户端将 drain 传递给 `initLogger({ drain })`
- drain 会批量处理事件，并在 `visibilitychange` 时使用 `fetch keepalive` + `sendBeacon`
- 在服务器上，接受包含 `DrainContext[]` 的请求体，并将其转发到我的 drain 流水线

文档：[https://www.evlog.dev/extend/drain-pipeline](https://www.evlog.dev/extend/drain-pipeline)

</prompt>

### 快速开始

```typescript [app.ts]
import { initLogger, log } from 'evlog'
import { createHttpLogDrain } from 'evlog/http'

const drain = createHttpLogDrain({
  drain: { endpoint: 'https://logs.example.com/v1/ingest' },
})
initLogger({ drain })

log.info({ action: 'page_view', path: location.pathname })
```

### 工作原理（浏览器特定）

1. `log.info()` / `log.warn()` / `log.error()` 将事件推入内存缓冲区
2. 事件会按大小（默认 25）或时间间隔（默认 2 秒）进行批处理
3. 批次通过带有 `keepalive: true` 的 `fetch` 发送，以确保请求在页面导航时仍能继续
4. 当页面变为隐藏状态（切换标签、导航）时，缓冲的事件会通过 `navigator.sendBeacon` 作为回退方式刷新发送
5. 你的服务器端点接收一个 `DrainContext[]` JSON 数组，并以你喜欢的方式处理它

### 选择所需层级

#### `createHttpLogDrain(options)`

高级、预组合：创建一个带有批处理、重试以及在 `visibilitychange` 时自动刷新的 pipeline。直接返回可与 `initLogger({ drain })` 一起使用的 `PipelineDrainFn<DrainContext>`。

```typescript [app.ts]
import { initLogger, log } from 'evlog'
import { createHttpLogDrain } from 'evlog/http'

const drain = createHttpLogDrain({
  drain: { endpoint: 'https://logs.example.com/v1/ingest' },
  pipeline: { batch: { size: 50, intervalMs: 5000 } },
})

initLogger({ drain })
log.info({ action: 'click', target: 'buy-button' })
```

#### `createHttpDrain(config)`

低层传输函数。当你希望完全控制 pipeline 配置时使用它：

```typescript [app.ts]
import { createHttpDrain } from 'evlog/http'
import { createDrainPipeline } from 'evlog/pipeline'
import type { DrainContext } from 'evlog'

const transport = createHttpDrain({
  endpoint: 'https://logs.example.com/v1/ingest',
})
const pipeline = createDrainPipeline<DrainContext>({
  batch: { size: 100, intervalMs: 10000 },
  retry: { maxAttempts: 5 },
})

const drain = pipeline(transport)
```

### 每个浏览器选项

#### `HttpDrainConfig`

<table>
<thead>
  <tr>
    <th>
      选项
    </th>
    
    <th>
      默认值
    </th>
    
    <th>
      说明
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        endpoint
      </code>
    </td>
    
    <td>
      -
    </td>
    
    <td>
      <strong>
        （必需）
      </strong>
      
       服务器 ingest 端点的完整 URL
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        headers
      </code>
    </td>
    
    <td>
      -
    </td>
    
    <td>
      随每个 <code>
        fetch
      </code>
      
       请求发送的自定义标头（例如 <code>
        Authorization
      </code>
      
      、<code>
        X-API-Key
      </code>
      
      ）
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        timeout
      </code>
    </td>
    
    <td>
      <code>
        5000
      </code>
    </td>
    
    <td>
      请求超时时间，单位为毫秒
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        useBeacon
      </code>
    </td>
    
    <td>
      <code>
        true
      </code>
    </td>
    
    <td>
      当页面隐藏时使用 <code>
        sendBeacon
      </code>
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        credentials
      </code>
    </td>
    
    <td>
      <code>
        'same-origin'
      </code>
    </td>
    
    <td>
      Fetch 凭据模式（<code>
        'omit'
      </code>
      
      、<code>
        'same-origin'
      </code>
      
      、<code>
        'include'
      </code>
      
      ）。对于跨域端点请设置为 <code>
        'include'
      </code>
    </td>
  </tr>
</tbody>
</table>

#### `HttpLogDrainOptions`

<table>
<thead>
  <tr>
    <th>
      选项
    </th>
    
    <th>
      默认值
    </th>
    
    <th>
      说明
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        drain
      </code>
    </td>
    
    <td>
      -
    </td>
    
    <td>
      <strong>
        （必需）
      </strong>
      
       <code>
        HttpDrainConfig
      </code>
      
       对象
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        pipeline
      </code>
    </td>
    
    <td>
      <code>
        { batch: { size: 25, intervalMs: 2000 }, retry: { maxAttempts: 2 } }
      </code>
    </td>
    
    <td>
      pipeline 配置覆盖项
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        autoFlush
      </code>
    </td>
    
    <td>
      <code>
        true
      </code>
    </td>
    
    <td>
      自动注册 <code>
        visibilitychange
      </code>
      
       刷新监听器
    </td>
  </tr>
</tbody>
</table>

### 让日志经受住标签页关闭

<callout color="info" icon="i-lucide-radio">

当启用 `useBeacon`（默认值）且页面变为隐藏时，drain 会自动从 `fetch` 切换到 `navigator.sendBeacon`。这可确保即使用户关闭标签页或离开页面，日志也能被发送。

</callout>

`sendBeacon` 受浏览器施加的负载限制（约 64 KB）。如果负载超过此限制，drain 会抛出错误。请保持批次大小合理（默认 25 远在限制范围内）。

### 验证浏览器请求

传递自定义标头以保护你的 ingest 端点：

```typescript [app.ts]
const drain = createHttpLogDrain({
  drain: {
    endpoint: 'https://logs.example.com/v1/ingest',
    headers: {
      'Authorization': 'Bearer ' + token,
    },
  },
})
```

<callout color="warning" icon="i-lucide-shield-alert">

`headers` 仅应用于 `fetch` 请求。`sendBeacon` API 不支持自定义标头，因此当页面隐藏并使用 `sendBeacon` 时，不会发送标头。如果你的端点需要身份验证，请通过会话 cookie 进行验证（对跨域端点设置 `credentials: 'include'`），或者通过 `useBeacon: false` 禁用 `sendBeacon`。

</callout>

### 服务器端点

你的服务器需要一个接受 `DrainContext[]` JSON body 的 POST 端点。常见框架示例如下：

<framework-tabs>

```typescript [Express]src/index.ts
app.post('/v1/ingest', express.json(), (req, res) => {
  for (const entry of req.body) {
    console.log('[BROWSER]', JSON.stringify(entry))
  }
  res.sendStatus(204)
})
```

```typescript [Hono]src/index.ts
app.post('/v1/ingest', async (c) => {
  const body = await c.req.json()
  for (const entry of body) {
    console.log('[BROWSER]', JSON.stringify(entry))
  }
  return c.body(null, 204)
})
```

</framework-tabs>

请参阅完整的 [browser example](https://github.com/evloghq/evlog/tree/main/examples/browser)，了解可运行的 Hono server + browser page。

## 会导致事件丢失的错误做法

- **不要忘记在关闭时调用 drain.flush()**：否则缓冲的事件会丢失
- **调整 batch.size 以匹配你的服务商建议的负载大小**：太小会浪费开销，太大则有被拒绝的风险
- **除非需要按目标分别重试，否则不要为每个 drain 运行一个流水线**：共享一个流水线可以让批处理 + 缓冲保持一致
- **在分发到与 serverless 不兼容的目标前先进行检查**：[stream server](https://evlog.zhcndoc.com/extend/stream) 通过进程内 stream 将事件发送给每个已连接的客户端；它不是 drain

## 下一步

- [Custom Drains](https://evlog.zhcndoc.com/extend/custom-drains)：使用 `defineHttpDrain` / `defineDrain` 为任意后端构建 drain
- [Adapters Overview](https://evlog.zhcndoc.com/integrate/adapters/overview)：开箱即用、可与流水线配合使用的内置适配器
- [Best Practices](https://evlog.zhcndoc.com/reference/best-practices)：安全与生产环境提示
- [Client logging](https://evlog.zhcndoc.com/use-cases/client-logging)：端到端的浏览器 → 服务器流程。

---

- [自定义 Drains](https://evlog.zhcndoc.com/extend/custom-drains)
- [Adapters 概览](https://evlog.zhcndoc.com/integrate/adapters/overview)


## Sitemap

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