Merge pull request #46042 from derekwaynecarr/quota-admission-registry

Automatic merge from submit-queue (batch tested with PRs 45346, 45903, 45958, 46042, 45975)

ResourceQuota admission control injects registry

**What this PR does / why we need it**:
The `ResourceQuota` admission controller works with a registry that maps a GroupKind to an Evaluator.  The registry used in the existing plug-in is not injectable, which makes usage of the ResourceQuota plug-in in other API server contexts difficult.  This PR updates the code to support late injection of the registry via a plug-in initializer.

cmd/kube-apiserver/app/BUILD

@@ -38,6 +38,7 @@ go_library(
         "//pkg/master:go_default_library",
         "//pkg/master/thirdparty:go_default_library",
         "//pkg/master/tunneler:go_default_library",
+        "//pkg/quota/install:go_default_library",
         "//pkg/registry/cachesize:go_default_library",
         "//pkg/registry/rbac/rest:go_default_library",
         "//pkg/version:go_default_library",

cmd/kube-apiserver/app/server.go

@@ -71,6 +71,7 @@ import (
 	kubeserver "k8s.io/kubernetes/pkg/kubeapiserver/server"
 	"k8s.io/kubernetes/pkg/master"
 	"k8s.io/kubernetes/pkg/master/tunneler"
+	quotainstall "k8s.io/kubernetes/pkg/quota/install"
 	"k8s.io/kubernetes/pkg/registry/cachesize"
 	rbacrest "k8s.io/kubernetes/pkg/registry/rbac/rest"
 	"k8s.io/kubernetes/pkg/version"
@@ -392,7 +393,12 @@ func BuildAdmissionPluginInitializer(s *options.ServerRunOptions, client interna
 
 	// TODO: use a dynamic restmapper. See https://github.com/kubernetes/kubernetes/pull/42615.
 	restMapper := api.Registry.RESTMapper()
-	pluginInitializer := kubeapiserveradmission.NewPluginInitializer(client, sharedInformers, apiAuthorizer, cloudConfig, restMapper)
+
+	// NOTE: we do not provide informers to the quota registry because admission level decisions
+	// do not require us to open watches for all items tracked by quota.
+	quotaRegistry := quotainstall.NewRegistry(nil, nil)
+
+	pluginInitializer := kubeapiserveradmission.NewPluginInitializer(client, sharedInformers, apiAuthorizer, cloudConfig, restMapper, quotaRegistry)
 	return pluginInitializer, nil
 }
 

federation/cmd/federation-apiserver/app/BUILD

@@ -49,6 +49,7 @@ go_library(
         "//pkg/kubeapiserver/admission:go_default_library",
         "//pkg/kubeapiserver/options:go_default_library",
         "//pkg/kubeapiserver/server:go_default_library",
+        "//pkg/quota/install:go_default_library",
         "//pkg/registry/autoscaling/horizontalpodautoscaler/storage:go_default_library",
         "//pkg/registry/batch/job/storage:go_default_library",
         "//pkg/registry/cachesize:go_default_library",

federation/cmd/federation-apiserver/app/server.go

@@ -49,6 +49,7 @@ import (
 	kubeapiserveradmission "k8s.io/kubernetes/pkg/kubeapiserver/admission"
 	kubeoptions "k8s.io/kubernetes/pkg/kubeapiserver/options"
 	kubeserver "k8s.io/kubernetes/pkg/kubeapiserver/server"
+	quotainstall "k8s.io/kubernetes/pkg/quota/install"
 	"k8s.io/kubernetes/pkg/registry/cachesize"
 	"k8s.io/kubernetes/pkg/routes"
 	"k8s.io/kubernetes/pkg/version"
@@ -192,7 +193,10 @@ func NonBlockingRun(s *options.ServerRunOptions, stopCh <-chan struct{}) error {
 		}
 	}
 
-	pluginInitializer := kubeapiserveradmission.NewPluginInitializer(client, sharedInformers, apiAuthorizer, cloudConfig, nil)
+	// NOTE: we do not provide informers to the quota registry because admission level decisions
+	// do not require us to open watches for all items tracked by quota.
+	quotaRegistry := quotainstall.NewRegistry(nil, nil)
+	pluginInitializer := kubeapiserveradmission.NewPluginInitializer(client, sharedInformers, apiAuthorizer, cloudConfig, nil, quotaRegistry)
 
 	err = s.Admission.ApplyTo(
 		genericConfig,

pkg/kubeapiserver/admission/BUILD

@@ -29,6 +29,7 @@ go_library(
     deps = [
         "//pkg/client/clientset_generated/internalclientset:go_default_library",
         "//pkg/client/informers/informers_generated/internalversion:go_default_library",
+        "//pkg/quota:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/meta:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/admission:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/authorization/authorizer:go_default_library",

pkg/kubeapiserver/admission/init_test.go

@@ -51,7 +51,7 @@ var _ WantsAuthorizer = &WantAuthorizerAdmission{}
 // TestWantsAuthorizer ensures that the authorizer is injected when the WantsAuthorizer
 // interface is implemented.
 func TestWantsAuthorizer(t *testing.T) {
-	initializer := NewPluginInitializer(nil, nil, &TestAuthorizer{}, nil, nil)
+	initializer := NewPluginInitializer(nil, nil, &TestAuthorizer{}, nil, nil, nil)
 	wantAuthorizerAdmission := &WantAuthorizerAdmission{}
 	initializer.Initialize(wantAuthorizerAdmission)
 	if wantAuthorizerAdmission.auth == nil {
@@ -73,7 +73,7 @@ func (self *WantsCloudConfigAdmissionPlugin) Validate() error
 
 func TestCloudConfigAdmissionPlugin(t *testing.T) {
 	cloudConfig := []byte("cloud-configuration")
-	initializer := NewPluginInitializer(nil, nil, &TestAuthorizer{}, cloudConfig, nil)
+	initializer := NewPluginInitializer(nil, nil, &TestAuthorizer{}, cloudConfig, nil, nil)
 	wantsCloudConfigAdmission := &WantsCloudConfigAdmissionPlugin{}
 	initializer.Initialize(wantsCloudConfigAdmission)
 

pkg/kubeapiserver/admission/initializer.go

@@ -22,6 +22,7 @@ import (
 	"k8s.io/apiserver/pkg/authorization/authorizer"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	informers "k8s.io/kubernetes/pkg/client/informers/informers_generated/internalversion"
+	"k8s.io/kubernetes/pkg/quota"
 )
 
 // TODO add a `WantsToRun` which takes a stopCh.  Might make it generic.
@@ -54,24 +55,37 @@ type WantsRESTMapper interface {
 	SetRESTMapper(meta.RESTMapper)
 }
 
+// WantsQuotaRegistry defines a function which sets quota registry for admission plugins that need it.
+type WantsQuotaRegistry interface {
+	SetQuotaRegistry(quota.Registry)
+	admission.Validator
+}
+
 type pluginInitializer struct {
 	internalClient internalclientset.Interface
 	informers      informers.SharedInformerFactory
 	authorizer     authorizer.Authorizer
 	cloudConfig    []byte
 	restMapper     meta.RESTMapper
+	quotaRegistry  quota.Registry
 }
 
 var _ admission.PluginInitializer = pluginInitializer{}
 
 // NewPluginInitializer constructs new instance of PluginInitializer
-func NewPluginInitializer(internalClient internalclientset.Interface, sharedInformers informers.SharedInformerFactory, authz authorizer.Authorizer, cloudConfig []byte, restMapper meta.RESTMapper) admission.PluginInitializer {
+func NewPluginInitializer(internalClient internalclientset.Interface,
+	sharedInformers informers.SharedInformerFactory,
+	authz authorizer.Authorizer,
+	cloudConfig []byte,
+	restMapper meta.RESTMapper,
+	quotaRegistry quota.Registry) admission.PluginInitializer {
 	return pluginInitializer{
 		internalClient: internalClient,
 		informers:      sharedInformers,
 		authorizer:     authz,
 		cloudConfig:    cloudConfig,
 		restMapper:     restMapper,
+		quotaRegistry:  quotaRegistry,
 	}
 }
 
@@ -97,4 +111,8 @@ func (i pluginInitializer) Initialize(plugin admission.Interface) {
 	if wants, ok := plugin.(WantsRESTMapper); ok {
 		wants.SetRESTMapper(i.restMapper)
 	}
+
+	if wants, ok := plugin.(WantsQuotaRegistry); ok {
+		wants.SetQuotaRegistry(i.quotaRegistry)
+	}
 }

plugin/pkg/admission/gc/gc_admission_test.go

@@ -74,7 +74,7 @@ func newGCPermissionsEnforcement() *gcPermissionsEnforcement {
 		Handler:   admission.NewHandler(admission.Create, admission.Update),
 		whiteList: whiteList,
 	}
-	pluginInitializer := kubeadmission.NewPluginInitializer(nil, nil, fakeAuthorizer{}, nil, api.Registry.RESTMapper())
+	pluginInitializer := kubeadmission.NewPluginInitializer(nil, nil, fakeAuthorizer{}, nil, api.Registry.RESTMapper(), nil)
 	pluginInitializer.Initialize(gcAdmit)
 	return gcAdmit
 }

plugin/pkg/admission/limitranger/admission_test.go

@@ -595,7 +595,7 @@ func newHandlerForTest(c clientset.Interface) (admission.Interface, informers.Sh
 	if err != nil {
 		return nil, f, err
 	}
-	pluginInitializer := kubeadmission.NewPluginInitializer(c, f, nil, nil, nil)
+	pluginInitializer := kubeadmission.NewPluginInitializer(c, f, nil, nil, nil, nil)
 	pluginInitializer.Initialize(handler)
 	err = admission.Validate(handler)
 	return handler, f, err

plugin/pkg/admission/namespace/autoprovision/admission_test.go

@@ -38,7 +38,7 @@ import (
 func newHandlerForTest(c clientset.Interface) (admission.Interface, informers.SharedInformerFactory, error) {
 	f := informers.NewSharedInformerFactory(c, 5*time.Minute)
 	handler := NewProvision()
-	pluginInitializer := kubeadmission.NewPluginInitializer(c, f, nil, nil, nil)
+	pluginInitializer := kubeadmission.NewPluginInitializer(c, f, nil, nil, nil, nil)
 	pluginInitializer.Initialize(handler)
 	err := admission.Validate(handler)
 	return handler, f, err

plugin/pkg/admission/namespace/exists/admission_test.go

@@ -37,7 +37,7 @@ import (
 func newHandlerForTest(c clientset.Interface) (admission.Interface, informers.SharedInformerFactory, error) {
 	f := informers.NewSharedInformerFactory(c, 5*time.Minute)
 	handler := NewExists()
-	pluginInitializer := kubeadmission.NewPluginInitializer(c, f, nil, nil, nil)
+	pluginInitializer := kubeadmission.NewPluginInitializer(c, f, nil, nil, nil, nil)
 	pluginInitializer.Initialize(handler)
 	err := admission.Validate(handler)
 	return handler, f, err

plugin/pkg/admission/namespace/lifecycle/admission_test.go

@@ -48,7 +48,7 @@ func newHandlerForTestWithClock(c clientset.Interface, cacheClock clock.Clock) (
 	if err != nil {
 		return nil, f, err
 	}
-	pluginInitializer := kubeadmission.NewPluginInitializer(c, f, nil, nil, nil)
+	pluginInitializer := kubeadmission.NewPluginInitializer(c, f, nil, nil, nil, nil)
 	pluginInitializer.Initialize(handler)
 	err = admission.Validate(handler)
 	return handler, f, err

plugin/pkg/admission/podnodeselector/admission_test.go

@@ -191,7 +191,7 @@ func TestHandles(t *testing.T) {
 func newHandlerForTest(c clientset.Interface) (*podNodeSelector, informers.SharedInformerFactory, error) {
 	f := informers.NewSharedInformerFactory(c, 5*time.Minute)
 	handler := NewPodNodeSelector(nil)
-	pluginInitializer := kubeadmission.NewPluginInitializer(c, f, nil, nil, nil)
+	pluginInitializer := kubeadmission.NewPluginInitializer(c, f, nil, nil, nil, nil)
 	pluginInitializer.Initialize(handler)
 	err := admission.Validate(handler)
 	return handler, f, err

plugin/pkg/admission/podtolerationrestriction/admission_test.go

@@ -193,7 +193,7 @@ func newHandlerForTest(c clientset.Interface) (*podTolerationsPlugin, informers.
 		return nil, nil, err
 	}
 	handler := NewPodTolerationsPlugin(pluginConfig)
-	pluginInitializer := kubeadmission.NewPluginInitializer(c, f, nil, nil, nil)
+	pluginInitializer := kubeadmission.NewPluginInitializer(c, f, nil, nil, nil, nil)
 	pluginInitializer.Initialize(handler)
 	err = admission.Validate(handler)
 	return handler, f, err

plugin/pkg/admission/resourcequota/BUILD

@@ -25,7 +25,6 @@ go_library(
         "//pkg/client/listers/core/internalversion:go_default_library",
         "//pkg/kubeapiserver/admission:go_default_library",
         "//pkg/quota:go_default_library",
-        "//pkg/quota/install:go_default_library",
         "//pkg/util/workqueue/prometheus:go_default_library",
         "//plugin/pkg/admission/resourcequota/apis/resourcequota:go_default_library",
         "//plugin/pkg/admission/resourcequota/apis/resourcequota/install:go_default_library",

plugin/pkg/admission/resourcequota/admission.go

@@ -27,7 +27,6 @@ import (
 	informers "k8s.io/kubernetes/pkg/client/informers/informers_generated/internalversion"
 	kubeapiserveradmission "k8s.io/kubernetes/pkg/kubeapiserver/admission"
 	"k8s.io/kubernetes/pkg/quota"
-	"k8s.io/kubernetes/pkg/quota/install"
 	resourcequotaapi "k8s.io/kubernetes/plugin/pkg/admission/resourcequota/apis/resourcequota"
 	"k8s.io/kubernetes/plugin/pkg/admission/resourcequota/apis/resourcequota/validation"
 )
@@ -51,10 +50,7 @@ func Register(plugins *admission.Plugins) {
 					return nil, errs.ToAggregate()
 				}
 			}
-			// NOTE: we do not provide informers to the registry because admission level decisions
+			return NewResourceQuota(configuration, 5, make(chan struct{}))
-			// does not require us to open watches for all items tracked by quota.
-			registry := install.NewRegistry(nil, nil)
-			return NewResourceQuota(registry, configuration, 5, make(chan struct{}))
 		})
 }
 
@@ -70,6 +66,7 @@ type quotaAdmission struct {
 }
 
 var _ = kubeapiserveradmission.WantsInternalKubeClientSet(&quotaAdmission{})
+var _ = kubeapiserveradmission.WantsQuotaRegistry(&quotaAdmission{})
 
 type liveLookupEntry struct {
 	expiry time.Time
@@ -79,7 +76,7 @@ type liveLookupEntry struct {
 // NewResourceQuota configures an admission controller that can enforce quota constraints
 // using the provided registry.  The registry must have the capability to handle group/kinds that
 // are persisted by the server this admission controller is intercepting
-func NewResourceQuota(registry quota.Registry, config *resourcequotaapi.Configuration, numEvaluators int, stopCh <-chan struct{}) (admission.Interface, error) {
+func NewResourceQuota(config *resourcequotaapi.Configuration, numEvaluators int, stopCh <-chan struct{}) (admission.Interface, error) {
 	quotaAccessor, err := newQuotaAccessor()
 	if err != nil {
 		return nil, err
@@ -88,11 +85,9 @@ func NewResourceQuota(registry quota.Registry, config *resourcequotaapi.Configur
 	return &quotaAdmission{
 		Handler:       admission.NewHandler(admission.Create, admission.Update),
 		stopCh:        stopCh,
-		registry:      registry,
 		numEvaluators: numEvaluators,
 		config:        config,
 		quotaAccessor: quotaAccessor,
-		evaluator:     NewQuotaEvaluator(quotaAccessor, registry, nil, config, numEvaluators, stopCh),
 	}, nil
 }
 
@@ -104,6 +99,11 @@ func (a *quotaAdmission) SetInternalKubeInformerFactory(f informers.SharedInform
 	a.quotaAccessor.lister = f.Core().InternalVersion().ResourceQuotas().Lister()
 }
 
+func (a *quotaAdmission) SetQuotaRegistry(r quota.Registry) {
+	a.registry = r
+	a.evaluator = NewQuotaEvaluator(a.quotaAccessor, a.registry, nil, a.config, a.numEvaluators, a.stopCh)
+}
+
 // Validate ensures an authorizer is set.
 func (a *quotaAdmission) Validate() error {
 	if a.quotaAccessor == nil {
@@ -115,6 +115,9 @@ func (a *quotaAdmission) Validate() error {
 	if a.quotaAccessor.lister == nil {
 		return fmt.Errorf("missing quotaAccessor.lister")
 	}
+	if a.registry == nil {
+		return fmt.Errorf("missing registry")
+	}
 	if a.evaluator == nil {
 		return fmt.Errorf("missing evaluator")
 	}

test/integration/quota/quota_test.go

@@ -66,13 +66,15 @@ func TestQuota(t *testing.T) {
 	clientset := clientset.NewForConfigOrDie(&restclient.Config{QPS: -1, Host: s.URL, ContentConfig: restclient.ContentConfig{GroupVersion: &api.Registry.GroupOrDie(v1.GroupName).GroupVersion}})
 	internalClientset := internalclientset.NewForConfigOrDie(&restclient.Config{QPS: -1, Host: s.URL, ContentConfig: restclient.ContentConfig{GroupVersion: &api.Registry.GroupOrDie(v1.GroupName).GroupVersion}})
 	config := &resourcequotaapi.Configuration{}
-	admission, err := resourcequota.NewResourceQuota(quotainstall.NewRegistry(nil, nil), config, 5, admissionCh)
+	admission, err := resourcequota.NewResourceQuota(config, 5, admissionCh)
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
 	admission.(kubeadmission.WantsInternalKubeClientSet).SetInternalKubeClientSet(internalClientset)
 	internalInformers := internalinformers.NewSharedInformerFactory(internalClientset, controller.NoResyncPeriodFunc())
 	admission.(kubeadmission.WantsInternalKubeInformerFactory).SetInternalKubeInformerFactory(internalInformers)
+	quotaRegistry := quotainstall.NewRegistry(nil, nil)
+	admission.(kubeadmission.WantsQuotaRegistry).SetQuotaRegistry(quotaRegistry)
 	defer close(admissionCh)
 
 	masterConfig := framework.NewIntegrationTestMasterConfig()
@@ -251,13 +253,15 @@ func TestQuotaLimitedResourceDenial(t *testing.T) {
 			},
 		},
 	}
-	admission, err := resourcequota.NewResourceQuota(quotainstall.NewRegistry(nil, nil), config, 5, admissionCh)
+	quotaRegistry := quotainstall.NewRegistry(nil, nil)
+	admission, err := resourcequota.NewResourceQuota(config, 5, admissionCh)
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
 	admission.(kubeadmission.WantsInternalKubeClientSet).SetInternalKubeClientSet(internalClientset)
 	internalInformers := internalinformers.NewSharedInformerFactory(internalClientset, controller.NoResyncPeriodFunc())
 	admission.(kubeadmission.WantsInternalKubeInformerFactory).SetInternalKubeInformerFactory(internalInformers)
+	admission.(kubeadmission.WantsQuotaRegistry).SetQuotaRegistry(quotaRegistry)
 	defer close(admissionCh)
 
 	masterConfig := framework.NewIntegrationTestMasterConfig()