*: handle auth invalid token and old revision errors in watch

Signed-off-by: Hitoshi Mitake <h.mitake@gmail.com>
mitake · Oct 4, 2022 · 0c6e466 · 0c6e466
1 parent d0a732f
commit 0c6e466
Show file tree

Hide file tree

Showing 2 changed files with 47 additions and 5 deletions.
diff --git 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
@@ -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 {