// Run sends a single probes out to all of the other cilium nodes to gather
// connectivity status for the cluster.
547226
func (p *prober) Run() error {
	err := p.Pinger.Run()
	p.runHTTPProbe()
	return err
}
403311
func (h *Hive) Run(p *BzzPeer) error {
	h.trackPeer(p)
	defer h.untrackPeer(p)

	dp := NewPeer(p, h.Kademlia)
	depth, changed := h.On(dp)
	// if we want discovery, advertise change of depth
	if h.Discovery {
		if changed {
			// if depth changed, send to all peers
			NotifyDepth(depth, h.Kademlia)
		} else {
			// otherwise just send depth to new peer
			dp.NotifyDepth(depth)
		}
		NotifyPeer(p.BzzAddr, h.Kademlia)
	}
	defer h.Off(dp)
	return dp.Run(dp.HandleMsg)
}
12264
func (c *CloudNodeLifecycleController) 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 check if any nodes have been
	// deleted or shutdown from the cloudprovider
	wait.Until(c.MonitorNodes, c.nodeMonitorPeriod, stopCh)
}
613690
func (k *Kafka) Run(sigCh <-chan os.Signal, ready chan<- struct{}) error {
	if k.Image == "" {
		k.Image = KafkaDefaultImage
	}

	if k.Name == "" {
		k.Name = DefaultNamer()
	}

	if k.HostIP == "" {
		k.HostIP = "127.0.0.1"
	}

	if k.ContainerPort == docker.Port("") {
		k.ContainerPort = docker.Port("9092/tcp")
	}

	if k.StartTimeout == 0 {
		k.StartTimeout = DefaultStartTimeout
	}

	if k.Client == nil {
		client, err := docker.NewClientFromEnv()
		if err != nil {
			return err
		}
		k.Client = client
	}

	if k.DefaultReplicationFactor == 0 {
		k.DefaultReplicationFactor = 1
	}

	if k.MinInsyncReplicas == 0 {
		k.MinInsyncReplicas = 1
	}

	if k.ZooKeeperConnect == "" {
		k.ZooKeeperConnect = "zookeeper:2181/kafka"
	}

	if k.MessageMaxBytes == 0 {
		k.MessageMaxBytes = 1000012
	}

	if k.ReplicaFetchMaxBytes == 0 {
		k.ReplicaFetchMaxBytes = 1048576
	}

	if k.ReplicaFetchResponseMaxBytes == 0 {
		k.ReplicaFetchResponseMaxBytes = 10485760
	}

	containerOptions := docker.CreateContainerOptions{
		Name: k.Name,
		Config: &docker.Config{
			Image: k.Image,
			Env:   k.buildEnv(),
		},
		HostConfig: &docker.HostConfig{
			AutoRemove: true,
			PortBindings: map[docker.Port][]docker.PortBinding{
				k.ContainerPort: {{
					HostIP:   k.HostIP,
					HostPort: strconv.Itoa(k.HostPort),
				}},
			},
		},
	}

	if k.NetworkName != "" {
		nw, err := k.Client.NetworkInfo(k.NetworkName)
		if err != nil {
			return err
		}

		containerOptions.NetworkingConfig = &docker.NetworkingConfig{
			EndpointsConfig: map[string]*docker.EndpointConfig{
				k.NetworkName: {
					NetworkID: nw.ID,
				},
			},
		}
	}

	container, err := k.Client.CreateContainer(containerOptions)
	if err != nil {
		return err
	}
	k.ContainerID = container.ID

	err = k.Client.StartContainer(container.ID, nil)
	if err != nil {
		return err
	}
	defer k.Stop()

	container, err = k.Client.InspectContainer(container.ID)
	if err != nil {
		return err
	}

	k.HostAddress = net.JoinHostPort(
		container.NetworkSettings.Ports[k.ContainerPort][0].HostIP,
		container.NetworkSettings.Ports[k.ContainerPort][0].HostPort,
	)
	k.ContainerAddress = net.JoinHostPort(
		container.NetworkSettings.Networks[k.NetworkName].IPAddress,
		k.ContainerPort.Port(),
	)

	logContext, cancelLogs := context.WithCancel(context.Background())
	defer cancelLogs()
	go k.streamLogs(logContext)

	containerExit := k.wait()
	ctx, cancel := context.WithTimeout(context.Background(), k.StartTimeout)
	defer cancel()

	select {
	case <-ctx.Done():
		return errors.Wrapf(ctx.Err(), "kafka broker in container %s did not start", k.ContainerID)
	case <-containerExit:
		return errors.New("container exited before ready")
	case <-k.ready(ctx, k.ContainerAddress):
		k.Address = k.ContainerAddress
	case <-k.ready(ctx, k.HostAddress):
		k.Address = k.HostAddress
	}

	cancel()
	close(ready)

	for {
		select {
		case err := <-containerExit:
			return err
		case <-sigCh:
			if err := k.Stop(); err != nil {
				return err
			}
		}
	}
}
887635
func (kubemarkController *KubemarkController) Run(stopCh chan struct{}) {
	nodeTemplate, err := kubemarkController.getNodeTemplate()
	if err != nil {
		klog.Fatalf("failed to get node template: %s", err)
	}
	kubemarkController.nodeTemplate = nodeTemplate

	go kubemarkController.runNodeCreation(stopCh)
	<-stopCh
}
1308072
func (k *KeyManager) Run(ctx context.Context) error {
	k.mu.Lock()
	ctx = log.WithModule(ctx, "keymanager")
	var (
		clusters []*api.Cluster
		err      error
	)
	k.store.View(func(readTx store.ReadTx) {
		clusters, err = store.FindClusters(readTx, store.ByName(k.config.ClusterName))
	})

	if err != nil {
		log.G(ctx).Errorf("reading cluster config failed, %v", err)
		k.mu.Unlock()
		return err
	}

	cluster := clusters[0]
	if len(cluster.NetworkBootstrapKeys) == 0 {
		for _, subsys := range k.config.Subsystems {
			for i := 0; i < keyringSize; i++ {
				k.keyRing.keys = append(k.keyRing.keys, k.allocateKey(ctx, subsys))
			}
		}
		if err := k.updateKey(cluster); err != nil {
			log.G(ctx).Errorf("store update failed %v", err)
		}
	} else {
		k.keyRing.lClock = cluster.EncryptionKeyLamportClock
		k.keyRing.keys = cluster.NetworkBootstrapKeys
	}

	ticker := time.NewTicker(k.config.RotationInterval)
	defer ticker.Stop()

	k.ctx, k.cancel = context.WithCancel(ctx)
	k.mu.Unlock()

	for {
		select {
		case <-ticker.C:
			k.rotateKey(ctx)
		case <-k.ctx.Done():
			return nil
		}
	}
}
1174906
func (m *cloudResourceSyncManager) Run(stopCh <-chan struct{}) {
	wait.Until(m.syncNodeAddresses, m.syncPeriod, stopCh)
}
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)
}
315369
func (s *Server) Run(ctx context.Context) error {
	s.mu.Lock()
	if s.isRunning() {
		s.mu.Unlock()
		return errors.New("CA signer is already running")
	}
	s.wg.Add(1)
	s.ctx, s.cancel = context.WithCancel(log.WithModule(ctx, "ca"))
	ctx = s.ctx
	s.mu.Unlock()
	defer s.wg.Done()
	defer func() {
		s.mu.Lock()
		s.mu.Unlock()
	}()

	// Retrieve the channels to keep track of changes in the cluster
	// Retrieve all the currently registered nodes
	var (
		nodes   []*api.Node
		cluster *api.Cluster
		err     error
	)
	updates, cancel, err := store.ViewAndWatch(
		s.store,
		func(readTx store.ReadTx) error {
			cluster = store.GetCluster(readTx, s.clusterID)
			if cluster == nil {
				return errors.New("could not find cluster object")
			}
			nodes, err = store.FindNodes(readTx, store.All)
			return err
		},
		api.EventCreateNode{},
		api.EventUpdateNode{},
		api.EventDeleteNode{},
		api.EventUpdateCluster{
			Cluster: &api.Cluster{ID: s.clusterID},
			Checks:  []api.ClusterCheckFunc{api.ClusterCheckID},
		},
	)

	// call once to ensure that the join tokens and local/external CA signer are always set
	rootReconciler := &rootRotationReconciler{
		ctx:                 log.WithField(ctx, "method", "(*Server).rootRotationReconciler"),
		clusterID:           s.clusterID,
		store:               s.store,
		batchUpdateInterval: s.rootReconciliationRetryInterval,
	}

	s.UpdateRootCA(ctx, cluster, rootReconciler)

	// Do this after updateCluster has been called, so Ready() and isRunning never returns true without
	// the join tokens and external CA/security config's root CA being set correctly
	s.mu.Lock()
	close(s.started)
	s.mu.Unlock()

	if err != nil {
		log.G(ctx).WithFields(logrus.Fields{
			"method": "(*Server).Run",
		}).WithError(err).Errorf("snapshot store view failed")
		return err
	}
	defer cancel()

	// We might have missed some updates if there was a leader election,
	// so let's pick up the slack.
	if err := s.reconcileNodeCertificates(ctx, nodes); err != nil {
		// We don't return here because that means the Run loop would
		// never run. Log an error instead.
		log.G(ctx).WithFields(logrus.Fields{
			"method": "(*Server).Run",
		}).WithError(err).Errorf("error attempting to reconcile certificates")
	}

	ticker := time.NewTicker(s.reconciliationRetryInterval)
	defer ticker.Stop()

	externalTLSCredsChange, externalTLSWatchCancel := s.securityConfig.Watch()
	defer externalTLSWatchCancel()

	// Watch for new nodes being created, new nodes being updated, and changes
	// to the cluster
	for {
		select {
		case <-ctx.Done():
			return nil
		default:
		}

		select {
		case event := <-updates:
			switch v := event.(type) {
			case api.EventCreateNode:
				s.evaluateAndSignNodeCert(ctx, v.Node)
				rootReconciler.UpdateNode(v.Node)
			case api.EventUpdateNode:
				// If this certificate is already at a final state
				// no need to evaluate and sign it.
				if !isFinalState(v.Node.Certificate.Status) {
					s.evaluateAndSignNodeCert(ctx, v.Node)
				}
				rootReconciler.UpdateNode(v.Node)
			case api.EventDeleteNode:
				rootReconciler.DeleteNode(v.Node)
			case api.EventUpdateCluster:
				if v.Cluster.ID == s.clusterID {
					s.UpdateRootCA(ctx, v.Cluster, rootReconciler)
				}
			}
		case <-externalTLSCredsChange:
			// The TLS certificates can rotate independently of the root CA (and hence which roots the
			// external CA trusts) and external CA URLs.  It's possible that the root CA update is received
			// before the external TLS cred change notification.  During that period, it is possible that
			// the TLS creds will expire or otherwise fail to authorize against external CAs.  However, in
			// that case signing will just fail with a recoverable connectivity error - the state of the
			// certificate issuance is left as pending, and on the next tick, the server will try to sign
			// all nodes with pending certs again (by which time the TLS cred change will have been
			// received).

			// Note that if the external CA changes, the new external CA *MUST* trust the current server's
			// certificate issuer, and this server's certificates should not be extremely close to expiry,
			// otherwise this server would not be able to get new TLS certificates and will no longer be
			// able to function.
			s.signingMu.Lock()
			s.externalCA.UpdateTLSConfig(NewExternalCATLSConfig(
				s.securityConfig.ClientTLSCreds.Config().Certificates, s.externalCAPool))
			s.signingMu.Unlock()
		case <-ticker.C:
			for _, node := range s.pending {
				if err := s.evaluateAndSignNodeCert(ctx, node); err != nil {
					// If this sign operation did not succeed, the rest are
					// unlikely to. Yield so that we don't hammer an external CA.
					// Since the map iteration order is randomized, there is no
					// risk of getting stuck on a problematic CSR.
					break
				}
			}
		case <-ctx.Done():
			return nil
		}
	}
}
96539
func (d *DockerMonitor) Run(ctx context.Context) error {

	if err := d.config.IsComplete(); err != nil {
		return fmt.Errorf("docker: %s", err)
	}

	err := d.waitForDockerDaemon(ctx)
	if err != nil {
		zap.L().Error("Docker daemon is not running - skipping container processing", zap.Error(err))
		return nil
	}

	if d.syncAtStart && d.config.Policy != nil {

		options := types.ContainerListOptions{All: true}
		containers, err := d.dockerClient.ContainerList(ctx, options)
		if err != nil {
			return fmt.Errorf("unable to get container list: %s", err)
		}

		// Starting the eventListener and wait to hear on channel for it to be ready.
		// Need to start before the resync process so that we don't loose any events.
		// They will be buffered. We don't want to start the listener before
		// getting the list from docker though, to avoid duplicates.
		listenerReady := make(chan struct{})
		go d.eventListener(ctx, listenerReady)
		<-listenerReady

		zap.L().Debug("Syncing all existing containers")
		// Syncing all Existing containers depending on MonitorSetting
		if err := d.resyncContainers(ctx, containers); err != nil {
			zap.L().Error("Unable to sync existing containers", zap.Error(err))
		}
	} else {
		// Starting the eventListener and wait to hear on channel for it to be ready.
		// We are not doing resync. We just start the listener.
		listenerReady := make(chan struct{})
		go d.eventListener(ctx, listenerReady)
		<-listenerReady
	}

	// Start processing the events
	go d.eventProcessors(ctx)

	return nil
}
71451
func (e *Epoller) Run(ctx context.Context) {
	var events [128]unix.EpollEvent
	for {
		select {
		case <-ctx.Done():
			e.Close()
			return
		default:
			n, err := unix.EpollWait(e.fd, events[:], -1)
			if err != nil {
				if err == unix.EINTR {
					continue
				}
				logrus.WithError(err).Error("cgroups: epoll wait")
			}
			for i := 0; i < n; i++ {
				e.process(ctx, uintptr(events[i].Fd))
			}
		}
	}
}
704311
func (c *PodController) Run() {
	glog.Infof("starting pod controller")

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

	<-stopCh
	glog.Infof("shutting down pod controller")
}
122723
func (r *Reset) Run(out io.Writer, client clientset.Interface, cfg *kubeadmapi.InitConfiguration) error {
	var dirsToClean []string

	// Reset the ClusterStatus for a given control-plane node.
	if isControlPlane() && cfg != nil {
		uploadconfig.ResetClusterStatusForNode(cfg.NodeRegistration.Name, client)
	}

	// Only clear etcd data when using local etcd.
	klog.V(1).Infoln("[reset] Checking for etcd config")
	etcdManifestPath := filepath.Join(kubeadmconstants.KubernetesDir, kubeadmconstants.ManifestsSubDirName, "etcd.yaml")
	etcdDataDir, err := getEtcdDataDir(etcdManifestPath, cfg)
	if err == nil {
		dirsToClean = append(dirsToClean, etcdDataDir)
		if cfg != nil {
			if err := etcdphase.RemoveStackedEtcdMemberFromCluster(client, cfg); err != nil {
				klog.Warningf("[reset] failed to remove etcd member: %v\n.Please manually remove this etcd member using etcdctl", err)
			}
		}
	} else {
		fmt.Println("[reset] No etcd config found. Assuming external etcd")
		fmt.Println("[reset] Please manually reset etcd to prevent further issues")
	}

	// Try to stop the kubelet service
	klog.V(1).Infoln("[reset] Getting init system")
	initSystem, err := initsystem.GetInitSystem()
	if err != nil {
		klog.Warningln("[reset] The kubelet service could not be stopped by kubeadm. Unable to detect a supported init system!")
		klog.Warningln("[reset] Please ensure kubelet is stopped manually")
	} else {
		fmt.Println("[reset] Stopping the kubelet service")
		if err := initSystem.ServiceStop("kubelet"); err != nil {
			klog.Warningf("[reset] The kubelet service could not be stopped by kubeadm: [%v]\n", err)
			klog.Warningln("[reset] Please ensure kubelet is stopped manually")
		}
	}

	// Try to unmount mounted directories under kubeadmconstants.KubeletRunDirectory in order to be able to remove the kubeadmconstants.KubeletRunDirectory directory later
	fmt.Printf("[reset] Unmounting mounted directories in %q\n", kubeadmconstants.KubeletRunDirectory)
	umountDirsCmd := fmt.Sprintf("awk '$2 ~ path {print $2}' path=%s/ /proc/mounts | xargs -r umount", kubeadmconstants.KubeletRunDirectory)

	klog.V(1).Infof("[reset] Executing command %q", umountDirsCmd)
	umountOutputBytes, err := exec.Command("sh", "-c", umountDirsCmd).Output()
	if err != nil {
		klog.Errorf("[reset] Failed to unmount mounted directories in %s: %s\n", kubeadmconstants.KubeletRunDirectory, string(umountOutputBytes))
	}

	klog.V(1).Info("[reset] Removing Kubernetes-managed containers")
	if err := removeContainers(utilsexec.New(), r.criSocketPath); err != nil {
		klog.Errorf("[reset] Failed to remove containers: %v", err)
	}

	dirsToClean = append(dirsToClean, []string{kubeadmconstants.KubeletRunDirectory, "/etc/cni/net.d", "/var/lib/dockershim", "/var/run/kubernetes"}...)

	// Then clean contents from the stateful kubelet, etcd and cni directories
	fmt.Printf("[reset] Deleting contents of stateful directories: %v\n", dirsToClean)
	for _, dir := range dirsToClean {
		klog.V(1).Infof("[reset] Deleting content of %s", dir)
		cleanDir(dir)
	}

	// Remove contents from the config and pki directories
	klog.V(1).Infoln("[reset] Removing contents from the config and pki directories")
	if r.certsDir != kubeadmapiv1beta2.DefaultCertificatesDir {
		klog.Warningf("[reset] WARNING: Cleaning a non-default certificates directory: %q\n", r.certsDir)
	}
	resetConfigDir(kubeadmconstants.KubernetesDir, r.certsDir)

	// Output help text instructing user how to remove iptables rules
	msg := dedent.Dedent(`
		The reset process does not reset or clean up iptables rules or IPVS tables.
		If you wish to reset iptables, you must do so manually.
		For example:
		iptables -F && iptables -t nat -F && iptables -t mangle -F && iptables -X

		If your cluster was setup to utilize IPVS, run ipvsadm --clear (or similar)
		to reset your system's IPVS tables.
		
		The reset process does not clean your kubeconfig files and you must remove them manually.
		Please, check the contents of the $HOME/.kube/config file.
	`)
	fmt.Print(msg)

	return nil
}
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
}
1068337
func (u *DropsondeUnmarshallerCollection) Run(inputChan <-chan []byte, outputChan chan<- *events.Envelope, waitGroup *sync.WaitGroup) {
	for _, unmarshaller := range u.unmarshallers {
		go func(um *DropsondeUnmarshaller) {
			defer waitGroup.Done()
			um.Run(inputChan, outputChan)
		}(unmarshaller)
	}
}
1346561
func (ar *allocRunner) Run() {
	// Close the wait channel on return
	defer close(ar.waitCh)

	// Start the task state update handler
	go ar.handleTaskStateUpdates()

	// Start the alloc update handler
	go ar.handleAllocUpdates()

	// If task update chan has been closed, that means we've been shutdown.
	select {
	case <-ar.taskStateUpdateHandlerCh:
		return
	default:
	}

	// Run the prestart hooks if non-terminal
	if ar.shouldRun() {
		if err := ar.prerun(); err != nil {
			ar.logger.Error("prerun failed", "error", err)
			goto POST
		}
	}

	// Run the runners (blocks until they exit)
	ar.runTasks()

POST:
	// Run the postrun hooks
	if err := ar.postrun(); err != nil {
		ar.logger.Error("postrun failed", "error", err)
	}

}
860771
func (p *EvaluatePool) run(stopCh chan struct{}) {
	for {
		select {
		case req := <-p.req:
			fit, reason, err := evaluateNodePlan(req.snap, req.plan, req.nodeID)
			p.res <- evaluateResult{req.nodeID, fit, reason, err}

		case <-stopCh:
			return
		}
	}
}