fuchencong

The way I am

0%

OpenMeter 学习 02：架构总览

发表于 分类于

一个按 Token 收费的 AI 服务，最初可能只需要一张用量表：请求完成后记下客户、模型和 Token 数量，月底求和，再乘上单价。随着事件增多，问题逐渐出现在记录和查询链路上：客户端重试可能产生重复记录，离线任务可能补报昨天的用量，逐条写入会拖慢存储，刚上报的事件也未必能立即查到。要得到可信的用量，首先需要弄清这些事件经过了哪些处理，以及每一步成功究竟意味着什么。

OpenMeter 把事件接入、后台写入和计量查询拆开，分别交给 API server、Kafka、Sink worker 和 ClickHouse 协作完成。业务配置及其他需要事务约束的状态保存在 PostgreSQL，内部消息和通知机制则连接后续处理。理解这些组件之间的数据流和一致性边界，比记住项目用了哪些数据库更有帮助。

本文以仓库提交 95794e610 为依据，围绕进程分工、代码组织、启动与请求处理、事件接入、批量落库、计量查询和消息投递展开，最后介绍运行观测。代码片段均摘自当前实现，示意配置和简化 SQL 会单独说明。

系统轮廓

数据与状态

使用记录和订阅状态对存储的要求差别很大。

使用事件数量多，主要操作是追加写入，以及按客户、时间和业务维度扫描聚合。订阅、价格、账单则需要维护实体之间的约束，一次修改往往涉及多张表，而且不能只成功一半。OpenMeter 因此将两类数据放在不同的存储中：

组件 主要职责 典型数据
Kafka 分离事件接收与后台处理，承载使用事件及领域事件 使用记录、订阅变更、账单变更
ClickHouse 保存使用事件，执行计量查询 Token、请求数、时长及其聚合结果
PostgreSQL 维护需要事务约束的业务状态 客户、计量器定义、订阅、额度快照、费用、账单、账务分录
Redis 按配置提供共享去重或查询进度状态 事件去重键、长查询进度
Svix 承接 webhook 的对外投递 投递请求、重试和投递结果

这里容易忽略的是，计量器定义和用量数据分属两个存储。

  • 计量器定义决定某类事件怎样解释，属于业务配置
  • 海量事件及其查询则交给 ClickHouse
  • Redis 是可选依赖，具体是否使用取决于配置

运行进程

项目把不同工作负载拆成了多个程序入口，但它们复用同一批 Go 领域包。部署上可以分开扩缩容，代码上仍然共享业务类型和服务实现。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
Usage ingestion

  Application / SDK
         |
         | CloudEvents
         v
    API server ---> Kafka (usage) ---> Sink worker ---> ClickHouse
                                           |
                                           +-- optional dedupe --> Redis

Domain processing

    API server -- outbox -----------+
                                    |
   Sink worker -- stored notice ----+--> Kafka (domain events / notices)
                                                      |
                         +----------------------------+------------+
                         |                            |            |
                         v                            v            v
                  Balance worker              Billing worker   Notification
                         |                            |          service
                         +-- value / invoice events --+
                                        |
                                        +--> Kafka (domain events / notices)

State and delivery

    API server / Balance worker / Billing worker -- query --> ClickHouse
    API server / Balance worker / Billing worker <-- state --> PostgreSQL
    Notification service -- notification records ----------> PostgreSQL
    API server -- notification delivery --> Svix -- webhook --> Application

图中分成三个部分，分别展示事件的接入路径、业务处理过程，以及各进程依赖的存储和对外投递关系：

  • Usage ingestion（使用事件接入）：这一部分负责把“客户消耗了一千二百 Token”这样的使用事实保存下来。业务应用通过 CloudEvents 向 API server 上报事件，API server 将其交给 Kafka 生产者；Sink worker 再消费消息，完成解析、校验、去重和批量写入 ClickHouse。图中的 Redis 是可选的去重存储。接收与落库异步进行，因此接入接口返回成功后，用量还需要经过后台处理才能被查询到。

  • Domain processing（后台业务处理）：这里的“领域”指额度、订阅、计费等业务范围。

    • 例如，用户通过 API 创建或修改订阅时，领域服务会在保存订阅状态的同一笔 PostgreSQL 事务中，将对应的变更事件写入 outbox（待发送消息表）；事务提交后，后台投递逻辑再将事件发送到 Kafka，供计费模块同步相关计费项目。
    • 另一类消息来自 Sink worker：它完成一批使用事件的 ClickHouse 写入后，按 namespace 和 subject 组织落库通知，携带受影响的计量器、时间标记和原始事件等信息，让下游知道哪些用量发生了更新，从而触发相关权益余额的重新计算
    • 各 worker 消费自己关心的消息：Balance worker 重算权益余额，Billing worker 执行计费同步等工作，Notification service 匹配通知规则并保存通知记录。余额和账单更新还会产生新事件，继续触发下游处理；返回 Kafka 的箭头表达的就是这层衔接。

  • State and delivery（状态存储与对外投递）：这一部分展示各进程共享的存储依赖，以及通知如何送到业务系统。API server、Balance worker 和 Billing worker 从 ClickHouse 查询用量，在 PostgreSQL 中读写客户、权益、订阅、费用和账单等业务状态；通知服务也将通知记录保存到 PostgreSQL。随后，API server 内的投递协调逻辑把待发送通知交给 Svix，由 Svix 向业务应用发送 webhook。图中这些连线贯穿前两部分，是它们工作时所依赖的数据和投递通道。

