基于 TiCDC 同步的主从集群数据校验
作者：eastfisher
背景
数据库作为最核心的基础组件之一，要求它能够安全运行和保障数据安全，这是一个刚需。另外，数据库服务本身的高可用，是我们实现整个对外数据服务连续性的最重要的基石。在这些基础上，光有高可用还是不够的，我们需要考虑到机房级的、数据中心级的、站点级的灾难导致的对业务的影响[1]。 TiDB作为分布式数据库，本身已经支持了高可用，然而对于金融级应用，往往还需要多数据中心容灾，这时就需要用到 TiDB 专用的数据同步组件了。TiDB提供了 tidb-binlog 和 TiCDC 两种数据同步组件，并且在最近的几个版本更加推荐使用 TiCDC 做主从TiDB集群的数据同步。
上下游数据一致性是数据同步的基本前提，tidb-binlog 和 TiCDC 都在实现机制上尽最大可能保障数据同步一致性。然而，我们仍有必要在业务数据层面，通过数据校验的方式来确认上下游数据一致，以避免由业务而非数据库本身导致的数据不一致发生（比如向灾备的从集群写入数据导致的主从数据不一致）。TiDB 提供了 sync-diff-inspector 数据校验工具，可对上下游 TiDB 集群进行数据校验，并且支持多种校验模式，官方文档也给出了基于 tidb-binlog 的主从集群，使用 sync-diff-inspector 工具进行数据校验的方案[2]。在该方案中，tidb-binlog 的 Drainer 组件在把数据同步到 TiDB 时，保存 checkpoint 的同时也会将上下游的 TSO 对应关系保存为 ts-map。sync-diff-inspector 利用 TiDB 的 snapshot read 特性，通过设置上下游的 tidb_snapshot 变量来指定校验查询时的上下游TSO，从而“对齐”上下游数据以进行校验。
遗憾的是，文档中并没有说明 TiCDC 是否支持类似的获取 ts-map 的机制。那么，TiCDC 是否也支持类似机制呢？我们尝试阅读 TiCDC 源码来找到答案。（注：以下所有源码以 v6.1.0 版本 TiCDC 进行展示，但相关逻辑在 v5.0 各个版本也适用）
实现
使用 TiCDC 做 TiDB 主从同步的架构如下：
TiCDC 的 MySQL Sink 提供了 SyncpointStore 功能，用于将 syncpoint 保存到下游 TiDB 的库表中，接口定义如下：
// https://github.com/pingcap/tiflow/blob/v6.1.0/cdc/sink/mysql/syncpointStore.go
type SyncpointStore interface {
// CreateSynctable create a table to record the syncpoints
CreateSynctable(ctx context.Context) error
// SinkSyncpoint record the syncpoint(a map with ts) in downstream db
SinkSyncpoint(ctx context.Context, id model.ChangeFeedID, checkpointTs uint64) error
// Close closes the SyncpointSink
Close() error
}
如果保存 syncpoint 的库表不存在，则首先创建库表。这里的库名和表名都是固定值： tidb_cdc.syncpoint_v1 。
// https://github.com/pingcap/tiflow/blob/v6.1.0/cdc/sink/mysql/mysql_syncpoint_store.go#L162
func (s *mysqlSyncpointStore) CreateSynctable(ctx context.Context) error {
database := mark.SchemaName
tx, err := s.db.BeginTx(ctx, nil)
... // err handling
_, err = tx.Exec("CREATE DATABASE IF NOT EXISTS " + database)
... // err handling
_, err = tx.Exec("USE " + database)
... // err handling
_, err = tx.Exec("CREATE TABLE IF NOT EXISTS " + syncpointTableName + " (cf varchar(255),primary_ts varchar(18),secondary_ts varchar(18),PRIMARY KEY ( `cf`, `primary_ts` ) )")
... // err handling
err = tx.Commit()
return cerror.WrapError(cerror.ErrMySQLTxnError, err)
}
Sink 会根据配置的时间间隔，定时触发保存 syncpoint。首先查询下游的tidb_current_ts系统变量，得到下游当前TSO，然后将当前上游同步的TSO和下游TSO写入到之前创建的表中，用 namespace 和 changefeed id 进行区分。
// https://github.com/pingcap/tiflow/blob/v6.1.0/cdc/sink/mysql/mysql_syncpoint_store.go#L174
func (s *mysqlSyncpointStore) SinkSyncpoint(ctx context.Context, id model.ChangeFeedID, checkpointTs uint64) error {
tx, err := s.db.BeginTx(ctx, nil)
... // err handling
row := tx.QueryRow("select @@tidb_current_ts")
... // err handling
query := "insert ignore into " + mark.SchemaName + "." + syncpointTableName
+ "(cf, primary_ts, secondary_ts) VALUES (?,?,?)"
_, err = tx.Exec(query, id.Namespace+"_"+id.ID, checkpointTs, secondaryTs)
... // err handling
err = tx.Commit()
return cerror.WrapError(cerror.ErrMySQLTxnError, err)
}
写到表中的数据格式如下：
+-----------------------------------+--------------------+--------------------+
| cf | primary_ts | secondary_ts |
+-----------------------------------+--------------------+--------------------+
| default_simple-replication-task | 435782288912416770 | 435782288872570881 |
| default_simple-replication-task | 435782291533856768 | 435782291546439681 |
+-----------------------------------+--------------------+--------------------+
默认情况下，创建 TiCDC 的 changefeed 是不会开启 syncpoint 写入的，需要在创建 changefeed 时额外指定 sync-point 和 sync-interval 这两个参数：
> cdc changefeed --sink-uri='mysql://root:@127.0.0.1:4000/' --changefeed-id="simple-replication-task" --sync-point=true --sync-interval=10s create
相关的代码如下：
// https://github.com/pingcap/tiflow/blob/v6.1.0/pkg/cmd/cli/cli_changefeed_create.go#L71
func (o *changefeedCommonOptions) addFlags(cmd *cobra.Command) {
if o == nil {
return
}
// ...
cmd.PersistentFlags().BoolVar(&o.syncPointEnabled, "sync-point", false, "(Experimental) Set and Record syncpoint in replication(default off)")
cmd.PersistentFlags().DurationVar(&o.syncPointInterval, "sync-interval", 10*time.Minute, "(Experimental) Set the interval for syncpoint in replication(default 10min)")
// ...
}
使用 sync-diff-inspector 工具执行校验时，可根据 changefeed id 查询 primary_ts 最大的一条记录：
> SELECT * FROM tidb_cdc.syncpoint_v1 WHERE cf = 'default_{changefeed_id}' ORDER BY primary_ts DESC LIMIT 1;
然后按照 sync-diff-inspector 的 Datasource 配置，在上下游的 snapshot 配置项中填写相应值。
######################### Datasource config ########################
[data-sources.uptidb]
host = "172.16.0.1"
port = 4000
user = "root"
password = ""
snapshot = "409621863377928194"
[data-sources.downtidb]
host = "172.16.0.2"
port = 4000
user = "root"
password = ""
snapshot = "409621863377928345"
注意事项
仅当使用 TiCDC 将数据写入下游 TiDB 时可以使用以上方式进行数据校验，如果下游是 MySQL，使用以上方式会导致 changefeed 同步报错，直接原因是 MySQL 没有 tidb_current_ts 这个系统变量使得查询下游 TSO 失败，本质原因是 MySQL 不支持 snapshot read。
此外，使用这种方式进行上下游校验，由于是读取到的 snapshot 数据，Sink 写入下游一直失败时，使用 snapshot 数据进行校验有可能仍然是一致的，这种情况的数据异常需要通过 TiCDC 的延迟监控发现，或者在做校验时，检查上游 TiDB 的当前 TSO 和 syncpoint 最新的上游 TSO 是否偏差过大。