clientv3: Add auth retry to retry interceptor

clientv3/integration/dial_test.go

@@ -83,7 +83,7 @@ func TestDialTLSNoConfig(t *testing.T) {
 
 	// TODO: this should not be required when we set grpc.WithBlock()
 	if c != nil {
-		ctx, cancel := context.WithTimeout(context.Background(), integration.RequestWaitTimeout)
+		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
 		_, err = c.KV.Get(ctx, "/")
 		cancel()
 	}

clientv3/retry.go

@@ -89,413 +89,204 @@ func isNonRepeatableStopError(err error) bool {
 	return desc != "there is no address available" && desc != "there is no connection available"
 }
 
-func (c *Client) newRetryWrapper() retryRPCFunc {
-	return func(rpcCtx context.Context, f rpcFunc, rp retryPolicy) error {
-		var isStop retryStopErrFunc
-		switch rp {
-		case repeatable:
-			isStop = isRepeatableStopError
-		case nonRepeatable:
-			isStop = isNonRepeatableStopError
-		}
-		for {
-			err := f(rpcCtx)
-			if err == nil {
-				return nil
-			}
-			lg.Lvl(4).Infof("clientv3/retry: error %q", err.Error())
-
-			if s, ok := status.FromError(err); ok && (s.Code() == codes.Unavailable || s.Code() == codes.DeadlineExceeded || s.Code() == codes.Internal) {
-				lg.Lvl(4).Infof("clientv3/retry: retrying due to error %q", err.Error())
-			}
-
-			if isStop(err) {
-				return err
-			}
-		}
-	}
-}
-
-func (c *Client) newAuthRetryWrapper(retryf retryRPCFunc) retryRPCFunc {
-	return func(rpcCtx context.Context, f rpcFunc, rp retryPolicy) error {
-		for {
-			err := retryf(rpcCtx, f, rp)
-			if err == nil {
-				return nil
-			}
-			lg.Lvl(4).Infof("clientv3/auth-retry: error %q", err.Error())
-			// always stop retry on etcd errors other than invalid auth token
-			if rpctypes.Error(err) == rpctypes.ErrInvalidAuthToken {
-				gterr := c.getToken(rpcCtx)
-				if gterr != nil {
-					lg.Lvl(4).Infof("clientv3/auth-retry: cannot retry due to error %q(%q)", err.Error(), gterr.Error())
-					return err // return the original error for simplicity
-				}
-				continue
-			}
-			return err
-		}
-	}
-}
-
 type retryKVClient struct {
 	kc pb.KVClient
-	retryf retryRPCFunc
 }
 
 // RetryKVClient implements a KVClient.
 func RetryKVClient(c *Client) pb.KVClient {
 	return &retryKVClient{
 		kc: pb.NewKVClient(c.conn),
-		retryf: c.newAuthRetryWrapper(c.newRetryWrapper()),
 	}
 }
 func (rkv *retryKVClient) Range(ctx context.Context, in *pb.RangeRequest, opts ...grpc.CallOption) (resp *pb.RangeResponse, err error) {
-	err = rkv.retryf(ctx, func(rctx context.Context) error {
+	return rkv.kc.Range(ctx, in, append(opts, withRetryPolicy(repeatable))...)
-		resp, err = rkv.kc.Range(rctx, in, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return resp, err
 }
 
 func (rkv *retryKVClient) Put(ctx context.Context, in *pb.PutRequest, opts ...grpc.CallOption) (resp *pb.PutResponse, err error) {
-	err = rkv.retryf(ctx, func(rctx context.Context) error {
+	return rkv.kc.Put(ctx, in, opts...)
-		resp, err = rkv.kc.Put(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 func (rkv *retryKVClient) DeleteRange(ctx context.Context, in *pb.DeleteRangeRequest, opts ...grpc.CallOption) (resp *pb.DeleteRangeResponse, err error) {
-	err = rkv.retryf(ctx, func(rctx context.Context) error {
+	return rkv.kc.DeleteRange(ctx, in, opts...)
-		resp, err = rkv.kc.DeleteRange(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 func (rkv *retryKVClient) Txn(ctx context.Context, in *pb.TxnRequest, opts ...grpc.CallOption) (resp *pb.TxnResponse, err error) {
-	// TODO: "repeatable" for read-only txn
+	return rkv.kc.Txn(ctx, in, opts...)
-	err = rkv.retryf(ctx, func(rctx context.Context) error {
-		resp, err = rkv.kc.Txn(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 func (rkv *retryKVClient) Compact(ctx context.Context, in *pb.CompactionRequest, opts ...grpc.CallOption) (resp *pb.CompactionResponse, err error) {
-	err = rkv.retryf(ctx, func(rctx context.Context) error {
+	return rkv.kc.Compact(ctx, in, opts...)
-		resp, err = rkv.kc.Compact(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 type retryLeaseClient struct {
 	lc pb.LeaseClient
-	retryf retryRPCFunc
 }
 
 // RetryLeaseClient implements a LeaseClient.
 func RetryLeaseClient(c *Client) pb.LeaseClient {
 	return &retryLeaseClient{
 		lc: pb.NewLeaseClient(c.conn),
-		retryf: c.newAuthRetryWrapper(c.newRetryWrapper()),
 	}
 }
 
 func (rlc *retryLeaseClient) LeaseTimeToLive(ctx context.Context, in *pb.LeaseTimeToLiveRequest, opts ...grpc.CallOption) (resp *pb.LeaseTimeToLiveResponse, err error) {
-	err = rlc.retryf(ctx, func(rctx context.Context) error {
+	return rlc.lc.LeaseTimeToLive(ctx, in, append(opts, withRetryPolicy(repeatable))...)
-		resp, err = rlc.lc.LeaseTimeToLive(rctx, in, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return resp, err
 }
 
 func (rlc *retryLeaseClient) LeaseLeases(ctx context.Context, in *pb.LeaseLeasesRequest, opts ...grpc.CallOption) (resp *pb.LeaseLeasesResponse, err error) {
-	err = rlc.retryf(ctx, func(rctx context.Context) error {
+	return rlc.lc.LeaseLeases(ctx, in, append(opts, withRetryPolicy(repeatable))...)
-		resp, err = rlc.lc.LeaseLeases(rctx, in, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return resp, err
 }
 
 func (rlc *retryLeaseClient) LeaseGrant(ctx context.Context, in *pb.LeaseGrantRequest, opts ...grpc.CallOption) (resp *pb.LeaseGrantResponse, err error) {
-	err = rlc.retryf(ctx, func(rctx context.Context) error {
+	return rlc.lc.LeaseGrant(ctx, in, append(opts, withRetryPolicy(repeatable))...)
-		resp, err = rlc.lc.LeaseGrant(rctx, in, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return resp, err
-
 }
 
 func (rlc *retryLeaseClient) LeaseRevoke(ctx context.Context, in *pb.LeaseRevokeRequest, opts ...grpc.CallOption) (resp *pb.LeaseRevokeResponse, err error) {
-	err = rlc.retryf(ctx, func(rctx context.Context) error {
+	return rlc.lc.LeaseRevoke(ctx, in, append(opts, withRetryPolicy(repeatable))...)
-		resp, err = rlc.lc.LeaseRevoke(rctx, in, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return resp, err
 }
 
 func (rlc *retryLeaseClient) LeaseKeepAlive(ctx context.Context, opts ...grpc.CallOption) (stream pb.Lease_LeaseKeepAliveClient, err error) {
-	err = rlc.retryf(ctx, func(rctx context.Context) error {
+	return rlc.lc.LeaseKeepAlive(ctx, append(opts, withRetryPolicy(repeatable))...)
-		stream, err = rlc.lc.LeaseKeepAlive(rctx, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return stream, err
 }
 
 type retryClusterClient struct {
 	cc pb.ClusterClient
-	retryf retryRPCFunc
 }
 
 // RetryClusterClient implements a ClusterClient.
 func RetryClusterClient(c *Client) pb.ClusterClient {
 	return &retryClusterClient{
 		cc: pb.NewClusterClient(c.conn),
-		retryf: c.newRetryWrapper(),
 	}
 }
 
 func (rcc *retryClusterClient) MemberList(ctx context.Context, in *pb.MemberListRequest, opts ...grpc.CallOption) (resp *pb.MemberListResponse, err error) {
-	err = rcc.retryf(ctx, func(rctx context.Context) error {
+	return rcc.cc.MemberList(ctx, in, append(opts, withRetryPolicy(repeatable))...)
-		resp, err = rcc.cc.MemberList(rctx, in, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return resp, err
 }
 
 func (rcc *retryClusterClient) MemberAdd(ctx context.Context, in *pb.MemberAddRequest, opts ...grpc.CallOption) (resp *pb.MemberAddResponse, err error) {
-	err = rcc.retryf(ctx, func(rctx context.Context) error {
+	return rcc.cc.MemberAdd(ctx, in, opts...)
-		resp, err = rcc.cc.MemberAdd(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 func (rcc *retryClusterClient) MemberRemove(ctx context.Context, in *pb.MemberRemoveRequest, opts ...grpc.CallOption) (resp *pb.MemberRemoveResponse, err error) {
-	err = rcc.retryf(ctx, func(rctx context.Context) error {
+	return rcc.cc.MemberRemove(ctx, in, opts...)
-		resp, err = rcc.cc.MemberRemove(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 func (rcc *retryClusterClient) MemberUpdate(ctx context.Context, in *pb.MemberUpdateRequest, opts ...grpc.CallOption) (resp *pb.MemberUpdateResponse, err error) {
-	err = rcc.retryf(ctx, func(rctx context.Context) error {
+	return rcc.cc.MemberUpdate(ctx, in, opts...)
-		resp, err = rcc.cc.MemberUpdate(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 type retryMaintenanceClient struct {
 	mc pb.MaintenanceClient
-	retryf retryRPCFunc
 }
 
 // RetryMaintenanceClient implements a Maintenance.
 func RetryMaintenanceClient(c *Client, conn *grpc.ClientConn) pb.MaintenanceClient {
 	return &retryMaintenanceClient{
 		mc: pb.NewMaintenanceClient(conn),
-		retryf: c.newRetryWrapper(),
 	}
 }
 
 func (rmc *retryMaintenanceClient) Alarm(ctx context.Context, in *pb.AlarmRequest, opts ...grpc.CallOption) (resp *pb.AlarmResponse, err error) {
-	err = rmc.retryf(ctx, func(rctx context.Context) error {
+	return rmc.mc.Alarm(ctx, in, append(opts, withRetryPolicy(repeatable))...)
-		resp, err = rmc.mc.Alarm(rctx, in, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return resp, err
 }
 
 func (rmc *retryMaintenanceClient) Status(ctx context.Context, in *pb.StatusRequest, opts ...grpc.CallOption) (resp *pb.StatusResponse, err error) {
-	err = rmc.retryf(ctx, func(rctx context.Context) error {
+	return rmc.mc.Status(ctx, in, append(opts, withRetryPolicy(repeatable))...)
-		resp, err = rmc.mc.Status(rctx, in, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return resp, err
 }
 
 func (rmc *retryMaintenanceClient) Hash(ctx context.Context, in *pb.HashRequest, opts ...grpc.CallOption) (resp *pb.HashResponse, err error) {
-	err = rmc.retryf(ctx, func(rctx context.Context) error {
+	return rmc.mc.Hash(ctx, in, append(opts, withRetryPolicy(repeatable))...)
-		resp, err = rmc.mc.Hash(rctx, in, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return resp, err
 }
 
 func (rmc *retryMaintenanceClient) HashKV(ctx context.Context, in *pb.HashKVRequest, opts ...grpc.CallOption) (resp *pb.HashKVResponse, err error) {
-	err = rmc.retryf(ctx, func(rctx context.Context) error {
+	return rmc.mc.HashKV(ctx, in, append(opts, withRetryPolicy(repeatable))...)
-		resp, err = rmc.mc.HashKV(rctx, in, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return resp, err
 }
 
 func (rmc *retryMaintenanceClient) Snapshot(ctx context.Context, in *pb.SnapshotRequest, opts ...grpc.CallOption) (stream pb.Maintenance_SnapshotClient, err error) {
-	err = rmc.retryf(ctx, func(rctx context.Context) error {
+	return rmc.mc.Snapshot(ctx, in, append(opts, withRetryPolicy(repeatable))...)
-		stream, err = rmc.mc.Snapshot(rctx, in, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return stream, err
 }
 
 func (rmc *retryMaintenanceClient) MoveLeader(ctx context.Context, in *pb.MoveLeaderRequest, opts ...grpc.CallOption) (resp *pb.MoveLeaderResponse, err error) {
-	err = rmc.retryf(ctx, func(rctx context.Context) error {
+	return rmc.mc.MoveLeader(ctx, in, append(opts, withRetryPolicy(repeatable))...)
-		resp, err = rmc.mc.MoveLeader(rctx, in, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return resp, err
 }
 
 func (rmc *retryMaintenanceClient) Defragment(ctx context.Context, in *pb.DefragmentRequest, opts ...grpc.CallOption) (resp *pb.DefragmentResponse, err error) {
-	err = rmc.retryf(ctx, func(rctx context.Context) error {
+	return rmc.mc.Defragment(ctx, in, opts...)
-		resp, err = rmc.mc.Defragment(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 type retryAuthClient struct {
 	ac pb.AuthClient
-	retryf retryRPCFunc
 }
 
 // RetryAuthClient implements a AuthClient.
 func RetryAuthClient(c *Client) pb.AuthClient {
 	return &retryAuthClient{
 		ac: pb.NewAuthClient(c.conn),
-		retryf: c.newRetryWrapper(),
 	}
 }
 
 func (rac *retryAuthClient) UserList(ctx context.Context, in *pb.AuthUserListRequest, opts ...grpc.CallOption) (resp *pb.AuthUserListResponse, err error) {
-	err = rac.retryf(ctx, func(rctx context.Context) error {
+	return rac.ac.UserList(ctx, in, append(opts, withRetryPolicy(repeatable))...)
-		resp, err = rac.ac.UserList(rctx, in, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return resp, err
 }
 
 func (rac *retryAuthClient) UserGet(ctx context.Context, in *pb.AuthUserGetRequest, opts ...grpc.CallOption) (resp *pb.AuthUserGetResponse, err error) {
-	err = rac.retryf(ctx, func(rctx context.Context) error {
+	return rac.ac.UserGet(ctx, in, append(opts, withRetryPolicy(repeatable))...)
-		resp, err = rac.ac.UserGet(rctx, in, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return resp, err
 }
 
 func (rac *retryAuthClient) RoleGet(ctx context.Context, in *pb.AuthRoleGetRequest, opts ...grpc.CallOption) (resp *pb.AuthRoleGetResponse, err error) {
-	err = rac.retryf(ctx, func(rctx context.Context) error {
+	return rac.ac.RoleGet(ctx, in, append(opts, withRetryPolicy(repeatable))...)
-		resp, err = rac.ac.RoleGet(rctx, in, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return resp, err
 }
 
 func (rac *retryAuthClient) RoleList(ctx context.Context, in *pb.AuthRoleListRequest, opts ...grpc.CallOption) (resp *pb.AuthRoleListResponse, err error) {
-	err = rac.retryf(ctx, func(rctx context.Context) error {
+	return rac.ac.RoleList(ctx, in, append(opts, withRetryPolicy(repeatable))...)
-		resp, err = rac.ac.RoleList(rctx, in, append(opts, withRetryPolicy(repeatable))...)
-		return err
-	}, repeatable)
-	return resp, err
 }
 
 func (rac *retryAuthClient) AuthEnable(ctx context.Context, in *pb.AuthEnableRequest, opts ...grpc.CallOption) (resp *pb.AuthEnableResponse, err error) {
-	err = rac.retryf(ctx, func(rctx context.Context) error {
+	return rac.ac.AuthEnable(ctx, in, opts...)
-		resp, err = rac.ac.AuthEnable(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 func (rac *retryAuthClient) AuthDisable(ctx context.Context, in *pb.AuthDisableRequest, opts ...grpc.CallOption) (resp *pb.AuthDisableResponse, err error) {
-	err = rac.retryf(ctx, func(rctx context.Context) error {
+	return rac.ac.AuthDisable(ctx, in, opts...)
-		resp, err = rac.ac.AuthDisable(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 func (rac *retryAuthClient) UserAdd(ctx context.Context, in *pb.AuthUserAddRequest, opts ...grpc.CallOption) (resp *pb.AuthUserAddResponse, err error) {
-	err = rac.retryf(ctx, func(rctx context.Context) error {
+	return rac.ac.UserAdd(ctx, in, opts...)
-		resp, err = rac.ac.UserAdd(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 func (rac *retryAuthClient) UserDelete(ctx context.Context, in *pb.AuthUserDeleteRequest, opts ...grpc.CallOption) (resp *pb.AuthUserDeleteResponse, err error) {
-	err = rac.retryf(ctx, func(rctx context.Context) error {
+	return rac.ac.UserDelete(ctx, in, opts...)
-		resp, err = rac.ac.UserDelete(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 func (rac *retryAuthClient) UserChangePassword(ctx context.Context, in *pb.AuthUserChangePasswordRequest, opts ...grpc.CallOption) (resp *pb.AuthUserChangePasswordResponse, err error) {
-	err = rac.retryf(ctx, func(rctx context.Context) error {
+	return rac.ac.UserChangePassword(ctx, in, opts...)
-		resp, err = rac.ac.UserChangePassword(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 func (rac *retryAuthClient) UserGrantRole(ctx context.Context, in *pb.AuthUserGrantRoleRequest, opts ...grpc.CallOption) (resp *pb.AuthUserGrantRoleResponse, err error) {
-	err = rac.retryf(ctx, func(rctx context.Context) error {
+	return rac.ac.UserGrantRole(ctx, in, opts...)
-		resp, err = rac.ac.UserGrantRole(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 func (rac *retryAuthClient) UserRevokeRole(ctx context.Context, in *pb.AuthUserRevokeRoleRequest, opts ...grpc.CallOption) (resp *pb.AuthUserRevokeRoleResponse, err error) {
-	err = rac.retryf(ctx, func(rctx context.Context) error {
+	return rac.ac.UserRevokeRole(ctx, in, opts...)
-		resp, err = rac.ac.UserRevokeRole(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 func (rac *retryAuthClient) RoleAdd(ctx context.Context, in *pb.AuthRoleAddRequest, opts ...grpc.CallOption) (resp *pb.AuthRoleAddResponse, err error) {
-	err = rac.retryf(ctx, func(rctx context.Context) error {
+	return rac.ac.RoleAdd(ctx, in, opts...)
-		resp, err = rac.ac.RoleAdd(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 func (rac *retryAuthClient) RoleDelete(ctx context.Context, in *pb.AuthRoleDeleteRequest, opts ...grpc.CallOption) (resp *pb.AuthRoleDeleteResponse, err error) {
-	err = rac.retryf(ctx, func(rctx context.Context) error {
+	return rac.ac.RoleDelete(ctx, in, opts...)
-		resp, err = rac.ac.RoleDelete(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 func (rac *retryAuthClient) RoleGrantPermission(ctx context.Context, in *pb.AuthRoleGrantPermissionRequest, opts ...grpc.CallOption) (resp *pb.AuthRoleGrantPermissionResponse, err error) {
-	err = rac.retryf(ctx, func(rctx context.Context) error {
+	return rac.ac.RoleGrantPermission(ctx, in, opts...)
-		resp, err = rac.ac.RoleGrantPermission(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 func (rac *retryAuthClient) RoleRevokePermission(ctx context.Context, in *pb.AuthRoleRevokePermissionRequest, opts ...grpc.CallOption) (resp *pb.AuthRoleRevokePermissionResponse, err error) {
-	err = rac.retryf(ctx, func(rctx context.Context) error {
+	return rac.ac.RoleRevokePermission(ctx, in, opts...)
-		resp, err = rac.ac.RoleRevokePermission(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }
 
 func (rac *retryAuthClient) Authenticate(ctx context.Context, in *pb.AuthenticateRequest, opts ...grpc.CallOption) (resp *pb.AuthenticateResponse, err error) {
-	err = rac.retryf(ctx, func(rctx context.Context) error {
+	return rac.ac.Authenticate(ctx, in, opts...)
-		resp, err = rac.ac.Authenticate(rctx, in, opts...)
-		return err
-	}, nonRepeatable)
-	return resp, err
 }

clientv3/retry_interceptor.go

@@ -23,6 +23,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
 	"github.com/grpc-ecosystem/go-grpc-middleware/util/backoffutils"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
@@ -57,10 +58,17 @@ func (c *Client) unaryClientInterceptor(logger *zap.Logger, optFuncs ...retryOpt
 				if ctx.Err() != nil {
 					// its the context deadline or cancellation.
 					return lastErr
-				} else {
+				}
 				// its the callCtx deadline or cancellation, in which case try again.
 				continue
 			}
+			if callOpts.retryAuth && rpctypes.Error(lastErr) == rpctypes.ErrInvalidAuthToken {
+				gterr := c.getToken(ctx)
+				if gterr != nil {
+					logger.Info("retry failed to fetch new auth token", zap.Error(gterr))
+					return lastErr // return the original error for simplicity
+				}
+				continue
 			}
 			if !isRetriable(lastErr, callOpts) {
 				return lastErr
@@ -97,6 +105,7 @@ func (c *Client) streamClientInterceptor(logger *zap.Logger, optFuncs ...retryOp
 			return nil, err
 		}
 		retryingStreamer := &serverStreamingRetryingStream{
+			client:       c,
 			ClientStream: newStreamer,
 			callOpts:     callOpts,
 			ctx:          ctx,
@@ -113,6 +122,7 @@ func (c *Client) streamClientInterceptor(logger *zap.Logger, optFuncs ...retryOp
 // a new ClientStream according to the retry policy.
 type serverStreamingRetryingStream struct {
 	grpc.ClientStream
+	client        *Client
 	bufferedSends []interface{} // single messsage that the client can sen
 	receivedGood  bool          // indicates whether any prior receives were successful
 	wasClosedSend bool          // indicates that CloseSend was closed
@@ -198,10 +208,18 @@ func (s *serverStreamingRetryingStream) receiveMsgAndIndicateRetry(m interface{}
 	if isContextError(err) {
 		if s.ctx.Err() != nil {
 			return false, err
-		} else {
+		}
 		// its the callCtx deadline or cancellation, in which case try again.
 		return true, err
 	}
+	if s.callOpts.retryAuth && rpctypes.Error(err) == rpctypes.ErrInvalidAuthToken {
+		gterr := s.client.getToken(s.ctx)
+		if gterr != nil {
+			logger.Info("retry failed to fetch new auth token", zap.Error(gterr))
+			return false, err // return the original error for simplicity
+		}
+		return true, err
+
 	}
 	return isRetriable(err, s.callOpts), err
 
@@ -278,6 +296,7 @@ var (
 		retryPolicy: nonRepeatable,
 		max:         0, // disabed
 		backoffFunc: backoffLinearWithJitter(50*time.Millisecond /*jitter*/, 0.10),
+		retryAuth:   true,
 	}
 )
 
@@ -296,6 +315,13 @@ func withRetryPolicy(rp retryPolicy) retryOption {
 	}}
 }
 
+// withAuthRetry sets enables authentication retries.
+func withAuthRetry(retryAuth bool) retryOption {
+	return retryOption{applyFunc: func(o *options) {
+		o.retryAuth = retryAuth
+	}}
+}
+
 // withMax sets the maximum number of retries on this call, or this interceptor.
 func withMax(maxRetries uint) retryOption {
 	return retryOption{applyFunc: func(o *options) {
@@ -314,6 +340,7 @@ type options struct {
 	retryPolicy retryPolicy
 	max         uint
 	backoffFunc backoffFunc
+	retryAuth   bool
 }
 
 // retryOption is a grpc.CallOption that is local to clientv3's retry interceptor.