图中省略了定时任务和观测设施。各可执行程序的职责如下：

  • server 接收事件，提供管理和查询 API，并运行通知投递协调逻辑。
  • sink-worker 消费使用事件，完成解析、校验、去重和批量写入。
  • balance-worker 在用量或额度相关状态变化后重新计算客户的功能权益值。
  • billing-worker 处理订阅同步、费用推进和账单相关后台工作。
  • notification-service 匹配通知规则，生成持久化通知记录。
  • jobs 承担迁移、定期同步和生命周期推进等任务。

这些职责之间通过事件和持久化状态衔接。一次请求无需同步等到所有 worker 完成，代价是不同阶段的完成时间并不相同。后文会重点分析接入确认、去重、消费位点和事务消息如何协调这些阶段。

系统还提供一个可选的外部采集程序 Collector，用于采集、变换、缓冲并转发事件。它与接入服务内部名为 Collector 的 Go 接口是两个层面的概念。

业务示例

以一个 AI 服务为例。工作区标识为 workspace-acme，Token 计量器使用 SUM 聚合事件中的 tokens 字段。假设此前没有上报用量，这次模型调用消耗了一千二百 Token。下面用这条记录串起图中的三个部分。

Usage ingestion 对应这次消费的记录过程。 模型网关完成调用后，发送一条事件，携带工作区标识、发生时间、Token 数量以及稳定的事件 ID。API server 将事件交给 Kafka 生产者，Sink worker 随后消费并批量写入 ClickHouse。处理完成后，按该工作区查询计量器，就能得到一千二百 Token 的用量。

Domain processing 对应用量更新后的后台协作。 各组件通过消息衔接处理：

  • Sink worker：在用量写入后发布落库通知，说明哪些工作区和计量器的用量发生了更新。
  • Balance worker：接到相关通知后，查询这份用量，更新功能权益的计算结果，并发布新的值事件。具体额度算法留到下一篇讨论。
  • Notification service：如果值事件满足已配置的提醒规则，就保存一条待投递通知。
  • 其他后台组件：消费各自关心的事件。比如订阅变更通过 outbox 触发计费处理，与使用事件的落库链路各自推进。

State and delivery 贯穿上述过程。 一千二百 Token 的原始事件保存在 ClickHouse，计量器定义、业务状态和通知记录保存在 PostgreSQL。查询接口和后台组件可以读取同一份用量。待投递通知先存入 PostgreSQL，再由 API server 内的投递协调逻辑交给 Svix，最终以 webhook 发送给业务应用。

这条记录说明了三个不同的完成点：API 接收了事件，ClickHouse 已经能查到新增用量，下游通知最终送达业务系统。它们由同一次消费关联起来，但分别经过生产者、消费写入和消息投递流程，完成时间并不相同。

代码组织

在跟踪业务之前，先认识源码的分工，可以省去大量在生成代码中寻找逻辑的时间。

目录总览

仓库同时包含服务端、外部采集器、API 定义与客户端，以及本地运行和部署配置。下面保留与本文主线有关的目录，省略各领域内部的完整文件清单：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
openmeter/
|-- cmd/                         # 可执行程序入口
|   |-- server/                  # HTTP API 与通知投递协调
|   |-- sink-worker/             # 使用事件消费与落库
|   |-- balance-worker/          # 权益余额重算
|   |-- billing-worker/          # 计费后台处理
|   |-- notification-service/    # 通知规则处理
|   `-- jobs/                    # 迁移与定时任务
|-- app/
|   |-- common/                  # 跨入口复用的依赖构造与组装
|   `-- config/                  # 配置解析、校验及默认值
|-- openmeter/                   # 核心领域与基础设施实现
|   |-- ingest/                  # 事件接入与 Kafka 转发
|   |-- sink/                    # 消费缓冲、去重、批量写入
|   |-- streaming/               # 用量查询接口与 ClickHouse 实现
|   |-- meter/                   # 计量器定义与管理
|   |-- customer/                # 客户与用量归属
|   |-- productcatalog/          # 功能、套餐、价格等产品配置
|   |-- subscription/            # 订阅时间表及变更流程
|   |-- entitlement/             # 客户功能权益
|   |-- credit/                  # 计量权益的额度消耗与余额快照
|   |-- billing/                 # 费用、账单、计费同步与生命周期
|   |-- ledger/                  # 账务分录、余额和更正
|   |-- notification/            # 通知规则、记录与投递协调
|   |-- watermill/               # 领域事件总线、Kafka 驱动与 outbox
|   |-- server/                  # 服务端路由与旧版 API 适配
|   `-- ent/
|       |-- schema/              # PostgreSQL 实体模型定义
|       `-- db/                  # Ent 生成的数据访问代码
|-- api/
|   |-- spec/                    # TypeSpec 定义及代码生成工具
|   |-- client/                  # 多语言客户端
|   `-- v3/
|       |-- handlers/            # v3 HTTP 操作实现
|       |-- server/              # v3 服务端组装与路由注册
|       `-- client/              # 独立 Go SDK 模块
|-- pkg/                         # 通用组件与工具包
|   `-- framework/               # HTTP 传输、事务、锁等公共机制
|-- collector/                   # 独立的外部事件采集程序与 Go 模块
|-- test/                        # 跨领域测试与集成场景
|-- e2e/                         # 通过 HTTP 访问服务的端到端测试模块
|-- quickstart/                  # 本地评估配置与 Docker Compose
|-- deploy/                     # Helm chart 等部署资源
|-- docs/                       # 架构、开发指南与迁移说明
|-- tools/                      # 开发及迁移辅助工具
|-- Makefile                    # 常用开发任务入口
`-- go.mod                      # 服务端主模块依赖

