Merge pull request #95207 from tkashem/apf-filter-latency

Add a new metric to measure apiserver request filter latency

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

@@ -110,6 +110,7 @@ filegroup(
         ":package-srcs",
         "//staging/src/k8s.io/apiserver/pkg/endpoints/deprecation:all-srcs",
         "//staging/src/k8s.io/apiserver/pkg/endpoints/discovery:all-srcs",
+        "//staging/src/k8s.io/apiserver/pkg/endpoints/filterlatency:all-srcs",
         "//staging/src/k8s.io/apiserver/pkg/endpoints/filters:all-srcs",
         "//staging/src/k8s.io/apiserver/pkg/endpoints/handlers:all-srcs",
         "//staging/src/k8s.io/apiserver/pkg/endpoints/metrics:all-srcs",

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

@@ -0,0 +1,35 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["filterlatency.go"],
+    importmap = "k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/endpoints/filterlatency",
+    importpath = "k8s.io/apiserver/pkg/endpoints/filterlatency",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//staging/src/k8s.io/apimachinery/pkg/util/clock:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/endpoints/metrics:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/endpoints/request:go_default_library",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["filterlatency_test.go"],
+    embed = [":go_default_library"],
+    deps = ["//staging/src/k8s.io/apimachinery/pkg/util/clock:go_default_library"],
+)

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

@@ -0,0 +1,96 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package filterlatency
+
+import (
+	"context"
+	"net/http"
+	"time"
+
+	utilclock "k8s.io/apimachinery/pkg/util/clock"
+	"k8s.io/apiserver/pkg/endpoints/metrics"
+	apirequest "k8s.io/apiserver/pkg/endpoints/request"
+)
+
+type requestFilterRecordKeyType int
+
+// requestFilterRecordKey is the context key for a request filter record struct.
+const requestFilterRecordKey requestFilterRecordKeyType = iota
+
+type requestFilterRecord struct {
+	name             string
+	startedTimestamp time.Time
+}
+
+// withRequestFilterRecord attaches the given request filter record to the parent context.
+func withRequestFilterRecord(parent context.Context, fr *requestFilterRecord) context.Context {
+	return apirequest.WithValue(parent, requestFilterRecordKey, fr)
+}
+
+// requestFilterRecordFrom returns the request filter record from the given context.
+func requestFilterRecordFrom(ctx context.Context) *requestFilterRecord {
+	fr, _ := ctx.Value(requestFilterRecordKey).(*requestFilterRecord)
+	return fr
+}
+
+// TrackStarted measures the timestamp the given handler has started execution
+// by attaching a handler to the chain.
+func TrackStarted(handler http.Handler, name string) http.Handler {
+	return trackStarted(handler, name, utilclock.RealClock{})
+}
+
+// TrackCompleted measures the timestamp the given handler has completed execution and then
+// it updates the corresponding metric with the filter latency duration.
+func TrackCompleted(handler http.Handler) http.Handler {
+	return trackCompleted(handler, utilclock.RealClock{}, func(fr *requestFilterRecord, completedAt time.Time) {
+		metrics.RecordFilterLatency(fr.name, completedAt.Sub(fr.startedTimestamp))
+	})
+}
+
+func trackStarted(handler http.Handler, name string, clock utilclock.PassiveClock) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ctx := r.Context()
+		if fr := requestFilterRecordFrom(ctx); fr != nil {
+			fr.name = name
+			fr.startedTimestamp = clock.Now()
+
+			handler.ServeHTTP(w, r)
+			return
+		}
+
+		fr := &requestFilterRecord{
+			name:             name,
+			startedTimestamp: clock.Now(),
+		}
+		r = r.WithContext(withRequestFilterRecord(ctx, fr))
+		handler.ServeHTTP(w, r)
+	})
+}
+
+func trackCompleted(handler http.Handler, clock utilclock.PassiveClock, action func(*requestFilterRecord, time.Time)) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// The previous filter has just completed.
+		completedAt := clock.Now()
+
+		defer handler.ServeHTTP(w, r)
+
+		ctx := r.Context()
+		if fr := requestFilterRecordFrom(ctx); fr != nil {
+			action(fr, completedAt)
+		}
+	})
+}

