Merge pull request #94903 from tkashem/request-received-timestamp

Take into account the latency incurred in server filters

staging/src/k8s.io/apiserver/pkg/audit/request.go

@@ -43,9 +43,9 @@ const (
 	userAgentTruncateSuffix = "...TRUNCATED"
 )
 
-func NewEventFromRequest(req *http.Request, level auditinternal.Level, attribs authorizer.Attributes) (*auditinternal.Event, error) {
+func NewEventFromRequest(req *http.Request, requestReceivedTimestamp time.Time, level auditinternal.Level, attribs authorizer.Attributes) (*auditinternal.Event, error) {
 	ev := &auditinternal.Event{
-		RequestReceivedTimestamp: metav1.NewMicroTime(time.Now()),
+		RequestReceivedTimestamp: metav1.NewMicroTime(requestReceivedTimestamp),
 		Verb:                     attribs.GetVerb(),
 		RequestURI:               req.URL.RequestURI(),
 		UserAgent:                maybeTruncateUserAgent(req),

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

@@ -16,6 +16,7 @@ go_test(
         "cachecontrol_test.go",
         "impersonation_test.go",
         "metrics_test.go",
+        "request_received_time_test.go",
         "requestinfo_test.go",
         "warning_test.go",
     ],
@@ -27,6 +28,7 @@ go_test(
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/serializer:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/clock:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/apis/audit:go_default_library",
@@ -54,6 +56,7 @@ go_library(
         "doc.go",
         "impersonation.go",
         "metrics.go",
+        "request_received_time.go",
         "requestinfo.go",
         "warning.go",
     ],
@@ -66,6 +69,7 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/clock:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/net:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/apis/audit:go_default_library",

staging/src/k8s.io/apiserver/pkg/endpoints/filters/audit.go

@@ -131,7 +131,11 @@ func createAuditEventAndAttachToContext(req *http.Request, policy policy.Checker
 		return req, nil, nil, nil
 	}
 
-	ev, err := audit.NewEventFromRequest(req, level, attribs)
+	requestReceivedTimestamp, ok := request.ReceivedTimestampFrom(ctx)
+	if !ok {
+		requestReceivedTimestamp = time.Now()
+	}
+	ev, err := audit.NewEventFromRequest(req, requestReceivedTimestamp, level, attribs)
 	if err != nil {
 		return req, nil, nil, fmt.Errorf("failed to complete audit event from request: %v", err)
 	}

staging/src/k8s.io/apiserver/pkg/endpoints/filters/request_received_time.go

@@ -0,0 +1,40 @@
+/*
+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 filters
+
+import (
+	"net/http"
+
+	utilclock "k8s.io/apimachinery/pkg/util/clock"
+	"k8s.io/apiserver/pkg/endpoints/request"
+)
+
+// WithRequestReceivedTimestamp attaches the ReceivedTimestamp (the time the request reached
+// the apiserver) to the context.
+func WithRequestReceivedTimestamp(handler http.Handler) http.Handler {
+	return withRequestReceivedTimestampWithClock(handler, utilclock.RealClock{})
+}
+
+// The clock is passed as a parameter, handy for unit testing.
+func withRequestReceivedTimestampWithClock(handler http.Handler, clock utilclock.PassiveClock) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		ctx := req.Context()
+		req = req.WithContext(request.WithReceivedTimestamp(ctx, clock.Now()))
+
+		handler.ServeHTTP(w, req)
+	})
+}

staging/src/k8s.io/apiserver/pkg/endpoints/filters/request_received_time_test.go

@@ -0,0 +1,63 @@
+/*
+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 filters
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	utilclock "k8s.io/apimachinery/pkg/util/clock"
+	"k8s.io/apiserver/pkg/endpoints/request"
+)
+
+func TestWithRequestReceivedTimestamp(t *testing.T) {
+	receivedTimestampExpected := time.Now()
+
+	var (
+		callCount            int
+		receivedTimestampGot time.Time
+		ok                   bool
+	)
+	handler := http.HandlerFunc(func(_ http.ResponseWriter, req *http.Request) {
+		callCount++
+
+		// does the handler chain that follows see the request received timestamp?
+		receivedTimestampGot, ok = request.ReceivedTimestampFrom(req.Context())
+	})
+
+	wrapped := withRequestReceivedTimestampWithClock(handler, utilclock.NewFakeClock(receivedTimestampExpected))
+
+	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("WithRequestReceivedTimestamp: expected the original handler to be invoked once, but was actually invoked %d times", callCount)
+	}
+	if !ok {
+		t.Error("WithRequestReceivedTimestamp: expected request.ReceivedTimestampFrom to return true, but got false")
+	}
+	if receivedTimestampExpected != receivedTimestampGot {
+		t.Errorf("WithRequestReceivedTimestamp: request received timestamp expected=%s but got=%s", receivedTimestampExpected, receivedTimestampGot)
+	}
+}

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

@@ -368,8 +368,11 @@ func MonitorRequest(req *http.Request, verb, group, version, resource, subresour
 // InstrumentRouteFunc works like Prometheus' InstrumentHandlerFunc but wraps
 // the go-restful RouteFunction instead of a HandlerFunc plus some Kubernetes endpoint specific information.
 func InstrumentRouteFunc(verb, group, version, resource, subresource, scope, component string, deprecated bool, removedRelease string, routeFunc restful.RouteFunction) restful.RouteFunction {
-	return restful.RouteFunction(func(request *restful.Request, response *restful.Response) {
-		now := time.Now()
+	return restful.RouteFunction(func(req *restful.Request, response *restful.Response) {
+		requestReceivedTimestamp, ok := request.ReceivedTimestampFrom(req.Request.Context())
+		if !ok {
+			requestReceivedTimestamp = time.Now()
+		}
 
 		delegate := &ResponseWriterDelegator{ResponseWriter: response.ResponseWriter}
 
@@ -384,16 +387,19 @@ func InstrumentRouteFunc(verb, group, version, resource, subresource, scope, com
 		}
 		response.ResponseWriter = rw
 
-		routeFunc(request, response)
+		routeFunc(req, response)
 
-		MonitorRequest(request.Request, verb, group, version, resource, subresource, scope, component, deprecated, removedRelease, delegate.Header().Get("Content-Type"), delegate.Status(), delegate.ContentLength(), time.Since(now))
+		MonitorRequest(req.Request, verb, group, version, resource, subresource, scope, component, deprecated, removedRelease, delegate.Header().Get("Content-Type"), delegate.Status(), delegate.ContentLength(), time.Since(requestReceivedTimestamp))
 	})
 }
 
 // InstrumentHandlerFunc works like Prometheus' InstrumentHandlerFunc but adds some Kubernetes endpoint specific information.
 func InstrumentHandlerFunc(verb, group, version, resource, subresource, scope, component string, deprecated bool, removedRelease string, handler http.HandlerFunc) http.HandlerFunc {
 	return func(w http.ResponseWriter, req *http.Request) {
-		now := time.Now()
+		requestReceivedTimestamp, ok := request.ReceivedTimestampFrom(req.Context())
+		if !ok {
+			requestReceivedTimestamp = time.Now()
+		}
 
 		delegate := &ResponseWriterDelegator{ResponseWriter: w}
 
@@ -408,7 +414,7 @@ func InstrumentHandlerFunc(verb, group, version, resource, subresource, scope, c
 
 		handler(w, req)
 
-		MonitorRequest(req, verb, group, version, resource, subresource, scope, component, deprecated, removedRelease, delegate.Header().Get("Content-Type"), delegate.Status(), delegate.ContentLength(), time.Since(now))
+		MonitorRequest(req, verb, group, version, resource, subresource, scope, component, deprecated, removedRelease, delegate.Header().Get("Content-Type"), delegate.Status(), delegate.ContentLength(), time.Since(requestReceivedTimestamp))
 	}
 }
 

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

@@ -10,6 +10,7 @@ go_test(
     name = "go_default_test",
     srcs = [
         "context_test.go",
+        "received_time_test.go",
         "requestinfo_test.go",
     ],
     embed = [":go_default_library"],
@@ -25,6 +26,7 @@ go_library(
     srcs = [
         "context.go",
         "doc.go",
+        "received_time.go",
         "requestinfo.go",
     ],
     importmap = "k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/endpoints/request",

staging/src/k8s.io/apiserver/pkg/endpoints/request/received_time.go

@@ -0,0 +1,45 @@
+/*
+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 request
+
+import (
+	"context"
+	"time"
+)
+
+type requestReceivedTimestampKeyType int
+
+// requestReceivedTimestampKey is the ReceivedTimestamp (the time the request reached the apiserver)
+// key for the context.
+const requestReceivedTimestampKey requestReceivedTimestampKeyType = iota
+
+// WithReceivedTimestamp returns a copy of parent context in which the ReceivedTimestamp
+// (the time the request reached the apiserver) is set.
+//
+// If the specified ReceivedTimestamp is zero, no value is set and the parent context is returned as is.
+func WithReceivedTimestamp(parent context.Context, receivedTimestamp time.Time) context.Context {
+	if receivedTimestamp.IsZero() {
+		return parent
+	}
+	return WithValue(parent, requestReceivedTimestampKey, receivedTimestamp)
+}
+
+// ReceivedTimestampFrom returns the value of the ReceivedTimestamp key from the specified context.
+func ReceivedTimestampFrom(ctx context.Context) (time.Time, bool) {
+	info, ok := ctx.Value(requestReceivedTimestampKey).(time.Time)
+	return info, ok
+}

staging/src/k8s.io/apiserver/pkg/endpoints/request/received_time_test.go

@@ -0,0 +1,62 @@
+/*
+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 request
+
+import (
+	"context"
+	"strconv"
+	"testing"
+	"time"
+)
+
+func TestWithRequestReceiveTime(t *testing.T) {
+	tests := []struct {
+		name              string
+		receivedTimestamp time.Time
+		expected          bool
+	}{
+		{
+			name:              "request received time is set",
+			receivedTimestamp: time.Now(),
+			expected:          true,
+		},
+		{
+			name:              "request received time is empty",
+			receivedTimestamp: time.Time{},
+			expected:          false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			parent := context.TODO()
+			ctx := WithReceivedTimestamp(parent, test.receivedTimestamp)
+			if ctx == nil {
+				t.Fatal("WithReceivedTimestamp: expected a non nil context, got nil")
+			}
+
+			receivedTimestampGot, ok := ReceivedTimestampFrom(ctx)
+			if test.expected != ok {
+				t.Errorf("ReceivedTimestampFrom: expected=%s got=%s", strconv.FormatBool(test.expected), strconv.FormatBool(ok))
+			}
+
+			if test.receivedTimestamp != receivedTimestampGot {
+				t.Errorf("ReceivedTimestampFrom: received timestamp expected=%s but got=%s", test.receivedTimestamp, receivedTimestampGot)
+			}
+		})
+	}
+}

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

@@ -690,6 +690,7 @@ func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler {
 	handler = genericapifilters.WithAuditAnnotations(handler, c.AuditBackend, c.AuditPolicyChecker)
 	handler = genericapifilters.WithWarningRecorder(handler)
 	handler = genericapifilters.WithCacheControl(handler)
+	handler = genericapifilters.WithRequestReceivedTimestamp(handler)
 	handler = genericfilters.WithPanicRecovery(handler)
 	return handler
 }