Merge pull request #14547 from mitake/3.5-backport-14322

Backport PR 14322 to release-3.5

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 {

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 {

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
 

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())
+}