Support migration data from source with different time_zone by default (

pingcap#1670)

chaos/cases/conf/task-optimistic.yaml

@@ -7,7 +7,6 @@ shard-mode: optimistic
 # so ignore sharding table check, seems has no effect on sync-diff-inspector
 # see https://dev.mysql.com/doc/relnotes/mysql/8.0/en/news-8-0-19.html
 ignore-checking-items: ["schema_of_shard_tables","auto_increment_ID"] # tables generated by go-sqlsmith may have auto increment ID
-timezone: "UTC"
 
 target-database:
  host: "tidb-0.tidb"

chaos/cases/conf/task-pessimistic.yaml

@@ -7,7 +7,6 @@ shard-mode: pessimistic
 # so ignore sharding table check, seems has no effect on sync-diff-inspector
 # see https://dev.mysql.com/doc/relnotes/mysql/8.0/en/news-8-0-19.html
 ignore-checking-items: ["schema_of_shard_tables","auto_increment_ID"] # tables generated by go-sqlsmith may have auto increment ID
-timezone: "UTC"
 
 target-database:
  host: "tidb-0.tidb"

chaos/cases/conf/task-single.yaml

@@ -1,7 +1,6 @@
 ---
 name: "task_single"
 task-mode: all
-timezone: "UTC"
 
 target-database:
  host: "tidb-0.tidb"

chaos/cases/source.go

@@ -45,16 +45,16 @@ func createSources(ctx context.Context, cli pb.MasterClient, cfg *config) error
  return err
  }
 
