---
title: "Telemetry Ingest"
description: "为 @evlog/telemetry 构建摄取端点：威胁模型、parseIngestBody 验证、框架路由、存储和限流。"
canonical_url: "https://evlog.zhcndoc.com/use-cases/telemetry/ingest"
---
# Telemetry Ingest

> 为 @evlog/telemetry 构建摄取端点：威胁模型、parseIngestBody 验证、框架路由、存储和限流。

你发布了一个 CLI。用户在自己的机器上运行它。你想知道 *运行了哪些命令、失败的频率，以及目前有哪些版本*，而不需要构建单独的分析 SDK 或读取原始的 `argv`

这意味着你需要一个 **服务器端点**，由你的 CLI 向其发送 POST 请求。本地 outbox 只是为了让短暂的 CI 作业和离线运行在 POST 成功之前不会丢失数据。

## 安全问题（先读这个）

<tip>

**CLI telemetry 不是保险库。** 你收集的是匿名使用计数器（命令名称、持续时间、结果以及少量数字），不是密码、不是文件路径、不是令牌。针对这一威胁模型进行设计，你就不需要不可能实现的客户端密钥。

</tip>

**“我把摄取 URL 放进了 CLI。它难道不是秘密吗？”**

不是。这个 URL 存在于你的 npm 包或二进制文件中。任何人都可以使用 `strings` 读取它，也可以通过安装软件包或观察网络流量获取它。CLI 中内置的 API key 或 `Authorization` header 也是一样，可以被提取出来，并不是真正的屏障。

**“那不是任何人都能发送伪造事件吗？”**

理论上是的：有人可以使用 `curl` 向你的端点发送伪造的 JSON。这就是为什么你要将端点视为 **半公开**（类似浏览器分析 key），并在服务器上进行防护：

<table>
<thead>
  <tr>
    <th>
      你可能担心的事
    </th>
    
    <th>
      实际做法
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      随机互联网机器人
    </td>
    
    <td>
      校验 payload 结构；对垃圾数据返回 <code>
        400
      </code>
    </td>
  </tr>
  
  <tr>
    <td>
      有人疯狂刷假的 <code>
        doctor
      </code>
      
       运行
    </td>
    
    <td>
      按 IP 限流；数据只是低价值计数器，不是钱
    </td>
  </tr>
  
  <tr>
    <td>
      伪造的 <code>
        tool.name
      </code>
    </td>
    
    <td>
      服务器端只允许你已发布的工具名称白名单
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        custom
      </code>
      
       中出现意外字段
    </td>
    
    <td>
      只过滤为你声明过的 key——和 CLI 端 <code>
        collect
      </code>
      
       的列表保持一致
    </td>
  </tr>
  
  <tr>
    <td>
      重试导致重复计数
    </td>
    
    <td>
      存储时按 <code>
        idempotencyKey
      </code>
      
       去重
    </td>
  </tr>
  
  <tr>
    <td>
      泄露用户秘密
    </td>
    
    <td>
      CLI 从不读取原始 <code>
        argv
      </code>
      
      ；你只允许列出的 flags 和 <code>
        telemetry.set()
      </code>
      
       计数器
    </td>
  </tr>
</tbody>
</table>

**“足够好”的方案是：** 验证每个 POST，以幂等方式存储，在边缘层（CDN、API gateway 或 middleware）进行限流，并让 payload 保持简单（从设计上不包含 PII）。这与客户端分析使用的是同一套方案：不是银行级身份验证，但可以提供可靠的产品洞察。

<callout color="info" icon="i-lucide-shield-check">

`@evlog/telemetry` 为你的服务器提供了 **parseIngestBody()**，使用本文档中记录的相同验证规则，因此你无需从文档中复制粘贴验证器。将它与框架路由以及你的数据库或 evlog drain 配合使用。

</callout>

## 第 1 步：使用 `parseIngestBody()` 进行验证

在服务器上镜像 CLI 配置：相同的 `name`，以及你通过 `collect` / `telemetry.set()` 允许的相同自定义键。

```ts [lib/telemetry-ingest.ts]
import { parseIngestBody } from '@evlog/telemetry/ingest'

export const ingestOptions = {
  allowedTools: ['my-tool'],
  allowedCustomKeys: {
    'my-tool': ['checksFailed', 'checksWarn', 'itemsSynced'],
  },
} as const

export function parseTelemetryBody(raw: string) {
  return parseIngestBody(raw, ingestOptions)
}
```

