Skip to content

Commit

Permalink
[pick-8.1] region_request: ignore resource group errors that not rela…
Browse files Browse the repository at this point in the history
…tive storage layer (#1354) (#1463)

ref #1322

Signed-off-by: nolouch <[email protected]>
  • Loading branch information
nolouch authored Sep 13, 2024
1 parent 1f18db9 commit 3777c38
Show file tree
Hide file tree
Showing 2 changed files with 51 additions and 11 deletions.
30 changes: 19 additions & 11 deletions internal/locate/region_request.go
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@ import (
"github.com/tikv/client-go/v2/metrics"
"github.com/tikv/client-go/v2/tikvrpc"
"github.com/tikv/client-go/v2/util"
"github.com/tikv/pd/client/errs"
pderr "github.com/tikv/pd/client/errs"
)

Expand Down Expand Up @@ -2124,7 +2125,9 @@ func (s *RegionRequestSender) sendReqToRegion(
}

if err != nil {
s.rpcError = err
if isRPCError(err) {
s.rpcError = err
}
if s.Stats != nil {
errStr := getErrMsg(err)
s.Stats.RecordRPCErrorStats(errStr)
Expand All @@ -2151,6 +2154,11 @@ func (s *RegionRequestSender) sendReqToRegion(
return
}

func isRPCError(err error) bool {
// exclude ErrClientResourceGroupThrottled
return err != nil && errs.ErrClientResourceGroupThrottled.NotEqual(err)
}

func storeIDLabel(rpcCtx *RPCContext) string {
if rpcCtx != nil && rpcCtx.Store != nil {
return strconv.FormatUint(rpcCtx.Store.storeID, 10)
Expand Down Expand Up @@ -2228,16 +2236,6 @@ func (s *RegionRequestSender) onSendFail(bo *retry.Backoffer, ctx *RPCContext, r
metrics.TiKVRPCErrorCounter.WithLabelValues("unknown", storeLabel).Inc()
}

if ctx.Store != nil && ctx.Store.storeType == tikvrpc.TiFlashCompute {
s.regionCache.InvalidateTiFlashComputeStoresIfGRPCError(err)
} else if ctx.Meta != nil {
if s.replicaSelector != nil {
s.replicaSelector.onSendFailure(bo, err)
} else {
s.regionCache.OnSendFail(bo, ctx, s.NeedReloadRegion(ctx), err)
}
}

// don't need to retry for ResourceGroup error
if errors.Is(err, pderr.ErrClientResourceGroupThrottled) {
return err
Expand All @@ -2250,6 +2248,16 @@ func (s *RegionRequestSender) onSendFail(bo *retry.Backoffer, ctx *RPCContext, r
return err
}

if ctx.Store != nil && ctx.Store.storeType == tikvrpc.TiFlashCompute {
s.regionCache.InvalidateTiFlashComputeStoresIfGRPCError(err)
} else if ctx.Meta != nil {
if s.replicaSelector != nil {
s.replicaSelector.onSendFailure(bo, err)
} else {
s.regionCache.OnSendFail(bo, ctx, s.NeedReloadRegion(ctx), err)
}
}

// Retry on send request failure when it's not canceled.
// When a store is not available, the leader of related region should be elected quickly.
// TODO: the number of retry time should be limited:since region may be unavailable
Expand Down
32 changes: 32 additions & 0 deletions internal/locate/region_request_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@ import (
"github.com/tikv/client-go/v2/internal/client/mockserver"
"github.com/tikv/client-go/v2/internal/mockstore/mocktikv"
"github.com/tikv/client-go/v2/tikvrpc"
pderr "github.com/tikv/pd/client/errs"
"google.golang.org/grpc"
)

Expand Down Expand Up @@ -160,6 +161,37 @@ func (s *testRegionRequestToSingleStoreSuite) TestOnRegionError() {
}()
}

func (s *testRegionRequestToSingleStoreSuite) TestOnSendFailByResourceGroupThrottled() {
req := tikvrpc.NewRequest(tikvrpc.CmdRawPut, &kvrpcpb.RawPutRequest{
Key: []byte("key"),
Value: []byte("value"),
})
region, err := s.cache.LocateRegionByID(s.bo, s.region)
s.Nil(err)
s.NotNil(region)

// test ErrClientResourceGroupThrottled handled by regionRequestSender
func() {
oc := s.regionRequestSender.client
defer func() {
s.regionRequestSender.client = oc
}()
storeOld, _ := s.regionRequestSender.regionCache.stores.get(1)
epoch := storeOld.epoch
s.regionRequestSender.client = &fnClient{fn: func(ctx context.Context, addr string, req *tikvrpc.Request, timeout time.Duration) (response *tikvrpc.Response, err error) {
return nil, pderr.ErrClientResourceGroupThrottled
}}
bo := retry.NewBackofferWithVars(context.Background(), 5, nil)
_, _, err := s.regionRequestSender.SendReq(bo, req, region.Region, time.Second)
s.NotNil(err)
storeNew, _ := s.regionRequestSender.regionCache.stores.get(1)
// not mark the store need be refill, then the epoch should not be changed.
s.Equal(epoch, storeNew.epoch)
// no rpc error if the error is ErrClientResourceGroupThrottled
s.Nil(s.regionRequestSender.rpcError)
}()
}

func (s *testRegionRequestToSingleStoreSuite) TestOnSendFailedWithStoreRestart() {
req := tikvrpc.NewRequest(tikvrpc.CmdRawPut, &kvrpcpb.RawPutRequest{
Key: []byte("key"),
Expand Down

0 comments on commit 3777c38

Please sign in to comment.