tso/local: remove local tso completely

client/client.go

@@ -568,7 +568,7 @@ func (c *client) GetTSAsync(ctx context.Context) tso.TSFuture {
 
 // GetLocalTSAsync implements the TSOClient interface.
 //
-// Deprecated: Local TSO will be completely removed in the future. Currently, regardless of the
+// Deprecated: Currently, regardless of the
 // parameters passed in, this method will default to returning the global TSO.
 func (c *client) GetLocalTSAsync(ctx context.Context, _ string) tso.TSFuture {
 	return c.GetTSAsync(ctx)
@@ -582,7 +582,7 @@ func (c *client) GetTS(ctx context.Context) (physical int64, logical int64, err
 
 // GetLocalTS implements the TSOClient interface.
 //
-// Deprecated: Local TSO will be completely removed in the future. Currently, regardless of the
+// Deprecated: Currently, regardless of the
 // parameters passed in, this method will default to returning the global TSO.
 func (c *client) GetLocalTS(ctx context.Context, _ string) (physical int64, logical int64, err error) {
 	return c.GetTS(ctx)

client/clients/tso/tso_client.go

@@ -58,13 +58,13 @@ type Client interface {
 
 	// GetLocalTS gets a local timestamp from PD or TSO microservice.
 	//
-	// Deprecated: Local TSO will be completely removed in the future. Currently, regardless of the
-	// parameters passed in, this method will default to returning the global TSO.
+	// Deprecated: Currently, regardless of the parameters passed in, this
+	// method will default to returning the global TSO.
 	GetLocalTS(ctx context.Context, _ string) (int64, int64, error)
 	// GetLocalTSAsync gets a local timestamp from PD or TSO microservice, without block the caller.
 	//
-	// Deprecated: Local TSO will be completely removed in the future. Currently, regardless of the
-	// parameters passed in, this method will default to returning the global TSO.
+	// Deprecated: Currently, regardless of the parameters passed in, this
+	// method will default to returning the global TSO.
 	GetLocalTSAsync(ctx context.Context, _ string) TSFuture
 }
 

pkg/mcs/tso/server/config.go

@@ -64,10 +64,7 @@ type Config struct {
 	// the primary/leader again. Etcd only supports seconds TTL, so here is second too.
 	LeaderLease int64 `toml:"lease" json:"lease"`
 
-	// 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.
+	// Deprecated
 	EnableLocalTSO bool `toml:"enable-local-tso" json:"enable-local-tso"`
 
 	// TSOSaveInterval is the interval to save timestamp.

pkg/mcs/tso/server/config_test.go

@@ -36,7 +36,6 @@ func TestConfigBasic(t *testing.T) {
 	re.Equal(defaultBackendEndpoints, cfg.BackendEndpoints)
 	re.Equal(defaultListenAddr, cfg.ListenAddr)
 	re.Equal(constant.DefaultLeaderLease, cfg.LeaderLease)
-	re.False(cfg.EnableLocalTSO)
 	re.True(cfg.EnableGRPCGateway)
 	re.Equal(defaultTSOSaveInterval, cfg.TSOSaveInterval.Duration)
 	re.Equal(defaultTSOUpdatePhysicalInterval, cfg.TSOUpdatePhysicalInterval.Duration)
@@ -48,7 +47,6 @@ func TestConfigBasic(t *testing.T) {
 	cfg.ListenAddr = "test-listen-addr"
 	cfg.AdvertiseListenAddr = "test-advertise-listen-addr"
 	cfg.LeaderLease = 123
-	cfg.EnableLocalTSO = true
 	cfg.TSOSaveInterval.Duration = time.Duration(10) * time.Second
 	cfg.TSOUpdatePhysicalInterval.Duration = time.Duration(100) * time.Millisecond
 	cfg.MaxResetTSGap.Duration = time.Duration(1) * time.Hour
@@ -58,7 +56,6 @@ func TestConfigBasic(t *testing.T) {
 	re.Equal("test-listen-addr", cfg.GetListenAddr())
 	re.Equal("test-advertise-listen-addr", cfg.GetAdvertiseListenAddr())
 	re.Equal(int64(123), cfg.GetLeaderLease())
-	re.True(cfg.EnableLocalTSO)
 	re.Equal(time.Duration(10)*time.Second, cfg.TSOSaveInterval.Duration)
 	re.Equal(time.Duration(100)*time.Millisecond, cfg.TSOUpdatePhysicalInterval.Duration)
 	re.Equal(time.Duration(1)*time.Hour, cfg.MaxResetTSGap.Duration)
@@ -74,7 +71,6 @@ name = "tso-test-name"
 data-dir = "/var/lib/tso"
 enable-grpc-gateway = false
 lease = 123
-enable-local-tso = true
 tso-save-interval = "10s"
 tso-update-physical-interval = "100ms"
 max-gap-reset-ts = "1h"
@@ -92,7 +88,6 @@ max-gap-reset-ts = "1h"
 	re.Equal("test-advertise-listen-addr", cfg.GetAdvertiseListenAddr())
 	re.Equal("/var/lib/tso", cfg.DataDir)
 	re.Equal(int64(123), cfg.GetLeaderLease())
-	re.True(cfg.EnableLocalTSO)
 	re.Equal(time.Duration(10)*time.Second, cfg.TSOSaveInterval.Duration)
 	re.Equal(time.Duration(100)*time.Millisecond, cfg.TSOUpdatePhysicalInterval.Duration)
 	re.Equal(time.Duration(1)*time.Hour, cfg.MaxResetTSGap.Duration)

pkg/mcs/tso/server/server.go

@@ -272,15 +272,6 @@ func (s *Server) GetTSOAllocatorManager(keyspaceGroupID uint32) (*tso.AllocatorM
 	return s.keyspaceGroupManager.GetAllocatorManager(keyspaceGroupID)
 }
 
-// IsLocalRequest checks if the forwarded host is the current host
-func (*Server) IsLocalRequest(forwardedHost string) bool {
-	// TODO: Check if the forwarded host is the current host.
-	// The logic is depending on etcd service mode -- if the TSO service
-	// uses the embedded etcd, check against ClientUrls; otherwise check
-	// against the cluster membership.
-	return forwardedHost == ""
-}
-
 // ValidateInternalRequest checks if server is closed, which is used to validate
 // the gRPC communication between TSO servers internally.
 // TODO: Check if the sender is from the global TSO allocator

pkg/storage/endpoint/tso.go

@@ -37,9 +37,9 @@ type TSOStorage interface {
 
 var _ TSOStorage = (*StorageEndpoint)(nil)
 
-// LoadTimestamp will get all time windows of Local/Global TSOs from etcd and return the biggest one.
-// For the Global TSO, loadTimestamp will get all Local and Global TSO time windows persisted in etcd and choose the biggest one.
-// For the Local TSO, loadTimestamp will only get its own dc-location time window persisted before.
+// LoadTimestamp will get all time windows of Global TSOs from etcd and return the biggest one.
+// TODO: Due to local TSO is deprecated, maybe we do not need to load timestamp
+// by prefix, we can just load the timestamp by the key.
 func (se *StorageEndpoint) LoadTimestamp(prefix string) (time.Time, error) {
 	prefixEnd := clientv3.GetPrefixRangeEnd(prefix)
 	keys, values, err := se.LoadRange(prefix, prefixEnd, 0)

pkg/tso/allocator_manager.go

@@ -17,7 +17,6 @@ package tso
 import (
 	"context"
 	"math"
-	"path"
 	"runtime/trace"
 	"strconv"
 	"sync"
@@ -43,8 +42,6 @@ const (
 	checkStep                   = time.Minute
 	patrolStep                  = time.Second
 	defaultAllocatorLeaderLease = 3
-	localTSOAllocatorEtcdPrefix = "lta"
-	localTSOSuffixEtcdPrefix    = "lts"
 )
 
 var (
@@ -217,17 +214,6 @@ func (am *AllocatorManager) getGroupIDStr() string {
 	return strconv.FormatUint(uint64(am.kgID), 10)
 }
 
-// GetTimestampPath returns the timestamp path in etcd.
-func (am *AllocatorManager) GetTimestampPath() string {
-	if am == nil {
-		return ""
-	}
-
-	am.mu.RLock()
-	defer am.mu.RUnlock()
-	return path.Join(am.rootPath, am.mu.allocatorGroup.allocator.GetTimestampPath())
-}
-
 // tsoAllocatorLoop is used to run the TSO Allocator updating daemon.
 func (am *AllocatorManager) tsoAllocatorLoop() {
 	defer logutil.LogPanic()

pkg/tso/global_allocator.go

@@ -47,14 +47,6 @@ type Allocator interface {
 	IsInitialize() bool
 	// UpdateTSO is used to update the TSO in memory and the time window in etcd.
 	UpdateTSO() error
-	// GetTimestampPath returns the timestamp path in etcd, which is:
-	// 1. for the default keyspace group:
-	//     a. timestamp in /pd/{cluster_id}/timestamp
-	//     b. lta/{dc-location}/timestamp in /pd/{cluster_id}/lta/{dc-location}/timestamp
-	// 1. for the non-default keyspace groups:
-	//     a. {group}/gts/timestamp in /ms/{cluster_id}/tso/{group}/gta/timestamp
-	//     b. {group}/lts/{dc-location}/timestamp in /ms/{cluster_id}/tso/{group}/lta/{dc-location}/timestamp
-	GetTimestampPath() string
 	// SetTSO sets the physical part with given TSO. It's mainly used for BR restore.
 	// Cannot set the TSO smaller than now in any case.
 	// if ignoreSmaller=true, if input ts is smaller than current, ignore silently, else return error
@@ -68,6 +60,8 @@ type Allocator interface {
 }
 
 // GlobalTSOAllocator is the global single point TSO allocator.
+// TODO: Local TSO allocator is deprecated now, we can update the name to
+// TSOAllocator and remove the `Global` concept.
 type GlobalTSOAllocator struct {
 	ctx    context.Context
 	cancel context.CancelFunc
@@ -132,19 +126,9 @@ func (gta *GlobalTSOAllocator) getGroupID() uint32 {
 	return gta.am.getGroupID()
 }
 
-// GetTimestampPath returns the timestamp path in etcd.
-func (gta *GlobalTSOAllocator) GetTimestampPath() string {
-	if gta == nil || gta.timestampOracle == nil {
-		return ""
-	}
-	return gta.timestampOracle.GetTimestampPath()
-}
-
 // Initialize will initialize the created global TSO allocator.
 func (gta *GlobalTSOAllocator) Initialize(int) error {
 	gta.tsoAllocatorRoleGauge.Set(1)
-	// The suffix of a Global TSO should always be 0.
-	gta.timestampOracle.suffix = 0
 	return gta.timestampOracle.SyncTimestamp()
 }
 

pkg/tso/keyspace_group_manager.go

@@ -66,7 +66,7 @@ const (
 type state struct {
 	syncutil.RWMutex
 	// ams stores the allocator managers of the keyspace groups. Each keyspace group is
-	// assigned with an allocator manager managing its global/local tso allocators.
+	// assigned with an allocator manager managing its global tso allocators.
 	// Use a fixed size array to maximize the efficiency of concurrent access to
 	// different keyspace groups for tso service.
 	ams [constant.MaxKeyspaceGroupCountInUse]*AllocatorManager
@@ -790,8 +790,7 @@ func (kgm *KeyspaceGroupManager) updateKeyspaceGroup(group *endpoint.KeyspaceGro
 	am := NewAllocatorManager(kgm.ctx, group.ID, participant, tsRootPath, storage, kgm.cfg)
 	am.startGlobalAllocatorLoop()
 	log.Info("created allocator manager",
-		zap.Uint32("keyspace-group-id", group.ID),
-		zap.String("timestamp-path", am.GetTimestampPath()))
+		zap.Uint32("keyspace-group-id", group.ID))
 	kgm.Lock()
 	group.KeyspaceLookupTable = make(map[uint32]struct{})
 	for _, kid := range group.Keyspaces {
@@ -1517,7 +1516,6 @@ func (kgm *KeyspaceGroupManager) deletedGroupCleaner() {
 			log.Info("delete the keyspace group tso key",
 				zap.Uint32("keyspace-group-id", groupID))
 			// Clean up the remaining TSO keys.
-			// TODO: support the Local TSO Allocator clean up.
 			err := kgm.tsoSvcStorage.DeleteTimestamp(
 				keypath.TimestampPath(
 					keypath.KeyspaceGroupGlobalTSPath(groupID),

pkg/tso/keyspace_group_manager_test.go

@@ -87,7 +87,6 @@ func (suite *keyspaceGroupManagerTestSuite) createConfig() *TestServiceConfig {
 		ListenAddr:                addr,
 		AdvertiseListenAddr:       addr,
 		LeaderLease:               constant.DefaultLeaderLease,
-		LocalTSOEnabled:           false,
 		TSOUpdatePhysicalInterval: 50 * time.Millisecond,
 		TSOSaveInterval:           time.Duration(constant.DefaultLeaderLease) * time.Second,
 		MaxResetTSGap:             time.Hour * 24,

pkg/tso/testutil.go

@@ -29,7 +29,6 @@ type TestServiceConfig struct {
 	ListenAddr                string              // Address the service listens on.
 	AdvertiseListenAddr       string              // Address the service advertises to the clients.
 	LeaderLease               int64               // Leader lease.
-	LocalTSOEnabled           bool                // Whether local TSO is enabled.
 	TSOUpdatePhysicalInterval time.Duration       // Interval to update TSO in physical storage.
 	TSOSaveInterval           time.Duration       // Interval to save TSO to physical storage.
 	MaxResetTSGap             time.Duration       // Maximum gap to reset TSO.

pkg/tso/tso.go

@@ -75,7 +75,6 @@
 	tsoMux *tsoObject
 	// last timestamp window stored in etcd
 	lastSavedTime atomic.Value // stored as time.Time
-	suffix        int
 
 	// pre-initialized metrics
 	metrics *tsoMetrics
@@ -106,7 +105,7 @@
 }
 
 // generateTSO will add the TSO's logical part with the given count and returns the new TSO result.
 func (t *timestampOracle) generateTSO(ctx context.Context, count int64, suffixBits int) (physical int64, logical int64, lastUpdateTime time.Time) {
 	defer trace.StartRegion(ctx, "timestampOracle.generateTSO").End()
 	t.tsoMux.Lock()
 	defer t.tsoMux.Unlock()
@@ -116,9 +115,6 @@
 	physical = t.tsoMux.physical.UnixNano() / int64(time.Millisecond)
 	t.tsoMux.logical += count
 	logical = t.tsoMux.logical
-	if suffixBits > 0 && t.suffix >= 0 {
-		logical = t.calibrateLogical(logical, suffixBits)
-	}
 	// Return the last update time
 	lastUpdateTime = t.tsoMux.updateTime
 	t.tsoMux.updateTime = time.Now()
@@ -133,28 +129,6 @@
 	return last.(time.Time)
 }
 
-// Because the Local TSO in each Local TSO Allocator is independent, so they are possible
-// to be the same at sometimes, to avoid this case, we need to use the logical part of the
-// Local TSO to do some differentiating work.
-// For example, we have three DCs: dc-1, dc-2 and dc-3. The bits of suffix is defined by
-// the const suffixBits. Then, for dc-2, the suffix may be 1 because it's persisted
-// in etcd with the value of 1.
-// Once we get a normal TSO like this (18 bits): xxxxxxxxxxxxxxxxxx. We will make the TSO's
-// low bits of logical part from each DC looks like:
-//
-//	global: xxxxxxxxxx00000000
-//	  dc-1: xxxxxxxxxx00000001
-//	  dc-2: xxxxxxxxxx00000010
-//	  dc-3: xxxxxxxxxx00000011
-func (t *timestampOracle) calibrateLogical(rawLogical int64, suffixBits int) int64 {
-	return rawLogical<<suffixBits + int64(t.suffix)
-}
-
-// GetTimestampPath returns the timestamp path in etcd.
-func (t *timestampOracle) GetTimestampPath() string {
-	return keypath.TimestampPath(t.tsPath)
-}
-
 // SyncTimestamp is used to synchronize the timestamp.
 func (t *timestampOracle) SyncTimestamp() error {
 	log.Info("start to sync timestamp", logutil.CondUint32("keyspace-group-id", t.keyspaceGroupID, t.keyspaceGroupID > 0))
@@ -209,7 +183,7 @@
 	})
 	save := next.Add(t.saveInterval)
 	start := time.Now()
-	if err = t.storage.SaveTimestamp(t.GetTimestampPath(), save); err != nil {
+	if err = t.storage.SaveTimestamp(keypath.TimestampPath(t.tsPath), save); err != nil {
 		t.metrics.errSaveSyncTSEvent.Inc()
 		return err
 	}
@@ -277,7 +251,7 @@
 	if typeutil.SubRealTimeByWallClock(t.getLastSavedTime(), nextPhysical) <= UpdateTimestampGuard {
 		save := nextPhysical.Add(t.saveInterval)
 		start := time.Now()
-		if err := t.storage.SaveTimestamp(t.GetTimestampPath(), save); err != nil {
+		if err := t.storage.SaveTimestamp(keypath.TimestampPath(t.tsPath), save); err != nil {
 			t.metrics.errSaveResetTSEvent.Inc()
 			return err
 		}
@@ -361,10 +335,10 @@
 	if typeutil.SubRealTimeByWallClock(t.getLastSavedTime(), next) <= UpdateTimestampGuard {
 		save := next.Add(t.saveInterval)
 		start := time.Now()
-		if err := t.storage.SaveTimestamp(t.GetTimestampPath(), save); err != nil {
+		if err := t.storage.SaveTimestamp(keypath.TimestampPath(t.tsPath), save); err != nil {
 			log.Warn("save timestamp failed",
 				logutil.CondUint32("keyspace-group-id", t.keyspaceGroupID, t.keyspaceGroupID > 0),
-				zap.String("timestamp-path", t.GetTimestampPath()),
+				zap.String("timestamp-path", keypath.TimestampPath(t.tsPath)),
 				zap.Error(err))
 			t.metrics.errSaveUpdateTSEvent.Inc()
 			return err

pkg/utils/keypath/key_path.go

@@ -370,20 +370,6 @@ func TimestampPath(tsPath string) string {
 	return path.Join(tsPath, TimestampKey)
 }
 
-// FullTimestampPath returns the full timestamp path.
-//  1. for the default keyspace group:
-//     /pd/{cluster_id}/timestamp
-//  2. for the non-default keyspace groups:
-//     /ms/{cluster_id}/tso/{group}/gta/timestamp
-func FullTimestampPath(groupID uint32) string {
-	rootPath := TSOSvcRootPath()
-	tsPath := TimestampPath(KeyspaceGroupGlobalTSPath(groupID))
-	if groupID == constant.DefaultKeyspaceGroupID {
-		rootPath = LegacyRootPath()
-	}
-	return path.Join(rootPath, tsPath)
-}
-
 const (
 	registryKey = "registry"
 )

server/config/config.go

@@ -96,10 +96,7 @@ type Config struct {
 	// be automatically clamped to the range.
 	TSOUpdatePhysicalInterval typeutil.Duration `toml:"tso-update-physical-interval" json:"tso-update-physical-interval"`
 
-	// 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.
+	// Deprecated
 	EnableLocalTSO bool `toml:"enable-local-tso" json:"enable-local-tso"`
 
 	Metric metricutil.MetricConfig `toml:"metric" json:"metric"`
@@ -115,8 +112,6 @@ type Config struct {
 	// Labels indicates the labels set for **this** PD server. The labels describe some specific properties
 	// like `zone`/`rack`/`host`. Currently, labels won't affect the PD server except for some special
 	// label keys. Now we have following special keys:
-	// 1. 'zone' is a special key that indicates the DC location of this PD server. If it is set, the value for this
-	// will be used to determine which DC's Local TSO service this PD will provide with if EnableLocalTSO is true.
 	Labels map[string]string `toml:"labels" json:"labels"`
 
 	// QuotaBackendBytes Raise alarms when backend size exceeds the given quota. 0 means use the default quota.

server/server.go

@@ -936,11 +936,6 @@ func (s *Server) GetServiceMiddlewareConfig() *config.ServiceMiddlewareConfig {
 	return cfg
 }
 
-// SetEnableLocalTSO sets enable-local-tso flag of Server. This function only for test.
-func (s *Server) SetEnableLocalTSO(enableLocalTSO bool) {
-	s.cfg.EnableLocalTSO = enableLocalTSO
-}
-
 // GetConfig gets the config information.
 func (s *Server) GetConfig() *config.Config {
 	cfg := s.cfg.Clone()

tests/cluster.go

@@ -185,11 +185,6 @@ func (s *TestServer) GetConfig() *config.Config {
 	return s.server.GetConfig()
 }
 
-// SetEnableLocalTSO sets the enable-local-tso flag of the TestServer.
-func (s *TestServer) SetEnableLocalTSO(enableLocalTSO bool) {
-	s.server.SetEnableLocalTSO(enableLocalTSO)
-}
-
 // GetPersistOptions returns the current TestServer's schedule option.
 func (s *TestServer) GetPersistOptions() *config.PersistOptions {
 	s.RLock()

tests/integrations/mcs/tso/keyspace_group_manager_test.go

@@ -223,8 +223,6 @@ func (suite *tsoKeyspaceGroupManagerTestSuite) TestKeyspacesServedByNonDefaultKe
 							ServiceName: constant.TSOServiceName,
 							GroupID:     param.keyspaceGroupID,
 						})
-						timestampPath := keypath.FullTimestampPath(param.keyspaceGroupID)
-						re.Equal(timestampPath, am.GetTimestampPath())
 						re.Equal(primaryPath, am.GetMember().GetLeaderPath())
 
 						served = true