OpenMeter 学习 01：快速入门

很多产品的价格与资源消耗有关：API 按调用次数收费，AI 产品按 Token 收费，CI 平台按构建分钟收费。真正麻烦的地方通常不在“加一”，而在于事件来源很多、需要按客户和时间窗口聚合，还要处理重试、重复上报、迟到事件以及后续的额度和账单。OpenMeter 把这部分工作拆成一条可观察的事件链路：应用发送使用事件，系统把事件可靠地写入消息队列，再由 worker 校验、去重和保存；查询接口从用量存储中按计量器计算结果。这样，业务代码只需要描述“发生了什么”，不用在每个服务里重复实现计数和聚合逻辑。

本文使用仓库中的 quickstart/ 目录完成一次真实部署，验证以下路径：

• Docker Compose 启动 OpenMeter API、worker 和依赖服务；
• 通过 CloudEvents 上报一次 API 请求；
• 等待异步处理完成并查询 COUNT 计量器；
• 验证相同 source 和 id 的重复事件不会被重复计量；
• 验证 SUM 计量器和 groupBy 查询。

文中的命令都从仓库根目录下的 quickstart/ 执行，读者可以直接复制运行。这个 Compose 配置使用 latest 镜像和开发级依赖，适合学习和本地评估，不是生产拓扑。

OpenMeter 简介

OpenMeter 是面向 AI、API 和其他使用量产品的开源变现基础设施。它把产品中的请求、Token、计算时长等使用记录转换为可查询的实时用量，并在此基础上提供额度控制和使用量计费能力。项目采用 API-first 的设计，可以独立承担从原始事件到发票的完整链路，也可以只接入其中的计量、额度或计费部分。

一个使用量产品通常要同时解决几件事：准确记录每次使用，按客户、时间窗口和业务维度聚合，判断客户是否还有可用额度，最后根据价格生成账单。OpenMeter 将这些问题拆成相互衔接的核心能力：

• 用量计量：接收符合 CloudEvents 规范的事件，通过计量器定义聚合次数、时长、Token 或其他数值。subject 标识使用者，groupBy 保存方法、路由、模型等维度；
• 产品与定价：定义 feature、plan、addon、rate card 和 subscription，支持固定费用、周期费用、按量、分层和套餐等价格形式；
• 额度与访问控制：根据客户的 entitlement、credit grant 和计量结果计算余额，供业务系统判断是否继续提供某项功能；
• 使用量计费：把订阅和用量转换为 charge、invoice，推进账单计算、收集和支付相关的生命周期；
• 开放集成：通过 REST API、JavaScript、Python 和 Go SDK 接入业务系统，并通过 webhook 或外部发票、支付和税务服务连接现有基础设施。

CloudEvents 规范

CloudEvents 是 CNCF 推动的一种事件描述规范，目的是让不同服务产生的事件拥有统一的外层格式。无论事件来自 API 网关、模型服务还是任务调度器，接收方都可以用相同的方式读取事件身份、类型和来源，而不必为每个生产者编写一套解析逻辑。

一个 OpenMeter 事件通常包含以下上下文字段：

• specversion：CloudEvents 规范版本，示例使用 1.0；
• id：事件在同一来源中的唯一标识，用于重试和去重；
• source：产生事件的服务或系统；
• type：事件类型，例如 request 或 prompt；
• subject：事件归属的使用者，例如客户或工作区；
• data：事件的业务数据，例如请求路由、执行时长或 Token 数量。

OpenMeter 当前通过 JSON 格式接收 CloudEvents，data 可以是任意合法的 JSON 对象。计量器使用 JsonPath 从 data 中取值，例如 $.duration_ms 表示读取 data.duration_ms。因此，CloudEvents 负责统一事件的外层结构，计量器负责解释业务数据，两者结合后，同一种事件格式就可以服务于计数、求和和维度分组等不同场景。

OpenMeter 负责用量和账单领域的核心状态，但不替代支付处理器、税务引擎或通用会计系统。它更像一层可组合的计费基础设施：产品可以先使用计量器和查询 API，之后再逐步加入 entitlement、订阅和 billing。本次入门先跑通最基础的事件计量链路，后面的组件和源码分析也会围绕这条链路展开。

