Grafana Loki 适配器

Loki 适配器通过其推送 API 将宽事件推送到 Grafana Loki。它适用于自托管的单租户实例、多租户部署以及 Grafana Cloud。

每个事件都会作为JSON 日志行推送，并归属于一组小型、低基数的标签。在 Loki 中，这一区别非常重要：标签会按基数建立索引并计费，而日志行则在查询时进行搜索。evlog 默认仅为 service、environment 和 level 添加标签，并将其他所有内容（requestId、path、user 以及你的自定义字段）保留在日志行中，可通过 | json 进行查询。

安装

Loki 适配器已随 evlog 一起提供：

import { createLokiDrain } from 'evlog/loki'

快速开始

设置 LOKI_ENDPOINT，并将日志排出器接入你的框架。

// server/plugins/evlog.ts
import { createLokiDrain } from 'evlog/loki'

export default defineNitroPlugin((nitroApp) => {
  nitroApp.hooks.hook('evlog:drain', createLokiDrain())
})

import { Hono } from 'hono'
import { evlog } from 'evlog/hono'
import { createLokiDrain } from 'evlog/loki'

const app = new Hono()
app.use(evlog({ drain: createLokiDrain() }))

import express from 'express'
import { evlog } from 'evlog/express'
import { createLokiDrain } from 'evlog/loki'

const app = express()
app.use(evlog({ drain: createLokiDrain() }))

import Fastify from 'fastify'
import { evlog } from 'evlog/fastify'
import { createLokiDrain } from 'evlog/loki'

const app = Fastify()
await app.register(evlog, { drain: createLokiDrain() })

import { Elysia } from 'elysia'
import { evlog } from 'evlog/elysia'
import { createLokiDrain } from 'evlog/loki'

const app = new Elysia().use(evlog({ drain: createLokiDrain() }))

import { Module } from '@nestjs/common'
import { EvlogModule } from 'evlog/nestjs'
import { createLokiDrain } from 'evlog/loki'

@Module({
  imports: [EvlogModule.forRoot({ drain: createLokiDrain() })],
})
export class AppModule {}

import { initLogger } from 'evlog'
import { createLokiDrain } from 'evlog/loki'

initLogger({
  env: { service: 'my-app' },
  drain: createLokiDrain(),
})

配置

环境变量

优先级

配置按从高到低的优先级解析：

1. 传递给 createLokiDrain() 的覆盖配置
2. runtimeConfig.evlog.loki（Nitro）
3. runtimeConfig.loki（Nitro）
4. 环境变量

选项

部署

Loki 以这两种方式中的任一种运行，适配器在两种情况下都相同。区别仅在于身份验证。

自托管

单租户实例只需要端点：

createLokiDrain({ endpoint: 'http://localhost:3100' })

LOKI_ENDPOINT=http://localhost:3100

对于多租户部署，请指定租户名称，该名称会作为 X-Scope-OrgID 发送：

createLokiDrain({
  endpoint: 'http://loki.internal:3100',
  tenantId: 'team-checkout',
})

LOKI_ENDPOINT=http://loki.internal:3100
LOKI_TENANT_ID=team-checkout

如果你的实例位于需要身份验证的代理之后，则仅发送 apiKey，格式为 Authorization: Bearer。

Grafana Cloud

Grafana Cloud 使用你的实例 ID和访问策略令牌进行身份验证，并将二者一起作为 HTTP Basic 发送：

createLokiDrain({
  endpoint: 'https://logs-prod-eu-west-0.grafana.net',
  user: '123456',
  apiKey: process.env.GRAFANA_API_KEY,
})

LOKI_ENDPOINT=https://logs-prod-eu-west-0.grafana.net
LOKI_USER=123456
LOKI_API_KEY=glc_xxx

应用哪种身份验证

tenantId 独立存在，可以与任一身份验证模式结合使用。

标签与基数

仅为你会进行筛选且取值集合有界的字段添加标签：

createLokiDrain({
  // `region` 的取值只有少数几个——安全
  labelFields: ['service', 'environment', 'level', 'region'],
  // 此部署中的所有内容使用的静态标签
  labels: { cluster: 'prod-eu' },
})

其他所有内容都会保留在 JSON 日志行中，并且完全可查询。请参见下文。

Grafana 中的查询

按标签筛选，然后使用 | json 深入查看宽事件：

{service="checkout", environment="production"}
  | json
  | status >= 500

# 某个路由上的慢请求
{service="checkout"} | json | path="/api/orders" | durationMs > 1000

# 端到端的单个请求
{service="checkout"} | json | requestId="4a8ff3a8-..."

# 每个服务的错误率
sum by (service) (rate({environment="production", level="error"}[5m]))

批处理

将 drain 与管道结合使用，以批量推送，而不是逐个请求推送：

import { createDrainPipeline } from 'evlog/pipeline'
import { createLokiDrain } from 'evlog/loki'
import type { DrainContext } from 'evlog'

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

export default defineNitroPlugin((nitroApp) => {
  nitroApp.hooks.hook('evlog:drain', pipeline(createLokiDrain()))
})

共享同一标签集的事件会被分组到单个 Loki 流中，并按时间戳排序，因为 Loki 会拒绝流内时间顺序不正确的推送。

在本地验证

在 Docker 中启动 Loki 并推送真实事件，无需云账户：

docker compose -f packages/evlog/test/e2e/docker-compose.yml up -d
LOKI_ENDPOINT=http://localhost:3100 pnpm run test:e2e
docker compose -f packages/evlog/test/e2e/docker-compose.yml down -v

该测试套件是一次往返测试：它推送事件，通过 Loki 的范围 API 将其查询回来，并断言标签集、JSON 日志行和时间戳均得以保留。如果未设置 LOKI_ENDPOINT，它会显示一个明确的标签并跳过自身，而不是静默通过。

如果想直接查看结果，可以将任意 Grafana 指向 http://localhost:3100，然后运行 {service="evlog-e2e"}。

故障排查

Missing endpoint。 未设置 LOKI_ENDPOINT，且没有传递 endpoint。drain 会记录一次 [evlog/loki] Missing endpoint，随后不再执行任何操作，因此请求永远不会被阻塞或失败。

Loki API error: 401。 在 Grafana Cloud 中，请检查 user 是否为 Loki 数据源的数字实例 ID，而不是你的账户邮箱。

提及时间顺序错误条目的 Loki API error: 400。 较旧版本的 Loki 会拒绝早于流中最近条目的条目。evlog 会在每次推送内进行排序；如果仍然遇到此问题，请在 Loki 中启用 unordered_writes，或缩短批处理间隔。

Grafana 中没有显示任何内容。 确认你查询的标签集。先运行 {service=~".+"}，查看哪些流已到达。

直接使用 API

绕过 drain，自行推送事件：

import { sendBatchToLoki, sendToLoki } from 'evlog/loki'

await sendToLoki(event, { endpoint: 'http://localhost:3100' })
await sendBatchToLoki(events, { endpoint: 'http://localhost:3100' })

buildLokiPayload()、toLokiLabels() 和 resolveLokiPushUrl() 也已导出，可用于自定义传输。

后续步骤

• 采样：在日志到达 Loki 之前控制日志量
• Enrichers：为每个事件添加派生字段
• Drain 管道：批处理、重试和扇出
• 适配器概览：所有可用的目标站点