// Run executes the controller's main loop and block until the service exits
// according to it's restart policy or Close() is called.
436436
func (c *Controller) Run() (err error) {
	defer c.shutdown()
	sigc := make(chan os.Signal, 1)
	signal.Notify(sigc,
		syscall.SIGINT,
		syscall.SIGTERM,
		syscall.SIGQUIT)

	instID, err := strconv.Atoi(c.options.Service.InstanceID)
	if err != nil {
		return err
	}

	env := os.Environ()
	env = append(env, "CONTROLPLANE=1")
	env = append(env, fmt.Sprintf("CONTROLPLANE_CONSUMER_URL=http://localhost%s/api/metrics/store", c.options.Metric.Address))
	env = append(env, fmt.Sprintf("CONTROLPLANE_HOST_ID=%s", c.hostID))
	env = append(env, fmt.Sprintf("CONTROLPLANE_TENANT_ID=%s", c.tenantID))
	env = append(env, fmt.Sprintf("CONTROLPLANE_INSTANCE_ID=%s", c.options.Service.InstanceID))
	env = append(env, fmt.Sprintf("CONTROLPLANE_SERVICED_ID=%s", c.options.Service.ID))

	if err := writeEnvFile(env); err != nil {
		return err
	}

	args := []string{"-c", "exec " + strings.Join(c.options.Service.Command, " ")}

	startService := func() (*subprocess.Instance, chan error) {
		service, serviceExited, _ := subprocess.New(time.Second*10, env, "/bin/sh", args...)
		return service, serviceExited
	}

	sendSignal := func(service *subprocess.Instance, sig os.Signal) bool {
		switch {
		case c.PIDFile != "":
			c.forwardSignal(sig)
		case service != nil:
			service.Notify(sig)
		default:
			return false
		}
		return true
	}

	rpcDead, err := c.rpcHealthCheck()
	if err != nil {
		glog.Errorf("Could not setup RPC ping check: %s", err)
		return err
	}

	storageDead, err := c.storageHealthCheck()
	if err != nil {
		glog.Errorf("Could not set up storage check: %s", err)
		return err
	}

	prereqsPassed := make(chan bool)
	var startAfter <-chan time.Time
	var exitAfter <-chan time.Time
	var service *subprocess.Instance = nil
	serviceExited := make(chan error, 1)
	endpointExit := make(chan struct{})
	c.endpoints.Run(endpointExit)

	// Start CC Rest API Proxy
	go c.ccApiProxy.run()

	// HACK: I guess this is how it used to work?  This code is horrible.
	go func() {
		errc := <-c.closing
		close(endpointExit)
		errc <- nil
	}()

	if err := c.handleControlCenterImports(rpcDead); err != nil {
		glog.Error("Could not setup Control Center specific imports: ", err)
		return err
	}
	go c.checkPrereqs(prereqsPassed, rpcDead)
	go c.reapZombies(rpcDead)
	healthExit := make(chan struct{})
	defer close(healthExit)

	if os.Getenv("SERVICED_IS_SERVICE_SHELL") != "true" {
		c.kickOffHealthChecks(healthExit)
	}

	exited := false

	var shutdownService = func(service *subprocess.Instance, sig os.Signal) {
		c.options.Service.Autorestart = false
		if sendSignal(service, sig) {
			// nil out all other channels because we're shutting down
			sigc = nil
			prereqsPassed = nil
			startAfter = nil
			rpcDead = nil
			storageDead = nil

			// exitAfter is the deadman switch for unresponsive processes and any processes that have already exited
			exitAfter = time.After(time.Second * 30)

			glog.Infof("Closing healthExit on signal %v for %q", sig, c.options.Service.ID)
			close(healthExit)
			glog.Infof("Closed healthExit on signal %v for %q", sig, c.options.Service.ID)
		} else {
			c.exitStatus = 1
			exited = true
			glog.Infof("Exited due to sendSignal(%v) failed for %q", sig, c.options.Service.ID)
		}
	}

	for !exited {
		select {
		case sig := <-sigc:
			glog.Infof("Notifying subprocess of signal %v for service %s", sig, c.options.Service.ID)
			shutdownService(service, sig)
			glog.Infof("Notification complete for signal %v for service %s", sig, c.options.Service.ID)

		case <-exitAfter:
			glog.Infof("Killing unresponsive subprocess for service %s", c.options.Service.ID)
			sendSignal(service, syscall.SIGKILL)
			glog.Infof("Kill signal sent for service %s", c.options.Service.ID)
			c.exitStatus = 1
			exited = true

		case <-prereqsPassed:
			startAfter = time.After(time.Millisecond * 1)
			prereqsPassed = nil

		case exitError := <-serviceExited:
			if !c.options.Service.Autorestart {
				exitStatus, _ := utils.GetExitStatus(exitError)
				if c.options.Logforwarder.Enabled {
					time.Sleep(c.options.Logforwarder.SettleTime)
				}
				glog.Infof("Service %s Exited with status:%d due to %+v", c.options.Service.ID, exitStatus, exitError)
				//set loop to end
				exited = true
				//exit with exit code, defer so that other cleanup can happen
				c.exitStatus = exitStatus

			} else {
				glog.Infof("Restarting service process for service %s in 10 seconds.", c.options.Service.ID)
				service = nil
				startAfter = time.After(time.Second * 10)
			}

		case <-startAfter:
			glog.Infof("Starting service process for service %s", c.options.Service.ID)
			service, serviceExited = startService()
			startAfter = nil
		case <-rpcDead:
			glog.Infof("RPC Server has gone away, cleaning up service %s", c.options.Service.ID)
			shutdownService(service, syscall.SIGTERM)
			glog.Infof("RPC Server shutdown for service %s complete", c.options.Service.ID)
		case <-storageDead:
			glog.Infof("Distributed storage for service %s has gone away; shutting down", c.options.Service.ID)
			shutdownService(service, syscall.SIGTERM)
			glog.Infof("Distributed storage shutdown for service %s complete", c.options.Service.ID)
		}
	}
	// Signal to health check registry that this instance is giving up the ghost.
	client, err := node.NewLBClient(c.options.ServicedEndpoint)
	if err != nil {
		glog.Errorf("Could not create a client to endpoint: %s, %s", c.options.ServicedEndpoint, err)
		return nil
	}
	defer client.Close()
	c.Close()
	req := master.ServiceInstanceRequest{
		ServiceID:  c.options.Service.ID,
		InstanceID: instID,
	}
	client.ReportInstanceDead(req, nil)
	return nil
}
406983
func (dc *DeploymentController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer dc.queue.ShutDown()

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

	if !controller.WaitForCacheSync("deployment", stopCh, dc.dListerSynced, dc.rsListerSynced, dc.podListerSynced) {
		return
	}

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

	<-stopCh
}
329182
func (c *migrateCommand) Run(ctx *cmd.Context) error {
	spec, err := c.getMigrationSpec()
	if err != nil {
		return err
	}
	modelName, err := c.ModelName()
	if err != nil {
		return errors.Trace(err)
	}
	uuids, err := c.ModelUUIDs([]string{modelName})
	if err != nil {
		return errors.Trace(err)
	}
	spec.ModelUUID = uuids[0]
	if err := c.checkMigrationFeasibility(spec); err != nil {
		return errors.Trace(err)
	}
	controllerName, err := c.ControllerName()
	if err != nil {
		return err
	}
	api, err := c.getMigrationAPI(controllerName)
	if err != nil {
		return err
	}
	defer func() { _ = api.Close() }()
	id, err := api.InitiateMigration(*spec)
	if err != nil {
		return err
	}
	ctx.Infof("Migration started with ID %q", id)
	return nil
}
564707
func (c *configCommand) Run(ctx *cmd.Context) error {
	client, err := c.getAPI()
	if err != nil {
		return errors.Trace(err)
	}
	defer func() { _ = client.Close() }()

	if len(c.resetKeys) > 0 {
		if err := c.resetConfig(client, ctx); err != nil {
			// We return this error naked as it is almost certainly going to be
			// cmd.ErrSilent and the cmd.Command framework expects that back
			// from cmd.Run if the process is blocked.
			return err
		}
	}
	if c.action == nil {
		// If we are reset only, we end up here; we have already done that.
		return nil
	}

	return c.action(client, ctx)
}
653402
func (c *grantCommand) Run(ctx *cmd.Context) error {
	if len(c.ModelNames) > 0 {
		return c.runForModel()
	}
	if len(c.OfferURLs) > 0 {
		if err := setUnsetUsers(c, c.OfferURLs); err != nil {
			return errors.Trace(err)
		}
		return c.runForOffers()
	}
	return c.runForController()
}
202045
func (svc *AsyncService) Run() error {
	for {
		e, open := <-svc.c
		if open {
			// Call event handler
			svc.i.HandleEvent(e)
		} else {
			// Exit runner
			break
		}
	}
	return nil
}
1183993
func (c *AvailableConditionController) Run(threadiness int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

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

	if !controllers.WaitForCacheSync("AvailableConditionController", stopCh, c.apiServiceSynced, c.servicesSynced, c.endpointsSynced) {
		return
	}

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

	<-stopCh
}
704311
func (c *PodController) Run() {
	glog.Infof("starting pod controller")

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

	<-stopCh
	glog.Infof("shutting down pod controller")
}
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
}
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
}
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
}
972900
func (s *Server) Run() {
	s.Log.Info("msg", "Starting HTTP Server")

	// start http server
	httpServer := &http.Server{Addr: s.Config.HTTPAddr, Handler: s}

	go func() {
		s.Log.Info("msg", "HTTP Server listening", "address", s.Config.HTTPAddr)
		httpServer.ListenAndServe()
	}()

	// wait for stop signal
	<-s.stopCh

	s.Log.Info("msg", "Shutting down HTTP Server...")

	httpServer.Shutdown(context.Background())
}
294865
func (e *TokensController) Run(workers int, stopCh <-chan struct{}) {
	// Shut down queues
	defer utilruntime.HandleCrash()
	defer e.syncServiceAccountQueue.ShutDown()
	defer e.syncSecretQueue.ShutDown()

	if !controller.WaitForCacheSync("tokens", stopCh, e.serviceAccountSynced, e.secretSynced) {
		return
	}

	klog.V(5).Infof("Starting workers")
	for i := 0; i < workers; i++ {
		go wait.Until(e.syncServiceAccount, 0, stopCh)
		go wait.Until(e.syncSecret, 0, stopCh)
	}
	<-stopCh
	klog.V(1).Infof("Shutting down")
}
987718
func (c *Controller) Run(ctx context.Context) error {
	// Watch CollectorSet objects
	err := c.watch(ctx)
	if err != nil {
		return err
	}

	<-ctx.Done()

	return ctx.Err()
}
1062213
func Run(service Service, sig ...os.Signal) error {
	env := environment{}
	if err := service.Init(env); err != nil {
		return err
	}

	if err := service.Start(); err != nil {
		return err
	}

	if len(sig) == 0 {
		sig = []os.Signal{syscall.SIGINT, syscall.SIGTERM}
	}

	signalChan := make(chan os.Signal, 1)
	signalNotify(signalChan, sig...)
	<-signalChan

	return service.Stop()
}
285948
func (c *listCommand) Run(ctx *cmd.Context) (err error) {
	if c.all {
		return c.listForController(ctx)
	}
	return c.listForModel(ctx)
}
1211035
func (brw *blockRetrievalWorker) run() {
	defer close(brw.doneCh)
	for {
		err := brw.HandleRequest()
		// Only io.EOF is relevant to the loop; other errors are handled in
		// FinalizeRequest
		if err == io.EOF {
			return
		}
	}
}