运行组件

OpenMeter 把高吞吐的事件数据和需要事务一致性的业务状态放在不同的存储中。这是架构中最重要的分界：ClickHouse 负责事件和聚合查询，PostgreSQL 负责客户、计量器、产品目录、订阅、额度和账单状态。

组件	在链路中的职责	quickstart 中的服务
API server	接收事件、管理计量器、提供用量查询和管理 API	openmeter
Kafka	把接收和处理解耦，承载使用事件及领域事件	kafka
Sink worker	消费使用事件，校验、去重、匹配计量器并写入 ClickHouse	sink-worker
ClickHouse	保存原始用量并执行计量聚合查询	clickhouse
PostgreSQL	保存计量器和产品、客户、订阅、额度、账单等事务状态	postgres
Redis	quickstart 中用于 sink 去重，也可保存查询进度	redis
Balance worker	依据使用量和 grant 重新计算 metered entitlement 余额	balance-worker
Billing worker	同步订阅、计算 charge 并推进 invoice 生命周期	billing-worker
Notification service	评估通知规则并记录待发送事件	notification-service
Jobs	执行迁移、订阅同步等定时维护任务	openmeter-jobs
Svix	提供 webhook 投递服务	svix

把这条运行链路按数据流拆开，可以分成几步：

1. 业务服务或 SDK 先把 CloudEvents 发送给 API server
2. API server 将事件写入 Kafka，同时从 PostgreSQL 读取和保存计量器、客户、订阅等业务状态
3. Sink worker 消费 Kafka 中的使用事件，完成校验、去重和计量器匹配后，把原始事件写入 ClickHouse，去重键保存在 Redis 中
4. 查询请求仍然由 API server 接收，再交给 ClickHouse 返回聚合后的用量
5. Balance worker 和 Billing worker 也消费 Kafka 中的相关事件，分别负责额度计算和账单处理。

Kafka 的意义是让 API 不必同步等待 ClickHouse 写入。API 返回成功，表示事件已经交给 collector；之后 sink worker 才会消费和落库。因此，刚上报事件时立刻查询到空数组是正常的，客户端需要等待一小段时间再查询。

本地环境

工具准备

需要安装 Git、Docker Engine 或 Docker Desktop，以及 Docker Compose plugin。下面的命令确认工具可用：

git --version
docker --version
docker compose version
curl --version

quickstart 将 API 映射到 127.0.0.1:48888；Kafka、ClickHouse、PostgreSQL、Redis 和 Svix 只在 Compose 网络中互相访问，不需要占用宿主机的默认端口。

服务启动

从仓库根目录执行：

git clone https://github.com/openmeterio/openmeter.git
cd openmeter/quickstart
docker compose up -d --wait

--wait 会等待 Compose 文件中定义的健康检查完成。检查服务状态：

docker compose ps

本次实测中，openmeter、sink-worker、balance-worker、billing-worker、notification-service、Kafka、ClickHouse、PostgreSQL、Redis 和 Svix 都进入了 healthy 状态；openmeter-jobs 没有 HTTP 健康检查，但容器保持运行。首次启动需要下载 OpenMeter、Kafka 和 ClickHouse 镜像，时间会明显长于后续启动。

如果启动期间看到 worker 报 Kafka topic 尚不存在或连接超时，先以 docker compose ps 的最终状态为准。topic 会由 API 和 worker 在初始化阶段创建；容器没有变为 healthy 时，再查看：

docker compose logs --tail=100 openmeter sink-worker kafka

计量器配置

业务服务上报的是一条条原始事件，但产品真正关心的通常不是事件本身，而是 某个客户在一段时间内用了多少。例如，同样一条 API 请求事件，既可以计入请求次数，也可以计入执行时长；同样一条模型调用事件，还可以分别统计输入 Token 和输出 Token。若把这些计算逻辑分散到各个业务服务中，产品改一次计费口径，就要同时修改生产事件的代码和查询逻辑。

