Merge pull request #35230 from deads2k/controller-12-sa-controller

Automatic merge from submit-queue

convert SA controller to shared informers

convert the SA controller to shared informer + workqueue.

I think one of @derekwaynecarr @ncdc or @liggitt

cmd/kube-controller-manager/app/controllermanager.go

@@ -529,10 +529,11 @@ func StartControllers(s *options.CMServer, kubeconfig *restclient.Config, rootCl
 		}
 	}
 
-	serviceaccountcontroller.NewServiceAccountsController(
+	go serviceaccountcontroller.NewServiceAccountsController(
+		sharedInformers.ServiceAccounts(), sharedInformers.Namespaces(),
 		client("service-account-controller"),
 		serviceaccountcontroller.DefaultServiceAccountsControllerOptions(),
-	).Run()
+	).Run(1, stop)
 	time.Sleep(wait.Jitter(s.ControllerStartInterval.Duration, ControllerStartJitter))
 
 	if s.EnableGarbageCollector {

pkg/client/cache/listers.go

@@ -450,23 +450,6 @@ func (s *StoreToCertificateRequestLister) List() (csrs certificates.CertificateS
 	return csrs, nil
 }
 
-// IndexerToNamespaceLister gives an Indexer List method
-type IndexerToNamespaceLister struct {
-	Indexer
-}
-
-// List returns a list of namespaces
-func (i *IndexerToNamespaceLister) List(selector labels.Selector) (namespaces []*api.Namespace, err error) {
-	for _, m := range i.Indexer.List() {
-		namespace := m.(*api.Namespace)
-		if selector.Matches(labels.Set(namespace.Labels)) {
-			namespaces = append(namespaces, namespace)
-		}
-	}
-
-	return namespaces, nil
-}
-
 type StoreToPodDisruptionBudgetLister struct {
 	Store
 }

pkg/client/cache/listers_core.go

@@ -322,3 +322,27 @@ func (s storePersistentVolumeClaimsNamespacer) Get(name string) (*api.Persistent
 	}
 	return obj.(*api.PersistentVolumeClaim), nil
 }
+
+// IndexerToNamespaceLister gives an Indexer List method
+type IndexerToNamespaceLister struct {
+	Indexer
+}
+
+// List returns a list of namespaces
+func (i *IndexerToNamespaceLister) List(selector labels.Selector) (ret []*api.Namespace, err error) {
+	err = ListAll(i.Indexer, selector, func(m interface{}) {
+		ret = append(ret, m.(*api.Namespace))
+	})
+	return ret, err
+}
+
+func (i *IndexerToNamespaceLister) Get(name string) (*api.Namespace, error) {
+	obj, exists, err := i.Indexer.GetByKey(name)
+	if err != nil {
+		return nil, err
+	}
+	if !exists {
+		return nil, errors.NewNotFound(api.Resource("namespace"), name)
+	}
+	return obj.(*api.Namespace), nil
+}

pkg/controller/serviceaccount/BUILD

