worker: use random value as server-id if server-id is not set

dm/config/subtask.go

@@ -145,7 +145,7 @@ type SubTaskConfig struct {
 	IgnoreCheckingItems []string `toml:"ignore-checking-items" json:"ignore-checking-items"`
 	// it represents a MySQL/MariaDB instance or a replica group
 	SourceID                string `toml:"source-id" json:"source-id"`
-	ServerID                int    `toml:"server-id" json:"server-id"`
+	ServerID                uint32  `toml:"server-id" json:"server-id"`
 	Flavor                  string `toml:"flavor" json:"flavor"`
 	MetaSchema              string `toml:"meta-schema" json:"meta-schema"`
 	RemoveMeta              bool   `toml:"remove-meta" json:"remove-meta"`

dm/worker/config.go

@@ -22,6 +22,7 @@ import (
 	"flag"
 	"fmt"
 	"io/ioutil"
+	"math"
 	"math/rand"
 	"strings"
 	"time"
@@ -46,7 +47,7 @@ const (
 	// flavorGetTimeout is timeout for getting some information from DB
 	dbGetTimeout = 30 * time.Second
 
-	maxServerID = 1<<32 - 1
+	maxServerID uint32 = math.MaxUint32
 )
 
 // SampleConfigFile is sample config file of dm-worker
@@ -100,7 +101,7 @@ type Config struct {
 	AutoFixGTID bool   `toml:"auto-fix-gtid" json:"auto-fix-gtid"`
 	RelayDir    string `toml:"relay-dir" json:"relay-dir"`
 	MetaDir     string `toml:"meta-dir" json:"meta-dir"`
-	ServerID    int    `toml:"server-id" json:"server-id"`
+	ServerID    uint32    `toml:"server-id" json:"server-id"`
 	Flavor      string `toml:"flavor" json:"flavor"`
 	Charset     string `toml:"charset" json:"charset"`
 
@@ -342,9 +343,9 @@ func (c *Config) adjustServerID(ctx context.Context, db *sql.DB) error {
 	}
 
 	for i := 0; i < 5; i++ {
-		randomValue := rand.Intn(100000)
+		randomValue := uint32(rand.Intn(100000))
 		randomServerID := maxServerID/10 + randomValue
-		if _, ok := serverIDs[int64(randomServerID)]; ok {
+		if _, ok := serverIDs[randomServerID]; ok {
 			continue
 		}
 

pkg/utils/db.go

@@ -53,7 +53,23 @@ func GetFlavor(ctx context.Context, db *sql.DB) (string, error) {
 }
 
 // GetAllServerID gets all slave server id and master server id
-func GetAllServerID(ctx context.Context, db *sql.DB) (map[int64]interface{}, error) {
+func GetAllServerID(ctx context.Context, db *sql.DB) (map[uint32]interface{}, error) {
+	serverIDs, err := GetSlaveServerID(ctx, db)
+	if err != nil {
+		return nil, err
+	}
+
+	masterServerID, err := GetServerID(db)
+	if err != nil {
+		return nil, err
+	}
+
+	serverIDs[masterServerID] = struct{}{}
+	return serverIDs, nil
+}
+
+// GetSlaveServerID gets all slave server id
+func GetSlaveServerID(ctx context.Context, db *sql.DB) (map[uint32]interface{}, error) {
 	rows, err := db.QueryContext(ctx, `SHOW SLAVE HOSTS`)
 	if err != nil {
 		return nil, terror.DBErrorAdapt(err, terror.ErrDBDriverError)
@@ -92,7 +108,7 @@ func GetAllServerID(ctx context.Context, db *sql.DB) (map[int64]interface{}, err
 		masterID  sql.NullInt64
 		slaveUUID sql.NullString
 	)
-	serverIDs := make(map[int64]interface{})
+	serverIDs := make(map[uint32]interface{})
 	for rows.Next() {
 		if len(rowColumns) == 5 {
 			err = rows.Scan(&serverID, &host, &port, &masterID, &slaveUUID)
@@ -104,20 +120,12 @@ func GetAllServerID(ctx context.Context, db *sql.DB) (map[int64]interface{}, err
 		}
 
 		if serverID.Valid {
-			serverIDs[serverID.Int64] = struct{}{}
+			serverIDs[uint32(serverID.Int64)] = struct{}{}
 		} else {
 			// should never happened
 			log.L().Warn("get invalid server_id when execute `SHOW SLAVE HOSTS;`")
 			continue
 		}
-
-		if masterID.Valid {
-			serverIDs[masterID.Int64] = struct{}{}
-		} else {
-			// should never happened
-			log.L().Warn("get invalid master_id when execute `SHOW SLAVE HOSTS;`")
-			continue
-		}
 	}
 
 	if rows.Err() != nil {
@@ -260,14 +268,14 @@ func GetGlobalVariable(db *sql.DB, variable string) (value string, err error) {
 }
 
 // GetServerID gets server's `server_id`
-func GetServerID(db *sql.DB) (int64, error) {
+func GetServerID(db *sql.DB) (uint32, error) {
 	serverIDStr, err := GetGlobalVariable(db, "server_id")
 	if err != nil {
 		return 0, err
 	}
 
-	serverID, err := strconv.ParseInt(serverIDStr, 10, 64)
-	return serverID, terror.ErrInvalidServerID.Delegate(err, serverIDStr)
+	serverID, err := strconv.ParseInt(serverIDStr, 10, 32)
+	return uint32(serverID), terror.ErrInvalidServerID.Delegate(err, serverIDStr)
 }
 
 // GetMariaDBGtidDomainID gets MariaDB server's `gtid_domain_id`

pkg/utils/db_test.go

@@ -23,11 +23,14 @@ import (
 
 func (t *testUtilsSuite) TestGetAllServerID(c *C) {
 	testCases := []struct {
+		masterID  int64
 		serverIDs []int64
 	}{
 		{
-			[]int64{1, 2, 3},
+			1,
+			[]int64{2, 3, 4},
 		}, {
+			2,
 			[]int64{},
 		},
 	}
@@ -39,24 +42,29 @@ func (t *testUtilsSuite) TestGetAllServerID(c *C) {
 
 	for _, testCase := range testCases {
 		for _, flavor := range flavors {
-			t.createMockResult(mock, testCase.serverIDs, flavor)
-			slaveHosts, err := GetAllServerID(context.Background(), db)
+			t.createMockResult(mock, testCase.masterID, testCase.serverIDs, flavor)
+			serverIDs, err := GetAllServerID(context.Background(), db)
 			c.Assert(err, IsNil)
 
 			for _, serverID := range testCase.serverIDs {
-				_, ok := slaveHosts[serverID]
+				_, ok := serverIDs[serverID]
 				c.Assert(ok, IsTrue)
 			}
+
+			_, ok := serverIDs[testCase.masterID]
+			c.Assert(ok, IsTrue)
 		}
 	}
+
+	err = mock.ExpectationsWereMet()
+	c.Assert(err, IsNil)
 }
 
-func (t *testUtilsSuite) createMockResult(mock sqlmock.Sqlmock, serverIDs []int64, flavor string) {
+func (t *testUtilsSuite) createMockResult(mock sqlmock.Sqlmock, masterID int64, serverIDs []int64, flavor string) {
 	expectQuery := mock.ExpectQuery("SHOW SLAVE HOSTS")
 
 	host := "test"
 	port := 3306
-	masterID := 1
 	slaveUUID := "test"
 
 	if flavor == mysql.MariaDBFlavor {
@@ -74,5 +82,8 @@ func (t *testUtilsSuite) createMockResult(mock sqlmock.Sqlmock, serverIDs []int6
 		expectQuery.WillReturnRows(rows)
 	}
 
+	mock.ExpectQuery("SHOW GLOBAL VARIABLES LIKE 'server_id'").WillReturnRows(sqlmock.NewRows([]string{"Variable_name", "Value"}).AddRow("server_id", masterID))
+
+
 	return
 }

relay/config.go

@@ -25,7 +25,7 @@ type Config struct {
 	EnableGTID  bool     `toml:"enable-gtid" json:"enable-gtid"`
 	AutoFixGTID bool     `toml:"auto-fix-gtid" json:"auto-fix-gtid"`
 	RelayDir    string   `toml:"relay-dir" json:"relay-dir"`
-	ServerID    int      `toml:"server-id" json:"server-id"`
+	ServerID    uint32   `toml:"server-id" json:"server-id"`
 	Flavor      string   `toml:"flavor" json:"flavor"`
 	Charset     string   `toml:"charset" json:"charset"`
 	From        DBConfig `toml:"data-source" json:"data-source"`

syncer/heartbeat.go

@@ -51,7 +51,7 @@ type HeartbeatConfig struct {
 	// serverID from dm-worker (relay)
 	// now, heartbeat not be synced to downstream
 	// so it will not be used by user directly and also enough to differ from other dm-worker's
-	serverID  int
+	serverID  uint32
 	masterCfg config.DBConfig // master server's DBConfig
 }
 
@@ -191,13 +191,13 @@ func (h *Heartbeat) TryUpdateTaskTs(taskName, schema, table string, data [][]int
 	}
 
 	latest := data[len(data)-1]
-	serverID, ok := latest[1].(int32)
+	serverID, ok := latest[1].(uint32)
 	if !ok {
 		h.logger.Warn("invalid data server_id for heartbeat", zap.Reflect("server ID", latest[1]))
 		return
 	}
-	if int(serverID) != h.cfg.serverID {
-		h.logger.Debug("ignore mismatched server_id for heartbeat", zap.Int32("obtained server ID", serverID), zap.Int("excepted server ID", h.cfg.serverID))
+	if serverID != h.cfg.serverID {
+		h.logger.Debug("ignore mismatched server_id for heartbeat", zap.Uint32("obtained server ID", serverID), zap.Uint32("excepted server ID", h.cfg.serverID))
 		return // only ignore
 	}
 
@@ -295,7 +295,7 @@ func (h *Heartbeat) updateTS() error {
 	// when we not need to support MySQL <=5.5, we can replace with `UTC_TIMESTAMP(6)`
 	query := fmt.Sprintf("REPLACE INTO `%s`.`%s` (`ts`, `server_id`) VALUES(UTC_TIMESTAMP(), ?)", h.schema, h.table)
 	_, err := h.master.Exec(query, h.cfg.serverID)
-	h.logger.Debug("update ts", zap.String("sql", query), zap.Int("server ID", h.cfg.serverID))
+	h.logger.Debug("update ts", zap.String("sql", query), zap.Uint32("server ID", h.cfg.serverID))
 	return terror.WithScope(terror.DBErrorAdapt(err, terror.ErrDBDriverError), terror.ScopeUpstream)
 }
 

syncer/online_ddl.go

@@ -100,7 +100,7 @@ func NewOnlineDDLStorage(newtctx *tcontext.Context, cfg *config.SubTaskConfig) *
 		cfg:    cfg,
 		schema: cfg.MetaSchema,
 		table:  fmt.Sprintf("%s_onlineddl", cfg.Name),
-		id:     strconv.Itoa(cfg.ServerID),
+		id:     strconv.Itoa(int(cfg.ServerID)),
 		ddls:   make(map[string]map[string]*GhostDDLInfo),
 		tctx:   newtctx,
 	}

syncer/syncer.go

@@ -2396,7 +2396,7 @@ func (s *Syncer) checkpointID() string {
 	if len(s.cfg.SourceID) > 0 {
 		return s.cfg.SourceID
 	}
-	return strconv.Itoa(s.cfg.ServerID)
+	return strconv.Itoa(int(s.cfg.ServerID))
 }
 
 // DDLInfo returns a chan from which can receive DDLInfo