Merge pull request #108927 from wojtek-t/unify_pf_metrics

Record dropped requests in apiserver_request_total metric

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

@@ -132,8 +132,11 @@ var (
 		},
 		[]string{"verb", "group", "version", "resource", "subresource", "scope", "component"},
 	)
-	// DroppedRequests is a number of requests dropped with 'Try again later' response"
-	DroppedRequests = compbasemetrics.NewCounterVec(
+	// droppedRequests is a number of requests dropped with 'Try again later' response"
+	//
+	// TODO(wojtek-t): This metric can be inferred both from requestTerminationsTotal as well as
+	// from requestCounter. We should deprecate and remove it.
+	droppedRequests = compbasemetrics.NewCounterVec(
 		&compbasemetrics.CounterOpts{
 			Name:           "apiserver_dropped_requests_total",
 			Help:           "Number of requests dropped with 'Try again later' response",
@@ -261,7 +264,7 @@ var (
 		requestLatencies,
 		requestSloLatencies,
 		responseSizes,
-		DroppedRequests,
+		droppedRequests,
 		TLSHandshakeErrors,
 		RegisteredWatchers,
 		WatchEvents,
@@ -403,6 +406,33 @@ func RecordRequestAbort(req *http.Request, requestInfo *request.RequestInfo) {
 	requestAbortsTotal.WithContext(req.Context()).WithLabelValues(reportedVerb, group, version, resource, subresource, scope).Inc()
 }
 
+// RecordDroppedRequest records that the request was rejected via http.TooManyRequests.
+func RecordDroppedRequest(req *http.Request, requestInfo *request.RequestInfo, component string, isMutatingRequest bool) {
+	if requestInfo == nil {
+		requestInfo = &request.RequestInfo{Verb: req.Method, Path: req.URL.Path}
+	}
+	scope := CleanScope(requestInfo)
+	dryRun := cleanDryRun(req.URL)
+
+	// We don't use verb from <requestInfo>, as this may be propagated from
+	// InstrumentRouteFunc which is registered in installer.go with predefined
+	// list of verbs (different than those translated to RequestInfo).
+	// However, we need to tweak it e.g. to differentiate GET from LIST.
+	reportedVerb := cleanVerb(CanonicalVerb(strings.ToUpper(req.Method), scope), getVerbIfWatch(req), req)
+
+	if requestInfo.IsResourceRequest {
+		requestCounter.WithContext(req.Context()).WithLabelValues(reportedVerb, dryRun, requestInfo.APIGroup, requestInfo.APIVersion, requestInfo.Resource, requestInfo.Subresource, scope, component, codeToString(http.StatusTooManyRequests)).Inc()
+	} else {
+		requestCounter.WithContext(req.Context()).WithLabelValues(reportedVerb, dryRun, "", "", "", requestInfo.Subresource, scope, component, codeToString(http.StatusTooManyRequests)).Inc()
+	}
+
+	if isMutatingRequest {
+		droppedRequests.WithContext(req.Context()).WithLabelValues(MutatingKind).Inc()
+	} else {
+		droppedRequests.WithContext(req.Context()).WithLabelValues(ReadOnlyKind).Inc()
+	}
+}
+
 // RecordRequestTermination records that the request was terminated early as part of a resource
 // preservation or apiserver self-defense mechanism (e.g. timeouts, maxinflight throttling,
 // proxyHandler errors). RecordRequestTermination should only be called zero or one times

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

@@ -19,10 +19,13 @@ package metrics
 import (
 	"net/http"
 	"net/url"
+	"strings"
 	"testing"
 
 	"k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/endpoints/responsewriter"
+	"k8s.io/component-base/metrics/legacyregistry"
+	"k8s.io/component-base/metrics/testutil"
 )
 
 func TestCleanVerb(t *testing.T) {
@@ -211,3 +214,114 @@ func TestResponseWriterDecorator(t *testing.T) {
 		t.Errorf("Expected the decorator to return the inner http.ResponseWriter object")
 	}
 }
+
+func TestRecordDroppedRequests(t *testing.T) {
+	testedMetrics := []string{
+		"apiserver_request_total",
+		"apiserver_dropped_requests_total",
+	}
+
+	testCases := []struct {
+		desc        string
+		request     *http.Request
+		requestInfo *request.RequestInfo
+		isMutating  bool
+		want        string
+	}{
+		{
+			desc: "list pods",
+			request: &http.Request{
+				Method: "GET",
+				URL: &url.URL{
+					RawPath: "/api/v1/pods",
+				},
+			},
+			requestInfo: &request.RequestInfo{
+				APIGroup:          "",
+				APIVersion:        "v1",
+				Resource:          "pods",
+				IsResourceRequest: true,
+			},
+			isMutating: false,
+			want: `
+			            # HELP apiserver_dropped_requests_total [ALPHA] Number of requests dropped with 'Try again later' response
+			            # TYPE apiserver_dropped_requests_total counter
+			            apiserver_dropped_requests_total{request_kind="readOnly"} 1
+			            # HELP apiserver_request_total [STABLE] Counter of apiserver requests broken out for each verb, dry run value, group, version, resource, scope, component, and HTTP response code.
+			            # TYPE apiserver_request_total counter
+			            apiserver_request_total{code="429",component="apiserver",dry_run="",group="",resource="pods",scope="cluster",subresource="",verb="LIST",version="v1"} 1
+				`,
+		},
+		{
+			desc: "post pods",
+			request: &http.Request{
+				Method: "POST",
+				URL: &url.URL{
+					RawPath: "/api/v1/namespaces/foo/pods",
+				},
+			},
+			requestInfo: &request.RequestInfo{
+				APIGroup:          "",
+				APIVersion:        "v1",
+				Resource:          "pods",
+				IsResourceRequest: true,
+			},
+			isMutating: true,
+			want: `
+			            # HELP apiserver_dropped_requests_total [ALPHA] Number of requests dropped with 'Try again later' response
+			            # TYPE apiserver_dropped_requests_total counter
+			            apiserver_dropped_requests_total{request_kind="mutating"} 1
+			            # HELP apiserver_request_total [STABLE] Counter of apiserver requests broken out for each verb, dry run value, group, version, resource, scope, component, and HTTP response code.
+			            # TYPE apiserver_request_total counter
+			            apiserver_request_total{code="429",component="apiserver",dry_run="",group="",resource="pods",scope="cluster",subresource="",verb="POST",version="v1"} 1
+				`,
+		},
+		{
+			desc: "dry-run patch job status",
+			request: &http.Request{
+				Method: "PATCH",
+				URL: &url.URL{
+					RawPath:  "/apis/batch/v1/namespaces/foo/pods/status",
+					RawQuery: "dryRun=All",
+				},
+			},
+			requestInfo: &request.RequestInfo{
+				APIGroup:          "batch",
+				APIVersion:        "v1",
+				Resource:          "jobs",
+				Subresource:       "status",
+				IsResourceRequest: true,
+			},
+			isMutating: true,
+			want: `
+			            # HELP apiserver_dropped_requests_total [ALPHA] Number of requests dropped with 'Try again later' response
+			            # TYPE apiserver_dropped_requests_total counter
+			            apiserver_dropped_requests_total{request_kind="mutating"} 1
+			            # HELP apiserver_request_total [STABLE] Counter of apiserver requests broken out for each verb, dry run value, group, version, resource, scope, component, and HTTP response code.
+			            # TYPE apiserver_request_total counter
+			            apiserver_request_total{code="429",component="apiserver",dry_run="All",group="batch",resource="jobs",scope="cluster",subresource="status",verb="PATCH",version="v1"} 1
+				`,
+		},
+	}
+
+	// Since prometheus' gatherer is global, other tests may have updated metrics already, so
+	// we need to reset them prior running this test.
+	// This also implies that we can't run this test in parallel with other tests.
+	Register()
+	requestCounter.Reset()
+	droppedRequests.Reset()
+
+	for _, test := range testCases {
+		t.Run(test.desc, func(t *testing.T) {
+			defer requestCounter.Reset()
+			defer droppedRequests.Reset()
+
+			RecordDroppedRequest(test.request, test.requestInfo, APIServerComponent, test.isMutating)
+
+			if err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, strings.NewReader(test.want), testedMetrics...); err != nil {
+				t.Fatal(err)
+			}
+
+		})
+	}
+}

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

@@ -198,11 +198,7 @@ func WithMaxInFlightLimit(
 					}
 				}
 				// We need to split this data between buckets used for throttling.
-				if isMutatingRequest {
-					metrics.DroppedRequests.WithContext(ctx).WithLabelValues(metrics.MutatingKind).Inc()
-				} else {
-					metrics.DroppedRequests.WithContext(ctx).WithLabelValues(metrics.ReadOnlyKind).Inc()
-				}
+				metrics.RecordDroppedRequest(r, requestInfo, metrics.APIServerComponent, isMutatingRequest)
 				metrics.RecordRequestTermination(r, requestInfo, metrics.APIServerComponent, http.StatusTooManyRequests)
 				tooManyRequests(r, w)
 			}

staging/src/k8s.io/apiserver/pkg/server/filters/priority-and-fairness.go

@@ -290,11 +290,7 @@ func WithPriorityAndFairness(
 		if !served {
 			setResponseHeaders(classification, w)
 
-			if isMutatingRequest {
-				epmetrics.DroppedRequests.WithContext(ctx).WithLabelValues(epmetrics.MutatingKind).Inc()
-			} else {
-				epmetrics.DroppedRequests.WithContext(ctx).WithLabelValues(epmetrics.ReadOnlyKind).Inc()
-			}
+			epmetrics.RecordDroppedRequest(r, requestInfo, epmetrics.APIServerComponent, isMutatingRequest)
 			epmetrics.RecordRequestTermination(r, requestInfo, epmetrics.APIServerComponent, http.StatusTooManyRequests)
 			tooManyRequests(r, w)
 		}