目录树最外层是整个项目，内部同名的 openmeter 目录承载主要业务实现。从调用关系看，HTTP 请求先经过服务入口和路由，再进入领域服务，最后由 adapter 访问数据库。公共装配层把这些实现连接起来，通用工具层则提供多个领域都会使用的机制。

项目包含四个 Go 模块，分别服务于核心程序、外部采集器、可发布的 v3 Go SDK 和端到端测试。它们独立管理依赖，根模块的构建与测试不会自动覆盖其他嵌套模块。

领域分层

大部分领域拥有自己的 Go 包。领域根包通常放类型、接口、输入和错误；service 层实现业务规则和事务编排；adapter 层负责数据库访问和实体转换。HTTP 层将协议输入映射成领域调用，底层由 Ent 访问 PostgreSQL。

例如，客户模块通过 Service 接口向调用方暴露能力，service 实现客户管理规则，adapter 处理客户实体的保存与查询。较早的模块还保留了 Connector、Repo、driver 等命名，阅读时应以实际调用关系为准。

服务装配

公共装配层和各程序入口共同完成依赖组装。以 API server 为例，Application 持有客户、计量、订阅、计费、消息发布等依赖，Wire 根据声明的构造器集合生成具体的组装代码。因此，同一个计费服务可以被 API 和后台 worker 共同使用，不需要复制业务逻辑。

HTTP 入口同时保留旧版 API 和 v3 API。服务端统一注册两套路由，分别由各自的 handler 完成协议适配，最终调用共享的领域服务。本文从较容易跟踪的 v1 事件接入路径切入，再进入共享的领域实现。

启动与退出

初始化顺序

API server 使用 Viper 处理配置，合并默认值、配置文件、环境变量及已绑定的命令行参数，再解码为配置结构并调用 Validate()。--validate 可以在校验后直接退出，这一步只检查配置，不等于依赖服务已经连通。

通过校验后，主程序调用 Wire 生成的 initializeApplication 构造依赖，再按顺序完成数据库迁移、namespace 处理器注册和默认 namespace 初始化，并准备默认应用、默认计费配置及配置文件中的计量器。这里调用的是统一迁移入口，具体迁移行为仍受配置控制。随后才构造 HTTP 路由，启动各个长期运行的任务。

1
2
3
4
5
6
7
8
9
10
11
12
13
Configuration loading and validation
                 |
                 v
Dependency construction (Wire)
                 |
                 v
Database / namespace / meter initialization
                 |
                 v
HTTP routes and runtime task registration
                 |
                 v
API + telemetry + Kafka reports + notification reconciler

这张图以 API server 为例。Sink worker 的准备工作有所不同：它在消费循环开始前加载计量器缓存、初始化落库回调，并订阅相关 Kafka topic。不同入口复用基础设施，但各自负责启动前的必要准备。

任务协调

一个进程内部并不只有 HTTP 服务。API server 还需要持续读取 Kafka 生产者的投递报告，运行通知投递协调循环，并提供独立的观测接口。主程序通过 run.Group 将这些工作注册为成对的“运行函数”和“停止函数”。一个任务退出时，运行组会调用其他任务的停止逻辑，再等待退出。

API server 和 Sink worker 都监听 SIGINT、SIGTERM，并使用反向停止顺序：

1
2
3
group.Add(run.SignalHandler(ctx, syscall.SIGINT, syscall.SIGTERM))

err = group.Run(run.WithReverseShutdownOrder())

API server 的退出流程还会将终止检查标记为正在退出；收到退出信号时，等待配置的传播时间，让就绪状态的变化有机会被外部负载均衡感知。之后 HTTP 服务通过带超时的 Shutdown 停止接收新连接，并等待活动请求结束。依赖清理函数则负责关闭相关资源。