- cfg1 := config2.NewSourceConfig()
- cfg2 := config2.NewSourceConfig()
- cfg3 := config2.NewSourceConfig()
- if err = cfg1.ParseYaml(string(s1Content)); err != nil {
+ cfg1, err := config2.ParseYaml(string(s1Content))
+ if err != nil {
  return err
  }
- if err = cfg2.ParseYaml(string(s2Content)); err != nil {
+ cfg2, err := config2.ParseYaml(string(s2Content))
+ if err != nil {
  return err
  }
- if err = cfg3.ParseYaml(string(s3Content)); err != nil {
+ cfg3, err := config2.ParseYaml(string(s3Content))
+ if err != nil {
  return err
  }
 

cmd/dm-syncer/config.go

@@ -17,7 +17,6 @@ import (
  "flag"
  "fmt"
  "os"
- "time"
 
  "github.com/BurntSushi/toml"
  "github.com/go-mysql-org/go-mysql/mysql"
@@ -27,6 +26,7 @@ import (
  router "github.com/pingcap/tidb-tools/pkg/table-router"
 
  "github.com/pingcap/dm/dm/config"
+ "github.com/pingcap/dm/pkg/log"
  "github.com/pingcap/dm/pkg/utils"
 )
 
@@ -54,6 +54,7 @@ type commonConfig struct {
  SafeMode bool
  MaxRetry int
 
+ // deprecated
  TimezoneStr string
 
  SyncerConfigFormat bool
@@ -76,13 +77,13 @@ func (c *commonConfig) newConfigFromSyncerConfig(args []string) (*config.SubTask
  EnableGTID: c.EnableGTID,
  SafeMode: c.SafeMode,
  MaxRetry: c.MaxRetry,
- TimezoneStr: c.TimezoneStr,
  }
 
  cfg.FlagSet = flag.NewFlagSet("dm-syncer", flag.ContinueOnError)
  fs := cfg.FlagSet
 
  var SyncerConfigFormat bool
+ var timezoneStr string
 
  fs.BoolVar(&cfg.printVersion, "V", false, "prints version and exit")
  fs.StringVar(&cfg.Name, "name", "", "the task name")
@@ -101,7 +102,7 @@ func (c *commonConfig) newConfigFromSyncerConfig(args []string) (*config.SubTask
  fs.BoolVar(&cfg.EnableGTID, "enable-gtid", false, "enable gtid mode")
  fs.BoolVar(&cfg.SafeMode, "safe-mode", false, "enable safe mode to make syncer reentrant")
  fs.IntVar(&cfg.MaxRetry, "max-retry", 100, "maxinum retry when network interruption")
- fs.StringVar(&cfg.TimezoneStr, "timezone", "", "target database timezone location string")
+ fs.StringVar(&timezoneStr, "timezone", "", "target database timezone location string")
  fs.BoolVar(&SyncerConfigFormat, "syncer-config-format", false, "read syncer config format")
 
  if err := fs.Parse(args); err != nil {
@@ -119,6 +120,10 @@ func (c *commonConfig) newConfigFromSyncerConfig(args []string) (*config.SubTask
  return nil, errors.Trace(err)
  }
 
+ if timezoneStr != "" {
+ log.L().Warn("'--timezone' is deprecated, needn't set it anymore.")
+ }
+
  return cfg.convertToNewFormat()
 }
 
@@ -146,6 +151,7 @@ func (c *commonConfig) newSubTaskConfig(args []string) (*config.SubTaskConfig, e
  var syncerConfigFormat bool
  var printVersion bool
  var serverID uint
+ var timezoneStr string
 
  fs.BoolVar(&printVersion, "V", false, "prints version and exit")
  fs.StringVar(&cfg.Name, "name", "", "the task name")
@@ -163,7 +169,7 @@ func (c *commonConfig) newSubTaskConfig(args []string) (*config.SubTaskConfig, e
  fs.BoolVar(&cfg.EnableGTID, "enable-gtid", false, "enable gtid mode")
  fs.BoolVar(&cfg.SafeMode, "safe-mode", false, "enable safe mode to make syncer reentrant")
  fs.IntVar(&cfg.MaxRetry, "max-retry", 100, "maxinum retry when network interruption")
- fs.StringVar(&cfg.Timezone, "timezone", "", "target database timezone location string")
+ fs.StringVar(&timezoneStr, "timezone", "", "target database timezone location string")
  fs.StringVar(&cfg.Name, "cp-table-prefix", "dm-syncer", "the prefix of the checkpoint table name")
  fs.BoolVar(&syncerConfigFormat, "syncer-config-format", false, "read syncer config format")
 
@@ -172,6 +178,9 @@ func (c *commonConfig) newSubTaskConfig(args []string) (*config.SubTaskConfig, e
  if err := cfg.Parse(args, false); err != nil {
  return nil, errors.Trace(err)
  }
+ if timezoneStr != "" {
+ log.L().Warn("'--timezone' is deprecated, needn't set it anymore.")
+ }
 
  if serverID != 101 {
  cfg.ServerID = uint32(serverID)
@@ -202,7 +211,6 @@ func newCommonConfig() *commonConfig {
  fs.BoolVar(&cfg.EnableGTID, "enable-gtid", false, "enable gtid mode")
  fs.BoolVar(&cfg.SafeMode, "safe-mode", false, "enable safe mode to make syncer reentrant")
  fs.IntVar(&cfg.MaxRetry, "max-retry", 100, "maxinum retry when network interruption")
- fs.StringVar(&cfg.TimezoneStr, "timezone", "", "target database timezone location string")
  fs.BoolVar(&cfg.SyncerConfigFormat, "syncer-config-format", false, "read syncer config format")
 
  return cfg
@@ -265,8 +273,7 @@ type syncerConfig struct {
  // MaxDMLConnectionTimeout string `toml:"execute-dml-timeout" json:"execute-dml-timeout"`
  // ExecutionQueueLength int `toml:"execute-queue-length" json:"execute-queue-length"`
 
- TimezoneStr string `toml:"timezone" json:"timezone"`
- Timezone *time.Location `json:"-"`
+ TimezoneStr string `toml:"timezone" json:"timezone"`
 
  printVersion bool
 }
@@ -343,7 +350,6 @@ func (oc *syncerConfig) convertToNewFormat() (*config.SubTaskConfig, error) {
  },
 
  ConfigFile: oc.ConfigFile,
- Timezone: oc.TimezoneStr,
  From: oc.From,
  To: oc.To,
  }

dm/config/source_config.go

@@ -82,6 +82,13 @@ type SourceConfig struct {
 
 // NewSourceConfig creates a new base config for upstream MySQL/MariaDB source.
 func NewSourceConfig() *SourceConfig {
+ c := newSourceConfig()
+ c.adjust()
+ return c
+}
+
+// NewSourceConfig creates a new base config without adjust.
+func newSourceConfig() *SourceConfig {
  c := &SourceConfig{
  Purge: PurgeConfig{
  Interval: 60 * 60,
@@ -94,7 +101,6 @@ func NewSourceConfig() *SourceConfig {
  BackoffMax: Duration{DefaultBackoffMax},
  },
  }
- c.adjust()
  return c
 }
 
@@ -140,12 +146,13 @@ func (c *SourceConfig) Parse(content string) error {
 }
 
 // ParseYaml parses flag definitions from the argument list, content should be yaml format.
-func (c *SourceConfig) ParseYaml(content string) error {
+func ParseYaml(content string) (*SourceConfig, error) {
+ c := newSourceConfig()
  if err := yaml.UnmarshalStrict([]byte(content), c); err != nil {
- return terror.ErrConfigYamlTransform.Delegate(err, "decode source config")
+ return nil, terror.ErrConfigYamlTransform.Delegate(err, "decode source config")
  }
  c.adjust()
- return nil
+ return c, nil
 }
 
 // EncodeToml encodes config.
@@ -311,16 +318,20 @@ func (c *SourceConfig) AdjustServerID(ctx context.Context, db *sql.DB) error {
 }
 
 // LoadFromFile loads config from file.
-func (c *SourceConfig) LoadFromFile(path string) error {
+func LoadFromFile(path string) (*SourceConfig, error) {
+ c := newSourceConfig()
  content, err := ioutil.ReadFile(path)
  if err != nil {
- return terror.ErrConfigReadCfgFromFile.Delegate(err, path)
+ return nil, terror.ErrConfigReadCfgFromFile.Delegate(err, path)
  }
- if err := yaml.UnmarshalStrict(content, c); err != nil {
- return terror.ErrConfigYamlTransform.Delegate(err, "decode source config")
+ if err = yaml.UnmarshalStrict(content, c); err != nil {
+ return nil, terror.ErrConfigYamlTransform.Delegate(err, "decode source config")
  }
  c.adjust()
- return c.Verify()
+ if err = c.Verify(); err != nil {
+ return nil, err
+ }
+ return c, nil
 }
 
 func (c *SourceConfig) check(metaData *toml.MetaData, err error) error {

dm/config/source_config_test.go

@@ -32,9 +32,8 @@ import (
 const sourceSampleFile = "../worker/source.yaml"
 
 func (t *testConfig) TestConfig(c *C) {
- cfg := NewSourceConfig()
-
- c.Assert(cfg.LoadFromFile(sourceSampleFile), IsNil)
+ cfg, err := LoadFromFile(sourceSampleFile)
+ c.Assert(err, IsNil)
  cfg.RelayDir = "./xx"
  c.Assert(cfg.RelayDir, Equals, "./xx")
  c.Assert(cfg.ServerID, Equals, uint32(101))
@@ -63,18 +62,24 @@ func (t *testConfig) TestConfig(c *C) {
  // test update config file and reload
  c.Assert(cfg.Parse(tomlStr), IsNil)
  c.Assert(cfg.ServerID, Equals, uint32(100))
- c.Assert(cfg.ParseYaml(yamlStr), IsNil)
- c.Assert(cfg.ServerID, Equals, uint32(100))
- c.Assert(cfg.Parse(originCfgStr), IsNil)
- c.Assert(cfg.ServerID, Equals, uint32(101))
- c.Assert(cfg.ParseYaml(originCfgYamlStr), IsNil)
- c.Assert(cfg.ServerID, Equals, uint32(101))
+ cfg1, err := ParseYaml(yamlStr)
+ c.Assert(err, IsNil)
+ c.Assert(cfg1.ServerID, Equals, uint32(100))
+ cfg.Filters = []*bf.BinlogEventRule{}
+ cfg.Tracer = map[string]interface{}{}
+
+ var cfg2 SourceConfig
+ c.Assert(cfg2.Parse(originCfgStr), IsNil)
+ c.Assert(cfg2.ServerID, Equals, uint32(101))
+
+ cfg3, err := ParseYaml(originCfgYamlStr)
+ c.Assert(err, IsNil)
+ c.Assert(cfg3.ServerID, Equals, uint32(101))
 
  // test decrypt password
  clone1.From.Password = "1234"
- clone1.ServerID = 101
  // fix empty map after marshal/unmarshal becomes nil
- clone1.From.Session = map[string]string{}
+ clone1.From.Session = cfg.From.Session
  clone1.Tracer = map[string]interface{}{}
  clone1.Filters = []*bf.BinlogEventRule{}
  clone2 := cfg.DecryptPassword()
@@ -96,16 +101,18 @@ func (t *testConfig) TestConfig(c *C) {
  c.Assert(err, IsNil)
  c.Assert(clone4toml, Matches, "(.|\n)*backoff-rollback = \"5m(.|\n)*")
  c.Assert(clone4toml, Matches, "(.|\n)*backoff-max = \"5m(.|\n)*")
+
  var clone5 SourceConfig
  c.Assert(clone5.Parse(clone4toml), IsNil)
  c.Assert(clone5, DeepEquals, *clone4)
  clone4yaml, err := clone4.Yaml()
  c.Assert(err, IsNil)
  c.Assert(clone4yaml, Matches, "(.|\n)*backoff-rollback: 5m(.|\n)*")
  c.Assert(clone4yaml, Matches, "(.|\n)*backoff-max: 5m(.|\n)*")
- var clone6 SourceConfig
- c.Assert(clone6.ParseYaml(clone4yaml), IsNil)
- c.Assert(clone6, DeepEquals, *clone4)
+
+ clone6, err := ParseYaml(clone4yaml)
+ c.Assert(err, IsNil)
+ c.Assert(clone6, DeepEquals, clone4)
 
  // test invalid config
  dir2 := c.MkDir()
@@ -116,15 +123,15 @@ aaa: xxx
 `)
  err = ioutil.WriteFile(configFile, configContent, 0o644)
  c.Assert(err, IsNil)
- err = cfg.LoadFromFile(configFile)
+ _, err = LoadFromFile(configFile)
  c.Assert(err, NotNil)
  c.Assert(err, ErrorMatches, "(.|\n)*field aaa not found in type config.SourceConfig(.|\n)*")
 }
 
 func (t *testConfig) TestConfigVerify(c *C) {
  newConfig := func() *SourceConfig {
- cfg := NewSourceConfig()
- c.Assert(cfg.LoadFromFile(sourceSampleFile), IsNil)
+ cfg, err := LoadFromFile(sourceSampleFile)
+ c.Assert(err, IsNil)
  cfg.RelayDir = "./xx"
  return cfg
  }
@@ -247,12 +254,12 @@ func subtestFlavor(c *C, cfg *SourceConfig, sqlInfo, expectedFlavor, expectedErr
 }
 
 func (t *testConfig) TestAdjustFlavor(c *C) {
- cfg := NewSourceConfig()
- c.Assert(cfg.LoadFromFile(sourceSampleFile), IsNil)
+ cfg, err := LoadFromFile(sourceSampleFile)
+ c.Assert(err, IsNil)
  cfg.RelayDir = "./xx"
 
  cfg.Flavor = "mariadb"
- err := cfg.AdjustFlavor(context.Background(), nil)
+ err = cfg.AdjustFlavor(context.Background(), nil)
  c.Assert(err, IsNil)
  c.Assert(cfg.Flavor, Equals, mysql.MariaDBFlavor)
  cfg.Flavor = "MongoDB"
@@ -270,8 +277,8 @@ func (t *testConfig) TestAdjustServerID(c *C) {
  }()
  getAllServerIDFunc = getMockServerIDs
 
- cfg := NewSourceConfig()
- c.Assert(cfg.LoadFromFile(sourceSampleFile), IsNil)
+ cfg, err := LoadFromFile(sourceSampleFile)
+ c.Assert(err, IsNil)
  cfg.RelayDir = "./xx"
 
  c.Assert(cfg.AdjustServerID(context.Background(), nil), IsNil)

dm/config/subtask.go

@@ -19,18 +19,17 @@ import (
  "flag"
  "fmt"
  "strings"
- "time"
-
- "github.com/pingcap/dm/pkg/log"
- "github.com/pingcap/dm/pkg/terror"
- "github.com/pingcap/dm/pkg/utils"
 
  "github.com/BurntSushi/toml"
  bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
  "github.com/pingcap/tidb-tools/pkg/column-mapping"
  "github.com/pingcap/tidb-tools/pkg/filter"
  router "github.com/pingcap/tidb-tools/pkg/table-router"
  "go.uber.org/zap"
+
+ "github.com/pingcap/dm/pkg/log"
+ "github.com/pingcap/dm/pkg/terror"
+ "github.com/pingcap/dm/pkg/utils"
 )
 
 // task modes.
@@ -118,6 +117,8 @@ func (db *DBConfig) Decode(data string) error {
 
 // Adjust adjusts the config.
 func (db *DBConfig) Adjust() {
+ // force set session time zone to UTC here.
+ AdjustTargetDBTimeZone(db)
 }
 
 // SubTaskConfig is the configuration for SubTask.
@@ -150,8 +151,8 @@ type SubTaskConfig struct {
  HeartbeatReportInterval int `toml:"heartbeat-report-interval" json:"heartbeat-report-interval"`
  EnableHeartbeat bool `toml:"enable-heartbeat" json:"enable-heartbeat"`
  Meta *Meta `toml:"meta" json:"meta"`
- Timezone string `toml:"timezone" josn:"timezone"`
-
+ // deprecated
+ Timezone string `toml:"timezone" json:"timezone"`
  // RelayDir get value from dm-worker config
  RelayDir string `toml:"relay-dir" json:"relay-dir"`
 
@@ -274,10 +275,8 @@ func (c *SubTaskConfig) Adjust(verifyDecryptPassword bool) error {
  }
 
  if c.Timezone != "" {
- _, err := time.LoadLocation(c.Timezone)
- if err != nil {
- return terror.ErrConfigInvalidTimezone.Delegate(err, c.Timezone)
- }
+ log.L().Warn("'timezone' is deprecated, please remove this field.")
+ c.Timezone = ""
  }
 
  dirSuffix := "." + c.Name