// Run starts the workers for this controller.
542341
func (c *NamespaceSCCAllocationController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	defer klog.V(1).Infof("Shutting down")

	// Wait for the stores to fill
	if !controller.WaitForCacheSync(controllerName, stopCh, c.nsListerSynced) {
		return
	}

	klog.V(1).Infof("Repairing SCC UID Allocations")
	if err := c.WaitForRepair(stopCh); err != nil {
		// this is consistent with previous behavior
		klog.Fatal(err)
	}
	klog.V(1).Infof("Repair complete")

	go c.worker()
	<-stopCh
}
127484
func (c *AddMachineCmd) Run(app subcommands.Application, args []string, env subcommands.Env) int {
	ctx := cli.GetContext(app, c, env)
	// TODO(smut): Validate required fields client-side.
	req := &crimson.CreateMachineRequest{
		Machine: &c.machine,
	}
	client := getClient(ctx)
	resp, err := client.CreateMachine(ctx, req)
	if err != nil {
		errors.Log(ctx, err)
		return 1
	}
	printMachines(c.f.tsv, resp)
	return 0
}
631665
func (ssc *StatefulSetController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer ssc.queue.ShutDown()

	klog.Infof("Starting stateful set controller")
	defer klog.Infof("Shutting down statefulset controller")

	if !controller.WaitForCacheSync("stateful set", stopCh, ssc.podListerSynced, ssc.setListerSynced, ssc.pvcListerSynced, ssc.revListerSynced) {
		return
	}

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

	<-stopCh
}
1295089
func (c *enableHACommand) Run(ctx *cmd.Context) error {
	controllerName, err := c.ControllerName()
	if err != nil {
		return errors.Trace(err)
	}
	if err := common.ValidateIaasController(c.CommandBase, c.Info().Name, controllerName, c.ClientStore()); err != nil {
		return errors.Trace(err)
	}

	c.Constraints, err = common.ParseConstraints(ctx, c.ConstraintsStr)
	if err != nil {
		return err
	}
	haClient, err := c.newHAClientFunc()
	if err != nil {
		return err
	}

	defer haClient.Close()
	enableHAResult, err := haClient.EnableHA(
		c.NumControllers,
		c.Constraints,
		c.Placement,
	)
	if err != nil {
		return block.ProcessBlockedError(err, block.BlockChange)
	}

	result := availabilityInfo{
		Added:      machineTagsToIds(enableHAResult.Added...),
		Removed:    machineTagsToIds(enableHAResult.Removed...),
		Maintained: machineTagsToIds(enableHAResult.Maintained...),
		Promoted:   machineTagsToIds(enableHAResult.Promoted...),
		Demoted:    machineTagsToIds(enableHAResult.Demoted...),
		Converted:  machineTagsToIds(enableHAResult.Converted...),
	}
	return c.out.Write(ctx, result)
}
998305
func (loop *Loop) Run(path string) {
	if path == "" {
		path = "./"
	}
	// lifecycle
	loop.lifecycle = jobsync.NewLifecycle(workers.DefaultTimeout, true)
	if loop.scope != nil {
		loop.scope.On(app.ErrorEvent, loop.KillSlot)
		loop.scope.On(app.KillEvent, loop.KillSlot)
	}
	// producer
	producentMaxJob := loop.loopData.Producents
	if producentMaxJob == 0 || producentMaxJob > workers.MaxJob {
		producentMaxJob = workers.MaxJob
	}
	producerPool := jobsync.NewPool(producentMaxJob)
	producer := &Producer{
		lifecycle: loop.lifecycle,
		pool:      producerPool,
		loopData:  loop.loopData,
		path:      path,
	}
	producerCounter := producerPool.Add(1)
	if producerCounter != 1 {
		panic("filesystem.Loop it is not possible to start producer job")
	}
	go producer.Loop()
	// consumer
	consumerMaxJob := loop.loopData.Consumers
	if consumerMaxJob == 0 || consumerMaxJob > workers.MaxJob {
		consumerMaxJob = workers.MaxJob
	}
	loop.consumerPool = jobsync.NewPool(consumerMaxJob)
	consumer := &Consumer{
		lifecycle: loop.lifecycle,
		pool:      loop.consumerPool,
		loopData:  loop.loopData,
	}
	consumerCounter := loop.consumerPool.Add(workers.MaxJob)
	for i := 0; i < consumerCounter; i++ {
		go consumer.Loop()
	}
	// lifecycle
	go func() {
		producerPool.Wait()
		loop.lifecycle.NextStep(StepClose)
		close(loop.loopData.chans.dirChan)
		close(loop.loopData.chans.fileChan)
	}()
}
548105
func (c *ServiceConfig) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()

	klog.Info("Starting service config controller")
	defer klog.Info("Shutting down service config controller")

	if !controller.WaitForCacheSync("service config", stopCh, c.listerSynced) {
		return
	}

	for i := range c.eventHandlers {
		klog.V(3).Info("Calling handler.OnServiceSynced()")
		c.eventHandlers[i].OnServiceSynced()
	}

	<-stopCh
}
465836
func (c *Client) Run(run params.RunParams) ([]params.ActionResult, error) {
	var results params.ActionResults
	err := c.facade.FacadeCall("Run", run, &results)
	return results.Results, err
}
133040
func (jm *Controller) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	klog.Infof("Starting CronJob Manager")
	// Check things every 10 second.
	go wait.Until(jm.syncAll, 10*time.Second, stopCh)
	<-stopCh
	klog.Infof("Shutting down CronJob Manager")
}
1057014
func (c *whoAmICommand) Run(ctx *cmd.Context) error {
	controllerName, err := modelcmd.DetermineCurrentController(c.store)
	if err != nil && !errors.IsNotFound(err) {
		return err
	}
	if err != nil {
		fmt.Fprintln(ctx.Stderr, "There is no current controller.\nRun juju list-controllers to see available controllers.")
		return nil
	}
	modelName, err := c.store.CurrentModel(controllerName)
	if err != nil && !errors.IsNotFound(err) {
		return err
	}
	userDetails, err := c.store.AccountDetails(controllerName)
	if err != nil && !errors.IsNotFound(err) {
		return err
	}
	if err != nil {
		fmt.Fprintf(ctx.Stderr, "You are not logged in to controller %q and model %q.\nRun juju login if you want to login.\n", controllerName, modelName)
		return nil
	}
	// Only qualify model name if there is a current model.
	if modelName != "" {
		if unqualifiedModelName, owner, err := jujuclient.SplitModelName(modelName); err == nil {
			user := names.NewUserTag(userDetails.User)
			modelName = common.OwnerQualifiedModelName(unqualifiedModelName, owner, user)
		}
	}

	result := whoAmI{
		ControllerName: controllerName,
		ModelName:      modelName,
		UserName:       userDetails.User,
	}
	return c.out.Write(ctx, result)
}
398680
func (c *autoRegisterController) Run(threadiness int, stopCh <-chan struct{}) {
	// don't let panics crash the process
	defer utilruntime.HandleCrash()
	// make sure the work queue is shutdown which will trigger workers to end
	defer c.queue.ShutDown()

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

	// wait for your secondary caches to fill before starting your work
	if !controllers.WaitForCacheSync("autoregister", stopCh, c.apiServiceSynced) {
		return
	}

	// record APIService objects that existed when we started
	if services, err := c.apiServiceLister.List(labels.Everything()); err == nil {
		for _, service := range services {
			c.apiServicesAtStart[service.Name] = true
		}
	}

	// start up your worker threads based on threadiness.  Some controllers have multiple kinds of workers
	for i := 0; i < threadiness; i++ {
		// runWorker will loop until "something bad" happens.  The .Until will then rekick the worker
		// after one second
		go wait.Until(c.runWorker, time.Second, stopCh)
	}

	// wait until we're told to stop
	<-stopCh
}
521826
func (c *ProjectFinalizerController) Run(stopCh <-chan struct{}, workers int) {
	defer runtime.HandleCrash()
	defer c.queue.ShutDown()

	// Wait for the stores to fill
	if !cache.WaitForCacheSync(stopCh, c.cacheSynced) {
		return
	}

	klog.V(5).Infof("Starting workers")
	for i := 0; i < workers; i++ {
		go c.worker()
	}
	<-stopCh
	klog.V(1).Infof("Shutting down")
}
849731
func (a *ActionAPI) Run(run params.RunParams) (results params.ActionResults, err error) {
	if err := a.checkCanAdmin(); err != nil {
		return results, err
	}

	if err := a.check.ChangeAllowed(); err != nil {
		return results, errors.Trace(err)
	}

	units, err := getAllUnitNames(a.state, run.Units, run.Applications)
	if err != nil {
		return results, errors.Trace(err)
	}

	machines := make([]names.Tag, len(run.Machines))
	for i, machineId := range run.Machines {
		if !names.IsValidMachine(machineId) {
			return results, errors.Errorf("invalid machine id %q", machineId)
		}
		machines[i] = names.NewMachineTag(machineId)
	}

	actionParams := a.createActionsParams(append(units, machines...), run.Commands, run.Timeout)

	return queueActions(a, actionParams)
}
437614
func (w *Worker) run() {
	for {
		// Dequeue a pending evaluation
		eval, token, waitIndex, shutdown := w.dequeueEvaluation(dequeueTimeout)
		if shutdown {
			return
		}

		// Check for a shutdown
		if w.srv.IsShutdown() {
			w.logger.Error("nacking eval because the server is shutting down", "eval", log.Fmt("%#v", eval))
			w.sendAck(eval.ID, token, false)
			return
		}

		// Wait for the raft log to catchup to the evaluation
		if err := w.waitForIndex(waitIndex, raftSyncLimit); err != nil {
			w.logger.Error("error waiting for Raft index", "error", err, "index", waitIndex)
			w.sendAck(eval.ID, token, false)
			continue
		}

		// Invoke the scheduler to determine placements
		if err := w.invokeScheduler(eval, token); err != nil {
			w.logger.Error("error invoking scheduler", "error", err)
			w.sendAck(eval.ID, token, false)
			continue
		}

		// Complete the evaluation
		w.sendAck(eval.ID, token, true)
	}
}
285948
func (c *listCommand) Run(ctx *cmd.Context) (err error) {
	if c.all {
		return c.listForController(ctx)
	}
	return c.listForModel(ctx)
}
78634
func (c *Controller) Run(ctx context.Context) {
	defer c.wg.Wait()
	defer c.crdContextCancel() // should be executed after stopping is set to true
	defer func() {
		c.wgLock.Lock()
		defer c.wgLock.Unlock()
		c.stopping = true
	}()

	c.Logger.Info("Starting Bundle controller")
	defer c.Logger.Info("Shutting down Bundle controller")

	sink := core_v1_client.EventSinkImpl{
		Interface: c.MainClient.CoreV1().Events(meta_v1.NamespaceNone),
	}
	recordingWatch := c.Broadcaster.StartRecordingToSink(&sink)
	defer recordingWatch.Stop()

	c.ReadyForWork()

	<-ctx.Done()
}
1003584
func (c *DeploymentController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	klog.Infof("Starting deployer controller")

	// Wait for the dc store to sync before starting any work in this controller.
	if !cache.WaitForCacheSync(stopCh, c.rcListerSynced, c.podListerSynced) {
		return
	}

	klog.Infof("Deployer controller caches are synced. Starting workers.")

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

	<-stopCh

	klog.Infof("Shutting down deployer controller")
}
517119
func (ccc *CSRCleanerController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()

	klog.Infof("Starting CSR cleaner controller")
	defer klog.Infof("Shutting down CSR cleaner controller")

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

	<-stopCh
}