退出阶段需要按组件判断数据是否已经处理完成。Sink 的 Close() 与 flush 共用互斥锁，会等待正在执行的 flush 离开临界区，再停止消费状态和定时器；它本身不会主动把所有剩余缓冲再写一遍。尚未保存处理位点的消息，在后续恢复消费时可能重新读取。Outbox 关闭同样不意味着队列已经清空，仍待发送的记录可以留在 PostgreSQL 中等待后续进程处理。

请求处理

协议适配

OpenMeter 使用 Chi 组织路由，并在请求外围安装请求标识、客户端地址处理、日志、异常恢复及 OpenTelemetry 等中间件。旧版 API 还使用 OpenAPI 请求校验中间件。两套 API 的适配细节有所不同，但共享的通用 HTTP 框架将处理过程拆成解码、操作执行和响应编码：

1
2
3
4
5
6
7
8
9
10
11
12
HTTP request
     |
     v
Routing and middleware
     |
     v
Request decoder --> Operation --> Response encoder --> HTTP response
        |               |                |
        +---------------+----------------+
                        |
                        v
              Error handling / diagnostics

以事件接入为例，decoder 解析内容类型和事件正文，operation 调用 IngestEvents，encoder 返回空的 204 响应。它们通过通用 handler 连接起来，而不是在一段函数中同时处理协议和转发逻辑。

这样的拆分也划定了校验边界：协议层处理格式与参数，领域服务维护自身约束，消费端继续校验 Kafka 消息。请求通过 HTTP 校验，并不能代替后续处理阶段的校验。

上下文与错误

通用 handler 从 r.Context() 取得请求上下文，并继续传给 decoder 和 operation。请求属性、日志关联信息以及调用取消信号，都可以沿着这条调用链传播。namespace 由 handler 使用注入的解码器解析，再放入服务输入；事件接入还会将其写进 Kafka 消息头，供 Sink 继续识别数据范围。

当前 OSS API server 装配的是默认 namespace 的静态解码器，因此“代码携带 namespace”与“部署已经具备按请求身份隔离的多租户认证”是两件事。前者是各层传递范围的机制，后者取决于具体接入与认证装配。

错误处理也集中在协议边界：decoder 或 operation 返回错误后，由错误编码器转换成 HTTP 响应；写响应失败则进入诊断处理，已经发送的响应头不会重新改写。框架还会同时检查请求上下文和错误类型，将客户端取消与内部操作取消区分开。这样，连接断开、参数错误和服务失败能够在日志与响应中保留不同含义。

事件接入

事件格式

不同业务服务需要统一表达使用记录，同时又要保留各自的数据。OpenMeter 使用 CloudEvents 组织事件：外层提供身份、来源、类型和归属，data 保存业务字段。

下面是一条用于说明后续链路的示例事件：

1
2
3
4
5
6
7
8
9
10
11
12
{
  "specversion": "1.0",
  "id": "completion-001",
  "source": "model-gateway",
  "type": "llm.completion",
  "subject": "workspace-acme",
  "time": "2026-10-01T10:00:00Z",
  "data": {
    "tokens": 1200,
    "model": "model-a"
  }
}

几个字段分别解决不同问题：

  • source 和 id 标识事件身份，生产者重试同一条事件时应保留它们。
  • type 表示事件类别，让计量器选择需要解释的事件。
  • subject 表示使用记录的归属键，例如工作区或租户在业务系统中的标识。
  • time 表示使用发生时间，data 中的 Token、模型等字段用于取值和分组。

OpenMeter 还在处理链路中携带 namespace。事件身份、计量器查找和数据库查询都带有这个范围。不过，namespace 是领域隔离手段，具体怎样从请求解析出来由应用装配决定；当前 OSS server 的主入口使用默认 namespace 的静态解码器。

接收语义

v1 的 POST /api/v1/events 请求先经过路由分发，再由事件接入 handler 处理。handler 解析单条或批量 JSON，调用 IngestEvents，成功后返回 204 No Content。

接入服务的 processEvent 方法在转发前统一时间：