@@ -26,6 +26,7 @@ go_library(
         "//pkg/client/cache:go_default_library",
         "//pkg/client/clientset_generated/internalclientset:go_default_library",
         "//pkg/client/retry:go_default_library",
+        "//pkg/controller/informers:go_default_library",
         "//pkg/fields:go_default_library",
         "//pkg/registry/core/secret:go_default_library",
         "//pkg/registry/core/secret/etcd:go_default_library",
@@ -59,8 +60,11 @@ go_test(
         "//pkg/api:go_default_library",
         "//pkg/api/errors:go_default_library",
         "//pkg/api/unversioned:go_default_library",
+        "//pkg/client/cache:go_default_library",
         "//pkg/client/clientset_generated/internalclientset/fake:go_default_library",
         "//pkg/client/testing/core:go_default_library",
+        "//pkg/controller:go_default_library",
+        "//pkg/controller/informers:go_default_library",
         "//pkg/runtime:go_default_library",
         "//pkg/util/rand:go_default_library",
         "//pkg/util/sets:go_default_library",

pkg/controller/serviceaccount/serviceaccounts_controller.go

@@ -26,10 +26,12 @@ import (
 	"k8s.io/kubernetes/pkg/api/meta"
 	"k8s.io/kubernetes/pkg/client/cache"
 	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
-	"k8s.io/kubernetes/pkg/fields"
+	"k8s.io/kubernetes/pkg/controller/informers"
-	"k8s.io/kubernetes/pkg/runtime"
+	utilerrors "k8s.io/kubernetes/pkg/util/errors"
 	"k8s.io/kubernetes/pkg/util/metrics"
-	"k8s.io/kubernetes/pkg/watch"
+	utilruntime "k8s.io/kubernetes/pkg/util/runtime"
+	"k8s.io/kubernetes/pkg/util/wait"
+	"k8s.io/kubernetes/pkg/util/workqueue"
 )
 
 // nameIndexFunc is an index function that indexes based on an object's name
@@ -66,193 +68,159 @@ func DefaultServiceAccountsControllerOptions() ServiceAccountsControllerOptions
 }
 
 // NewServiceAccountsController returns a new *ServiceAccountsController.
-func NewServiceAccountsController(cl clientset.Interface, options ServiceAccountsControllerOptions) *ServiceAccountsController {
+func NewServiceAccountsController(saInformer informers.ServiceAccountInformer, nsInformer informers.NamespaceInformer, cl clientset.Interface, options ServiceAccountsControllerOptions) *ServiceAccountsController {
 	e := &ServiceAccountsController{
 		client:                  cl,
 		serviceAccountsToEnsure: options.ServiceAccounts,
+		queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "serviceaccount"),
 	}
 	if cl != nil && cl.Core().RESTClient().GetRateLimiter() != nil {
 		metrics.RegisterMetricAndTrackRateLimiterUsage("serviceaccount_controller", cl.Core().RESTClient().GetRateLimiter())
 	}
-	accountSelector := fields.Everything()
-	if len(options.ServiceAccounts) == 1 {
-		// If we're maintaining a single account, we can scope the accounts we watch to just that name
-		accountSelector = fields.SelectorFromSet(map[string]string{api.ObjectNameField: options.ServiceAccounts[0].Name})
-	}
-	e.serviceAccounts, e.serviceAccountController = cache.NewIndexerInformer(
-		&cache.ListWatch{
-			ListFunc: func(options api.ListOptions) (runtime.Object, error) {
-				options.FieldSelector = accountSelector
-				return e.client.Core().ServiceAccounts(api.NamespaceAll).List(options)
-			},
-			WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
-				options.FieldSelector = accountSelector
-				return e.client.Core().ServiceAccounts(api.NamespaceAll).Watch(options)
-			},
-		},
-		&api.ServiceAccount{},
-		options.ServiceAccountResync,
-		cache.ResourceEventHandlerFuncs{
-			DeleteFunc: e.serviceAccountDeleted,
-		},
-		cache.Indexers{"namespace": cache.MetaNamespaceIndexFunc},
-	)
 
-	e.namespaces, e.namespaceController = cache.NewIndexerInformer(
+	saInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		&cache.ListWatch{
+		DeleteFunc: e.serviceAccountDeleted,
-			ListFunc: func(options api.ListOptions) (runtime.Object, error) {
+	})
-				return e.client.Core().Namespaces().List(options)
+	nsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-			},
-			WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
-				return e.client.Core().Namespaces().Watch(options)
-			},
-		},
-		&api.Namespace{},
-		options.NamespaceResync,
-		cache.ResourceEventHandlerFuncs{
 		AddFunc:    e.namespaceAdded,
 		UpdateFunc: e.namespaceUpdated,
-		},
+	})
-		cache.Indexers{"name": nameIndexFunc},
+
-	)
+	e.saSynced = saInformer.Informer().HasSynced
+	e.saLister = saInformer.Lister()
+	e.nsSynced = nsInformer.Informer().HasSynced
+	e.nsLister = nsInformer.Lister()
+
+	e.syncHandler = e.syncNamespace
 
 	return e
 }
 
 // ServiceAccountsController manages ServiceAccount objects inside Namespaces
 type ServiceAccountsController struct {
-	stopChan chan struct{}
-
 	client                  clientset.Interface
 	serviceAccountsToEnsure []api.ServiceAccount
 
-	serviceAccounts cache.Indexer
+	// To allow injection for testing.
-	namespaces      cache.Indexer
+	syncHandler func(key string) error
 
-	// Since we join two objects, we'll watch both of them with controllers.
+	saLister *cache.StoreToServiceAccountLister
-	serviceAccountController *cache.Controller
+	nsLister *cache.IndexerToNamespaceLister
-	namespaceController      *cache.Controller
+
+	saSynced cache.InformerSynced
+	nsSynced cache.InformerSynced
+
+	queue workqueue.RateLimitingInterface
 }
 
