UPSTREAM: <carry>: kube-apiserver: wire through isTerminating into handler chain

Origin-commit: 5772e7285acbe901762d8cd8cb1fc33d8b459d04

openshift-rebase(v1.24):source=c7c48fdacb0

openshift-rebase(v1.24):source=c7c48fdacb0

openshift-rebase(v1.24):source=c7c48fdacb0

UPSTREAM: <carry>: use lifeCycleSignals for isTerminating

openshift-rebase(v1.24):source=d3045350e43

Author: sttts

cmd/kube-apiserver/app/server.go

@@ -33,7 +33,6 @@ import (
 	"k8s.io/kubernetes/openshift-kube-apiserver/enablement"
 	"k8s.io/kubernetes/openshift-kube-apiserver/openshiftkubeapiserver"
 
-	"github.com/go-openapi/spec"
 	"github.com/spf13/cobra"
 
 	oteltrace "go.opentelemetry.io/otel/trace"
@@ -389,7 +388,6 @@ func CreateKubeAPIServerConfig(s completedServerRunOptions) (
 func buildGenericConfig(
 	s *options.ServerRunOptions,
 	proxyTransport *http.Transport,
-
 ) (
 	genericConfig *genericapiserver.Config,
 	versionedInformers clientgoinformers.SharedInformerFactory,

cmd/kube-scheduler/app/server.go

@@ -247,7 +247,7 @@ func buildHandlerChain(handler http.Handler, authn authenticator.Request, authz
 	handler = genericapifilters.WithAuthentication(handler, authn, failedHandler, nil)
 	handler = genericapifilters.WithRequestInfo(handler, requestInfoResolver)
 	handler = genericapifilters.WithCacheControl(handler)
-	handler = genericfilters.WithHTTPLogging(handler)
+	handler = genericfilters.WithHTTPLogging(handler, nil)
 	handler = genericfilters.WithPanicRecovery(handler, requestInfoResolver)
 
 	return handler

pkg/kubelet/server/server.go

@@ -1024,7 +1024,7 @@ var statusesNoTracePred = httplog.StatusIsNot(
 
 // ServeHTTP responds to HTTP requests on the Kubelet.
 func (s *Server) ServeHTTP(w http.ResponseWriter, req *http.Request) {
-	handler := httplog.WithLogging(s.restfulCont, statusesNoTracePred)
+	handler := httplog.WithLogging(s.restfulCont, statusesNoTracePred, nil)
 
 	// monitor http requests
 	var serverType string

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

@@ -856,7 +856,7 @@ func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler {
 	if c.ShutdownSendRetryAfter {
 		handler = genericfilters.WithRetryAfter(handler, c.lifecycleSignals.NotAcceptingNewRequest.Signaled())
 	}
-	handler = genericfilters.WithHTTPLogging(handler)
+	handler = genericfilters.WithHTTPLogging(handler, c.newIsTerminatingFunc())
 	if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.APIServerTracing) {
 		handler = genericapifilters.WithTracing(handler, c.TracerProvider)
 	}

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

@@ -341,6 +341,9 @@ func TestTimeoutWithLogging(t *testing.T) {
 					},
 				),
 			),
+			func() bool {
+				return false
+			},
 		),
 	)
 	defer ts.Close()

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

@@ -59,8 +59,8 @@ func WithPanicRecovery(handler http.Handler, resolver request.RequestInfoResolve
 }
 
 // WithHTTPLogging enables logging of incoming requests.
-func WithHTTPLogging(handler http.Handler) http.Handler {
-	return httplog.WithLogging(handler, httplog.DefaultStacktracePred)
+func WithHTTPLogging(handler http.Handler, isTerminating func() bool) http.Handler {
+	return httplog.WithLogging(handler, httplog.DefaultStacktracePred, isTerminating)
 }
 
 func withPanicRecovery(handler http.Handler, crashHandler func(http.ResponseWriter, *http.Request, interface{})) http.Handler {

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

@@ -65,6 +65,7 @@ type respLogger struct {
 	addedInfo          strings.Builder
 	addedKeyValuePairs []interface{}
 	startTime          time.Time
+	isTerminating      bool
 
 	captureErrorOutput bool
 
@@ -98,13 +99,13 @@ func DefaultStacktracePred(status int) bool {
 const withLoggingLevel = 3
 
 // WithLogging wraps the handler with logging.
-func WithLogging(handler http.Handler, pred StacktracePred) http.Handler {
+func WithLogging(handler http.Handler, pred StacktracePred, isTerminatingFn func() bool) http.Handler {
 	return withLogging(handler, pred, func() bool {
 		return klog.V(withLoggingLevel).Enabled()
-	})
+	}, isTerminatingFn)
 }
 
-func withLogging(handler http.Handler, stackTracePred StacktracePred, shouldLogRequest ShouldLogRequestPred) http.Handler {
+func withLogging(handler http.Handler, stackTracePred StacktracePred, shouldLogRequest ShouldLogRequestPred, isTerminatingFn func() bool) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 		if !shouldLogRequest() {
 			handler.ServeHTTP(w, req)
@@ -115,17 +116,22 @@ func withLogging(handler http.Handler, stackTracePred StacktracePred, shouldLogR
 		if old := respLoggerFromRequest(req); old != nil {
 			panic("multiple WithLogging calls!")
 		}
-
 		startTime := time.Now()
 		if receivedTimestamp, ok := request.ReceivedTimestampFrom(ctx); ok {
 			startTime = receivedTimestamp
 		}
 
-		rl := newLoggedWithStartTime(req, w, startTime)
-		rl.StacktraceWhen(stackTracePred)
+		isTerminating := false
+		if isTerminatingFn != nil {
+			isTerminating = isTerminatingFn()
+		}
+		rl := newLoggedWithStartTime(req, w, startTime).StacktraceWhen(stackTracePred).IsTerminating(isTerminating)
 		req = req.WithContext(context.WithValue(ctx, respLoggerContextKey, rl))
 		defer rl.Log()
 
+		if klog.V(3).Enabled() || (rl.isTerminating && klog.V(1).Enabled()) {
+			defer rl.Log()
+		}
 		w = responsewriter.WrapForHTTP1Or2(rl)
 		handler.ServeHTTP(w, req)
 	})
@@ -185,6 +191,12 @@ func (rl *respLogger) StacktraceWhen(pred StacktracePred) *respLogger {
 	return rl
 }
 
+// IsTerminating informs the logger that the server is terminating.
+func (rl *respLogger) IsTerminating(is bool) *respLogger {
+	rl.isTerminating = is
+	return rl
+}
+
 // StatusIsNot returns a StacktracePred which will cause stacktraces to be logged
 // for any status *not* in the given list.
 func StatusIsNot(statuses ...int) StacktracePred {

staging/src/k8s.io/apiserver/pkg/server/httplog/httplog_test.go

@@ -67,7 +67,7 @@ func TestWithLogging(t *testing.T) {
 	shouldLogRequest := func() bool { return true }
 	var handler http.Handler
 	handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})
-	handler = withLogging(withLogging(handler, DefaultStacktracePred, shouldLogRequest), DefaultStacktracePred, shouldLogRequest)
+	handler = withLogging(withLogging(handler, DefaultStacktracePred, shouldLogRequest, nil), DefaultStacktracePred, shouldLogRequest, nil)
 
 	func() {
 		defer func() {
@@ -111,7 +111,7 @@ func TestLogOf(t *testing.T) {
 					t.Errorf("Expected %v, got %v", test.want, got)
 				}
 			})
-			handler = withLogging(handler, DefaultStacktracePred, func() bool { return test.shouldLogRequest })
+			handler = withLogging(handler, DefaultStacktracePred, func() bool { return test.shouldLogRequest }, nil)
 			w := httptest.NewRecorder()
 			handler.ServeHTTP(w, req)
 		})
@@ -135,7 +135,7 @@ func TestUnlogged(t *testing.T) {
 			}
 		})
 		if makeLogger {
-			handler = WithLogging(handler, DefaultStacktracePred)
+			handler = WithLogging(handler, DefaultStacktracePred, nil)
 		}
 
 		handler.ServeHTTP(origWriter, req)
@@ -216,7 +216,7 @@ func TestRespLoggerWithDecoratedResponseWriter(t *testing.T) {
 				}
 			})
 