计量器（meter）就是 OpenMeter 对这种计算口径的声明。它把四件事写在一起：要匹配哪种事件、从事件数据中取什么值、使用哪种聚合方式，以及按哪些维度拆分结果。事件生产方只负责如实上报使用事实，计量器则负责把这些事实解释成“请求数”“请求时长”或“Token 用量”等指标。这样，同一份事件可以被多个计量器复用，计费规则变化时也可以优先调整配置和后续的价格模型。

在配置文件中，meters 是计量器列表，每个计量器通过 slug 定义稳定的查询名称。常见字段的关系如下：

• eventType 决定计量器接收哪类事件，必须与 CloudEvent 的 type 对应；
• aggregation 决定如何合并匹配到的事件，支持 COUNT、SUM、AVG、MIN、MAX、UNIQUE_COUNT 和 LATEST；
• valueProperty 用 JsonPath 指向数值字段，COUNT 不需要它，其他需要数值的聚合通常都需要它；
• groupBy 把事件中的字段映射为查询维度，例如按 HTTP 方法、路由或模型分别统计。

quickstart 通过 quickstart/config.yaml 挂载配置。它预置了三个计量器：

meters:
  - slug: api_requests_total
    description: API Requests
    eventType: request
    aggregation: COUNT
    groupBy:
      method: $.method
      route: $.route

  - slug: api_requests_duration
    description: API Request Duration
    eventType: request
    aggregation: SUM
    valueProperty: $.duration_ms
    groupBy:
      method: $.method
      route: $.route

  - slug: tokens_total
    description: AI Token Usage
    eventType: prompt
    aggregation: SUM
    valueProperty: $.tokens
    groupBy:
      model: $.model
      type: $.type

这几项配置表达了不同的业务含义：

• eventType 先筛选事件类型，api_requests_total 只处理 type=request 的事件；
• COUNT 每个匹配事件贡献 1，不要求事件数据中存在数值字段；
• SUM 需要通过 valueProperty 找到数值，$.duration_ms 对应事件 data.duration_ms；
• groupBy 从事件数据中提取维度，最终查询结果会把 method 和 route 原样带回来。

服务启动时，API 会把配置中的计量器写入 PostgreSQL。日志中可以看到 api_requests_total、api_requests_duration 和 tokens_total 被创建；sink worker 定期刷新计量器缓存，所以事件到达时能够知道应该匹配哪些计量器。

CloudEvents 事件

上报请求使用 CloudEvents JSON 格式。下面的事件表示 blog-customer 发起了一次 GET 请求：

{
  "specversion": "1.0",
  "type": "request",
  "id": "blog-quickstart-1",
  "source": "blog",
  "subject": "blog-customer",
  "data": {
    "method": "GET",
    "route": "/hello"
  }
}

几个字段分别承担不同职责：type 与计量器的 eventType 匹配，subject 是使用量归属对象，data 保存业务字段，source 和 id 组成事件的稳定身份。生产环境应该保证同一个来源不会为不同事件重复使用同一个 ID。

这个请求体里没有 namespace 字段。quickstart 启动时会创建默认 namespace，API server 通过静态 namespace decoder 把请求放入该 namespace；所以示例中的计量器和事件都属于同一个默认空间。多 namespace 部署时，namespace decoder 是选择目标 namespace 的入口，具体方式由 router 配置决定。

API 路由是 POST /api/v1/events。openmeter/ingest/httpdriver/ingest.go 根据 Content-Type 解析单个事件或批量事件，成功后返回 204 No Content。解析完成后，openmeter/ingest/service.go 会补齐缺失的时间戳，并把事件交给 collector：

if event.Time().IsZero() {
    event.SetTime(time.Now().UTC())
} else {
    event.SetTime(event.Time().UTC())
}

err := s.collector.Ingest(ctx, namespace, event)

这里的 204 代表 API 已经接收并转发事件，不代表 ClickHouse 立刻可查。这个区别正是后面轮询查询的原因。

第一次查询

发送事件：

curl -sS -o /dev/null -w 'HTTP %{http_code}\n' \
  -X POST http://localhost:48888/api/v1/events \
  -H 'Content-Type: application/cloudevents+json' \
  --data-raw '{
    "specversion": "1.0",
    "type": "request",
    "id": "blog-quickstart-1",
    "source": "blog",
    "subject": "blog-customer",
    "data": { "method": "GET", "route": "/hello" }
  }'

