add temporary feature gate to allow disabling aggregated discovery timeout

Author: deads2k

cmd/kube-apiserver/app/aggregator.go

@@ -41,7 +41,7 @@ import (
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	kubeexternalinformers "k8s.io/client-go/informers"
 	"k8s.io/client-go/tools/cache"
-	"k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
+	v1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
 	v1helper "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1/helper"
 	"k8s.io/kube-aggregator/pkg/apis/apiregistration/v1beta1"
 	aggregatorapiserver "k8s.io/kube-aggregator/pkg/apiserver"
@@ -50,6 +50,7 @@ import (
 	informers "k8s.io/kube-aggregator/pkg/client/informers/externalversions/apiregistration/v1"
 	"k8s.io/kube-aggregator/pkg/controllers/autoregister"
 	"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
+	kubefeatures "k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/master/controller/crdregistration"
 )
 
@@ -109,10 +110,11 @@ func createAggregatorConfig(
 			SharedInformerFactory: externalInformers,
 		},
 		ExtraConfig: aggregatorapiserver.ExtraConfig{
-			ProxyClientCert: certBytes,
-			ProxyClientKey:  keyBytes,
-			ServiceResolver: serviceResolver,
-			ProxyTransport:  proxyTransport,
+			ProxyClientCert:                  certBytes,
+			ProxyClientKey:                   keyBytes,
+			ServiceResolver:                  serviceResolver,
+			ProxyTransport:                   proxyTransport,
+			EnableAggregatedDiscoveryTimeout: utilfeature.DefaultFeatureGate.Enabled(kubefeatures.EnableAggregatedDiscoveryTimeout),
 		},
 	}
 

pkg/features/kube_features.go

