本文是 2023 年 3 月 4 日在第 12 届 PostgreSQL 中国技术大会发表主题演讲《基于 PostgreSQL 逻辑复制和 CDC 实现企业级分布式系统》的文本内容 (数据库管理运维与最佳实践专场)

大会视频回放和 PPT 下载链接：https://mp.weixin.qq.com/s/4azY8cm9iA_mcKWPw1jhBw

背景与调研

产品背景

XSKY 有 SDS 和 SDDC 两款产品，SDS 诞生于 2015 年，SDDC 诞生于 2021 年。这次分享的是 SDDC 产品的管理面的架构设计。

SDS 产品基于 Postgres 9.6，为了控制产品的复杂性，我们没有引入数据库消息队列组件。但是在产品中又得依赖于消息队列这样的机制，因此我们使用了两个方案：

  • 任务表 + 定时轮询
    • 消息传递及时性较低
  • Trigger
    • 效率低，性能消耗大 -因为没有直接回调，还是需要依赖于定时轮询

逻辑复制的相关概念

逻辑复制

  • 逻辑复制是根据复制标识（通常是主键）复制数据对象及其变更的一种方法。
  • 传送的是数据库的一种与存储格式无关的表达格式
    • 允许跨 Postgres 版本传递数据
    • 甚至允许向非 Postgres 程序传递数据

CDC (Change Data Capture)

  • 近实时捕获数据源的变更并且发送给下游的数据消费者
  • 不能等价于消息队列
    • 只能表达和数据源有关的数据变化
    • 产生的是顺序事件，不能按照随意顺序消费

逻辑复制调研

调研过程

  • Postgres 10 开始
  • Postgres 13 开始预研
    • 设计并实现了一个小项目，验证逻辑复制和 CDC 方案的可行性
    • 向 <github.com/jackc/pglogrepl> 贡献 pgoutput 协议解析代码 By @diabloneo
    • 性能测试
      • Postgres 13
      • Intel(R) Xeon(R) Gold 5218R CPU @ 2.10GHz
      • 每分钟可以发送超过 50,000 个简单的事务
  • 生产版本使用的是 Postgres 14

问题预判

  • 逻辑事件只能包含部分的数据库操作
    • 缺少的那些，在我们的系统里都可以通过带外的方式来解决。
  • 逻辑复制事件不会包含一行记录的所有内容。
    • 我们只会依赖消息中的 id 和几个时间戳字段，整个记录的内容会使用 ORM 从数据库重新读取。
  • 事件丢失 我们一定要做好事件可能会丢失的准备，提供后备方案。
  • 处理阻塞导致 WAL 写满的情况

架构设计

整体架构

architecture

  • API Server
    • 负责和用户交互，并进行数据库读写
    • 消费 LR 消息，用于发送 websocket 消息等
  • Controller
    • 消费 LR 消息，用于实现业务逻辑
  • Agent
    • LR 消息会触发 informer 重新载入数据
    • 会根据 cache 中的数据对业务做收敛

App 设计

app_1

  • 代表一个业务逻辑的分组，例如虚拟机，块存储等
  • App 是在另外一个维度上把 apiserver, controller 和 agent 联系在了一起
  • API server 和 controller 之间使用 LR 作为联系方式
  • Controller 和 agent 之间使用 informer 作为联系方式

app_2

  • 每个 app 会注册一个独立的 publication + slot
  • App 按照顺序处理自己订阅的事件
  • 不同的 app 会处理同一个事件
    • 更新数据库时，需要使用 etag 这类乐观锁机制
    • 遇到 etag 冲突时，自动重试

CDC 封装与应用

CDC Event

cdc_event_flow

  • Postgres 的 LR 消息过于原始，不利于应用开发
  • Event and EventGroup
    • Event: Insert/Update 消息，Relation 用于触发一个 cache 的更新，Commit 被映射为 FlushLSN
    • EventGroup: 一个事务中的所有 数据操作 Event 的集合
    • CDC Manager 会将 LR 消息转化为对应的 ORM Model

CDC 的应用

  • App
    • 消费 event，根据 event 执行数据库的 update 操作
  • API Client Manager
    • 监听 Node 和 Service 资源的 event，对所管理的 API Client 进行操作：创建、删除、failover 等
  • Websocket 通知
    • 监听所有资源的 event，一旦资源有变动就可以发送 websocket 通知。
  • Informer Monitor
    • 监听所有资源的 event，通过 etcd 通知 agent reload 相关的缓存数据
    • Agent 不直接消费 CDC event 的原因
      • 为了实现 agent 的 scale-out，agent 不直接访问数据库
      • Agent 中的 executor 需要一次载入某个时刻 (RR Transaction) 的多个表的数据
      • Executor 的运行需要综合定时器触发和 CDC event 触发等多种原因

informer_architecture_1