预期输出：

HTTP 204

随后轮询计量器。一次查询返回 {"data":[]} 并不表示事件丢失，只说明 sink 尚未完成消费：

response=''
for attempt in $(seq 1 20); do
  response=$(curl -fsS \
    'http://localhost:48888/api/v1/meters/api_requests_total/query?subject=blog-customer')
  printf 'attempt=%s %s\n' "$attempt" "$response"
  if printf '%s' "$response" | grep -q '"value":1'; then
    break
  fi
  sleep 1
done

本次实测第一次查询返回空数组，第二次查询得到：

{
  "data": [
    {
      "groupBy": {},
      "subject": "blog-customer",
      "value": 1,
      "windowStart": "2026-10-12T10:25:00Z",
      "windowEnd": "2026-10-12T10:26:00Z"
    }
  ]
}

时间窗口会随实际运行时间变化；结果中的关键是 subject 为 blog-customer，value 为 1。

事件链路

现在沿源码把刚才的结果串起来。

API 与 Kafka

openmeter/ingest/kafkaingest/collector.go 的 Collector.Ingest 先根据 namespace 解析 Kafka topic，再序列化 CloudEvent 并发布消息。默认 topic 模板来自配置 om_%s_events，所以不同 namespace 的事件会进入不同 topic。这样既能让租户数据隔离，也能让 sink worker 按 namespace 刷新计量器。

Sink 与 ClickHouse

Sink worker 从 Kafka 消费消息，先放入 buffer；达到 minCommitCount 或等待时间后触发 flush。flush 阶段会依次完成：

• 解析 Kafka 消息中的 CloudEvent 和 namespace；
• 用 source + id 检查重复事件；
• 从 meter cache 找出受影响的计量器；
• 校验事件数据并把结果写入 ClickHouse；
• 提交 Kafka offset，并发布 ingest notification。

仓库同时提供 API collector 层的可选去重包装，逻辑在 openmeter/ingest/dedupe.go 中很直接：

isUnique, err := d.Deduplicator.IsUnique(ctx, namespace, ev)
if isUnique {
    return d.Collector.Ingest(ctx, namespace, ev)
}
return nil

本次 quickstart 没有打开 API 层的全局 dedupe，而是在 quickstart/config.yaml 的 sink.dedupe 中启用了 Redis 去重，过期时间是 768h（32 天）。sink worker 在 flush 前用 openmeter/sink/models/models.go 生成包含 namespace、source 和 id 的去重键，落库成功后再写入 Redis。客户端因此可以在网络错误后安全重试同一个 source 和 id。

事件解析

计量器匹配后，openmeter/meter/parse.go 负责从 data 中提取数值和分组字段。COUNT 计量器不需要读取数值，源码直接让每个事件贡献 1.0：

if meter.Aggregation == MeterAggregationCount {
    parsedEvent.Value = lo.ToPtr(1.0)
    return parsedEvent, nil
}

对于 SUM、AVG、MIN、MAX 和 LATEST，ValueProperty 必须存在且能解析为有限的浮点数；groupBy 字段缺失时会被记为空字符串，而不会让整个事件因为缺一个维度就失败。

ClickHouse 查询

查询入口是 GET /api/v1/meters/{meterIdOrSlug}/query，路由定义位于 openmeter/server/router/meter.go。handler 最终调用 streaming.Connector.QueryMeter，ClickHouse connector 再根据计量器聚合方式、时间窗口、subject 和 groupBy 生成查询。

在 openmeter/streaming/clickhouse/meter_query.go 中，COUNT 对应 count(*)，小时窗口会使用 tumbleStart 和 tumbleEnd 计算窗口边界。也就是说，接口返回的 windowStart 和 windowEnd 并不是 API 层随手拼出的字段，而是 ClickHouse 聚合查询的结果。

重复事件

再次发送完全相同的 source=blog 和 id=blog-quickstart-1：