-// Runs controller loops and returns immediately
+func (c *ServiceAccountsController) Run(workers int, stopCh <-chan struct{}) {
-func (e *ServiceAccountsController) Run() {
+	defer utilruntime.HandleCrash()
-	if e.stopChan == nil {
+	defer c.queue.ShutDown()
-		e.stopChan = make(chan struct{})
-		go e.serviceAccountController.Run(e.stopChan)
-		go e.namespaceController.Run(e.stopChan)
-	}
-}
 
-// Stop gracefully shuts down this controller
+	glog.Infof("Starting ServiceAccount controller")
-func (e *ServiceAccountsController) Stop() {
+
-	if e.stopChan != nil {
+	if !cache.WaitForCacheSync(stopCh, c.saSynced) {
-		close(e.stopChan)
+		return
-		e.stopChan = nil
 	}
+
+	for i := 0; i < workers; i++ {
+		go wait.Until(c.runWorker, time.Second, stopCh)
+	}
+
+	<-stopCh
+	glog.Infof("Shutting down ServiceAccount controller")
 }
 
 // serviceAccountDeleted reacts to a ServiceAccount deletion by recreating a default ServiceAccount in the namespace if needed
-func (e *ServiceAccountsController) serviceAccountDeleted(obj interface{}) {
+func (c *ServiceAccountsController) serviceAccountDeleted(obj interface{}) {
-	serviceAccount, ok := obj.(*api.ServiceAccount)
+	sa, ok := obj.(*api.ServiceAccount)
 	if !ok {
-		// Unknown type. If we missed a ServiceAccount deletion, the
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
-		// corresponding secrets will be cleaned up during the Secret re-list
+		if !ok {
+			utilruntime.HandleError(fmt.Errorf("Couldn't get object from tombstone %#v", obj))
 			return
 		}
-	// If the deleted service account is one we're maintaining, recreate it
+		sa, ok = tombstone.Obj.(*api.ServiceAccount)
-	for _, sa := range e.serviceAccountsToEnsure {
+		if !ok {
-		if sa.Name == serviceAccount.Name {
+			utilruntime.HandleError(fmt.Errorf("Tombstone contained object that is not a ServiceAccount %#v", obj))
-			e.createServiceAccountIfNeeded(sa, serviceAccount.Namespace)
+			return
 		}
 	}
+	c.queue.Add(sa.Namespace)
 }
 
 // namespaceAdded reacts to a Namespace creation by creating a default ServiceAccount object
-func (e *ServiceAccountsController) namespaceAdded(obj interface{}) {
+func (c *ServiceAccountsController) namespaceAdded(obj interface{}) {
 	namespace := obj.(*api.Namespace)
-	for _, sa := range e.serviceAccountsToEnsure {
+	c.queue.Add(namespace.Name)
-		e.createServiceAccountIfNeeded(sa, namespace.Name)
-	}
 }
 
 // namespaceUpdated reacts to a Namespace update (or re-list) by creating a default ServiceAccount in the namespace if needed
-func (e *ServiceAccountsController) namespaceUpdated(oldObj interface{}, newObj interface{}) {
+func (c *ServiceAccountsController) namespaceUpdated(oldObj interface{}, newObj interface{}) {
 	newNamespace := newObj.(*api.Namespace)
-	for _, sa := range e.serviceAccountsToEnsure {
+	c.queue.Add(newNamespace.Name)
-		e.createServiceAccountIfNeeded(sa, newNamespace.Name)
+}
+
+func (c *ServiceAccountsController) runWorker() {
+	for c.processNextWorkItem() {
 	}
 }
 
-// createServiceAccountIfNeeded creates a ServiceAccount with the given name in the given namespace if:
+// processNextWorkItem deals with one key off the queue.  It returns false when it's time to quit.
-// * the named ServiceAccount does not already exist
+func (c *ServiceAccountsController) processNextWorkItem() bool {
-// * the specified namespace exists
+	key, quit := c.queue.Get()
-// * the specified namespace is in the ACTIVE phase
+	if quit {
-func (e *ServiceAccountsController) createServiceAccountIfNeeded(sa api.ServiceAccount, namespace string) {
+		return false
-	existingServiceAccount, err := e.getServiceAccount(sa.Name, namespace)
-	if err != nil {
-		glog.Error(err)
-		return
 	}
-	if existingServiceAccount != nil {
+	defer c.queue.Done(key)
-		// If service account already exists, it doesn't need to be created
+
-		return
+	err := c.syncHandler(key.(string))
+	if err == nil {
+		c.queue.Forget(key)
+		return true
 	}
 
-	namespaceObj, err := e.getNamespace(namespace)
+	utilruntime.HandleError(fmt.Errorf("%v failed with : %v", key, err))
+	c.queue.AddRateLimited(key)
+
+	return true
+}
+func (c *ServiceAccountsController) syncNamespace(key string) error {
+	startTime := time.Now()
+	defer func() {
+		glog.V(4).Infof("Finished syncing namespace %q (%v)", key, time.Now().Sub(startTime))
+	}()
+
+	ns, err := c.nsLister.Get(key)
+	if apierrs.IsNotFound(err) {
+		return nil
+	}
 	if err != nil {
-		glog.Error(err)
+		return err
-		return
 	}
-	if namespaceObj == nil {
+	if ns.Status.Phase != api.NamespaceActive {
-		// If namespace does not exist, no service account is needed
-		return
-	}
-	if namespaceObj.Status.Phase != api.NamespaceActive {
 		// If namespace is not active, we shouldn't try to create anything
-		return
+		return nil
 	}
 
-	e.createServiceAccount(sa, namespace)
+	createFailures := []error{}
-}
+	for i := range c.serviceAccountsToEnsure {
-
+		sa := c.serviceAccountsToEnsure[i]
-// createDefaultServiceAccount creates a default ServiceAccount in the specified namespace
+		switch _, err := c.saLister.ServiceAccounts(ns.Name).Get(sa.Name); {
-func (e *ServiceAccountsController) createServiceAccount(sa api.ServiceAccount, namespace string) {
+		case err == nil:
-	sa.Namespace = namespace
+			continue
-	if _, err := e.client.Core().ServiceAccounts(namespace).Create(&sa); err != nil && !apierrs.IsAlreadyExists(err) {
+		case apierrs.IsNotFound(err):
-		glog.Error(err)
+		case err != nil:
-	}
+			return err
-}
+		}
-
+		// this is only safe because we never read it and we always write it
-// getServiceAccount returns the ServiceAccount with the given name for the given namespace
+		// TODO eliminate this once the fake client can handle creation without NS
-func (e *ServiceAccountsController) getServiceAccount(name, namespace string) (*api.ServiceAccount, error) {
+		sa.Namespace = ns.Name
-	key := &api.ServiceAccount{ObjectMeta: api.ObjectMeta{Namespace: namespace}}
+
-	accounts, err := e.serviceAccounts.Index("namespace", key)
+		if _, err := c.client.Core().ServiceAccounts(ns.Name).Create(&sa); err != nil && !apierrs.IsAlreadyExists(err) {
-	if err != nil {
+			createFailures = append(createFailures, err)
-		return nil, err
+		}
-	}
+	}
-
+
-	for _, obj := range accounts {
+	return utilerrors.Flatten(utilerrors.NewAggregate(createFailures))
-		serviceAccount := obj.(*api.ServiceAccount)
-		if name == serviceAccount.Name {
-			return serviceAccount, nil
-		}
-	}
-	return nil, nil
-}
-
-// getNamespace returns the Namespace with the given name
-func (e *ServiceAccountsController) getNamespace(name string) (*api.Namespace, error) {
-	key := &api.Namespace{ObjectMeta: api.ObjectMeta{Name: name}}
-	namespaces, err := e.namespaces.Index("name", key)
-	if err != nil {
-		return nil, err
-	}
-
-	if len(namespaces) == 0 {
-		return nil, nil
-	}
-	if len(namespaces) == 1 {
-		return namespaces[0].(*api.Namespace), nil
-	}
-	return nil, fmt.Errorf("%d namespaces with the name %s indexed", len(namespaces), name)
 }

pkg/controller/serviceaccount/serviceaccounts_controller_test.go

@@ -18,10 +18,14 @@ package serviceaccount
 
 import (
 	"testing"
+	"time"
 
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/client/cache"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/fake"
 	"k8s.io/kubernetes/pkg/client/testing/core"
+	"k8s.io/kubernetes/pkg/controller"
+	"k8s.io/kubernetes/pkg/controller/informers"
 	"k8s.io/kubernetes/pkg/util/sets"
 )
 
@@ -128,6 +132,7 @@ func TestServiceAccountCreation(t *testing.T) {
 			ExpectCreatedServiceAccounts: []string{},
 		},
 		"deleted serviceaccount with active namespace": {
+			ExistingServiceAccounts:      []*api.ServiceAccount{managedServiceAccount},
 			ExistingNamespace:            activeNS,
 			DeletedServiceAccount:        defaultServiceAccount,
 			ExpectCreatedServiceAccounts: []string{defaultName},
@@ -138,6 +143,7 @@ func TestServiceAccountCreation(t *testing.T) {
 			ExpectCreatedServiceAccounts: []string{},
 		},
 		"deleted unmanaged serviceaccount with active namespace": {
+			ExistingServiceAccounts:      []*api.ServiceAccount{defaultServiceAccount, managedServiceAccount},
 			ExistingNamespace:            activeNS,
 			DeletedServiceAccount:        unmanagedServiceAccount,
 			ExpectCreatedServiceAccounts: []string{},
@@ -151,32 +157,58 @@ func TestServiceAccountCreation(t *testing.T) {
 
 	for k, tc := range testcases {
 		client := fake.NewSimpleClientset(defaultServiceAccount, managedServiceAccount)
+		informers := informers.NewSharedInformerFactory(fake.NewSimpleClientset(), controller.NoResyncPeriodFunc())
 		options := DefaultServiceAccountsControllerOptions()
 		options.ServiceAccounts = []api.ServiceAccount{
 			{ObjectMeta: api.ObjectMeta{Name: defaultName}},
 			{ObjectMeta: api.ObjectMeta{Name: managedName}},
 		}
-		controller := NewServiceAccountsController(client, options)
+		controller := NewServiceAccountsController(informers.ServiceAccounts(), informers.Namespaces(), client, options)
+		controller.saLister = &cache.StoreToServiceAccountLister{Indexer: cache.NewIndexer(cache.DeletionHandlingMetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})}
+		controller.nsLister = &cache.IndexerToNamespaceLister{Indexer: cache.NewIndexer(cache.DeletionHandlingMetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})}
+		controller.saSynced = alwaysReady
+		controller.nsSynced = alwaysReady
+
+		syncCalls := make(chan struct{})
+		controller.syncHandler = func(key string) error {
+			err := controller.syncNamespace(key)
+			if err != nil {
+				t.Logf("%s: %v", k, err)
+			}
+
+			syncCalls <- struct{}{}
+			return err
+		}
+		stopCh := make(chan struct{})
+		defer close(stopCh)
+		go controller.Run(1, stopCh)
 
 		if tc.ExistingNamespace != nil {
-			controller.namespaces.Add(tc.ExistingNamespace)
+			controller.nsLister.Add(tc.ExistingNamespace)
 		}
 		for _, s := range tc.ExistingServiceAccounts {
-			controller.serviceAccounts.Add(s)
+			controller.saLister.Indexer.Add(s)
 		}
 
 		if tc.AddedNamespace != nil {
-			controller.namespaces.Add(tc.AddedNamespace)
+			controller.nsLister.Add(tc.AddedNamespace)
 			controller.namespaceAdded(tc.AddedNamespace)
 		}
 		if tc.UpdatedNamespace != nil {
-			controller.namespaces.Add(tc.UpdatedNamespace)
+			controller.nsLister.Add(tc.UpdatedNamespace)
 			controller.namespaceUpdated(nil, tc.UpdatedNamespace)
 		}
 		if tc.DeletedServiceAccount != nil {
 			controller.serviceAccountDeleted(tc.DeletedServiceAccount)
 		}
 
+		// wait to be called
+		select {
+		case <-syncCalls:
+		case <-time.After(10 * time.Second):
+			t.Errorf("%s: took too long", k)
+		}
+
 		actions := client.Actions()
 		if len(tc.ExpectCreatedServiceAccounts) != len(actions) {
 			t.Errorf("%s: Expected to create accounts %#v. Actual actions were: %#v", k, tc.ExpectCreatedServiceAccounts, actions)
@@ -195,3 +227,5 @@ func TestServiceAccountCreation(t *testing.T) {
 		}
 	}
 }
+
+var alwaysReady = func() bool { return true }

test/integration/serviceaccount/service_account_test.go

@@ -40,6 +40,8 @@ import (
 	"k8s.io/kubernetes/pkg/auth/user"
 	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	"k8s.io/kubernetes/pkg/client/restclient"
+	"k8s.io/kubernetes/pkg/controller"
+	"k8s.io/kubernetes/pkg/controller/informers"
 	serviceaccountcontroller "k8s.io/kubernetes/pkg/controller/serviceaccount"
 	"k8s.io/kubernetes/pkg/serviceaccount"
 	"k8s.io/kubernetes/pkg/util/sets"
@@ -416,14 +418,16 @@ func startServiceAccountTestServer(t *testing.T) (*clientset.Clientset, restclie
 	stopCh := make(chan struct{})
 	tokenController := serviceaccountcontroller.NewTokensController(rootClientset, serviceaccountcontroller.TokensControllerOptions{TokenGenerator: serviceaccount.JWTTokenGenerator(serviceAccountKey)})
 	go tokenController.Run(1, stopCh)
-	serviceAccountController := serviceaccountcontroller.NewServiceAccountsController(rootClientset, serviceaccountcontroller.DefaultServiceAccountsControllerOptions())
+
-	serviceAccountController.Run()
+	informers := informers.NewSharedInformerFactory(rootClientset, controller.NoResyncPeriodFunc())
+	serviceAccountController := serviceaccountcontroller.NewServiceAccountsController(informers.ServiceAccounts(), informers.Namespaces(), rootClientset, serviceaccountcontroller.DefaultServiceAccountsControllerOptions())
+	informers.Start(stopCh)
+	go serviceAccountController.Run(5, stopCh)
 	// Start the admission plugin reflectors
 	serviceAccountAdmission.Run()
 
 	stop := func() {
 		close(stopCh)
-		serviceAccountController.Stop()
 		serviceAccountAdmission.Stop()
 		apiServer.Close()
 	}