staging/src/k8s.io/apiserver/pkg/endpoints/filterlatency/filterlatency_test.go

@@ -0,0 +1,177 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package filterlatency
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	utilclock "k8s.io/apimachinery/pkg/util/clock"
+)
+
+func TestTrackStartedWithContextAlreadyHasFilterRecord(t *testing.T) {
+	filterName := "my-filter"
+	var (
+		callCount    int
+		filterRecord *requestFilterRecord
+	)
+	handler := http.HandlerFunc(func(_ http.ResponseWriter, req *http.Request) {
+		// we expect the handler to be invoked just once.
+		callCount++
+
+		// we expect the filter record to be set in the context
+		filterRecord = requestFilterRecordFrom(req.Context())
+	})
+
+	requestFilterStarted := time.Now()
+	wrapped := trackStarted(handler, filterName, utilclock.NewFakeClock(requestFilterStarted))
+
+	testRequest, err := http.NewRequest(http.MethodGet, "/api/v1/namespaces", nil)
+	if err != nil {
+		t.Fatalf("failed to create new http request - %v", err)
+	}
+	testRequest = testRequest.WithContext(withRequestFilterRecord(testRequest.Context(), &requestFilterRecord{
+		name:             "foo",
+		startedTimestamp: time.Now(),
+	}))
+
+	w := httptest.NewRecorder()
+	wrapped.ServeHTTP(w, testRequest)
+
+	if callCount != 1 {
+		t.Errorf("expected the given handler to be invoked once, but was actually invoked %d times", callCount)
+	}
+	if filterRecord == nil {
+		t.Fatal("expected a filter record in the request context, but got nil")
+	}
+	if filterName != filterRecord.name {
+		t.Errorf("expected filter name=%s but got=%s", filterName, filterRecord.name)
+	}
+	if requestFilterStarted != filterRecord.startedTimestamp {
+		t.Errorf("expected filter started timestamp=%s but got=%s", requestFilterStarted, filterRecord.startedTimestamp)
+	}
+}
+
+func TestTrackStartedWithContextDoesNotHaveFilterRecord(t *testing.T) {
+	filterName := "my-filter"
+	var (
+		callCount    int
+		filterRecord *requestFilterRecord
+	)
+	handler := http.HandlerFunc(func(_ http.ResponseWriter, req *http.Request) {
+		// we expect the handler to be invoked just once.
+		callCount++
+
+		// we expect the filter record to be set in the context
+		filterRecord = requestFilterRecordFrom(req.Context())
+	})
+
+	requestFilterStarted := time.Now()
+	wrapped := trackStarted(handler, filterName, utilclock.NewFakeClock(requestFilterStarted))
+
+	testRequest, err := http.NewRequest(http.MethodGet, "/api/v1/namespaces", nil)
+	if err != nil {
+		t.Fatalf("failed to create new http request - %v", err)
+	}
+
+	w := httptest.NewRecorder()
+	wrapped.ServeHTTP(w, testRequest)
+
+	if callCount != 1 {
+		t.Errorf("expected the given handler to be invoked once, but was actually invoked %d times", callCount)
+	}
+	if filterRecord == nil {
+		t.Fatal("expected a filter record in the request context, but got nil")
+	}
+	if filterName != filterRecord.name {
+		t.Errorf("expected filter name=%s but got=%s", filterName, filterRecord.name)
+	}
+	if requestFilterStarted != filterRecord.startedTimestamp {
+		t.Errorf("expected filter started timestamp=%s but got=%s", requestFilterStarted, filterRecord.startedTimestamp)
+	}
+}
+
+func TestTrackCompletedContextHasFilterRecord(t *testing.T) {
+	var (
+		handlerCallCount     int
+		actionCallCount      int
+		filterRecordGot      *requestFilterRecord
+		filterCompletedAtGot time.Time
+	)
+	handler := http.HandlerFunc(func(_ http.ResponseWriter, req *http.Request) {
+		// we expect the handler to be invoked just once.
+		handlerCallCount++
+	})
+
+	requestFilterEndedAt := time.Now()
+	wrapped := trackCompleted(handler, utilclock.NewFakeClock(requestFilterEndedAt), func(fr *requestFilterRecord, completedAt time.Time) {
+		actionCallCount++
+		filterRecordGot = fr
+		filterCompletedAtGot = completedAt
+	})
+
+	testRequest, err := http.NewRequest(http.MethodGet, "/api/v1/namespaces", nil)
+	if err != nil {
+		t.Fatalf("failed to create new http request - %v", err)
+	}
+
+	testRequest = testRequest.WithContext(withRequestFilterRecord(testRequest.Context(), &requestFilterRecord{}))
+
+	w := httptest.NewRecorder()
+	wrapped.ServeHTTP(w, testRequest)
+
+	if handlerCallCount != 1 {
+		t.Errorf("expected the given handler to be invoked once, but was actually invoked %d times", handlerCallCount)
+	}
+	if actionCallCount != 1 {
+		t.Errorf("expected the action callback to be invoked once, but was actually invoked %d times", actionCallCount)
+	}
+	if filterRecordGot == nil {
+		t.Fatal("expected a filter record in the request context, but got nil")
+	}
+	if requestFilterEndedAt != filterCompletedAtGot {
+		t.Errorf("expected filter ended timestamp=%s but got=%s", requestFilterEndedAt, filterCompletedAtGot)
+	}
+}
+
+func TestTrackCompletedContextDoesNotHaveFilterRecord(t *testing.T) {
+	var actionCallCount, handlerCallCount int
+	handler := http.HandlerFunc(func(_ http.ResponseWriter, req *http.Request) {
+		handlerCallCount++
+	})
+
+	wrapped := trackCompleted(handler, utilclock.NewFakeClock(time.Now()), func(_ *requestFilterRecord, _ time.Time) {
+		actionCallCount++
+	})
+
+	testRequest, err := http.NewRequest(http.MethodGet, "/api/v1/namespaces", nil)
+	if err != nil {
+		t.Fatalf("failed to create new http request - %v", err)
+	}
+
+	w := httptest.NewRecorder()
+	wrapped.ServeHTTP(w, testRequest)
+
+	if handlerCallCount != 1 {
+		t.Errorf("expected the given handler to be invoked once, but was actually invoked %d times", handlerCallCount)
+	}
+	if actionCallCount != 0 {
+		t.Errorf("expected the callback to not be invoked, but was actually invoked %d times", actionCallCount)
+	}
+}

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