curl -sS -o /dev/null -w 'HTTP %{http_code}\n' \
  -X POST http://localhost:48888/api/v1/events \
  -H 'Content-Type: application/cloudevents+json' \
  --data-raw '{
    "specversion": "1.0",
    "type": "request",
    "id": "blog-quickstart-1",
    "source": "blog",
    "subject": "blog-customer",
    "data": { "method": "GET", "route": "/hello" }
  }'

API 仍会返回 204，因为请求格式和转发流程都没有问题。等待两秒后再次查询，结果仍然是 value: 1，而不是 2。去重发生在消费和落库链路中，所以客户端不需要为了避免重复计量而改变正常的重试策略。

sleep 2
curl 'http://localhost:48888/api/v1/meters/api_requests_total/query?subject=blog-customer'

为了让验证结果更明确，下一节再发送一个不同 id 的事件。最终按路由分组时，GET /hello 和 POST /items 各为 1；如果重复事件被计量，两行中的某一行就会变成 2。

聚合维度

再发送一个带时长的请求，验证 SUM 计量器：

curl -sS -o /dev/null -w 'HTTP %{http_code}\n' \
  -X POST http://localhost:48888/api/v1/events \
  -H 'Content-Type: application/cloudevents+json' \
  --data-raw '{
    "specversion": "1.0",
    "type": "request",
    "id": "blog-quickstart-2",
    "source": "blog",
    "subject": "blog-customer",
    "data": { "method": "POST", "route": "/items", "duration_ms": "40" }
  }'

轮询下面的接口，直到返回 value: 40：

curl 'http://localhost:48888/api/v1/meters/api_requests_duration/query?subject=blog-customer'

本次实测返回了 value: 40。注意 duration_ms 在事件里是字符串也可以工作，因为 ParseEvent 会把字符串解析为浮点数；生产事件最好统一字段类型，减少跨语言 SDK 的差异。

按照小时、方法和路由分组：

curl 'http://localhost:48888/api/v1/meters/api_requests_total/query?subject=blog-customer&windowSize=HOUR&groupBy=method&groupBy=route'

本次实测得到两个分组：

{
  "data": [
    { "groupBy": { "method": "POST", "route": "/items" }, "subject": "blog-customer", "value": 1 },
    { "groupBy": { "method": "GET", "route": "/hello" }, "subject": "blog-customer", "value": 1 }
  ],
  "windowSize": "HOUR"
}

省略的窗口字段会随当前时间变化。这里可以看到，COUNT 计量器只负责数事件，维度由 groupBy 决定；同一个 subject 的不同路由不会被混成一行。

计量与计费边界

到这里只使用了 meter 和 streaming 两个核心部分。它们回答的是“客户用了多少”。如果产品还要回答“客户现在能不能继续用”或“这次使用应收多少钱”，则会进入另外两条链路：

• entitlement 将 feature、customer attribution 和 grant 联系起来，balance worker 在使用量变化后重新计算余额；
• product catalog 和 subscription 描述产品、价格和客户订阅，billing worker 再把可计费工作转换为 charge 和 invoice；
• ledger 保存不可变的会计分录，负责余额和资金流的账务表达。

这些模块共享事件和 PostgreSQL 状态，但没有把所有逻辑塞进 sink worker。这样的边界让高吞吐的用量写入可以独立扩展，也让账单和余额的事务生命周期保持清晰。

环境清理

实验完成后删除容器和本地数据卷：

docker compose down -v

小结

本文从本地 Docker Compose 部署开始，完成了事件上报、异步查询、重复事件去重，以及 COUNT、SUM 和维度分组的验证。CloudEvents 统一事件格式，计量器定义统计口径，两者共同把业务中的使用记录变成可以查询的用量。

沿着源码可以看到这条链路的分工：API 接收事件并交给 Kafka，sink worker 消费、去重后写入 ClickHouse，查询接口再根据计量器定义执行聚合；PostgreSQL 则保存计量器和其他业务状态。由于接收与落库分开执行，204 返回后需要等待异步处理完成，才能在查询中看到新增用量。

这次实验覆盖的是计量能力。额度控制、订阅和账单在此基础上继续处理访问权限与费用，各自有独立的模块和生命周期。对于首次接入 OpenMeter 的产品，可以先确认事件字段、使用者归属和计量口径，再根据业务需要加入这些能力。