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

刘陈泓 CDC封装与应用 关键问题处理 总结 https://ww.xsky.com 刘陈泓|资深研发架构师负责SDS产品和SDDC产品管理面的架构设计 XSKY有SDS和SDDC两款产品，SDS诞生于2015年，SDDC诞生于2021年。这次分享的是SDDC产品的管理面的架构设计。 SDS产品基于Postgres9.6，为了控制产品的复杂性，我们没有引入数据库消息队列组件。但是在产品中又得依赖于消息队列这样的机制，因此我们使用了两个方案： 任务表+定时轮询•消息传递及时性较低 Trigger •效率低，性能消耗大•因为没有直接回调，还是需要依赖于定时轮询 逻辑复制 •逻辑复制是根据复制标识（通常是主键）复制数据对象及其变更的一种方法。 •传送的是数据库的一种与存储格式无关的表达格式•允许跨Postgres版本传递数据•甚至允许向非Postgres程序传递数据 CDC(ChangeData Capture) •近实时捕获数据源的变更并且发送给下游的数据消费者•不能等价于消息队列•只能表达和数据源有关的数据变化•产生的是顺序事件，不能按照随意顺序消费 设计并实现了一个小项目，验证逻辑复制和CDC方案的可行性向github.com/jackc/pglogrepl贡献pgoutput协议解析代码•By@diabloneo性能测试•Postgres13•Intel(R)Xeon(R)Gold5218RCPU@2.10GHz•每分钟可以发送超过50,000个简单的事务 问题预判 •逻辑事件只能包含部分的数据库操作•缺少的那些，在我们的系统里都可以通过带外的方式来解决。 •我们只会依赖消息中的id和几个时间戳字段，整个记录的内容会使用ORM从数据库重新读取。 •事件丢失 •我们一定要做好事件可能会丢失的准备，提供后备方案。 •处理阻塞导致WAL写满的情况 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冲突时，自动重试 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触发等多种原因 三个问题： 在我们的controller程序中进行管理（在controllerleader节点进行管理）。 第一次启动耗时间较长，可能会丢CDCevent Controllerfailover时会漏掉一些CDCevent l基本的做法是在controller启动时，检查app对应的slot是否存在，如果不存在则创建。 Patroni会尝试drop掉它不认识的slot 我们将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实现导致的) 丢失原因 •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 当前版本的slot数量： •Persistent:40•Temporary:6 一个内部使用的生产环境。Controllerleader运行了18天，接收的LR消息数量： •Update:966961•Insert:6276•Relation:1149 亮点： •Postgres的逻辑复制很适合在分布式系统中使用 •可以在很大程度上免去对消息队列的使用，简化系统架构•性能不错•稳定性不错 •Golang的生态对于逻辑复制的支持已经比较不错 需要注意的地方： •需要了解逻辑复制的原理，并且能够管理publication和slot•消费LR消息的时候，尽可能的不阻塞，避免WAL被drop•需要理解CDC的思想，不能将逻辑复制当成消息队列来使用•LR消息的消费者，尽可能实现幂等操作