Skip to content

Commit

Permalink
Remove string dependent error handling in watch and auth
Browse files Browse the repository at this point in the history
Signed-off-by: Hitoshi Mitake <h.mitake@gmail.com>
  • Loading branch information
mitake committed Feb 5, 2024
1 parent 43d6759 commit 1db2eb8
Show file tree
Hide file tree
Showing 3 changed files with 0 additions and 99 deletions.
27 changes: 0 additions & 27 deletions client/v3/watch.go
Original file line number Diff line number Diff line change
Expand Up @@ -593,25 +593,6 @@ func (w *watchGrpcStream) run() {

switch {
case pbresp.Created:
if pbresp.Canceled && shouldRetryWatch(pbresp.CancelReason) {
var newErr error
if wc, newErr = w.newWatchClient(); newErr != nil {
w.lg.Error("failed to create a new watch client", zap.Error(newErr))
return
}

if len(w.resuming) != 0 {
if ws := w.resuming[0]; ws != nil {
if err := wc.Send(ws.initReq.toPB()); err != nil {
w.lg.Debug("error when sending request", zap.Error(err))
}
}
}

cur = nil
continue
}

// response to head of queue creation
if len(w.resuming) != 0 {
if ws := w.resuming[0]; ws != nil {
Expand Down Expand Up @@ -721,14 +702,6 @@ func (w *watchGrpcStream) run() {
}
}

func shouldRetryWatch(cancelReason string) bool {
if cancelReason == "" {
return false
}
return (cancelReason == errMsgGRPCInvalidAuthToken) ||
(cancelReason == errMsgGRPCAuthOldRevision)
}

// nextResume chooses the next resuming to register with the grpc stream. Abandoned
// streams are marked as nil in the queue since the head must wait for its inflight registration.
func (w *watchGrpcStream) nextResume() *watcherStream {
Expand Down
39 changes: 0 additions & 39 deletions client/v3/watch_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,7 @@ package clientv3
import (
"testing"

"github.com/stretchr/testify/assert"

"go.etcd.io/etcd/api/v3/mvccpb"
"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
)

func TestEvent(t *testing.T) {
Expand Down Expand Up @@ -56,39 +53,3 @@ 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))
})
}
}
33 changes: 0 additions & 33 deletions tests/integration/v3_auth_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -420,39 +420,6 @@ func TestV3AuthOldRevConcurrent(t *testing.T) {
wg.Wait()
}

func TestV3AuthWatchAndTokenExpire(t *testing.T) {
integration.BeforeTest(t)
clus := integration.NewCluster(t, &integration.ClusterConfig{Size: 1, AuthTokenTTL: 3})
defer clus.Terminate(t)

ctx, cancel := context.WithTimeout(context.TODO(), 10*time.Second)
defer cancel()

authSetupRoot(t, integration.ToGRPC(clus.Client(0)).Auth)

c, cerr := integration.NewClient(t, clientv3.Config{Endpoints: clus.Client(0).Endpoints(), Username: "root", Password: "123"})
if cerr != nil {
t.Fatal(cerr)
}
defer c.Close()

_, err := c.Put(ctx, "key", "val")
if err != nil {
t.Fatalf("Unexpected error from Put: %v", err)
}

// The first watch gets a valid auth token through watcher.newWatcherGrpcStream()
// We should discard the first one by waiting TTL after the first watch.
wChan := c.Watch(ctx, "key", clientv3.WithRev(1))
<-wChan

time.Sleep(5 * time.Second)

wChan = c.Watch(ctx, "key", clientv3.WithRev(1))
watchResponse := <-wChan
testutil.AssertNil(t, watchResponse.Err())
}

func TestV3AuthWatchErrorAndWatchId0(t *testing.T) {
integration.BeforeTest(t)
clus := integration.NewCluster(t, &integration.ClusterConfig{Size: 1})
Expand Down

0 comments on commit 1db2eb8

Please sign in to comment.