diff --git a/client/v3/watch.go b/client/v3/watch.go index b73925ba1..5bd2d4cd0 100644 --- a/client/v3/watch.go +++ b/client/v3/watch.go @@ -18,6 +18,7 @@ import ( "context" "errors" "fmt" + "strings" "sync" "time" @@ -580,6 +581,26 @@ func (w *watchGrpcStream) run() { switch { case pbresp.Created: + cancelReasonError := v3rpc.Error(errors.New(pbresp.CancelReason)) + if shouldRetryWatch(cancelReasonError) { + 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 { @@ -688,6 +709,11 @@ func (w *watchGrpcStream) run() { } } +func shouldRetryWatch(cancelReasonError error) bool { + return (strings.Compare(cancelReasonError.Error(), v3rpc.ErrGRPCInvalidAuthToken.Error()) == 0) || + (strings.Compare(cancelReasonError.Error(), v3rpc.ErrGRPCAuthOldRevision.Error()) == 0) +} + // 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 { diff --git a/server/etcdserver/api/v3rpc/watch.go b/server/etcdserver/api/v3rpc/watch.go index c56ec4b26..6de11caab 100644 --- a/server/etcdserver/api/v3rpc/watch.go +++ b/server/etcdserver/api/v3rpc/watch.go @@ -223,16 +223,16 @@ func (ws *watchServer) Watch(stream pb.Watch_WatchServer) (err error) { return err } -func (sws *serverWatchStream) isWatchPermitted(wcr *pb.WatchCreateRequest) bool { +func (sws *serverWatchStream) isWatchPermitted(wcr *pb.WatchCreateRequest) error { authInfo, err := sws.ag.AuthInfoFromCtx(sws.gRPCStream.Context()) if err != nil { - return false + return err } if authInfo == nil { // if auth is enabled, IsRangePermitted() can cause an error authInfo = &auth.AuthInfo{} } - return sws.ag.AuthStore().IsRangePermitted(authInfo, wcr.Key, wcr.RangeEnd) == nil + return sws.ag.AuthStore().IsRangePermitted(authInfo, wcr.Key, wcr.RangeEnd) } func (sws *serverWatchStream) recvLoop() error { @@ -266,13 +266,29 @@ func (sws *serverWatchStream) recvLoop() error { creq.RangeEnd = []byte{} } - if !sws.isWatchPermitted(creq) { + err := sws.isWatchPermitted(creq) + if err != nil { + var cancelReason string + switch err { + case auth.ErrInvalidAuthToken: + cancelReason = rpctypes.ErrGRPCInvalidAuthToken.Error() + case auth.ErrAuthOldRevision: + cancelReason = rpctypes.ErrGRPCAuthOldRevision.Error() + case auth.ErrUserEmpty: + cancelReason = rpctypes.ErrGRPCUserEmpty.Error() + default: + if err != auth.ErrPermissionDenied { + sws.lg.Error("unexpected error code", zap.Error(err)) + } + cancelReason = rpctypes.ErrGRPCPermissionDenied.Error() + } + wr := &pb.WatchResponse{ Header: sws.newResponseHeader(sws.watchStream.Rev()), WatchId: creq.WatchId, Canceled: true, Created: true, - CancelReason: rpctypes.ErrGRPCPermissionDenied.Error(), + CancelReason: cancelReason, } select { diff --git a/tests/integration/cluster.go b/tests/integration/cluster.go index 0d7b47ce4..a99c554ba 100644 --- a/tests/integration/cluster.go +++ b/tests/integration/cluster.go @@ -136,7 +136,8 @@ type ClusterConfig struct { DiscoveryURL string - AuthToken string + AuthToken string + AuthTokenTTL uint UseGRPC bool @@ -314,6 +315,7 @@ func (c *cluster) mustNewMember(t testutil.TB, memberNumber int64) *member { name: c.generateMemberName(), memberNumber: memberNumber, authToken: c.cfg.AuthToken, + authTokenTTL: c.cfg.AuthTokenTTL, peerTLS: c.cfg.PeerTLS, clientTLS: c.cfg.ClientTLS, quotaBackendBytes: c.cfg.QuotaBackendBytes, @@ -624,6 +626,7 @@ type memberConfig struct { peerTLS *transport.TLSInfo clientTLS *transport.TLSInfo authToken string + authTokenTTL uint quotaBackendBytes int64 maxTxnOps uint maxRequestBytes uint @@ -715,6 +718,9 @@ func mustNewMember(t testutil.TB, mcfg memberConfig) *member { if mcfg.authToken != "" { m.AuthToken = mcfg.authToken } + if mcfg.authTokenTTL != 0 { + m.TokenTTL = mcfg.authTokenTTL + } m.BcryptCost = uint(bcrypt.MinCost) // use min bcrypt cost to speedy up integration testing diff --git a/tests/integration/v3_auth_test.go b/tests/integration/v3_auth_test.go index 89a1377b5..21c87a4f9 100644 --- a/tests/integration/v3_auth_test.go +++ b/tests/integration/v3_auth_test.go @@ -498,3 +498,36 @@ func TestV3AuthRestartMember(t *testing.T) { _, err = c2.Put(context.TODO(), "foo", "bar2") testutil.AssertNil(t, err) } + +func TestV3AuthWatchAndTokenExpire(t *testing.T) { + BeforeTest(t) + clus := NewClusterV3(t, &ClusterConfig{Size: 1, AuthTokenTTL: 3}) + defer clus.Terminate(t) + + ctx, cancel := context.WithTimeout(context.TODO(), 10*time.Second) + defer cancel() + + authSetupRoot(t, toGRPC(clus.Client(0)).Auth) + + c, cerr := 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)) + watchResponse := <-wChan + + time.Sleep(5 * time.Second) + + wChan = c.Watch(ctx, "key", clientv3.WithRev(1)) + watchResponse = <-wChan + testutil.AssertNil(t, watchResponse.Err()) +}