关键问题处理

Slot 的管理 – 未使用 Patroni 时

  • 在我们的 controller 程序中进行管理（在 controller leader 节点进行管理）。
    • 基本的做法是在 controller 启动时，检查 app 对应的 slot 是否存在，如果不存在则创建。

存在三个问题：

  1. 第一次启动耗时间较长，可能会丢 CDC event
  2. Controller failover 时会漏掉一些 CDC event
  3. Patroni 会尝试 drop 掉它不认识的 slot

Slot 的管理 – 使用 Patroni

  • 我们将 slot 管理交给 Patroni 来做，同时解决了上面这些问题：
    • 我们实现了一个 slot sync 命令，会在系统安装时通过 Patroni 创建好 slot。因为所有的程序都是在这个过程之后启动的，所以避免了 CDC 事件的丢失。
    • Patroni 管理 slots 后，它会在 primary/replicas 之间自动同步 slot 的 restart_lsn(10s 一次)。
      • Failover 后会收到重复的 CDC 事件，需要做幂等处理。
    • 所有 slot 受 Patroni 管控，所以 Patroni 也不会再去 drop slots。

临时 slot 的应用

  • 临时 slot
    • 不会将 slot 的信息持久化
    • 会话结束或者发生错误时，会自动销毁。
  • 使用场景
    • 允许 CDC 事件丢失的场景
      • Websocket
      • API Client Manager
  • 更换为临时 slot 的原因
    • 非 Patroni 管理的持久化 slot，会被 Patroni 尝试 drop，会产生很多 log。
    • Patroni 在管理 slot 的时候，会过滤掉所有的临时 slot。

升级

upgrade

Publication 必须在 slot 之前创建，否则 subscribe 时，server 会报找不到 publication (Postgres 实现导致的):

Thread: How is this possible “publication does not exist”

CDC 事件丢失

  • 丢失原因
    • WAL 满了，drop 掉老数据
    • Bug
  • CDC 事件重放机制
    • 避免线上出现数据丢失的情况时，需要手动去做数据库操作
    • 需要能够区分原始事件和重放的事件

CDC 事件重放机制 - Insert

数据库的 insert 事件只能通过插入新的记录来触发。如果我们要触发一个 insert 事件，那么就得把记录先删除，然后再插入一次。这个方案有几个问题：

  • 因为重新插入记录，所以无法区分一条记录是否被重放了。
  • 删除记录会导致一条需要回收的记录产生，这会增大数据库的空间。虽然这个数量不会很多，但是还是尽量避免。

方案：在 model 中统一加入一个字段 CdcInserted ，类型是 *time.Time 。重放的流程如下：

cdc_replay_insert

CDC 事件重放机制 - Update

Update event 的重放其实可以直接通过更新 UpdatedAt 字段来触发，不过这样不会保存重放的记录，也无法标记是一个重放的事件。

方案：在 model 中统一加入一个字段 CdcUpdated ，类型是 *time.Time 。重放的流程如下：

cdc_replay_update

CDC 事件重放机制 – replay 命令

为了可以在线上方便的进行操作，我们按照资源的视角开发了一个命令行工具：

$ sddc-manage cdc replay --help 
NAME: 
    manage cdc replay - Replay CDC events 
USAGE: 
    manage cdc replay [command options] [arguments...] 
OPTIONS: 
    --object-type value, -t value Object type to be replayed (required) 
    --event value, -e value Replay event type (required) (insert|update) 
    --id value Filter: ID of object to be replayed 
    --from-id value Filter: Object id >= from-id will be replayed 
    --to-id value Filter: Object id <= to-id will be replayed 
    --help, -h show help

$ sddc-manage cdc replay --id 1 -t VmImageSpec -e insert 
$ sddc-manage cdc replay --id 1 -t VirtualMachineSpec -e update

总结

运行数据

当前版本的 slot 数量：

  • Persistent: 40
  • Temporary: 6

一个内部使用的生产环境。Controller leader 运行了 18 天，接收的 LR 消息数量：

  • Update: 966961
  • Insert: 6276
  • Relation: 1149

总结

亮点：

  • Postgres 的逻辑复制很适合在分布式系统中使用
    • 可以在很大程度上免去对消息队列的使用，简化系统架构
    • 性能不错
    • 稳定性不错
  • Golang 的生态对于逻辑复制的支持已经比较不错

需要注意的地方：

  • 需要了解逻辑复制的原理，并且能够管理 publication 和 slot
  • 消费 LR 消息的时候，尽可能的不阻塞，避免 WAL 被 drop
  • 需要理解 CDC 的思想，不能将逻辑复制当成消息队列来使用
  • LR 消息的消费者，尽可能实现幂等操作

知识共享许可协议本作品采用知识共享署名-非商业性使用-禁止演绎 4.0 国际许可协议进行许可。