// waitForPod watches the pod it until it finishes and send all events on the
// pod to the PV.
495945
func waitForPod(pod *v1.Pod, recyclerClient recyclerClient, podCh <-chan watch.Event) error {
	for {
		event, ok := <-podCh
		if !ok {
			return fmt.Errorf("recycler pod %q watch channel had been closed", pod.Name)
		}
		switch event.Object.(type) {
		case *v1.Pod:
			// POD changed
			pod := event.Object.(*v1.Pod)
			klog.V(4).Infof("recycler pod update received: %s %s/%s %s", event.Type, pod.Namespace, pod.Name, pod.Status.Phase)
			switch event.Type {
			case watch.Added, watch.Modified:
				if pod.Status.Phase == v1.PodSucceeded {
					// Recycle succeeded.
					return nil
				}
				if pod.Status.Phase == v1.PodFailed {
					if pod.Status.Message != "" {
						return fmt.Errorf(pod.Status.Message)
					}
					return fmt.Errorf("pod failed, pod.Status.Message unknown")
				}

			case watch.Deleted:
				return fmt.Errorf("recycler pod was deleted")

			case watch.Error:
				return fmt.Errorf("recycler pod watcher failed")
			}

		case *v1.Event:
			// Event received
			podEvent := event.Object.(*v1.Event)
			klog.V(4).Infof("recycler event received: %s %s/%s %s/%s %s", event.Type, podEvent.Namespace, podEvent.Name, podEvent.InvolvedObject.Namespace, podEvent.InvolvedObject.Name, podEvent.Message)
			if event.Type == watch.Added {
				recyclerClient.Event(podEvent.Type, podEvent.Message)
			}
		}
	}
}
601544
func WaitForRunningDeployerPod(podClient corev1client.PodsGetter, rc *corev1.ReplicationController, timeout time.Duration) error {
	podName := appsutil.DeployerPodNameForDeployment(rc.Name)
	canGetLogs := func(p *corev1.Pod) bool {
		return corev1.PodSucceeded == p.Status.Phase || corev1.PodFailed == p.Status.Phase || corev1.PodRunning == p.Status.Phase
	}

	fieldSelector := fields.OneTermEqualSelector("metadata.name", podName).String()
	lw := &cache.ListWatch{
		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
			options.FieldSelector = fieldSelector
			return podClient.Pods(rc.Namespace).List(options)
		},
		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
			options.FieldSelector = fieldSelector
			return podClient.Pods(rc.Namespace).Watch(options)
		},
	}

	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	_, err := watchtools.UntilWithSync(ctx, lw, &corev1.Pod{}, nil, func(e watch.Event) (bool, error) {
		switch e.Type {
		case watch.Added, watch.Modified:
			newPod, ok := e.Object.(*corev1.Pod)
			if !ok {
				return true, fmt.Errorf("unknown event object %#v", e.Object)
			}

			return canGetLogs(newPod), nil

		case watch.Deleted:
			return true, fmt.Errorf("pod got deleted %#v", e.Object)

		case watch.Error:
			return true, fmt.Errorf("encountered error while watching for pod: %v", e.Object)

		default:
			return true, fmt.Errorf("unexpected event type: %T", e.Type)
		}
	})
	return err
}
122286
func (c *Conn) WaitForNotification(ctx context.Context) (notification *Notification, err error) {
	// Return already received notification immediately
	if len(c.notifications) > 0 {
		notification := c.notifications[0]
		c.notifications = c.notifications[1:]
		return notification, nil
	}

	err = c.waitForPreviousCancelQuery(ctx)
	if err != nil {
		return nil, err
	}

	err = c.initContext(ctx)
	if err != nil {
		return nil, err
	}
	defer func() {
		err = c.termContext(err)
	}()

	if err = c.lock(); err != nil {
		return nil, err
	}
	defer func() {
		if unlockErr := c.unlock(); unlockErr != nil && err == nil {
			err = unlockErr
		}
	}()

	if err := c.ensureConnectionReadyForQuery(); err != nil {
		return nil, err
	}

	for {
		msg, err := c.rxMsg()
		if err != nil {
			return nil, err
		}

		err = c.processContextFreeMsg(msg)
		if err != nil {
			return nil, err
		}

		if len(c.notifications) > 0 {
			notification := c.notifications[0]
			c.notifications = c.notifications[1:]
			return notification, nil
		}
	}
}
1004806
func (p *Pod) WaitFinished(ctx context.Context) error {
	f := func() bool {
		switch err := p.TrySharedLock(); err {
		case nil:
			// the pod is now locked successfully, hence one of the running phases passed.
			// continue with unlocking the pod immediately below.
		case lock.ErrLocked:
			// pod is still locked, hence we are still in a running phase.
			// i.e. in pepare, run, exitedGarbage, garbage state.
			return false
		default:
			// some error occurred, bail out.
			return true
		}

		// unlock immediately
		if err := p.Unlock(); err != nil {
			return true
		}

		if err := p.refreshState(); err != nil {
			return true
		}

		// if we're in the gap between preparing and running in a split prepare/run-prepared usage, take a nap
		if p.isPrepared {
			time.Sleep(time.Second)
		}

		return p.IsFinished()
	}

	return retry(ctx, f, 100*time.Millisecond)
}
67665
func WaitForPodsWithLabelRunning(c kubernetes.Interface, ns string, label labels.Selector) error {
	glog.Infof("Waiting for pod with label %q in ns %q ...", ns, label)
	lastKnownPodNumber := -1
	return wait.PollImmediate(constants.APICallRetryInterval, ReasonableStartTime, func() (bool, error) {
		listOpts := metav1.ListOptions{LabelSelector: label.String()}
		pods, err := c.CoreV1().Pods(ns).List(listOpts)
		if err != nil {
			glog.Infof("error getting Pods with label selector %q [%v]\n", label.String(), err)
			return false, nil
		}

		if lastKnownPodNumber != len(pods.Items) {
			glog.Infof("Found %d Pods for label selector %s\n", len(pods.Items), label.String())
			lastKnownPodNumber = len(pods.Items)
		}

		if len(pods.Items) == 0 {
			return false, nil
		}

		for _, pod := range pods.Items {
			if pod.Status.Phase != v1.PodRunning {
				return false, nil
			}
		}

		return true, nil
	})
}
983069
func (w *Waiter) WaitForStaticPodControlPlaneHashes(_ string) (map[string]string, error) {
	return map[string]string{
		constants.KubeAPIServer:         "",
		constants.KubeControllerManager: "",
		constants.KubeScheduler:         "",
	}, nil
}
670862
func (we *WorkflowExecutor) waitMainContainerStart() (string, error) {
	for {
		podsIf := we.ClientSet.CoreV1().Pods(we.Namespace)
		fieldSelector := fields.ParseSelectorOrDie(fmt.Sprintf("metadata.name=%s", we.PodName))
		opts := metav1.ListOptions{
			FieldSelector: fieldSelector.String(),
		}
		watchIf, err := podsIf.Watch(opts)
		if err != nil {
			return "", errors.InternalWrapErrorf(err, "Failed to establish pod watch: %v", err)
		}
		for watchEv := range watchIf.ResultChan() {
			if watchEv.Type == watch.Error {
				return "", errors.InternalErrorf("Pod watch error waiting for main to start: %v", watchEv.Object)
			}
			pod, ok := watchEv.Object.(*apiv1.Pod)
			if !ok {
				log.Warnf("Pod watch returned non pod object: %v", watchEv.Object)
				continue
			}
			for _, ctrStatus := range pod.Status.ContainerStatuses {
				if ctrStatus.Name == common.MainContainerName {
					log.Debug(ctrStatus)
					if ctrStatus.ContainerID != "" {
						we.mainContainerID = containerID(ctrStatus.ContainerID)
						return containerID(ctrStatus.ContainerID), nil
					} else if ctrStatus.State.Waiting == nil && ctrStatus.State.Running == nil && ctrStatus.State.Terminated == nil {
						// status still not ready, wait
					} else if ctrStatus.State.Waiting != nil {
						// main container is still in waiting status
					} else {
						// main container in running or terminated state but missing container ID
						return "", errors.InternalError("Main container ID cannot be found")
					}
				}
			}
		}
		log.Warnf("Pod watch closed unexpectedly")
	}
}
661220
func WaitForListen(u *url.URL, timeout time.Duration) error {
	timedout := time.After(timeout)

	b := backoff.Backoff{
		Min:    100 * time.Millisecond,
		Max:    time.Second,
		Factor: 1.5,
	}

	for {
		select {
		case <-timedout:
			return errors.Errorf("timed out after %s", timeout)
		case <-time.After(b.Duration()):
			if IsListening(u) {
				return nil
			}
		}
	}
}
363942
func (client *Client) WaitForListener(loadBalancerId string, port int, listenerType ListenerType) (status ListenerStatus, err error) {
	timeout := DefaultTimeout

	args := &CommonLoadBalancerListenerArgs{
		LoadBalancerId: loadBalancerId,
		ListenerPort:   port,
	}

	method := fmt.Sprintf("DescribeLoadBalancer%sListenerAttribute", listenerType)
	response := &DescribeLoadBalancerListenerAttributeResponse{}

	for {
		timeout = timeout - DefaultWaitForInterval
		if timeout <= 0 {
			return response.Status, common.GetClientErrorFromString("Timeout")
		}
		time.Sleep(DefaultWaitForInterval * time.Second)
		//Sleep first to ensure the previous request is sent
		err = client.Invoke(method, args, response)
		if err != nil {
			return "", err
		}
		if response.Status == Running || response.Status == Stopped {
			break
		}
	}
	return response.Status, nil
}
774152
func WaitForStart(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
	if ok, missing := p.RequiresValue("client"); !ok {
		return nil, &cookoo.FatalError{"Missing required fields: " + strings.Join(missing, ", ")}
	}
	cli := p.Get("client", nil).(*docli.Client)
	timeout := p.Get("timeout", 30*time.Second).(time.Duration)

	keepon := true
	timer := time.AfterFunc(timeout, func() {
		keepon = false
	})

	for keepon == true {
		if err := cli.Ping(); err == nil {
			timer.Stop()
			log.Infof(c, "Docker is running.")
			return true, nil
		}
		time.Sleep(time.Second)
	}
	return false, fmt.Errorf("Docker timed out after waiting %s for server.", timeout)
}
742170
func (w *Waiter) WaitForStaticPodSingleHash(_ string, _ string) (string, error) {
	return "", nil
}
1344678
func (c *Client) WaitAndGetCompletedPodPhase(namespace string, reader io.Reader, timeout time.Duration) (v1.PodPhase, error) {
	infos, err := c.Build(namespace, reader)
	if err != nil {
		return v1.PodUnknown, err
	}
	info := infos[0]

	kind := info.Mapping.GroupVersionKind.Kind
	if kind != "Pod" {
		return v1.PodUnknown, fmt.Errorf("%s is not a Pod", info.Name)
	}

	if err := c.watchPodUntilComplete(timeout, info); err != nil {
		return v1.PodUnknown, err
	}

	if err := info.Get(); err != nil {
		return v1.PodUnknown, err
	}
	status := info.Object.(*v1.Pod).Status.Phase

	return status, nil
}
138090
func (w *Waiter) WaitForPodsWithLabel(kvLabel string) error {
	fmt.Printf("[dryrun] Would wait for the Pods with the label %q in the %s namespace to become Running\n", kvLabel, metav1.NamespaceSystem)
	return nil
}
44263
func (pgCtrl *podGroupController) waitLastPodHealth(instanceNo int) bool {
	maxRetries := 5
	retryTimes := 0
	if instanceNo > 1 {
		sleepTime := DefaultSetUpTime
		podSpec := pgCtrl.spec.Pod
		if podSpec.GetSetupTime() > DefaultSetUpTime {
			sleepTime = podSpec.GetSetupTime()
		}
		lastPodCtrl := pgCtrl.podCtrls[instanceNo-2]
		// wait some seconds for new instance's initialization completed, before we update next one
		if lastPodCtrl.pod.Healthst == HealthStateNone {
			time.Sleep(time.Second * time.Duration(podSpec.GetSetupTime()))
		} else {
			tick := time.Tick(time.Duration(sleepTime) * time.Second)
		Loop:
			for {
				select {
				case <-tick:
					retryTimes++
					// wait until to healthy state
					lastPodCtrl.Refresh(pgCtrl.engine.cluster)
					log.Infof("emit with loop :%v", lastPodCtrl.pod.Healthst)
					if lastPodCtrl.pod.Healthst == HealthStateHealthy {
						break Loop
					}
				case <-lastPodCtrl.event:
					log.Infof("emit with event:%v", lastPodCtrl.pod.Healthst)
					if lastPodCtrl.pod.Healthst == HealthStateHealthy {
						break Loop
					}
				}
				if retryTimes >= maxRetries {
					break
				}
			}
		}
	}
	// reset health state to starting
	if pgCtrl.podCtrls[instanceNo-1].pod.Healthst != HealthStateNone {
		pgCtrl.podCtrls[instanceNo-1].pod.Healthst = HealthStateStarting
	}
	if retryTimes >= maxRetries && pgCtrl.podCtrls[instanceNo-2].pod.Healthst != HealthStateHealthy {
		return false
	}
	return true
}
316868
func (s *SnapshotCreateService) WaitForCompletion(waitForCompletion bool) *SnapshotCreateService {
	s.waitForCompletion = &waitForCompletion
	return s
}
1079918
func (c *Ctx) WaitForSlotEvent(flags uint) chan SlotEvent {
	sl := make(chan SlotEvent, 1) // hold one element
	go c.waitForSlotEventHelper(flags, sl)
	return sl
}
29625
func (wi *Instance) WaitForCommand(wrk Worker, done chan struct{}) error {
	// display the status every second
	timer := time.NewTicker(wi.commandDisplayInterval)
	defer timer.Stop()
	for {
		select {
		case <-done:
			log.Info(wrk.StatusAsText())
			wi.currentWorkerMutex.Lock()
			err := wi.lastRunError
			wi.currentWorkerMutex.Unlock()
			if err != nil {
				return err
			}
			return nil
		case <-timer.C:
			log.Info(wrk.StatusAsText())
		}
	}
}