add binlog socket config for compatible with kafka version pump (#8747)

pingcap · Dec 19, 2018 · 79180dc · 79180dc
1 parent 0c7ca13
commit 79180dc
Show file tree

Hide file tree

Showing 6 changed files with 24 additions and 5 deletions.
diff --git a/config/config.go b/config/config.go
@@ -246,6 +246,8 @@ type Binlog struct {
 	// If IgnoreError is true, when writting binlog meets error, TiDB would
 	// ignore the error.
 	IgnoreError bool `toml:"ignore-error" json:"ignore-error"`
+	// Use socket file to write binlog, for compatible with kafka version tidb-binlog.
+	BinlogSocket string `toml:"binlog-socket" json:"binlog-socket"`
 }
 
 var defaultConf = Config{

diff --git a/config/config.toml.example b/config/config.toml.example
@@ -245,3 +245,6 @@ write-timeout = "15s"
 # If IgnoreError is true, when writting binlog meets error, TiDB would stop writting binlog,
 # but still provide service.
 ignore-error = false
+
+# use socket file to write binlog, for compatible with kafka version tidb-binlog.
+binlog-socket = ""
diff --git a/config/config_test.go b/config/config_test.go
@@ -35,6 +35,7 @@ func (s *testConfigSuite) TestConfig(c *C) {
 	conf := new(Config)
 	conf.Binlog.Enable = true
 	conf.Binlog.IgnoreError = true
+	conf.Binlog.BinlogSocket = "/tmp/socket"
 	conf.TiKVClient.CommitTimeout = "10s"
 
 	configFile := "config.toml"
@@ -53,6 +54,7 @@ commit-timeout="41s"`)
 
 	// Test that the original value will not be clear by load the config file that does not contain the option.
 	c.Assert(conf.Binlog.Enable, Equals, true)
+	c.Assert(conf.Binlog.BinlogSocket, Equals, "/tmp/socket")
 
 	c.Assert(conf.TiKVClient.CommitTimeout, Equals, "41s")
 	c.Assert(f.Close(), IsNil)

diff --git a/go.mod b/go.mod
@@ -50,7 +50,7 @@ require (
 	github.com/pingcap/kvproto v0.0.0-20181109035735-8e3f33ac4929
 	github.com/pingcap/parser v0.0.0-20181207085916-6c21d4344dfa
 	github.com/pingcap/pd v2.1.0-rc.4+incompatible
-	github.com/pingcap/tidb-tools v0.0.0-20181112132202-4860a0d5de03
+	github.com/pingcap/tidb-tools v2.1.1-0.20181218072513-b2235d442b06+incompatible
 	github.com/pingcap/tipb v0.0.0-20180910045846-371b48b15d93
 	github.com/pkg/errors v0.8.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect

diff --git a/go.sum b/go.sum
@@ -96,8 +96,8 @@ github.com/pingcap/parser v0.0.0-20181207085916-6c21d4344dfa h1:m6vqGJWtMKHr2Riy
 github.com/pingcap/parser v0.0.0-20181207085916-6c21d4344dfa/go.mod h1:1FNvfp9+J0wvc4kl8eGNh7Rqrxveg15jJoWo/a0uHwA=
 github.com/pingcap/pd v2.1.0-rc.4+incompatible h1:/buwGk04aHO5odk/+O8ZOXGs4qkUjYTJ2UpCJXna8NE=
 github.com/pingcap/pd v2.1.0-rc.4+incompatible/go.mod h1:nD3+EoYes4+aNNODO99ES59V83MZSI+dFbhyr667a0E=
-github.com/pingcap/tidb-tools v0.0.0-20181112132202-4860a0d5de03 h1:xVuo5U+l6XAWHsb+xhkZ8zz3jerIwDfCHAO6kR2Kaog=
-github.com/pingcap/tidb-tools v0.0.0-20181112132202-4860a0d5de03/go.mod h1:XGdcy9+yqlDSEMTpOXnwf3hiTeqrV6MN/u1se9N8yIM=
+github.com/pingcap/tidb-tools v2.1.1-0.20181218072513-b2235d442b06+incompatible h1:Bsd+NHosPVowEGB3BCx+2d8wUQGDTXSSC5ljeNS6cXo=
+github.com/pingcap/tidb-tools v2.1.1-0.20181218072513-b2235d442b06+incompatible/go.mod h1:XGdcy9+yqlDSEMTpOXnwf3hiTeqrV6MN/u1se9N8yIM=
 github.com/pingcap/tipb v0.0.0-20180910045846-371b48b15d93 h1:gI5bOzLMxjUq6ui+md/JnT4pYpkzrABJ/PeYORWiYYs=
 github.com/pingcap/tipb v0.0.0-20180910045846-371b48b15d93/go.mod h1:RtkHW8WbcNxj8lsbzjaILci01CtYnYbIkQhjyZWrWVI=
 github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw=

diff --git a/tidb-server/main.go b/tidb-server/main.go
@@ -176,11 +176,23 @@ func setupBinlogClient() {
 		binloginfo.SetIgnoreError(true)
 	}
 
-	client, err := pumpcli.NewPumpsClient(cfg.Path, parseDuration(cfg.Binlog.WriteTimeout), pd.SecurityOption{
+	var (
+		client *pumpcli.PumpsClient
+		err    error
+	)
+
+	securityOption := pd.SecurityOption{
 		CAPath:   cfg.Security.ClusterSSLCA,
 		CertPath: cfg.Security.ClusterSSLCert,
 		KeyPath:  cfg.Security.ClusterSSLKey,
-	})
+	}
+
+	if len(cfg.Binlog.BinlogSocket) == 0 {
+		client, err = pumpcli.NewPumpsClient(cfg.Path, parseDuration(cfg.Binlog.WriteTimeout), securityOption)
+	} else {
+		client, err = pumpcli.NewLocalPumpsClient(cfg.Path, cfg.Binlog.BinlogSocket, parseDuration(cfg.Binlog.WriteTimeout), securityOption)
+	}
+
 	terror.MustNil(err)
 
 	binloginfo.SetPumpsClient(client)