Merge pull request #51415 from jpbetz/timeout-param-minimal

Automatic merge from submit-queue (batch tested with PRs 50932, 49610, 51312, 51415, 50705)

Add --request-timeout to kube-apiserver to make global request timeout configurable

**What this PR does / why we need it**:

Make the currently hard coded 60 global request timeout in apiserver configurable via a --request-timeout command line flag.

**Which issue this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close that issue when PR gets merged)*: fixes #

https://github.com/kubernetes/kubernetes/issues/51355

**Special notes for your reviewer**:

We plan to backport this to 1.6 and 1.7. To ease the backport work, we've kept this change to a minimum. After this PR we will submit an additional PR to rename `--min-request-timeout` to something that more clearly describes it's purpose, such as `--long-running-request-timeout-min`.

**Release note**:

```release-note
Add --request-timeout to kube-apiserver to make global request timeout configurable.
```

cmd/kube-apiserver/app/options/options_test.go

@@ -81,6 +81,7 @@ func TestAddFlags(t *testing.T) {
 		"--kubelet-certificate-authority=/var/run/kubernetes/caserver.crt",
 		"--proxy-client-cert-file=/var/run/kubernetes/proxy.crt",
 		"--proxy-client-key-file=/var/run/kubernetes/proxy.key",
+		"--request-timeout=2m",
 		"--storage-backend=etcd2",
 	}
 	f.Parse(args)
