Instrument watchstream send loop

This PR adds a few more timing metrics to the watch send loop.
We're currently trying to track down some sort of contention affecting
all gRPC endpoints, we see a hard QPS ceiling when the apiserver restarts with
watch cache inits.

There's currently no metric to help us pinpoint where the time is going
to create such a ceiling, thus adding some more metrics here.

Somewhat related to etcd-io#15402.

Signed-off-by: Thomas Jungblut <tjungblu@redhat.com>

Author: tjungblu

server/etcdserver/api/v3rpc/metrics.go

@@ -52,11 +52,60 @@ var (
 		},
 		[]string{"type", "client_api_version"},
 	)
+
+	watchSendLoopWatchStreamTime = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "etcd_debugging",
+			Subsystem: "server",
+			Name:      "watch_send_loop_watch_stream_time_seconds",
+			Help:      "The total duration in seconds of running through the send loop watch stream response all events.",
+		},
+	)
+
+	watchSendLoopWatchStreamTimePerEvent = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "etcd_debugging",
+			Subsystem: "server",
+			Name:      "watch_send_loop_watch_stream_time_per_event_seconds",
+			Help:      "The average duration in seconds of running through the send loop watch stream response, per event.",
+			// lowest bucket start of upper bound 0.0001 sec (0.1 ms) with factor 2
+			// highest bucket start of 0.0001 sec * 2^15 == 3.2768 sec
+			Buckets: prometheus.ExponentialBuckets(0.0001, 2, 16),
+		},
+	)
+
+	watchSendLoopControlStreamTime = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "etcd_debugging",
+			Subsystem: "server",
+			Name:      "watch_send_loop_control_stream_time_seconds",
+			Help:      "The total duration in seconds of running through the send loop control stream response.",
+			// lowest bucket start of upper bound 0.0001 sec (0.1 ms) with factor 2
+			// highest bucket start of 0.0001 sec * 2^15 == 3.2768 sec
+			Buckets: prometheus.ExponentialBuckets(0.0001, 2, 16),
+		},
+	)
+
+	watchSendLoopProgressTime = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "etcd_debugging",
+			Subsystem: "server",
+			Name:      "watch_send_loop_progress_time_seconds",
+			Help:      "The total duration in seconds of running through the progress loop control stream response.",
+			// lowest bucket start of upper bound 0.0001 sec (0.1 ms) with factor 2
+			// highest bucket start of 0.0001 sec * 2^15 == 3.2768 sec
+			Buckets: prometheus.ExponentialBuckets(0.0001, 2, 16),
+		},
+	)
 )
 
 func init() {
 	prometheus.MustRegister(sentBytes)
 	prometheus.MustRegister(receivedBytes)
 	prometheus.MustRegister(streamFailures)
 	prometheus.MustRegister(clientRequests)
+	prometheus.MustRegister(watchSendLoopWatchStreamTime)
+	prometheus.MustRegister(watchSendLoopWatchStreamTimePerEvent)
+	prometheus.MustRegister(watchSendLoopControlStreamTime)
+	prometheus.MustRegister(watchSendLoopProgressTime)
 }

server/etcdserver/api/v3rpc/watch.go

@@ -405,6 +405,7 @@ func (sws *serverWatchStream) sendLoop() {
 				return
 			}
 
+			start := time.Now()
 			// TODO: evs is []mvccpb.Event type
 			// either return []*mvccpb.Event from the mvcc package
 			// or define protocol buffer with []mvccpb.Event.
@@ -475,11 +476,16 @@ func (sws *serverWatchStream) sendLoop() {
 			}
 			sws.mu.Unlock()
 
+			totalDur := time.Since(start)
+			sws.lg.Info("watch response to gRPC stream", zap.Duration("duration", totalDur))
+			watchSendLoopWatchStreamTime.Observe(totalDur.Seconds())
+			watchSendLoopWatchStreamTimePerEvent.Observe(totalDur.Seconds() / float64(len(evs)))
+
 		case c, ok := <-sws.ctrlStream:
 			if !ok {
 				return
 			}
-
+			start := time.Now()
 			if err := sws.gRPCStream.Send(c); err != nil {
 				if isClientCtxErr(sws.gRPCStream.Context().Err(), err) {
 					sws.lg.Debug("failed to send watch control response to gRPC stream", zap.Error(err))
@@ -517,7 +523,11 @@ func (sws *serverWatchStream) sendLoop() {
 				delete(pending, wid)
 			}
 
+			watchSendLoopControlStreamTime.Observe(time.Since(start).Seconds())
+
 		case <-progressTicker.C:
+			start := time.Now()
+
 			sws.mu.Lock()
 			for id, ok := range sws.progress {
 				if ok {
@@ -526,6 +536,7 @@ func (sws *serverWatchStream) sendLoop() {
 				sws.progress[id] = true
 			}
 			sws.mu.Unlock()
+			watchSendLoopProgressTime.Observe(time.Since(start).Seconds())
 
 		case <-sws.closec:
 			return