@@ -196,6 +196,16 @@ var (
 		[]string{"verb", "resource", "subresource"},
 	)
 
+	requestFilterDuration = compbasemetrics.NewHistogramVec(
+		&compbasemetrics.HistogramOpts{
+			Name:           "apiserver_request_filter_duration_seconds",
+			Help:           "Request filter latency distribution in seconds, for each filter type",
+			Buckets:        []float64{0.0, 0.0003, 0.001, 0.003, 0.01, 0.03, 0.1, 0.3, 1.0, 5.0},
+			StabilityLevel: compbasemetrics.ALPHA,
+		},
+		[]string{"filter"},
+	)
+
 	kubectlExeRegexp = regexp.MustCompile(`^.*((?i:kubectl\.exe))`)
 
 	metrics = []resettableCollector{
@@ -213,6 +223,7 @@ var (
 		currentInqueueRequests,
 		requestTerminationsTotal,
 		apiSelfRequestCounter,
+		requestFilterDuration,
 	}
 
 	// these are the known (e.g. whitelisted/known) content types which we will report for
@@ -302,6 +313,10 @@ func UpdateInflightRequestMetrics(phase string, nonmutating, mutating int) {
 	}
 }
 
+func RecordFilterLatency(name string, elapsed time.Duration) {
+	requestFilterDuration.WithLabelValues(name).Observe(elapsed.Seconds())
+}
+
 // 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/server/BUILD