@@ -95,6 +96,7 @@ func TestAddFlags(t *testing.T) {
 			CorsAllowedOriginList:       []string{"10.10.10.100", "10.10.10.200"},
 			MaxRequestsInFlight:         400,
 			MaxMutatingRequestsInFlight: 200,
+			RequestTimeout:              time.Duration(2) * time.Minute,
 			MinRequestTimeout:           1800,
 		},
 		Admission: &apiserveroptions.AdmissionOptions{

pkg/kubeapiserver/server/insecure_handler.go

@@ -46,7 +46,7 @@ func BuildInsecureHandlerChain(apiHandler http.Handler, c *server.Config) http.H
 	handler = genericapifilters.WithAuthentication(handler, c.RequestContextMapper, insecureSuperuser{}, nil)
 	handler = genericfilters.WithCORS(handler, c.CorsAllowedOriginList, nil, nil, nil, "true")
 	handler = genericfilters.WithPanicRecovery(handler)
-	handler = genericfilters.WithTimeoutForNonLongRunningRequests(handler, c.RequestContextMapper, c.LongRunningFunc)
+	handler = genericfilters.WithTimeoutForNonLongRunningRequests(handler, c.RequestContextMapper, c.LongRunningFunc, c.RequestTimeout)
 	handler = genericfilters.WithMaxInFlightLimit(handler, c.MaxRequestsInFlight, c.MaxMutatingRequestsInFlight, c.RequestContextMapper, c.LongRunningFunc)
 	handler = genericapifilters.WithRequestInfo(handler, server.NewRequestInfoResolver(c), c.RequestContextMapper)
 	handler = apirequest.WithRequestContext(handler, c.RequestContextMapper)

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

@@ -148,8 +148,11 @@ type Config struct {
 	// RESTOptionsGetter is used to construct RESTStorage types via the generic registry.
 	RESTOptionsGetter genericregistry.RESTOptionsGetter
 
-	// If specified, requests will be allocated a random timeout between this value, and twice this value.
-	// Note that it is up to the request handlers to ignore or honor this timeout. In seconds.
+	// If specified, all requests except those which match the LongRunningFunc predicate will timeout
+	// after this duration.
+	RequestTimeout time.Duration
+	// If specified, long running requests such as watch will be allocated a random timeout between this value, and
+	// twice this value.  Note that it is up to the request handlers to ignore or honor this timeout. In seconds.
 	MinRequestTimeout int
 	// MaxRequestsInFlight is the maximum number of parallel non-long-running requests. Every further
 	// request has to wait. Applies only to non-mutating requests.
@@ -222,6 +225,7 @@ func NewConfig(codecs serializer.CodecFactory) *Config {
 		EnableProfiling:              true,
 		MaxRequestsInFlight:          400,
 		MaxMutatingRequestsInFlight:  200,
+		RequestTimeout:               time.Duration(60) * time.Second,
 		MinRequestTimeout:            1800,
 		EnableAPIResponseCompression: utilfeature.DefaultFeatureGate.Enabled(features.APIResponseCompression),
 
@@ -477,7 +481,7 @@ func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler {
 	}
 	handler = genericapifilters.WithAuthentication(handler, c.RequestContextMapper, c.Authenticator, genericapifilters.Unauthorized(c.RequestContextMapper, c.Serializer, c.SupportsBasicAuth))
 	handler = genericfilters.WithCORS(handler, c.CorsAllowedOriginList, nil, nil, nil, "true")
-	handler = genericfilters.WithTimeoutForNonLongRunningRequests(handler, c.RequestContextMapper, c.LongRunningFunc)
+	handler = genericfilters.WithTimeoutForNonLongRunningRequests(handler, c.RequestContextMapper, c.LongRunningFunc, c.RequestTimeout)
 	handler = genericapifilters.WithRequestInfo(handler, NewRequestInfoResolver(c), c.RequestContextMapper)
 	handler = apirequest.WithRequestContext(handler, c.RequestContextMapper)
 	handler = genericfilters.WithPanicRecovery(handler)

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

@@ -31,12 +31,10 @@ import (
 	apirequest "k8s.io/apiserver/pkg/endpoints/request"
 )
 
-const globalTimeout = time.Minute
-
 var errConnKilled = fmt.Errorf("kill connection/stream")
 
-// WithTimeoutForNonLongRunningRequests times out non-long-running requests after the time given by globalTimeout.
-func WithTimeoutForNonLongRunningRequests(handler http.Handler, requestContextMapper apirequest.RequestContextMapper, longRunning apirequest.LongRunningRequestCheck) http.Handler {
+// WithTimeoutForNonLongRunningRequests times out non-long-running requests after the time given by timeout.
+func WithTimeoutForNonLongRunningRequests(handler http.Handler, requestContextMapper apirequest.RequestContextMapper, longRunning apirequest.LongRunningRequestCheck, timeout time.Duration) http.Handler {
 	if longRunning == nil {
 		return handler
 	}
@@ -45,13 +43,13 @@ func WithTimeoutForNonLongRunningRequests(handler http.Handler, requestContextMa
 		ctx, ok := requestContextMapper.Get(req)
 		if !ok {
 			// if this happens, the handler chain isn't setup correctly because there is no context mapper
-			return time.After(globalTimeout), func() {}, apierrors.NewInternalError(fmt.Errorf("no context found for request during timeout"))
+			return time.After(timeout), func() {}, apierrors.NewInternalError(fmt.Errorf("no context found for request during timeout"))
 		}
 
 		requestInfo, ok := apirequest.RequestInfoFrom(ctx)
 		if !ok {
 			// if this happens, the handler chain isn't setup correctly because there is no request info
-			return time.After(globalTimeout), func() {}, apierrors.NewInternalError(fmt.Errorf("no request info found for request during timeout"))
+			return time.After(timeout), func() {}, apierrors.NewInternalError(fmt.Errorf("no request info found for request during timeout"))
 		}
 
 		if longRunning(req, requestInfo) {
@@ -69,7 +67,7 @@ func WithTimeoutForNonLongRunningRequests(handler http.Handler, requestContextMa
 				metrics.MonitorRequest(req, strings.ToUpper(requestInfo.Verb), "", requestInfo.Path, "", scope, http.StatusGatewayTimeout, 0, now)
 			}
 		}
-		return time.After(globalTimeout), metricFn, apierrors.NewTimeoutError(fmt.Sprintf("request did not complete within %s", globalTimeout), 0)
+		return time.After(timeout), metricFn, apierrors.NewTimeoutError(fmt.Sprintf("request did not complete within %s", timeout), 0)
 	}
 	return WithTimeout(handler, timeoutFunc)
 }

staging/src/k8s.io/apiserver/pkg/server/options/server_run_options.go

@@ -19,6 +19,7 @@ package options
 import (
 	"fmt"
 	"net"
+	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
@@ -39,6 +40,7 @@ type ServerRunOptions struct {
 	ExternalHost                string
 	MaxRequestsInFlight         int
 	MaxMutatingRequestsInFlight int
+	RequestTimeout              time.Duration
 	MinRequestTimeout           int
 	TargetRAMMB                 int
 	WatchCacheSizes             []string
@@ -49,6 +51,7 @@ func NewServerRunOptions() *ServerRunOptions {
 	return &ServerRunOptions{
 		MaxRequestsInFlight:         defaults.MaxRequestsInFlight,
 		MaxMutatingRequestsInFlight: defaults.MaxMutatingRequestsInFlight,
+		RequestTimeout:              defaults.RequestTimeout,
 		MinRequestTimeout:           defaults.MinRequestTimeout,
 	}
 }
@@ -59,6 +62,7 @@ func (s *ServerRunOptions) ApplyTo(c *server.Config) error {
 	c.ExternalAddress = s.ExternalHost
 	c.MaxRequestsInFlight = s.MaxRequestsInFlight
 	c.MaxMutatingRequestsInFlight = s.MaxMutatingRequestsInFlight
+	c.RequestTimeout = s.RequestTimeout
 	c.MinRequestTimeout = s.MinRequestTimeout
 	c.PublicAddress = s.AdvertiseAddress
 
@@ -93,7 +97,11 @@ func (s *ServerRunOptions) Validate() []error {
 		errors = append(errors, fmt.Errorf("--max-requests-inflight can not be negative value"))
 	}
 	if s.MaxMutatingRequestsInFlight < 0 {
-		errors = append(errors, fmt.Errorf("--min-request-timeout can not be negative value"))
+		errors = append(errors, fmt.Errorf("--max-mutating-requests-inflight can not be negative value"))
+	}
+
+	if s.RequestTimeout.Nanoseconds() < 0 {
+		errors = append(errors, fmt.Errorf("--request-timeout can not be negative value"))
 	}
 
 	return errors
@@ -132,6 +140,11 @@ func (s *ServerRunOptions) AddUniversalFlags(fs *pflag.FlagSet) {
 		"The maximum number of mutating requests in flight at a given time. When the server exceeds this, "+
 		"it rejects requests. Zero for no limit.")
 
+	fs.DurationVar(&s.RequestTimeout, "request-timeout", s.RequestTimeout, ""+
+		"An optional field indicating the duration a handler must keep a request open before timing "+
+		"it out. This is the default request timeout for requests but may be overridden by flags such as "+
+		"--min-request-timeout for specific types of requests.")
+
 	fs.IntVar(&s.MinRequestTimeout, "min-request-timeout", s.MinRequestTimeout, ""+
 		"An optional field indicating the minimum number of seconds a handler must keep "+
 		"a request open before timing it out. Currently only honored by the watch request "+