sync_diff_inspector: support check between clusters with different time_zone setting (#436)

Author: glorv

pkg/dbutil/common.go

@@ -18,6 +18,7 @@ import (
 	"database/sql"
 	"encoding/json"
 	"fmt"
+	"net/url"
 	"os"
 	"strconv"
 	"strings"
@@ -109,7 +110,7 @@ func GetDBConfigFromEnv(schema string) DBConfig {
 }
 
 // OpenDB opens a mysql connection FD
-func OpenDB(cfg DBConfig) (*sql.DB, error) {
+func OpenDB(cfg DBConfig, vars map[string]string) (*sql.DB, error) {
 	var dbDSN string
 	if len(cfg.Snapshot) != 0 {
 		log.Info("create connection with snapshot", zap.String("snapshot", cfg.Snapshot))
@@ -118,6 +119,11 @@ func OpenDB(cfg DBConfig) (*sql.DB, error) {
 		dbDSN = fmt.Sprintf("%s:%s@tcp(%s:%d)/?charset=utf8mb4", cfg.User, cfg.Password, cfg.Host, cfg.Port)
 	}
 
+	for key, val := range vars {
+		// key='val'. add single quote for better compatibility.
+		dbDSN += fmt.Sprintf("&%s=%%27%s%%27", key, url.QueryEscape(val))
+	}
+
 	dbConn, err := sql.Open("mysql", dbDSN)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -286,6 +292,47 @@ func GetMinMaxValue(ctx context.Context, db *sql.DB, schema, table, column strin
 	return min.String, max.String, errors.Trace(rows.Err())
 }
 
+func GetTimeZoneOffset(ctx context.Context, db *sql.DB) (time.Duration, error) {
+	var timeStr string
+	err := db.QueryRowContext(ctx, "SELECT cast(TIMEDIFF(NOW(6), UTC_TIMESTAMP(6)) as time);").Scan(&timeStr)
+	if err != nil {
+		return 0, errors.Trace(err)
+	}
+	factor := time.Duration(1)
+	if timeStr[0] == '-' || timeStr[0] == '+' {
+		if timeStr[0] == '-' {
+			factor *= -1
+		}
+		timeStr = timeStr[1:]
+	}
+	t, err := time.Parse("15:04:05", timeStr)
+	if err != nil {
+		return 0, errors.Trace(err)
+	}
+
+	if t.IsZero() {
+		return 0, nil
+	}
+
+	hour, minute, second := t.Clock()
+
+	d := time.Duration(hour*3600+minute*60+second) * time.Second * factor
+	return d, nil
+}
+
+func FormatTimeZoneOffset(offset time.Duration) string {
+	prefix := "+"
+	if offset < 0 {
+		prefix = "-"
+		offset *= -1
+	}
+	hours := offset / time.Hour
+	minutes := (offset % time.Hour) / time.Minute
+
+	return fmt.Sprintf("%s%02d:%02d", prefix, hours, minutes)
+
+}
+
 func queryTables(ctx context.Context, db *sql.DB, q string) (tables []string, err error) {
 	log.Debug("query tables", zap.String("query", q))
 	rows, err := db.QueryContext(ctx, q)

pkg/dbutil/common_test.go

@@ -15,6 +15,7 @@ package dbutil
 
 import (
 	"context"
+	"time"
 
 	sqlmock "github.com/DATA-DOG/go-sqlmock"
 	"github.com/go-sql-driver/mysql"
@@ -227,3 +228,18 @@ func (s *testDBSuite) TestAnalyzeValuesFromBuckets(c *C) {
 		c.Assert(val[0], Equals, ca.expect)
 	}
 }
+
+func (s *testDBSuite) TestFormatTimeZoneOffset(c *C) {
+	cases := map[string]time.Duration{
+		"+00:00": 0,
+		"+01:00": time.Hour,
+		"-08:03": -1 * (8*time.Hour + 3*time.Minute),
+		"-12:59": -1 * (12*time.Hour + 59*time.Minute),
+		"+12:59": 12*time.Hour + 59*time.Minute,
+	}
+
+	for k, v := range cases {
+		offset := FormatTimeZoneOffset(v)
+		c.Assert(k, Equals, offset)
+	}
+}

pkg/ddl-checker/ddl_syncer.go

@@ -29,7 +29,7 @@ type DDLSyncer struct {
 
 // NewDDLSyncer create a new DDLSyncer
 func NewDDLSyncer(cfg *dbutil.DBConfig, executableChecker *ExecutableChecker) (*DDLSyncer, error) {
-	db, err := dbutil.OpenDB(*cfg)
+	db, err := dbutil.OpenDB(*cfg, nil)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}

pkg/diff/conn.go

@@ -22,8 +22,8 @@ import (
 )
 
 // CreateDB creates sql.DB used for select data
-func CreateDB(ctx context.Context, dbConfig dbutil.DBConfig, num int) (db *sql.DB, err error) {
-	db, err = dbutil.OpenDB(dbConfig)
+func CreateDB(ctx context.Context, dbConfig dbutil.DBConfig, vars map[string]string, num int) (db *sql.DB, err error) {
+	db, err = dbutil.OpenDB(dbConfig, vars)
 	if err != nil {
 		return nil, errors.Errorf("create db connections %s error %v", dbConfig.String(), err)
 	}
@@ -40,7 +40,7 @@ func CreateDB(ctx context.Context, dbConfig dbutil.DBConfig, num int) (db *sql.D
 func CreateDBForCP(ctx context.Context, dbConfig dbutil.DBConfig) (cpDB *sql.DB, err error) {
 	// set snapshot to empty, this DB used for write checkpoint data
 	dbConfig.Snapshot = ""
-	cpDB, err = dbutil.OpenDB(dbConfig)
+	cpDB, err = dbutil.OpenDB(dbConfig, nil)
 	if err != nil {
 		return nil, errors.Errorf("create db connections %+v error %v", dbConfig, err)
 	}

pkg/diff/diff_test.go

@@ -289,7 +289,7 @@ func createTableDiff(conn *sql.DB, schema string, sourceTableNames []string, tar
 }
 
 func createConn() (*sql.DB, error) {
-	return dbutil.OpenDB(dbutil.GetDBConfigFromEnv(""))
+	return dbutil.OpenDB(dbutil.GetDBConfigFromEnv(""), nil)
 }
 
 func generateData(ctx context.Context, db *sql.DB, dbCfg dbutil.DBConfig, sourceTables []string, targetTable string, hasEmptyTable bool) error {

sync_diff_inspector/diff.go

@@ -122,21 +122,30 @@ func (df *Diff) init(cfg *Config) (err error) {
 
 // CreateDBConn creates db connections for source and target.
 func (df *Diff) CreateDBConn(cfg *Config) (err error) {
-	for _, source := range cfg.SourceDBCfg {
-		source.Conn, err = diff.CreateDB(df.ctx, source.DBConfig, cfg.CheckThreadCount)
-		if err != nil {
-			return errors.Errorf("create source db %s error %v", source.DBConfig.String(), err)
-		}
-		df.sourceDBs[source.InstanceID] = source
-	}
-
 	// create connection for target.
-	cfg.TargetDBCfg.Conn, err = diff.CreateDB(df.ctx, cfg.TargetDBCfg.DBConfig, cfg.CheckThreadCount)
+	cfg.TargetDBCfg.Conn, err = diff.CreateDB(df.ctx, cfg.TargetDBCfg.DBConfig, nil, cfg.CheckThreadCount)
 	if err != nil {
 		return errors.Errorf("create target db %s error %v", cfg.TargetDBCfg.DBConfig.String(), err)
 	}
 	df.targetDB = cfg.TargetDBCfg
 
+	targetTZOffset, err := dbutil.GetTimeZoneOffset(df.ctx, cfg.TargetDBCfg.Conn)
+	if err != nil {
+		return errors.Annotatef(err, "fetch target db %s time zone offset failed", cfg.TargetDBCfg.DBConfig.String())
+	}
+	vars := map[string]string{
+		"time_zone": dbutil.FormatTimeZoneOffset(targetTZOffset),
+	}
+
+	for _, source := range cfg.SourceDBCfg {
+		// connect source db with target db time_zone
+		source.Conn, err = diff.CreateDB(df.ctx, source.DBConfig, vars, cfg.CheckThreadCount)
+		if err != nil {
+			return errors.Annotatef(err, "create source db %s failed", source.DBConfig.String())
+		}
+		df.sourceDBs[source.InstanceID] = source
+	}
+
 	df.cpDB, err = diff.CreateDBForCP(df.ctx, cfg.TargetDBCfg.DBConfig)
 	if err != nil {
 		return errors.Errorf("create checkpoint db %s error %v", cfg.TargetDBCfg.DBConfig.String(), err)

tests/sync_diff_inspector/time_zone/config.toml

@@ -0,0 +1,47 @@
+# diff Configuration.
+
+log-level = "debug"
+
+# for example, the whole data is [1...100]
+# we can split these data to [1...10], [11...20], ..., [91...100]
+# the [1...10] is a chunk, and it's chunk size is 10
+# size of the split chunk
+chunk-size = 1000
+
+# how many goroutines are created to check data
+check-thread-count = 4
+
+# sampling check percent, for example 10 means only check 10% data
+sample-percent = 100
+
+# calculate the data's checksum, and compare data by checksum.
+# set false if want to comapre the data directly
+use-checksum = true
+
+# set true will continue check from the latest checkpoint
+use-checkpoint = false
+
+# the name of the file which saves sqls used to fix different data.
+fix-sql-file = "/tmp/tidb_tools_test/sync_diff_inspector/fix.sql"
+
+# tables need to check.
+[[check-tables]]
+# schema name in target database.
+schema = "tz_test"
+
+# table list which need check in target database.
+tables = ["diff"]
+
+[[source-db]]
+host = "127.0.0.1"
+port = 4001
+user = "root"
+password = ""
+instance-id = "source-1"
+
+[target-db]
+host = "127.0.0.1"
+port = 4000
+user = "root"
+password = ""
+instance-id = "target-1"

tests/sync_diff_inspector/time_zone/run.sh

@@ -0,0 +1,30 @@
+#!/bin/sh
+
+set -e
+
+cd "$(dirname "$0")"
+OUT_DIR=/tmp/tidb_tools_test/sync_diff_inspector
+
+mysql -uroot -h 127.0.0.1 -P 4000 -e "create database if not exists tz_test"
+mysql -uroot -h 127.0.0.1 -P 4000 -e "create table tz_test.diff(id int, dt datetime, ts timestamp);"
+mysql -uroot -h 127.0.0.1 -P 4000 -e "insert into table tz_test.diff values (1, now(), now());"
+
+echo "dump data and then load to tidb"
+rm -rf $OUT_DIR/dump_tz_diff
+mydumper --host 127.0.0.1 --port 4000 --user root --outputdir $OUT_DIR/dump_tz_diff -B tz_test -T diff
+loader -h 127.0.0.1 -P 4001 -u root -d $OUT_DIR/dump_tz_diff
+
+echo "check with the same time_zone, check result should be pass"
+sync_diff_inspector --config=./config.toml > $OUT_DIR/time_zone_diff.log
+check_contains "check pass!!!" $OUT_DIR/time_zone_diff.log
+
+# check upstream and downstream time_zone
+mysql -uroot -h 127.0.0.1 -P 4000 -e "SET @@global.time_zone = '+08:00'";
+mysql -uroot -h 127.0.0.1 -P 4001 -e "SET @@global.time_zone = '+00:00'";
+echo "check with different time_zone, check result should be pass again"
+sync_diff_inspector --config=./config.toml > $OUT_DIR/time_zone_diff.log
+check_contains "check pass!!!" $OUT_DIR/time_zone_diff.log
+
+# reset time_zone
+mysql -uroot -h 127.0.0.1 -P 4000 -e "SET @@global.time_zone = 'SYSTEM'";
+mysql -uroot -h 127.0.0.1 -P 4001 -e "SET @@global.time_zone = 'SYSTEM'";