`parseIngestBody()` 会检查批次大小、JSON 结构、`event: 'run'`、工具白名单、信封类型、ISO 时间戳，并移除你未声明的 `custom` 键。失败时会抛出 `IngestValidationError`，因此应从你的路由返回 `400`。

## 第 2 步：接入你的路由

任何框架都使用相同的契约：读取原始请求体，进行验证，存储，并返回 **204**，这样 CLI 就会清空其 outbox：

<framework-tabs>

```ts [Nitro]server/api/telemetry/ingest.post.ts
import { defineHandler } from 'nitro/h3'
import { readRawBody, setResponseStatus } from 'h3'
import { IngestValidationError } from '@evlog/telemetry/ingest'
import { parseTelemetryBody } from '~/lib/telemetry-ingest'
import { storeRunEvents } from '~/lib/telemetry-store'

export default defineHandler(async (event) => {
  const raw = await readRawBody(event, 'utf8')
  if (!raw) {
    setResponseStatus(event, 400)
    return { error: '请求体为空' }
  }

  try {
    const events = parseTelemetryBody(raw)
    await storeRunEvents(events)
    setResponseStatus(event, 204)
    return null
  } catch (err) {
    setResponseStatus(event, err instanceof IngestValidationError ? 400 : 500)
    return { error: '无效载荷' }
  }
})
```

```ts [Next.js App Router]app/api/telemetry/ingest/route.ts
import { IngestValidationError } from '@evlog/telemetry/ingest'
import { parseTelemetryBody } from '@/lib/telemetry-ingest'
import { storeRunEvents } from '@/lib/telemetry-store'

export async function POST(request: Request) {
  const raw = await request.text()
  try {
    const events = parseTelemetryBody(raw)
    await storeRunEvents(events)
    return new Response(null, { status: 204 })
  } catch (err) {
    const status = err instanceof IngestValidationError ? 400 : 500
    return Response.json({ error: '无效载荷' }, { status })
  }
}
```

```ts [Hono]src/routes/telemetry.ts
import { Hono } from 'hono'
import { IngestValidationError } from '@evlog/telemetry/ingest'
import { parseTelemetryBody } from '../lib/telemetry-ingest'
import { storeRunEvents } from '../lib/telemetry-store'

const app = new Hono()

app.post('/api/telemetry/ingest', async (c) => {
  const raw = await c.req.text()
  try {
    const events = parseTelemetryBody(raw)
    await storeRunEvents(events)
    return c.body(null, 204)
  } catch (err) {
    const status = err instanceof IngestValidationError ? 400 : 500
    return c.json({ error: '无效载荷' }, status)
  }
})

export default app
```

</framework-tabs>

## 第 3 步：以幂等方式存储

重试和积压清理可能会用相同的 `idempotencyKey` 发送两次 POST。使用 upsert（或跳过）以确保指标保持正确：

```ts [lib/telemetry-store.ts]
import type { RunEvent } from '@evlog/telemetry'

export async function storeRunEvents(events: RunEvent[]): Promise<void> {
  for (const run of events) {
    await db.telemetryRun.upsert({
      where: { idempotencyKey: run.idempotencyKey },
      create: {
        tool: run.tool.name,
        version: run.tool.version,
        command: run.command,
        outcome: run.outcome,
        durationMs: run.durationMs,
        custom: run.custom,
        recordedAt: run.timestamp,
      },
      update: {},
    })
  }

  // 或者转发到 evlog 的 drain pipeline，供 Axiom / Datadog / OTLP 使用：
  // await ingestWideEvents(events.map(run => ({ source: 'telemetry', ...run })))
}
```

## 第 4 步：在边缘层进行限流

`parseIngestBody()` 是你的最后一道 schema 防线，而不是唯一防线。在流量进入的位置（CDN、API gateway 或 middleware）添加限流，尤其是按 IP 限流。可选：如果需要更严格的控制，也可以在 handler 内按 `machineId` 哈希进行分桶。

非 2xx 响应会将事件保留在用户的 outbox 中；它们会在下一次 CLI 调用时重试。只有在存储成功时才返回 **204**（或任何 `2xx`）。

---

- [概览](https://evlog.zhcndoc.com/use-cases/telemetry/overview)
- [设置](https://evlog.zhcndoc.com/use-cases/telemetry/setup)
- [参考](https://evlog.zhcndoc.com/use-cases/telemetry/reference)


## Sitemap

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