基于PostgreSQL逻辑复制和CDC实现企业级分布式系统

基于PostgreSQL逻辑复制和CDC实现企业级分布式系统 刘陈泓 1背景与调研 目2架构设计 Contents 录 3CDC封装与应用 4关键问题处理 5总结 https://ww.xsky.com 刘陈泓|资深研发架构师 负责SDS产品和SDDC产品管理面的架构设计 01 XSKY有SDS和SDDC两款产品，SDS诞生于2015年，SDDC诞生于2021年。这次分享的是SDDC产品的管理面的架构设计。 SDS产品基于Postgres9.6，为了控制产品的复杂性，我们没有引入数据库消息队列组件。但是在产品中又得依赖于消息队列这样的机制，因此我们使用了两个方案： 任务表+定时轮询 •消息传递及时性较低 Trigger •效率低，性能消耗大 •因为没有直接回调，还是需要依赖于定时轮询 逻辑复制 •逻辑复制是根据复制标识（通常是主键）复制数据对象及其变更的一种方法。 •传送的是数据库的一种与存储格式无关的表达格式 •允许跨Postgres版本传递数据 •甚至允许向非Postgres程序传递数据 CDC(ChangeDataCapture) •近实时捕获数据源的变更并且发送给下游的数据消费者 •不能等价于消息队列 •只能表达和数据源有关的数据变化 •产生的是顺序事件，不能按照随意顺序消费 Postgres10 开始关注 Postgres13 预研项目 Postgres14 生产版本 设计并实现了一个小项目，验证逻辑复制和CDC方案的可行性 向github.com/jackc/pglogrepl贡献pgoutput 协议解析代码 •By@diabloneo 性能测试 •Postgres13 •Intel(R)Xeon(R)Gold5218RCPU@2.10GHz •每分钟可以发送超过50,000个简单的事务 问题预判 •逻辑事件只能包含部分的数据库操作 •缺少的那些，在我们的系统里都可以通过带外的方式来解决。 •逻辑复制事件不会包含一行记录的所有内容。 •我们只会依赖消息中的id和几个时间戳字段，整个记录的内容会使用ORM从数据库重新读取。 •事件丢失 •我们一定要做好事件可能会丢失的准备，提供后备方案。 •处理阻塞导致WAL写满的情况 02 APIServer •负责和用户交互，并进行数据库读写 •消费LR消息，用于发送websocket消息等 Controller •消费LR消息，用于实现业务逻辑 Agent •LR消息会触发informer重新载入数据 •会根据cache中的数据对业务做收敛 •代表一个业务逻辑的分组，例如虚拟机，块存储等 •App是在另外一个维度上把apiserver,controller和agent联系在了一起 •APIserver和controller之间使用LR作为联系方式 •Controller和agent之间使用informer作为联系方式 每个app会注册一个独立的publication+slot App按照顺序处理自己订阅的事件 不同的app会处理同一个事件 •更新数据库时，需要使用etag这类乐观锁机制 •遇到etag冲突时，自动重试 03 Postgres的LR消息过于原始，不利于应用开发 EventandEventGroup •Event:Insert/Update消息，Relation用于触发一个cache的更新，Commit被映射为FlushLSN •EventGroup:一个事务中的所有数据操作Event的集合 •CDCManager会将LR消息转化为对应的ORMModel App •消费event，根据event执行数据库的update操作 APIClientManager •监听Node和Service资源的event，对所管理的APIClient进行操作：创建、删除、failover等 Websocket通知 •监听所有资源的event，一旦资源有变动就可以发送websocket通知。 InformerMonitor •监听所有资源的event，通过etcd通知agentreload相关的缓存数据 •Agent不直接消费CDCevent的原因 •为了实现agent的scale-out，agent不直接访问数据库 •Agent中的executor需要一次载入某个 时刻(RRTransaction)的多个表的数据 •Executor的运行需要综合定时器触发和CDCevent触发等多种原因 04 在我们的controller程序中进行管理（在controllerleader节点进行管理）。 基本的做法是在controller启动时， Patroni会尝试drop掉它不认识的slot 检查app对应的slot是否存在，如果不存在则创建。 三个问题： 第一次启动耗时间较长，可能会丢CDCevent Controllerfailover时会漏掉一些CDCevent 我们将slot管理交给Patroni来做，同时解决了上面这些问题： •我们实现了一个slotsync命令，会在系统安装时通过Patroni创建好slot。因为所有的程序都是在这个过程之后启动的，所以避免了CDC事件的丢失。 •Patroni管理slots后，它会在primary/replicas之间自动同步slot的restart_lsn(10s一次)。 •Failover后会收到重复的CDC事件，需要做幂等处理。 •所有slot受Patroni管控，所以Patroni也不会再去dropslots。 临时slot •不会将slot的信息持久化 •会话结束或者发生错误时，会自动销毁。 使用场景 •允许CDC事件丢失的场景 •Websocket •APIClientManager 更换为临时slot的原因 •非Patroni管理的持久化slot，会被Patroni尝试drop，会产生很多log。 •Patroni在管理slot的时候，会过滤掉所有的临时slot。 Publication必须在slot之前创建，否则subscribe时，server会报找不到publication(Postgres实现导致的) Thread:Howisthispossible"publicationdoesnotexist" 丢失原因 •WAL满了，drop掉老数据 •Bug CDC事件重放机制 •避免线上出现数据丢失的情况时，需要手动去做数据库操作 •需要能够区分原始事件和重放的事件 数据库的insert事件只能通过插入新的记录来触发。如果我们要触发一个insert事件，那么就得把记录先删除，然后再插入一次。这个方案有几个问题： •因为重新插入记录，所以无法区分一条记录是否被重放了。 •删除记录会导致一条需要回收的记录产生，这会增大数据库的空间。虽然这个数量不会很多，但是还是尽量避免。 方案：在model中统一加入一个字段CdcInserted，类型是*time.Time。重放的流程如下： Updateevent的重放其实可以直接通过更新UpdatedAt字段来触发，不过这样不会保存重放的 记录，也无法标记是一个重放的事件。 方案：在model中统一加入一个字段CdcUpdated，类型是*time.Time。重放的流程如下： 为了可以在线上方便的进行操作，我们按照资源的视角开发了一个命令行工具： $sddc-managecdcreplay--helpNAME: managecdcreplay-ReplayCDCeventsUSAGE: managecdcreplay[commandoptions][arguments...]OPTIONS: --object-typevalue,-tvalueObjecttypetobereplayed(required) --eventvalue,-evalueReplayeventtype(required)(insert|update) --idvalueFilter:IDofobjecttobereplayed --from-idvalueFilter:Objectid>=from-idwillbereplayed --to-idvalueFilter:Objectid<=to-idwillbereplayed --help,-hshowhelp $sddc-managecdcreplay--id1-tVmImageSpec-einsert $sddc-managecdcreplay--id1-tVirtualMachineSpec-eupdate 05 当前版本的slot数量： •Persistent:40 •Temporary:6 一个内部使用的生产环境。Controllerleader运行了18天，接收的LR消息数量： •Update:966961 •Insert:6276 •Relation:1149 亮点： •Postgres的逻辑复制很适合在分布式系统中使用 •可以在很大程度上免去对消息队列的使用，简化系统架构 •性能不错 •稳定性不错 •Golang的生态对于逻辑复制的支持已经比较不错 需要注意的地方： •需要了解逻辑复制的原理，并且能够管理publication和slot •消费LR消息的时候，尽可能的不阻塞，避免WAL被drop •需要理解CDC的思想，不能将逻辑复制当成消息队列来使用 •LR消息的消费者，尽可能实现幂等操作