Track number of registered watchers in apiserver

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch.go

@@ -30,6 +30,7 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/apiserver/pkg/endpoints/handlers/negotiation"
+	"k8s.io/apiserver/pkg/endpoints/metrics"
 	"k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/server/httplog"
 	"k8s.io/apiserver/pkg/util/wsstream"
@@ -140,6 +141,10 @@ type WatchServer struct {
 // ServeHTTP serves a series of encoded events via HTTP with Transfer-Encoding: chunked
 // or over a websocket connection.
 func (s *WatchServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	kind := s.Scope.Kind
+	metrics.RegisteredWatchers.WithLabelValues(kind.Group, kind.Version, kind.Kind).Inc()
+	defer metrics.RegisteredWatchers.WithLabelValues(kind.Group, kind.Version, kind.Kind).Dec()
+
 	w = httplog.Unlogged(w)
 
 	if wsstream.IsWebSocketRequest(req) {

staging/src/k8s.io/apiserver/pkg/endpoints/metrics/metrics.go

@@ -93,6 +93,14 @@ var (
 		},
 		[]string{"requestKind"},
 	)
+	// RegisteredWatchers is a number of currently registered watchers splitted by resource.
+	RegisteredWatchers = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "apiserver_registered_watchers",
+			Help: "Number of currently registered watchers for a given resources",
+		},
+		[]string{"group", "version", "kind"},
+	)
 	// Because of volatality of the base metric this is pre-aggregated one. Instead of reporing current usage all the time
 	// it reports maximal usage during the last second.
 	currentInflightRequests = prometheus.NewGaugeVec(
@@ -111,6 +119,7 @@ var (
 		requestLatenciesSummary,
 		responseSizes,
 		DroppedRequests,
+		RegisteredWatchers,
 		currentInflightRequests,
 	}
 )