dm-worker: refine db operation (pingcap#236)

* syncer db conn

* loader db refine

* remove useless code

* add db retry policy

* remove syncer max retry config

* add normal retry policy

* update retry strategy

* add context for retry

* update to terror

* add invalid strategy

* update retry invalid connection

* update retry strategy

* add retry strategy comment

* add reset connection logic

* use same baseConn in Conns

* udpate syncer dbs

* unify query in log

* share same *DB by all conns

* split load execute sql with different retryFn

* unify retry logic to pkg

* put retry error together

* unify syncer conn helper function

* update retry logic and some tests error formats

* add baseConn test

* make baseconn a pkg

* abstract retry strategy

* add querySQL args

* add whiteList error retry logic

* unify retry error in syncer/loader

* keep max-retry in syncerConifg

Author: 3pointer

dm/config/subtask.go

@@ -219,10 +219,6 @@ func (c *SubTaskConfig) Adjust() error {
 		c.MetaSchema = defaultMetaSchema
 	}
 
-	if c.MaxRetry == 0 {
-		c.MaxRetry = 1
-	}
-
 	if !c.DisableHeartbeat {
 		c.EnableHeartbeat = true
 	}

dm/config/task.go

@@ -55,7 +55,6 @@ var (
 	// SyncerConfig
 	defaultWorkerCount = 16
 	defaultBatch       = 100
-	defaultMaxRetry    = 100
 )
 
 // Meta represents binlog's meta pos
@@ -184,7 +183,8 @@ type SyncerConfig struct {
 	MetaFile    string `yaml:"meta-file" toml:"meta-file" json:"meta-file"` // meta filename, used only when load SubConfig directly
 	WorkerCount int    `yaml:"worker-count" toml:"worker-count" json:"worker-count"`
 	Batch       int    `yaml:"batch" toml:"batch" json:"batch"`
-	MaxRetry    int    `yaml:"max-retry" toml:"max-retry" json:"max-retry"`
+	// deprecated
+	MaxRetry int `yaml:"max-retry" toml:"max-retry" json:"max-retry"`
 
 	// refine following configs to top level configs?
 	AutoFixGTID      bool `yaml:"auto-fix-gtid" toml:"auto-fix-gtid" json:"auto-fix-gtid"`
@@ -198,7 +198,6 @@ func defaultSyncerConfig() SyncerConfig {
 	return SyncerConfig{
 		WorkerCount: defaultWorkerCount,
 		Batch:       defaultBatch,
-		MaxRetry:    defaultMaxRetry,
 	}
 }
 

dm/dm-ansible/conf/task.yaml.example

@@ -74,7 +74,6 @@ mysql-instances:             # one or more source database, config more source d
     syncer:
       worker-count: 16
       batch: 100
-      max-retry: 100
 
 # other common configs shared by all instances
 
@@ -150,4 +149,3 @@ syncers:                     # syncer process unit specific configs, mysql insta
   global:
     worker-count: 16
     batch: 100
-    max-retry: 100

dm/master/server_test.go

@@ -118,7 +118,6 @@ syncers:
   global:
     worker-count: 16
     batch: 100
-    max-retry: 100
 `
 
 var (

dm/master/task.yaml

@@ -74,7 +74,6 @@ mysql-instances:             # one or more source database, config more source d
     syncer:
       worker-count: 16
       batch: 100
-      max-retry: 100
 
 # other common configs shared by all instances
 
@@ -150,4 +149,3 @@ syncers:                     # syncer process unit specific configs, mysql insta
   global:
     worker-count: 16
     batch: 100
-    max-retry: 100

dm/worker/subtask.toml

@@ -58,8 +58,6 @@ meta-file = "./syncer.subTaskA.meta"
 worker-count = 16
 batch = 1000
 
-# max-retry is used for retry when network interruption.
-max-retry = 100
 
 # target database timezone, all timestamp event in binlog will translate to format time based on this timezone, default use local timezone
 # timezone = "Asia/Shanghai"

loader/checkpoint.go

@@ -112,7 +112,7 @@ func (cp *RemoteCheckPoint) prepare() error {
 
 func (cp *RemoteCheckPoint) createSchema() error {
 	sql2 := fmt.Sprintf("CREATE SCHEMA IF NOT EXISTS `%s`", cp.schema)
-	err := cp.conn.executeSQL(cp.tctx, []string{sql2}, true)
+	err := cp.conn.executeSQL(cp.tctx, []string{sql2})
 	return terror.WithScope(err, terror.ScopeDownstream)
 }
 
@@ -131,7 +131,7 @@ func (cp *RemoteCheckPoint) createTable() error {
 	);
 `
 	sql2 := fmt.Sprintf(createTable, tableName)
-	err := cp.conn.executeSQL(cp.tctx, []string{sql2}, true)
+	err := cp.conn.executeSQL(cp.tctx, []string{sql2})
 	return terror.WithScope(err, terror.ScopeDownstream)
 }
 
@@ -143,7 +143,7 @@ func (cp *RemoteCheckPoint) Load() error {
 	}()
 
 	query := fmt.Sprintf("SELECT `filename`,`cp_schema`,`cp_table`,`offset`,`end_pos` from `%s`.`%s` where `id`=?", cp.schema, cp.table)
-	rows, err := cp.conn.querySQL(cp.tctx, query, queryRetryCount, cp.id)
+	rows, err := cp.conn.querySQL(cp.tctx, query, cp.id)
 	if err != nil {
 		return terror.WithScope(err, terror.ScopeDownstream)
 	}
@@ -279,7 +279,8 @@ func (cp *RemoteCheckPoint) Init(filename string, endPos int64) error {
 		zap.String("table", fields[1]),
 		zap.Int64("offset", 0),
 		zap.Int64("end position", endPos))
-	err := cp.conn.executeCheckpointSQL(cp.tctx, sql2, maxRetryCount, cp.id, filename, fields[0], fields[1], 0, endPos)
+	args := []interface{}{cp.id, filename, fields[0], fields[1], 0, endPos}
+	err := cp.conn.executeSQL(cp.tctx, []string{sql2}, args)
 	if err != nil {
 		if isErrDupEntry(err) {
 			cp.tctx.L().Info("checkpoint record already exists, skip it.", zap.String("id", cp.id), zap.String("filename", filename))
@@ -305,14 +306,14 @@ func (cp *RemoteCheckPoint) GenSQL(filename string, offset int64) string {
 // Clear implements CheckPoint.Clear
 func (cp *RemoteCheckPoint) Clear() error {
 	sql2 := fmt.Sprintf("DELETE FROM `%s`.`%s` WHERE `id` = '%s'", cp.schema, cp.table, cp.id)
-	err := cp.conn.executeSQL(cp.tctx, []string{sql2}, true)
+	err := cp.conn.executeSQL(cp.tctx, []string{sql2})
 	return terror.WithScope(err, terror.ScopeDownstream)
 }
 
 // Count implements CheckPoint.Count
 func (cp *RemoteCheckPoint) Count() (int, error) {
 	query := fmt.Sprintf("SELECT COUNT(id) FROM `%s`.`%s` WHERE `id` = ?", cp.schema, cp.table)
-	rows, err := cp.conn.querySQL(cp.tctx, query, queryRetryCount, cp.id)
+	rows, err := cp.conn.querySQL(cp.tctx, query, cp.id)
 	if err != nil {
 		return 0, terror.WithScope(err, terror.ScopeDownstream)
 	}

loader/checkpoint_test.go

@@ -118,7 +118,7 @@ func (t *testCheckPointSuite) TestForDB(c *C) {
 	defer closeConn(conn)
 	for _, cs := range cases {
 		sql2 := cp.GenSQL(cs.filename, cs.endPos)
-		err = conn.executeSQL(tctx, []string{sql2}, true)
+		err = conn.executeSQL(tctx, []string{sql2})
 		c.Assert(err, IsNil)
 	}