Repository: tikv/client-go
Pull request #891 (2023-07-17 08:48:06)

handle `mismatch peer id` region error #885
cherry-pick https://github.com/tikv/client-go/pull/888


Patch:
diff --git a/internal/locate/region_request.go b/internal/locate/region_request.go
index b39eb5a..ee48e06 100644
--- a/internal/locate/region_request.go
+++ b/internal/locate/region_request.go
@@ -35,6 +35,7 @@
 package locate
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"math/rand"
@@ -259,6 +260,7 @@ type replicaSelector struct {
 	targetIdx AccessIndex
 	// replicas[proxyIdx] is the store used to redirect requests this time
 	proxyIdx AccessIndex
+	reqTxnTs uint64
 }
 
 // selectorState is the interface of states of the replicaSelector.
@@ -557,11 +559,45 @@ func (state *accessFollower) next(bo *retry.Backoffer, selector *replicaSelector
 	}
 	// If there is no candidate, fallback to the leader.
 	if selector.targetIdx < 0 {
-		if len(state.option.labels) > 0 {
-			logutil.BgLogger().Warn("unable to find stores with given labels")
-		}
 		leader := selector.replicas[state.leaderIdx]
-		if leader.isEpochStale() || leader.isExhausted(1) {
+		leaderInvalid := leader.isEpochStale() || leader.isExhausted(1)
+		if len(state.option.labels) > 0 {
+			var regionID uint64
+			var labels []string
+			var replicaStatus []string
+			if selector.region != nil {
+				regionID = selector.region.GetID()
+			}
+			for _, label := range state.option.labels {
+				labels = append(labels, label.Key+": "+label.Value)
+			}
+			allMismatch := true
+			for _, replica := range selector.replicas {
+				isLabelsMatch := replica.store.IsLabelsMatch(state.option.labels)
+				if isLabelsMatch {
+					allMismatch = false
+				}
+				replicaStatus = append(replicaStatus, fmt.Sprintf("peer: %v, store: %v, isEpochStale: %v, IsLabelsMatch: %v, isExhausted: %v, attempts: %v",
+					replica.peer.GetId(),
+					replica.store.storeID,
+					replica.isEpochStale(),
+					isLabelsMatch,
+					replica.isExhausted(1),
+					replica.attempts))
+
+			}
+			logutil.BgLogger().Warn("unable to find stores with given labels",
+				zap.Uint64("txn-ts", selector.reqTxnTs),
+				zap.Uint64("region-id", regionID),
+				zap.Bool("all-replica-labels-mismatch", allMismatch),
+				zap.Uint64("leader-id", leader.peer.GetId()),
+				zap.Bool("leader-invalid", leaderInvalid),
+				zap.Bool("leaderOnly", state.option.leaderOnly),
+				zap.String("req-labels", strings.Join(labels, ",")),
+				zap.String("replica-status", strings.Join(replicaStatus, ";")),
+			)
+		}
+		if leaderInvalid {
 			metrics.TiKVReplicaSelectorFailureCounter.WithLabelValues("exhausted").Inc()
 			selector.invalidateRegion()
 			return nil, nil
@@ -644,6 +680,7 @@ func newReplicaSelector(regionCache *RegionCache, regionID RegionVerID, req *tik
 		}
 	}
 
+	txnTs := getReqTxnTs(req)
 	return &replicaSelector{
 		regionCache,
 		cachedRegion,
@@ -652,6 +689,7 @@ func newReplicaSelector(regionCache *RegionCache, regionID RegionVerID, req *tik
 		state,
 		-1,
 		-1,
+		txnTs,
 	}, nil
 }
 
@@ -976,6 +1014,7 @@ func (s *RegionRequestSender) SendReqCtx(
 			metrics.TiKVRequestRetryTimesHistogram.Observe(float64(tryTimes))
 		}
 	}()
+	totalErrors := make(map[string]int)
 	for {
 		if tryTimes > 0 {
 			req.IsRetryRequest = true
@@ -1004,7 +1043,7 @@ func (s *RegionRequestSender) SendReqCtx(
 
 			// TODO: Change the returned error to something like "region missing in cache",
 			// and handle this error like EpochNotMatch, which means to re-split the request and retry.
-			logutil.Logger(bo.GetCtx()).Debug("throwing pseudo region error due to region not found in cache", zap.Stringer("region", &regionID))
+			s.logReqError(bo, "throwing pseudo region error due to no replica available", regionID, tryTimes, req, totalErrors)
 			resp, err = tikvrpc.GenRegionErrorResp(req, &errorpb.Error{EpochNotMatch: &errorpb.EpochNotMatch{}})
 			return resp, nil, err
 		}
@@ -1046,6 +1085,8 @@ func (s *RegionRequestSender) SendReqCtx(
 			return nil, nil, err
 		}
 		if regionErr != nil {
+			errStr := getRegionErrorString(regionErr)
+			totalErrors[errStr]++
 			retry, err = s.onRegionError(bo, rpcCtx, req, regionErr)
 			if err != nil {
 				return nil, nil, err
@@ -1054,6 +1095,7 @@ func (s *RegionRequestSender) SendReqCtx(
 				tryTimes++
 				continue
 			}
+			s.logReqError(bo, "send req failed without retry", regionID, tryTimes, req, totalErrors)
 		} else {
 			if s.replicaSelector != nil {
 				s.replicaSelector.onSendSuccess()
@@ -1063,6 +1105,135 @@ func (s *RegionRequestSender) SendReqCtx(
 	}
 }
 
+func getRegionErrorString(regionErr *errorpb.Error) string {
+	if regionErr == nil {
+		return ""
+	}
+	errStr := regionErrorToLabel(regionErr)
+	if errStr == "unknown" {
+		if regionErr.Message != "" {
+			errStr = regionErr.Message
+		} else {
+			errStr = regionErr.String()
+		}
+	}
+	return errStr
+}
+
+func (s *RegionRequestSender) logReqError(bo *retry.Backoffer, msg string, regionID RegionVerID, tryTimes int, req *tikvrpc.Request, totalErrors map[string]int) {
+	var replicaStatus []string
+	replicaSelectorState := "nil"
+	cachedRegionInfo := "nil"
+	if s.replicaSelector != nil {
+		switch s.replicaSelector.state.(type) {
+		case *accessKnownLeader:
+			replicaSelectorState = "accessKnownLeader"
+		case *accessFollower:
+			replicaSelectorState = "accessFollower"
+		case *accessByKnownProxy:
+			replicaSelectorState = "accessByKnownProxy"
+		case *tryFollower:
+			replicaSelectorState = "tryFollower"
+		case *tryNewProxy:
+			replicaSelectorState = "tryNewProxy"
+		case *invalidLeader:
+			replicaSelectorState = "invalidLeader"
+		case *invalidStore:
+			replicaSelectorState = "invalidStore"
+		case *stateBase:
+			replicaSelectorState = "stateBase"
+		case nil:
+			replicaSelectorState = "nil"
+		}
+		for _, replica := range s.replicaSelector.replicas {
+			replicaStatus = append(replicaStatus, fmt.Sprintf("peer: %v, store: %v, isEpochStale: %v, attempts: %v",
+				replica.peer.GetId(),
+				replica.store.storeID,
+				replica.isEpochStale(),
+				replica.attempts))
+		}
+		if s.replicaSelector.region != nil {
+			cachedRegionInfo = fmt.Sprintf("region-id: %v, conf-ver: %v, ver: %v, isValid: %v, checkNeedReload: %v",
+				regionID.GetID(),
+				regionID.GetConfVer(),
+				regionID.GetVer(),
+				s.replicaSelector.region.isValid(),
+				s.replicaSelector.region.checkNeedReload(),
+			)
+		}
+	}
+	txnTs := getReqTxnTs(req)
+
+	var totalErrorStr bytes.Buffer
+	for err, cnt := range totalErrors {
+		if totalErrorStr.Len() > 0 {
+			totalErrorStr.WriteString(", ")
+		}
+		totalErrorStr.WriteString(err)
+		totalErrorStr.WriteString(":")
+		totalErrorStr.WriteString(strconv.Itoa(cnt))
+	}
+	var replicaReadType string
+	switch req.ReplicaReadType {
+	case kv.ReplicaReadLeader:
+		replicaReadType = "ReplicaReadLeader"
+	case kv.ReplicaReadFollower:
+		replicaReadType = "ReplicaReadFollower"
+	case kv.ReplicaReadMixed:
+		replicaReadType = "ReplicaReadMixed"
+	default:
+		replicaReadType	= fmt.Sprintf("unknown(%d)", int(req.ReplicaReadType))
+	}
+	var backoffDetail bytes.Buffer
+	totalBackOff := bo.GetTotalSleep()
+	backoffTimes := bo.GetBackoffTimes()
+	for k,v := range backoffTimes{
+		if backoffDetail.Len() > 0{
+			backoffDetail.WriteString(", ")
+		}
+		backoffDetail.WriteString(k)
+		backoffDetail.WriteString(":")
+		backoffDetail.WriteString(strconv.Itoa(v))
+	}
+	logutil.Logger(bo.GetCtx()).Info(msg,
+		zap.Uint64("txn-ts", txnTs),
+		zap.String("region-info", cachedRegionInfo),
+		zap.Int("try-times", tryTimes),
+		zap.String("replica-read-type", replicaReadType),
+		zap.String("replica-selector-state", replicaSelectorState),
+		zap.String("total-region-errors", totalErrorStr.String()),
+		zap.String("replica-status", strings.Join(replicaStatus, ";")),
+		zap.Int("total-backoff-ms", totalBackOff),
+		zap.Int("total-backoff-times", bo.GetTotalBackoffTimes()),
+		zap.String("backoff-detail", backoffDetail.String()),
+		)
+}
+
+func getReqTxnTs(req *tikvrpc.Request) uint64{
+	txnTs := uint64(0)
+	switch req.Type {
+	case tikvrpc.CmdGet:
+		r := req.Get()
+		txnTs = r.GetVersion()
+	case tikvrpc.CmdBatchGet:
+		r := req.BatchGet()
+		txnTs = r.GetVersion()
+	case tikvrpc.CmdCop:
+		r := req.Cop()
+		txnTs = r.StartTs
+	case tikvrpc.CmdBatchCop:
+		r := req.BatchCop()
+		txnTs = r.StartTs
+	case tikvrpc.CmdPrewrite:
+		r := req.Prewrite()
+		txnTs = r.StartVersion
+	case tikvrpc.CmdScan:
+		r := req.Scan()
+		txnTs = r.GetVersion()
+	}
+	return txnTs
+}
+
 // RPCCancellerCtxKey is context key attach rpc send cancelFunc collector to ctx.
 type RPCCancellerCtxKey struct{}
 
@@ -1399,6 +1570,11 @@ func regionErrorToLabel(e *errorpb.Error) string {
 		return "flashback_in_progress"
 	} else if e.GetFlashbackNotPrepared() != nil {
 		return "flashback_not_prepared"
+	} else if strings.Contains(e.Message, "mismatch peer id") {
+		// the error message is like "[components/raftstore/src/store/util.rs:428]: mismatch peer id ? != ?"
+		// the `mismatch peer id` error does not has a specific error type, so we have to match the error message.
+		// TODO: add a specific error type for `mismatch peer id`.
+		return "mismatch_peer_id"
 	}
 	return "unknown"
 }
@@ -1411,7 +1587,8 @@ func (s *RegionRequestSender) onRegionError(bo *retry.Backoffer, ctx *RPCContext
 	}
 
 	// NOTE: Please add the region error handler in the same order of errorpb.Error.
-	metrics.TiKVRegionErrorCounter.WithLabelValues(regionErrorToLabel(regionErr)).Inc()
+	errLabel := regionErrorToLabel(regionErr)
+	metrics.TiKVRegionErrorCounter.WithLabelValues(errLabel).Inc()
 
 	if notLeader := regionErr.GetNotLeader(); notLeader != nil {
 		// Retry if error is `NotLeader`.
@@ -1612,6 +1789,10 @@ func (s *RegionRequestSender) onRegionError(bo *retry.Backoffer, ctx *RPCContext
 		zap.Stringer("ctx", ctx))
 
 	if s.replicaSelector != nil {
+		if errLabel == "mismatch_peer_id" {
+			s.replicaSelector.invalidateRegion()
+			return false, nil
+		}
 		// Try the next replica.
 		return true, nil
 	}


Test patch:
diff --git a/internal/locate/region_request3_test.go b/internal/locate/region_request3_test.go
index c0f579a..29edd95 100644
--- a/internal/locate/region_request3_test.go
+++ b/internal/locate/region_request3_test.go
@@ -873,3 +873,48 @@ func (s *testRegionRequestToThreeStoresSuite) TestSendReqWithReplicaSelector() {
 		s.True(totalAttempts <= 2)
 	}
 }
+
+func (s *testRegionRequestToThreeStoresSuite) TestReplicaReadFallbackToLeaderRegionError() {
+	regionLoc, err := s.cache.LocateRegionByID(s.bo, s.regionID)
+	s.Nil(err)
+	s.NotNil(regionLoc)
+
+	s.regionRequestSender.client = &fnClient{fn: func(ctx context.Context, addr string, req *tikvrpc.Request, timeout time.Duration) (response *tikvrpc.Response, err error) {
+		select {
+		case <-ctx.Done():
+			return nil, errors.New("timeout")
+		default:
+		}
+		// Return `mismatch peer id` when accesses the leader.
+		if addr == s.cluster.GetStore(s.storeIDs[0]).Address {
+			return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{RegionError: &errorpb.Error{
+				Message: `"[components/raftstore/src/store/util.rs:428]: mismatch peer id 1 != 2"`,
+			}}}, nil
+		}
+		return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{RegionError: &errorpb.Error{
+			DataIsNotReady: &errorpb.DataIsNotReady{},
+		}}}, nil
+	}}
+
+	region := s.cache.getRegionByIDFromCache(regionLoc.Region.GetID())
+	s.True(region.isValid())
+
+	req := tikvrpc.NewReplicaReadRequest(tikvrpc.CmdGet, &kvrpcpb.GetRequest{Key: []byte("key")}, kv.ReplicaReadLeader, nil)
+	req.ReadReplicaScope = oracle.GlobalTxnScope
+	req.TxnScope = oracle.GlobalTxnScope
+	req.EnableStaleRead()
+	req.ReplicaReadType = kv.ReplicaReadFollower
+
+	ctx, _ := context.WithTimeout(context.Background(), 10*time.Second)
+	bo := retry.NewBackoffer(ctx, -1)
+	s.Nil(err)
+	resp, err := s.regionRequestSender.SendReq(bo, req, regionLoc.Region, time.Second)
+	s.Nil(err)
+	regionErr, err := resp.GetRegionError()
+	s.Nil(err)
+	s.Equal(regionErrorToLabel(regionErr), "mismatch_peer_id")
+	// return non-epoch-not-match region error and the upper layer can auto retry.
+	s.Nil(regionErr.GetEpochNotMatch())
+	// after region error returned, the region should be invalidated.
+	s.False(region.isValid())
+}