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

dm/worker/config.go

@@ -16,11 +16,13 @@ package worker
 import (
 	"bytes"
 	"context"
+	"database/sql"
 	"encoding/base64"
 	"encoding/json"
 	"flag"
 	"fmt"
 	"io/ioutil"
+	"math/rand"
 	"strings"
 	"time"
 
@@ -39,18 +41,22 @@ import (
 )
 
 const (
-	// flavorReadTimeout is readTimeout for DB connection in adjustFlavor
-	flavorReadTimeout = "30s"
-	// flavorGetTimeout is timeout for getting version info from DB
-	flavorGetTimeout = 30 * time.Second
+	// dbReadTimeout is readTimeout for DB connection in adjust
+	dbReadTimeout = "30s"
+	// flavorGetTimeout is timeout for getting some information from DB
+	dbGetTimeout = 30 * time.Second
+
+	maxServerID = 1<<32 - 1
 )
 
 // SampleConfigFile is sample config file of dm-worker
 // later we can read it from dm/worker/dm-worker.toml
 // and assign it to SampleConfigFile while we build dm-worker
 var SampleConfigFile string
 
-var applyNewBaseDB = conn.DefaultDBProvider.Apply
+var (
+	getAllServerIDFunc = utils.GetAllServerID
+)
 
 // NewConfig creates a new base config for worker.
 func NewConfig() *Config {
@@ -199,9 +205,7 @@ func (c *Config) Parse(arguments []string) error {
 	// assign tracer id to source id
 	c.Tracer.Source = c.SourceID
 
-	c.From.Adjust()
-	c.Checker.adjust()
-	err = c.adjustFlavor()
+	err = c.adjust()
 	if err != nil {
 		return err
 	}
@@ -256,8 +260,54 @@ func (c *Config) configFromFile(path string) error {
 	return c.verify()
 }
 
+func (c *Config) adjust() error {
+	c.From.Adjust()
+	c.Checker.adjust()
+
+	if c.Flavor == "" || c.ServerID == 0 {
+		return nil
+
+		fromDB, err := c.createBaseDB()
+		if err != nil {
+			return err
+		}
+		defer fromDB.Close()
+
+		ctx, cancel := context.WithTimeout(context.Background(), dbGetTimeout)
+		defer cancel()
+
+		err = c.adjustFlavor(ctx, fromDB.DB)
+		if err != nil {
+			return err
+		}
+
+		err = c.adjustServerID(ctx, fromDB.DB)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c *Config) createBaseDB() (*conn.BaseDB, error) {
+	// decrypt password
+	clone, err := c.DecryptPassword()
+	if err != nil {
+		return nil, err
+	}
+	from := clone.From
+	from.RawDBCfg = config.DefaultRawDBConfig().SetReadTimeout(dbReadTimeout)
+	fromDB, err := conn.DefaultDBProvider.Apply(from)
+	if err != nil {
+		return nil, terror.WithScope(err, terror.ScopeUpstream)
+	}
+
+	return fromDB, nil
+}
+
 // adjustFlavor adjusts flavor through querying from given database
-func (c *Config) adjustFlavor() error {
+func (c *Config) adjustFlavor(ctx context.Context, db *sql.DB) (err error) {
 	if c.Flavor != "" {
 		switch c.Flavor {
 		case mysql.MariaDBFlavor, mysql.MySQLFlavor:
@@ -266,26 +316,43 @@ func (c *Config) adjustFlavor() error {
 			return terror.ErrNotSupportedFlavor.Generate(c.Flavor)
 		}
 	}
-	// decrypt password
-	clone, err := c.DecryptPassword()
-	if err != nil {
-		return err
+
+	c.Flavor, err = utils.GetFlavor(ctx, db)
+	if ctx.Err() != nil {
+		err = terror.Annotatef(err, "time cost to get flavor info exceeds %s", dbGetTimeout)
+	}
+	return terror.WithScope(err, terror.ScopeUpstream)
+}
+
+func (c *Config) adjustServerID(ctx context.Context, db *sql.DB) error {
+	if c.ServerID != 0 {
+		if c.ServerID < 0 || c.ServerID > maxServerID {
+			return terror.ErrInvalidServerID.Generate(c.ServerID)
+		}
+
+		return nil
+	}
+
+	serverIDs, err := getAllServerIDFunc(ctx, db)
+	if ctx.Err() != nil {
+		err = terror.Annotatef(err, "time cost to get flavor info exceeds %s", dbGetTimeout)
 	}
-	from := clone.From
-	from.RawDBCfg = config.DefaultRawDBConfig().SetReadTimeout(flavorReadTimeout)
-	fromDB, err := applyNewBaseDB(from)
 	if err != nil {
 		return terror.WithScope(err, terror.ScopeUpstream)
 	}
-	defer fromDB.Close()
 
-	ctx, cancel := context.WithTimeout(context.Background(), flavorGetTimeout)
-	defer cancel()
-	c.Flavor, err = utils.GetFlavor(ctx, fromDB.DB)
-	if ctx.Err() != nil {
-		err = terror.Annotatef(err, "time cost to get flavor info exceeds %s", flavorGetTimeout)
+	for i := 0; i < 5; i++ {
+		randomValue := rand.Intn(100000)
+		randomServerID := maxServerID/10 + randomValue
+		if _, ok := serverIDs[int64(randomServerID)]; ok {
+			continue
+		}
+
+		c.ServerID = randomServerID
+		return nil
 	}
-	return terror.WithScope(err, terror.ScopeUpstream)
+
+	return terror.ErrInvalidServerID.Generatef("can't find a random available server ID")
 }
 
 // UpdateConfigFile write configure to local file

dm/worker/config_test.go

@@ -14,17 +14,18 @@
 package worker
 
 import (
+	"context"
+	"database/sql"
 	"fmt"
 	"io/ioutil"
 	"path"
 	"strings"
 
-	"github.com/DATA-DOG/go-sqlmock"
+	sqlmock "github.com/DATA-DOG/go-sqlmock"
 	. "github.com/pingcap/check"
 	"github.com/siddontang/go-mysql/mysql"
 
 	"github.com/pingcap/dm/dm/config"
-	"github.com/pingcap/dm/pkg/conn"
 )
 
 func (t *testServer) TestConfig(c *C) {
@@ -172,10 +173,8 @@ func subtestFlavor(c *C, cfg *Config, sqlInfo, expectedFlavor, expectedError str
 		WillReturnRows(sqlmock.NewRows([]string{"Variable_name", "Value"}).
 			AddRow("version", sqlInfo))
 	mock.ExpectClose()
-	applyNewBaseDB = func(config config.DBConfig) (*conn.BaseDB, error) {
-		return &conn.BaseDB{DB: db}, nil
-	}
-	err = cfg.adjustFlavor()
+
+	err = cfg.adjustFlavor(context.Background(), db)
 	if expectedError == "" {
 		c.Assert(err, IsNil)
 		c.Assert(cfg.Flavor, Equals, expectedFlavor)
@@ -189,18 +188,38 @@ func (t *testServer) TestAdjustFlavor(c *C) {
 	c.Assert(cfg.Parse([]string{"-config=./dm-worker.toml", "-relay-dir=./xx"}), IsNil)
 
 	cfg.Flavor = "mariadb"
-	err := cfg.adjustFlavor()
+	err := cfg.adjustFlavor(context.Background(), nil)
 	c.Assert(err, IsNil)
 	c.Assert(cfg.Flavor, Equals, mysql.MariaDBFlavor)
 	cfg.Flavor = "MongoDB"
-	err = cfg.adjustFlavor()
+	err = cfg.adjustFlavor(context.Background(), nil)
 	c.Assert(err, ErrorMatches, ".*flavor MongoDB not supported")
 
-	var origApplyNewBaseDB = applyNewBaseDB
+	subtestFlavor(c, cfg, "10.4.8-MariaDB-1:10.4.8+maria~bionic", mysql.MariaDBFlavor, "")
+	subtestFlavor(c, cfg, "5.7.26-log", mysql.MySQLFlavor, "")
+}
+
+func (t *testServer) TestAdjustServerID(c *C) {
+	var originGetAllServerIDFunc = getAllServerIDFunc
 	defer func() {
-		applyNewBaseDB = origApplyNewBaseDB
+		getAllServerIDFunc = originGetAllServerIDFunc
 	}()
+	getAllServerIDFunc = getMockServerIDs
 
-	subtestFlavor(c, cfg, "10.4.8-MariaDB-1:10.4.8+maria~bionic", mysql.MariaDBFlavor, "")
-	subtestFlavor(c, cfg, "5.7.26-log", mysql.MySQLFlavor, "")
+	cfg := NewConfig()
+	c.Assert(cfg.Parse([]string{"-config=./dm-worker.toml", "-relay-dir=./xx"}), IsNil)
+
+	cfg.adjustServerID(context.Background(), nil)
+	c.Assert(cfg.ServerID, Equals, 101)
+
+	cfg.ServerID = 0
+	cfg.adjustServerID(context.Background(), nil)
+	c.Assert(cfg.ServerID, Not(Equals), 0)
+}
+
+func getMockServerIDs(ctx context.Context, db *sql.DB) (map[int64]interface{}, error) {
+	return map[int64]interface{}{
+		1: struct{}{},
+		2: struct{}{},
+	}, nil
 }

pkg/utils/db.go

@@ -52,6 +52,81 @@ func GetFlavor(ctx context.Context, db *sql.DB) (string, error) {
 	return gmysql.MySQLFlavor, nil
 }
 
+// GetAllServerID gets all slave server id and master server id
+func GetAllServerID(ctx context.Context, db *sql.DB) (map[int64]interface{}, error) {
+	rows, err := db.QueryContext(ctx, `SHOW SLAVE HOSTS`)
+	if err != nil {
+		return nil, terror.DBErrorAdapt(err, terror.ErrDBDriverError)
+	}
+	defer rows.Close()
+
+	rowColumns, err := rows.Columns()
+	if err != nil {
+		return nil, terror.DBErrorAdapt(err, terror.ErrDBDriverError)
+	}
+
+	/*
+		in MySQL:
+		mysql> SHOW SLAVE HOSTS;
+		+------------+-----------+------+-----------+--------------------------------------+
+		| Server_id  | Host      | Port | Master_id | Slave_UUID                           |
+		+------------+-----------+------+-----------+--------------------------------------+
+		|  192168010 | iconnect2 | 3306 | 192168011 | 14cb6624-7f93-11e0-b2c0-c80aa9429562 |
+		| 1921680101 | athena    | 3306 | 192168011 | 07af4990-f41f-11df-a566-7ac56fdaf645 |
+		+------------+-----------+------+-----------+--------------------------------------+
+
+		in MariaDB:
+		mysql> SHOW SLAVE HOSTS;
+		+------------+-----------+------+-----------+
+		| Server_id  | Host      | Port | Master_id |
+		+------------+-----------+------+-----------+
+		|  192168010 | iconnect2 | 3306 | 192168011 |
+		| 1921680101 | athena    | 3306 | 192168011 |
+		+------------+-----------+------+-----------+
+	*/
+
+	var (
+		serverID  sql.NullInt64
+		host      sql.NullString
+		port      sql.NullInt64
+		masterID  sql.NullInt64
+		slaveUUID sql.NullString
+	)
+	serverIDs := make(map[int64]interface{})
+	for rows.Next() {
+		if len(rowColumns) == 5 {
+			err = rows.Scan(&serverID, &host, &port, &masterID, &slaveUUID)
+		} else {
+			err = rows.Scan(&serverID, &host, &port, &masterID)
+		}
+		if err != nil {
+			return nil, terror.DBErrorAdapt(err, terror.ErrDBDriverError)
+		}
+
+		if serverID.Valid {
+			serverIDs[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 {
+		return nil, terror.DBErrorAdapt(rows.Err(), terror.ErrDBDriverError)
+	}
+
+	return serverIDs, nil
+}
+
 // GetMasterStatus gets status from master
 func GetMasterStatus(db *sql.DB, flavor string) (gmysql.Position, gtid.Set, error) {
 	var (
@@ -61,7 +136,7 @@ func GetMasterStatus(db *sql.DB, flavor string) (gmysql.Position, gtid.Set, erro
 
 	rows, err := db.Query(`SHOW MASTER STATUS`)
 	if err != nil {
-		return binlogPos, gs, err
+		return binlogPos, gs, terror.DBErrorAdapt(err, terror.ErrDBDriverError)
 	}
 	defer rows.Close()