// startControllers starts the cloud specific controller loops.
965440
func startControllers(c *cloudcontrollerconfig.CompletedConfig, stopCh <-chan struct{}, cloud cloudprovider.Interface, controllers map[string]initFunc) error {
	// Initialize the cloud provider with a reference to the clientBuilder
	cloud.Initialize(c.ClientBuilder, stopCh)
	// Set the informer on the user cloud object
	if informerUserCloud, ok := cloud.(cloudprovider.InformerUser); ok {
		informerUserCloud.SetInformers(c.SharedInformers)
	}

	for controllerName, initFn := range controllers {
		if !genericcontrollermanager.IsControllerEnabled(controllerName, ControllersDisabledByDefault, c.ComponentConfig.Generic.Controllers) {
			klog.Warningf("%q is disabled", controllerName)
			continue
		}

		klog.V(1).Infof("Starting %q", controllerName)
		_, started, err := initFn(c, cloud, stopCh)
		if err != nil {
			klog.Errorf("Error starting %q", controllerName)
			return err
		}
		if !started {
			klog.Warningf("Skipping %q", controllerName)
			continue
		}
		klog.Infof("Started %q", controllerName)

		time.Sleep(wait.Jitter(c.ComponentConfig.Generic.ControllerStartInterval.Duration, ControllerStartJitter))
	}

	// If apiserver is not running we should wait for some time and fail only then. This is particularly
	// important when we start apiserver and controller manager at the same time.
	if err := genericcontrollermanager.WaitForAPIServer(c.VersionedClient, 10*time.Second); err != nil {
		klog.Fatalf("Failed to wait for apiserver being healthy: %v", err)
	}

	c.SharedInformers.Start(stopCh)

	select {}
}
1034554
func NewController(bk *buckets.Bucket) *Controller {
	// map of days to integers
	daynum := map[string]int{
		"mon": 1, // monday is the first day of the week
		"tue": 2,
		"wed": 3,
		"thu": 4,
		"fri": 5,
		"sat": 6,
		"sun": 7,
	}
	// map of scanners for iterating over keys subsets of keys
	scan := map[string]buckets.Scanner{
		"mon": bk.NewPrefixScanner([]byte("1")),
		"tue": bk.NewPrefixScanner([]byte("2")),
		"wed": bk.NewPrefixScanner([]byte("3")),
		"thu": bk.NewPrefixScanner([]byte("4")),
		"fri": bk.NewPrefixScanner([]byte("5")),
		"sat": bk.NewPrefixScanner([]byte("6")),
		"sun": bk.NewPrefixScanner([]byte("7")),
		// weekdays are mon to fri: 1 <= key < 6.
		"weekday": bk.NewRangeScanner([]byte("1"), []byte("6")),
		// weekends are sat to sun: 6 <= key < 8.
		"weekend": bk.NewRangeScanner([]byte("6"), []byte("8")),
	}
	return &Controller{bk, daynum, scan}
}
692658
func CreateControllerContext(s *config.CompletedConfig, rootClientBuilder, clientBuilder controller.ControllerClientBuilder, stop <-chan struct{}) (ControllerContext, error) {
	versionedClient := rootClientBuilder.ClientOrDie("shared-informers")
	sharedInformers := informers.NewSharedInformerFactory(versionedClient, ResyncPeriod(s)())

	dynamicClient := dynamic.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("dynamic-informers"))
	dynamicInformers := dynamicinformer.NewDynamicSharedInformerFactory(dynamicClient, ResyncPeriod(s)())

	// If apiserver is not running we should wait for some time and fail only then. This is particularly
	// important when we start apiserver and controller manager at the same time.
	if err := genericcontrollermanager.WaitForAPIServer(versionedClient, 10*time.Second); err != nil {
		return ControllerContext{}, fmt.Errorf("failed to wait for apiserver being healthy: %v", err)
	}

	// Use a discovery client capable of being refreshed.
	discoveryClient := rootClientBuilder.ClientOrDie("controller-discovery")
	cachedClient := cacheddiscovery.NewMemCacheClient(discoveryClient.Discovery())
	restMapper := restmapper.NewDeferredDiscoveryRESTMapper(cachedClient)
	go wait.Until(func() {
		restMapper.Reset()
	}, 30*time.Second, stop)

	availableResources, err := GetAvailableResources(rootClientBuilder)
	if err != nil {
		return ControllerContext{}, err
	}

	cloud, loopMode, err := createCloudProvider(s.ComponentConfig.KubeCloudShared.CloudProvider.Name, s.ComponentConfig.KubeCloudShared.ExternalCloudVolumePlugin,
		s.ComponentConfig.KubeCloudShared.CloudProvider.CloudConfigFile, s.ComponentConfig.KubeCloudShared.AllowUntaggedCloud, sharedInformers)
	if err != nil {
		return ControllerContext{}, err
	}

	ctx := ControllerContext{
		ClientBuilder:          clientBuilder,
		InformerFactory:        sharedInformers,
		GenericInformerFactory: controller.NewInformerFactory(sharedInformers, dynamicInformers),
		ComponentConfig:        s.ComponentConfig,
		RESTMapper:             restMapper,
		AvailableResources:     availableResources,
		Cloud:                  cloud,
		LoopMode:               loopMode,
		Stop:                   stop,
		InformersStarted:       make(chan struct{}),
		ResyncPeriod:           ResyncPeriod(s),
	}
	return ctx, nil
}
750965
func NewController(kubeClient clientset.Interface) (*Controller, error) {
	eventBroadcaster := record.NewBroadcaster()
	eventBroadcaster.StartLogging(klog.Infof)
	eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: kubeClient.CoreV1().Events("")})

	if kubeClient != nil && kubeClient.CoreV1().RESTClient().GetRateLimiter() != nil {
		if err := metrics.RegisterMetricAndTrackRateLimiterUsage("cronjob_controller", kubeClient.CoreV1().RESTClient().GetRateLimiter()); err != nil {
			return nil, err
		}
	}

	jm := &Controller{
		kubeClient: kubeClient,
		jobControl: realJobControl{KubeClient: kubeClient},
		sjControl:  &realSJControl{KubeClient: kubeClient},
		podControl: &realPodControl{KubeClient: kubeClient},
		recorder:   eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "cronjob-controller"}),
	}

	return jm, nil
}
494885
func (wfc *WorkflowController) RunTTLController(ctx context.Context) {
	ttlCtrl := ttlcontroller.NewController(
		wfc.restConfig,
		wfc.wfclientset,
		wfc.Config.Namespace,
		wfc.Config.InstanceID,
	)
	err := ttlCtrl.Run(ctx.Done())
	if err != nil {
		panic(err)
	}
}
156785
func (ec *EstablishingController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer ec.queue.ShutDown()

	klog.Infof("Starting EstablishingController")
	defer klog.Infof("Shutting down EstablishingController")

	if !cache.WaitForCacheSync(stopCh, ec.crdSynced) {
		return
	}

	// only start one worker thread since its a slow moving API
	go wait.Until(ec.runWorker, time.Second, stopCh)

	<-stopCh
}
654761
func (s *Factory) ReplicationControllers() corev1.ReplicationControllerInterface {
	return s.coreClientSet.ReplicationControllers(s.namespace)
}
1139856
func (c *Controller) resetAppControllerFields() {
	appController := reflect.ValueOf(c.AppController).Elem()
	// Zero out controller
	for _, index := range c.Type.ControllerIndexes {
		appController.FieldByIndex(index).Set(reflect.Zero(reflect.TypeOf(c.AppController).Elem().FieldByIndex(index).Type))
	}
}
438613
func (c *Controller) SetController(context ServerContext) {

	c.Request.SetRequest(context.GetRequest())
	c.Response.SetResponse(context.GetResponse())
	c.Request.controller = c
	c.Params = new(Params)
	c.Args = map[string]interface{}{}
	c.ViewArgs = map[string]interface{}{
		"RunMode": RunMode,
		"DevMode": DevMode,
	}

}
19657
func NewController(backend Backend, t *api.Task) (*Controller, error) {
	spec, err := readSpec(t)
	if err != nil {
		return nil, err
	}
	return &Controller{
		backend:   backend,
		spec:      spec,
		serviceID: t.ServiceID,
		logger: logrus.WithFields(logrus.Fields{
			"controller": "plugin",
			"task":       t.ID,
			"plugin":     spec.Name,
		})}, nil
}
1153856
func initControllerStack() {
	RevelConfig.Controller.Reuse = Config.BoolDefault("revel.controller.reuse",true)

	if RevelConfig.Controller.Reuse {
		RevelConfig.Controller.Stack = utils.NewStackLock(
			Config.IntDefault("revel.controller.stack", 10),
			Config.IntDefault("revel.controller.maxstack", 200), func() interface{} {
				return NewControllerEmpty()
			})
		RevelConfig.Controller.CachedStackSize = Config.IntDefault("revel.cache.controller.stack", 10)
		RevelConfig.Controller.CachedStackMaxSize = Config.IntDefault("revel.cache.controller.maxstack", 100)
		RevelConfig.Controller.CachedMap = map[string]*utils.SimpleLockStack{}
	}
}
154696
func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc {
	controllers := map[string]InitFunc{}
	controllers["endpoint"] = startEndpointController
	controllers["replicationcontroller"] = startReplicationController
	controllers["podgc"] = startPodGCController
	controllers["resourcequota"] = startResourceQuotaController
	controllers["namespace"] = startNamespaceController
	controllers["serviceaccount"] = startServiceAccountController
	controllers["garbagecollector"] = startGarbageCollectorController
	controllers["daemonset"] = startDaemonSetController
	controllers["job"] = startJobController
	controllers["deployment"] = startDeploymentController
	controllers["replicaset"] = startReplicaSetController
	controllers["horizontalpodautoscaling"] = startHPAController
	controllers["disruption"] = startDisruptionController
	controllers["statefulset"] = startStatefulSetController
	controllers["cronjob"] = startCronJobController
	controllers["csrsigning"] = startCSRSigningController
	controllers["csrapproving"] = startCSRApprovingController
	controllers["csrcleaner"] = startCSRCleanerController
	controllers["ttl"] = startTTLController
	controllers["bootstrapsigner"] = startBootstrapSignerController
	controllers["tokencleaner"] = startTokenCleanerController
	controllers["nodeipam"] = startNodeIpamController
	controllers["nodelifecycle"] = startNodeLifecycleController
	if loopMode == IncludeCloudLoops {
		controllers["service"] = startServiceController
		controllers["route"] = startRouteController
		controllers["cloud-node-lifecycle"] = startCloudNodeLifecycleController
		// TODO: volume controller into the IncludeCloudLoops only set.
	}
	controllers["persistentvolume-binder"] = startPersistentVolumeBinderController
	controllers["attachdetach"] = startAttachDetachController
	controllers["persistentvolume-expander"] = startVolumeExpandController
	controllers["clusterrole-aggregation"] = startClusterRoleAggregrationController
	controllers["pvc-protection"] = startPVCProtectionController
	controllers["pv-protection"] = startPVProtectionController
	controllers["ttl-after-finished"] = startTTLAfterFinishedController
	controllers["root-ca-cert-publisher"] = startRootCACertPublisher

	return controllers
}
751726
func NewController(config ControllerConfig) (*Controller, error) {
	c, err := newController(config, newResidentManager(config.Changes))
	return c, errors.Trace(err)
}
619684
func (c *Controller) setAppControllerFields() {
	appController := reflect.ValueOf(c.AppController).Elem()
	cValue := reflect.ValueOf(c)
	for _, index := range c.Type.ControllerIndexes {
		appController.FieldByIndex(index).Set(cValue)
	}
}
222759
func (c *Controller) Start() {
	if c.runner != nil {
		return
	}

	// Reconcile during first run removing itself until server is ready.
	endpointPorts := createEndpointPortSpec(c.PublicServicePort, "https", c.ExtraEndpointPorts)
	if err := c.EndpointReconciler.RemoveEndpoints(kubernetesServiceName, c.PublicIP, endpointPorts); err != nil {
		klog.Errorf("Unable to remove old endpoints from kubernetes service: %v", err)
	}

	repairClusterIPs := servicecontroller.NewRepair(c.ServiceClusterIPInterval, c.ServiceClient, c.EventClient, &c.ServiceClusterIPRange, c.ServiceClusterIPRegistry)
	repairNodePorts := portallocatorcontroller.NewRepair(c.ServiceNodePortInterval, c.ServiceClient, c.EventClient, c.ServiceNodePortRange, c.ServiceNodePortRegistry)

	// run all of the controllers once prior to returning from Start.
	if err := repairClusterIPs.RunOnce(); err != nil {
		// If we fail to repair cluster IPs apiserver is useless. We should restart and retry.
		klog.Fatalf("Unable to perform initial IP allocation check: %v", err)
	}
	if err := repairNodePorts.RunOnce(); err != nil {
		// If we fail to repair node ports apiserver is useless. We should restart and retry.
		klog.Fatalf("Unable to perform initial service nodePort check: %v", err)
	}

	c.runner = async.NewRunner(c.RunKubernetesNamespaces, c.RunKubernetesService, repairClusterIPs.RunUntil, repairNodePorts.RunUntil)
	c.runner.Start()
}
570731
func (config ManifoldConfig) start(context dependency.Context) (_ worker.Worker, err error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}

	var hub *pubsub.StructuredHub
	if err := context.Get(config.HubName, &hub); err != nil {
		return nil, errors.Trace(err)
	}

	var getCertificate func() *tls.Certificate
	if err := context.Get(config.CertWatcherName, &getCertificate); err != nil {
		return nil, errors.Trace(err)
	}

	var mux *apiserverhttp.Mux
	if err := context.Get(config.MuxName, &mux); err != nil {
		return nil, errors.Trace(err)
	}

	// We don't actually need anything from these workers, but we
	// shouldn't start until they're available.
	if err := context.Get(config.APIServerName, nil); err != nil {
		return nil, errors.Trace(err)
	}
	if err := context.Get(config.RaftTransportName, nil); err != nil {
		return nil, errors.Trace(err)
	}

	var stTracker workerstate.StateTracker
	if err := context.Get(config.StateName, &stTracker); err != nil {
		return nil, errors.Trace(err)
	}
	statePool, err := stTracker.Use()
	if err != nil {
		return nil, errors.Trace(err)
	}
	defer func() {
		if err != nil {
			stTracker.Done()
		}
	}()

	systemState := statePool.SystemState()
	tlsConfig, err := config.NewTLSConfig(systemState, getCertificate)
	if err != nil {
		return nil, errors.Trace(err)
	}
	controllerConfig, err := config.GetControllerConfig(systemState)
	if err != nil {
		return nil, errors.Annotate(err, "unable to get controller config")
	}

	w, err := config.NewWorker(Config{
		AgentName:            config.AgentName,
		Clock:                config.Clock,
		PrometheusRegisterer: config.PrometheusRegisterer,
		Hub:                  hub,
		TLSConfig:            tlsConfig,
		Mux:                  mux,
		MuxShutdownWait:      config.MuxShutdownWait,
		LogDir:               config.LogDir,
		APIPort:              controllerConfig.APIPort(),
		APIPortOpenDelay:     controllerConfig.APIPortOpenDelay(),
		ControllerAPIPort:    controllerConfig.ControllerAPIPort(),
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	return common.NewCleanupWorker(w, func() { stTracker.Done() }), nil
}
1176718
func NewCloudController(ip, username, password, deploymentName string, cloudControllers CloudControllerJobs) (*CloudController, error) {
	return &CloudController{
		deploymentName:   deploymentName,
		cloudControllers: cloudControllers,
		ip:               ip,
		username:         username,
		password:         password,
	}, nil
}