fix(tso): do not exit allocatorUpdater if the updating fails (tikv#9180)

close tikv#9179

Do not exit `allocatorUpdater` if the updating fails to prevent the TSO from stuck.

Signed-off-by: JmPotato <github@ipotato.me>

Author: JmPotato

pkg/tso/allocator.go

@@ -43,9 +43,17 @@ import (
 	"github.com/tikv/pd/pkg/utils/logutil"
 )
 
-// GlobalDCLocation is the Global TSO Allocator's DC location label.
-// Deprecated: This is a legacy label, it should be removed in the future.
-const GlobalDCLocation = "global"
+const (
+	// GlobalDCLocation is the Global TSO Allocator's DC location label.
+	// Deprecated: This is a legacy label, it should be removed in the future.
+	GlobalDCLocation = "global"
+	// maxUpdateTSORetryCount is the max retry count for updating TSO.
+	// When encountering a network partition, manually retrying may help the next request succeed with the new endpoint according to https://github.com/etcd-io/etcd/issues/8711
+	maxUpdateTSORetryCount = 3
+	// Etcd client retry with `roundRobinQuorumBackoff` (https://github.com/etcd-io/etcd/blob/d62cdeee4863001b09e772ed013eb1342a1d0f89/client/v3/client.go#L488),
+	// whose default interval is 25ms, so we sleep 50ms here. (https://github.com/etcd-io/etcd/blob/d62cdeee4863001b09e772ed013eb1342a1d0f89/client/v3/options.go#L53)
+	updateTSORetryInterval = 50 * time.Millisecond
+)
 
 // ElectionMember defines the interface for the election related logic.
 type ElectionMember interface {
@@ -175,7 +183,8 @@ func (a *Allocator) allocatorUpdater() {
 			if err := a.UpdateTSO(); err != nil {
 				log.Warn("failed to update allocator's timestamp", append(a.logFields, errs.ZapError(err))...)
 				a.Reset(true)
-				return
+				// To wait for the allocator to be re-initialized next time.
+				continue
 			}
 		case <-a.ctx.Done():
 			a.Reset(false)
@@ -185,9 +194,9 @@ func (a *Allocator) allocatorUpdater() {
 	}
 }
 
-// close is used to shutdown the primary election loop.
+// Close is used to close the allocator and shutdown all the daemon loops.
 // tso service call this function to shutdown the loop here, but pd manages its own loop.
-func (a *Allocator) close() {
+func (a *Allocator) Close() {
 	log.Info("closing the allocator", a.logFields...)
 	a.cancel()
 	a.wg.Wait()
@@ -207,18 +216,14 @@ func (a *Allocator) IsInitialize() bool {
 
 // UpdateTSO is used to update the TSO in memory and the time window in etcd.
 func (a *Allocator) UpdateTSO() (err error) {
-	// When meet network partition, we need to manually retry to update the tso,
-	// next request succeeds with the new endpoint, according to https://github.com/etcd-io/etcd/issues/8711
-	maxRetryCount := 3
-	for range maxRetryCount {
+	for i := range maxUpdateTSORetryCount {
 		err = a.timestampOracle.updateTimestamp()
 		if err == nil {
 			return nil
 		}
-		log.Warn("try to update the tso but failed", errs.ZapError(err))
-		// Etcd client retry with roundRobinQuorumBackoff https://github.com/etcd-io/etcd/blob/d62cdeee4863001b09e772ed013eb1342a1d0f89/client/v3/client.go#L488
-		// And its default interval is 25ms, so we sleep 50ms here. https://github.com/etcd-io/etcd/blob/d62cdeee4863001b09e772ed013eb1342a1d0f89/client/v3/options.go#L53
-		time.Sleep(50 * time.Millisecond)
+		log.Warn("try to update the tso but failed",
+			zap.Int("retry-count", i), zap.Duration("retry-interval", updateTSORetryInterval), errs.ZapError(err))
+		time.Sleep(updateTSORetryInterval)
 	}
 	return
 }

pkg/tso/keyspace_group_manager.go

@@ -113,7 +113,7 @@ func (s *state) deInitialize() {
 		go func(allocator *Allocator) {
 			defer logutil.LogPanic()
 			defer wg.Done()
-			allocator.close()
+			allocator.Close()
 			log.Info("keyspace group closed", zap.Uint32("keyspace-group-id", allocator.keyspaceGroupID))
 		}(allocator)
 	}
@@ -938,7 +938,7 @@ func (kgm *KeyspaceGroupManager) deleteKeyspaceGroup(groupID uint32) {
 
 	allocator := kgm.allocators[groupID]
 	if allocator != nil {
-		allocator.close()
+		allocator.Close()
 		kgm.allocators[groupID] = nil
 	}
 
@@ -955,7 +955,7 @@ func (kgm *KeyspaceGroupManager) exitElectionMembership(group *endpoint.Keyspace
 
 	allocator := kgm.allocators[group.ID]
 	if allocator != nil {
-		allocator.close()
+		allocator.Close()
 		kgm.allocators[group.ID] = nil
 	}
 

server/server.go

@@ -533,6 +533,9 @@ func (s *Server) Close() {
 	if s.IsKeyspaceGroupEnabled() {
 		s.keyspaceGroupManager.Close()
 	}
+	if s.tsoAllocator != nil {
+		s.tsoAllocator.Close()
+	}
 
 	if s.client != nil {
 		if err := s.client.Close(); err != nil {

tests/integrations/tso/client_test.go

@@ -92,6 +92,7 @@ func TestMicroserviceTSOClientSuite(t *testing.T) {
 
 func (suite *tsoClientTestSuite) SetupSuite() {
 	re := suite.Require()
+	re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/tso/fastUpdatePhysicalInterval", "return(true)"))
 
 	var err error
 	suite.ctx, suite.cancel = context.WithCancel(context.Background())
@@ -207,6 +208,7 @@ func (suite *tsoClientTestSuite) TearDownTest() {
 }
 
 func (suite *tsoClientTestSuite) TearDownSuite() {
+	failpoint.Disable("github.com/tikv/pd/pkg/tso/fastUpdatePhysicalInterval")
 	suite.cancel()
 	if !suite.legacy {
 		suite.tsoCluster.Destroy()
@@ -379,10 +381,8 @@ func (suite *tsoClientTestSuite) TestUpdateAfterResetTSO() {
 
 func (suite *tsoClientTestSuite) TestRandomResignLeader() {
 	re := suite.Require()
-	re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/tso/fastUpdatePhysicalInterval", "return(true)"))
 	re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/member/skipCampaignLeaderCheck", "return(true)"))
 	defer func() {
-		re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/tso/fastUpdatePhysicalInterval"))
 		re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/member/skipCampaignLeaderCheck"))
 	}()
 
@@ -426,7 +426,6 @@ func (suite *tsoClientTestSuite) TestRandomResignLeader() {
 
 func (suite *tsoClientTestSuite) TestRandomShutdown() {
 	re := suite.Require()
-	re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/tso/fastUpdatePhysicalInterval", "return(true)"))
 
 	parallelAct := func() {
 		// After https://github.com/tikv/pd/issues/6376 is fixed, we can use a smaller number here.
@@ -445,7 +444,6 @@ func (suite *tsoClientTestSuite) TestRandomShutdown() {
 	mcs.CheckMultiKeyspacesTSO(suite.ctx, re, suite.clients, parallelAct)
 	suite.TearDownSuite()
 	suite.SetupSuite()
-	re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/tso/fastUpdatePhysicalInterval"))
 }
 
 func (suite *tsoClientTestSuite) TestGetTSWhileResettingTSOClient() {