From cffe156a77dc6c561b130d6e79949abde2a27edf Mon Sep 17 00:00:00 2001 From: mani Date: Thu, 23 Jul 2026 18:18:17 +0530 Subject: [PATCH 1/3] [YUNIKORN-3330] Run predicate checks in Required node preemption --- pkg/scheduler/objects/application_test.go | 94 ++++++++++++++ pkg/scheduler/objects/predicates.go | 7 +- pkg/scheduler/objects/preemption.go | 5 +- pkg/scheduler/objects/preemption_test.go | 28 +++-- .../objects/required_node_preemptor.go | 116 +++++++++++++----- 5 files changed, 204 insertions(+), 46 deletions(-) diff --git a/pkg/scheduler/objects/application_test.go b/pkg/scheduler/objects/application_test.go index 42dc28011..1f963e0d5 100644 --- a/pkg/scheduler/objects/application_test.go +++ b/pkg/scheduler/objects/application_test.go @@ -19,6 +19,7 @@ package objects import ( + "errors" "fmt" "math" "strings" @@ -3892,6 +3893,99 @@ func TestRequiredNodePreemption(t *testing.T) { } } +func TestRequiredNodePreemptionWithPredicates(t *testing.T) { + node := newNode(nodeID1, map[string]resources.Quantity{"first": 20}) + iterator := getNodeIteratorFn(node) + getNode := func(nodeID string) *Node { + return node + } + appQueueMapping := NewAppQueueMapping() + + // set queue + rootQ, err := createRootQueue(map[string]string{"first": "20"}) + assert.NilError(t, err) + childQ, err := createManagedQueueWithAppQueueMapping(rootQ, "default", false, map[string]string{"first": "20"}, appQueueMapping) + assert.NilError(t, err) + + // register predicate handler + preemptions := []mockCommon.Preemption{ + mockCommon.NewPreemption(true, "ask-2", nodeID1, []string{"ask-1"}, 0, 0), + } + wrongNodes := make(map[string]int, 1) + wrongNodes[nodeID2] = 10 + rightNodes := make(map[string]int, 1) + rightNodes[nodeID1] = 10 + tests := []struct { + name string + mockPlugin *mockCommon.PreemptionPredicatePlugin + result bool + mockPluginError error + }{ + {"nil plugin, so no predicate checks", nil, true, nil}, + {"prefilter fails", mockCommon.NewPreemptionPredicatePlugin(preemptions, nil, true, false), false, errors.New("fail")}, + {"prefilter passes but none of the node from iterator is available in feasible nodes", mockCommon.NewPreemptionPredicatePlugin(preemptions, wrongNodes, false, false), false, nil}, + {"prefilter pass with expected feasible nodes, filter fails", mockCommon.NewPreemptionPredicatePlugin(preemptions, rightNodes, false, true), false, errors.New("fail")}, + {"both prefilter and filter passes with correct feasible nodes", mockCommon.NewPreemptionPredicatePlugin(preemptions, rightNodes, false, false), true, nil}, + {"both prefilter and filter passes with empty feasible nodes", mockCommon.NewPreemptionPredicatePlugin(preemptions, nil, false, false), true, nil}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + app := newApplication(appID1, "default", "root.default") + app.SetQueue(childQ) + childQ.AddApplication(app) + appQueueMapping.AddAppQueueMapping(app.ApplicationID, childQ) + + // add an ask + askRes := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 15}) + ask1 := newAllocationAsk("ask-1", "app-1", askRes) + err = app.AddAllocationAsk(ask1) + assert.NilError(t, err, "could not add ask-1") + preemptionAttemptsRemaining := 1 + + // allocate ask + headRoom := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 50}) + result := app.tryAllocate(headRoom, true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode) + assert.Equal(t, result.ResultType, Allocated, "could not allocate ask-1") + assert.Equal(t, result.Request.allocationKey, "ask-1", "unexpected allocation key") + + // add ask2 with required node + ask2 := newAllocationAsk("ask-2", "app-1", askRes) + ask2.requiredNode = nodeID1 + err = app.AddAllocationAsk(ask2) + assert.NilError(t, err, "could not add ask-2") + + // try to allocate ask2 with node being full - expect a reservation + result = app.tryAllocate(headRoom, true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode) + assert.Equal(t, result.ResultType, Reserved, "allocation result is not reserved") + assert.Equal(t, result.Request.allocationKey, "ask-2", "unexpected allocation key") + err = app.Reserve(node, ask2) + assert.NilError(t, err, "reservation failed") + + if tt.mockPlugin != nil { + plugins.RegisterSchedulerPlugin(tt.mockPlugin) + } + app.tryReservedAllocate(headRoom, iterator) + if tt.result { + assert.Assert(t, ask1.IsPreempted(), "ask1 has not been preempted") + assert.Assert(t, ask2.HasTriggeredPreemption(), "ask2 has not triggered preemption") + } else { + assert.Assert(t, !ask1.IsPreempted(), "ask1 preempted") + assert.Assert(t, !ask2.HasTriggeredPreemption(), "ask2 triggered preemption") + } + if tt.mockPluginError != nil { + assert.ErrorContains(t, tt.mockPlugin.GetPredicateError(), tt.mockPluginError.Error()) + } + // reset + resetQ(t, childQ) + resetNode(node) + appQueueMapping.RemoveAppQueueMapping(app.ApplicationID) + if tt.mockPlugin != nil { + plugins.UnregisterSchedulerPlugins() + } + }) + } +} + func TestRequiredNodePreemptionFailed(t *testing.T) { // tests RequiredNode (DaemonSet) preemption where the victim pod has a high priority, hence preemption is not possible app := newApplication(appID0, "default", "root.default") diff --git a/pkg/scheduler/objects/predicates.go b/pkg/scheduler/objects/predicates.go index 179b01dc3..d641421f6 100644 --- a/pkg/scheduler/objects/predicates.go +++ b/pkg/scheduler/objects/predicates.go @@ -21,7 +21,6 @@ package objects import ( "fmt" "strings" - "sync" "go.uber.org/zap" @@ -114,9 +113,7 @@ func (pcr *predicateCheckResult) populateVictims(victimsByNode map[string][]*All } } -// preemptPredicateCheck performs a single predicate check and reports the resultType on a channel -func preemptPredicateCheck(plugin api.ResourceManagerCallback, ch chan<- *predicateCheckResult, wg *sync.WaitGroup, args *si.PreemptionPredicatesArgs) { - defer wg.Done() +func PredicateChecks(plugin api.ResourceManagerCallback, args *si.PreemptionPredicatesArgs) *predicateCheckResult { result := &predicateCheckResult{ allocationKey: args.AllocationKey, nodeID: args.NodeID, @@ -149,7 +146,7 @@ func preemptPredicateCheck(plugin api.ResourceManagerCallback, ch chan<- *predic result.predicateErrors = response.GetErrorMessage() } } - ch <- result + return result } func (p *predicateCheckResult) String() string { diff --git a/pkg/scheduler/objects/preemption.go b/pkg/scheduler/objects/preemption.go index 32a443c20..e19f8e03e 100644 --- a/pkg/scheduler/objects/preemption.go +++ b/pkg/scheduler/objects/preemption.go @@ -407,7 +407,10 @@ func (p *Preemptor) checkPreemptionPredicates(predicateChecks []*si.PreemptionPr // add goroutine for checking preemption wg.Add(1) expected++ - go preemptPredicateCheck(plugin, ch, &wg, args) + go func() { + defer wg.Done() + ch <- PredicateChecks(plugin, args) + }() } // wait for completion and close channel go func() { diff --git a/pkg/scheduler/objects/preemption_test.go b/pkg/scheduler/objects/preemption_test.go index c001ea65a..3cfe21d39 100644 --- a/pkg/scheduler/objects/preemption_test.go +++ b/pkg/scheduler/objects/preemption_test.go @@ -96,6 +96,7 @@ func creatApp1WithTwoDifferentAllocations( } else { alloc2 = newAllocationWithKey("alloc2", appID1, nodeID1, resources.NewResourceFromMap(app2Rec)) alloc2.createTime = ask2.createTime + app1.AddAllocation(alloc2) if !node1.TryAddAllocation(alloc2) { return nil, nil, fmt.Errorf("node alloc2 failed") } @@ -133,15 +134,23 @@ func resetNode(node *Node) { for _, v := range node.allocations { node.RemoveAllocation(v.allocationKey) } + node.reservations = make(map[string]*reservation) } func resetQ(t *testing.T, queue *Queue) { - for _, v := range queue.applications { + for _, v := range queue.GetCopyOfApps() { for _, a := range v.allocations { - v.RemoveAllocationAsk(a.allocationKey) + v.RemoveAllocation(a.allocationKey, si.TerminationType_STOPPED_BY_RM) err := queue.DecAllocatedResource(a.GetAllocatedResource()) assert.NilError(t, err) } + for _, req := range v.requests { + v.RemoveAllocationAsk(req.allocationKey) + } + for _, r := range v.reservations { + v.unReserveInternal(r) + } + v.queue.UnReserve(v.ApplicationID, len(v.reservations)) queue.RemoveApplication(v) } queue.applications = make(map[string]*Application) @@ -363,7 +372,7 @@ func TestTryPreemption(t *testing.T) { // register predicate handler preemptions := []mock.Preemption{ - mock.NewPreemption(true, "alloc3", nodeID1, []string{"alloc1"}, 0, 0), + mock.NewPreemption(true, "alloc3", nodeID1, []string{"alloc2"}, 0, 0), } wrongNodes := make(map[string]int, 1) wrongNodes[nodeID2] = 10 @@ -438,7 +447,7 @@ func TestTryPreemption_SendEvent(t *testing.T) { eventSystem := evtMock.NewEventSystem() events := schedEvt.NewAskEvents(eventSystem) - alloc1.askEvents = events + alloc2.askEvents = events app2, ask3, err := creatApp2(childQ2, map[string]resources.Quantity{"first": 5, "pods": 1}, "alloc3", appQueueMapping) assert.NilError(t, err) @@ -449,7 +458,7 @@ func TestTryPreemption_SendEvent(t *testing.T) { // register predicate handler preemptions := []mock.Preemption{ - mock.NewPreemption(true, "alloc3", nodeID1, []string{"alloc1"}, 0, 0), + mock.NewPreemption(true, "alloc3", nodeID1, []string{"alloc2"}, 0, 0), } plugin := mock.NewPreemptionPredicatePlugin(preemptions, nil, false, false) plugins.RegisterSchedulerPlugin(plugin) @@ -460,12 +469,12 @@ func TestTryPreemption_SendEvent(t *testing.T) { assert.NilError(t, plugin.GetPredicateError()) assert.Assert(t, ok, "no victims found") assert.Equal(t, "alloc3", result.Request.GetAllocationKey(), "wrong alloc") - assert.Check(t, alloc1.IsPreempted(), "alloc1 not preempted") - assert.Check(t, !alloc2.IsPreempted(), "alloc2 preempted") + assert.Check(t, !alloc1.IsPreempted(), "alloc1 not preempted") + assert.Check(t, alloc2.IsPreempted(), "alloc2 preempted") assert.Equal(t, 1, len(eventSystem.Events)) event := eventSystem.Events[0] - assert.Equal(t, alloc1.applicationID, event.ReferenceID) - assert.Equal(t, alloc1.allocationKey, event.ObjectID) + assert.Equal(t, alloc2.applicationID, event.ReferenceID) + assert.Equal(t, alloc2.allocationKey, event.ObjectID) assert.Equal(t, si.EventRecord_NONE, event.EventChangeType) assert.Equal(t, si.EventRecord_DETAILS_NONE, event.EventChangeDetail) assert.Equal(t, si.EventRecord_REQUEST, event.Type) @@ -753,6 +762,7 @@ func TestTryPreemptionOnQueue(t *testing.T) { assert.Check(t, ok == true, "either node1 or node2 chosen") assert.Check(t, alloc1.IsPreempted() || alloc2.IsPreempted(), "either alloc1 or alloc2 preempted, but not both") assert.Check(t, !alloc1.IsPreempted() || !alloc2.IsPreempted(), "either alloc1 or alloc2 not preempted, but not both") + childQ1.DecPreemptingResource(alloc1.GetAllocatedResource()) assert.Equal(t, len(ask3.GetAllocationLog()), 0) } else { assert.Assert(t, result == nil, "no result") diff --git a/pkg/scheduler/objects/required_node_preemptor.go b/pkg/scheduler/objects/required_node_preemptor.go index 9f99a07a9..a9328f2d4 100644 --- a/pkg/scheduler/objects/required_node_preemptor.go +++ b/pkg/scheduler/objects/required_node_preemptor.go @@ -26,6 +26,7 @@ import ( "github.com/apache/yunikorn-core/pkg/common" "github.com/apache/yunikorn-core/pkg/common/resources" "github.com/apache/yunikorn-core/pkg/log" + "github.com/apache/yunikorn-core/pkg/plugins" "github.com/apache/yunikorn-scheduler-interface/lib/go/si" ) @@ -67,42 +68,41 @@ func (p *PreemptionContext) tryPreemption() { p.sortAllocations() // Are there any victims/asks to preempt? - victims := p.GetVictims() + var victims []*Allocation + var finalVictims []*Allocation + victims = p.GetVictims() if len(victims) > 0 { - log.Log(log.SchedRequiredNodePreemption).Info("Found victims for required node preemption", - zap.String("ds allocation key", p.requiredAsk.GetAllocationKey()), - zap.String("allocation name", p.requiredAsk.GetAllocationName()), - zap.Int("no.of victims", len(victims))) - for _, victim := range victims { - err := victim.MarkPreempted() - if err != nil { - log.Log(log.SchedRequiredNodePreemption).Warn("allocation is already released, so not proceeding further on the daemon set preemption process", - zap.String("applicationID", p.requiredAsk.GetApplicationID()), - zap.String("allocationKey", victim.GetAllocationKey())) - continue + finalVictims = p.runPredicates(victims) + if len(finalVictims) > 0 { + for _, victim := range finalVictims { + err := victim.MarkPreempted() + if err != nil { + log.Log(log.SchedRequiredNodePreemption).Warn("allocation is already released, so not proceeding further on the daemon set preemption process", + zap.String("applicationID", p.requiredAsk.GetApplicationID()), + zap.String("allocationKey", victim.GetAllocationKey())) + continue + } + if victimQueue := p.application.queue.GetQueueByAppID(victim.GetApplicationID()); victimQueue != nil { + victimQueue.IncPreemptingResource(victim.GetAllocatedResource()) + } else { + log.Log(log.SchedRequiredNodePreemption).Warn("BUG: Queue not found for daemon set preemption victim", + zap.String("queue", p.application.queue.Name), + zap.String("victimApplicationID", victim.GetApplicationID()), + zap.String("victimAllocationKey", victim.GetAllocationKey())) + } + victim.SendPreemptedBySchedulerEvent(p.requiredAsk.GetAllocationKey(), p.requiredAsk.GetApplicationID(), p.application.queuePath) } - if victimQueue := p.application.queue.GetQueueByAppID(victim.GetApplicationID()); victimQueue != nil { - victimQueue.IncPreemptingResource(victim.GetAllocatedResource()) - } else { - log.Log(log.SchedRequiredNodePreemption).Warn("BUG: Queue not found for daemon set preemption victim", - zap.String("queue", p.application.queue.Name), - zap.String("victimApplicationID", victim.GetApplicationID()), - zap.String("victimAllocationKey", victim.GetAllocationKey())) - } - victim.SendPreemptedBySchedulerEvent(p.requiredAsk.GetAllocationKey(), p.requiredAsk.GetApplicationID(), p.application.queuePath) + p.requiredAsk.MarkTriggeredPreemption() + p.application.notifyRMAllocationReleased(victims, si.TerminationType_PREEMPTED_BY_SCHEDULER, + "preempting allocations to free up resources to run daemon set ask: "+p.requiredAsk.GetAllocationKey()) } - p.requiredAsk.MarkTriggeredPreemption() - p.application.notifyRMAllocationReleased( - victims, - si.TerminationType_PREEMPTED_BY_SCHEDULER, - "preempting allocations to free up resources to run daemon set ask: "+p.requiredAsk.GetAllocationKey(), - ) - } else { + } + if len(victims) == 0 || len(finalVictims) == 0 { p.requiredAsk.LogAllocationFailure(common.NoVictimForRequiredNode, true) p.requiredAsk.SendRequiredNodePreemptionFailedEvent(p.node.NodeID) getRateLimitedReqNodeLog().Info("no victim found for required node preemption", - zap.String("allocation key", p.requiredAsk.GetAllocationKey()), - zap.String("allocation name", p.requiredAsk.GetAllocationName()), + zap.String("allocationKey", p.requiredAsk.GetAllocationKey()), + zap.String("allocationName", p.requiredAsk.GetAllocationName()), zap.String("node", p.node.NodeID), zap.Int("total allocations", result.totalAllocations), zap.Int("requiredNode allocations", result.requiredNodeAllocations), @@ -174,7 +174,7 @@ func (p *PreemptionContext) GetVictims() []*Allocation { } } - // Did we found the useful set of victims? + // Did we find the useful set of victims? if len(victims) > 0 && resources.StrictlyGreaterThanOrEquals( resources.Add(currentResource, p.node.GetAvailableResource()), p.requiredAsk.GetAllocatedResource()) { return victims @@ -182,6 +182,60 @@ func (p *PreemptionContext) GetVictims() []*Allocation { return nil } +// runPredicates Run Predicate checks to confirm whether collected victims from the required node is +// good enough to move forward on the preemption further or not +func (p *PreemptionContext) runPredicates(victims []*Allocation) []*Allocation { + finalVictims := make([]*Allocation, 0) + plugin := plugins.GetResourceManagerCallbackPlugin() + if plugin == nil { + log.Log(log.SchedRequiredNodePreemption).Debug("No RM callback plugin registered, using chosen victims as is", + zap.String("node", p.node.NodeID), + zap.String("allocationKey", p.requiredAsk.GetAllocationKey())) + return victims + } else { + // run predicates for this pod before in hand and fetch feasible nodes + feasibleNodes, predicatesResult := p.requiredAsk.preAllocateConditions(true) + if !predicatesResult { + return finalVictims + } + // Is this node suitable to run the pod? + if len(feasibleNodes) > 0 { + if _, ok := feasibleNodes[p.node.NodeID]; !ok { + log.Log(log.SchedRequiredNodePreemption).Debug("skipping node as it is not feasible to run the pod", + zap.String("allocationKey", p.requiredAsk.GetAllocationKey()), + zap.String("node", p.node.NodeID)) + return finalVictims + } + } + keys := make([]string, 0) + for _, victim := range victims { + keys = append(keys, victim.GetAllocationKey()) + } + args := &si.PreemptionPredicatesArgs{ + AllocationKey: p.requiredAsk.GetAllocationKey(), + NodeID: p.node.NodeID, + PreemptAllocationKeys: keys, + StartIndex: int32(0), + } + predicateResult := PredicateChecks(plugin, args) + if !predicateResult.success || predicateResult.index == -1 { + log.Log(log.SchedRequiredNodePreemption).Debug("running predicate checks failed", + zap.String("allocationKey", p.requiredAsk.GetAllocationKey()), + zap.String("node", p.node.NodeID)) + return finalVictims + } + log.Log(log.SchedRequiredNodePreemption).Info("Found victims for required node preemption", + zap.String("allocationKey", p.requiredAsk.GetAllocationKey()), + zap.String("allocationName", p.requiredAsk.GetAllocationName()), + zap.Int("total victims", len(victims))) + victimsByNode := make(map[string][]*Allocation) + victimsByNode[p.node.NodeID] = victims + predicateResult.populateVictims(victimsByNode) + finalVictims = predicateResult.victims + } + return finalVictims +} + // for test only func (p *PreemptionContext) getAllocations() []*Allocation { return p.allocations From a6750794d602a0b411b9b9a3ab4bf5de12511651 Mon Sep 17 00:00:00 2001 From: mani Date: Wed, 7 Oct 2026 18:23:31 +0530 Subject: [PATCH 2/3] Addressed review comments --- pkg/scheduler/objects/application_test.go | 65 +++++++++---- pkg/scheduler/objects/predicates.go | 38 -------- pkg/scheduler/objects/preemption.go | 2 +- pkg/scheduler/objects/preemption_test.go | 7 +- pkg/scheduler/objects/preemption_utilities.go | 44 +++++++++ .../objects/required_node_preemptor.go | 91 +++++++++---------- .../objects/required_node_preemptor_test.go | 32 +++---- 7 files changed, 155 insertions(+), 124 deletions(-) diff --git a/pkg/scheduler/objects/application_test.go b/pkg/scheduler/objects/application_test.go index 1f963e0d5..913887476 100644 --- a/pkg/scheduler/objects/application_test.go +++ b/pkg/scheduler/objects/application_test.go @@ -3911,22 +3911,40 @@ func TestRequiredNodePreemptionWithPredicates(t *testing.T) { preemptions := []mockCommon.Preemption{ mockCommon.NewPreemption(true, "ask-2", nodeID1, []string{"ask-1"}, 0, 0), } + preemptions1 := []mockCommon.Preemption{ + mockCommon.NewPreemption(true, "ask-2", nodeID1, []string{"ask-3", "ask-1"}, 0, 0), + } wrongNodes := make(map[string]int, 1) wrongNodes[nodeID2] = 10 rightNodes := make(map[string]int, 1) rightNodes[nodeID1] = 10 + + res1 := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 15}) + res2 := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 15}) + res3 := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}) + + ask1Info := make(map[string]*resources.Resource) + ask1Info["ask-1"] = res1 + + ask2Info := make(map[string]*resources.Resource) + ask2Info["ask-1"] = res3 + ask2Info["ask-3"] = res2 + tests := []struct { name string + asks map[string]*resources.Resource mockPlugin *mockCommon.PreemptionPredicatePlugin result bool mockPluginError error + expectedVictims []string }{ - {"nil plugin, so no predicate checks", nil, true, nil}, - {"prefilter fails", mockCommon.NewPreemptionPredicatePlugin(preemptions, nil, true, false), false, errors.New("fail")}, - {"prefilter passes but none of the node from iterator is available in feasible nodes", mockCommon.NewPreemptionPredicatePlugin(preemptions, wrongNodes, false, false), false, nil}, - {"prefilter pass with expected feasible nodes, filter fails", mockCommon.NewPreemptionPredicatePlugin(preemptions, rightNodes, false, true), false, errors.New("fail")}, - {"both prefilter and filter passes with correct feasible nodes", mockCommon.NewPreemptionPredicatePlugin(preemptions, rightNodes, false, false), true, nil}, - {"both prefilter and filter passes with empty feasible nodes", mockCommon.NewPreemptionPredicatePlugin(preemptions, nil, false, false), true, nil}, + {"nil plugin, so no predicate checks", ask1Info, nil, true, nil, []string{"ask-1"}}, + {"prefilter fails", ask1Info, mockCommon.NewPreemptionPredicatePlugin(preemptions, nil, true, false), false, errors.New("fail"), nil}, + {"prefilter passes but none of the node from iterator is available in feasible nodes", ask1Info, mockCommon.NewPreemptionPredicatePlugin(preemptions, wrongNodes, false, false), false, nil, nil}, + {"prefilter pass with expected feasible nodes, filter fails", ask1Info, mockCommon.NewPreemptionPredicatePlugin(preemptions, rightNodes, false, true), false, errors.New("fail"), nil}, + {"both prefilter and filter passes with correct feasible nodes", ask1Info, mockCommon.NewPreemptionPredicatePlugin(preemptions, rightNodes, false, false), true, nil, []string{"ask-1"}}, + {"both prefilter and filter passes with correct feasible nodes but predicate checks failed for 1 victim", ask2Info, mockCommon.NewPreemptionPredicatePlugin(preemptions1, rightNodes, false, false), true, nil, []string{"ask-3"}}, + {"both prefilter and filter passes with empty feasible nodes", ask1Info, mockCommon.NewPreemptionPredicatePlugin(preemptions, nil, false, false), true, nil, []string{"ask-1"}}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -3935,27 +3953,30 @@ func TestRequiredNodePreemptionWithPredicates(t *testing.T) { childQ.AddApplication(app) appQueueMapping.AddAppQueueMapping(app.ApplicationID, childQ) - // add an ask - askRes := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 15}) - ask1 := newAllocationAsk("ask-1", "app-1", askRes) - err = app.AddAllocationAsk(ask1) - assert.NilError(t, err, "could not add ask-1") + headRoom := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 50}) preemptionAttemptsRemaining := 1 - // allocate ask - headRoom := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 50}) - result := app.tryAllocate(headRoom, true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode) - assert.Equal(t, result.ResultType, Allocated, "could not allocate ask-1") - assert.Equal(t, result.Request.allocationKey, "ask-1", "unexpected allocation key") + // add an ask + askInfo := make(map[string]*Allocation) + for k, v := range tt.asks { + ask := newAllocationAsk(k, "app-1", v) + err = app.AddAllocationAsk(ask) + assert.NilError(t, err, "could not add ask "+k) + askInfo[k] = ask + + result := app.tryAllocate(headRoom, true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode) + assert.Equal(t, result.ResultType, Allocated, "could not allocate ask") + assert.Equal(t, result.Request.allocationKey, k, "unexpected allocation key") + } // add ask2 with required node - ask2 := newAllocationAsk("ask-2", "app-1", askRes) + ask2 := newAllocationAsk("ask-2", "app-1", res1) ask2.requiredNode = nodeID1 err = app.AddAllocationAsk(ask2) assert.NilError(t, err, "could not add ask-2") // try to allocate ask2 with node being full - expect a reservation - result = app.tryAllocate(headRoom, true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode) + result := app.tryAllocate(headRoom, true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode) assert.Equal(t, result.ResultType, Reserved, "allocation result is not reserved") assert.Equal(t, result.Request.allocationKey, "ask-2", "unexpected allocation key") err = app.Reserve(node, ask2) @@ -3966,10 +3987,14 @@ func TestRequiredNodePreemptionWithPredicates(t *testing.T) { } app.tryReservedAllocate(headRoom, iterator) if tt.result { - assert.Assert(t, ask1.IsPreempted(), "ask1 has not been preempted") + for _, expVictim := range tt.expectedVictims { + assert.Assert(t, askInfo[expVictim].IsPreempted(), "ask1 has not been preempted") + } assert.Assert(t, ask2.HasTriggeredPreemption(), "ask2 has not triggered preemption") } else { - assert.Assert(t, !ask1.IsPreempted(), "ask1 preempted") + for _, expVictim := range tt.expectedVictims { + assert.Assert(t, !askInfo[expVictim].IsPreempted(), "ask1 has not been preempted") + } assert.Assert(t, !ask2.HasTriggeredPreemption(), "ask2 triggered preemption") } if tt.mockPluginError != nil { diff --git a/pkg/scheduler/objects/predicates.go b/pkg/scheduler/objects/predicates.go index d641421f6..1d485e4a5 100644 --- a/pkg/scheduler/objects/predicates.go +++ b/pkg/scheduler/objects/predicates.go @@ -25,8 +25,6 @@ import ( "go.uber.org/zap" "github.com/apache/yunikorn-core/pkg/log" - "github.com/apache/yunikorn-scheduler-interface/lib/go/api" - "github.com/apache/yunikorn-scheduler-interface/lib/go/si" ) type predicateCheckResult struct { @@ -113,42 +111,6 @@ func (pcr *predicateCheckResult) populateVictims(victimsByNode map[string][]*All } } -func PredicateChecks(plugin api.ResourceManagerCallback, args *si.PreemptionPredicatesArgs) *predicateCheckResult { - result := &predicateCheckResult{ - allocationKey: args.AllocationKey, - nodeID: args.NodeID, - success: false, - index: -1, - predicateErrors: make(map[string]int32), - } - if len(args.PreemptAllocationKeys) == 0 { - // normal check; there are sufficient resources to run on this node - if err := plugin.Predicates(&si.PredicatesArgs{ - AllocationKey: args.AllocationKey, - NodeID: args.NodeID, - Allocate: true, - }); err == nil { - result.success = true - result.index = -1 - } else { - log.Log(log.SchedPreemption).Debug("Normal predicate check failed", - zap.String("AllocationKey", args.AllocationKey), - zap.String("NodeID", args.NodeID), - zap.Error(err)) - result.predicateErrors[err.Error()]++ - } - } else if response := plugin.PreemptionPredicates(args); response != nil { - // preemption check; at least one allocation will need preemption - result.success = response.GetSuccess() - if result.success { - result.index = int(response.GetIndex()) - } else { - result.predicateErrors = response.GetErrorMessage() - } - } - return result -} - func (p *predicateCheckResult) String() string { if p.nodeID == "" { return "" diff --git a/pkg/scheduler/objects/preemption.go b/pkg/scheduler/objects/preemption.go index e19f8e03e..02f7837cb 100644 --- a/pkg/scheduler/objects/preemption.go +++ b/pkg/scheduler/objects/preemption.go @@ -409,7 +409,7 @@ func (p *Preemptor) checkPreemptionPredicates(predicateChecks []*si.PreemptionPr expected++ go func() { defer wg.Done() - ch <- PredicateChecks(plugin, args) + ch <- runPredicateChecks(plugin, args) }() } // wait for completion and close channel diff --git a/pkg/scheduler/objects/preemption_test.go b/pkg/scheduler/objects/preemption_test.go index 3cfe21d39..91363988a 100644 --- a/pkg/scheduler/objects/preemption_test.go +++ b/pkg/scheduler/objects/preemption_test.go @@ -408,8 +408,9 @@ func TestTryPreemption(t *testing.T) { assert.NilError(t, tt.mockPlugin.GetPredicateError()) assert.Assert(t, ok, "no victims found") assert.Equal(t, "alloc3", result.Request.GetAllocationKey(), "wrong alloc") - assert.Check(t, alloc1.IsPreempted(), "alloc1 not preempted") - assert.Check(t, !alloc2.IsPreempted(), "alloc2 preempted") + assert.Check(t, !alloc1.IsPreempted(), "alloc1 not preempted") + assert.Check(t, alloc2.IsPreempted(), "alloc2 preempted") + childQ1.DecPreemptingResource(alloc1.GetAllocatedResource()) assert.Equal(t, len(ask3.GetAllocationLog()), 0) } else { assert.Assert(t, result == nil, "no result") @@ -502,7 +503,7 @@ func TestTryAllocateDoesNotWaitForPreemptionRMReply(t *testing.T) { ask.allowPreemptOther = true plugin := mock.NewPreemptionPredicatePlugin([]mock.Preemption{ - mock.NewPreemption(true, "alloc3", nodeID1, []string{"alloc1"}, 0, 0), + mock.NewPreemption(true, "alloc3", nodeID1, []string{"alloc2"}, 0, 0), }, nil, false, false) plugins.RegisterSchedulerPlugin(plugin) defer plugins.UnregisterSchedulerPlugins() diff --git a/pkg/scheduler/objects/preemption_utilities.go b/pkg/scheduler/objects/preemption_utilities.go index b7425baa1..e65533dd7 100644 --- a/pkg/scheduler/objects/preemption_utilities.go +++ b/pkg/scheduler/objects/preemption_utilities.go @@ -22,7 +22,12 @@ import ( "sort" "time" + "go.uber.org/zap" + "github.com/apache/yunikorn-core/pkg/common/resources" + "github.com/apache/yunikorn-core/pkg/log" + "github.com/apache/yunikorn-scheduler-interface/lib/go/api" + "github.com/apache/yunikorn-scheduler-interface/lib/go/si" ) var ( @@ -141,3 +146,42 @@ func scoreAllocationBasedOnAsk(allocation *Allocation, ask *resources.Resource) score += allocation.GetAllocatedResource().TypeMatching(ask) return score } + +func runPredicateChecks(plugin api.ResourceManagerCallback, args *si.PreemptionPredicatesArgs) *predicateCheckResult { + result := &predicateCheckResult{ + allocationKey: args.AllocationKey, + nodeID: args.NodeID, + success: false, + index: -1, + predicateErrors: make(map[string]int32), + } + if len(args.PreemptAllocationKeys) == 0 { + // normal check; there are sufficient resources to run on this node + if err := plugin.Predicates(&si.PredicatesArgs{ + AllocationKey: args.AllocationKey, + NodeID: args.NodeID, + Allocate: true, + }); err == nil { + result.success = true + result.index = -1 + } else { + log.Log(log.SchedPreemption).Debug("Normal predicate check failed", + zap.String("AllocationKey", args.AllocationKey), + zap.String("NodeID", args.NodeID), + zap.Error(err)) + result.predicateErrors[err.Error()]++ + } + } else if response := plugin.PreemptionPredicates(args); response != nil { + // preemption check; at least one allocation will need preemption + result.success = response.GetSuccess() + if result.success { + result.index = int(response.GetIndex()) + } else { + log.Log(log.SchedPreemption).Debug("Preemption predicate check failed", + zap.String("AllocationKey", args.AllocationKey), + zap.String("NodeID", args.NodeID)) + result.predicateErrors = response.GetErrorMessage() + } + } + return result +} diff --git a/pkg/scheduler/objects/required_node_preemptor.go b/pkg/scheduler/objects/required_node_preemptor.go index a9328f2d4..c1f30bdb7 100644 --- a/pkg/scheduler/objects/required_node_preemptor.go +++ b/pkg/scheduler/objects/required_node_preemptor.go @@ -70,34 +70,9 @@ func (p *PreemptionContext) tryPreemption() { // Are there any victims/asks to preempt? var victims []*Allocation var finalVictims []*Allocation - victims = p.GetVictims() - if len(victims) > 0 { - finalVictims = p.runPredicates(victims) - if len(finalVictims) > 0 { - for _, victim := range finalVictims { - err := victim.MarkPreempted() - if err != nil { - log.Log(log.SchedRequiredNodePreemption).Warn("allocation is already released, so not proceeding further on the daemon set preemption process", - zap.String("applicationID", p.requiredAsk.GetApplicationID()), - zap.String("allocationKey", victim.GetAllocationKey())) - continue - } - if victimQueue := p.application.queue.GetQueueByAppID(victim.GetApplicationID()); victimQueue != nil { - victimQueue.IncPreemptingResource(victim.GetAllocatedResource()) - } else { - log.Log(log.SchedRequiredNodePreemption).Warn("BUG: Queue not found for daemon set preemption victim", - zap.String("queue", p.application.queue.Name), - zap.String("victimApplicationID", victim.GetApplicationID()), - zap.String("victimAllocationKey", victim.GetAllocationKey())) - } - victim.SendPreemptedBySchedulerEvent(p.requiredAsk.GetAllocationKey(), p.requiredAsk.GetApplicationID(), p.application.queuePath) - } - p.requiredAsk.MarkTriggeredPreemption() - p.application.notifyRMAllocationReleased(victims, si.TerminationType_PREEMPTED_BY_SCHEDULER, - "preempting allocations to free up resources to run daemon set ask: "+p.requiredAsk.GetAllocationKey()) - } - } - if len(victims) == 0 || len(finalVictims) == 0 { + victims = p.runPredicates() + finalVictims = p.pruneVictims(victims) + if len(finalVictims) == 0 { p.requiredAsk.LogAllocationFailure(common.NoVictimForRequiredNode, true) p.requiredAsk.SendRequiredNodePreemptionFailedEvent(p.node.NodeID) getRateLimitedReqNodeLog().Info("no victim found for required node preemption", @@ -109,7 +84,29 @@ func (p *PreemptionContext) tryPreemption() { zap.Int("allocations already preempted", result.alreadyPreemptedAllocations), zap.Int("higher priority allocations", result.higherPriorityAllocations), zap.Int("allocations with non-matching resources", result.atLeastOneResNotMatched)) + return + } + for _, victim := range finalVictims { + err := victim.MarkPreempted() + if err != nil { + log.Log(log.SchedRequiredNodePreemption).Warn("allocation is already released, so not proceeding further on the daemon set preemption process", + zap.String("applicationID", p.requiredAsk.GetApplicationID()), + zap.String("allocationKey", victim.GetAllocationKey())) + continue + } + if victimQueue := p.application.queue.GetQueueByAppID(victim.GetApplicationID()); victimQueue != nil { + victimQueue.IncPreemptingResource(victim.GetAllocatedResource()) + } else { + log.Log(log.SchedRequiredNodePreemption).Warn("BUG: Queue not found for daemon set preemption victim", + zap.String("queue", p.application.queue.Name), + zap.String("victimApplicationID", victim.GetApplicationID()), + zap.String("victimAllocationKey", victim.GetAllocationKey())) + } + victim.SendPreemptedBySchedulerEvent(p.requiredAsk.GetAllocationKey(), p.requiredAsk.GetApplicationID(), p.application.queuePath) } + p.requiredAsk.MarkTriggeredPreemption() + p.application.notifyRMAllocationReleased(finalVictims, si.TerminationType_PREEMPTED_BY_SCHEDULER, + "preempting allocations to free up resources to run daemon set ask: "+p.requiredAsk.GetAllocationKey()) } func (p *PreemptionContext) filterAllocations() filteringResult { @@ -162,30 +159,31 @@ func (p *PreemptionContext) sortAllocations() { SortAllocations(p.allocations) } -func (p *PreemptionContext) GetVictims() []*Allocation { - var victims []*Allocation +func (p *PreemptionContext) pruneVictims(victims []*Allocation) []*Allocation { + var prunedVictims []*Allocation var currentResource = resources.NewResource() - for _, allocation := range p.allocations { + for _, allocation := range victims { if !resources.StrictlyGreaterThanOrEquals(currentResource, p.requiredAsk.GetAllocatedResource()) { currentResource.AddTo(allocation.GetAllocatedResource()) - victims = append(victims, allocation) + prunedVictims = append(prunedVictims, allocation) } else { break } } // Did we find the useful set of victims? - if len(victims) > 0 && resources.StrictlyGreaterThanOrEquals( + if len(prunedVictims) > 0 && resources.StrictlyGreaterThanOrEquals( resources.Add(currentResource, p.node.GetAvailableResource()), p.requiredAsk.GetAllocatedResource()) { - return victims + return prunedVictims } return nil } // runPredicates Run Predicate checks to confirm whether collected victims from the required node is // good enough to move forward on the preemption further or not -func (p *PreemptionContext) runPredicates(victims []*Allocation) []*Allocation { - finalVictims := make([]*Allocation, 0) +func (p *PreemptionContext) runPredicates() []*Allocation { + victims := p.allocations + passedVictims := make([]*Allocation, 0) plugin := plugins.GetResourceManagerCallbackPlugin() if plugin == nil { log.Log(log.SchedRequiredNodePreemption).Debug("No RM callback plugin registered, using chosen victims as is", @@ -196,7 +194,7 @@ func (p *PreemptionContext) runPredicates(victims []*Allocation) []*Allocation { // run predicates for this pod before in hand and fetch feasible nodes feasibleNodes, predicatesResult := p.requiredAsk.preAllocateConditions(true) if !predicatesResult { - return finalVictims + return passedVictims } // Is this node suitable to run the pod? if len(feasibleNodes) > 0 { @@ -204,7 +202,7 @@ func (p *PreemptionContext) runPredicates(victims []*Allocation) []*Allocation { log.Log(log.SchedRequiredNodePreemption).Debug("skipping node as it is not feasible to run the pod", zap.String("allocationKey", p.requiredAsk.GetAllocationKey()), zap.String("node", p.node.NodeID)) - return finalVictims + return passedVictims } } keys := make([]string, 0) @@ -217,23 +215,24 @@ func (p *PreemptionContext) runPredicates(victims []*Allocation) []*Allocation { PreemptAllocationKeys: keys, StartIndex: int32(0), } - predicateResult := PredicateChecks(plugin, args) + predicateResult := runPredicateChecks(plugin, args) if !predicateResult.success || predicateResult.index == -1 { log.Log(log.SchedRequiredNodePreemption).Debug("running predicate checks failed", zap.String("allocationKey", p.requiredAsk.GetAllocationKey()), zap.String("node", p.node.NodeID)) - return finalVictims + return passedVictims } - log.Log(log.SchedRequiredNodePreemption).Info("Found victims for required node preemption", - zap.String("allocationKey", p.requiredAsk.GetAllocationKey()), - zap.String("allocationName", p.requiredAsk.GetAllocationName()), - zap.Int("total victims", len(victims))) victimsByNode := make(map[string][]*Allocation) victimsByNode[p.node.NodeID] = victims predicateResult.populateVictims(victimsByNode) - finalVictims = predicateResult.victims + passedVictims = predicateResult.victims + log.Log(log.SchedRequiredNodePreemption).Info("Found victims for required node preemption", + zap.String("allocationKey", p.requiredAsk.GetAllocationKey()), + zap.String("allocationName", p.requiredAsk.GetAllocationName()), + zap.Int("victims", len(victims)), + zap.Int("passed victims", len(passedVictims))) } - return finalVictims + return passedVictims } // for test only diff --git a/pkg/scheduler/objects/required_node_preemptor_test.go b/pkg/scheduler/objects/required_node_preemptor_test.go index 01244f58c..69948d93b 100644 --- a/pkg/scheduler/objects/required_node_preemptor_test.go +++ b/pkg/scheduler/objects/required_node_preemptor_test.go @@ -146,16 +146,16 @@ func TestGetVictims(t *testing.T) { asks := prepareAllocationAsks(t, node) p.filterAllocations() p.sortAllocations() - victims := p.GetVictims() - assert.Equal(t, len(victims), 4) - assert.Equal(t, victims[0].GetAllocationKey(), "ask5") - assert.Equal(t, resources.Equals(victims[0].GetAllocatedResource(), resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})), true) - assert.Equal(t, victims[1].GetAllocationKey(), "ask1") - assert.Equal(t, resources.Equals(victims[1].GetAllocatedResource(), resources.NewResourceFromMap(map[string]resources.Quantity{"first": 10})), true) - assert.Equal(t, victims[2].GetAllocationKey(), "ask4") - assert.Equal(t, resources.Equals(victims[2].GetAllocatedResource(), resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})), true) - assert.Equal(t, victims[3].GetAllocationKey(), "ask2") - assert.Equal(t, resources.Equals(victims[3].GetAllocatedResource(), resources.NewResourceFromMap(map[string]resources.Quantity{"first": 8})), true) + prunedVictims := p.pruneVictims(p.allocations) + assert.Equal(t, len(prunedVictims), 4) + assert.Equal(t, prunedVictims[0].GetAllocationKey(), "ask5") + assert.Equal(t, resources.Equals(prunedVictims[0].GetAllocatedResource(), resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})), true) + assert.Equal(t, prunedVictims[1].GetAllocationKey(), "ask1") + assert.Equal(t, resources.Equals(prunedVictims[1].GetAllocatedResource(), resources.NewResourceFromMap(map[string]resources.Quantity{"first": 10})), true) + assert.Equal(t, prunedVictims[2].GetAllocationKey(), "ask4") + assert.Equal(t, resources.Equals(prunedVictims[2].GetAllocatedResource(), resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})), true) + assert.Equal(t, prunedVictims[3].GetAllocationKey(), "ask2") + assert.Equal(t, resources.Equals(prunedVictims[3].GetAllocatedResource(), resources.NewResourceFromMap(map[string]resources.Quantity{"first": 8})), true) removeAllocationAsks(node, asks) // case 2: victims are available and its resources are matching with ds request ask (but with different quantity) @@ -165,8 +165,8 @@ func TestGetVictims(t *testing.T) { asks = prepareAllocationAsks(t, node) p2.filterAllocations() p2.sortAllocations() - victims2 := p2.GetVictims() - assert.Equal(t, len(victims2), 1) + prunedVictims2 := p2.pruneVictims(p2.allocations) + assert.Equal(t, len(prunedVictims2), 1) removeAllocationAsks(node, asks) // case 3: allocations are available and its resources are matching partially with ds request ask (because of different resource types), hence no victims @@ -182,8 +182,8 @@ func TestGetVictims(t *testing.T) { p3.sortAllocations() // allocations are available but no exact match for choosing victims - victims3 := p3.GetVictims() - assert.Equal(t, len(victims3), 0) + prunedVictims3 := p3.pruneVictims(p3.allocations) + assert.Equal(t, len(prunedVictims3), 0) removeAllocationAsks(node, asks) // case 4: victim chosen earlier released in the meantime @@ -195,8 +195,8 @@ func TestGetVictims(t *testing.T) { p4.sortAllocations() err = asks[1].SetReleased(true) assert.NilError(t, err) - victims = p4.GetVictims() - assert.Equal(t, len(victims), 4) + prunedVictims4 := p4.pruneVictims(p4.allocations) + assert.Equal(t, len(prunedVictims4), 4) removeAllocationAsks(node, asks) } From 77695f3a0ce21e2900d63735b0b7d17f78270e7f Mon Sep 17 00:00:00 2001 From: mani Date: Thu, 8 Oct 2026 12:03:42 +0530 Subject: [PATCH 3/3] Addressed review comments * Fixed lint warnings --- pkg/scheduler/objects/application_test.go | 19 ++++++++++++------- pkg/scheduler/objects/preemption_test.go | 5 ++++- .../objects/required_node_preemptor.go | 4 ++-- 3 files changed, 18 insertions(+), 10 deletions(-) diff --git a/pkg/scheduler/objects/application_test.go b/pkg/scheduler/objects/application_test.go index 913887476..140d8bac6 100644 --- a/pkg/scheduler/objects/application_test.go +++ b/pkg/scheduler/objects/application_test.go @@ -3919,20 +3919,24 @@ func TestRequiredNodePreemptionWithPredicates(t *testing.T) { rightNodes := make(map[string]int, 1) rightNodes[nodeID1] = 10 + type resInfo struct { + res *resources.Resource + createTime time.Time + } res1 := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 15}) res2 := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 15}) res3 := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}) - ask1Info := make(map[string]*resources.Resource) - ask1Info["ask-1"] = res1 + ask1Info := make(map[string]*resInfo) + ask1Info["ask-1"] = &resInfo{res1, time.Now().Add(-30 * time.Second)} - ask2Info := make(map[string]*resources.Resource) - ask2Info["ask-1"] = res3 - ask2Info["ask-3"] = res2 + ask2Info := make(map[string]*resInfo) + ask2Info["ask-1"] = &resInfo{res3, time.Now().Add(-60 * time.Second)} + ask2Info["ask-3"] = &resInfo{res2, time.Now().Add(-30 * time.Second)} tests := []struct { name string - asks map[string]*resources.Resource + asks map[string]*resInfo mockPlugin *mockCommon.PreemptionPredicatePlugin result bool mockPluginError error @@ -3959,7 +3963,8 @@ func TestRequiredNodePreemptionWithPredicates(t *testing.T) { // add an ask askInfo := make(map[string]*Allocation) for k, v := range tt.asks { - ask := newAllocationAsk(k, "app-1", v) + ask := newAllocationAsk(k, "app-1", v.res) + ask.createTime = v.createTime err = app.AddAllocationAsk(ask) assert.NilError(t, err, "could not add ask "+k) askInfo[k] = ask diff --git a/pkg/scheduler/objects/preemption_test.go b/pkg/scheduler/objects/preemption_test.go index 91363988a..c9243c312 100644 --- a/pkg/scheduler/objects/preemption_test.go +++ b/pkg/scheduler/objects/preemption_test.go @@ -96,7 +96,10 @@ func creatApp1WithTwoDifferentAllocations( } else { alloc2 = newAllocationWithKey("alloc2", appID1, nodeID1, resources.NewResourceFromMap(app2Rec)) alloc2.createTime = ask2.createTime - app1.AddAllocation(alloc2) + err := app1.AddAllocation(alloc2) + if err != nil { + return nil, nil, err + } if !node1.TryAddAllocation(alloc2) { return nil, nil, fmt.Errorf("node alloc2 failed") } diff --git a/pkg/scheduler/objects/required_node_preemptor.go b/pkg/scheduler/objects/required_node_preemptor.go index c1f30bdb7..7dec22e28 100644 --- a/pkg/scheduler/objects/required_node_preemptor.go +++ b/pkg/scheduler/objects/required_node_preemptor.go @@ -160,7 +160,7 @@ func (p *PreemptionContext) sortAllocations() { } func (p *PreemptionContext) pruneVictims(victims []*Allocation) []*Allocation { - var prunedVictims []*Allocation + prunedVictims := make([]*Allocation, 0, len(victims)) var currentResource = resources.NewResource() for _, allocation := range victims { if !resources.StrictlyGreaterThanOrEquals(currentResource, p.requiredAsk.GetAllocatedResource()) { @@ -230,7 +230,7 @@ func (p *PreemptionContext) runPredicates() []*Allocation { zap.String("allocationKey", p.requiredAsk.GetAllocationKey()), zap.String("allocationName", p.requiredAsk.GetAllocationName()), zap.Int("victims", len(victims)), - zap.Int("passed victims", len(passedVictims))) + zap.Int("passedVictims", len(passedVictims))) } return passedVictims }