Merge pull request #17855 from deads2k/controller-16-wait

wait for kube controllers to be ready before starting informers

Author: deads2k

pkg/cmd/server/origin/controller/interfaces.go

@@ -43,6 +43,10 @@ type ControllerContext struct {
 
 	// Stop is the stop channel
 	Stop <-chan struct{}
+
+	// InformersStarted is closed after all of the controllers have been initialized and are running.  After this point it is safe,
+	// for an individual controller to start the shared informers. Before it is closed, they should not.
+	InformersStarted chan struct{}
 }
 
 // OpenshiftControllerOptions contain the options used to run the controllers.  Eventually we need to construct a way to properly

pkg/cmd/server/start/controllers.go

@@ -28,6 +28,7 @@ func newControllerContext(
 	kubeExternal kclientsetexternal.Interface,
 	informers *informers,
 	stopCh <-chan struct{},
+	informersStarted chan struct{},
 ) origincontrollers.ControllerContext {
 
 	// divide up the QPS since it re-used separately for every client
@@ -60,6 +61,7 @@ func newControllerContext(
 		SecurityInformers:      informers.securityInformers,
 		TemplateInformers:      informers.templateInformers,
 		Stop:                   stopCh,
+		InformersStarted:       informersStarted,
 	}
 
 	return openshiftControllerContext

pkg/cmd/server/start/start_kube_controller_manager.go

@@ -27,7 +27,7 @@ import (
 )
 
 // newKubeControllerContext provides a function which overrides the default and plugs a different set of informers in
-func newKubeControllerContext(informers *informers) func(s *controlleroptions.CMServer, rootClientBuilder, clientBuilder controller.ControllerClientBuilder, stop <-chan struct{}) (controllerapp.ControllerContext, error) {
+func newKubeControllerContext(informers *informers, informersStarted chan struct{}) func(s *controlleroptions.CMServer, rootClientBuilder, clientBuilder controller.ControllerClientBuilder, stop <-chan struct{}) (controllerapp.ControllerContext, error) {
 	oldContextFunc := controllerapp.CreateControllerContext
 	return func(s *controlleroptions.CMServer, rootClientBuilder, clientBuilder controller.ControllerClientBuilder, stop <-chan struct{}) (controllerapp.ControllerContext, error) {
 		ret, err := oldContextFunc(s, rootClientBuilder, clientBuilder, stop)
@@ -38,6 +38,7 @@ func newKubeControllerContext(informers *informers) func(s *controlleroptions.CM
 		// Overwrite the informers.  Since nothing accessed the existing informers that we're overwriting, they are inert.
 		// TODO Remove this.  It keeps in-process memory utilization down, but we shouldn't do it.
 		ret.InformerFactory = newGenericInformers(informers)
+		ret.InformersStarted = informersStarted
 
 		return ret, nil
 	}
@@ -219,10 +220,11 @@ func createRecylerTemplate(recyclerImage string) (string, error) {
 }
 
 func runEmbeddedKubeControllerManager(kubeconfigFile, saPrivateKeyFile, saRootCAFile, podEvictionTimeout string, dynamicProvisioningEnabled bool, cmdLineArgs map[string][]string,
-	recyclerImage string, informers *informers) {
-	controllerapp.CreateControllerContext = newKubeControllerContext(informers)
+	recyclerImage string, informers *informers, informersStarted chan struct{}, kubeControllersStarted chan struct{}) {
+	controllerapp.CreateControllerContext = newKubeControllerContext(informers, informersStarted)
 	controllerapp.StartInformers = func(stop <-chan struct{}) {
-		informers.Start(stop)
+		// We'll start everything out of band, BUT we don't want to start until kube has called this
+		close(kubeControllersStarted)
 	}
 
 	// TODO we need a real identity for this.  Right now it's just using the loopback connection like it used to.

pkg/cmd/server/start/start_master.go

@@ -475,6 +475,9 @@ func (m *Master) Start() error {
 			// continuously run the scheduler while we have the primary lease
 			go runEmbeddedScheduler(m.config.MasterClients.OpenShiftLoopbackKubeConfig, m.config.KubernetesMasterConfig.SchedulerConfigFile, m.config.KubernetesMasterConfig.SchedulerArguments)
 
+			informersStarted := make(chan struct{})
+			kubeControllersStarted := make(chan struct{})
+
 			go runEmbeddedKubeControllerManager(
 				m.config.MasterClients.OpenShiftLoopbackKubeConfig,
 				m.config.ServiceAccountConfig.PrivateKeyFile,
@@ -483,18 +486,23 @@ func (m *Master) Start() error {
 				m.config.VolumeConfig.DynamicProvisioningEnabled,
 				m.config.KubernetesMasterConfig.ControllerArguments,
 				recyclerImage,
-				informers)
+				informers,
+				informersStarted,
+				kubeControllersStarted)
 
 			openshiftControllerOptions, err := getOpenshiftControllerOptions(m.config.KubernetesMasterConfig.ControllerArguments)
 			if err != nil {
 				glog.Fatal(err)
 			}
-			controllerContext := newControllerContext(openshiftControllerOptions, privilegedLoopbackConfig, kubeExternal, informers, utilwait.NeverStop)
+			controllerContext := newControllerContext(openshiftControllerOptions, privilegedLoopbackConfig, kubeExternal, informers, utilwait.NeverStop, informersStarted)
 			if err := startControllers(*m.config, allocationController, controllerContext); err != nil {
 				glog.Fatal(err)
 			}
 
+			// don't start any controllers until kube is ready
+			<-kubeControllersStarted
 			informers.Start(utilwait.NeverStop)
+			close(informersStarted)
 		}()
 	}