1
2
3
4
5
6
7
if event.Time().IsZero() {
    logger.DebugContext(ctx, "event does not have a timestamp")

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

这意味着客户端最好明确传入业务发生时间。否则，补报一条昨天的事件时，系统会把当前接收时间作为事件时间，后续聚合也就会进入不同的业务区间。

之后，服务将事件交给 Kafka collector。它的 Ingest 方法通过以下调用转发消息：

1
2
3
4
5
6
7
err = s.Producer.Produce(msg, nil)
if err != nil {
    err = fmt.Errorf("producing kafka message: %w", err)
    return err
}

return nil

这里调用的是 Kafka 客户端的异步生产接口。最终投递报告由另外的循环处理，投递失败会在那里记录。因此，这条接入路径返回成功，表示事件已被本地生产者接受，不能直接解释为 Kafka 服务端已确认持久化，更不能解释为 ClickHouse 已经可以查到用量。

批量接入也不是一笔跨事件事务。IngestEvents 逐条转发，某条失败就返回；此前的事件可能已经交给生产者。调用方重试整批时，稳定的事件身份和正确配置的去重机制就很重要。

批量落库

缓冲与分区

事件已经进入 Kafka，下一步是把消息高效地转成可查询数据。如果每收到一条就写一次 ClickHouse，会产生大量小批次写入；如果一直等到大批次，又会拉长低流量时的查询延迟。

Sink worker 用两个条件控制刷新：MinCommitCount 达到消息数门槛，或者 MaxCommitWait 达到最长等待时间，任一条件满足就触发 flush。计时器通过 channel 通知消费主循环执行刷新，因此刷新错误可以向上传递，停止消费，而不会被计时器回调悄悄吞掉。

另一个细节在 Kafka 消息键中。JSON 序列化器的 SerializeKey 方法直接复用了去重身份：

1
2
3
4
5
6
7
dedupeItem := dedupe.Item{
    Namespace: namespace,
    ID:        ev.ID(),
    Source:    ev.Source(),
}

return []byte(dedupeItem.Key()), nil

在分区集合稳定、使用一致散列分区器的情况下，同一身份的重试消息会进入同一分区。这有助于减少多个消费者同时处理同一事件的竞争。它保证的范围是事件身份，不能由此推导出同一客户的全部事件都有全局顺序。

分区交接

增加 Sink 副本后，同一消费组会重新分配分区。扩容能分担消费，但也需要处理原进程缓冲中尚未完成的消息，否则新旧消费者可能在交接期间各自继续处理同一批数据。

Sink 在 flush 前暂停当前分区，完成后再恢复消费；flush 与关闭流程之间还使用互斥锁协调。ClickHouse 写入变慢时，这个 worker 的消费推进也会放慢，已进入 Kafka 的待处理数据需要等待后续消费。批次大小和等待时间因此同时影响吞吐与延迟，不能只追求更大的批次。

分区交接由 rebalance 回调处理：

  • 获得分区：使用 OffsetStored 从已保存的消费位点恢复，增量加入新分区，并恢复当前分区集合的消费。
  • 失去分区：先暂停消费，再增量撤销分区，移除缓冲中属于这些分区的消息。尚未提交位点的部分可由接手的消费者重新读取。
  • 意外失去所有权：记录警告，因为分区可能已由其他消费者接手，位点提交未必还能成功。

如果分区没有可用的已提交位点，还要看起始策略。当前 Sink 装配明确设置 AutoOffsetResetLatest，这时会从末尾开始。因此，创建新消费组不能直接视为从头重放历史事件；恢复行为需要结合已有位点和起始策略判断。

去重范围

Sink 的 deduplicateAndResolveMeters 会先去掉当前批次中的重复事件，再按配置查询共享或本地去重器。两者的覆盖范围不同：

  • 批内去重处理一次 flush 内的重复消息，即使没有配置外部去重器，这一步也存在。
  • 内存去重器依赖当前进程中的有限缓存，无法覆盖其他副本和进程重启。
  • Redis 去重器可以共享去重状态，其覆盖时间取决于过期配置。

当前默认配置中，接入侧的 dedupe.enabled 和 sink 侧的 sink.dedupe.enabled 都是 false；driver 默认名称是 memory。因此，不能只根据“项目支持 Redis”就认为部署已经具有跨批次去重。这两处开关需要分别配置，不能把一侧启用理解为整条链路都已启用。

计量器匹配也发生在这里。Sink 按 namespace 和事件 type 从缓存中找到受影响的计量器，用于后续落库通知。当前存储实现保存的是原始事件，并不会为每个匹配到的计量器各写一份聚合值。 这一点会直接影响我们对查询架构的理解。

配置传播

计量器定义保存在 PostgreSQL，但高频事件处理不适合为每条消息都查询一次定义。Sink 因此通过 NamespacedMeterCache 在启动时加载计量器，按 namespace、事件类型建立索引，再根据 MeterRefetchInterval 周期刷新。

刷新先构造新的索引，再在写锁保护下整体替换；读取匹配结果时使用读锁。如果后台刷新失败，程序会记录错误并保留旧缓存，等待下一轮刷新；首次加载失败则会阻止消费循环启动。这样避免了瞬时查询失败直接清空已有配置，但也意味着配置变更不会立即出现在所有 Sink 副本中。

1
2
3
4
5
6
7
8
Meter definitions (PostgreSQL)
             |
             | initial load / periodic refresh
             v
Sink cache: namespace --> event type --> meters
             |
             v
Affected meters in stored-event notifications

这份缓存用于 Sink 匹配受影响的计量器，不是所有查询共享的全局缓存。新增计量器之后，Sink 何时能在落库通知中带上它，取决于本地缓存刷新；用量查询则根据自身解析得到的计量器定义生成 SQL。原始事件的保存也不要求每条事件都先找到匹配计量器。

Kafka topic 的发现是另一套周期任务。Sink 根据 topic 名称规则筛选订阅对象，只有目标集合与当前订阅不同才调用 SubscribeTopics。因为即使 topic 集合没有变化，重复订阅仍会触发消费组重新加入；每个副本都周期性重复订阅，会使正常消费不断被分区重分配打断。

写入与位点

去重之后，flush 依次执行几个关键步骤：

  1. 调用 persistToStorage，将有效事件批量写入存储。
  2. 调用消费者的 StoreMessage，记录已处理位点。
  3. 若配置了去重器，则写入去重身份。
  4. 调用刷新成功回调，发布后续用量处理通知。

StoreMessage 是更新客户端维护的位点，正常情况下由消费者自动提交。Sink worker 创建消费者时明确设置：

1
2
consumerConfig.EnableAutoCommit = true
consumerConfig.EnableAutoOffsetStore = false

禁用自动存储位点，是为了避免“消息刚取出来，还没写入 ClickHouse，就被当成处理完成”。先写入再记录位点，可以让未完成的工作在恢复后重放。

但 ClickHouse、Kafka 位点和 Redis 不在同一个事务中。如果写入已完成，进程却在记录位点或去重状态前崩溃，重放仍可能产生重复。这段实现包含针对部分失败的补救和日志，也明确存在需要根据日志协调状态的情形。将整条链路概括为无条件的 exactly-once，会掩盖这些失败窗口。

落库后的通知同样有自己的边界：OnFlushSuccess 在 goroutine 中运行，失败会记录日志。它没有与原始事件写入组成原子提交。这解释了为什么后续领域不能只依赖一次通知就假设所有衍生状态永久正确。

异常事件

重试适合处理临时故障，但一条格式损坏的消息，无论读多少次也不会自行恢复。Sink 的 parseMessage 因此区分消息内容问题与操作失败：缺少 namespace、JSON 解析失败或 CloudEvents 校验失败时，返回带 DROP 状态的消息，由后续批处理跳过写入。批次继续完成后，这些消息的位点也会被记录，避免同一条坏消息反复阻塞消费。

正常去重发现的重复消息也会被过滤。它与存储故障的处理不同：写入失败会向上返回，消费主循环结束本次运行，尚未提交位点的工作可以在恢复后重读。丢弃状态会进入处理指标，不能把“消费者仍在运行”当成“所有收到的事件都已有效落库”。

计量查询

原始事件表

接下来要解决的是：同一批事件怎样服务于不同的计量方式。以 Token 为例，有的页面需要查询一个客户本月总量，有的需要按模型和小时分组，有的功能只统计特定模型。如果写入时就把事件压成一个计数，后续查询会失去许多可用信息。

当前 ClickHouse connector 的做法是保留原始事件，由查询侧根据计量器定义生成 SQL。事件表建表逻辑的核心是：

1
2
3
4
5
6
7
8
9
sb.Define("time", "DateTime")
sb.Define("data", "String")
sb.Define("ingested_at", "DateTime")
sb.Define("stored_at", "DateTime")
sb.Define(fmt.Sprintf("INDEX %s_stored_at stored_at TYPE minmax GRANULARITY 4", d.EventsTableName))
sb.Define("store_row_id", "String")
sb.SQL("ENGINE = MergeTree")

sb.SQL("PARTITION BY toYYYYMM(time)")

同一函数随后指定排序键：

1
sb.SQL("ORDER BY (namespace, type, subject, toStartOfHour(time))")

月分区对应事件时间范围；排序键则贴近典型查询中的 namespace、事件类型、归属和时间条件。data 保存为字符串，查询时再解析 JSON 字段。这个实现依靠 ClickHouse 的列式扫描和聚合能力，代价是查询阶段仍需读取事件并解释相关字段。

这里使用普通 MergeTree。CloudEvents 的 id 也只是一个列，不能把它当作关系数据库的唯一键，更不能指望表引擎自动兜底所有重复写入。

计量器与 SQL

计量器描述“怎样从事件中取值并聚合”。下面是说明本文示例的配置片段：

1
2
3
4
5
6
7
meters:
  - slug: llm_tokens
    eventType: llm.completion
    aggregation: SUM
    valueProperty: $.tokens
    groupBy:
      model: $.model

valueProperty 定义数值路径，groupBy 定义可查询的维度路径。计量器不关心这些 Token 价值多少钱，也不决定客户拥有多少额度。

计量查询构造器将计量器聚合类型映射为 ClickHouse 函数：SUM 对应 sum，COUNT 对应 count，UNIQUE_COUNT 对应 uniqExact，LATEST 对应按事件时间选择值的 argMax。数值提取有浮点和可选的 Decimal 精度分支，JSON 路径和值通过查询构造器传参。

例如，查询一个工作区按小时、按模型分组的 Token 用量，在浮点分支下可以理解为下面的简化 SQL。这里用字面值替代实际参数，省略设置项和部分投影：

1
2
3
4
5
6
7
8
9
10
11
12
SELECT
    tumbleStart(time, toIntervalHour(1), 'UTC') AS windowstart,
    sum(ifNotFinite(toFloat64OrNull(JSON_VALUE(data, '$.tokens')), NULL)) AS value,
    JSON_VALUE(data, '$.model') AS model
FROM events
WHERE namespace = 'default'
  AND type = 'llm.completion'
  AND subject = 'workspace-acme'
  AND time >= toUnixTimestamp('2026-10-01 00:00:00', 'UTC')
  AND time < toUnixTimestamp('2026-10-02 00:00:00', 'UTC')
GROUP BY windowstart, model
ORDER BY windowstart

查询时间范围采用左闭右开区间 [from, to)。这样，连续两个区间在边界上的事件不会被重复包含。计量器本身还可以配置 EventFrom，实际起点取查询起点和该边界中较晚的一个。

由此也能看出，不能把当前 connector 描述成“每个 meter 都拥有一个按分钟聚合的物化视图”。实际查询入口 Connector.QueryMeter 构造 queryMeter，再对事件表执行聚合。

时间的含义

原始表同时保存三个时间：

字段 含义 主要作用
time 使用发生时间 决定用量属于哪个服务区间
ingested_at 接入链路记录的时间 观察接入延迟、排查补报
stored_at sink 为本次写入标记的时间 限定一次计费计算可见的事件集合

stored_at 在 flush 中赋值，随后随原始事件写入；它是处理链路记录的写入时间标记，不是数据库另行生成的事务提交时刻。

假设一条十点发生的事件十一点才补报。按 time 查询，它仍属于十点的用量；但如果查询同时要求 stored_at 早于十一点，就不会包含这条后到的记录。查询构造器支持同时使用业务时间范围和写入时间过滤，从而区分“哪个时间段的消费”和“截至某个处理时刻已经记录了哪些事件”。

消息与通知

事务消息

前面多次提到领域事件。对订阅等事务状态来说，最危险的情形是：数据库已经提交，Kafka 发布却失败，后台处理永远不知道发生了变化。

当前装配为 system-events topic 启用了 PostgreSQL outbox。Outbox 发布器复用调用方事务，写入消息，并注册提交后唤醒：

1
2
3
4
5
6
7
8
9
10
11
_, err = client.EventOutbox.Create().
    SetMessageID(msg.UUID).
    SetTransactionID(tx.ID()).
    SetTopic(topic).
    SetPayload(msg.Payload).
    SetMetadata(map[string]string(msg.Metadata)).Save(ctx)
if err != nil {
    return fmt.Errorf("enqueue system event: %w", err)
}

return tx.AfterCommit(p.signal)

这样，业务变更和待发送消息要么一起提交，要么一起回滚。提交后后台 worker 将消息发往 Kafka，也会通过周期重试继续处理积压。当前实现并非仅在后续有新消息发布时才重试。

Outbox 仍有明确范围和限制：

  • 应用装配只为 system-events topic 开启 outbox，其他 topic 直接发布。
  • 同一个源事务的消息按序处理；其中一条失败时，后续消息暂缓，其他事务仍可继续。
  • 成功投递的记录会删除，但崩溃或数据库提交失败可能造成再次投递，消费者需要容忍重复。
  • 重试次数有限，默认达到十次后保留失败记录并跳过，需通过日志和队列状态发现与处理。

因此，outbox 解决的是数据库变更与待发送消息的原子保存，并不意味着所有消费者一定且只会执行一次。

消费重试

消息成功投递到 Kafka，只解决了发布侧的问题。后台消费者执行时仍可能遇到依赖超时或临时错误，需要自己的恢复策略。使用默认 Watermill 路由的消费者将恢复、重试、超时和死信处理组合成中间件，围绕具体 handler 执行。

重试间隔采用退避并加入随机扰动，次数和最大耗时由配置控制。如果设置了 ProcessingTimeout，每次尝试还会取得带超时的上下文。路由通过 RestoreContext 在尝试结束后恢复原上下文，避免第一次超时留下的取消状态，使下一次重试一开始就失效。超时信号仍需要 handler 及其依赖遵守，并不等于强制终止任意执行中的代码。

1
2
3
4
5
6
7
Kafka message --> Handler attempt
                       |
                       +-- success --> normal acknowledgement
                       |
                       +-- failure --> retry with backoff
                                            |
                                            +-- still failing --> DLQ

重试后仍失败的消息，由外层失败处理发送到配置的死信 topic（DLQ），留下独立检查和后续恢复的入口。路由正在关闭时会跳过这条死信路径，让消息保持可重新投递的状态，避免把关停导致的取消误判成不可恢复的坏消息。发送到死信队列也不等于业务已处理成功，仍需要后续排查。

这里讨论的是后台消息路由，与前面的 Sink 主消费循环不同。Sink 对无效使用事件标记 DROP，并不是自动把它们都转入同一个死信队列；发布侧 outbox 的重试也独立于消费者的执行重试。区分这三层，才能知道失败消息最终留在哪里。

规则与投递

领域事件到了通知服务，还要经过用户配置的规则。例如，用量达到百分之八十可以发提醒，账单创建也可以触发 webhook。通知处理拥有独立的记录与投递流程，使外部 webhook 的延迟和失败可以单独处理。

通知消费者负责匹配规则，生成保存历史内容的通知事件，并在同一事务中创建每个通知渠道（channel）的初始投递状态。没有匹配规则，就不会为该来源事件保存通知。

实际向 Svix 发送和协调投递状态的，是 API server 内带主实例选举的投递协调循环。Svix 承担外部 HTTP 投递及重试，OpenMeter 维护自己的规则、渠道、通知记录和投递状态。这也是为什么不能仅凭进程名，就认为 notification-service 独自完成了 webhook 全流程。

保存下来的消息内容是通知历史的依据。重发一条旧通知时，不应重新查询业务对象的当前状态来拼装，否则同一个通知 ID 在不同时间可能代表不同内容。通过持久化消息内容，投递和重发可以基于同一份历史记录完成。

通知也不是逐请求阻断机制。余额阈值规则依赖重新计算后的值事件，既受上游异步延迟影响，也有自己的规则匹配和投递周期。它适合提醒与业务联动，不能替代业务调用路径上的准入判断。

运行观测

健康状态

异步链路的运维不能只看 HTTP 成功率。接口可能正常返回，后台却在等待写入或重试消息。因此，OpenMeter 为 API 和 worker 提供独立的 telemetry HTTP 服务，暴露健康检查、可选的 Prometheus 指标，以及调试分析接口。

健康接口表达的含义并不相同：

接口 当前实现 能说明的范围
/healthz/live 直接返回 ok 该 HTTP 处理路径仍能响应
/healthz/ready 返回已注册健康检查的汇总结果 当前注册的检查是否通过，包括 API 的终止检查
/healthz 使用同一健康汇总处理器 各检查的状态与诊断信息

就绪检查不直接证明 Kafka 没有积压，也不保证最近上报的事件已经可查询；这些属于数据处理进度，需要通过指标和实际查询另外观察。进程退出时将终止检查变为失败，作用是让外部看到它正在退出，而不是报告某条事件的处理结果。

日志与指标

项目使用结构化日志和 OpenTelemetry 组织运行信息，HTTP 中间件记录请求指标与追踪信息，Sink 对接收、处理结果和接入延迟分别计数。下面三个指标直接对应前文的处理阶段；表中名称是源码中的 OpenTelemetry instrument 名称，Prometheus 导出时名称可能经过转换。

指标 记录内容 排查用途
sink.kafka.messages 从 Kafka 收到的消息数，带 namespace 维度 消费端是否仍在取得消息
sink.flush.events flush 处理的事件数，带 namespace 和状态维度 处理是否推进，是否出现 DROP
sink.flush.ingest.delay stored_at - ingested_at 的毫秒分布 已处理事件从接入到本批写入标记的延迟

延迟指标的计算直接使用前文介绍的时间标记：

1
2
3
4
5
6
7
delay := message.StoredAt.Sub(*message.IngestedAt).Milliseconds()

if delay < 0 {
    delay = 0
}

s.flushIngestDelay.Record(ctx, delay, metric.WithAttributes(namespaceAttr))

这里的 stored_at 在 flush 开始处理批次时赋值，因此该指标不等于完整的 ClickHouse 写入耗时，也不覆盖之后的业务计算和 webhook 投递。它只对符合记录条件的已处理消息采样；完全停滞的消费者可能没有新样本，不能把图表上最后一次较低的延迟当成当前仍然正常。

HTTP 追踪也有自己的成本控制。通用 handler 可以额外创建操作级子 span，但默认关闭，避免每个请求都增加一层追踪记录。日志、指标和追踪分别适合定位具体错误、观察总体趋势和拆分调用耗时，不需要把每条事件内容都变成指标标签。

排查线索

假设客户端上报成功，但稍后仍查不到那一千二百 Token，可以沿着已有的阶段逐步缩小范围：

  • 接入与生产者：确认事件身份、namespace 和发生时间，检查异步投递报告是否失败。HTTP 成功本身不足以证明 Kafka 服务端已接收。
  • 消费与写入：查看消息接收数是否增长、flush 是否继续完成，以及是否有分区交接、写入错误或 DROP。结合 Kafka 客户端指标与消费位点判断是否积压。
  • 查询条件：核对计量器的事件类型、取值路径、subject 和时间区间。事件落库与查询能够选中事件，是两个检查点。
  • 后续处理：若用量已经可查，继续检查落库通知、消费者重试或死信记录；若通知已经保存，则检查 Svix 投递和通知状态协调。

这些检查分别对应接入确认、持久化、查询可见和外部投递，避免把不同阶段的问题都归结为“事件丢了”。

小结

回到开头的 AI 服务，一条包含一千二百 Token 的使用记录，先由 API 接收并交给 Kafka 生产者，再由 Sink worker 校验、去重和批量写入 ClickHouse。查询时，计量器定义被转换成聚合 SQL；后台组件则通过内部事件获知数据变化。启动顺序、分区交接、配置刷新和退出协调维持各进程的运行，消息恢复与观测机制则帮助定位尚未完成的处理阶段。

客户与产品、订阅时间表、计费同步、额度计算、费用与定价、账单生命周期，以及相关账务记录，属于这套基础架构承载的业务领域。它们如何使用计量结果、维护业务约束并协同工作，将在下一篇文章中展开。