skip update member when request reach the limit

Signed-off-by: Ryan Leung <[email protected]>
rleungx · Jul 27, 2023 · 65b693d · 65b693d
1 parent a034984
commit 65b693d
Show file tree

Hide file tree

Showing 5 changed files with 10 additions and 11 deletions.
diff --git a/client/client.go b/client/client.go
@@ -33,7 +33,6 @@ import (
 	"github.com/tikv/pd/client/grpcutil"
 	"github.com/tikv/pd/client/tlsutil"
 	"github.com/tikv/pd/client/tsoutil"
-	"github.com/tikv/pd/pkg/errs"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -960,6 +959,7 @@ func (c *client) GetRegionFromMember(ctx context.Context, key []byte, memberURLs
 	start := time.Now()
 	defer func() { cmdDurationGetRegion.Observe(time.Since(start).Seconds()) }()
 
+	var err error
 	var resp *pdpb.GetRegionResponse
 	for _, url := range memberURLs {
 		conn, err := c.pdSvcDiscovery.GetOrCreateGRPCConn(url)

diff --git a/client/client_test.go b/client/client_test.go
@@ -92,7 +92,7 @@ func TestGRPCDialOption(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.TODO(), 500*time.Millisecond)
 	defer cancel()
 	cli := &pdServiceDiscovery{
-		checkMembershipCh: make(chan struct{}, 1),
+		checkMembershipCh: make(chan error, 1),
 		ctx:               ctx,
 		cancel:            cancel,
 		tlsCfg:            &tlsutil.TLSConfig{},

diff --git a/client/pd_service_discovery.go b/client/pd_service_discovery.go
@@ -30,7 +30,6 @@ import (
 	"github.com/tikv/pd/client/errs"
 	"github.com/tikv/pd/client/grpcutil"
 	"github.com/tikv/pd/client/tlsutil"
-	"github.com/tikv/pd/pkg/errs"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 )
@@ -84,7 +83,7 @@ type ServiceDiscovery interface {
 	// ScheduleCheckMemberChanged is used to trigger a check to see if there is any membership change
 	// among the leader/followers in a quorum-based cluster or among the primary/secondaries in a
 	// primary/secondary configured cluster.
-	ScheduleCheckMemberChanged()
+	ScheduleCheckMemberChanged(err error)
 	// CheckMemberChanged immediately check if there is any membership change among the leader/followers
 	// in a quorum-based cluster or among the primary/secondaries in a primary/secondary configured cluster.
 	CheckMemberChanged() error
@@ -142,7 +141,7 @@ type pdServiceDiscovery struct {
 	// leader is updated.
 	tsoGlobalAllocLeaderUpdatedCb tsoGlobalServAddrUpdatedFunc
 
-	checkMembershipCh chan err
+	checkMembershipCh chan error
 
 	wg        *sync.WaitGroup
 	ctx       context.Context
@@ -246,7 +245,7 @@ func (c *pdServiceDiscovery) updateMemberLoop() {
 			return
 		case <-ticker.C:
 		case err := <-c.checkMembershipCh:
-			if err != nil && strings.Contains(err.Error(), errs.ErrRateLimitExceeded) {
+			if err != nil && strings.Contains(err.Error(), errs.ErrRateLimitExceeded.Error()) {
 				continue
 			}
 		}

diff --git a/client/tso_dispatcher.go b/client/tso_dispatcher.go
@@ -76,7 +76,7 @@ func (c *tsoClient) dispatchRequest(dcLocation string, request *tsoRequest) erro
 	if !ok {
 		err := errs.ErrClientGetTSO.FastGenByArgs(fmt.Sprintf("unknown dc-location %s to the client", dcLocation))
 		log.Error("[tso] dispatch tso request error", zap.String("dc-location", dcLocation), errs.ZapError(err))
-		c.svcDiscovery.ScheduleCheckMemberChanged()
+		c.svcDiscovery.ScheduleCheckMemberChanged(nil)
 		return err
 	}
 	dispatcher.(*tsoDispatcher).tsoBatchController.tsoRequestCh <- request
@@ -439,7 +439,7 @@ tsoBatchLoop:
 				case <-streamLoopTimer.C:
 					err = errs.ErrClientCreateTSOStream.FastGenByArgs(errs.RetryTimeoutErr)
 					log.Error("[tso] create tso stream error", zap.String("dc-location", dc), errs.ZapError(err))
-					c.svcDiscovery.ScheduleCheckMemberChanged()
+					c.svcDiscovery.ScheduleCheckMemberChanged(nil)
 					c.finishRequest(tbc.getCollectedRequests(), 0, 0, 0, errors.WithStack(err))
 					timer.Stop()
 					continue tsoBatchLoop
@@ -483,7 +483,7 @@ tsoBatchLoop:
 				return
 			default:
 			}
-			c.svcDiscovery.ScheduleCheckMemberChanged()
+			c.svcDiscovery.ScheduleCheckMemberChanged(nil)
 			log.Error("[tso] getTS error",
 				zap.String("dc-location", dc),
 				zap.String("stream-addr", streamAddr),
@@ -589,7 +589,7 @@ func (c *tsoClient) tryConnectToTSO(
 	ticker := time.NewTicker(retryInterval)
 	defer ticker.Stop()
 	for i := 0; i < maxRetryTimes; i++ {
-		c.svcDiscovery.ScheduleCheckMemberChanged()
+		c.svcDiscovery.ScheduleCheckMemberChanged(nil)
 		cc, url = c.GetTSOAllocatorClientConnByDCLocation(dc)
 		cctx, cancel := context.WithCancel(dispatcherCtx)
 		stream, err = c.tsoStreamBuilderFactory.makeBuilder(cc).build(cctx, cancel, c.option.timeout)

diff --git a/client/tso_service_discovery.go b/client/tso_service_discovery.go
@@ -344,7 +344,7 @@ func (c *tsoServiceDiscovery) GetOrCreateGRPCConn(addr string) (*grpc.ClientConn
 }
 
 // ScheduleCheckMemberChanged is used to trigger a check to see if there is any change in service endpoints.
-func (c *tsoServiceDiscovery) ScheduleCheckMemberChanged() {
+func (c *tsoServiceDiscovery) ScheduleCheckMemberChanged(err error) {
 	select {
 	case c.checkMembershipCh <- struct{}{}:
 	default: