TiCDC核心原理解析

作者： Jellybean

架构简介

基本架构

TiCDC 集群由多个对等节点组成，是一种分布式无状态的架构设计。当 TiDB 集群内部有数据变更的时候，就会产生 KV change log。

KV change log 是 TiKV 提供的隐藏大部分内部实现细节的的 row changed event，TiCDC 会实时从 TiKV 拉取这些 Event 完成扫描和拼装，再同步到下游节点。同步任务将会按照一定的调度规则被划分给一个或者多个 Capture 处理。

为了方便深入了解 Capture 的执行过程，需要理解一些概念：

• Owner

• 可以理解为是 TiCDC 集群的 leader 节点，它负责响应用户的请求、调度集群和同步 DDL 等任务。

• Processor

• Capture 内部的逻辑线程，一个 Capture 节点中可以运行多个 Processor。

• Table Pipeline

• Processor 内部的数据同步管道，每个 TablePipeline 负责处理一张表，表的数据会在这个管道中处理和流转，最后被发送到下游。

• Changefeed

• 是由用户启动同步任务，一个同步任务中可能包含多张表，这些表会被 Owner 划分为多个子任务分配到不同的 Capture 进行处理。每个 Processor 负责处理 ChangeFeed 的一个子任务。

执行流程

内部详细的执行过程如下图：

TiCDC 的 DML 同步流和 DDL 同步流是分开的。从上面的架构图中可以看到， DML 的同步是由 Processor 进行的，数据流从上游的 TiKV 流入经过 Processor 内的 TablePipeline ，最后被同步到下游。而 DDL 同步则是由 Owner 进行的，OwnerDDLPuller 拉取上游发生的 DDL 事件，然后在内部经过一系列的处理之后，通过 DDLSink 同步到下游。

• OwnerSchemaStorage：由 Owner 持有，维护了当前所有表最新的 schema 信息，这些表的 schema 信息主要会被 scheduler 所使用，用来感知同步任务的表数量的变化；此外，还会被 owner 用来解析 ddlPuller 拉取到的 DDL 事件。
• ProcessorSchemaStorage：由 Processor 持有，维护了当前所有表的多个版本的 schema 信息，这些信息会被 Mounter 用来解析行变更信息。
• BarrierTs：由 Owner 向 Processor 发送的控制信息，它会让 Processor 把同步进度阻塞到 BarrierTs 所指定的值。TiCDC 内部有几种不同类型的 BarrierTs，为了简化叙述，本文中提到的 BarrierTs 仅表示 DDL 事件产生的 DDLBarrierTs。
• OwnerDDLPuller：由 Owner 持有，负责拉取和过滤上游 TiDB 集群的 DDL 事件，并把它们缓存在一个队列中，等待 Owner 处理；此外，它还会维护一个 ResolvedTs，该值为上游 TiKV 发送过来的最新的 ResolvedTs，在没有 DDL 事件到来的时候，Owner 将会使用它来推进 DDLBirrierTs。
• ProcessorDDLPuller：由 Processor 持有，负责拉取和过滤上游 TiDB 集群的 DDL 事件，然后把它们发送给 Processor 去更新 ProcessorSchemaStorage。
• DDLSink：由 Owner 持有，负责执行 DDL 到下游。

TiCDC与ETCD

我们还需要知道，TiCDC 集群的元数据都会被存储到 PD 内置的 Etcd 中并定期更新。当一个 TiCDC 集群被部署起来时，每个 Capture 都会向 Etcd 注册自己的信息，这样 Capture 就能够发现彼此的存在，从而完成 Owner 的选举。

竞选到 Owner 角色的 Capture 会作为集群的管理者，也负责监听和响应来自用户的请求。

这里有一个风险点，因为 etcd 的多版本并发控制 (MVCC) 以及 PD 默认的 compaction 间隔是 1 小时，TiCDC 占用的 PD 存储空间与 1 小时内元数据的版本数量成正比，在 v4.0.5、v4.0.6、v4.0.7 三个版本中 TiCDC 存在元数据写入频繁的问题，如果 1 小时内有 1000 张表创建或调度，就会用尽 etcd 的存储空间，导致集群不可用。

特别说明：etcd 存储空间耗尽会出现 etcdserver: mvcc: database space exceeded 错误，需及时清理 etcd 存储空间，否则集群不可用。参考 etcd maintenance space-quota。如果TiCDC 版本为 v4.0.5、v4.0.6 或 v4.0.7，会有比较大的使用风险，强烈建议升级到 v4.0.9 及以后版本。

同步状态

Changefeed 是 TiCDC 中的单个同步任务，负责将一个表或者多个表的变更数据输出到一个指定的下游。TiCDC 集群可以运行和管理多个 Changefeed。

在 TiCDC 运行过程中，同步任务可能会运行出错、手动暂停、恢复，或达到指定的 TargetTs，这些行为都可以导致同步任务状态发生变化。同步任务的状态有：

• Normal：同步任务正常进行，checkpoint-ts 正常推进。
• Stopped：同步任务停止，由于用户手动暂停 (pause) 任务。

• 处于这个状态的 changefeed 会阻挡 GC 推进。

• Warning：同步任务报错，由于某些可恢复的错误导致同步无法继续进行。

• 处于这个状态的 changefeed 会阻挡集群 GC 推进。
• 此时同步任务会不断尝试继续推进，直到状态转为 Normal。超过最大重试时间 30 分钟，changefeed 会进入 failed 状态。

