add a config to manage tso switch

Signed-off-by: Ryan Leung <rleungx@gmail.com>

server/cluster/cluster.go

@@ -403,11 +403,30 @@ func (c *RaftCluster) checkSchedulingService() {
 // checkTSOService checks the TSO service.
 func (c *RaftCluster) checkTSOService() {
 	if c.isAPIServiceMode {
+		if c.opt.GetMicroServiceConfig().IsTSODynamicSwitchingEnabled() {
+			servers, err := discovery.Discover(c.etcdClient, strconv.FormatUint(c.clusterID, 10), constant.TSOServiceName)
+			if err != nil || len(servers) == 0 {
+				if err := c.startTSOJobs(); err != nil {
+					log.Error("failed to start TSO jobs", errs.ZapError(err))
+					return
+				}
+				log.Info("TSO is provided by PD")
+				c.UnsetServiceIndependent(constant.TSOServiceName)
+			} else {
+				if err := c.stopTSOJobs(); err != nil {
+					log.Error("failed to stop TSO jobs", errs.ZapError(err))
+					return
+				}
+				log.Info("TSO is provided by TSO server")
+				if !c.IsServiceIndependent(constant.TSOServiceName) {
+					c.SetServiceIndependent(constant.TSOServiceName)
+				}
+			}
+		}
 		return
 	}
 
 	if err := c.startTSOJobs(); err != nil {
-		// If there is an error, need to wait for the next check.
 		log.Error("failed to start TSO jobs", errs.ZapError(err))
 		return
 	}
@@ -422,6 +441,8 @@ func (c *RaftCluster) runServiceCheckJob() {
 		schedulingTicker.Reset(time.Millisecond)
 	})
 	defer schedulingTicker.Stop()
+	tsoTicker := time.NewTicker(tsoServiceCheckInterval)
+	defer tsoTicker.Stop()
 
 	for {
 		select {
@@ -430,6 +451,8 @@ func (c *RaftCluster) runServiceCheckJob() {
 			return
 		case <-schedulingTicker.C:
 			c.checkSchedulingService()
+		case <-tsoTicker.C:
+			c.checkTSOService()
 		}
 	}
 }

server/config/config.go

@@ -252,7 +252,8 @@ const (
 	minCheckRegionSplitInterval     = 1 * time.Millisecond
 	maxCheckRegionSplitInterval     = 100 * time.Millisecond
 
-	defaultEnableSchedulingFallback = true
+	defaultEnableSchedulingFallback  = true
+	defaultEnableTSODynamicSwitching = false
 )
 
 // Special keys for Labels
@@ -854,13 +855,17 @@ func (c *DRAutoSyncReplicationConfig) adjust(meta *configutil.ConfigMetaData) {
 
 // MicroServiceConfig is the configuration for micro service.
 type MicroServiceConfig struct {
-	EnableSchedulingFallback bool `toml:"enable-scheduling-fallback" json:"enable-scheduling-fallback,string"`
+	EnableSchedulingFallback  bool `toml:"enable-scheduling-fallback" json:"enable-scheduling-fallback,string"`
+	EnableTSODynamicSwitching bool `toml:"enable-tso-dynamic-switching" json:"enable-tso-dynamic-switching,string"`
 }
 
 func (c *MicroServiceConfig) adjust(meta *configutil.ConfigMetaData) {
 	if !meta.IsDefined("enable-scheduling-fallback") {
 		c.EnableSchedulingFallback = defaultEnableSchedulingFallback
 	}
+	if !meta.IsDefined("enable-dynamic-tso") {
+		c.EnableTSODynamicSwitching = defaultEnableTSODynamicSwitching
+	}
 }
 
 // Clone returns a copy of micro service config.
@@ -874,6 +879,11 @@ func (c *MicroServiceConfig) IsSchedulingFallbackEnabled() bool {
 	return c.EnableSchedulingFallback
 }
 
+// IsTSODynamicSwitchingEnabled returns whether to enable TSO dynamic switching.
+func (c *MicroServiceConfig) IsTSODynamicSwitchingEnabled() bool {
+	return c.EnableTSODynamicSwitching
+}
+
 // KeyspaceConfig is the configuration for keyspace management.
 type KeyspaceConfig struct {
 	// PreAlloc contains the keyspace to be allocated during keyspace manager initialization.

server/server.go

@@ -1420,10 +1420,6 @@ func (s *Server) GetRaftCluster() *cluster.RaftCluster {
 // IsServiceIndependent returns whether the service is independent.
 func (s *Server) IsServiceIndependent(name string) bool {
 	if s.mode == APIServiceMode && !s.IsClosed() {
-		// TODO: remove it after we support tso discovery
-		if name == constant.TSOServiceName {
-			return true
-		}
 		return s.cluster.IsServiceIndependent(name)
 	}
 	return false