-			handler = withLogging(handler, DefaultStacktracePred, func() bool { return true })
+			handler = withLogging(handler, DefaultStacktracePred, func() bool { return true }, nil)
 			handler.ServeHTTP(test.r(), req)
 		})
 	}

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

@@ -0,0 +1,39 @@
+/*
+Copyright 2021 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 server
+
+// newIsTerminatingFunc returns a 'func() bool' that relies on the
+// 'ShutdownInitiated' life cycle signal of answer if the apiserver
+// has started the termination process.
+func (c *Config) newIsTerminatingFunc() func() bool {
+	var shutdownCh <-chan struct{}
+	// TODO: a properly initialized Config object should always have lifecycleSignals
+	//  initialized, but some config unit tests leave lifecycleSignals as nil.
+	//  Fix the unit tests upstream and then we can remove this check.
+	if c.lifecycleSignals.ShutdownInitiated != nil {
+		shutdownCh = c.lifecycleSignals.ShutdownInitiated.Signaled()
+	}
+
+	return func() bool {
+		select {
+		case <-shutdownCh:
+			return true
+		default:
+			return false
+		}
+	}
+}

staging/src/k8s.io/controller-manager/app/serve.go

@@ -48,7 +48,7 @@ func BuildHandlerChain(apiHandler http.Handler, authorizationInfo *apiserver.Aut
 	}
 	handler = genericapifilters.WithRequestInfo(handler, requestInfoResolver)
 	handler = genericapifilters.WithCacheControl(handler)
-	handler = genericfilters.WithHTTPLogging(handler)
+	handler = genericfilters.WithHTTPLogging(handler, nil)
 	handler = genericfilters.WithPanicRecovery(handler, requestInfoResolver)
 
 	return handler