in pkg/capacityscheduling/capacity_scheduling.go [433:612]
func selectVictimsOnNode(
ctx context.Context,
fh framework.Handle,
state *framework.CycleState,
pod *v1.Pod,
nodeInfo *framework.NodeInfo,
pdbs []*policy.PodDisruptionBudget,
) ([]*v1.Pod, int, *framework.Status) {
elasticQuotaSnapshotState, err := getElasticQuotaSnapshotState(state)
if err != nil {
msg := fmt.Sprintf("error reading %q from cycleState: %v", ElasticQuotaSnapshotKey, err)
klog.Errorf(msg)
return nil, 0, framework.NewStatus(framework.Unschedulable, msg)
}
preFilterState, err := getPreFilterState(state)
if err != nil {
msg := fmt.Sprintf("error reading %q from cycleState: %v", preFilterStateKey, err)
klog.Errorf(msg)
return nil, 0, framework.NewStatus(framework.Unschedulable, msg)
}
removePod := func(rpi *framework.PodInfo) error {
if err := nodeInfo.RemovePod(rpi.Pod); err != nil {
return err
}
status := fh.RunPreFilterExtensionRemovePod(ctx, state, pod, rpi, nodeInfo)
if !status.IsSuccess() {
return status.AsError()
}
return nil
}
addPod := func(api *framework.PodInfo) error {
nodeInfo.AddPodInfo(api)
status := fh.RunPreFilterExtensionAddPod(ctx, state, pod, api, nodeInfo)
if !status.IsSuccess() {
return status.AsError()
}
return nil
}
elasticQuotaInfos := elasticQuotaSnapshotState.elasticQuotaInfos
podPriority := corev1helpers.PodPriority(pod)
preemptorElasticQuotaInfo, preemptorWithElasticQuota := elasticQuotaInfos[pod.Namespace]
var moreThanMinWithPreemptor bool
// Check if there is elastic quota in the preemptor's namespace.
if preemptorWithElasticQuota {
moreThanMinWithPreemptor = preemptorElasticQuotaInfo.overUsed(preFilterState.Resource, preemptorElasticQuotaInfo.Min)
}
// sort the pods in node by the priority class
sort.Slice(nodeInfo.Pods, func(i, j int) bool { return !util.MoreImportantPod(nodeInfo.Pods[i].Pod, nodeInfo.Pods[j].Pod) })
var potentialVictims []*framework.PodInfo
if preemptorWithElasticQuota {
for _, p := range nodeInfo.Pods {
pElasticQuotaInfo, pWithElasticQuota := elasticQuotaInfos[p.Pod.Namespace]
if !pWithElasticQuota {
continue
}
if moreThanMinWithPreemptor {
// If Preemptor.Request + Quota.Used > Quota.Min:
// It means that its guaranteed isn't borrowed by other
// quotas. So that we will select the pods which subject to the
// same quota(namespace) with the lower priority than the
// preemptor's priority as potential victims in a node.
if p.Pod.Namespace == pod.Namespace && corev1helpers.PodPriority(p.Pod) < podPriority {
potentialVictims = append(potentialVictims, p)
if err := removePod(p); err != nil {
return nil, 0, framework.AsStatus(err)
}
}
} else {
// If Preemptor.Request + Quota.allocated <= Quota.min: It
// means that its min(guaranteed) resource is used or
// `borrowed` by other Quota. Potential victims in a node
// will be chosen from Quotas that allocates more resources
// than its min, i.e., borrowing resources from other
// Quotas.
if p.Pod.Namespace != pod.Namespace && moreThanMin(*pElasticQuotaInfo.Used, *pElasticQuotaInfo.Min) {
potentialVictims = append(potentialVictims, p)
if err := removePod(p); err != nil {
return nil, 0, framework.AsStatus(err)
}
}
}
}
} else {
for _, p := range nodeInfo.Pods {
_, pWithElasticQuota := elasticQuotaInfos[p.Pod.Namespace]
if pWithElasticQuota {
continue
}
if corev1helpers.PodPriority(p.Pod) < podPriority {
potentialVictims = append(potentialVictims, p)
if err := removePod(p); err != nil {
return nil, 0, framework.AsStatus(err)
}
}
}
}
// No potential victims are found, and so we don't need to evaluate the node again since its state didn't change.
if len(potentialVictims) == 0 {
message := fmt.Sprintf("No victims found on node %v for preemptor pod %v", nodeInfo.Node().Name, pod.Name)
return nil, 0, framework.NewStatus(framework.UnschedulableAndUnresolvable, message)
}
// If the new pod does not fit after removing all the lower priority pods,
// we are almost done and this node is not suitable for preemption. The only
// condition that we could check is if the "pod" is failing to schedule due to
// inter-pod affinity to one or more victims, but we have decided not to
// support this case for performance reasons. Having affinity to lower
// priority pods is not a recommended configuration anyway.
if s := fh.RunFilterPluginsWithNominatedPods(ctx, state, pod, nodeInfo); !s.IsSuccess() {
return nil, 0, s
}
// If the quota.used + pod.request > quota.max or sum(quotas.used) + pod.request > sum(quotas.min)
// after removing all the lower priority pods,
// we are almost done and this node is not suitable for preemption.
if preemptorWithElasticQuota {
if preemptorElasticQuotaInfo.overUsed(preFilterState.Resource, preemptorElasticQuotaInfo.Max) ||
elasticQuotaInfos.aggregatedMinOverUsedWithPod(preFilterState.Resource) {
return nil, 0, framework.NewStatus(framework.Unschedulable, "global quota max exceeded")
}
}
var victims []*v1.Pod
numViolatingVictim := 0
sort.Slice(potentialVictims, func(i, j int) bool { return util.MoreImportantPod(potentialVictims[i].Pod, potentialVictims[j].Pod) })
// Try to reprieve as many pods as possible. We first try to reprieve the PDB
// violating victims and then other non-violating ones. In both cases, we start
// from the highest priority victims.
violatingVictims, nonViolatingVictims := filterPodsWithPDBViolation(potentialVictims, pdbs)
reprievePod := func(pi *framework.PodInfo) (bool, error) {
p := pi.Pod
if err := addPod(pi); err != nil {
return false, err
}
s := fh.RunFilterPluginsWithNominatedPods(ctx, state, pod, nodeInfo)
fits := s.IsSuccess()
if !fits {
if err := removePod(pi); err != nil {
return false, err
}
victims = append(victims, p)
klog.V(5).Infof("Pod %v/%v is a potential preemption victim on node %v.", p.Namespace, p.Name, nodeInfo.Node().Name)
}
if preemptorWithElasticQuota && (preemptorElasticQuotaInfo.overUsed(preFilterState.Resource, preemptorElasticQuotaInfo.Max) || elasticQuotaInfos.aggregatedMinOverUsedWithPod(preFilterState.Resource)) {
if err := removePod(pi); err != nil {
return false, err
}
victims = append(victims, p)
klog.V(5).Infof("Pod %v/%v is a potential preemption victim on node %v.", p.Namespace, p.Name, nodeInfo.Node().Name)
}
return fits, nil
}
for _, pi := range violatingVictims {
if fits, err := reprievePod(pi); err != nil {
klog.Warningf("Failed to reprieve pod %q: %v", pi.Pod.Name, err)
return nil, 0, framework.AsStatus(err)
} else if !fits {
numViolatingVictim++
}
}
// Now we try to reprieve non-violating victims.
for _, pi := range nonViolatingVictims {
if _, err := reprievePod(pi); err != nil {
klog.Warningf("Failed to reprieve pod %q: %v", pi.Pod.Name, err)
return nil, 0, framework.AsStatus(err)
}
}
return victims, numViolatingVictim, framework.NewStatus(framework.Success)
}