// RunTTLController runs the workflow TTL controller
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)
	}
}
414011
func NewWorkflowController(
	restConfig *rest.Config,
	kubeclientset kubernetes.Interface,
	wfclientset wfclientset.Interface,
	namespace,
	executorImage,
	executorImagePullPolicy,
	configMap string,
) *WorkflowController {
	wfc := WorkflowController{
		restConfig:                 restConfig,
		kubeclientset:              kubeclientset,
		wfclientset:                wfclientset,
		configMap:                  configMap,
		namespace:                  namespace,
		cliExecutorImage:           executorImage,
		cliExecutorImagePullPolicy: executorImagePullPolicy,
		wfQueue:                    workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
		podQueue:                   workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
		completedPods:              make(chan string, 512),
	}
	wfc.throttler = NewThrottler(0, wfc.wfQueue)
	return &wfc
}
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")
}
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")
}
695313
func NewTokenController(service *goa.Service, app application.Application, tokenManager manager.TokenManager,
	configuration TokenControllerConfiguration) *TokenController {
	return &TokenController{
		Controller:    service.NewController("token"),
		TokenManager:  tokenManager,
		Configuration: configuration,
		app:           app,
	}
}
840985
func RunTTL(ttl TTL, cancel <-chan interface{}, min, max time.Duration) {

	logger := plog.WithFields(log.Fields{
		"name": ttl.Name(),
		"min":  int(min.Minutes()),
		"max":  int(max.Minutes()),
	})
	logger.Debug("Start TTL routine")
	original_min := min
	for {
		var repeatSooner bool
		wait, err := ttl.Purge(max)
		if err != nil {
			wait = min
			repeatSooner = true
			logger.WithField("wait", int(wait.Minutes())).WithError(err).
				Warning("Could not purge; trying again after wait period")
		} else {
			min = original_min
			repeatSooner = false
		}

		logger.WithField("wait", int(wait.Minutes())).Debug("Waiting for next purge cycle")
		select {
		case <-time.After(wait):
			// Use exponential backoff if repeating check after a failure
			if repeatSooner {
				min = 2 * min
				if min > max {
					min = max
				}
			}
		case <-cancel:
			return
		}
	}
}
1121314
func (c *NodeController) Run() {
	glog.Infof("starting node controller")

	stopCh := c.StopChannel()
	go c.runWatcher(stopCh)

	<-stopCh
	glog.Infof("shutting down node controller")
}
282764
func (c *snapshotController) Run(ctx <-chan struct{}) {
	glog.Infof("Starting snapshot controller")

	go c.snapshotController.Run(ctx)

	if !kcontroller.WaitForCacheSync("snapshot-controller", ctx, c.snapshotController.HasSynced) {
		return
	}

	go c.reconciler.Run(ctx)
	go c.desiredStateOfWorldPopulator.Run(ctx)

}
219137
func (t *Tango) RunTLS(certFile, keyFile string, args ...interface{}) {
	addr := getAddress(args...)

	t.logger.Info("Listening on https://" + addr)

	t.Server.Addr = addr
	t.Server.Handler = t

	err := t.ListenAndServeTLS(certFile, keyFile)
	if err != nil {
		t.logger.Error(err)
	}
}
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)
}
441606
func (t *WaitForFilteredReplicationTask) Run(parameters map[string]string) ([]*automationpb.TaskContainer, string, error) {
	keyspaceAndShard := topoproto.KeyspaceShardString(parameters["keyspace"], parameters["shard"])
	output, err := ExecuteVtctl(context.TODO(), parameters["vtctld_endpoint"],
		[]string{"WaitForFilteredReplication", "-max_delay", parameters["max_delay"], keyspaceAndShard})
	return nil, output, err
}
655741
func (w *Workflow) Run(ctx context.Context, manager *workflow.Manager, wi *topo.WorkflowInfo) error {
	w.uiUpdate()
	w.rootUINode.Display = workflow.NodeDisplayDeterminate
	w.rootUINode.BroadcastChanges(false /* updateChildren */)

	// Run all the validators. They may add fixers.
	for name, v := range validators {
		w.logger.Infof("Running validator: %v", name)
		w.uiUpdate()
		w.rootUINode.BroadcastChanges(false /* updateChildren */)
		err := v.Audit(ctx, manager.TopoServer(), w)
		if err != nil {
			w.logger.Errorf("Validator %v failed: %v", name, err)
		} else {
			w.logger.Infof("Validator %v successfully finished", name)
		}
		w.runCount++
	}

	// Now for each Fixer, add a sub node.
	if len(w.fixers) == 0 {
		w.logger.Infof("No problem found")
	}
	for i, f := range w.fixers {
		w.wg.Add(1)
		f.node = workflow.NewNode()
		w.rootUINode.Children = append(w.rootUINode.Children, f.node)
		f.node.PathName = fmt.Sprintf("%v", i)
		f.node.Name = f.name
		f.node.Message = f.message
		f.node.Display = workflow.NodeDisplayIndeterminate
		for _, action := range f.actions {
			f.node.Actions = append(f.node.Actions, &workflow.Action{
				Name:  action,
				State: workflow.ActionStateEnabled,
				Style: workflow.ActionStyleNormal,
			})
		}
		f.node.Listener = f
	}
	w.uiUpdate()
	w.rootUINode.BroadcastChanges(true /* updateChildren */)

	// And wait for the workflow to be done.
	fixersChan := make(chan struct{})
	go func(wg *sync.WaitGroup) {
		wg.Wait()
		close(fixersChan)
	}(&w.wg)
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-fixersChan:
		return nil
	}
}
644668
func (s *ScheduledImageStreamController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()

	klog.Infof("Starting scheduled import controller")

	// Wait for the stream store to sync before starting any work in this controller.
	if !cache.WaitForCacheSync(stopCh, s.listerSynced) {
		utilruntime.HandleError(fmt.Errorf("timed out waiting for caches to sync"))
		return
	}

	go s.scheduler.RunUntil(stopCh)

	metrics.InitializeImportCollector(true, s.importCounter.Collect)

	<-stopCh
	klog.Infof("Shutting down image stream 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
}
753901
func (tc *Controller) processTTL(job *batch.Job) (expired bool, err error) {
	// We don't care about the Jobs that are going to be deleted, or the ones that don't need clean up.
	if job.DeletionTimestamp != nil || !needsCleanup(job) {
		return false, nil
	}

	now := tc.clock.Now()
	t, err := timeLeft(job, &now)
	if err != nil {
		return false, err
	}

	// TTL has expired
	if *t <= 0 {
		return true, nil
	}

	tc.enqueueAfter(job, *t)
	return false, nil
}
605744
func (c *ClusterQuotaReconcilationController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()

	klog.Infof("Starting the cluster quota reconciliation controller")

	// the controllers that replenish other resources to respond rapidly to state changes
	go c.quotaMonitor.Run(stopCh)

	if !controller.WaitForCacheSync("cluster resource quota", stopCh, c.informerSyncedFuncs...) {
		return
	}

	// the workers that chug through the quota calculation backlog
	for i := 0; i < workers; i++ {
		go wait.Until(c.worker, time.Second, stopCh)
	}

	// the timer for how often we do a full recalculation across all quotas
	go wait.Until(func() { c.calculateAll() }, c.resyncPeriod, stopCh)

	<-stopCh
	klog.Infof("Shutting down ClusterQuotaReconcilationController")
	c.queue.ShutDown()
}
165116
func RunInTransaction(c context.Context, f func(tc context.Context) error,
	opts *datastore.TransactionOptions) error {

	return datastore.RunInTransaction(c, func(tc context.Context) error {
		tx := &transaction{}
		tc = context.WithValue(tc, &transactionKey, tx)
		if err := f(tc); err != nil {
			return err
		}

		// tx.Unlock() is not called as the tx context should never be called
		//again so we rather block than allow people to misuse the context.
		tx.Lock()
		memcacheCtx, err := memcacheContext(tc)
		if err != nil {
			return err
		}
		return memcacheSetMulti(memcacheCtx, tx.lockMemcacheItems)
	}, opts)
}