• Finished：同步任务完成，同步任务进度已经达到预设的目标时间戳 TargetTs。
• Failed：同步任务失败。

• 由于发生了某些不可恢复的错误，导致同步无法继续进行，并且无法自动恢复。
• 为了让用户有足够的时间处理故障，处于这个状态的 changefeed 会阻塞 GC 推进，阻塞时长为 gc-ttl 所设置的值，默认 24 小时。
• 如果 changefeed 遭遇错误码为 ErrGCTTLExceeded, ErrSnapshotLostByGC 或者 ErrStartTsBeforeGC 类型的错误，任务直接失败，不阻塞集群 GC 推进。

以上状态流转图中的编号说明如下：

• ① 执行 pause 暂停同步任务。
• ② 执行 resume 恢复同步任务。
• ③ 同步任务运行过程中发生可恢复的错误，自动重试。
• ④ 同步任务自动重试成功，checkpoint-ts 已经继续推进。
• ⑤ 同步任务自动重试超过 30 分钟，重试失败，进入 failed 状态。此时 changefeed 会继续阻塞上游 GC，阻塞时长为 gc-ttl 所配置的时长。
• ⑥ 同步任务遇到不可重试错误，直接进入 failed 状态。此时 changefeed 会继续阻塞上游 GC，阻塞时长为 gc-ttl 所配置的时长。
• ⑦ 同步任务的同步进度到达 target-ts 设置的值，完成同步。
• ⑧ 同步任务停滞时间超过 gc-ttl 所指定的时长，因集群会继续推进 GC 而让任务直接失败。

Table Pipeline

每个同步任务，负责同步一张或者多张表，TiCDC 的同步对于单表来说是单线程的，对于多表之间的同步是并行的。对于每个表的处理过程，会放在一个 Table Pipeline 流程内执行完成。TablePipeline 就是一个表数据流动和处理的管道。

TiCDC 的 Processor 接收到一个同步子任务之后，会为每一张表自动创建出一个 TablePipeline，它主要由 Puller、Sorter、Mounter 和 Sink 构成。各个模块之间是串行的关系，组合在一起完成从上游拉取、排序、加载和同步数据到下游的过程。

• Puller： 负责拉取对应表在上游的变更数据，它隐藏了内部大量的实现细节，包括与 TiKV CDC 模块建立 gRPC 连接和反解码数据流等。Puller 从 KV-Client 接收数据并写入到 Sorter 中，并持续推进表级别的 Resovled Ts，标识该表当前接收数据的进度。
• Sorter： 负责对 Puller 输出的乱序数据进行排序，并且会把 Sink 来不及消费的数据进行落盘，起到一个蓄水池的作用。此时输出的数据是从 TiKV 中扫描出的 key-value，是 bytes 数据。TiCDC 使用 Pebble 作为默认的排序引擎，是基于 LSM 树的 golang 开源实现 。在 TiCDC 中，通过一些手段极大缓解了 LSM 树读写放大的情况。
• Mounter：根据对应的表的 Schema 信息将行转化为按照表结构组织的数据。Mounter 进行的是一项 CPU 密集型工作，当一个表中所包含的字段较多时，Mounter 会消耗大量的计算资源。
• Sink：将 Mounter 处理过后的数据进行编解码，转化为 SQL 语句或者 Kafka 消息发送到对应下游。

• TableSink 作为一种 Table 级别的管理单位，缓存着要下发到 ProcessorSink 的数据，它的主要作用是方便 TiCDC 按照表为单位管理资源和进行调度
• ProcessorSink 作为真实要与数据库或者 Kafka 建立连接的 Sink 负责 SQL/Kafka 消息的转换和同步

多表同步任务

假设我们创建一个 Changefeed 任务，要同步 test1.tab1、test1.tab2、test3.tab3 和 test4.tab4 四张表。TiCDC 接收到这个命令之后的处理流程如下：

1. TiCDC 将这个任务发送给 Owner Capture 进程。
2. Owner Capture 进程将这个任务的相关定义信息保存在 PD 的 etcd 中。
3. Owner Capture 将这个任务拆分成若干个 Task，并通知其他 Capture 进程各个 Task 需要完成的任务。
4. 各个 Capture 进程开始从对应的 TiKV 节点拉取信息，进行处理后完成同步。

上图创建了一个同步任务，这个 Changefeed 包含 4 张表，其被拆分成了 3 个任务，以表为单位同步数据为每张表创建一个 Table Pipeline。均匀的分发到了 TiCDC 集群的 3 个 Capture 节点上，在 TiCDC 对这些数据进行了处理之后，数据同步到了下游的系统。

总结

• TiCDC 内部通过 Puller、Sorter、Mounter 和 Sink 这几个环节串行操作，实现变更数据的拉取、排序、处理和写入下游。
• 对于单表来说，是一个独立的 Pipeline 处理，可以简单理解为内部是单并发处理；对于多表，是多并发处理。

• 单个表的同步任务是单并发处理，多个表的同步是多并发处理。

• TiCDC 会在 PD 注册自己的 GC Safe Point，最多可以阻塞集群 24 小时 GC不推进，即延迟最大是一天。超过一条不处理，PD 会强制推进 GC ，此时 TiCDC 的同步任务会立即失败。

• 所以，对于 TiCDC 同步的延迟情况，尽量早介入处理，否则会导致超时而出现任务失败的现象。

• TiCDC 内部会有 Owner 角色进行统一的调度管理，类似 TiKV 的 Leader 角色，保证任务处理的高可用。