Merge pull request #52302 from smarterclayton/simplify_metrics_registration

Automatic merge from submit-queue. If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>..

Collapse all metrics handlers into common code

Remove the MonitorRequest method and replace with a method that takes
request.RequestInfo, which is our default way to talk about API objects.
Preserves existing semantics for calls.

Not for 1.8, but fixes the ugliness and code duplication in #52237

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

@@ -54,22 +54,18 @@ type ProxyHandler struct {
 
 func (r *ProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	reqStart := time.Now()
-	proxyHandlerTraceID := rand.Int63()
 
-	var verb, apiResource, subresource, scope string
 	var httpCode int
-
+	var requestInfo *request.RequestInfo
 	defer func() {
 		responseLength := 0
 		if rw, ok := w.(*metrics.ResponseWriterDelegator); ok {
 			responseLength = rw.ContentLength()
+			if httpCode == 0 {
+				httpCode = rw.Status()
+			}
 		}
-		metrics.Monitor(
+		metrics.Record(req, requestInfo, w.Header().Get("Content-Type"), httpCode, responseLength, time.Now().Sub(reqStart))
-			verb, apiResource, subresource, scope,
-			net.GetHTTPClient(req),
-			w.Header().Get("Content-Type"),
-			httpCode, responseLength, reqStart,
-		)
 	}()
 
 	ctx, ok := r.Mapper.Get(req)
@@ -79,32 +75,32 @@ func (r *ProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	requestInfo, ok := request.RequestInfoFrom(ctx)
+	requestInfo, ok = request.RequestInfoFrom(ctx)
 	if !ok {
 		responsewriters.InternalError(w, req, errors.New("Error getting RequestInfo from context"))
 		httpCode = http.StatusInternalServerError
 		return
 	}
+
+	metrics.RecordLongRunning(req, requestInfo, func() {
+		httpCode = r.serveHTTP(w, req, ctx, requestInfo)
+	})
+}
+
+// serveHTTP performs proxy handling and returns the status code of the operation.
+func (r *ProxyHandler) serveHTTP(w http.ResponseWriter, req *http.Request, ctx request.Context, requestInfo *request.RequestInfo) int {
+	proxyHandlerTraceID := rand.Int63()
+
 	if !requestInfo.IsResourceRequest {
 		responsewriters.NotFound(w, req)
-		httpCode = http.StatusNotFound
+		return http.StatusNotFound
-		return
-	}
-	verb = requestInfo.Verb
-	namespace, resource, subresource, parts := requestInfo.Namespace, requestInfo.Resource, requestInfo.Subresource, requestInfo.Parts
-	scope = "cluster"
-	if namespace != "" {
-		scope = "namespace"
-	}
-	if requestInfo.Name != "" {
-		scope = "resource"
 	}
+	namespace, resource, parts := requestInfo.Namespace, requestInfo.Resource, requestInfo.Parts
 
 	ctx = request.WithNamespace(ctx, namespace)
 	if len(parts) < 2 {
 		responsewriters.NotFound(w, req)
-		httpCode = http.StatusNotFound
+		return http.StatusNotFound
-		return
 	}
 	id := parts[1]
 	remainder := ""
@@ -122,31 +118,26 @@ func (r *ProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	if !ok {
 		httplog.LogOf(req, w).Addf("'%v' has no storage object", resource)
 		responsewriters.NotFound(w, req)
-		httpCode = http.StatusNotFound
+		return http.StatusNotFound
-		return
 	}
-	apiResource = resource
 
 	gv := schema.GroupVersion{Group: requestInfo.APIGroup, Version: requestInfo.APIVersion}
 
 	redirector, ok := storage.(rest.Redirector)
 	if !ok {
 		httplog.LogOf(req, w).Addf("'%v' is not a redirector", resource)
-		httpCode = responsewriters.ErrorNegotiated(ctx, apierrors.NewMethodNotSupported(schema.GroupResource{Resource: resource}, "proxy"), r.Serializer, gv, w, req)
+		return responsewriters.ErrorNegotiated(ctx, apierrors.NewMethodNotSupported(schema.GroupResource{Resource: resource}, "proxy"), r.Serializer, gv, w, req)
-		return
 	}
 
 	location, roundTripper, err := redirector.ResourceLocation(ctx, id)
 	if err != nil {
 		httplog.LogOf(req, w).Addf("Error getting ResourceLocation: %v", err)
-		httpCode = responsewriters.ErrorNegotiated(ctx, err, r.Serializer, gv, w, req)
+		return responsewriters.ErrorNegotiated(ctx, err, r.Serializer, gv, w, req)
-		return
 	}
 	if location == nil {
 		httplog.LogOf(req, w).Addf("ResourceLocation for %v returned nil", id)
 		responsewriters.NotFound(w, req)
-		httpCode = http.StatusNotFound
+		return http.StatusNotFound
-		return
 	}
 
 	if roundTripper != nil {
@@ -179,7 +170,7 @@ func (r *ProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	// https://github.com/openshift/origin/blob/master/pkg/util/httpproxy/upgradeawareproxy.go.
 	// That proxy needs to be modified to support multiple backends, not just 1.
 	if r.tryUpgrade(ctx, w, req, newReq, location, roundTripper, gv) {
-		return
+		return http.StatusSwitchingProtocols
 	}
 
 	// Redirect requests of the form "/{resource}/{name}" to "/{resource}/{name}/"
@@ -192,7 +183,7 @@ func (r *ProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		}
 		w.Header().Set("Location", req.URL.Path+"/"+queryPart)
 		w.WriteHeader(http.StatusMovedPermanently)
-		return
+		return http.StatusMovedPermanently
 	}
 
 	start := time.Now()
@@ -224,6 +215,7 @@ func (r *ProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	proxy.Transport = roundTripper
 	proxy.FlushInterval = 200 * time.Millisecond
 	proxy.ServeHTTP(w, newReq)
+	return 0
 }
 
 // tryUpgrade returns true if the request was handled.

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/responsewriters/BUILD

@@ -42,6 +42,7 @@ go_library(
         "//vendor/k8s.io/apiserver/pkg/audit:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/authorization/authorizer:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/endpoints/handlers/negotiation:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/endpoints/metrics:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/endpoints/request:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/registry/rest:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/storage:go_default_library",

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/responsewriters/writers.go

@@ -28,6 +28,7 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apiserver/pkg/audit"
 	"k8s.io/apiserver/pkg/endpoints/handlers/negotiation"
+	"k8s.io/apiserver/pkg/endpoints/metrics"
 	"k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/registry/rest"
 	"k8s.io/apiserver/pkg/util/flushwriter"
@@ -42,7 +43,10 @@ import (
 func WriteObject(ctx request.Context, statusCode int, gv schema.GroupVersion, s runtime.NegotiatedSerializer, object runtime.Object, w http.ResponseWriter, req *http.Request) {
 	stream, ok := object.(rest.ResourceStreamer)
 	if ok {
-		StreamObject(ctx, statusCode, gv, s, stream, w, req)
+		requestInfo, _ := request.RequestInfoFrom(ctx)
+		metrics.RecordLongRunning(req, requestInfo, func() {
+			StreamObject(ctx, statusCode, gv, s, stream, w, req)
+		})
 		return
 	}
 	WriteObjectNegotiated(ctx, s, gv, w, req, statusCode, object)

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

@@ -46,6 +46,7 @@ import (
 	"k8s.io/apiserver/pkg/audit"
 	"k8s.io/apiserver/pkg/endpoints/handlers/negotiation"
 	"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
+	"k8s.io/apiserver/pkg/endpoints/metrics"
 	"k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/registry/rest"
 	utiltrace "k8s.io/apiserver/pkg/util/trace"
@@ -261,12 +262,15 @@ func ConnectResource(connecter rest.Connecter, scope RequestScope, admit admissi
 				return
 			}
 		}
-		handler, err := connecter.Connect(ctx, name, opts, &responder{scope: scope, req: req, w: w})
+		requestInfo, _ := request.RequestInfoFrom(ctx)
-		if err != nil {
+		metrics.RecordLongRunning(req, requestInfo, func() {
-			scope.err(err, w, req)
+			handler, err := connecter.Connect(ctx, name, opts, &responder{scope: scope, req: req, w: w})
-			return
+			if err != nil {
-		}
+				scope.err(err, w, req)
-		handler.ServeHTTP(w, req)
+				return
+			}
+			handler.ServeHTTP(w, req)
+		})
 	}
 }
 
@@ -366,7 +370,10 @@ func ListResource(r rest.Lister, rw rest.Watcher, scope RequestScope, forceWatch
 				scope.err(err, w, req)
 				return
 			}
-			serveWatch(watcher, scope, req, w, timeout)
+			requestInfo, _ := request.RequestInfoFrom(ctx)
+			metrics.RecordLongRunning(req, requestInfo, func() {
+				serveWatch(watcher, scope, req, w, timeout)
+			})
 			return
 		}
 

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

@@ -19,6 +19,7 @@ go_library(
         "//vendor/github.com/emicklei/go-restful:go_default_library",
         "//vendor/github.com/prometheus/client_golang/prometheus:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/net:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/endpoints/request:go_default_library",
     ],
 )
 

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

@@ -27,7 +27,7 @@ import (
 	"time"
 
 	utilnet "k8s.io/apimachinery/pkg/util/net"
-	//utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apiserver/pkg/endpoints/request"
 
 	"github.com/emicklei/go-restful"
 	"github.com/prometheus/client_golang/prometheus"
@@ -43,6 +43,13 @@ var (
 		},
 		[]string{"verb", "resource", "subresource", "scope", "client", "contentType", "code"},
 	)
+	longRunningRequestGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "apiserver_longrunning_gauge",
+			Help: "Gauge of all active long-running apiserver requests broken out by verb, API resource, and scope. Not all requests are tracked this way.",
+		},
+		[]string{"verb", "resource", "subresource", "scope"},
+	)
 	requestLatencies = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
 			Name: "apiserver_request_latencies",
@@ -77,43 +84,59 @@ var (
 // Register all metrics.
 func Register() {
 	prometheus.MustRegister(requestCounter)
+	prometheus.MustRegister(longRunningRequestGauge)
 	prometheus.MustRegister(requestLatencies)
 	prometheus.MustRegister(requestLatenciesSummary)
 	prometheus.MustRegister(responseSizes)
 }
 
-// Monitor records a request to the apiserver endpoints that follow the Kubernetes API conventions.  verb must be
+// Record records a single request to the standard metrics endpoints. For use by handlers that perform their own
-// uppercase to be backwards compatible with existing monitoring tooling.
+// processing. All API paths should use InstrumentRouteFunc implicitly. Use this instead of MonitorRequest if
-func Monitor(verb, resource, subresource, scope, client, contentType string, httpCode, respSize int, reqStart time.Time) {
+// you already have a RequestInfo object.
-	elapsed := float64((time.Since(reqStart)) / time.Microsecond)
+func Record(req *http.Request, requestInfo *request.RequestInfo, contentType string, code int, responseSizeInBytes int, elapsed time.Duration) {
-	requestCounter.WithLabelValues(verb, resource, subresource, scope, client, contentType, codeToString(httpCode)).Inc()
+	if requestInfo == nil {
-	requestLatencies.WithLabelValues(verb, resource, subresource, scope).Observe(elapsed)
+		requestInfo = &request.RequestInfo{Verb: req.Method, Path: req.URL.Path}
-	requestLatenciesSummary.WithLabelValues(verb, resource, subresource, scope).Observe(elapsed)
-	// We are only interested in response sizes of read requests.
-	if verb == "GET" || verb == "LIST" {
-		responseSizes.WithLabelValues(verb, resource, subresource, scope).Observe(float64(respSize))
 	}
+	scope := cleanScope(requestInfo)
+	if requestInfo.IsResourceRequest {
+		MonitorRequest(req, strings.ToUpper(requestInfo.Verb), requestInfo.Resource, requestInfo.Subresource, contentType, scope, code, responseSizeInBytes, elapsed)
+	} else {
+		MonitorRequest(req, strings.ToUpper(requestInfo.Verb), "", requestInfo.Path, contentType, scope, code, responseSizeInBytes, elapsed)
+	}
+}
+
+// RecordLongRunning tracks the execution of a long running request against the API server. It provides an accurate count
+// of the total number of open long running requests. requestInfo may be nil if the caller is not in the normal request flow.
+func RecordLongRunning(req *http.Request, requestInfo *request.RequestInfo, fn func()) {
+	if requestInfo == nil {
+		requestInfo = &request.RequestInfo{Verb: req.Method, Path: req.URL.Path}
+	}
+	var g prometheus.Gauge
+	scope := cleanScope(requestInfo)
+	reportedVerb := cleanVerb(strings.ToUpper(requestInfo.Verb), req)
+	if requestInfo.IsResourceRequest {
+		g = longRunningRequestGauge.WithLabelValues(reportedVerb, requestInfo.Resource, requestInfo.Subresource, scope)
+	} else {
+		g = longRunningRequestGauge.WithLabelValues(reportedVerb, "", requestInfo.Path, scope)
+	}
+	g.Inc()
+	defer g.Dec()
+	fn()
 }
 
 // MonitorRequest handles standard transformations for client and the reported verb and then invokes Monitor to record
 // a request. verb must be uppercase to be backwards compatible with existing monitoring tooling.
-func MonitorRequest(request *http.Request, verb, resource, subresource, scope, contentType string, httpCode, respSize int, reqStart time.Time) {
+func MonitorRequest(req *http.Request, verb, resource, subresource, scope, contentType string, httpCode, respSize int, elapsed time.Duration) {
-	reportedVerb := verb
+	reportedVerb := cleanVerb(verb, req)
-	if verb == "LIST" {
+	client := cleanUserAgent(utilnet.GetHTTPClient(req))
-		// see apimachinery/pkg/runtime/conversion.go Convert_Slice_string_To_bool
+	elapsedMicroseconds := float64(elapsed / time.Microsecond)
-		if values := request.URL.Query()["watch"]; len(values) > 0 {
+	requestCounter.WithLabelValues(reportedVerb, resource, subresource, scope, client, contentType, codeToString(httpCode)).Inc()
-			if value := strings.ToLower(values[0]); value != "0" && value != "false" {
+	requestLatencies.WithLabelValues(reportedVerb, resource, subresource, scope).Observe(elapsedMicroseconds)
-				reportedVerb = "WATCH"
+	requestLatenciesSummary.WithLabelValues(reportedVerb, resource, subresource, scope).Observe(elapsedMicroseconds)
-			}
+	// We are only interested in response sizes of read requests.
-		}
+	if verb == "GET" || verb == "LIST" {
+		responseSizes.WithLabelValues(reportedVerb, resource, subresource, scope).Observe(float64(respSize))
 	}
-	// normalize the legacy WATCHLIST to WATCH to ensure users aren't surprised by metrics
-	if verb == "WATCHLIST" {
-		reportedVerb = "WATCH"
-	}
-
-	client := cleanUserAgent(utilnet.GetHTTPClient(request))
-	Monitor(reportedVerb, resource, subresource, scope, client, contentType, httpCode, respSize, reqStart)
 }
 
 func Reset() {
@@ -144,10 +167,41 @@ func InstrumentRouteFunc(verb, resource, subresource, scope string, routeFunc re
 
 		routeFunc(request, response)
 
-		MonitorRequest(request.Request, verb, resource, subresource, scope, delegate.Header().Get("Content-Type"), delegate.Status(), delegate.ContentLength(), now)
+		MonitorRequest(request.Request, verb, resource, subresource, scope, delegate.Header().Get("Content-Type"), delegate.Status(), delegate.ContentLength(), time.Now().Sub(now))
 	})
 }
 
+func cleanScope(requestInfo *request.RequestInfo) string {
+	if requestInfo.Namespace != "" {
+		return "namespace"
+	}
+	if requestInfo.Name != "" {
+		return "resource"
+	}
+	if requestInfo.IsResourceRequest {
+		return "cluster"
+	}
+	// this is the empty scope
+	return ""
+}
+
+func cleanVerb(verb string, request *http.Request) string {
+	reportedVerb := verb
+	if verb == "LIST" {
+		// see apimachinery/pkg/runtime/conversion.go Convert_Slice_string_To_bool
+		if values := request.URL.Query()["watch"]; len(values) > 0 {
+			if value := strings.ToLower(values[0]); value != "0" && value != "false" {
+				reportedVerb = "WATCH"
+			}
+		}
+	}
+	// normalize the legacy WATCHLIST to WATCH to ensure users aren't surprised by metrics
+	if verb == "WATCHLIST" {
+		reportedVerb = "WATCH"
+	}
+	return reportedVerb
+}
+
 func cleanUserAgent(ua string) string {
 	// We collapse all "web browser"-type user agents into one "browser" to reduce metric cardinality.
 	if strings.HasPrefix(ua, "Mozilla/") {

staging/src/k8s.io/apiserver/pkg/server/filters/maxinflight.go

@@ -19,8 +19,6 @@ package filters
 import (
 	"fmt"
 	"net/http"
-	"strings"
-	"time"
 
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apiserver/pkg/authentication/user"
@@ -108,18 +106,7 @@ func WithMaxInFlightLimit(
 						}
 					}
 				}
-				scope := "cluster"
+				metrics.Record(r, requestInfo, "", http.StatusTooManyRequests, 0, 0)
-				if requestInfo.Namespace != "" {
-					scope = "namespace"
-				}
-				if requestInfo.Name != "" {
-					scope = "resource"
-				}
-				if requestInfo.IsResourceRequest {
-					metrics.MonitorRequest(r, strings.ToUpper(requestInfo.Verb), requestInfo.Resource, requestInfo.Subresource, "", scope, http.StatusTooManyRequests, 0, time.Now())
-				} else {
-					metrics.MonitorRequest(r, strings.ToUpper(requestInfo.Verb), "", requestInfo.Path, "", scope, http.StatusTooManyRequests, 0, time.Now())
-				}
 				tooManyRequests(r, w)
 			}
 		}

staging/src/k8s.io/apiserver/pkg/server/filters/timeout.go

@@ -22,7 +22,6 @@ import (
 	"fmt"
 	"net"
 	"net/http"
-	"strings"
 	"sync"
 	"time"
 
@@ -55,20 +54,8 @@ func WithTimeoutForNonLongRunningRequests(handler http.Handler, requestContextMa
 		if longRunning(req, requestInfo) {
 			return nil, nil, nil
 		}
-		now := time.Now()
 		metricFn := func() {
-			scope := "cluster"
+			metrics.Record(req, requestInfo, "", http.StatusGatewayTimeout, 0, 0)
-			if requestInfo.Namespace != "" {
-				scope = "namespace"
-			}
-			if requestInfo.Name != "" {
-				scope = "resource"
-			}
-			if requestInfo.IsResourceRequest {
-				metrics.MonitorRequest(req, strings.ToUpper(requestInfo.Verb), requestInfo.Resource, requestInfo.Subresource, "", scope, http.StatusGatewayTimeout, 0, now)
-			} else {
-				metrics.MonitorRequest(req, strings.ToUpper(requestInfo.Verb), "", requestInfo.Path, "", scope, http.StatusGatewayTimeout, 0, now)
-			}
 		}
 		return time.After(timeout), metricFn, apierrors.NewTimeoutError(fmt.Sprintf("request did not complete within %s", timeout), 0)
 	}