binlog: revert commit for variable tidb_log_bin && add send binlog …

…strategy config (#9864)

config/config.go

@@ -251,13 +251,14 @@ type TiKVClient struct {
 // Binlog is the config for binlog.
 type Binlog struct {
 	Enable       bool   `toml:"enable" json:"enable"`
-	AutoMode     bool   `toml:"auto-mode" json:"auto-mode"`
 	WriteTimeout string `toml:"write-timeout" json:"write-timeout"`
 	// If IgnoreError is true, when writing 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"`
+	// The strategy for sending binlog to pump, value can be "range" or "hash" now.
+	Strategy string `toml:"strategy" json:"strategy"`
 }
 
 // Plugin is the config for plugin
@@ -345,6 +346,7 @@ var defaultConf = Config{
 	},
 	Binlog: Binlog{
 		WriteTimeout: "15s",
+		Strategy:     "range",
 	},
 }
 

config/config.toml.example

@@ -261,12 +261,9 @@ enabled = true
 capacity = 2048000
 
 [binlog]
-# Enable to write binlog. This config will be disabled if auto-mode is true.
+# enable to write binlog.
 enable = false
 
-# If auto-mode is true, will enable binlog according to the system variables 'tidb_log_bin'.
-auto-mode = false
-
 # WriteTimeout specifies how long it will wait for writing binlog to pump.
 write-timeout = "15s"
 
@@ -276,3 +273,6 @@ ignore-error = false
 
 # use socket file to write binlog, for compatible with kafka version tidb-binlog.
 binlog-socket = ""
+
+# the strategy for sending binlog to pump, value can be "range" or "hash" now.
+strategy = "range"

config/config_test.go

@@ -34,8 +34,8 @@ func TestT(t *testing.T) {
 func (s *testConfigSuite) TestConfig(c *C) {
 	conf := new(Config)
 	conf.Binlog.Enable = true
-	conf.Binlog.AutoMode = true
 	conf.Binlog.IgnoreError = true
+	conf.Binlog.Strategy = "hash"
 	conf.TiKVClient.CommitTimeout = "10s"
 	conf.CheckMb4ValueInUtf8 = true
 	configFile := "config.toml"
@@ -56,7 +56,7 @@ max-batch-size=128
 
 	// 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.AutoMode, Equals, true)
+	c.Assert(conf.Binlog.Strategy, Equals, "hash")
 
 	c.Assert(conf.TiKVClient.CommitTimeout, Equals, "41s")
 	c.Assert(conf.TiKVClient.MaxBatchSize, Equals, uint(128))

executor/set_test.go

@@ -18,8 +18,8 @@ import (
 
 	. "github.com/pingcap/check"
 	"github.com/pingcap/parser/terror"
+	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/sessionctx"
-	"github.com/pingcap/tidb/sessionctx/binloginfo"
 	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/util/testkit"
 	"github.com/pingcap/tidb/util/testutil"
@@ -239,13 +239,8 @@ func (s *testSuite2) TestSetVar(c *C) {
 	tk.MustExec("set @@sql_log_bin = on")
 	tk.MustQuery(`select @@session.sql_log_bin;`).Check(testkit.Rows("1"))
 
-	tk.MustQuery(`select @@global.log_bin;`).Check(testkit.Rows(variable.BoolToIntStr(binloginfo.ShouldEnableBinlog())))
-	tk.MustQuery(`select @@log_bin;`).Check(testkit.Rows(variable.BoolToIntStr(binloginfo.ShouldEnableBinlog())))
-
-	tk.MustExec("set global tidb_log_bin = on")
-	tk.MustQuery(`select @@global.tidb_log_bin;`).Check(testkit.Rows("1"))
-	tk.MustExec("set global tidb_log_bin = off")
-	tk.MustQuery(`select @@global.tidb_log_bin;`).Check(testkit.Rows("0"))
+	tk.MustQuery(`select @@global.log_bin;`).Check(testkit.Rows(variable.BoolToIntStr(config.GetGlobalConfig().Binlog.Enable)))
+	tk.MustQuery(`select @@log_bin;`).Check(testkit.Rows(variable.BoolToIntStr(config.GetGlobalConfig().Binlog.Enable)))
 
 	tk.MustExec("set @@tidb_general_log = 1")
 	tk.MustExec("set @@tidb_general_log = 0")

go.mod

@@ -53,7 +53,7 @@ require (
 	github.com/pingcap/log v0.0.0-20190307075452-bd41d9273596
 	github.com/pingcap/parser v0.0.0-20190321052000-f9a452f8f24e
 	github.com/pingcap/pd v2.1.0-rc.4+incompatible
-	github.com/pingcap/tidb-tools v2.1.3-0.20190116051332-34c808eef588+incompatible
+	github.com/pingcap/tidb-tools v2.1.3-0.20190321065848-1e8b48f5c168+incompatible
 	github.com/pingcap/tipb v0.0.0-20190107072121-abbec73437b7
 	github.com/pquerna/ffjson v0.0.0-20181028064349-e517b90714f7 // indirect
 	github.com/prometheus/client_golang v0.9.0

go.sum

@@ -123,8 +123,8 @@ github.com/pingcap/parser v0.0.0-20190321052000-f9a452f8f24e h1:Evw2H5BmAGqHTKbb
 github.com/pingcap/parser v0.0.0-20190321052000-f9a452f8f24e/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 v2.1.3-0.20190116051332-34c808eef588+incompatible h1:e9Gi/LP9181HT3gBfSOeSBA+5JfemuE4aEAhqNgoE4k=
-github.com/pingcap/tidb-tools v2.1.3-0.20190116051332-34c808eef588+incompatible/go.mod h1:XGdcy9+yqlDSEMTpOXnwf3hiTeqrV6MN/u1se9N8yIM=
+github.com/pingcap/tidb-tools v2.1.3-0.20190321065848-1e8b48f5c168+incompatible h1:MkWCxgZpJBgY2f4HtwWMMFzSBb3+JPzeJgF3VrXE/bU=
+github.com/pingcap/tidb-tools v2.1.3-0.20190321065848-1e8b48f5c168+incompatible/go.mod h1:XGdcy9+yqlDSEMTpOXnwf3hiTeqrV6MN/u1se9N8yIM=
 github.com/pingcap/tipb v0.0.0-20190107072121-abbec73437b7 h1:wnjdQRhybddDesBVBKyOLUPgDaOFdtqA92pduBgWvVQ=
 github.com/pingcap/tipb v0.0.0-20190107072121-abbec73437b7/go.mod h1:RtkHW8WbcNxj8lsbzjaILci01CtYnYbIkQhjyZWrWVI=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=

session/session.go

@@ -1361,13 +1361,6 @@ func BootstrapSession(store kv.Storage) (*domain.Domain, error) {
 		return nil, errors.Trace(err)
 	}
 
-	// get global system variable tidb_log_bin from mysql.GLOBAL_VARIABLES
-	tidbLogBin, err := se1.GetGlobalSysVar(variable.TiDBLogBin)
-	if err != nil {
-		return nil, errors.Trace(err)
-	}
-	variable.SysVars[variable.TiDBLogBin].Value = tidbLogBin
-
 	if len(cfg.Plugin.Load) > 0 {
 		plugin.InitWatchLoops(dom.GetEtcdClient())
 	}

sessionctx/binloginfo/binloginfo.go

@@ -24,11 +24,9 @@ import (
 	"github.com/pingcap/parser/terror"
 	"github.com/pingcap/tidb-tools/tidb-binlog/node"
 	pumpcli "github.com/pingcap/tidb-tools/tidb-binlog/pump_client"
-	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/metrics"
 	"github.com/pingcap/tidb/sessionctx"
-	"github.com/pingcap/tidb/sessionctx/variable"
 	binlog "github.com/pingcap/tipb/go-binlog"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
@@ -95,15 +93,6 @@ func SetIgnoreError(on bool) {
 	}
 }
 
-// ShouldEnableBinlog returns true if Binlog.AutoMode is false and Binlog.Enable is true, or Binlog.AutoMode is true and tidb_log_bin's value is "1"
-func ShouldEnableBinlog() bool {
-	if config.GetGlobalConfig().Binlog.AutoMode {
-		return variable.SysVars[variable.TiDBLogBin].Value == "1"
-	}
-
-	return config.GetGlobalConfig().Binlog.Enable
-}
-
 // WriteBinlog writes a binlog to Pump.
 func (info *BinlogInfo) WriteBinlog(clusterID uint64) error {
 	skip := atomic.LoadUint32(&skipBinlog)

sessionctx/variable/sysvar.go

@@ -388,7 +388,6 @@ var defaultSysVars = []*SysVar{
 	{ScopeSession, "last_insert_id", ""},
 	{ScopeNone, "innodb_ft_cache_size", "8000000"},
 	{ScopeNone, LogBin, "0"},
-	{ScopeGlobal, TiDBLogBin, "0"},
 	{ScopeGlobal, "innodb_disable_sort_file_cache", "OFF"},
 	{ScopeGlobal, "log_error_verbosity", ""},
 	{ScopeNone, "performance_schema_hosts_size", "100"},
@@ -749,8 +748,6 @@ const (
 	SQLLogBin = "sql_log_bin"
 	// LogBin is the name for 'log_bin' system variable.
 	LogBin = "log_bin"
-	// TiDBLogBin is the name for 'tidb_log_bin' system variable.
-	TiDBLogBin = "tidb_log_bin"
 	// MaxSortLength is the name for 'max_sort_length' system variable.
 	MaxSortLength = "max_sort_length"
 	// MaxSpRecursionDepth is the name for 'max_sp_recursion_depth' system variable.

sessionctx/variable/varsutil.go

@@ -350,7 +350,7 @@ func ValidateSetSystemVar(vars *SessionVars, name string, value string) (string,
 	case WarningCount, ErrorCount:
 		return value, ErrReadOnly.GenWithStackByArgs(name)
 	case GeneralLog, TiDBGeneralLog, AvoidTemporalUpgrade, BigTables, CheckProxyUsers, LogBin,
-		CoreFile, EndMakersInJSON, SQLLogBin, TiDBLogBin, OfflineMode, PseudoSlaveMode, LowPriorityUpdates,
+		CoreFile, EndMakersInJSON, SQLLogBin, OfflineMode, PseudoSlaveMode, LowPriorityUpdates,
 		SkipNameResolve, SQLSafeUpdates, TiDBConstraintCheckInPlace:
 		if strings.EqualFold(value, "ON") || value == "1" {
 			return "1", nil

tidb-server/main.go

@@ -146,10 +146,9 @@ func main() {
 	setupLog()
 	setupTracing() // Should before createServer and after setup config.
 	printInfo()
+	setupBinlogClient()
 	setupMetrics()
 	createStoreAndDomain()
-	// setupBinlogClient should run after bootstrap
-	setupBinlogClient()
 	createServer()
 	signal.SetupSignalHandler(serverShutdown)
 	runServer()
@@ -188,7 +187,7 @@ func createStoreAndDomain() {
 }
 
 func setupBinlogClient() {
-	if !binloginfo.ShouldEnableBinlog() {
+	if !cfg.Binlog.Enable {
 		return
 	}
 
@@ -208,7 +207,7 @@ func setupBinlogClient() {
 	}
 
 	if len(cfg.Binlog.BinlogSocket) == 0 {
-		client, err = pumpcli.NewPumpsClient(cfg.Path, parseDuration(cfg.Binlog.WriteTimeout), securityOption)
+		client, err = pumpcli.NewPumpsClient(cfg.Path, cfg.Binlog.Strategy, parseDuration(cfg.Binlog.WriteTimeout), securityOption)
 	} else {
 		client, err = pumpcli.NewLocalPumpsClient(cfg.Path, cfg.Binlog.BinlogSocket, parseDuration(cfg.Binlog.WriteTimeout), securityOption)
 	}
@@ -453,7 +452,7 @@ func setGlobalVars() {
 
 	variable.SysVars[variable.TIDBMemQuotaQuery].Value = strconv.FormatInt(cfg.MemQuotaQuery, 10)
 	variable.SysVars["lower_case_table_names"].Value = strconv.Itoa(cfg.LowerCaseTableNames)
-	variable.SysVars[variable.LogBin].Value = variable.BoolToIntStr(binloginfo.ShouldEnableBinlog())
+	variable.SysVars[variable.LogBin].Value = variable.BoolToIntStr(config.GetGlobalConfig().Binlog.Enable)
 
 	variable.SysVars[variable.Port].Value = fmt.Sprintf("%d", cfg.Port)
 	variable.SysVars[variable.Socket].Value = cfg.Socket