Repository: kubernetes-sigs/kueue

Issue:
[FairSharing] Preemption algorithm is non-deterministic leading to excessive preemption
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!

If the matter is security related, please disclose it privately via https://kubernetes.io/security/
-->


**What happened**:
When there are two CQ that have equal DRS, FairSharing preemption algorithm picks one randomly ([based on the hash function in a set](https://github.com/kubernetes-sigs/kueue/blob/main/pkg/cache/hierarchy/cohort.go#L45)). Since the algorithm is run over and over for the same preemptor until preemption actually executes this leads to excessive preemption, where a preemptor Workload preempts more Workloads than needed if they come from drawing ClusterQueues

**What you expected to happen**:
Make the preemption algorithm deterministic, ideally Workloads should be ordered by the same less function as across the codebase

**How to reproduce it (as minimally and precisely as possible)**:

**Anything else we need to know?**:

**Environment**:
- Kubernetes version (use `kubectl version`):
- Kueue version (use `git describe --tags --dirty --always`):
- Cloud provider or hardware configuration:
- OS (e.g: `cat /etc/os-release`):
- Kernel (e.g. `uname -a`):
- Install tools:
- Others:

Hints from issue comments:

Solution patch:
diff --git a/pkg/scheduler/preemption/common/ordering.go b/pkg/scheduler/preemption/common/ordering.go
new file mode 100644
index 000000000..3c34ea3d6
--- /dev/null
+++ b/pkg/scheduler/preemption/common/ordering.go
@@ -0,0 +1,88 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package preemptioncommon
+
+import (
+	"time"
+
+	"github.com/go-logr/logr"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/klog/v2"
+
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/util/priority"
+	"sigs.k8s.io/kueue/pkg/workload"
+)
+
+// CandidatesOrdering criteria:
+// 0. Workloads already marked for preemption first.
+// 1. Workloads from other ClusterQueues in the cohort before the ones in the
+// same ClusterQueue as the preemptor.
+// 2. (AdmissionFairSharing only) Workloads with lower LocalQueue's usage first
+// 3. Workloads with lower priority first.
+// 4. Workloads admitted more recently first.
+func CandidatesOrdering(log logr.Logger, afsEnabled bool, a, b *workload.Info, cq kueue.ClusterQueueReference, now time.Time) bool {
+	aEvicted := meta.IsStatusConditionTrue(a.Obj.Status.Conditions, kueue.WorkloadEvicted)
+	bEvicted := meta.IsStatusConditionTrue(b.Obj.Status.Conditions, kueue.WorkloadEvicted)
+	if aEvicted != bEvicted {
+		return aEvicted
+	}
+	aInCQ := a.ClusterQueue == cq
+	bInCQ := b.ClusterQueue == cq
+	if aInCQ != bInCQ {
+		return !aInCQ
+	}
+
+	if afsEnabled && resourceUsagePreemptionEnabled(a, b) {
+		if a.LocalQueueFSUsage != b.LocalQueueFSUsage {
+			log.V(3).Info("Comparing workloads by LocalQueue fair sharing usage",
+				"workloadA", klog.KObj(a.Obj), "queueA", a.Obj.Spec.QueueName, "usageA", a.LocalQueueFSUsage,
+				"workloadB", klog.KObj(b.Obj), "queueB", b.Obj.Spec.QueueName, "usageB", b.LocalQueueFSUsage)
+			return *a.LocalQueueFSUsage > *b.LocalQueueFSUsage
+		}
+	}
+	pa := priority.Priority(a.Obj)
+	pb := priority.Priority(b.Obj)
+	if pa != pb {
+		return pa < pb
+	}
+	timeA := quotaReservationTime(a.Obj, now)
+	timeB := quotaReservationTime(b.Obj, now)
+	if !timeA.Equal(timeB) {
+		return timeA.After(timeB)
+	}
+	// Arbitrary comparison for deterministic sorting.
+	return a.Obj.UID < b.Obj.UID
+}
+
+func resourceUsagePreemptionEnabled(a, b *workload.Info) bool {
+	// If both workloads are in the same ClusterQueue, but different LocalQueues,
+	// we can compare their LocalQueue usage.
+	// If the LocalQueueUsage is not nil for both Workloads, it means the feature gate has been enabled, and the
+	// AdmissionScope of the ClusterQueue is set to UsageBasedFairSharing. We inherit this information from the snapshot initialization.
+	return a.ClusterQueue == b.ClusterQueue && a.Obj.Spec.QueueName != b.Obj.Spec.QueueName && a.LocalQueueFSUsage != nil && b.LocalQueueFSUsage != nil
+}
+
+func quotaReservationTime(wl *kueue.Workload, now time.Time) time.Time {
+	cond := meta.FindStatusCondition(wl.Status.Conditions, kueue.WorkloadQuotaReserved)
+	if cond == nil || cond.Status != metav1.ConditionTrue {
+		// The condition wasn't populated yet, use the current time.
+		return now
+	}
+	return cond.LastTransitionTime.Time
+}
diff --git a/pkg/scheduler/preemption/fairsharing/ordering.go b/pkg/scheduler/preemption/fairsharing/ordering.go
index 9327b94a1..da6558e47 100644
--- a/pkg/scheduler/preemption/fairsharing/ordering.go
+++ b/pkg/scheduler/preemption/fairsharing/ordering.go
@@ -18,11 +18,14 @@ package fairsharing
 
 import (
 	"iter"
+	"time"
 
+	"github.com/go-logr/logr"
 	"k8s.io/apimachinery/pkg/util/sets"
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	schdcache "sigs.k8s.io/kueue/pkg/cache/scheduler"
+	preemptioncommon "sigs.k8s.io/kueue/pkg/scheduler/preemption/common"
 	"sigs.k8s.io/kueue/pkg/workload"
 )
 
@@ -54,9 +57,10 @@ type TargetClusterQueueOrdering struct {
 	// preemption target candidates.
 	prunedClusterQueues sets.Set[*schdcache.ClusterQueueSnapshot]
 	prunedCohorts       sets.Set[*schdcache.CohortSnapshot]
+	log                 logr.Logger
 }
 
-func MakeClusterQueueOrdering(cq *schdcache.ClusterQueueSnapshot, candidates []*workload.Info) TargetClusterQueueOrdering {
+func MakeClusterQueueOrdering(cq *schdcache.ClusterQueueSnapshot, candidates []*workload.Info, log logr.Logger) TargetClusterQueueOrdering {
 	t := TargetClusterQueueOrdering{
 		preemptorCq:        cq,
 		preemptorAncestors: sets.New[*schdcache.CohortSnapshot](),
@@ -65,6 +69,7 @@ func MakeClusterQueueOrdering(cq *schdcache.ClusterQueueSnapshot, candidates []*
 
 		prunedClusterQueues: sets.New[*schdcache.ClusterQueueSnapshot](),
 		prunedCohorts:       sets.New[*schdcache.CohortSnapshot](),
+		log:                 log,
 	}
 
 	for ancestor := range cq.PathParentToRoot() {
@@ -141,9 +146,16 @@ func (t *TargetClusterQueueOrdering) nextTarget(cohort *schdcache.CohortSnapshot
 		drs := cq.DominantResourceShare()
 		// we can't prune the preemptor ClusterQueue itself,
 		// until it runs out of candidates.
-		if (drs == 0 && cq != t.preemptorCq) || !t.hasWorkload(cq) {
+		switch {
+		case (drs == 0 && cq != t.preemptorCq) || !t.hasWorkload(cq):
 			t.prunedClusterQueues.Insert(cq)
-		} else if drs >= highestCqDrs {
+		case drs == highestCqDrs:
+			newCandWl := t.clusterQueueToTarget[cq.GetName()][0]
+			currentCandWl := t.clusterQueueToTarget[highestCq.GetName()][0]
+			if preemptioncommon.CandidatesOrdering(t.log, false, newCandWl, currentCandWl, t.preemptorCq.Name, time.Now()) {
+				highestCq = cq
+			}
+		case drs > highestCqDrs:
 			highestCqDrs = drs
 			highestCq = cq
 		}
diff --git a/pkg/scheduler/preemption/preemption.go b/pkg/scheduler/preemption/preemption.go
index 0500b3a02..89d102f66 100644
--- a/pkg/scheduler/preemption/preemption.go
+++ b/pkg/scheduler/preemption/preemption.go
@@ -21,12 +21,10 @@ import (
 	"fmt"
 	"sort"
 	"sync/atomic"
-	"time"
 
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
@@ -42,6 +40,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/resources"
 	"sigs.k8s.io/kueue/pkg/scheduler/flavorassigner"
 	"sigs.k8s.io/kueue/pkg/scheduler/preemption/classical"
+	preemptioncommon "sigs.k8s.io/kueue/pkg/scheduler/preemption/common"
 	"sigs.k8s.io/kueue/pkg/scheduler/preemption/fairsharing"
 	"sigs.k8s.io/kueue/pkg/util/logging"
 	"sigs.k8s.io/kueue/pkg/util/priority"
@@ -221,7 +220,7 @@ func (p *Preemptor) classicalPreemptions(preemptionCtx *preemptionCtx) []*Target
 		Requests:          preemptionCtx.workloadUsage.Quota,
 		WorkloadOrdering:  p.workloadOrdering,
 	}
-	candidatesGenerator := classical.NewCandidateIterator(hierarchicalReclaimCtx, p.enabledAfs, preemptionCtx.frsNeedPreemption, preemptionCtx.snapshot, p.clock, CandidatesOrdering)
+	candidatesGenerator := classical.NewCandidateIterator(hierarchicalReclaimCtx, p.enabledAfs, preemptionCtx.frsNeedPreemption, preemptionCtx.snapshot, p.clock, preemptioncommon.CandidatesOrdering)
 	var attemptPossibleOpts []preemptionAttemptOpts
 	borrowWithinCohortForbidden, _ := classical.IsBorrowingWithinCohortForbidden(preemptionCtx.preemptorCQ)
 	// We have three types of candidates:
@@ -312,7 +311,8 @@ func parseStrategies(s []config.PreemptionStrategy) []fairsharing.Strategy {
 // and returns (fits, targets, retryCandidates) retryCandidates may be
 // used if rule S2-b is configured.
 func runFirstFsStrategy(preemptionCtx *preemptionCtx, candidates []*workload.Info, strategy fairsharing.Strategy) (bool, []*Target, []*workload.Info) {
-	ordering := fairsharing.MakeClusterQueueOrdering(preemptionCtx.preemptorCQ, candidates)
+	ordering := fairsharing.MakeClusterQueueOrdering(preemptionCtx.preemptorCQ, candidates, preemptionCtx.log)
+
 	var targets []*Target
 	var retryCandidates []*workload.Info
 	for candCQ := range ordering.Iter() {
@@ -357,7 +357,7 @@ func runFirstFsStrategy(preemptionCtx *preemptionCtx, candidates []*workload.Inf
 // runSecondFsStrategy implements Fair Sharing Rule S2-b. It returns
 // (fits, targets).
 func runSecondFsStrategy(retryCandidates []*workload.Info, preemptionCtx *preemptionCtx, targets []*Target) (bool, []*Target) {
-	ordering := fairsharing.MakeClusterQueueOrdering(preemptionCtx.preemptorCQ, retryCandidates)
+	ordering := fairsharing.MakeClusterQueueOrdering(preemptionCtx.preemptorCQ, retryCandidates, preemptionCtx.log)
 	for candCQ := range ordering.Iter() {
 		preemptorNewShare, targetOldShare := candCQ.ComputeShares()
 		// Due to API validation, we can only reach here if the second strategy is LessThanInitialShare,
@@ -387,7 +387,7 @@ func (p *Preemptor) fairPreemptions(preemptionCtx *preemptionCtx, strategies []f
 		return nil
 	}
 	sort.Slice(candidates, func(i, j int) bool {
-		return CandidatesOrdering(preemptionCtx.log, p.enabledAfs, candidates[i], candidates[j], preemptionCtx.preemptorCQ.Name, p.clock.Now())
+		return preemptioncommon.CandidatesOrdering(preemptionCtx.log, p.enabledAfs, candidates[i], candidates[j], preemptionCtx.preemptorCQ.Name, p.clock.Now())
 	})
 	if logV := preemptionCtx.log.V(5); logV.Enabled() {
 		logV.Info("Simulating fair preemption", "candidates", workload.References(candidates), "resourcesRequiringPreemption", preemptionCtx.frsNeedPreemption.UnsortedList(), "preemptingWorkload", klog.KObj(preemptionCtx.preemptor.Obj))
@@ -550,61 +550,3 @@ func queueUnderNominalInResourcesNeedingPreemption(preemptionCtx *preemptionCtx)
 	}
 	return true
 }
-
-func resourceUsagePreemptionEnabled(a, b *workload.Info) bool {
-	// If both workloads are in the same ClusterQueue, but different LocalQueues,
-	// we can compare their LocalQueue usage.
-	// If the LocalQueueUsage is not nil for both Workloads, it means the feature gate has been enabled, and the
-	// AdmissionScope of the ClusterQueue is set to UsageBasedFairSharing. We inherit this information from the snapshot initialization.
-	return a.ClusterQueue == b.ClusterQueue && a.Obj.Spec.QueueName != b.Obj.Spec.QueueName && a.LocalQueueFSUsage != nil && b.LocalQueueFSUsage != nil
-}
-
-// CandidatesOrdering criteria:
-// 0. Workloads already marked for preemption first.
-// 1. Workloads from other ClusterQueues in the cohort before the ones in the
-// same ClusterQueue as the preemptor.
-// 2. (AdmissionFairSharing only) Workloads with lower LocalQueue's usage first
-// 3. Workloads with lower priority first.
-// 4. Workloads admitted more recently first.
-func CandidatesOrdering(log logr.Logger, afsEnabled bool, a, b *workload.Info, cq kueue.ClusterQueueReference, now time.Time) bool {
-	aEvicted := meta.IsStatusConditionTrue(a.Obj.Status.Conditions, kueue.WorkloadEvicted)
-	bEvicted := meta.IsStatusConditionTrue(b.Obj.Status.Conditions, kueue.WorkloadEvicted)
-	if aEvicted != bEvicted {
-		return aEvicted
-	}
-	aInCQ := a.ClusterQueue == cq
-	bInCQ := b.ClusterQueue == cq
-	if aInCQ != bInCQ {
-		return !aInCQ
-	}
-
-	if afsEnabled && resourceUsagePreemptionEnabled(a, b) {
-		if a.LocalQueueFSUsage != b.LocalQueueFSUsage {
-			log.V(3).Info("Comparing workloads by LocalQueue fair sharing usage",
-				"workloadA", klog.KObj(a.Obj), "queueA", a.Obj.Spec.QueueName, "usageA", a.LocalQueueFSUsage,
-				"workloadB", klog.KObj(b.Obj), "queueB", b.Obj.Spec.QueueName, "usageB", b.LocalQueueFSUsage)
-			return *a.LocalQueueFSUsage > *b.LocalQueueFSUsage
-		}
-	}
-	pa := priority.Priority(a.Obj)
-	pb := priority.Priority(b.Obj)
-	if pa != pb {
-		return pa < pb
-	}
-	timeA := quotaReservationTime(a.Obj, now)
-	timeB := quotaReservationTime(b.Obj, now)
-	if !timeA.Equal(timeB) {
-		return timeA.After(timeB)
-	}
-	// Arbitrary comparison for deterministic sorting.
-	return a.Obj.UID < b.Obj.UID
-}
-
-func quotaReservationTime(wl *kueue.Workload, now time.Time) time.Time {
-	cond := meta.FindStatusCondition(wl.Status.Conditions, kueue.WorkloadQuotaReserved)
-	if cond == nil || cond.Status != metav1.ConditionTrue {
-		// The condition wasn't populated yet, use the current time.
-		return now
-	}
-	return cond.LastTransitionTime.Time
-}



Test patch:
diff --git a/pkg/scheduler/preemption/preemption_test.go b/pkg/scheduler/preemption/preemption_test.go
index ba0affe30..13c3b8368 100644
--- a/pkg/scheduler/preemption/preemption_test.go
+++ b/pkg/scheduler/preemption/preemption_test.go
@@ -45,6 +45,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/resources"
 	"sigs.k8s.io/kueue/pkg/scheduler/flavorassigner"
+	preemptioncommon "sigs.k8s.io/kueue/pkg/scheduler/preemption/common"
 	"sigs.k8s.io/kueue/pkg/util/slices"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
 	"sigs.k8s.io/kueue/pkg/workload"
@@ -2897,7 +2898,7 @@ func TestCandidatesOrdering(t *testing.T) {
 	for _, tc := range cases {
 		features.SetFeatureGateDuringTest(t, features.AdmissionFairSharing, tc.admissionFairSharingEnabled)
 		sort.Slice(tc.candidates, func(i int, j int) bool {
-			return CandidatesOrdering(log, tc.admissionFairSharingEnabled, &tc.candidates[i], &tc.candidates[j], kueue.ClusterQueueReference(preemptorCq), now)
+			return preemptioncommon.CandidatesOrdering(log, tc.admissionFairSharingEnabled, &tc.candidates[i], &tc.candidates[j], kueue.ClusterQueueReference(preemptorCq), now)
 		})
 		got := slices.Map(tc.candidates, func(c *workload.Info) workload.Reference {
 			return workload.Reference(c.Obj.Name)
diff --git a/test/integration/singlecluster/scheduler/fairsharing/fair_sharing_test.go b/test/integration/singlecluster/scheduler/fairsharing/fair_sharing_test.go
index 299550600..22c16885d 100644
--- a/test/integration/singlecluster/scheduler/fairsharing/fair_sharing_test.go
+++ b/test/integration/singlecluster/scheduler/fairsharing/fair_sharing_test.go
@@ -614,6 +614,56 @@ var _ = ginkgo.Describe("Scheduler", ginkgo.Ordered, ginkgo.ContinueOnFailure, f
 			util.ExpectReservingActiveWorkloadsMetric(cqB, 1)
 		})
 	})
+
+	ginkgo.When("Preemption is enabled in fairsharing and there are best effort and guaranteed workloads", func() {
+		var (
+			bestEffortCQA *kueue.ClusterQueue
+			bestEffortCQB *kueue.ClusterQueue
+		)
+		ginkgo.BeforeEach(func() {
+			bestEffortCQA = createQueue(testing.MakeClusterQueue("best-effort-a").
+				Cohort("all").
+				ResourceGroup(
+					*testing.MakeFlavorQuotas("default").Resource(corev1.ResourceCPU, "0").Obj(),
+				).Obj())
+
+			bestEffortCQB = createQueue(testing.MakeClusterQueue("best-effort-b").
+				Cohort("all").
+				ResourceGroup(
+					*testing.MakeFlavorQuotas("default").Resource(corev1.ResourceCPU, "0").Obj(),
+				).Obj())
+
+			createQueue(testing.MakeClusterQueue("guaranteed").
+				Cohort("all").
+				ResourceGroup(
+					*testing.MakeFlavorQuotas("default").Resource(corev1.ResourceCPU, "8").Obj(),
+				).Preemption(kueue.ClusterQueuePreemption{
+				ReclaimWithinCohort: kueue.PreemptionPolicyAny}).
+				Obj())
+
+			features.SetFeatureGateDuringTest(ginkgo.GinkgoTB(), features.AdmissionFairSharing, false)
+		})
+
+		ginkgo.It("Guaranteed workloads cause preemption of a single best effort workload", func() {
+			ginkgo.By("Creating two best effort workloads in each best effort CQ")
+			wlBestEffortA := createWorkload("best-effort-a", "4")
+			util.WaitForNextSecondAfterCreation(wlBestEffortA)
+			util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wlBestEffortA)
+			util.ExpectReservingActiveWorkloadsMetric(bestEffortCQA, 1)
+			wlBestEffortB := createWorkload("best-effort-b", "4")
+			util.ExpectReservingActiveWorkloadsMetric(bestEffortCQB, 1)
+
+			ginkgo.By("Creating a guaranteed workload in the guaranteed CQ, that should reclaim quota")
+			wlGuaranteed := createWorkload("guaranteed", "4")
+
+			util.ExpectWorkloadsToBePreempted(ctx, k8sClient, wlBestEffortB)
+			util.FinishEvictionForWorkloads(ctx, k8sClient, wlBestEffortB)
+			util.ExpectWorkloadsToBeAdmitted(ctx, k8sClient, wlGuaranteed)
+
+			util.ExpectEvictedWorkloadsTotalMetric(bestEffortCQA.Name, kueue.WorkloadEvictedByPreemption, "", 0)
+			util.ExpectEvictedWorkloadsTotalMetric(bestEffortCQB.Name, kueue.WorkloadEvictedByPreemption, "", 1)
+		})
+	})
 })
 
 func expectCohortWeightedShare(cohortName string, weightedShare int64) {