// Controller returns a docker container runner.
172388
func (e *executor) Controller(t *api.Task) (exec.Controller, error) {
	dependencyGetter := template.NewTemplatedDependencyGetter(agent.Restrict(e.dependencies, t), t, nil)

	// Get the node description from the executor field
	e.mutex.Lock()
	nodeDescription := e.node
	e.mutex.Unlock()

	if t.Spec.GetAttachment() != nil {
		return newNetworkAttacherController(e.backend, e.imageBackend, e.volumeBackend, t, nodeDescription, dependencyGetter)
	}

	var ctlr exec.Controller
	switch r := t.Spec.GetRuntime().(type) {
	case *api.TaskSpec_Generic:
		logrus.WithFields(logrus.Fields{
			"kind":     r.Generic.Kind,
			"type_url": r.Generic.Payload.TypeUrl,
		}).Debug("custom runtime requested")
		runtimeKind, err := naming.Runtime(t.Spec)
		if err != nil {
			return ctlr, err
		}
		switch runtimeKind {
		case string(swarmtypes.RuntimePlugin):
			if !e.backend.HasExperimental() {
				return ctlr, fmt.Errorf("runtime type %q only supported in experimental", swarmtypes.RuntimePlugin)
			}
			c, err := plugin.NewController(e.pluginBackend, t)
			if err != nil {
				return ctlr, err
			}
			ctlr = c
		default:
			return ctlr, fmt.Errorf("unsupported runtime type: %q", runtimeKind)
		}
	case *api.TaskSpec_Container:
		c, err := newController(e.backend, e.imageBackend, e.volumeBackend, t, nodeDescription, dependencyGetter)
		if err != nil {
			return ctlr, err
		}
		ctlr = c
	default:
		return ctlr, fmt.Errorf("unsupported runtime: %q", r)
	}

	return ctlr, nil
}
880337
func (o *OpenShiftControllerManager) RunControllerManager() error {
	// try to decode into our new types first.  right now there is no validation, no file path resolution.  this unsticks the operator to start.
	// TODO add those things
	configContent, err := ioutil.ReadFile(o.ConfigFilePath)
	if err != nil {
		return err
	}
	scheme := runtime.NewScheme()
	utilruntime.Must(openshiftcontrolplanev1.Install(scheme))
	codecs := serializer.NewCodecFactory(scheme)
	obj, err := runtime.Decode(codecs.UniversalDecoder(openshiftcontrolplanev1.GroupVersion, configv1.GroupVersion), configContent)
	switch {
	case runtime.IsMissingVersion(err): // fall through to legacy master config
	case runtime.IsMissingKind(err): // fall through to legacy master config
	case runtime.IsNotRegisteredError(err): // fall through to legacy master config
	case err != nil:
		return err
	case err == nil:
		// Resolve relative to CWD
		absoluteConfigFile, err := api.MakeAbs(o.ConfigFilePath, "")
		if err != nil {
			return err
		}
		configFileLocation := path.Dir(absoluteConfigFile)

		config := obj.(*openshiftcontrolplanev1.OpenShiftControllerManagerConfig)
		/// this isn't allowed to be nil when by itself.
		// TODO remove this when the old path is gone.
		if config.ServingInfo == nil {
			config.ServingInfo = &configv1.HTTPServingInfo{}
		}
		if err := helpers.ResolvePaths(configconversion.GetOpenShiftControllerConfigFileReferences(config), configFileLocation); err != nil {
			return err
		}
		configdefault.SetRecommendedOpenShiftControllerConfigDefaults(config)

		clientConfig, err := helpers.GetKubeClientConfig(config.KubeClientConfig)
		if err != nil {
			return err
		}
		return RunOpenShiftControllerManager(config, clientConfig)
	}

	masterConfig, err := configapilatest.ReadAndResolveMasterConfig(o.ConfigFilePath)
	if err != nil {
		return err
	}

	validationResults := validation.ValidateMasterConfig(masterConfig, nil)
	if len(validationResults.Warnings) != 0 {
		for _, warning := range validationResults.Warnings {
			klog.Warningf("%v", warning)
		}
	}
	if len(validationResults.Errors) != 0 {
		return kerrors.NewInvalid(configapi.Kind("MasterConfig"), "master-config.yaml", validationResults.Errors)
	}

	// round trip to external
	externalMasterConfig, err := configapi.Scheme.ConvertToVersion(masterConfig, legacyconfigv1.LegacySchemeGroupVersion)
	if err != nil {
		return err
	}
	config := ConvertMasterConfigToOpenshiftControllerConfig(externalMasterConfig.(*legacyconfigv1.MasterConfig))
	clientConfig, err := helpers.GetKubeClientConfig(config.KubeClientConfig)
	if err != nil {
		return err
	}

	return RunOpenShiftControllerManager(config, clientConfig)
}
1281428
func NewController(options ControllerOptions) (*Controller, error) {
	c := &Controller{
		options: options,
	}
	c.closing = make(chan chan error)

	if len(options.ServicedEndpoint) <= 0 {
		return nil, ErrInvalidEndpoint
	}

	// Load keys
	keyshutdown := make(chan interface{})
	go func() {
		// This is a ridiculous shutdown pattern I'll follow for the sake of
		// not destabilizing the controller this late in a release.
		// TODO: Make this not stupid
		errc := <-c.closing
		close(keyshutdown)
		errc <- nil
	}()

	go auth.WatchDelegateKeyFile(containerDelegateKeyFile, keyshutdown)
	go auth.WatchTokenFile(containerTokenFile, keyshutdown)

	// Load the delegate keys and auth tokens first so there's no race btwn starting the watcher routines
	//    and making the first RPC call in getService()
	<-auth.WaitForDelegateKeys(nil)
	<-auth.WaitForAuthToken(nil)

	// get service
	instanceID, err := strconv.Atoi(options.Service.InstanceID)
	if err != nil {
		glog.Errorf("Invalid instance from instanceID:%s", options.Service.InstanceID)
		return c, fmt.Errorf("Invalid instance from instanceID:%s", options.Service.InstanceID)
	}
	service, tenantID, svcPath, err := getService(options.ServicedEndpoint, options.Service.ID, instanceID)
	if err != nil {
		glog.Errorf("%+v", err)
		glog.Errorf("Invalid service from serviceID:%s", options.Service.ID)
		return c, ErrInvalidService
	}
	c.healthChecks = service.HealthChecks
	c.tenantID = tenantID
	options.ServiceNamePath = svcPath

	if service.PIDFile != "" {
		if strings.HasPrefix(service.PIDFile, "exec ") {
			cmd := service.PIDFile[5:len(service.PIDFile)]
			out, err := exec.Command("sh", "-c", cmd).Output()
			if err != nil {
				glog.Errorf("Unable to run command '%s'", cmd)
			} else {
				c.PIDFile = strings.Trim(string(out), "\n ")
			}
		} else {
			c.PIDFile = service.PIDFile
		}
	}

	// create config files
	if err := setupConfigFiles(service); err != nil {
		glog.Errorf("Could not setup config files error:%s", err)
		return c, fmt.Errorf("container: invalid ConfigFiles error:%s", err)
	}

	// get host id
	c.hostID, err = getAgentHostID(options.ServicedEndpoint)
	if err != nil {
		glog.Errorf("Invalid hostID")
		return c, ErrInvalidHostID
	}

	if options.Logforwarder.Enabled && len(service.LogConfigs) > 0 {
		if err := setupLogstashFiles(c.hostID, options.HostIPs, options.ServiceNamePath, service,
			options.Service.InstanceID, options.Logforwarder); err != nil {
			glog.Errorf("Could not setup logstash files error:%s", err)
			return c, fmt.Errorf("container: invalid LogStashFiles error:%s", err)
		}

		logforwarder, exited, err := subprocess.New(time.Second,
			nil,
			options.Logforwarder.Path,
			"-e", // Log to stderr
			"-c", options.Logforwarder.ConfigFile)
		if err != nil {
			return nil, err
		}
		c.logforwarder = logforwarder
		c.logforwarderExited = exited
	}

	//build metric redirect url -- assumes 8444 is port mapped
	metricRedirect := options.Metric.RemoteEndoint
	if len(metricRedirect) == 0 {
		glog.V(1).Infof("container.Controller does not have metric forwarding")
	} else if !options.MetricForwarding {
		glog.V(1).Infof("Not forwarding metrics for this container (%v)", c.tenantID)
	} else {
		if len(c.tenantID) <= 0 {
			return nil, ErrInvalidTenantID
		}
		if len(c.hostID) <= 0 {
			return nil, ErrInvalidHostID
		}
		if len(options.Service.ID) <= 0 {
			return nil, ErrInvalidServiceID
		}

		metricRedirect += "?controlplane_tenant_id=" + c.tenantID
		metricRedirect += "&controlplane_service_id=" + options.Service.ID
		metricRedirect += "&controlplane_host_id=" + c.hostID
		metricRedirect += "&controlplane_instance_id=" + options.Service.InstanceID

		//build and serve the container metric forwarder
		forwarder, err := NewMetricForwarder(options.Metric.Address, metricRedirect)
		if err != nil {
			return c, err
		}
		c.metricForwarder = forwarder

		// setup network stats
		destination := fmt.Sprintf("http://localhost%s/api/metrics/store", options.Metric.Address)
		glog.Infof("pushing network stats to: %s", destination)
		go statReporter(destination, time.Second*15)
	}

	// Keep a copy of the service prerequisites in the Controller object.
	c.prereqs = service.Prereqs

	// set up the zookeeper client
	c.zkInfo, err = getAgentZkInfo(options.ServicedEndpoint)
	if err != nil {
		glog.Errorf("Invalid zk info: %v", err)
		return c, err
	}
	glog.Infof(" c.zkInfo: %+v", c.zkInfo)

	// endpoints are created at the root level (not pool aware)
	rootBasePath := ""
	zClient, err := coordclient.New("zookeeper", c.zkInfo.ZkDSN, rootBasePath, nil)
	if err != nil {
		glog.Errorf("failed create a new coordclient: %v", err)
		return c, err
	}

	zzk.InitializeLocalClient(zClient)

	// get endpoints
	opts := ContainerEndpointsOptions{
		HostID:               c.hostID,
		TenantID:             c.tenantID,
		InstanceID:           instanceID,
		IsShell:              os.Getenv("SERVICED_IS_SERVICE_SHELL") == "true",
		TCPMuxPort:           uint16(options.Mux.Port),
		UseTLS:               !options.Mux.DisableTLS,
		VirtualAddressSubnet: options.VirtualAddressSubnet,
	}
	c.endpoints, err = NewContainerEndpoints(service, opts)
	if err != nil {
		return c, err
	}

	// CC Rest API proxy
	c.ccApiProxy = newServicedApiProxy()

	// check command
	glog.Infof("command: %v [%d]", options.Service.Command, len(options.Service.Command))
	if len(options.Service.Command) < 1 {
		glog.Errorf("Invalid commandif ")
		return c, ErrInvalidCommand
	}

	return c, nil
}
1288043
func (r *resolver) container() *block {
	for b := r.env; ; b = b.parent {
		if b.function != nil || b.isModule() {
			return b
		}
	}
}
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
}
936962
func (c *MemStore) AllControllers() (map[string]ControllerDetails, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	result := make(map[string]ControllerDetails)
	for name, details := range c.Controllers {
		result[name] = details
	}
	return result, nil
}
984627
func NewShowControllerCommand() cmd.Command {
	cmd := &showControllerCommand{
		store: jujuclient.NewFileClientStore(),
	}
	return modelcmd.WrapBase(cmd)
}
911124
func (e *DockerRegistryServiceController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer e.registryLocationQueue.ShutDown()

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

	// Wait for the store to sync before starting any work in this controller.
	ready := make(chan struct{})
	go e.waitForDockerURLs(ready, stopCh)
	select {
	case <-ready:
	case <-stopCh:
		return
	}
	klog.V(1).Infof("caches synced")

	go wait.Until(e.watchForDockerURLChanges, time.Second, stopCh)
	for i := 0; i < workers; i++ {
		go wait.Until(e.watchForDockercfgSecretUpdates, time.Second, stopCh)
	}
	<-stopCh
}
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
}
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,
	}

}
950539
func (c *Client) GrantController(user, access string) error {
	return c.modifyControllerUser(params.GrantControllerAccess, user, access)
}
390161
func (c *DNSController) Run() {
	glog.Infof("starting DNS controller")

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

	<-stopCh
	glog.Infof("shutting down DNS controller")
}
476023
func Docker(cmd icmd.Cmd, cmdOperators ...CmdOperator) *icmd.Result {
	for _, op := range cmdOperators {
		deferFn := op(&cmd)
		if deferFn != nil {
			defer deferFn()
		}
	}
	appendDocker(&cmd)
	if err := validateArgs(cmd.Command...); err != nil {
		return &icmd.Result{
			Error: err,
		}
	}
	return icmd.RunCmd(cmd)
}
1090832
func controllerCinderVolumes(storageAdapter OpenstackStorage, controllerUUID string) ([]cinder.Volume, error) {
	return cinderVolumes(storageAdapter, func(v *cinder.Volume) bool {
		return v.Metadata[tags.JujuController] == controllerUUID
	})
}
767903
func Container() (c *ContainerController) {
	return &ContainerController{
		List:      containerList,
		Detail:    containerDetail,
		Raw:       containerRaw,
		Logs:      containerLogs,
		FetchLogs: containerFetchLogs,
		Delete:    containerDelete,
		Exec:      containerExec,
		Connect:   containerConnect,
	}
}
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()
}
179108
func (rc *externalController) ControllerInfo() crossmodel.ControllerInfo {
	return crossmodel.ControllerInfo{
		ControllerTag: names.NewControllerTag(rc.doc.Id),
		Alias:         rc.doc.Alias,
		Addrs:         rc.doc.Addrs,
		CACert:        rc.doc.CACert,
	}
}