binlog: update binlog enable config && add tidb_log_bin system variab…

…le (#9625)

config/config.go

@@ -250,7 +250,7 @@ type TiKVClient struct {
 
 // Binlog is the config for binlog.
 type Binlog struct {
-	Enable       bool   `toml:"enable" json:"enable"`
+	Enable       string `toml:"enable" json:"enable"`
 	WriteTimeout string `toml:"write-timeout" json:"write-timeout"`
 	// If IgnoreError is true, when writing binlog meets error, TiDB would
 	// ignore the error.
@@ -343,6 +343,7 @@ var defaultConf = Config{
 		BatchWaitSize:     8,
 	},
 	Binlog: Binlog{
+		Enable:       "auto",
 		WriteTimeout: "15s",
 	},
 }

config/config.toml.example

@@ -258,8 +258,9 @@ enabled = true
 capacity = 2048000
 
 [binlog]
-# enable to write binlog.
-enable = false
+# Enable to write binlog. Values can be "on", "off" or "auto".
+# If value is "auto", will enable binlog according to the system variables 'tidb_log_bin'.
+enable = "auto"
 
 # WriteTimeout specifies how long it will wait for writing binlog to pump.
 write-timeout = "15s"

config/config_test.go

@@ -33,7 +33,7 @@ func TestT(t *testing.T) {
 
 func (s *testConfigSuite) TestConfig(c *C) {
 	conf := new(Config)
-	conf.Binlog.Enable = true
+	conf.Binlog.Enable = "auto"
 	conf.Binlog.IgnoreError = true
 	conf.TiKVClient.CommitTimeout = "10s"
 	conf.CheckMb4ValueInUtf8 = true
@@ -54,7 +54,7 @@ max-batch-size=128
 	c.Assert(conf.Load(configFile), IsNil)
 
 	// 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.Enable, Equals, "auto")
 
 	c.Assert(conf.TiKVClient.CommitTimeout, Equals, "41s")
 	c.Assert(conf.TiKVClient.MaxBatchSize, Equals, uint(128))

executor/set_test.go

@@ -239,8 +239,13 @@ 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(config.GetGlobalConfig().Binlog.Enable)))
-	tk.MustQuery(`select @@log_bin;`).Check(testkit.Rows(variable.BoolToIntStr(config.GetGlobalConfig().Binlog.Enable)))
+	tk.MustQuery(`select @@global.log_bin;`).Check(testkit.Rows(variable.BoolToIntStr(config.GetGlobalConfig().Binlog.Enable == "on")))
+	tk.MustQuery(`select @@log_bin;`).Check(testkit.Rows(variable.BoolToIntStr(config.GetGlobalConfig().Binlog.Enable == "on")))
+
+	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.MustExec("set @@tidb_general_log = 1")
 	tk.MustExec("set @@tidb_general_log = 0")

session/session.go

@@ -1319,6 +1319,13 @@ func BootstrapSession(store kv.Storage) (*domain.Domain, error) {
 		return nil, errors.Trace(err)
 	}
 
+	// get global system 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,9 +24,11 @@ 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"
@@ -93,6 +95,11 @@ func SetIgnoreError(on bool) {
 	}
 }
 
+// ShouldEnableBinlog returns true if binlog.enable is "on", or binlog.enable is "auto" and tidb_log_bin's value is "1"
+func ShouldEnableBinlog() bool {
+	return config.GetGlobalConfig().Binlog.Enable == "on" || (config.GetGlobalConfig().Binlog.Enable == "auto" && variable.SysVars[variable.TiDBLogBin].Value == "1")
+}
+
 // WriteBinlog writes a binlog to Pump.
 func (info *BinlogInfo) WriteBinlog(clusterID uint64) error {
 	skip := atomic.LoadUint32(&skipBinlog)

sessionctx/variable/sysvar.go

@@ -388,6 +388,7 @@ 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"},
@@ -748,6 +749,8 @@ 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, OfflineMode, PseudoSlaveMode, LowPriorityUpdates,
+		CoreFile, EndMakersInJSON, SQLLogBin, TiDBLogBin, OfflineMode, PseudoSlaveMode, LowPriorityUpdates,
 		SkipNameResolve, SQLSafeUpdates, TiDBConstraintCheckInPlace:
 		if strings.EqualFold(value, "ON") || value == "1" {
 			return "1", nil

tidb-server/main.go

@@ -100,7 +100,7 @@ var (
 	port             = flag.String(nmPort, "4000", "tidb server port")
 	cors             = flag.String(nmCors, "", "tidb server allow cors origin")
 	socket           = flag.String(nmSocket, "", "The socket file to use for connection.")
-	enableBinlog     = flagBoolean(nmEnableBinlog, false, "enable generate binlog")
+	enableBinlog     = flag.String(nmEnableBinlog, "auto", "enable generate binlog")
 	runDDL           = flagBoolean(nmRunDDL, true, "run ddl worker on this tidb-server")
 	ddlLease         = flag.String(nmDdlLease, "45s", "schema lease duration, very dangerous to change only if you know what you do")
 	tokenLimit       = flag.Int(nmTokenLimit, 1000, "the limit of concurrent executed sessions")
@@ -146,9 +146,10 @@ 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()
@@ -187,7 +188,7 @@ func createStoreAndDomain() {
 }
 
 func setupBinlogClient() {
-	if !cfg.Binlog.Enable {
+	if !binloginfo.ShouldEnableBinlog() {
 		return
 	}
 
@@ -452,7 +453,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(config.GetGlobalConfig().Binlog.Enable)
+	variable.SysVars[variable.LogBin].Value = variable.BoolToIntStr(binloginfo.ShouldEnableBinlog())
 
 	variable.SysVars[variable.Port].Value = fmt.Sprintf("%d", cfg.Port)
 	variable.SysVars[variable.Socket].Value = cfg.Socket