Telemetry

Telemetry Ingest

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

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

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

安全问题（先读这个）

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

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

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

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

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

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

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

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

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

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

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：

import { defineEventHandler, readRawBody, setResponseStatus } from 'h3'
import { IngestValidationError } from '@evlog/telemetry/ingest'
import { parseTelemetryBody } from '~/lib/telemetry-ingest'
import { storeRunEvents } from '~/lib/telemetry-store'

export default defineEventHandler(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: '无效载荷' }
  }
})

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

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

第 3 步：以幂等方式存储

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

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

Edit this pageorReport an issue

设置

将 @evlog/telemetry 接入 citty CLI、独立脚本和 GitHub Actions——投递配置、自动捕获标志、telemetry.set() 和调试模式。

参考

RunEvent 信封、使用 collect 和自定义字段扩展模式、隐私规则、TELEMETRY.md 披露，以及同意 / outbox 可靠性。