You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@yunikorn.apache.org by ma...@apache.org on 2026/10/01 08:13:15 UTC
(yunikorn-core) branch master updated: [YUNIKORN-3453] Send the plugin errors back to core for preemption ask (#1160)
This is an automated email from the ASF dual-hosted git repository.
manirajv06 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-core.git
The following commit(s) were added to refs/heads/master by this push:
new fc4c1981 [YUNIKORN-3453] Send the plugin errors back to core for preemption ask (#1160)
fc4c1981 is described below
commit fc4c198191aadb885a25dbfec63ec554b20433e4
Author: mani <ma...@gmail.com>
AuthorDate: Thu Oct 1 13:43:03 2026 +0530
[YUNIKORN-3453] Send the plugin errors back to core for preemption ask (#1160)
Closes: #1160
Signed-off-by: mani <ma...@gmail.com>
---
pkg/mock/preemption_predicate_plugin.go | 7 +++++--
pkg/scheduler/objects/predicates.go | 23 ++++++++++++++---------
pkg/scheduler/objects/preemption.go | 18 +++++++++++++-----
pkg/scheduler/objects/preemption_test.go | 11 +++++++++++
4 files changed, 43 insertions(+), 16 deletions(-)
diff --git a/pkg/mock/preemption_predicate_plugin.go b/pkg/mock/preemption_predicate_plugin.go
index 407a7f4e..cfb34b04 100644
--- a/pkg/mock/preemption_predicate_plugin.go
+++ b/pkg/mock/preemption_predicate_plugin.go
@@ -60,6 +60,7 @@ func (m *PreemptionPredicatePlugin) PreFilterPredicates(args *si.PreFilterPredic
if m.mustPreFilterFail {
m.errHolder.err = fmt.Errorf("fake preemption predicate prefilter plugin failed")
+ result.ErrorMessage = m.errHolder.err.Error()
return result
}
for k, v := range m.nodes {
@@ -86,11 +87,13 @@ func (m *PreemptionPredicatePlugin) PreemptionPredicates(args *si.PreemptionPred
m.Lock()
defer m.Unlock()
result := &si.PreemptionPredicatesResponse{
- Success: false,
- Index: -1,
+ Success: false,
+ Index: -1,
+ ErrorMessage: make(map[string]int32),
}
if m.mustFilterFail {
m.errHolder.err = fmt.Errorf("fake preemption predicate filter plugin failed")
+ result.ErrorMessage[m.errHolder.err.Error()]++
return result
}
for _, preemption := range m.preemptions {
diff --git a/pkg/scheduler/objects/predicates.go b/pkg/scheduler/objects/predicates.go
index e1730e68..179b01dc 100644
--- a/pkg/scheduler/objects/predicates.go
+++ b/pkg/scheduler/objects/predicates.go
@@ -31,11 +31,12 @@ import (
)
type predicateCheckResult struct {
- allocationKey string
- nodeID string
- success bool
- index int
- victims []*Allocation
+ allocationKey string
+ nodeID string
+ success bool
+ index int
+ victims []*Allocation
+ predicateErrors map[string]int32
}
func (pcr *predicateCheckResult) betterThan(other *predicateCheckResult, allocationsByNode map[string][]*Allocation) bool {
@@ -117,10 +118,11 @@ func (pcr *predicateCheckResult) populateVictims(victimsByNode map[string][]*All
func preemptPredicateCheck(plugin api.ResourceManagerCallback, ch chan<- *predicateCheckResult, wg *sync.WaitGroup, args *si.PreemptionPredicatesArgs) {
defer wg.Done()
result := &predicateCheckResult{
- allocationKey: args.AllocationKey,
- nodeID: args.NodeID,
- success: false,
- index: -1,
+ 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
@@ -136,12 +138,15 @@ func preemptPredicateCheck(plugin api.ResourceManagerCallback, ch chan<- *predic
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()
}
}
ch <- result
diff --git a/pkg/scheduler/objects/preemption.go b/pkg/scheduler/objects/preemption.go
index 2dbdd18e..3dd95e7e 100644
--- a/pkg/scheduler/objects/preemption.go
+++ b/pkg/scheduler/objects/preemption.go
@@ -380,10 +380,10 @@ func (p *Preemptor) duplicateQueueSnapshots() map[string]*QueuePreemptionSnapsho
}
// checkPreemptionPredicates calls the shim via the SI to evaluate nodes for preemption
-func (p *Preemptor) checkPreemptionPredicates(predicateChecks []*si.PreemptionPredicatesArgs, victimsByNode map[string][]*Allocation) *predicateCheckResult {
+func (p *Preemptor) checkPreemptionPredicates(predicateChecks []*si.PreemptionPredicatesArgs, victimsByNode map[string][]*Allocation) (*predicateCheckResult, map[string]int) {
// don't process empty list
if len(predicateChecks) == 0 {
- return nil
+ return nil, nil
}
// sort predicate checks by number of expected preempted tasks
@@ -411,12 +411,13 @@ func (p *Preemptor) checkPreemptionPredicates(predicateChecks []*si.PreemptionPr
index: int(check.StartIndex),
}
result.populateVictims(victimsByNode)
- return result
+ return result, nil
}
// process each batch of checks by sending to the RM
batches := batchPreemptionChecks(predicateChecks, preemptCheckConcurrency)
var bestResult *predicateCheckResult = nil
+ predicateErrors := make(map[string]int)
for _, batch := range batches {
var wg sync.WaitGroup
ch := make(chan *predicateCheckResult, len(batch))
@@ -440,6 +441,10 @@ func (p *Preemptor) checkPreemptionPredicates(predicateChecks []*si.PreemptionPr
} else if result.betterThan(bestResult, p.allocationsByNode) {
bestResult = result
}
+ } else {
+ for e := range result.predicateErrors {
+ predicateErrors[e]++
+ }
}
}
// if the best resultType we have from this batch meets all our criteria, don't run another batch
@@ -448,7 +453,7 @@ func (p *Preemptor) checkPreemptionPredicates(predicateChecks []*si.PreemptionPr
}
}
bestResult.populateVictims(victimsByNode)
- return bestResult
+ return bestResult, predicateErrors
}
// calculateAdditionalVictims finds additional preemption victims necessary to ensure
@@ -589,10 +594,13 @@ func (p *Preemptor) tryNodes() (string, []*Allocation, bool) {
}
}
// call predicates to evaluate each node
- result := p.checkPreemptionPredicates(predicateChecks, victimsByNode)
+ result, predicateErrors := p.checkPreemptionPredicates(predicateChecks, victimsByNode)
if result != nil && result.success {
return result.nodeID, result.victims, true
}
+ if len(predicateErrors) > 0 {
+ p.ask.SendPredicatesFailedEvent(predicateErrors)
+ }
return "", nil, false
}
diff --git a/pkg/scheduler/objects/preemption_test.go b/pkg/scheduler/objects/preemption_test.go
index 220cfe2b..d3406a6c 100644
--- a/pkg/scheduler/objects/preemption_test.go
+++ b/pkg/scheduler/objects/preemption_test.go
@@ -22,6 +22,7 @@ import (
"errors"
"fmt"
"strconv"
+ "strings"
"testing"
"time"
@@ -382,6 +383,8 @@ func TestTryPreemption(t *testing.T) {
app2, ask3, err := creatApp2(childQ2, map[string]resources.Quantity{"first": 5, "pods": 1}, "alloc3", appQueueMapping)
assert.NilError(t, err)
childQ2.incPendingResource(ask3.GetAllocatedResource())
+ eventSystem := evtMock.NewEventSystem()
+ ask3.askEvents = schedEvt.NewAskEvents(eventSystem)
headRoom := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 10, "pods": 3})
preemptor := NewPreemptor(app2, headRoom, 30*time.Second, ask3, iterator(), false)
plugins.RegisterSchedulerPlugin(tt.mockPlugin)
@@ -399,6 +402,9 @@ func TestTryPreemption(t *testing.T) {
}
if tt.mockPluginError != nil {
assert.ErrorContains(t, tt.mockPlugin.GetPredicateError(), tt.mockPluginError.Error())
+ assert.Equal(t, 1, len(eventSystem.Events))
+ event := eventSystem.Events[0]
+ assert.Assert(t, strings.Contains(event.Message, "plugin failed"))
}
// reset
resetNode(node)
@@ -665,6 +671,8 @@ func TestTryPreemptionOnQueue(t *testing.T) {
assert.NilError(t, err)
app2, ask3, err := creatApp2(childQ2, map[string]resources.Quantity{"first": 5, "pods": 1}, "alloc3", appQueueMapping)
assert.NilError(t, err)
+ eventSystem := evtMock.NewEventSystem()
+ ask3.askEvents = schedEvt.NewAskEvents(eventSystem)
headRoom := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 10, "pods": 3})
preemptor := NewPreemptor(app2, headRoom, 30*time.Second, ask3, iterator(), false)
plugins.RegisterSchedulerPlugin(tt.mockPlugin)
@@ -686,6 +694,9 @@ func TestTryPreemptionOnQueue(t *testing.T) {
}
if tt.mockPluginError != nil {
assert.ErrorContains(t, tt.mockPlugin.GetPredicateError(), tt.mockPluginError.Error())
+ assert.Equal(t, 1, len(eventSystem.Events))
+ event := eventSystem.Events[0]
+ assert.Assert(t, strings.Contains(event.Message, "plugin failed"))
}
// reset
resetNode(node1)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@yunikorn.apache.org
For additional commands, e-mail: issues-help@yunikorn.apache.org