Merge pull request kubernetes#46672 from smarterclayton/initializer_w…

…ith_config

Automatic merge from submit-queue (batch tested with PRs 46967, 46992, 43338, 46717, 46672)

Select initializers from the dynamic configuration

Continues kubernetes#36721

kubernetes/enhancements#209

cmd/kube-apiserver/app/BUILD

@@ -25,6 +25,7 @@ go_library(
         "//pkg/apis/extensions:go_default_library",
         "//pkg/apis/networking:go_default_library",
         "//pkg/capabilities:go_default_library",
+        "//pkg/client/clientset_generated/clientset:go_default_library",
         "//pkg/client/clientset_generated/internalclientset:go_default_library",
         "//pkg/client/informers/informers_generated/internalversion:go_default_library",
         "//pkg/cloudprovider:go_default_library",

cmd/kube-apiserver/app/server.go

@@ -61,6 +61,7 @@ import (
 	"k8s.io/kubernetes/pkg/apis/extensions"
 	"k8s.io/kubernetes/pkg/apis/networking"
 	"k8s.io/kubernetes/pkg/capabilities"
+	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	informers "k8s.io/kubernetes/pkg/client/informers/informers_generated/internalversion"
 	"k8s.io/kubernetes/pkg/cloudprovider"
@@ -380,6 +381,10 @@ func BuildGenericConfig(s *options.ServerRunOptions) (*genericapiserver.Config,
 		// TODO: get rid of KUBE_API_VERSIONS or define sane behaviour if set
 		glog.Errorf("Failed to create clientset with KUBE_API_VERSIONS=%q. KUBE_API_VERSIONS is only for testing. Things will break.", kubeAPIVersions)
 	}
+	externalClient, err := clientset.NewForConfig(genericConfig.LoopbackClientConfig)
+	if err != nil {
+		return nil, nil, nil, fmt.Errorf("failed to create external clientset: %v", err)
+	}
 	sharedInformers := informers.NewSharedInformerFactory(client, 10*time.Minute)
 
 	genericConfig.Authenticator, genericConfig.OpenAPIConfig.SecurityDefinitions, err = BuildAuthenticator(s, storageFactory, client, sharedInformers)
@@ -398,6 +403,7 @@ func BuildGenericConfig(s *options.ServerRunOptions) (*genericapiserver.Config,
 	pluginInitializer, err := BuildAdmissionPluginInitializer(
 		s,
 		client,
+		externalClient,
 		sharedInformers,
 		genericConfig.Authorizer,
 	)
@@ -415,7 +421,7 @@ func BuildGenericConfig(s *options.ServerRunOptions) (*genericapiserver.Config,
 }
 
 // BuildAdmissionPluginInitializer constructs the admission plugin initializer
-func BuildAdmissionPluginInitializer(s *options.ServerRunOptions, client internalclientset.Interface, sharedInformers informers.SharedInformerFactory, apiAuthorizer authorizer.Authorizer) (admission.PluginInitializer, error) {
+func BuildAdmissionPluginInitializer(s *options.ServerRunOptions, client internalclientset.Interface, externalClient clientset.Interface, sharedInformers informers.SharedInformerFactory, apiAuthorizer authorizer.Authorizer) (admission.PluginInitializer, error) {
 	var cloudConfig []byte
 
 	if s.CloudProvider.CloudConfigFile != "" {
@@ -433,7 +439,7 @@ func BuildAdmissionPluginInitializer(s *options.ServerRunOptions, client interna
 	// 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)
+	pluginInitializer := kubeapiserveradmission.NewPluginInitializer(client, externalClient, sharedInformers, apiAuthorizer, cloudConfig, restMapper, quotaRegistry)
 
 	// Read client cert/key for plugins that need to make calls out
 	if len(s.ProxyClientCertFile) > 0 && len(s.ProxyClientKeyFile) > 0 {

federation/cluster/common.sh

@@ -258,7 +258,7 @@ function create-federation-api-objects {
     export FEDERATION_APISERVER_KEY_BASE64="${FEDERATION_APISERVER_KEY_BASE64}"
 
     # Enable the NamespaceLifecycle admission control by default.
-    export FEDERATION_ADMISSION_CONTROL="${FEDERATION_ADMISSION_CONTROL:-Initializers,NamespaceLifecycle}"
+    export FEDERATION_ADMISSION_CONTROL="${FEDERATION_ADMISSION_CONTROL:-NamespaceLifecycle}"
 
     for file in federation-etcd-pvc.yaml federation-apiserver-{deployment,secrets}.yaml federation-controller-manager-deployment.yaml; do
       echo "Creating manifest: ${file}"

federation/cmd/federation-apiserver/app/BUILD

@@ -41,6 +41,7 @@ go_library(
         "//pkg/apis/extensions:go_default_library",
         "//pkg/apis/extensions/install:go_default_library",
         "//pkg/apis/extensions/v1beta1:go_default_library",
+        "//pkg/client/clientset_generated/clientset:go_default_library",
         "//pkg/client/clientset_generated/internalclientset:go_default_library",
         "//pkg/client/informers/informers_generated/internalversion:go_default_library",
         "//pkg/cloudprovider/providers:go_default_library",

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

@@ -42,6 +42,7 @@ import (
 	"k8s.io/kubernetes/pkg/api"
 	apiv1 "k8s.io/kubernetes/pkg/api/v1"
 	extensionsapiv1beta1 "k8s.io/kubernetes/pkg/apis/extensions/v1beta1"
+	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	informers "k8s.io/kubernetes/pkg/client/informers/informers_generated/internalversion"
 	"k8s.io/kubernetes/pkg/generated/openapi"
@@ -180,6 +181,10 @@ func NonBlockingRun(s *options.ServerRunOptions, stopCh <-chan struct{}) error {
 	if err != nil {
 		return fmt.Errorf("failed to create clientset: %v", err)
 	}
+	externalClient, err := clientset.NewForConfig(genericConfig.LoopbackClientConfig)
+	if err != nil {
+		return fmt.Errorf("failed to create external clientset: %v", err)
+	}
 	sharedInformers := informers.NewSharedInformerFactory(client, 10*time.Minute)
 
 	authorizationConfig := s.Authorization.ToAuthorizationConfig(sharedInformers)
@@ -199,7 +204,7 @@ func NonBlockingRun(s *options.ServerRunOptions, stopCh <-chan struct{}) error {
 	// 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)
+	pluginInitializer := kubeapiserveradmission.NewPluginInitializer(client, externalClient, sharedInformers, apiAuthorizer, cloudConfig, nil, quotaRegistry)
 
 	err = s.Admission.ApplyTo(
 		genericConfig,

federation/pkg/kubefed/init/init.go

@@ -692,7 +692,7 @@ func createAPIServer(clientset client.Interface, namespace, name, federationName
 		"--client-ca-file":       "/etc/federation/apiserver/ca.crt",
 		"--tls-cert-file":        "/etc/federation/apiserver/server.crt",
 		"--tls-private-key-file": "/etc/federation/apiserver/server.key",
-		"--admission-control":    "Initializers,NamespaceLifecycle",
+		"--admission-control":    "NamespaceLifecycle",
 	}
 
 	if advertiseAddress != "" {

federation/pkg/kubefed/init/init_test.go

@@ -869,7 +869,7 @@ func fakeInitHostFactory(apiserverServiceType v1.ServiceType, federationName, na
 		fmt.Sprintf("--secure-port=%d", apiServerSecurePort),
 		"--tls-cert-file=/etc/federation/apiserver/server.crt",
 		"--tls-private-key-file=/etc/federation/apiserver/server.key",
-		"--admission-control=Initializers,NamespaceLifecycle",
+		"--admission-control=NamespaceLifecycle",
 		fmt.Sprintf("--advertise-address=%s", address),
 	}
 

hack/make-rules/test-federation-cmd.sh

@@ -34,7 +34,7 @@ function run_federation_apiserver() {
   kube::log::status "Starting federation-apiserver"
 
   # Admission Controllers to invoke prior to persisting objects in cluster
-  ADMISSION_CONTROL="Initializers,NamespaceLifecycle"
+  ADMISSION_CONTROL="NamespaceLifecycle"
 
   "${KUBE_OUTPUT_HOSTBIN}/federation-apiserver" \
     --insecure-port="${API_PORT}" \

pkg/kubeapiserver/admission/BUILD

@@ -26,6 +26,7 @@ go_library(
     tags = ["automanaged"],
     deps = [
         "//pkg/apis/admissionregistration:go_default_library",
+        "//pkg/client/clientset_generated/clientset:go_default_library",
         "//pkg/client/clientset_generated/internalclientset:go_default_library",
         "//pkg/client/informers/informers_generated/internalversion:go_default_library",
         "//pkg/quota:go_default_library",

pkg/kubeapiserver/admission/configuration/BUILD

@@ -29,6 +29,8 @@ go_library(
     tags = ["automanaged"],
     deps = [
         "//pkg/apis/admissionregistration/v1alpha1:go_default_library",
+        "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",

pkg/kubeapiserver/admission/configuration/configuration_manager.go

@@ -30,6 +30,11 @@ const (
 	defaultFailureThreshold = 5
 )
 
+var (
+	ErrNotReady = fmt.Errorf("configuration is not ready")
+	ErrDisabled = fmt.Errorf("disabled")
+)
+
 type getFunc func() (runtime.Object, error)
 
 // When running, poller calls `get` every `interval`. If `get` is
@@ -52,7 +57,8 @@ type poller struct {
 	ready               bool
 	mergedConfiguration runtime.Object
 	// lock much be hold when reading ready or mergedConfiguration
-	lock sync.RWMutex
+	lock    sync.RWMutex
+	lastErr error
 }
 
 func newPoller(get getFunc) *poller {
@@ -63,6 +69,12 @@ func newPoller(get getFunc) *poller {
 	}
 }
 
+func (a *poller) lastError(err error) {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+	a.lastErr = err
+}
+
 func (a *poller) notReady() {
 	a.lock.Lock()
 	defer a.lock.Unlock()
@@ -73,7 +85,10 @@ func (a *poller) configuration() (runtime.Object, error) {
 	a.lock.RLock()
 	defer a.lock.RUnlock()
 	if !a.ready {
-		return nil, fmt.Errorf("configuration is not ready")
+		if a.lastErr != nil {
+			return nil, a.lastErr
+		}
+		return nil, ErrNotReady
 	}
 	return a.mergedConfiguration, nil
 }
@@ -83,6 +98,7 @@ func (a *poller) setConfigurationAndReady(value runtime.Object) {
 	defer a.lock.Unlock()
 	a.mergedConfiguration = value
 	a.ready = true
+	a.lastErr = nil
 }
 
 func (a *poller) Run(stopCh <-chan struct{}) {
@@ -93,6 +109,7 @@ func (a *poller) sync() {
 	configuration, err := a.get()
 	if err != nil {
 		a.failures++
+		a.lastError(err)
 		if a.failures >= a.failureThreshold {
 			a.notReady()
 		}

pkg/kubeapiserver/admission/configuration/external_admission_hook_manager.go

@@ -33,6 +33,20 @@ type ExternalAdmissionHookConfigurationManager struct {
 	*poller
 }
 
+func NewExternalAdmissionHookConfigurationManager(c ExternalAdmissionHookConfigurationLister) *ExternalAdmissionHookConfigurationManager {
+	getFn := func() (runtime.Object, error) {
+		list, err := c.List(metav1.ListOptions{})
+		if err != nil {
+			return nil, err
+		}
+		return mergeExternalAdmissionHookConfigurations(list), nil
+	}
+
+	return &ExternalAdmissionHookConfigurationManager{
+		newPoller(getFn),
+	}
+}
+
 // ExternalAdmissionHooks returns the merged ExternalAdmissionHookConfiguration.
 func (im *ExternalAdmissionHookConfigurationManager) ExternalAdmissionHooks() (*v1alpha1.ExternalAdmissionHookConfiguration, error) {
 	configuration, err := im.poller.configuration()
@@ -46,17 +60,8 @@ func (im *ExternalAdmissionHookConfigurationManager) ExternalAdmissionHooks() (*
 	return externalAdmissionHookConfiguration, nil
 }
 
-func NewExternalAdmissionHookConfigurationManager(c ExternalAdmissionHookConfigurationLister) *ExternalAdmissionHookConfigurationManager {
-	getFn := func() (runtime.Object, error) {
-		list, err := c.List(metav1.ListOptions{})
-		if err != nil {
-			return nil, err
-		}
-		return mergeExternalAdmissionHookConfigurations(list), nil
-	}
-
-	return &ExternalAdmissionHookConfigurationManager{
-		newPoller(getFn)}
+func (im *ExternalAdmissionHookConfigurationManager) Run(stopCh <-chan struct{}) {
+	im.poller.Run(stopCh)
 }
 
 func mergeExternalAdmissionHookConfigurations(

pkg/kubeapiserver/admission/configuration/initializer_manager.go

@@ -21,6 +21,9 @@ import (
 	"reflect"
 	"sort"
 
+	"github.com/golang/glog"
+
+	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/kubernetes/pkg/apis/admissionregistration/v1alpha1"
@@ -34,6 +37,23 @@ type InitializerConfigurationManager struct {
 	*poller
 }
 
+func NewInitializerConfigurationManager(c InitializerConfigurationLister) *InitializerConfigurationManager {
+	getFn := func() (runtime.Object, error) {
+		list, err := c.List(metav1.ListOptions{})
+		if err != nil {
+			if errors.IsNotFound(err) || errors.IsForbidden(err) {
+				glog.V(5).Infof("Initializers are disabled due to an error: %v", err)
+				return nil, ErrDisabled
+			}
+			return nil, err
+		}
+		return mergeInitializerConfigurations(list), nil
+	}
+	return &InitializerConfigurationManager{
+		newPoller(getFn),
+	}
+}
+
 // Initializers returns the merged InitializerConfiguration.
 func (im *InitializerConfigurationManager) Initializers() (*v1alpha1.InitializerConfiguration, error) {
 	configuration, err := im.poller.configuration()
@@ -47,16 +67,8 @@ func (im *InitializerConfigurationManager) Initializers() (*v1alpha1.Initializer
 	return initializerConfiguration, nil
 }
 
-func NewInitializerConfigurationManager(c InitializerConfigurationLister) *InitializerConfigurationManager {
-	getFn := func() (runtime.Object, error) {
-		list, err := c.List(metav1.ListOptions{})
-		if err != nil {
-			return nil, err
-		}
-		return mergeInitializerConfigurations(list), nil
-	}
-	return &InitializerConfigurationManager{
-		newPoller(getFn)}
+func (im *InitializerConfigurationManager) Run(stopCh <-chan struct{}) {
+	im.poller.Run(stopCh)
 }
 
 func mergeInitializerConfigurations(initializerConfigurationList *v1alpha1.InitializerConfigurationList) *v1alpha1.InitializerConfiguration {

pkg/kubeapiserver/admission/init_test.go

@@ -57,7 +57,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, nil)
+	initializer := NewPluginInitializer(nil, nil, nil, &TestAuthorizer{}, nil, nil, nil)
 	wantAuthorizerAdmission := &WantAuthorizerAdmission{}
 	initializer.Initialize(wantAuthorizerAdmission)
 	if wantAuthorizerAdmission.auth == nil {
@@ -76,7 +76,7 @@ func (self *WantsCloudConfigAdmissionPlugin) SetCloudConfig(cloudConfig []byte)
 
 func TestCloudConfigAdmissionPlugin(t *testing.T) {
 	cloudConfig := []byte("cloud-configuration")
-	initializer := NewPluginInitializer(nil, nil, &TestAuthorizer{}, cloudConfig, nil, nil)
+	initializer := NewPluginInitializer(nil, nil, nil, &TestAuthorizer{}, cloudConfig, nil, nil)
 	wantsCloudConfigAdmission := &WantsCloudConfigAdmissionPlugin{}
 	initializer.Initialize(wantsCloudConfigAdmission)
 

pkg/kubeapiserver/admission/initializer.go

@@ -23,6 +23,7 @@ import (
 	"k8s.io/apiserver/pkg/admission"
 	"k8s.io/apiserver/pkg/authorization/authorizer"
 	"k8s.io/kubernetes/pkg/apis/admissionregistration"
+	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	informers "k8s.io/kubernetes/pkg/client/informers/informers_generated/internalversion"
 	"k8s.io/kubernetes/pkg/quota"
@@ -36,6 +37,12 @@ type WantsInternalKubeClientSet interface {
 	admission.Validator
 }
 
+// WantsExternalKubeClientSet defines a function which sets ClientSet for admission plugins that need it
+type WantsExternalKubeClientSet interface {
+	SetExternalKubeClientSet(clientset.Interface)
+	admission.Validator
+}
+
 // WantsInternalKubeInformerFactory defines a function which sets InformerFactory for admission plugins that need it
 type WantsInternalKubeInformerFactory interface {
 	SetInternalKubeInformerFactory(informers.SharedInformerFactory)
@@ -95,6 +102,7 @@ type WebhookSource interface {
 
 type PluginInitializer struct {
 	internalClient  internalclientset.Interface
+	externalClient  clientset.Interface
 	informers       informers.SharedInformerFactory
 	authorizer      authorizer.Authorizer
 	cloudConfig     []byte
@@ -113,14 +121,18 @@ var _ admission.PluginInitializer = &PluginInitializer{}
 // NewPluginInitializer constructs new instance of PluginInitializer
 // TODO: switch these parameters to use the builder pattern or just make them
 // all public, this construction method is pointless boilerplate.
-func NewPluginInitializer(internalClient internalclientset.Interface,
+func NewPluginInitializer(
+	internalClient internalclientset.Interface,
+	externalClient clientset.Interface,
 	sharedInformers informers.SharedInformerFactory,
 	authz authorizer.Authorizer,
 	cloudConfig []byte,
 	restMapper meta.RESTMapper,
-	quotaRegistry quota.Registry) *PluginInitializer {
+	quotaRegistry quota.Registry,
+) *PluginInitializer {
 	return &PluginInitializer{
 		internalClient: internalClient,
+		externalClient: externalClient,
 		informers:      sharedInformers,
 		authorizer:     authz,
 		cloudConfig:    cloudConfig,
@@ -157,6 +169,10 @@ func (i *PluginInitializer) Initialize(plugin admission.Interface) {
 		wants.SetInternalKubeClientSet(i.internalClient)
 	}
 
+	if wants, ok := plugin.(WantsExternalKubeClientSet); ok {
+		wants.SetExternalKubeClientSet(i.externalClient)
+	}
+
 	if wants, ok := plugin.(WantsInternalKubeInformerFactory); ok {
 		wants.SetInternalKubeInformerFactory(i.informers)
 	}

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(), nil)
+	pluginInitializer := kubeadmission.NewPluginInitializer(nil, nil, nil, fakeAuthorizer{}, nil, api.Registry.RESTMapper(), nil)
 	pluginInitializer.Initialize(gcAdmit)
 	return gcAdmit
 }