sync_diff_inspector: add auto-position of tidb_snapshot (#664)

close #663

Author: morgo

sync_diff_inspector/config/config.go

@@ -21,6 +21,7 @@ import (
 	"net/url"
 	"os"
 	"path/filepath"
+	"strings"
 	"syscall"
 	"time"
 
@@ -97,6 +98,18 @@ type DataSource struct {
 	// SourceType string `toml:"source-type" json:"source-type"`
 }
 
+// IsAutoSnapshot returns true if the tidb_snapshot is expected to automatically
+// be set from the syncpoint from the target TiDB instance.
+func (d *DataSource) IsAutoSnapshot() bool {
+	return strings.EqualFold(d.Snapshot, "auto")
+}
+
+// SetSnapshot changes the snapshot in configuration. This is typically
+// used with the auto-snapshot feature.
+func (d *DataSource) SetSnapshot(newSnapshot string) {
+	d.Snapshot = newSnapshot
+}
+
 func (d *DataSource) ToDBConfig() *dbutil.DBConfig {
 	return &dbutil.DBConfig{
 		Host:     d.Host,

sync_diff_inspector/config/config.toml

@@ -30,6 +30,8 @@ check-struct-only = false
     # remove comment if use tidb's snapshot data
     # snapshot = "2016-10-08 16:45:26"
     # snapshot = "386902609362944000"
+    # When using TiCDC syncpoint source and target can be set to auto
+    # snapshot = "auto"
 
 ######################### Task config #########################
 # Required

sync_diff_inspector/source/source.go

@@ -45,8 +45,9 @@ const (
 const UnifiedTimeZone string = "+0:00"
 
 const (
-	ShieldDBName    = "_no__exists__db_"
-	ShieldTableName = "_no__exists__table_"
+	ShieldDBName      = "_no__exists__db_"
+	ShieldTableName   = "_no__exists__table_"
+	GetSyncPointQuery = "SELECT primary_ts, secondary_ts FROM tidb_cdc.syncpoint_v1 ORDER BY primary_ts DESC LIMIT 1"
 )
 
 type ChecksumInfo struct {
@@ -226,11 +227,43 @@ func buildSourceFromCfg(ctx context.Context, tableDiffs []*common.TableDiff, che
 	return NewMySQLSources(ctx, tableDiffs, dbs, checkThreadCount, f)
 }
 
+func getAutoSnapshotPosition(dbConfig *dbutil.DBConfig, vars map[string]string) (string, string, error) {
+	tmpConn, err := dbutil.OpenDB(*dbConfig, vars)
+	if err != nil {
+		return "", "", errors.Annotatef(err, "connecting to auto-position tidb_snapshot failed")
+	}
+	defer tmpConn.Close()
+	var primaryTs, secondaryTs string
+	err = tmpConn.QueryRow(GetSyncPointQuery).Scan(&primaryTs, &secondaryTs)
+	if err != nil {
+		return "", "", errors.Annotatef(err, "fetching auto-position tidb_snapshot failed")
+	}
+	return primaryTs, secondaryTs, nil
+}
+
 func initDBConn(ctx context.Context, cfg *config.Config) error {
 	// Unified time zone
 	vars := map[string]string{
 		"time_zone": UnifiedTimeZone,
 	}
+	// Fill in tidb_snapshot if it is set to AUTO
+	// This is only supported when set to auto on both target/source.
+	if cfg.Task.TargetInstance.IsAutoSnapshot() {
+		if len(cfg.Task.SourceInstances) > 1 {
+			return errors.Errorf("'auto' snapshot only supports one tidb source")
+		}
+		if !cfg.Task.SourceInstances[0].IsAutoSnapshot() {
+			return errors.Errorf("'auto' snapshot should be set on both target and source")
+		}
+		tmpDBConfig := cfg.Task.TargetInstance.ToDBConfig()
+		tmpDBConfig.Snapshot = ""
+		primaryTs, secondaryTs, err := getAutoSnapshotPosition(tmpDBConfig, vars)
+		if err != nil {
+			return err
+		}
+		cfg.Task.TargetInstance.SetSnapshot(secondaryTs)
+		cfg.Task.SourceInstances[0].SetSnapshot(primaryTs)
+	}
 	// we had 3 producers and `cfg.CheckThreadCount` consumer to use db connections.
 	// so the connection count need to be cfg.CheckThreadCount + 3.
 	targetConn, err := common.CreateDB(ctx, cfg.Task.TargetInstance.ToDBConfig(), vars, cfg.CheckThreadCount+3)
@@ -241,6 +274,11 @@ func initDBConn(ctx context.Context, cfg *config.Config) error {
 	cfg.Task.TargetInstance.Conn = targetConn
 
 	for _, source := range cfg.Task.SourceInstances {
+		// If it is still set to AUTO it means it was not set on the target.
+		// We require it to be set to AUTO on both.
+		if source.IsAutoSnapshot() {
+			return errors.Errorf("'auto' snapshot should be set on both target and source")
+		}
 		// connect source db with target db time_zone
 		conn, err := common.CreateDB(ctx, source.ToDBConfig(), vars, cfg.CheckThreadCount+1)
 		if err != nil {