tso/local: remove local tso completely (#8864)

close #8802

Signed-off-by: okJiang <[email protected]>

Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com>

client/client.go

@@ -509,10 +509,10 @@ func (c *client) GetTSAsync(ctx context.Context) tso.TSFuture {
 	return c.inner.dispatchTSORequestWithRetry(ctx)
 }
 
-// GetLocalTSAsync implements the TSOClient interface.
-//
-// 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: the Local TSO feature has been deprecated. Regardless of the
+// parameters passed, the behavior of this interface will be equivalent to
+// `GetTSAsync`. If you want to use a separately deployed TSO service,
+// please refer to the deployment of the TSO microservice.
 func (c *client) GetLocalTSAsync(ctx context.Context, _ string) tso.TSFuture {
 	return c.GetTSAsync(ctx)
 }
@@ -523,10 +523,10 @@ func (c *client) GetTS(ctx context.Context) (physical int64, logical int64, err
 	return resp.Wait()
 }
 
-// GetLocalTS implements the TSOClient interface.
-//
-// 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: the Local TSO feature has been deprecated. Regardless of the
+// parameters passed, the behavior of this interface will be equivalent to
+// `GetTS`. If you want to use a separately deployed TSO service,
+// please refer to the deployment of the TSO microservice.
 func (c *client) GetLocalTS(ctx context.Context, _ string) (physical int64, logical int64, err error) {
 	return c.GetTS(ctx)
 }

client/clients/tso/client.go

@@ -56,15 +56,15 @@ type Client interface {
 	// the TSO microservice.
 	GetMinTS(ctx context.Context) (int64, int64, error)
 
-	// 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: the Local TSO feature has been deprecated. Regardless of the
+	// parameters passed, the behavior of this interface will be equivalent to
+	// `GetTS`. If you want to use a separately deployed TSO service,
+	// please refer to the deployment of the TSO microservice.
 	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: the Local TSO feature has been deprecated. Regardless of the
+	// parameters passed, the behavior of this interface will be equivalent to
+	// `GetTSAsync`. If you want to use a separately deployed TSO service,
+	// please refer to the deployment of the TSO microservice.
 	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/service_middleware.go

@@ -29,7 +29,7 @@ type ServiceMiddlewareStorage interface {
 
 var _ ServiceMiddlewareStorage = (*StorageEndpoint)(nil)
 
-// LoadServiceMiddlewareConfig loads service middleware config from keypath.KeyspaceGroupLocalTSPath then unmarshal it to cfg.
+// LoadServiceMiddlewareConfig loads service middleware config from ServiceMiddlewarePath then unmarshal it to cfg.
 func (se *StorageEndpoint) LoadServiceMiddlewareConfig(cfg any) (bool, error) {
 	value, err := se.Load(keypath.ServiceMiddlewarePath)
 	if err != nil || value == "" {
@@ -42,7 +42,7 @@ func (se *StorageEndpoint) LoadServiceMiddlewareConfig(cfg any) (bool, error) {
 	return true, nil
 }
 
-// SaveServiceMiddlewareConfig stores marshallable cfg to the keypath.KeyspaceGroupLocalTSPath.
+// SaveServiceMiddlewareConfig stores marshallable cfg to the ServiceMiddlewarePath.
 func (se *StorageEndpoint) SaveServiceMiddlewareConfig(cfg any) error {
 	return se.saveJSON(keypath.ServiceMiddlewarePath, cfg)
 }

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

@@ -16,8 +16,6 @@ package tso
 
 import (
 	"context"
-	"math"
-	"path"
 	"runtime/trace"
 	"strconv"
 	"sync"
@@ -43,8 +41,6 @@ const (
 	checkStep                   = time.Minute
 	patrolStep                  = time.Second
 	defaultAllocatorLeaderLease = 3
-	localTSOAllocatorEtcdPrefix = "lta"
-	localTSOSuffixEtcdPrefix    = "lts"
 )
 
 var (
@@ -217,17 +213,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()
@@ -254,21 +239,6 @@ func (am *AllocatorManager) GetMember() ElectionMember {
 	return am.member
 }
 
-// GetSuffixBits calculates the bits of suffix sign
-// by the max number of suffix so far,
-// which will be used in the TSO logical part.
-func (am *AllocatorManager) GetSuffixBits() int {
-	am.mu.RLock()
-	defer am.mu.RUnlock()
-	return CalSuffixBits(am.mu.maxSuffix)
-}
-
-// CalSuffixBits calculates the bits of suffix by the max suffix sign.
-func CalSuffixBits(maxSuffix int32) int {
-	// maxSuffix + 1 because we have the Global TSO holds 0 as the suffix sign
-	return int(math.Ceil(math.Log2(float64(maxSuffix + 1))))
-}
-
 // AllocatorDaemon is used to update every allocator's TSO and check whether we have
 // any new local allocator that needs to be set up.
 func (am *AllocatorManager) AllocatorDaemon(ctx context.Context) {

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()
 }
 
@@ -175,7 +159,7 @@ func (gta *GlobalTSOAllocator) GenerateTSO(ctx context.Context, count uint32) (p
 		return pdpb.Timestamp{}, errs.ErrGenerateTimestamp.FastGenByArgs(fmt.Sprintf("requested pd %s of cluster", errs.NotLeaderErr))
 	}
 
-	return gta.timestampOracle.getTS(ctx, gta.member.GetLeadership(), count, 0)
+	return gta.timestampOracle.getTS(ctx, gta.member.GetLeadership(), count)
 }
 
 // Reset is used to reset the TSO allocator.

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.