Merge pull request #21030 from tjungblu/www_lls

Author: fuweid

server/etcdserver/api/v3rpc/metrics.go

@@ -52,11 +52,63 @@ var (
 		},
 		[]string{"type", "client_api_version"},
 	)
+
+	watchSendLoopWatchStreamDuration = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "etcd_debugging",
+			Subsystem: "server",
+			Name:      "watch_send_loop_watch_stream_duration_seconds",
+			Help:      "The total duration in seconds of running through the send loop watch stream response all events.",
+			// lowest bucket start of upper bound 0.001 sec (1 ms) with factor 2
+			// highest bucket start of 0.001 sec * 2^13 == 8.192 sec
+			Buckets: prometheus.ExponentialBuckets(0.001, 2, 14),
+		},
+	)
+
+	watchSendLoopWatchStreamDurationPerEvent = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "etcd_debugging",
+			Subsystem: "server",
+			Name:      "watch_send_loop_watch_stream_duration_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.001 sec (1 ms) with factor 2
+			// highest bucket start of 0.001 sec * 2^13 == 8.192 sec
+			Buckets: prometheus.ExponentialBuckets(0.001, 2, 14),
+		},
+	)
+
+	watchSendLoopControlStreamDuration = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "etcd_debugging",
+			Subsystem: "server",
+			Name:      "watch_send_loop_control_stream_duration_seconds",
+			Help:      "The total duration in seconds of running through the send loop control stream response.",
+			// lowest bucket start of upper bound 0.001 sec (1 ms) with factor 2
+			// highest bucket start of 0.001 sec * 2^13 == 8.192 sec
+			Buckets: prometheus.ExponentialBuckets(0.001, 2, 14),
+		},
+	)
+
+	watchSendLoopProgressDuration = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "etcd_debugging",
+			Subsystem: "server",
+			Name:      "watch_send_loop_progress_duration_seconds",
+			Help:      "The total duration in seconds of running through the progress loop control stream response.",
+			// lowest bucket start of upper bound 0.001 sec (1 ms) with factor 2
+			// highest bucket start of 0.001 sec * 2^13 == 8.192 sec
+			Buckets: prometheus.ExponentialBuckets(0.001, 2, 14),
+		},
+	)
 )
 
 func init() {
 	prometheus.MustRegister(sentBytes)
 	prometheus.MustRegister(receivedBytes)
 	prometheus.MustRegister(streamFailures)
 	prometheus.MustRegister(clientRequests)
+	prometheus.MustRegister(watchSendLoopWatchStreamDuration)
+	prometheus.MustRegister(watchSendLoopWatchStreamDurationPerEvent)
+	prometheus.MustRegister(watchSendLoopControlStreamDuration)
+	prometheus.MustRegister(watchSendLoopProgressDuration)
 }

server/etcdserver/api/v3rpc/watch.go

@@ -432,6 +432,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.
@@ -502,11 +503,15 @@ func (sws *serverWatchStream) sendLoop() {
 			}
 			sws.mu.Unlock()
 
+			totalDur := time.Since(start)
+			watchSendLoopWatchStreamDuration.Observe(totalDur.Seconds())
+			watchSendLoopWatchStreamDurationPerEvent.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))
@@ -544,7 +549,11 @@ func (sws *serverWatchStream) sendLoop() {
 				delete(pending, wid)
 			}
 
+			watchSendLoopControlStreamDuration.Observe(time.Since(start).Seconds())
+
 		case <-progressTicker.C:
+			start := time.Now()
+
 			sws.mu.Lock()
 			for id, ok := range sws.progress {
 				if ok {
@@ -553,6 +562,7 @@ func (sws *serverWatchStream) sendLoop() {
 				sws.progress[id] = true
 			}
 			sws.mu.Unlock()
+			watchSendLoopProgressDuration.Observe(time.Since(start).Seconds())
 
 		case <-sws.closec:
 			return

tests/e2e/metrics_test.go

@@ -160,6 +160,10 @@ func TestNoMetricsMissing(t *testing.T) {
 			"etcd_debugging_mvcc_watch_stream_total",
 			"etcd_debugging_mvcc_watcher_total",
 			"etcd_debugging_server_lease_expired_total",
+			"etcd_debugging_server_watch_send_loop_watch_stream_duration_seconds",
+			"etcd_debugging_server_watch_send_loop_watch_stream_duration_per_event_seconds",
+			"etcd_debugging_server_watch_send_loop_control_stream_duration_seconds",
+			"etcd_debugging_server_watch_send_loop_progress_duration_seconds",
 			"etcd_debugging_snap_save_marshalling_duration_seconds",
 			"etcd_debugging_snap_save_total_duration_seconds",
 			"etcd_debugging_store_expires_total",