扩展
食谱
具体的可复制粘贴方案——构建你自己的最小 devtool，pipe 到 curl + jq，先回放历史再切换到实时，并在消费端聚合。
结合了 进程内总线和流服务器 与 文件系统读取器 的真实世界模式。
构建一个自定义的 evlog devtool / 仪表盘
1. 构建一个最小 devtool
一个实时事件面板本质上就是 EventSource + 一个列表。完整的传输格式和发现规则——.evlog/stream.url、/api/_evlog/stream-info、{ evlog: '1', type, data } 包装，以及认证——都记录在 流页面 中。下面每个方案都假设你已经通过上述任一机制获取了 URL。
纯 HTML + JS（直接放进任意页面）
<!doctype html>
<html>
<head>
<meta charset="utf-8">
<title>evlog mini devtool</title>
<style>
body { font: 13px ui-sans-serif, system-ui; margin: 0; padding: 0; }
table { width: 100%; border-collapse: collapse; }
td, th { padding: 6px 10px; border-bottom: 1px solid #eee; text-align: left; }
.lvl-error { color: #ef4444 }
.lvl-warn { color: #f59e0b }
.lvl-info { color: #3b82f6 }
</style>
</head>
<body>
<table id="t">
<thead><tr><th>时间</th><th>级别</th><th>服务</th><th>操作</th></tr></thead>
<tbody></tbody>
</table>
<script>
// 替换为启动时打印的 URL，或者从 /api/_evlog/stream-info 获取
const STREAM_URL = 'http://127.0.0.1:51203'
const tbody = document.querySelector('#t tbody')
const es = new EventSource(STREAM_URL)
es.onmessage = (e) => {
const env = JSON.parse(e.data)
if (env.evlog !== '1') return
if (env.type !== 'event' && env.type !== 'replay') return
const w = env.data
const tr = document.createElement('tr')
tr.innerHTML = `
<td>${new Date(w.timestamp).toLocaleTimeString()}</td>
<td class="lvl-${w.level}">${w.level}</td>
<td>${w.service ?? ''}</td>
<td>${w.action ?? w.message ?? w.path ?? ''}</td>
`
tbody.prepend(tr)
while (tbody.children.length > 200) tbody.lastElementChild.remove()
}
</script>
</body>
</html>
保存为 devtool.html，在 evlog 已注入的开发服务器运行时，用任意浏览器标签页打开即可。这就是整个 MVP。
Vue 3 组件
<script setup lang="ts">
import { onBeforeUnmount, onMounted, ref } from 'vue'
import type { WideEvent } from 'evlog'
const events = ref<WideEvent[]>([])
let es: EventSource | null = null
onMounted(async () => {
// 通过同源信息端点发现 URL（Nuxt）
const { url } = await $fetch<{ url: string | null }>('/api/_evlog/stream-info')
if (!url) return
es = new EventSource(url)
es.onmessage = (e) => {
const env = JSON.parse(e.data)
if (env.evlog !== '1') return
if (env.type === 'event' || env.type === 'replay') {
events.value.unshift(env.data as WideEvent)
if (events.value.length > 500) events.value.length = 500
}
}
})
onBeforeUnmount(() => es?.close())
</script>
<template>
<ul>
<li v-for="(e, i) in events" :key="`${e.timestamp}-${i}`">
<code>{{ e.level }}</code>
<strong>{{ e.service }}</strong>
<span>{{ e.action ?? e.message ?? e.path }}</span>
</li>
</ul>
</template>
React hook
import { useEffect, useState } from 'react'
import type { WideEvent } from 'evlog'
export function useEvlogStream(url: string) {
const [events, setEvents] = useState<WideEvent[]>([])
useEffect(() => {
if (!url) return
const es = new EventSource(url)
es.onmessage = (e) => {
const env = JSON.parse(e.data)
if (env.evlog !== '1') return
if (env.type === 'event' || env.type === 'replay') {
setEvents(prev => [env.data, ...prev].slice(0, 500))
}
}
return () => es.close()
}, [url])
return events
}
这就是完整的集成面。没有 SDK，也没有超出 evlog 导出的 WideEvent 之外的特殊类型。
2. 使用 curl + jq 进行快速 CLI 检查
URL 在 .evlog/stream.url 中：
URL=$(cat .evlog/stream.url)
curl -N "$URL" | jq -c 'select(.type == "event") | .data'
按需在客户端过滤：
# 仅错误
curl -sN "$URL" | jq -c 'select(.type == "event" and .data.level == "error") | .data'
# 仅某个服务
curl -sN "$URL" | jq -c 'select(.type == "event" and .data.service == "checkout") | .data'
# 慢请求
curl -sN "$URL" | jq -c 'select(.type == "event" and .data.duration > 500) | .data'
-N 让 curl 保持流式模式（不缓冲）。-s 表示静默。
3. 先回放历史，再切换到实时
磁盘历史（文件系统 drain）+ 来自流服务器的实时更新 = 从任意时间点都能获得完整视图。
import { readFsLogs } from 'evlog/fs'
import { readFile } from 'node:fs/promises'
import type { WideEvent } from 'evlog'
async function bootstrap(handle: (e: WideEvent) => void) {
// 1. 回放来自 `.evlog/logs/` 的最近一小时
const since = new Date(Date.now() - 60 * 60 * 1000)
for await (const event of readFsLogs({ since })) {
handle(event)
}
// 2. 切换到实时 SSE 流
const url = (await readFile('.evlog/stream.url', 'utf-8')).trim()
const es = new EventSource(url)
es.onmessage = (e) => {
const env = JSON.parse(e.data)
if (env.evlog !== '1') return
if (env.type === 'event' || env.type === 'replay') {
handle(env.data)
}
}
return () => es.close()
}
readFsLogs 会跳过日期范围之外的文件，因此即使你保留了数周历史，回放步骤也很快。若要仅尾随、不做磁盘回放，则改为用 ?since=<iso> 访问流服务器，以复用进程内环形缓冲区。
4. Node / Bun 客户端（fetch + ReadableStream）
相同协议，无需 EventSource polyfill：
import { readFile } from 'node:fs/promises'
const url = (await readFile('.evlog/stream.url', 'utf-8')).trim()
const res = await fetch(url)
const reader = res.body!.getReader()
const decoder = new TextDecoder()
let buffer = ''
while (true) {
const { value, done } = await reader.read()
if (done) break
buffer += decoder.decode(value, { stream: true })
let idx
while ((idx = buffer.indexOf('\n\n')) !== -1) {
const frame = buffer.slice(0, idx)
buffer = buffer.slice(idx + 2)
const dataLine = frame.split('\n').find(l => l.startsWith('data:'))
if (!dataLine) continue
const env = JSON.parse(dataLine.slice(5).trim())
if (env.type === 'event') console.log(env.data)
}
}
5. 在消费端进行过滤、转换、聚合
让服务器保持“傻瓜”——每个消费者按需选择自己关心的内容：
// 仅错误
const errors = events.filter(e => e.level === 'error')
// 慢请求
const slowReqs = events.filter(e => typeof e.duration === 'number' && e.duration > 500)
// 按服务分组
const byService = Object.groupBy(events, e => e.service)
// 滚动错误率（最近 100 条事件）
const last100 = events.slice(0, 100)
const errorRate = last100.filter(e => e.level === 'error').length / last100.length
// 临时成本分析——之所以可行，是因为 evlog/ai 会在每次 AI 调用时写入 ai.* 字段
const totalCost = events
.filter(e => typeof e.ai?.estimatedCost === 'number')
.reduce((sum, e) => sum + (e.ai?.estimatedCost as number), 0)
6. 自托管的 "tail -f" 替代方案
如果消费者运行在同一台机器上，可以完全跳过网络：
import { tailFsLogs } from 'evlog/fs'
const ac = new AbortController()
process.on('SIGINT', () => ac.abort())
for await (const event of tailFsLogs({ signal: ac.signal })) {
if (event.level === 'error') notifyOps(event)
}
即使不对正在运行的应用进行注入也能工作——这对监视某个目录的 sidecar / observer 进程很有用。
不要这样做
- 不要在 Vercel Functions / Cloudflare Workers / Lambda 上运行流服务器。 每次调用都是一个独立的 isolate；一个 isolate 中的订阅者永远看不到其他 isolate 产生的事件。跨实例扇出请使用真正的 broker（Redis Streams、NATS、Pub/Sub）。
- 除非你的 evlog 配置会对其进行脱敏，否则不要在宽事件中放入与认证相关的敏感数据。 服务器会原样转发应用发出的内容——包括任何未脱敏的 PII。
- 不要在服务器端过滤（“只要错误事件”）。服务器的设计目标就是透明。请在消费端过滤；这样一个过滤条件就不会饿死另一个消费者。