Merge pull request #111988 from kkkkun/update-connnect-metrics

Differentiate CONNECT requests from other requests

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

@@ -321,6 +321,14 @@ var (
 		"UPDATE",
 		"WATCH",
 		"WATCHLIST")
+
+	// These are the valid connect requests which we report in our metrics.
+	validConnectRequests = utilsets.NewString(
+		"log",
+		"exec",
+		"portforward",
+		"attach",
+		"proxy")
 )
 
 const (
@@ -427,7 +435,7 @@ func RecordRequestAbort(req *http.Request, requestInfo *request.RequestInfo) {
 	}
 
 	scope := CleanScope(requestInfo)
-	reportedVerb := cleanVerb(CanonicalVerb(strings.ToUpper(req.Method), scope), "", req)
+	reportedVerb := cleanVerb(CanonicalVerb(strings.ToUpper(req.Method), scope), "", req, requestInfo)
 	resource := requestInfo.Resource
 	subresource := requestInfo.Subresource
 	group := requestInfo.APIGroup
@@ -448,7 +456,7 @@ func RecordDroppedRequest(req *http.Request, requestInfo *request.RequestInfo, c
 	// 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), "", req)
+	reportedVerb := cleanVerb(CanonicalVerb(strings.ToUpper(req.Method), scope), "", req, requestInfo)
 
 	if requestInfo.IsResourceRequest {
 		requestCounter.WithContext(req.Context()).WithLabelValues(reportedVerb, dryRun, requestInfo.APIGroup, requestInfo.APIVersion, requestInfo.Resource, requestInfo.Subresource, scope, component, codeToString(http.StatusTooManyRequests)).Inc()
@@ -471,7 +479,7 @@ func RecordRequestTermination(req *http.Request, requestInfo *request.RequestInf
 	// 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), "", req)
+	reportedVerb := cleanVerb(CanonicalVerb(strings.ToUpper(req.Method), scope), "", req, requestInfo)
 
 	if requestInfo.IsResourceRequest {
 		requestTerminationsTotal.WithContext(req.Context()).WithLabelValues(reportedVerb, requestInfo.APIGroup, requestInfo.APIVersion, requestInfo.Resource, requestInfo.Subresource, scope, component, codeToString(code)).Inc()
@@ -493,7 +501,7 @@ func RecordLongRunning(req *http.Request, requestInfo *request.RequestInfo, comp
 	// 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), "", req)
+	reportedVerb := cleanVerb(CanonicalVerb(strings.ToUpper(req.Method), scope), "", req, requestInfo)
 
 	if requestInfo.IsResourceRequest {
 		g = longRunningRequestsGauge.WithContext(req.Context()).WithLabelValues(reportedVerb, requestInfo.APIGroup, requestInfo.APIVersion, requestInfo.Resource, requestInfo.Subresource, scope, component)
@@ -512,7 +520,7 @@ func MonitorRequest(req *http.Request, verb, group, version, resource, subresour
 	// 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), verb, req)
+	reportedVerb := cleanVerb(CanonicalVerb(strings.ToUpper(req.Method), scope), verb, req, nil)
 
 	dryRun := cleanDryRun(req.URL)
 	elapsedSeconds := elapsed.Seconds()
@@ -585,15 +593,16 @@ func InstrumentHandlerFunc(verb, group, version, resource, subresource, scope, c
 // NormalizedVerb returns normalized verb
 func NormalizedVerb(req *http.Request) string {
 	verb := req.Method
-	if requestInfo, ok := request.RequestInfoFrom(req.Context()); ok {
+	requestInfo, ok := request.RequestInfoFrom(req.Context())
+	if ok {
 		// If we can find a requestInfo, we can get a scope, and then
 		// we can convert GETs to LISTs when needed.
 		scope := CleanScope(requestInfo)
 		verb = CanonicalVerb(strings.ToUpper(verb), scope)
 	}
 
-	// mark APPLY requests and WATCH requests correctly.
+	// mark APPLY requests, WATCH requests and CONNECT requests correctly.
-	return CleanVerb(verb, req)
+	return CleanVerb(verb, req, requestInfo)
 }
 
 // CleanScope returns the scope of the request.
@@ -626,8 +635,8 @@ func CanonicalVerb(verb string, scope string) string {
 }
 
 // CleanVerb returns a normalized verb, so that it is easy to tell WATCH from
-// LIST and APPLY from PATCH.
+// LIST, APPLY from PATCH and CONNECT from others.
-func CleanVerb(verb string, request *http.Request) string {
+func CleanVerb(verb string, request *http.Request, requestInfo *request.RequestInfo) string {
 	reportedVerb := verb
 	if suggestedVerb := getVerbIfWatch(request); suggestedVerb == "WATCH" {
 		reportedVerb = "WATCH"
@@ -639,11 +648,14 @@ func CleanVerb(verb string, request *http.Request) string {
 	if verb == "PATCH" && request.Header.Get("Content-Type") == string(types.ApplyPatchType) {
 		reportedVerb = "APPLY"
 	}
+	if requestInfo != nil && requestInfo.IsResourceRequest && len(requestInfo.Subresource) > 0 && validConnectRequests.Has(requestInfo.Subresource) {
+		reportedVerb = "CONNECT"
+	}
 	return reportedVerb
 }
 
 // cleanVerb additionally ensures that unknown verbs don't clog up the metrics.
-func cleanVerb(verb, suggestedVerb string, request *http.Request) string {
+func cleanVerb(verb, suggestedVerb string, request *http.Request, requestInfo *request.RequestInfo) string {
 	// CanonicalVerb (being an input for this function) doesn't handle correctly the
 	// deprecated path pattern for watch of:
 	//   GET /api/{version}/watch/{resource}
@@ -651,7 +663,7 @@ func cleanVerb(verb, suggestedVerb string, request *http.Request) string {
 	if suggestedVerb == "WATCH" || suggestedVerb == "WATCHLIST" {
 		return "WATCH"
 	}
-	reportedVerb := CleanVerb(verb, request)
+	reportedVerb := CleanVerb(verb, request, requestInfo)
 	if validRequestMethods.Has(reportedVerb) {
 		return reportedVerb
 	}

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

@@ -34,6 +34,7 @@ func TestCleanVerb(t *testing.T) {
 		initialVerb   string
 		suggestedVerb string
 		request       *http.Request
+		requestInfo   *request.RequestInfo
 		expectedVerb  string
 	}{
 		{
@@ -142,6 +143,89 @@ func TestCleanVerb(t *testing.T) {
 			request:      nil,
 			expectedVerb: "other",
 		},
+		{
+			desc:        "Pod logs should be transformed to CONNECT",
+			initialVerb: "GET",
+			request: &http.Request{
+				Method: "GET",
+				URL: &url.URL{
+					RawQuery: "/api/v1/namespaces/default/pods/test-pod/log",
+				},
+			},
+			requestInfo: &request.RequestInfo{
+				Verb:              "GET",
+				Resource:          "pods",
+				IsResourceRequest: true,
+				Subresource:       "log",
+			},
+			expectedVerb: "CONNECT",
+		},
+		{
+			desc:        "Pod exec should be transformed to CONNECT",
+			initialVerb: "POST",
+			request: &http.Request{
+				Method: "POST",
+				URL: &url.URL{
+					RawQuery: "/api/v1/namespaces/default/pods/test-pod/exec?command=sh",
+				},
+				Header: map[string][]string{
+					"Connection": {"Upgrade"},
+					"Upgrade":    {"SPDY/3.1"},
+					"X-Stream-Protocol-Version": {
+						"v4.channel.k8s.io", "v3.channel.k8s.io", "v2.channel.k8s.io", "channel.k8s.io",
+					},
+				},
+			},
+			requestInfo: &request.RequestInfo{
+				Verb:              "POST",
+				Resource:          "pods",
+				IsResourceRequest: true,
+				Subresource:       "exec",
+			},
+			expectedVerb: "CONNECT",
+		},
+		{
+			desc:        "Pod portforward should be transformed to CONNECT",
+			initialVerb: "POST",
+			request: &http.Request{
+				Method: "POST",
+				URL: &url.URL{
+					RawQuery: "/api/v1/namespaces/default/pods/test-pod/portforward",
+				},
+				Header: map[string][]string{
+					"Connection": {"Upgrade"},
+					"Upgrade":    {"SPDY/3.1"},
+					"X-Stream-Protocol-Version": {
+						"v4.channel.k8s.io", "v3.channel.k8s.io", "v2.channel.k8s.io", "channel.k8s.io",
+					},
+				},
+			},
+			requestInfo: &request.RequestInfo{
+				Verb:              "POST",
+				Resource:          "pods",
+				IsResourceRequest: true,
+				Subresource:       "portforward",
+			},
+			expectedVerb: "CONNECT",
+		},
+		{
+			desc:        "Deployment scale should not be transformed to CONNECT",
+			initialVerb: "PUT",
+			request: &http.Request{
+				Method: "PUT",
+				URL: &url.URL{
+					RawQuery: "/apis/apps/v1/namespaces/default/deployments/test-1/scale",
+				},
+				Header: map[string][]string{},
+			},
+			requestInfo: &request.RequestInfo{
+				Verb:              "PUT",
+				Resource:          "deployments",
+				IsResourceRequest: true,
+				Subresource:       "scale",
+			},
+			expectedVerb: "PUT",
+		},
 	}
 	for _, tt := range testCases {
 		t.Run(tt.initialVerb, func(t *testing.T) {
@@ -149,7 +233,7 @@ func TestCleanVerb(t *testing.T) {
 			if tt.request != nil {
 				req = tt.request
 			}
-			cleansedVerb := cleanVerb(tt.initialVerb, tt.suggestedVerb, req)
+			cleansedVerb := cleanVerb(tt.initialVerb, tt.suggestedVerb, req, tt.requestInfo)
 			if cleansedVerb != tt.expectedVerb {
 				t.Errorf("Got %s, but expected %s", cleansedVerb, tt.expectedVerb)
 			}