refactor tso config

Signed-off-by: Ryan Leung <[email protected]>

pkg/member/member.go

@@ -33,7 +33,6 @@ import (
 	"github.com/tikv/pd/pkg/errs"
 	"github.com/tikv/pd/pkg/storage/kv"
 	"github.com/tikv/pd/pkg/utils/etcdutil"
-	"github.com/tikv/pd/server/config"
 	"go.etcd.io/etcd/clientv3"
 	"go.etcd.io/etcd/embed"
 	"go.uber.org/zap"
@@ -250,12 +249,12 @@ func (m *Member) isSameLeader(leader *pdpb.Member) bool {
 }
 
 // MemberInfo initializes the member info.
-func (m *Member) MemberInfo(cfg *config.Config, name string, rootPath string) {
+func (m *Member) MemberInfo(advertiseClientUrls, advertisePeerUrls, name string, rootPath string) {
 	leader := &pdpb.Member{
 		Name:       name,
 		MemberId:   m.ID(),
-		ClientUrls: strings.Split(cfg.AdvertiseClientUrls, ","),
-		PeerUrls:   strings.Split(cfg.AdvertisePeerUrls, ","),
+		ClientUrls: strings.Split(advertiseClientUrls, ","),
+		PeerUrls:   strings.Split(advertisePeerUrls, ","),
 	}
 
 	data, err := leader.Marshal()

pkg/tso/allocator_manager.go

@@ -134,17 +134,18 @@ type AllocatorManager struct {
 func NewAllocatorManager(
 	m *member.Member,
 	rootPath string,
-	cfg config,
+	cfg Config,
+	tlsConfig *grpcutil.TLSConfig,
 	maxResetTSGap func() time.Duration,
 ) *AllocatorManager {
 	allocatorManager := &AllocatorManager{
-		enableLocalTSO:         cfg.IsLocalTSOEnabled(),
+		enableLocalTSO:         cfg.EnableLocalTSO,
 		member:                 m,
 		rootPath:               rootPath,
-		saveInterval:           cfg.GetTSOSaveInterval(),
-		updatePhysicalInterval: cfg.GetTSOUpdatePhysicalInterval(),
+		saveInterval:           cfg.SaveInterval.Duration,
+		updatePhysicalInterval: cfg.UpdatePhysicalInterval.Duration,
 		maxResetTSGap:          maxResetTSGap,
-		securityConfig:         cfg.GetTLSConfig(),
+		securityConfig:         tlsConfig,
 	}
 	allocatorManager.mu.allocatorGroups = make(map[string]*allocatorGroup)
 	allocatorManager.mu.clusterDCLocations = make(map[string]*DCLocationInfo)

pkg/tso/config.go

@@ -15,14 +15,24 @@
 package tso
 
 import (
-	"time"
-
-	"github.com/tikv/pd/pkg/utils/grpcutil"
+	"github.com/tikv/pd/pkg/utils/typeutil"
 )
 
-type config interface {
-	IsLocalTSOEnabled() bool
-	GetTSOSaveInterval() time.Duration
-	GetTSOUpdatePhysicalInterval() time.Duration
-	GetTLSConfig() *grpcutil.TLSConfig
+// Config is the configuration for the TSO.
+type Config struct {
+	// EnableLocalTSO is used to enable the Local TSO Allocator feature,
+	// which allows the PD server to generate Local TSO for certain DC-level transactions.
+	// To make this feature meaningful, user has to set the "zone" label for the PD server
+	// to indicate which DC this PD belongs to.
+	EnableLocalTSO bool `toml:"enable-local-tso" json:"enable-local-tso"`
+
+	// SaveInterval is the interval to save timestamp.
+	SaveInterval typeutil.Duration `toml:"save-interval" json:"save-interval"`
+
+	// The interval to update physical part of timestamp. Usually, this config should not be set.
+	// At most 1<<18 (262144) TSOs can be generated in the interval. The smaller the value, the
+	// more TSOs provided, and at the same time consuming more CPU time.
+	// This config is only valid in 1ms to 10s. If it's configured too long or too short, it will
+	// be automatically clamped to the range.
+	UpdatePhysicalInterval typeutil.Duration `toml:"update-physical-interval" json:"update-physical-interval"`
 }

pkg/tso/tso.go

@@ -391,7 +391,7 @@ func (t *timestampOracle) getTS(leadership *election.Leadership, count uint32, s
 			return pdpb.Timestamp{}, errs.ErrGenerateTimestamp.FastGenByArgs("timestamp in memory has been reset")
 		}
 		if resp.GetLogical() >= maxLogical {
-			log.Warn("logical part outside of max logical interval, please check ntp time, or adjust config item `tso-update-physical-interval`",
+			log.Warn("logical part outside of max logical interval, please check ntp time, or adjust config item `tso.update-physical-interval`",
 				zap.Reflect("response", resp),
 				zap.Int("retry-count", i), errs.ZapError(errs.ErrLogicOverflow))
 			tsoCounter.WithLabelValues("logical_overflow", t.dcLocation).Inc()

server/api/member_test.go

@@ -45,7 +45,7 @@ func TestMemberTestSuite(t *testing.T) {
 
 func (suite *memberTestSuite) SetupSuite() {
 	suite.cfgs, suite.servers, suite.clean = mustNewCluster(suite.Require(), 3, func(cfg *config.Config) {
-		cfg.EnableLocalTSO = true
+		cfg.TSOConfig.EnableLocalTSO = true
 		cfg.Labels = map[string]string{
 			config.ZoneLabel: "dc-1",
 		}

server/api/tso_test.go

@@ -39,7 +39,7 @@ func TestTSOTestSuite(t *testing.T) {
 func (suite *tsoTestSuite) SetupSuite() {
 	re := suite.Require()
 	suite.svr, suite.cleanup = mustNewServer(re, func(cfg *config.Config) {
-		cfg.EnableLocalTSO = true
+		cfg.TSOConfig.EnableLocalTSO = true
 		cfg.Labels[config.ZoneLabel] = "dc-1"
 	})
 	server.MustWaitLeader(re, []*server.Server{suite.svr})

server/config/config.go

@@ -30,6 +30,7 @@ import (
 	"github.com/tikv/pd/pkg/core/storelimit"
 	"github.com/tikv/pd/pkg/encryption"
 	"github.com/tikv/pd/pkg/errs"
+	"github.com/tikv/pd/pkg/tso"
 	"github.com/tikv/pd/pkg/utils/grpcutil"
 	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/tikv/pd/pkg/utils/metricutil"
@@ -111,6 +112,8 @@ type Config struct {
 
 	PDServerCfg PDServerConfig `toml:"pd-server" json:"pd-server"`
 
+	TSOConfig tso.Config `toml:"tso" json:"tso"`
+
 	ClusterVersion semver.Version `toml:"cluster-version" json:"cluster-version"`
 
 	// Labels indicates the labels set for **this** PD server. The labels describe some specific properties
@@ -409,6 +412,27 @@ func (c *Config) Parse(arguments []string) error {
 				c.Log.Level = c.LogLevelDeprecated
 			}
 		}
+		if c.EnableLocalTSO {
+			msg := fmt.Sprintf("enable-local-tso in %s is deprecated, use [tso.enable-local-tso] instead", c.configFile)
+			c.WarningMsgs = append(c.WarningMsgs, msg)
+			if !meta.IsDefined("tso", "enable-local-tso") {
+				c.TSOConfig.EnableLocalTSO = c.EnableLocalTSO
+			}
+		}
+		if c.TSOSaveInterval.Duration != defaultTSOSaveInterval {
+			msg := fmt.Sprintf("tso-save-interval in %s is deprecated, use [tso.save-interval] instead", c.configFile)
+			c.WarningMsgs = append(c.WarningMsgs, msg)
+			if !meta.IsDefined("tso", "save-interval") {
+				c.TSOConfig.SaveInterval = c.TSOSaveInterval
+			}
+		}
+		if c.TSOUpdatePhysicalInterval.Duration != defaultTSOUpdatePhysicalInterval {
+			msg := fmt.Sprintf("tso-update-physical-interval in %s is deprecated, use [tso.update-physical-interval] instead", c.configFile)
+			c.WarningMsgs = append(c.WarningMsgs, msg)
+			if !meta.IsDefined("tso", "update-physical-interval") {
+				c.TSOConfig.UpdatePhysicalInterval = c.TSOUpdatePhysicalInterval
+			}
+		}
 		if meta.IsDefined("schedule", "disable-raft-learner") {
 			msg := fmt.Sprintf("disable-raft-learner in %s is deprecated", c.configFile)
 			c.WarningMsgs = append(c.WarningMsgs, msg)
@@ -589,6 +613,7 @@ func (c *Config) Adjust(meta *toml.MetaData, reloading bool) error {
 	}
 
 	c.adjustLog(configMetaData.Child("log"))
+	c.adjustTSOConfig(configMetaData.Child("tso"))
 	adjustDuration(&c.HeartbeatStreamBindInterval, defaultHeartbeatStreamRebindInterval)
 
 	adjustDuration(&c.LeaderPriorityCheckInterval, defaultLeaderPriorityCheckInterval)
@@ -619,6 +644,21 @@ func (c *Config) adjustLog(meta *configMetaData) {
 	}
 }
 
+func (c *Config) adjustTSOConfig(meta *configMetaData) {
+	if !meta.IsDefined("save-interval") {
+		c.TSOConfig.SaveInterval.Duration = defaultTSOSaveInterval
+	}
+
+	if !meta.IsDefined("update-physical-interval") {
+		c.TSOConfig.UpdatePhysicalInterval.Duration = defaultTSOUpdatePhysicalInterval
+	}
+	if c.TSOConfig.UpdatePhysicalInterval.Duration > maxTSOUpdatePhysicalInterval {
+		c.TSOConfig.UpdatePhysicalInterval.Duration = maxTSOUpdatePhysicalInterval
+	} else if c.TSOConfig.UpdatePhysicalInterval.Duration < minTSOUpdatePhysicalInterval {
+		c.TSOConfig.UpdatePhysicalInterval.Duration = minTSOUpdatePhysicalInterval
+	}
+}
+
 // Clone returns a cloned configuration.
 func (c *Config) Clone() *Config {
 	cfg := *c
@@ -1306,19 +1346,14 @@ func (c *Config) GetConfigFile() string {
 	return c.configFile
 }
 
-// IsLocalTSOEnabled returns if the local TSO is enabled.
-func (c *Config) IsLocalTSOEnabled() bool {
-	return c.EnableLocalTSO
-}
-
 // GetTSOUpdatePhysicalInterval returns TSO update physical interval.
 func (c *Config) GetTSOUpdatePhysicalInterval() time.Duration {
-	return c.TSOUpdatePhysicalInterval.Duration
+	return c.TSOConfig.UpdatePhysicalInterval.Duration
 }
 
 // GetTSOSaveInterval returns TSO save interval.
 func (c *Config) GetTSOSaveInterval() time.Duration {
-	return c.TSOSaveInterval.Duration
+	return c.TSOConfig.SaveInterval.Duration
 }
 
 // GetTLSConfig returns the TLS config.

server/config/config_test.go

@@ -281,6 +281,37 @@ tso-update-physical-interval = "15s"
 	re.NoError(err)
 
 	re.Equal(maxTSOUpdatePhysicalInterval, cfg.TSOUpdatePhysicalInterval.Duration)
+
+	// TSO config testings
+	cfgData = `
+[tso]
+update-physical-interval = "3m"
+save-interval = "1m"
+enable-local-tso = true
+`
+	cfg = NewConfig()
+	meta, err = toml.Decode(cfgData, &cfg)
+	re.NoError(err)
+	err = cfg.Adjust(&meta, false)
+	re.NoError(err)
+
+	re.Equal(maxTSOUpdatePhysicalInterval, cfg.TSOConfig.UpdatePhysicalInterval.Duration)
+	re.Equal(1*time.Minute, cfg.TSOConfig.SaveInterval.Duration)
+	re.True(cfg.TSOConfig.EnableLocalTSO)
+
+	cfgData = `
+[tso]
+update-physical-interval = "1ns"
+	`
+	cfg = NewConfig()
+	meta, err = toml.Decode(cfgData, &cfg)
+	re.NoError(err)
+	err = cfg.Adjust(&meta, false)
+	re.NoError(err)
+
+	re.Equal(minTSOUpdatePhysicalInterval, cfg.TSOConfig.UpdatePhysicalInterval.Duration)
+	re.Equal(defaultTSOSaveInterval, cfg.TSOConfig.SaveInterval.Duration)
+	re.False(cfg.TSOConfig.EnableLocalTSO)
 }
 
 func TestMigrateFlags(t *testing.T) {

server/server.go

@@ -358,7 +358,7 @@ func (s *Server) startServer(ctx context.Context) error {
 	serverInfo.WithLabelValues(versioninfo.PDReleaseVersion, versioninfo.PDGitHash).Set(float64(time.Now().Unix()))
 
 	s.rootPath = path.Join(pdRootPath, strconv.FormatUint(s.clusterID, 10))
-	s.member.MemberInfo(s.cfg, s.Name(), s.rootPath)
+	s.member.MemberInfo(s.cfg.AdvertiseClientUrls, s.cfg.AdvertisePeerUrls, s.Name(), s.rootPath)
 	s.member.SetMemberDeployPath(s.member.ID())
 	s.member.SetMemberBinaryVersion(s.member.ID(), versioninfo.PDReleaseVersion)
 	s.member.SetMemberGitHash(s.member.ID(), versioninfo.PDGitHash)
@@ -370,17 +370,17 @@ func (s *Server) startServer(ctx context.Context) error {
 		Member:    s.member.MemberValue(),
 	})
 	s.tsoAllocatorManager = tso.NewAllocatorManager(
-		s.member, s.rootPath, s.cfg,
+		s.member, s.rootPath, s.cfg.TSOConfig, s.cfg.GetTLSConfig(),
 		func() time.Duration { return s.persistOptions.GetMaxResetTSGap() })
 	// Set up the Global TSO Allocator here, it will be initialized once the PD campaigns leader successfully.
 	s.tsoAllocatorManager.SetUpAllocator(ctx, tso.GlobalDCLocation, s.member.GetLeadership())
 	// When disabled the Local TSO, we should clean up the Local TSO Allocator's meta info written in etcd if it exists.
-	if !s.cfg.EnableLocalTSO {
+	if !s.cfg.TSOConfig.EnableLocalTSO {
 		if err = s.tsoAllocatorManager.CleanUpDCLocation(); err != nil {
 			return err
 		}
 	}
-	if zone, exist := s.cfg.Labels[config.ZoneLabel]; exist && zone != "" && s.cfg.EnableLocalTSO {
+	if zone, exist := s.cfg.Labels[config.ZoneLabel]; exist && zone != "" && s.cfg.TSOConfig.EnableLocalTSO {
 		if err = s.tsoAllocatorManager.SetLocalTSOConfig(zone); err != nil {
 			return err
 		}
@@ -825,7 +825,7 @@ func (s *Server) GetServiceMiddlewareConfig() *config.ServiceMiddlewareConfig {
 
 // SetEnableLocalTSO sets enable-local-tso flag of Server. This function only for test.
 func (s *Server) SetEnableLocalTSO(enableLocalTSO bool) {
-	s.cfg.EnableLocalTSO = enableLocalTSO
+	s.cfg.TSOConfig.EnableLocalTSO = enableLocalTSO
 }
 
 // GetConfig gets the config information.

server/testutil.go

@@ -76,6 +76,7 @@ func NewTestSingleConfig(c *assertutil.Checker) *config.Config {
 		TSOSaveInterval: typeutil.NewDuration(200 * time.Millisecond),
 	}
 
+	cfg.TSOConfig.SaveInterval = typeutil.NewDuration(200 * time.Millisecond)
 	cfg.AdvertiseClientUrls = cfg.ClientUrls
 	cfg.AdvertisePeerUrls = cfg.PeerUrls
 	cfg.DataDir, _ = os.MkdirTemp("/tmp", "test_pd")

tests/client/client_test.go

@@ -260,7 +260,7 @@ func TestTSOAllocatorLeader(t *testing.T) {
 	}
 	dcLocationNum := len(dcLocationConfig)
 	cluster, err := tests.NewTestCluster(ctx, dcLocationNum, func(conf *config.Config, serverName string) {
-		conf.EnableLocalTSO = true
+		conf.TSOConfig.EnableLocalTSO = true
 		conf.Labels[config.ZoneLabel] = dcLocationConfig[serverName]
 	})
 	re.NoError(err)
@@ -421,7 +421,7 @@ func TestGlobalAndLocalTSO(t *testing.T) {
 	}
 	dcLocationNum := len(dcLocationConfig)
 	cluster, err := tests.NewTestCluster(ctx, dcLocationNum, func(conf *config.Config, serverName string) {
-		conf.EnableLocalTSO = true
+		conf.TSOConfig.EnableLocalTSO = true
 		conf.Labels[config.ZoneLabel] = dcLocationConfig[serverName]
 	})
 	re.NoError(err)
@@ -435,7 +435,7 @@ func TestGlobalAndLocalTSO(t *testing.T) {
 
 	// Join a new dc-location
 	pd4, err := cluster.Join(ctx, func(conf *config.Config, serverName string) {
-		conf.EnableLocalTSO = true
+		conf.TSOConfig.EnableLocalTSO = true
 		conf.Labels[config.ZoneLabel] = "dc-4"
 	})
 	re.NoError(err)

tests/server/member/member_test.go

@@ -53,7 +53,7 @@ func TestMemberDelete(t *testing.T) {
 	}
 	dcLocationNum := len(dcLocationConfig)
 	cluster, err := tests.NewTestCluster(ctx, dcLocationNum, func(conf *config.Config, serverName string) {
-		conf.EnableLocalTSO = true
+		conf.TSOConfig.EnableLocalTSO = true
 		conf.Labels[config.ZoneLabel] = dcLocationConfig[serverName]
 	})
 	defer cluster.Destroy()

tests/server/tso/allocator_test.go

@@ -47,7 +47,7 @@ func TestAllocatorLeader(t *testing.T) {
 	dcLocationNum := len(dcLocationConfig)
 	cluster, err := tests.NewTestCluster(ctx, dcLocationNum*2, func(conf *config.Config, serverName string) {
 		if zoneLabel, ok := dcLocationConfig[serverName]; ok {
-			conf.EnableLocalTSO = true
+			conf.TSOConfig.EnableLocalTSO = true
 			conf.Labels[config.ZoneLabel] = zoneLabel
 		}
 	})
@@ -119,7 +119,7 @@ func TestPriorityAndDifferentLocalTSO(t *testing.T) {
 	}
 	dcLocationNum := len(dcLocationConfig)
 	cluster, err := tests.NewTestCluster(ctx, dcLocationNum, func(conf *config.Config, serverName string) {
-		conf.EnableLocalTSO = true
+		conf.TSOConfig.EnableLocalTSO = true
 		conf.Labels[config.ZoneLabel] = dcLocationConfig[serverName]
 	})
 	defer cluster.Destroy()
@@ -133,7 +133,7 @@ func TestPriorityAndDifferentLocalTSO(t *testing.T) {
 
 	// Join a new dc-location
 	pd4, err := cluster.Join(ctx, func(conf *config.Config, serverName string) {
-		conf.EnableLocalTSO = true
+		conf.TSOConfig.EnableLocalTSO = true
 		conf.Labels[config.ZoneLabel] = "dc-4"
 	})
 	re.NoError(err)