@@ -495,6 +495,13 @@ const (
 	//
 	// Enables the startupProbe in kubelet worker.
 	StartupProbe featuregate.Feature = "StartupProbe"
+
+	// owner @deads2k
+	// deprecated: v1.16
+	//
+	// Enable the aggregated discovery timeout to ensure client responsiveness. Note this feature is present
+	// only for backward compatibility, it will be removed in the 1.17 release.
+	EnableAggregatedDiscoveryTimeout featuregate.Feature = "EnableAggregatedDiscoveryTimeout"
 )
 
 func init() {
@@ -597,6 +604,8 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS
 	apiextensionsfeatures.CustomResourcePublishOpenAPI:    {Default: true, PreRelease: featuregate.GA, LockToDefault: true},
 	apiextensionsfeatures.CustomResourceDefaulting:        {Default: true, PreRelease: featuregate.Beta},
 
+	EnableAggregatedDiscoveryTimeout: {Default: true, PreRelease: featuregate.Deprecated},
+
 	// features that enable backwards compatibility but are scheduled to be removed
 	// ...
 	HPAScaleToZero:         {Default: false, PreRelease: featuregate.Alpha},

staging/src/k8s.io/kube-aggregator/pkg/apiserver/apiserver.go

@@ -71,6 +71,8 @@ type ExtraConfig struct {
 
 	// Mechanism by which the Aggregator will resolve services. Required.
 	ServiceResolver ServiceResolver
+
+	EnableAggregatedDiscoveryTimeout bool
 }
 
 // Config represents the configuration needed to create an APIAggregator.
@@ -132,6 +134,8 @@ type APIAggregator struct {
 
 	// openAPIAggregationController downloads and merges OpenAPI specs.
 	openAPIAggregationController *openapicontroller.AggregationController
+
+	enableAggregatedDiscoveryTimeout bool
 }
 
 // Complete fills in any fields not set that are required to have valid data. It's mutating the receiver.
@@ -172,17 +176,18 @@ func (c completedConfig) NewWithDelegate(delegationTarget genericapiserver.Deleg
 	)
 
 	s := &APIAggregator{
-		GenericAPIServer:         genericServer,
-		delegateHandler:          delegationTarget.UnprotectedHandler(),
-		proxyClientCert:          c.ExtraConfig.ProxyClientCert,
-		proxyClientKey:           c.ExtraConfig.ProxyClientKey,
-		proxyTransport:           c.ExtraConfig.ProxyTransport,
-		proxyHandlers:            map[string]*proxyHandler{},
-		handledGroups:            sets.String{},
-		lister:                   informerFactory.Apiregistration().V1().APIServices().Lister(),
-		APIRegistrationInformers: informerFactory,
-		serviceResolver:          c.ExtraConfig.ServiceResolver,
-		openAPIConfig:            openAPIConfig,
+		GenericAPIServer:                 genericServer,
+		delegateHandler:                  delegationTarget.UnprotectedHandler(),
+		proxyClientCert:                  c.ExtraConfig.ProxyClientCert,
+		proxyClientKey:                   c.ExtraConfig.ProxyClientKey,
+		proxyTransport:                   c.ExtraConfig.ProxyTransport,
+		proxyHandlers:                    map[string]*proxyHandler{},
+		handledGroups:                    sets.String{},
+		lister:                           informerFactory.Apiregistration().V1().APIServices().Lister(),
+		APIRegistrationInformers:         informerFactory,
+		serviceResolver:                  c.ExtraConfig.ServiceResolver,
+		openAPIConfig:                    openAPIConfig,
+		enableAggregatedDiscoveryTimeout: c.ExtraConfig.EnableAggregatedDiscoveryTimeout,
 	}
 
 	apiGroupInfo := apiservicerest.NewRESTStorage(c.GenericConfig.MergedResourceConfig, c.GenericConfig.RESTOptionsGetter)
@@ -286,11 +291,12 @@ func (s *APIAggregator) AddAPIService(apiService *v1.APIService) error {
 
 	// register the proxy handler
 	proxyHandler := &proxyHandler{
-		localDelegate:   s.delegateHandler,
-		proxyClientCert: s.proxyClientCert,
-		proxyClientKey:  s.proxyClientKey,
-		proxyTransport:  s.proxyTransport,
-		serviceResolver: s.serviceResolver,
+		localDelegate:                    s.delegateHandler,
+		proxyClientCert:                  s.proxyClientCert,
+		proxyClientKey:                   s.proxyClientKey,
+		proxyTransport:                   s.proxyTransport,
+		serviceResolver:                  s.serviceResolver,
+		enableAggregatedDiscoveryTimeout: s.enableAggregatedDiscoveryTimeout,
 	}
 	proxyHandler.updateAPIService(apiService)
 	if s.openAPIAggregationController != nil {

staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_proxy.go

@@ -20,8 +20,6 @@ import (
 	"context"
 	"net/http"
 	"net/url"
-	"os"
-	"strconv"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -31,7 +29,6 @@ import (
 	"k8s.io/apimachinery/pkg/util/httpstream/spdy"
 	utilnet "k8s.io/apimachinery/pkg/util/net"
 	"k8s.io/apimachinery/pkg/util/proxy"
-	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
 	endpointmetrics "k8s.io/apiserver/pkg/endpoints/metrics"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
@@ -50,19 +47,6 @@ const (
 	aggregatedDiscoveryTimeout = 5 * time.Second
 )
 
-var (
-	// TODO this should be unconditionally true once we remove the env var override
-	enableAggregatedDiscoveryTimeout = true
-)
-
-func init() {
-	disableAggregatedDiscoveryTimeout, err := strconv.ParseBool(os.Getenv("DEPRECATED_DISABLE_AGGREGATOR_DISCOVERY_TIMEOUT"))
-	if err != nil {
-		utilruntime.HandleError(err)
-	}
-	enableAggregatedDiscoveryTimeout = !disableAggregatedDiscoveryTimeout
-}
-
 // proxyHandler provides a http.Handler which will proxy traffic to locations
 // specified by items implementing Redirector.
 type proxyHandler struct {
@@ -79,6 +63,8 @@ type proxyHandler struct {
 	serviceResolver ServiceResolver
 
 	handlingInfo atomic.Value
+
+	enableAggregatedDiscoveryTimeout bool
 }
 
 type proxyHandlingInfo struct {
@@ -162,7 +148,7 @@ func (r *proxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	location.Path = req.URL.Path
 	location.RawQuery = req.URL.Query().Encode()
 
-	newReq, cancelFn := newRequestForProxy(location, req)
+	newReq, cancelFn := newRequestForProxy(location, req, r.enableAggregatedDiscoveryTimeout)
 	defer cancelFn()
 
 	if handlingInfo.proxyRoundTripper == nil {
@@ -191,7 +177,7 @@ func (r *proxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 }
 
 // newRequestForProxy returns a shallow copy of the original request with a context that may include a timeout for discovery requests
-func newRequestForProxy(location *url.URL, req *http.Request) (*http.Request, context.CancelFunc) {
+func newRequestForProxy(location *url.URL, req *http.Request, enableAggregatedDiscoveryTimeout bool) (*http.Request, context.CancelFunc) {
 	newCtx := context.Background()
 	cancelFn := func() {}
 

staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_proxy_test.go

@@ -540,7 +540,7 @@ func TestGetContextForNewRequest(t *testing.T) {
 		}
 		location.Path = req.URL.Path
 
-		newReq, cancelFn := newRequestForProxy(location, req)
+		newReq, cancelFn := newRequestForProxy(location, req, true)
 		defer cancelFn()
 
 		theproxy := proxy.NewUpgradeAwareHandler(location, server.Client().Transport, true, false, &responder{w: w})