source: check whether GTID is ON in the upstream ealier (#946)

_utils/terror_gen/errors_release.txt

@@ -173,6 +173,7 @@ ErrTaskCheckGenTableRouter,[code=26003:class=task-check:scope=internal:level=med
 ErrTaskCheckGenColumnMapping,[code=26004:class=task-check:scope=internal:level=medium], "Message: generate column mapping error, Workaround: Please check the `column-mappings` config in task configuration file."
 ErrTaskCheckSyncConfigError,[code=26005:class=task-check:scope=internal:level=medium], "Message: %s: %v\n detail: %v"
 ErrTaskCheckGenBAList,[code=26006:class=task-check:scope=internal:level=medium], "Message: generate block allow list error, Workaround: Please check the `block-allow-list` config in task configuration file."
+ErrSourceCheckGTID,[code=26007:class=task-check:scope=internal:level=medium], "Message: %s has GTID_MODE = %s instead of ON, Workaround: Please check the `enable-gtid` config in source configuration file."
 ErrRelayParseUUIDIndex,[code=28001:class=relay-event-lib:scope=internal:level=high], "Message: parse server-uuid.index"
 ErrRelayParseUUIDSuffix,[code=28002:class=relay-event-lib:scope=internal:level=high], "Message: UUID (with suffix) %s not valid"
 ErrRelayUUIDWithSuffixNotFound,[code=28003:class=relay-event-lib:scope=internal:level=high], "Message: no UUID (with suffix) matched %s found in %s, all UUIDs are %v"

dm/config/source_config.go

@@ -241,6 +241,16 @@ func (c *SourceConfig) Adjust(db *sql.DB) (err error) {
 		}
 	}
 
+	if c.EnableGTID {
+		val, err := utils.GetGTID(db)
+		if err != nil {
+			return err
+		}
+		if val != "ON" {
+			return terror.ErrSourceCheckGTID.Generate(c.SourceID, val)
+		}
+	}
+
 	return nil
 }
 

dm/master/server.go

@@ -1351,6 +1351,7 @@ func parseAndAdjustSourceConfig(contents []string) ([]*config.SourceConfig, erro
 		if _, err = cfg.Yaml(); err != nil {
 			return cfgs, err
 		}
+
 		cfgs[i] = cfg
 	}
 	return cfgs, nil

errors.toml

@@ -1048,6 +1048,12 @@ description = ""
 workaround = "Please check the `block-allow-list` config in task configuration file."
 tags = ["internal", "medium"]
 
+[error.DM-task-check-26007]
+message = "%s has GTID_MODE = %s instead of ON"
+description = ""
+workaround = "Please check the `enable-gtid` config in source configuration file."
+tags = ["internal", "medium"]
+
 [error.DM-relay-event-lib-28001]
 message = "parse server-uuid.index"
 description = ""

pkg/terror/error_list.go

@@ -238,6 +238,7 @@ const (
 	codeTaskCheckGenColumnMapping
 	codeTaskCheckSyncConfigError
 	codeTaskCheckGenBAList
+	codeSourceCheckGTID
 )
 
 // Relay log utils error code
@@ -811,6 +812,7 @@ var (
 	ErrTaskCheckGenColumnMapping = New(codeTaskCheckGenColumnMapping, ClassTaskCheck, ScopeInternal, LevelMedium, "generate column mapping error", "Please check the `column-mappings` config in task configuration file.")
 	ErrTaskCheckSyncConfigError  = New(codeTaskCheckSyncConfigError, ClassTaskCheck, ScopeInternal, LevelMedium, "%s: %v\n detail: %v", "")
 	ErrTaskCheckGenBAList        = New(codeTaskCheckGenBAList, ClassTaskCheck, ScopeInternal, LevelMedium, "generate block allow list error", "Please check the `block-allow-list` config in task configuration file.")
+	ErrSourceCheckGTID           = New(codeSourceCheckGTID, ClassTaskCheck, ScopeInternal, LevelMedium, "%s has GTID_MODE = %s instead of ON", "Please check the `enable-gtid` config in source configuration file.")
 
 	// Relay log basic API error
 	ErrRelayParseUUIDIndex         = New(codeRelayParseUUIDIndex, ClassRelayEventLib, ScopeInternal, LevelHigh, "parse server-uuid.index", "")

pkg/terror/terror.go

@@ -38,6 +38,7 @@ const (
 	ClassBinlogOp
 	ClassCheckpoint
 	ClassTaskCheck
+	ClassSourceCheck
 	ClassRelayEventLib
 	ClassRelayUnit
 	ClassDumpUnit
@@ -59,6 +60,7 @@ var errClass2Str = map[ErrClass]string{
 	ClassBinlogOp:      "binlog-op",
 	ClassCheckpoint:    "checkpoint",
 	ClassTaskCheck:     "task-check",
+	ClassSourceCheck:   "source-check",
 	ClassRelayEventLib: "relay-event-lib",
 	ClassRelayUnit:     "relay-unit",
 	ClassDumpUnit:      "dump-unit",

pkg/utils/db.go

@@ -403,3 +403,9 @@ func IsErrBinlogPurged(err error) bool {
 func IsNoSuchThreadError(err error) bool {
 	return IsMySQLError(err, tmysql.ErrNoSuchThread)
 }
+
+// GetGTID return GTID_MODE
+func GetGTID(db *sql.DB) (string, error) {
+	val, err := GetGlobalVariable(db, "GTID_MODE")
+	return val, err
+}