// Runs controller loops and returns on shutdown
162807
func (e *DockercfgTokenDeletedController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	klog.Infof("Starting DockercfgTokenDeletedController controller")
	defer klog.Infof("Shutting down DockercfgTokenDeletedController controller")

	// Wait for the stores to fill
	if !cache.WaitForCacheSync(stopCh, e.secretController.HasSynced) {
		return
	}
	klog.V(1).Infof("caches synced")

	<-stopCh
}
416267
func (nc *Controller) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()

	klog.Infof("Starting node controller")
	defer klog.Infof("Shutting down node controller")

	if !controller.WaitForCacheSync("taint", stopCh, nc.leaseInformerSynced, nc.nodeInformerSynced, nc.podInformerSynced, nc.daemonSetInformerSynced) {
		return
	}

	if nc.runTaintManager {
		go nc.taintManager.Run(stopCh)
	}

	// Close node update queue to cleanup go routine.
	defer nc.nodeUpdateQueue.ShutDown()

	// Start workers to reconcile labels and/or update NoSchedule taint for nodes.
	for i := 0; i < scheduler.UpdateWorkerSize; i++ {
		// Thanks to "workqueue", each worker just need to get item from queue, because
		// the item is flagged when got from queue: if new event come, the new item will
		// be re-queued until "Done", so no more than one worker handle the same item and
		// no event missed.
		go wait.Until(nc.doNodeProcessingPassWorker, time.Second, stopCh)
	}

	if nc.useTaintBasedEvictions {
		// Handling taint based evictions. Because we don't want a dedicated logic in TaintManager for NC-originated
		// taints and we normally don't rate limit evictions caused by taints, we need to rate limit adding taints.
		go wait.Until(nc.doNoExecuteTaintingPass, scheduler.NodeEvictionPeriod, stopCh)
	} else {
		// Managing eviction of nodes:
		// When we delete pods off a node, if the node was not empty at the time we then
		// queue an eviction watcher. If we hit an error, retry deletion.
		go wait.Until(nc.doEvictionPass, scheduler.NodeEvictionPeriod, stopCh)
	}

	// Incorporate the results of node health signal pushed from kubelet to master.
	go wait.Until(func() {
		if err := nc.monitorNodeHealth(); err != nil {
			klog.Errorf("Error monitoring node health: %v", err)
		}
	}, nc.nodeMonitorPeriod, stopCh)

	<-stopCh
}
6305
func (c *DefaultRoleBindingController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

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

	if !controller.WaitForCacheSync("DefaultRoleBindingController", stopCh, c.roleBindingSynced, c.namespaceSynced) {
		return
	}

	for i := 0; i < workers; i++ {
		go wait.Until(c.runWorker, time.Second, stopCh)
	}

	<-stopCh
}
833902
func (c *listControllersCommand) Run(ctx *cmd.Context) error {
	controllers, err := c.store.AllControllers()
	if err != nil {
		return errors.Annotate(err, "failed to list controllers")
	}
	if len(controllers) == 0 && c.out.Name() == "tabular" {
		return errors.Trace(modelcmd.ErrNoControllersDefined)
	}
	if c.refresh && len(controllers) > 0 {
		var wg sync.WaitGroup
		wg.Add(len(controllers))
		for controllerName := range controllers {
			name := controllerName
			go func() {
				defer wg.Done()
				client, err := c.getAPI(name)
				if err != nil {
					fmt.Fprintf(ctx.GetStderr(), "error connecting to api for %q: %v\n", name, err)
					return
				}
				defer client.Close()
				if err := c.refreshControllerDetails(client, name); err != nil {
					fmt.Fprintf(ctx.GetStderr(), "error updating cached details for %q: %v\n", name, err)
				}
			}()
		}
		wg.Wait()
		// Reload controller details
		controllers, err = c.store.AllControllers()
		if err != nil {
			return errors.Annotate(err, "failed to list controllers")
		}
	}
	details, errs := c.convertControllerDetails(controllers)
	if len(errs) > 0 {
		fmt.Fprintln(ctx.Stderr, strings.Join(errs, "\n"))
	}
	currentController, err := modelcmd.DetermineCurrentController(c.store)
	if errors.IsNotFound(err) {
		currentController = ""
	} else if err != nil {
		return errors.Annotate(err, "getting current controller")
	}
	controllerSet := ControllerSet{
		Controllers:       details,
		CurrentController: currentController,
	}
	return c.out.Write(ctx, controllerSet)
}
121885
func (c *destroyCommand) Run(ctx *cmd.Context) error {
	controllerName, err := c.ControllerName()
	if err != nil {
		return errors.Trace(err)
	}
	store := c.ClientStore()
	if !c.assumeYes {
		if err := confirmDestruction(ctx, controllerName); err != nil {
			return err
		}
	}

	// Attempt to connect to the API.  If we can't, fail the destroy.  Users will
	// need to use the controller kill command if we can't connect.
	api, err := c.getControllerAPI()
	if err != nil {
		return c.ensureUserFriendlyErrorLog(errors.Annotate(err, "cannot connect to API"), ctx, nil)
	}
	defer api.Close()

	if api.BestAPIVersion() < 4 {
		// Versions before 4 support only destroying the storage,
		// and will not raise an error if there is storage in the
		// controller. Force the user to specify up-front.
		if c.releaseStorage {
			return errors.New("this juju controller only supports destroying storage")
		}
		if !c.destroyStorage {
			models, err := api.AllModels()
			if err != nil {
				return errors.Trace(err)
			}
			var anyStorage bool
			for _, model := range models {
				hasStorage, err := c.modelHasStorage(model.Name)
				if err != nil {
					return errors.Trace(err)
				}
				if hasStorage {
					anyStorage = true
					break
				}
			}
			if anyStorage {
				return errors.Errorf(`cannot destroy controller %q

Destroying this controller will destroy the storage,
but you have not indicated that you want to do that.

Please run the the command again with --destroy-storage
to confirm that you want to destroy the storage along
with the controller.

If instead you want to keep the storage, you must first
upgrade the controller to version 2.3 or greater.

`, controllerName)
			}
			c.destroyStorage = true
		}
	}

	// Obtain controller environ so we can clean up afterwards.
	controllerEnviron, err := c.getControllerEnviron(ctx, store, controllerName, api)
	if err != nil {
		return errors.Annotate(err, "getting controller environ")
	}

	cloudCallCtx := cloudCallContext(c.controllerCredentialAPIFunc)

	for {
		// Attempt to destroy the controller.
		ctx.Infof("Destroying controller")
		var hasHostedModels bool
		var hasPersistentStorage bool
		var destroyStorage *bool
		if c.destroyStorage || c.releaseStorage {
			// Set destroyStorage to true or false, if
			// --destroy-storage or --release-storage
			// is specified, respectively.
			destroyStorage = &c.destroyStorage
		}
		err = api.DestroyController(controllerapi.DestroyControllerParams{
			DestroyModels:  c.destroyModels,
			DestroyStorage: destroyStorage,
		})
		if err != nil {
			if params.IsCodeHasHostedModels(err) {
				hasHostedModels = true
			} else if params.IsCodeHasPersistentStorage(err) {
				hasPersistentStorage = true
			} else {
				return c.ensureUserFriendlyErrorLog(
					errors.Annotate(err, "cannot destroy controller"),
					ctx, api,
				)
			}
		}

		updateStatus := newTimedStatusUpdater(ctx, api, controllerEnviron.Config().UUID(), clock.WallClock)
		// wait for 2 seconds to let empty hosted models changed from alive to dying.
		modelStatus := updateStatus(0)
		if !c.destroyModels {
			if err := c.checkNoAliveHostedModels(ctx, modelStatus.models); err != nil {
				return errors.Trace(err)
			}
			if hasHostedModels && !hasUnDeadModels(modelStatus.models) {
				// When we called DestroyController before, we were
				// informed that there were hosted models remaining.
				// When we checked just now, there were none. We should
				// try destroying again.
				continue
			}
		}
		if !c.destroyStorage && !c.releaseStorage && hasPersistentStorage {
			if err := c.checkNoPersistentStorage(ctx, modelStatus); err != nil {
				return errors.Trace(err)
			}
			// When we called DestroyController before, we were
			// informed that there was persistent storage remaining.
			// When we checked just now, there was none. We should
			// try destroying again.
			continue
		}

		// Even if we've not just requested for hosted models to be destroyed,
		// there may be some being destroyed already. We need to wait for them.
		// Check for both undead models and live machines, as machines may be
		// in the controller model.
		ctx.Infof("Waiting for hosted model resources to be reclaimed")
		for ; hasUnreclaimedResources(modelStatus); modelStatus = updateStatus(2 * time.Second) {
			ctx.Infof(fmtCtrStatus(modelStatus.controller))
			for _, model := range modelStatus.models {
				ctx.Verbosef(fmtModelStatus(model))
			}
		}
		ctx.Infof("All hosted models reclaimed, cleaning up controller machines")
		return c.environsDestroy(controllerName, controllerEnviron, cloudCallCtx, store)
	}
}
1335730
func (ctrl *PersistentVolumeController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer ctrl.claimQueue.ShutDown()
	defer ctrl.volumeQueue.ShutDown()

	klog.Infof("Starting persistent volume controller")
	defer klog.Infof("Shutting down persistent volume controller")

	if !controller.WaitForCacheSync("persistent volume", stopCh, ctrl.volumeListerSynced, ctrl.claimListerSynced, ctrl.classListerSynced, ctrl.podListerSynced, ctrl.NodeListerSynced) {
		return
	}

	ctrl.initializeCaches(ctrl.volumeLister, ctrl.claimLister)

	go wait.Until(ctrl.resync, ctrl.resyncPeriod, stopCh)
	go wait.Until(ctrl.volumeWorker, time.Second, stopCh)
	go wait.Until(ctrl.claimWorker, time.Second, stopCh)

	metrics.Register(ctrl.volumes.store, ctrl.claims)

	<-stopCh
}
964777
func (c *TemplateInstanceController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	if !cache.WaitForCacheSync(stopCh, c.informer.HasSynced) {
		return
	}

	klog.V(2).Infof("Starting TemplateInstance controller")

	for i := 0; i < workers; i++ {
		go wait.Until(c.runWorker, time.Second, stopCh)
	}

	<-stopCh
}
84635
func (a *HorizontalController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer a.queue.ShutDown()

	klog.Infof("Starting HPA controller")
	defer klog.Infof("Shutting down HPA controller")

	if !controller.WaitForCacheSync("HPA", stopCh, a.hpaListerSynced, a.podListerSynced) {
		return
	}

	// start a single worker (we may wish to start more in the future)
	go wait.Until(a.worker, time.Second, stopCh)

	<-stopCh
}
272024
func (bc *BuildController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer bc.buildQueue.ShutDown()
	defer bc.buildConfigQueue.ShutDown()
	defer bc.controllerConfigQueue.ShutDown()

	// Wait for the controller stores to sync before starting any work in this controller.
	if !cache.WaitForCacheSync(stopCh,
		bc.buildStoreSynced,
		bc.podStoreSynced,
		bc.secretStoreSynced,
		bc.imageStreamStoreSynced,
		bc.configMapStoreSynced) {
		utilruntime.HandleError(fmt.Errorf("timed out waiting for caches to sync"))
		return
	}

	// Integration tests currently do not support cache sync for operator-installed custom resource definitions
	if os.Getenv("OS_INTEGRATION_TEST") != "true" {
		if !cache.WaitForCacheSync(stopCh,
			bc.buildControllerConfigStoreSynced,
			bc.imageConfigStoreSynced) {
			utilruntime.HandleError(fmt.Errorf("timed out waiting for caches to sync"))
			return
		}
	}

	// Sync build controller config on start up.
	if errs := bc.handleControllerConfig(); len(errs) > 0 {
		utilruntime.HandleError(fmt.Errorf("errors syncing build controller config: %v", errs))
	}

	klog.Infof("Starting build controller")

	// Only need one worker to manage config
	go wait.Until(bc.controllerConfigWorker, time.Second, stopCh)

	for i := 0; i < workers; i++ {
		go wait.Until(bc.buildWorker, time.Second, stopCh)
	}

	for i := 0; i < workers; i++ {
		go wait.Until(bc.buildConfigWorker, time.Second, stopCh)
	}

	metrics.IntializeMetricsCollector(bc.buildLister)

	<-stopCh
	klog.Infof("Shutting down build controller")
}
907592
func (cnc *CloudNodeController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()

	// The following loops run communicate with the APIServer with a worst case complexity
	// of O(num_nodes) per cycle. These functions are justified here because these events fire
	// very infrequently. DO NOT MODIFY this to perform frequent operations.

	// Start a loop to periodically update the node addresses obtained from the cloud
	wait.Until(cnc.UpdateNodeStatus, cnc.nodeStatusUpdateFrequency, stopCh)
}
1107838
func (e *BackgroundTask) Run(m libkb.MetaContext) (err error) {
	defer m.Trace(e.Name(), func() error { return err })()

	// use a new background context with a saved cancel function
	var cancel func()
	m, cancel = m.BackgroundWithCancel()

	e.Lock()
	defer e.Unlock()

	e.shutdownFunc = cancel
	if e.shutdown {
		// Shutdown before started
		cancel()
		e.meta("early-shutdown")
		return nil
	}

	// start the loop and return
	go func() {
		err := e.loop(m)
		if err != nil {
			e.log(m, "loop error: %s", err)
		}
		cancel()
		e.meta("loop-exit")
	}()

	return nil
}
322450
func (c *Controller) Run(staticSpec *spec.Swagger, openAPIService *handler.OpenAPIService, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()
	defer klog.Infof("Shutting down OpenAPI controller")

	klog.Infof("Starting OpenAPI controller")

	c.staticSpec = staticSpec
	c.openAPIService = openAPIService

	if !cache.WaitForCacheSync(stopCh, c.crdsSynced) {
		utilruntime.HandleError(fmt.Errorf("timed out waiting for caches to sync"))
		return
	}

	// create initial spec to avoid merging once per CRD on startup
	crds, err := c.crdLister.List(labels.Everything())
	if err != nil {
		utilruntime.HandleError(fmt.Errorf("failed to initially list all CRDs: %v", err))
		return
	}
	for _, crd := range crds {
		if !apiextensions.IsCRDConditionTrue(crd, apiextensions.Established) {
			continue
		}
		newSpecs, changed, err := buildVersionSpecs(crd, nil)
		if err != nil {
			utilruntime.HandleError(fmt.Errorf("failed to build OpenAPI spec of CRD %s: %v", crd.Name, err))
		} else if !changed {
			continue
		}
		c.crdSpecs[crd.Name] = newSpecs
	}
	if err := c.updateSpecLocked(); err != nil {
		utilruntime.HandleError(fmt.Errorf("failed to initially create OpenAPI spec for CRDs: %v", err))
		return
	}

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

	<-stopCh
}
217941
func (c *restoreCommand) Run(ctx *cmd.Context) error {
	if err := c.validateIaasController(c.Info().Name); err != nil {
		return errors.Trace(err)
	}
	if c.Log != nil {
		if err := c.Log.Start(ctx); err != nil {
			return err
		}
	}

	// Don't allow restore in an HA environment
	controllerModelUUID, modelStatus, err := c.modelStatus()
	if err != nil {
		return errors.Trace(err)
	}
	activeCount, _ := controller.ControllerMachineCounts(controllerModelUUID, modelStatus)
	if activeCount > 1 {
		return errors.Errorf("unable to restore backup in HA configuration.  For help see https://docs.jujucharms.com/stable/controllers-backup")
	}

	var archive ArchiveReader
	var meta *params.BackupsMetadataResult
	target := c.BackupId
	if c.Filename != "" {
		// Read archive specified by the Filename
		target = c.Filename
		var err error
		archive, meta, err = getArchive(c.Filename)
		if err != nil {
			return errors.Trace(err)
		}
		defer archive.Close()
	}

	client, err := c.NewAPIClient()
	if err != nil {
		return errors.Trace(err)
	}
	defer client.Close()

	// We have a backup client, now use the relevant method
	// to restore the backup.
	if c.Filename != "" {
		err = client.RestoreReader(archive, meta, c.newClient)
	} else {
		err = client.Restore(c.BackupId, c.newClient)
	}
	if err != nil {
		return errors.Trace(err)
	}
	fmt.Fprintf(ctx.Stdout, "restore from %q completed\n", target)
	return nil
}
699456
func (c *Controller) Shutdown(ctx context.Context) error {
	return c.c.Shutdown(ctx)
}
136275
func (e *EndpointController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer e.queue.ShutDown()

	klog.Infof("Starting endpoint controller")
	defer klog.Infof("Shutting down endpoint controller")

	if !controller.WaitForCacheSync("endpoint", stopCh, e.podsSynced, e.servicesSynced, e.endpointsSynced) {
		return
	}

	for i := 0; i < workers; i++ {
		go wait.Until(e.worker, e.workerLoopPeriod, stopCh)
	}

	go func() {
		defer utilruntime.HandleCrash()
		e.checkLeftoverEndpoints()
	}()

	<-stopCh
}
1194922
func (c *Controller) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	klog.Infof("Starting PV protection controller")
	defer klog.Infof("Shutting down PV protection controller")

	if !controller.WaitForCacheSync("PV protection", stopCh, c.pvListerSynced) {
		return
	}

	for i := 0; i < workers; i++ {
		go wait.Until(c.runWorker, time.Second, stopCh)
	}

	<-stopCh
}
829803
func (c *showCommand) Run(ctx *cmd.Context) (err error) {
	controllerName, err := c.ControllerName()
	if err != nil {
		return err
	}
	url, err := crossmodel.ParseOfferURL(c.url)
	if err != nil {
		store := c.ClientStore()
		currentModel, err := store.CurrentModel(controllerName)
		if err != nil {
			return errors.Trace(err)
		}
		url, err = makeURLFromCurrentModel(c.url, controllerName, currentModel)
		if err != nil {
			return errors.Trace(err)
		}
	}
	if url.Source != "" {
		controllerName = url.Source
	}
	accountDetails, err := c.CurrentAccountDetails()
	if err != nil {
		return err
	}
	loggedInUser := accountDetails.User

	api, err := c.newAPIFunc(controllerName)
	if err != nil {
		return err
	}
	defer api.Close()

	url.Source = ""
	found, err := api.ApplicationOffer(url.String())
	if err != nil {
		return err
	}

	output, err := convertOffers(controllerName, names.NewUserTag(loggedInUser), found)
	if err != nil {
		return err
	}
	return c.out.Write(ctx, output)
}