add node-id cmd to specify drainer's node-id

cmd/drainer/README.md

@@ -41,6 +41,8 @@ Usage of drainer:
       prometheus pushgateway address, leaves it empty will disable prometheus push
   -metrics-interval int
       prometheus client push interval in second, set "0" to disable prometheus push (default 15)
+  -node-id string
+      the ID of drainer node; if not specified, we will generate one from hostname and the listening port
   -pd-urls string
       a comma separated list of PD endpoints (default "http://127.0.0.1:2379")
   -safe-mode

cmd/pump/README.md

@@ -34,7 +34,7 @@ pump is a daemon that receives realtime binlog from tidb-server and writes in se
   -metrics-interval int
       prometheus client push interval in second, set "0" to disable prometheus push (default 15)
   -node-id string
-      the ID of pump node; if not specify, we will generate one from hostname and the listening port
+      the ID of pump node; if not specified, we will generate one from hostname and the listening port
   -pd-urls string
       a comma separated list of the PD endpoints (default "http://127.0.0.1:2379")
   -zookeeper-addrs string

drainer/config.go

@@ -65,6 +65,7 @@ type SyncerConfig struct {
 type Config struct {
 	*flag.FlagSet   `json:"-"`
 	LogLevel        string          `toml:"log-level" json:"log-level"`
+	NodeID          string          `toml:"node-id" json:"node-id"`
 	ListenAddr      string          `toml:"addr" json:"addr"`
 	AdvertiseAddr   string          `toml:"advertise-addr" json:"advertise-addr"`
 	DataDir         string          `toml:"data-dir" json:"data-dir"`
@@ -98,6 +99,7 @@ func NewConfig() *Config {
 		fmt.Fprintln(os.Stderr, "Usage of drainer:")
 		fs.PrintDefaults()
 	}
+	fs.StringVar(&cfg.NodeID, "node-id", "", "the ID of drainer node; if not specified, we will generate one from hostname and the listening port")
 	fs.StringVar(&cfg.ListenAddr, "addr", util.DefaultListenAddr(8249), "addr (i.e. 'host:port') to listen on for drainer connections")
 	fs.StringVar(&cfg.AdvertiseAddr, "advertise-addr", "", "addr(i.e. 'host:port') to advertise to the public, default to be the same value as -addr")
 	fs.StringVar(&cfg.DataDir, "data-dir", defaultDataDir, "drainer data directory path (default data.drainer)")

drainer/server.go

@@ -16,7 +16,7 @@ import (
 	"github.com/ngaut/log"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/parser/model"
-	"github.com/pingcap/pd/client"
+	pd "github.com/pingcap/pd/client"
 	"github.com/pingcap/tidb-binlog/drainer/checkpoint"
 	"github.com/pingcap/tidb-binlog/pkg/flags"
 	"github.com/pingcap/tidb-binlog/pkg/node"
@@ -72,13 +72,16 @@ func init() {
 
 // NewServer return a instance of binlog-server
 func NewServer(cfg *Config) (*Server, error) {
-	ID, err := genDrainerID(cfg.ListenAddr)
-	if err != nil {
-		return nil, errors.Trace(err)
+	if cfg.NodeID == "" {
+		var err error
+		cfg.NodeID, err = genDrainerID(cfg.ListenAddr)
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
 	}
 
 	if err1 := os.MkdirAll(cfg.DataDir, 0700); err1 != nil {
-		return nil, err
+		return nil, err1
 	}
 
 	// get pd client and cluster ID
@@ -133,10 +136,10 @@ func NewServer(cfg *Config) (*Server, error) {
 		return nil, errors.Annotatef(err, "invalid configuration of advertise addr(%s)", cfg.AdvertiseAddr)
 	}
 
-	status := node.NewStatus(ID, advURL.Host, node.Online, 0, syncer.GetLatestCommitTS(), util.GetApproachTS(latestTS, latestTime))
+	status := node.NewStatus(cfg.NodeID, advURL.Host, node.Online, 0, syncer.GetLatestCommitTS(), util.GetApproachTS(latestTS, latestTime))
 
 	return &Server{
-		ID:        ID,
+		ID:        cfg.NodeID,
 		host:      advURL.Host,
 		cfg:       cfg,
 		collector: c,

pump/config.go

@@ -81,7 +81,7 @@ func NewConfig() *Config {
 		fs.PrintDefaults()
 	}
 
-	fs.StringVar(&cfg.NodeID, "node-id", "", "the ID of pump node; if not specify, we will generate one from hostname and the listening port")
+	fs.StringVar(&cfg.NodeID, "node-id", "", "the ID of pump node; if not specified, we will generate one from hostname and the listening port")
 	fs.StringVar(&cfg.ListenAddr, "addr", util.DefaultListenAddr(8250), "addr(i.e. 'host:port') to listen on for client traffic")
 	fs.StringVar(&cfg.AdvertiseAddr, "advertise-addr", "", "addr(i.e. 'host:port') to advertise to the public")
 	fs.StringVar(&cfg.Socket, "socket", "", "unix socket addr to listen on for client traffic")