// MakeNextPodFunc returns a function to retrieve the next pod from a given
// scheduling queue
430789
func MakeNextPodFunc(queue SchedulingQueue) func() *v1.Pod {
	return func() *v1.Pod {
		pod, err := queue.Pop()
		if err == nil {
			klog.V(4).Infof("About to try and schedule pod %v/%v", pod.Namespace, pod.Name)
			return pod
		}
		klog.Errorf("Error while retrieving next pod from scheduling queue: %v", err)
		return nil
	}
}
901055
func podMatchesScopeFunc(selector corev1.ScopedResourceSelectorRequirement, object runtime.Object) (bool, error) {
	pod, err := toExternalPodOrError(object)
	if err != nil {
		return false, err
	}
	switch selector.ScopeName {
	case corev1.ResourceQuotaScopeTerminating:
		return isTerminating(pod), nil
	case corev1.ResourceQuotaScopeNotTerminating:
		return !isTerminating(pod), nil
	case corev1.ResourceQuotaScopeBestEffort:
		return isBestEffort(pod), nil
	case corev1.ResourceQuotaScopeNotBestEffort:
		return !isBestEffort(pod), nil
	case corev1.ResourceQuotaScopePriorityClass:
		return podMatchesSelector(pod, selector)
	}
	return false, nil
}
342327
func PodMutateLimitFunc(limitRange *corev1.LimitRange, pod *api.Pod) error {
	defaultResources := defaultContainerResourceRequirements(limitRange)
	mergePodResourceRequirements(pod, &defaultResources)
	return nil
}
758669
func (pcond *ProxyConds) DoFunc(f func(resp *http.Response, ctx *ProxyCtx) *http.Response) {
	pcond.Do(FuncRespHandler(f))
}
195382
func (fc *FakeClock) AfterFunc(d time.Duration, f func()) Timer {
	return fc.FakeAfterFunc(d, f)
}
10712
func makeWalkFunc(router *mux.Router) mux.WalkFunc {
	return mux.WalkFunc(func(r *mux.Route, _ *mux.Router, _ []*mux.Route) error {
		// Does a route with this name exist in router?
		route := router.Get(r.GetName())
		if route == nil {
			return fmt.Errorf("no route by name %q in router", r.GetName())
		}
		// Does the route have a handler?
		handler := route.GetHandler()
		if handler == nil {
			return fmt.Errorf("no handler for route %q in router", r.GetName())
		}
		return nil
	})
}
833229
func newUnschedulablePodsMap(metricRecorder metrics.MetricRecorder) *UnschedulablePodsMap {
	return &UnschedulablePodsMap{
		podInfoMap:     make(map[string]*podInfo),
		keyFunc:        util.GetPodFullName,
		metricRecorder: metricRecorder,
	}
}
99276
func (m *MockClock) AfterFunc(arg0 time.Duration, arg1 func()) clock.Timer {
	ret := m.ctrl.Call(m, "AfterFunc", arg0, arg1)
	ret0, _ := ret[0].(clock.Timer)
	return ret0
}
1138254
func (m *MockClock) AfterFunc(d time.Duration, f func()) Timer {
	t := m.NewTimer(d)
	go func() {
		<-t.Chan()
		go f()
	}()

	return t
}
692791
func (s *Selection) NextUntil(selector string) *Selection {
	return pushStack(s, getSiblingNodes(s.Nodes, siblingNextUntil,
		compileMatcher(selector), nil))
}
1303155
func (stm *StoreTaskMeta) createNextRunFromQueue(now, nextDue int64, sch cron.Schedule, makeID func() (platform.ID, error)) (RunCreation, error) {
	if len(stm.ManualRuns) == 0 {
		return RunCreation{}, errors.New("cannot create run from empty queue")
	}

	q := stm.ManualRuns[0]
	latest := q.LatestCompleted
	for _, r := range stm.CurrentlyRunning {
		if r.RangeStart != q.Start || r.RangeEnd != q.End || r.RequestedAt != q.RequestedAt {
			// Doesn't match our queue.
			continue
		}
		if r.Now > latest {
			latest = r.Now
		}
	}

	runNow := sch.Next(time.Unix(latest, 0)).Unix()

	// Already validated that we have room to create another run, in CreateNextRun.
	id := platform.ID(q.RunID)

	if !id.Valid() {
		var err error
		id, err = makeID()
		if err != nil {
			return RunCreation{}, err
		}
	}

	stm.CurrentlyRunning = append(stm.CurrentlyRunning, &StoreTaskMetaRun{
		Now:   runNow,
		Try:   1,
		RunID: uint64(id),

		RangeStart:  q.Start,
		RangeEnd:    q.End,
		RequestedAt: q.RequestedAt,
	})

	if runNow >= q.End {
		// Drop the queue.
		stm.ManualRuns = append(stm.ManualRuns[:0], stm.ManualRuns[1:]...)
	}

	return RunCreation{
		Created: QueuedRun{
			RunID:       id,
			Now:         runNow,
			RequestedAt: q.RequestedAt,
		},
		NextDue:  nextDue,
		HasQueue: len(stm.ManualRuns) > 0,
	}, nil
}
837350
func (k *Kite) FinalFunc(f FinalFunc) {
	k.finalFuncs = append(k.finalFuncs, f)
}
627593
func GetFirstPod(client coreclient.PodsGetter, namespace string, selector string, timeout time.Duration, sortBy func([]*corev1.Pod) sort.Interface) (*corev1.Pod, int, error) {
	options := metav1.ListOptions{LabelSelector: selector}

	podList, err := client.Pods(namespace).List(options)
	if err != nil {
		return nil, 0, err
	}
	pods := []*corev1.Pod{}
	for i := range podList.Items {
		pod := podList.Items[i]
		pods = append(pods, &pod)
	}
	if len(pods) > 0 {
		sort.Sort(sortBy(pods))
		return pods[0], len(podList.Items), nil
	}

	// Watch until we observe a pod
	options.ResourceVersion = podList.ResourceVersion
	w, err := client.Pods(namespace).Watch(options)
	if err != nil {
		return nil, 0, err
	}
	defer w.Stop()

	condition := func(event watch.Event) (bool, error) {
		return event.Type == watch.Added || event.Type == watch.Modified, nil
	}

	ctx, cancel := watchtools.ContextWithOptionalTimeout(context.Background(), timeout)
	defer cancel()
	event, err := watchtools.UntilWithoutRetry(ctx, w, condition)
	if err != nil {
		return nil, 0, err
	}
	pod, ok := event.Object.(*corev1.Pod)
	if !ok {
		return nil, 0, fmt.Errorf("%#v is not a pod event", event)
	}
	return pod, 1, nil
}
317084
func NewInterPodAffinityPriority(
	info predicates.NodeInfo,
	nodeLister algorithm.NodeLister,
	podLister algorithm.PodLister,
	hardPodAffinityWeight int32) PriorityFunction {
	interPodAffinity := &InterPodAffinity{
		info:                  info,
		nodeLister:            nodeLister,
		podLister:             podLister,
		hardPodAffinityWeight: hardPodAffinityWeight,
	}
	return interPodAffinity.CalculateInterPodAffinityPriority
}
796824
func (p *PriorityQueue) NumUnschedulablePods() int {
	p.lock.RLock()
	defer p.lock.RUnlock()
	return len(p.unschedulableQ.podInfoMap)
}
777767
func (wfc *WorkflowController) processNextPodItem() bool {
	key, quit := wfc.podQueue.Get()
	if quit {
		return false
	}
	defer wfc.podQueue.Done(key)

	obj, exists, err := wfc.podInformer.GetIndexer().GetByKey(key.(string))
	if err != nil {
		log.Errorf("Failed to get pod '%s' from informer index: %+v", key, err)
		return true
	}
	if !exists {
		// we can get here if pod was queued into the pod workqueue,
		// but it was either deleted or labeled completed by the time
		// we dequeued it.
		return true
	}
	pod, ok := obj.(*apiv1.Pod)
	if !ok {
		log.Warnf("Key '%s' in index is not a pod", key)
		return true
	}
	if pod.Labels == nil {
		log.Warnf("Pod '%s' did not have labels", key)
		return true
	}
	workflowName, ok := pod.Labels[common.LabelKeyWorkflow]
	if !ok {
		// Ignore pods unrelated to workflow (this shouldn't happen unless the watch is setup incorrectly)
		log.Warnf("watch returned pod unrelated to any workflow: %s", pod.ObjectMeta.Name)
		return true
	}
	// TODO: currently we reawaken the workflow on *any* pod updates.
	// But this could be be much improved to become smarter by only
	// requeue the workflow when there are changes that we care about.
	wfc.wfQueue.Add(pod.ObjectMeta.Namespace + "/" + workflowName)
	return true
}
568857
func GetUniqueVolumeNameFromSpecWithPod(
	podName types.UniquePodName, volumePlugin volume.VolumePlugin, volumeSpec *volume.Spec) v1.UniqueVolumeName {
	return v1.UniqueVolumeName(
		fmt.Sprintf("%s/%v-%s", volumePlugin.GetPluginName(), podName, volumeSpec.Name()))
}