Method scaleInWorkerQueue

pkg/workerpool/worker_pool.go:300–356

()

}

func (w *WorkerPool) scaleInWorkerQueue() {
	workers, err := w.podLister.List(labels.Set{constants.WorkerOwnerLabelKey: ""}.AsSelector())
	if err != nil {
		klog.ErrorS(err, "error when listing pod from informer cache")
	}

	idelWorkers := []*corev1.Pod{}
	for _, pod := range workers {
		// the tolerance time of not available pod is 30 minutes, if
		// time out, we need to delete these pods.
		if util.IsPodNotAvailable(pod, 30*60) {
			if err := w.deleteWorker(pod); err != nil {
				klog.ErrorS(err, "failed to delete worker that is not available for 30 minute", klog.KObjs(pod))
			}
		}

		if util.IsPodReady(pod) {
			if req := w.matchRequestFor(pod); req == nil {
				idelWorkers = append(idelWorkers, pod)
			}
		}
	}

	if len(idelWorkers) > w.coreWorkerLimit {
		// Sort by number of bindings first, and then by chronological
		// order of creation if they are equal.
		sort.Slice(idelWorkers, func(i, j int) bool {
			one, two := idelWorkers[i], idelWorkers[j]

			c1 := one.Labels[constants.WorkerBindingCountLabelKey]
			c2 := two.Labels[constants.WorkerBindingCountLabelKey]
			n1, _ := strconv.ParseInt(c1, 10, 32)
			n2, _ := strconv.ParseInt(c2, 10, 32)

			if n1 == n2 {
				return one.CreationTimestamp.Before(&two.CreationTimestamp)
			}

			return n1 < n2
		})

		scaleNumber := float64(len(idelWorkers)-w.coreWorkerLimit) * ScaleInQueueThreshold
		l := int(math.Floor(scaleNumber + 0.5))

		for i := 0; i < l; i++ {
			worker := idelWorkers[i]
			if err := w.deleteWorker(worker); err != nil {
				klog.ErrorS(err, "failed to delete worker that needs to be scaled in", klog.KObjs(worker))
			} else {
				key, _ := WorkerKeyFunc(worker)
				w.workerQueue.Remove(key)
			}
		}
	}

	klog.V(2).Infof("the worker pool size is %d", w.workerQueue.Len())
}