@@ -100,6 +100,7 @@ go_library(
         "//staging/src/k8s.io/apiserver/pkg/authorization/union:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/endpoints:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/endpoints/discovery:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/endpoints/filterlatency:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/endpoints/filters:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/endpoints/handlers/responsewriters:go_default_library",

staging/src/k8s.io/apiserver/pkg/server/config.go

@@ -50,6 +50,7 @@ import (
 	"k8s.io/apiserver/pkg/authorization/authorizerfactory"
 	authorizerunion "k8s.io/apiserver/pkg/authorization/union"
 	"k8s.io/apiserver/pkg/endpoints/discovery"
+	"k8s.io/apiserver/pkg/endpoints/filterlatency"
 	genericapifilters "k8s.io/apiserver/pkg/endpoints/filters"
 	apiopenapi "k8s.io/apiserver/pkg/endpoints/openapi"
 	apirequest "k8s.io/apiserver/pkg/endpoints/request"
@@ -669,17 +670,34 @@ func (c completedConfig) New(name string, delegationTarget DelegationTarget) (*G
 }
 
 func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler {
-	handler := genericapifilters.WithAuthorization(apiHandler, c.Authorization.Authorizer, c.Serializer)
+	handler := filterlatency.TrackCompleted(apiHandler)
+	handler = genericapifilters.WithAuthorization(handler, c.Authorization.Authorizer, c.Serializer)
+	handler = filterlatency.TrackStarted(handler, "authorization")
+
 	if c.FlowControl != nil {
+		handler = filterlatency.TrackCompleted(handler)
 		handler = genericfilters.WithPriorityAndFairness(handler, c.LongRunningFunc, c.FlowControl)
+		handler = filterlatency.TrackStarted(handler, "priorityandfairness")
 	} else {
 		handler = genericfilters.WithMaxInFlightLimit(handler, c.MaxRequestsInFlight, c.MaxMutatingRequestsInFlight, c.LongRunningFunc)
 	}
+
+	handler = filterlatency.TrackCompleted(handler)
 	handler = genericapifilters.WithImpersonation(handler, c.Authorization.Authorizer, c.Serializer)
+	handler = filterlatency.TrackStarted(handler, "impersonation")
+
+	handler = filterlatency.TrackCompleted(handler)
 	handler = genericapifilters.WithAudit(handler, c.AuditBackend, c.AuditPolicyChecker, c.LongRunningFunc)
+	handler = filterlatency.TrackStarted(handler, "audit")
+
 	failedHandler := genericapifilters.Unauthorized(c.Serializer)
 	failedHandler = genericapifilters.WithFailedAuthenticationAudit(failedHandler, c.AuditBackend, c.AuditPolicyChecker)
+
+	failedHandler = filterlatency.TrackCompleted(failedHandler)
+	handler = filterlatency.TrackCompleted(handler)
 	handler = genericapifilters.WithAuthentication(handler, c.Authentication.Authenticator, failedHandler, c.Authentication.APIAudiences)
+	handler = filterlatency.TrackStarted(handler, "authentication")
+
 	handler = genericfilters.WithCORS(handler, c.CorsAllowedOriginList, nil, nil, nil, "true")
 	handler = genericfilters.WithTimeoutForNonLongRunningRequests(handler, c.LongRunningFunc, c.RequestTimeout)
 	handler = genericfilters.WithWaitGroup(handler, c.LongRunningFunc, c.HandlerChainWaitGroup)

vendor/modules.txt

@@ -1779,6 +1779,7 @@ k8s.io/apiserver/pkg/authorization/union
 k8s.io/apiserver/pkg/endpoints
 k8s.io/apiserver/pkg/endpoints/deprecation
 k8s.io/apiserver/pkg/endpoints/discovery
+k8s.io/apiserver/pkg/endpoints/filterlatency
 k8s.io/apiserver/pkg/endpoints/filters
 k8s.io/apiserver/pkg/endpoints/handlers
 k8s.io/apiserver/pkg/endpoints/handlers/fieldmanager