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 {