From a66d7c3ba56dbf48bfe5b8224bc71bd4e7b41191 Mon Sep 17 00:00:00 2001
From: "Sahdev P. Zala" <spzala@us.ibm.com>
Date: Fri, 5 Jan 2018 11:24:50 -0500
Subject: [PATCH] etcdserver/api/v3rpc: debug user cancellation and log warning
 for rest

The context error with cancel code is typically for user cancellation which
should be at debug level. For other error codes we should display a warning.

Fixes #9085
---
 etcdserver/api/v3rpc/lease.go | 12 ++++++++++--
 etcdserver/api/v3rpc/util.go  | 13 +++++++++++++
 etcdserver/api/v3rpc/watch.go | 24 ++++++++++++++++++++----
 3 files changed, 43 insertions(+), 6 deletions(-)

diff --git a/etcdserver/api/v3rpc/lease.go b/etcdserver/api/v3rpc/lease.go
index 13602b4e652..5b4f2b14228 100644
--- a/etcdserver/api/v3rpc/lease.go
+++ b/etcdserver/api/v3rpc/lease.go
@@ -107,7 +107,11 @@ func (ls *LeaseServer) leaseKeepAlive(stream pb.Lease_LeaseKeepAliveServer) erro
 			return nil
 		}
 		if err != nil {
-			plog.Debugf("failed to receive lease keepalive request from gRPC stream (%q)", err.Error())
+			if isClientCtxErr(stream.Context().Err(), err) {
+				plog.Debugf("failed to receive lease keepalive request from gRPC stream (%q)", err.Error())
+			} else {
+				plog.Warningf("failed to receive lease keepalive request from gRPC stream (%q)", err.Error())
+			}
 			return err
 		}
 
@@ -133,7 +137,11 @@ func (ls *LeaseServer) leaseKeepAlive(stream pb.Lease_LeaseKeepAliveServer) erro
 		resp.TTL = ttl
 		err = stream.Send(resp)
 		if err != nil {
-			plog.Debugf("failed to send lease keepalive response to gRPC stream (%q)", err.Error())
+			if isClientCtxErr(stream.Context().Err(), err) {
+				plog.Debugf("failed to send lease keepalive response to gRPC stream (%q)", err.Error())
+			} else {
+				plog.Warningf("failed to send lease keepalive response to gRPC stream (%q)", err.Error())
+			}
 			return err
 		}
 	}
diff --git a/etcdserver/api/v3rpc/util.go b/etcdserver/api/v3rpc/util.go
index 71a549b0555..e4bb85ae755 100644
--- a/etcdserver/api/v3rpc/util.go
+++ b/etcdserver/api/v3rpc/util.go
@@ -81,3 +81,16 @@ func togRPCError(err error) error {
 	}
 	return grpcErr
 }
+
+func isClientCtxErr(ctxErr error, err error) bool {
+	if ctxErr != nil {
+		return true
+	}
+
+	ev, ok := status.FromError(err)
+	if !ok {
+		return false
+	}
+	code := ev.Code()
+	return code == codes.Canceled || code == codes.DeadlineExceeded
+}
diff --git a/etcdserver/api/v3rpc/watch.go b/etcdserver/api/v3rpc/watch.go
index b2f8169ded9..dd4f329515d 100644
--- a/etcdserver/api/v3rpc/watch.go
+++ b/etcdserver/api/v3rpc/watch.go
@@ -140,7 +140,11 @@ func (ws *watchServer) Watch(stream pb.Watch_WatchServer) (err error) {
 	// deadlock when calling sws.close().
 	go func() {
 		if rerr := sws.recvLoop(); rerr != nil {
-			plog.Debugf("failed to receive watch request from gRPC stream (%q)", rerr.Error())
+			if isClientCtxErr(stream.Context().Err(), rerr) {
+				plog.Debugf("failed to receive watch request from gRPC stream (%q)", rerr.Error())
+			} else {
+				plog.Warningf("failed to receive watch request from gRPC stream (%q)", rerr.Error())
+			}
 			errc <- rerr
 		}
 	}()
@@ -339,7 +343,11 @@ func (sws *serverWatchStream) sendLoop() {
 
 			mvcc.ReportEventReceived(len(evs))
 			if err := sws.gRPCStream.Send(wr); err != nil {
-				plog.Debugf("failed to send watch response to gRPC stream (%q)", err.Error())
+				if isClientCtxErr(sws.gRPCStream.Context().Err(), err) {
+					plog.Debugf("failed to send watch response to gRPC stream (%q)", err.Error())
+				} else {
+					plog.Warningf("failed to send watch response to gRPC stream (%q)", err.Error())
+				}
 				return
 			}
 
@@ -356,7 +364,11 @@ func (sws *serverWatchStream) sendLoop() {
 			}
 
 			if err := sws.gRPCStream.Send(c); err != nil {
-				plog.Debugf("failed to send watch control response to gRPC stream (%q)", err.Error())
+				if isClientCtxErr(sws.gRPCStream.Context().Err(), err) {
+					plog.Debugf("failed to send watch control response to gRPC stream (%q)", err.Error())
+				} else {
+					plog.Warningf("failed to send watch control response to gRPC stream (%q)", err.Error())
+				}
 				return
 			}
 
@@ -372,7 +384,11 @@ func (sws *serverWatchStream) sendLoop() {
 				for _, v := range pending[wid] {
 					mvcc.ReportEventReceived(len(v.Events))
 					if err := sws.gRPCStream.Send(v); err != nil {
-						plog.Debugf("failed to send pending watch response to gRPC stream (%q)", err.Error())
+						if isClientCtxErr(sws.gRPCStream.Context().Err(), err) {
+							plog.Debugf("failed to send pending watch response to gRPC stream (%q)", err.Error())
+						} else {
+							plog.Warningf("failed to send pending watch response to gRPC stream (%q)", err.Error())
+						}
 						return
 					}
 				}