Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #14935 from ahrtr/minor_enhance_error_20221213

Browse Source 
client: enhance the function shouldRetryWatch and added unit test
This commit is contained in:
Hitoshi Mitake 2022-12-13 23:56:21 +09:00 committed by GitHub
commit 6429c044a6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 51 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
client/v3/watch.go
View File

@ -18,7 +18,6 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"strings"
"sync" "sync"
"time" "time"
@ -46,6 +45,11 @@ const (
InvalidWatchID = -1 InvalidWatchID = -1
) )
var (
errMsgGRPCInvalidAuthToken = v3rpc.ErrGRPCInvalidAuthToken.Error()
errMsgGRPCAuthOldRevision = v3rpc.ErrGRPCAuthOldRevision.Error()
)
type Event mvccpb.Event type Event mvccpb.Event
type WatchChan <-chan WatchResponse type WatchChan <-chan WatchResponse
@ -588,8 +592,7 @@ func (w *watchGrpcStream) run() {
switch { switch {
case pbresp.Created: case pbresp.Created:
cancelReasonError := v3rpc.Error(errors.New(pbresp.CancelReason)) if pbresp.Canceled && shouldRetryWatch(pbresp.CancelReason) {
if shouldRetryWatch(cancelReasonError) {
var newErr error var newErr error
if wc, newErr = w.newWatchClient(); newErr != nil { if wc, newErr = w.newWatchClient(); newErr != nil {
w.lg.Error("failed to create a new watch client", zap.Error(newErr)) w.lg.Error("failed to create a new watch client", zap.Error(newErr))
@ -717,9 +720,12 @@ func (w *watchGrpcStream) run() {
} }
} }
func shouldRetryWatch(cancelReasonError error) bool { func shouldRetryWatch(cancelReason string) bool {
return (strings.Compare(cancelReasonError.Error(), v3rpc.ErrGRPCInvalidAuthToken.Error()) == 0) || if cancelReason == "" {
(strings.Compare(cancelReasonError.Error(), v3rpc.ErrGRPCAuthOldRevision.Error()) == 0) return false
}
return (cancelReason == errMsgGRPCInvalidAuthToken) ||
(cancelReason == errMsgGRPCAuthOldRevision)
} }
// nextResume chooses the next resuming to register with the grpc stream. Abandoned // nextResume chooses the next resuming to register with the grpc stream. Abandoned

39
client/v3/watch_test.go
View File

@ -17,7 +17,10 @@ package clientv3
import ( import (
"testing" "testing"
"github.com/stretchr/testify/assert"
"go.etcd.io/etcd/api/v3/mvccpb" "go.etcd.io/etcd/api/v3/mvccpb"
"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
) )
func TestEvent(t *testing.T) { func TestEvent(t *testing.T) {
@ -53,3 +56,39 @@ func TestEvent(t *testing.T) {
} }
} }
} }
func TestShouldRetryWatch(t *testing.T) {
testCases := []struct {
name string
msg string
expectedRetry bool
}{
{
name: "equal to ErrGRPCInvalidAuthToken",
msg: rpctypes.ErrGRPCInvalidAuthToken.Error(),
expectedRetry: true,
},
{
name: "equal to ErrGRPCAuthOldRevision",
msg: rpctypes.ErrGRPCAuthOldRevision.Error(),
expectedRetry: true,
},
{
name: "valid grpc error but not equal to ErrGRPCInvalidAuthToken or ErrGRPCAuthOldRevision",
msg: rpctypes.ErrGRPCUserEmpty.Error(),
expectedRetry: false,
},
{
name: "invalid grpc error and not equal to ErrGRPCInvalidAuthToken or ErrGRPCAuthOldRevision",
msg: "whatever error message",
expectedRetry: false,
},
}
for _, tc := range testCases {
tc := tc
t.Run(tc.name, func(t *testing.T) {
assert.Equal(t, tc.expectedRetry, shouldRetryWatch(tc.msg))
})
}
}