Merge pull request #9760 from gyuho/stream

etcdserver/api/v3rpc: add "etcd_network_server_stream_failures_total"

etcdserver/api/v3rpc/lease.go

@@ -122,6 +122,7 @@ func (ls *LeaseServer) leaseKeepAlive(stream pb.Lease_LeaseKeepAliveServer) erro
 				} else {
 					plog.Warningf("failed to receive lease keepalive request from gRPC stream (%q)", err.Error())
 				}
+				streamFailures.WithLabelValues("receive", "lease-keepalive").Inc()
 			}
 			return err
 		}
@@ -160,6 +161,7 @@ func (ls *LeaseServer) leaseKeepAlive(stream pb.Lease_LeaseKeepAliveServer) erro
 				} else {
 					plog.Warningf("failed to send lease keepalive response to gRPC stream (%q)", err.Error())
 				}
+				streamFailures.WithLabelValues("send", "lease-keepalive").Inc()
 			}
 			return err
 		}

etcdserver/api/v3rpc/metrics.go

@@ -30,9 +30,19 @@ var (
 		Name:      "client_grpc_received_bytes_total",
 		Help:      "The total number of bytes received from grpc clients.",
 	})
+
+	streamFailures = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "etcd",
+		Subsystem: "network",
+		Name:      "server_stream_failures_total",
+		Help:      "The total number of stream failures from the local server.",
+	},
+		[]string{"Type", "API"},
+	)
 )
 
 func init() {
 	prometheus.MustRegister(sentBytes)
 	prometheus.MustRegister(receivedBytes)
+	prometheus.MustRegister(streamFailures)
 }

etcdserver/api/v3rpc/watch.go

@@ -183,6 +183,7 @@ func (ws *watchServer) Watch(stream pb.Watch_WatchServer) (err error) {
 				} else {
 					plog.Warningf("failed to receive watch request from gRPC stream (%q)", rerr.Error())
 				}
+				streamFailures.WithLabelValues("receive", "watch").Inc()
 			}
 			errc <- rerr
 		}
@@ -416,6 +417,7 @@ func (sws *serverWatchStream) sendLoop() {
 					} else {
 						plog.Warningf("failed to send watch response to gRPC stream (%q)", serr.Error())
 					}
+					streamFailures.WithLabelValues("send", "watch").Inc()
 				}
 				return
 			}
@@ -445,6 +447,7 @@ func (sws *serverWatchStream) sendLoop() {
 					} else {
 						plog.Warningf("failed to send watch control response to gRPC stream (%q)", err.Error())
 					}
+					streamFailures.WithLabelValues("send", "watch").Inc()
 				}
 				return
 			}
@@ -473,6 +476,7 @@ func (sws *serverWatchStream) sendLoop() {
 							} else {
 								plog.Warningf("failed to send pending watch response to gRPC stream (%q)", err.Error())
 							}
+							streamFailures.WithLabelValues("send", "watch").Inc()
 						}
 						return
 					}