diff --git a/clientv3/watch.go b/clientv3/watch.go index 66e16ad63..c064858c6 100644 --- a/clientv3/watch.go +++ b/clientv3/watch.go @@ -18,6 +18,7 @@ import ( "context" "errors" "fmt" + "strings" "sync" "time" @@ -577,6 +578,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 ws := w.resuming[0]; ws != nil { w.addSubstream(pbresp, ws) @@ -695,6 +716,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/etcdserver/api/v3rpc/watch.go b/etcdserver/api/v3rpc/watch.go index 6a9204024..dbbe75414 100644 --- a/etcdserver/api/v3rpc/watch.go +++ b/etcdserver/api/v3rpc/watch.go @@ -232,16 +232,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 { @@ -275,13 +275,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/integration/cluster.go b/integration/cluster.go index 11d44c965..d881f8137 100644 --- a/integration/cluster.go +++ b/integration/cluster.go @@ -126,7 +126,8 @@ type ClusterConfig struct { DiscoveryURL string - AuthToken string + AuthToken string + AuthTokenTTL uint UseGRPC bool @@ -285,6 +286,7 @@ func (c *cluster) mustNewMember(t testing.TB) *member { memberConfig{ name: c.name(rand.Int()), authToken: c.cfg.AuthToken, + authTokenTTL: c.cfg.AuthTokenTTL, peerTLS: c.cfg.PeerTLS, clientTLS: c.cfg.ClientTLS, quotaBackendBytes: c.cfg.QuotaBackendBytes, @@ -579,6 +581,7 @@ type memberConfig struct { peerTLS *transport.TLSInfo clientTLS *transport.TLSInfo authToken string + authTokenTTL uint quotaBackendBytes int64 maxTxnOps uint maxRequestBytes uint @@ -665,6 +668,9 @@ func mustNewMember(t testing.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/integration/v3_auth_test.go b/integration/v3_auth_test.go index 03c01ef6d..bccc77ed5 100644 --- a/integration/v3_auth_test.go +++ b/integration/v3_auth_test.go @@ -471,3 +471,36 @@ func TestV3AuthRestartMember(t *testing.T) { _, err = c2.Put(context.TODO(), "foo", "bar2") testutil.AssertNil(t, err) } + +func TestV3AuthWatchAndTokenExpire(t *testing.T) { + defer testutil.AfterTest(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 := clientv3.New(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()) +}