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 11d44cca [YUNIKORN-3319] Track reservations released by preemption
(#1120)
11d44cca is described below
commit 11d44cca4ca528c89e03c0e1476e496d829834c9
Author: stantheman0128 <[email protected]>
AuthorDate: Mon Aug 10 13:47:51 2026 +0530
[YUNIKORN-3319] Track reservations released by preemption (#1120)
initWorkingState() cancels stale reservations while building the working
state for TryPreemption, but the released count was discarded on every
return path, including the success path. Report it through the
reservation-released callback added in YUNIKORN-3321 so the partition
level reservation counter stays accurate whether or not preemption
itself succeeds.
Closes: #1120
Signed-off-by: mani <[email protected]>
---
pkg/scheduler/objects/preemption.go | 6 +-
pkg/scheduler/objects/preemption_test.go | 99 ++++++++++++++++++++++++++++++++
2 files changed, 104 insertions(+), 1 deletion(-)
diff --git a/pkg/scheduler/objects/preemption.go
b/pkg/scheduler/objects/preemption.go
index 211e883b..12c0a08b 100644
--- a/pkg/scheduler/objects/preemption.go
+++ b/pkg/scheduler/objects/preemption.go
@@ -590,7 +590,11 @@ func (p *Preemptor) TryPreemption() (*AllocationResult,
bool) {
}
// ensure required data structures are populated
- p.initWorkingState()
+ // building the data structures can cancel reservations on the nodes we
walk. Those reservations are
+ // gone whether or not the preemption below succeeds, so the release
must be accounted for here
+ // instead of on the return paths: all but one of them bail out without
an allocation result.
+ released := p.initWorkingState()
+ p.application.executeReservationReleasedCallback(released)
// try to find a node to schedule on and victims to preempt
nodeID, victims, ok := p.tryNodes()
diff --git a/pkg/scheduler/objects/preemption_test.go
b/pkg/scheduler/objects/preemption_test.go
index 627eaa34..333f6cd3 100644
--- a/pkg/scheduler/objects/preemption_test.go
+++ b/pkg/scheduler/objects/preemption_test.go
@@ -2267,6 +2267,14 @@ func Test_PreemptForAppOnReservedNode(t *testing.T) {
app.SetQueue(childQ)
childQ.AddApplication(app)
+ // the stale reservation below is cancelled while preemption builds its
working state, and this run
+ // does not preempt anything. The release must still be reported so the
partition level counter can
+ // be corrected. The callback runs on its own go routine, so collect
the value over a channel.
+ releasedCh := make(chan int, 2)
+ app.SetReservationReleasedCallback(func(released int) {
+ releasedCh <- released
+ })
+
// (1) no-priority ask, RESERVED on node1 ask does not fit in free node
resources
askLow := newAllocationAsk(aKey, appID1,
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}))
// must be more than wait timeout (default 60min) + delay (default
2sec) ago
@@ -2292,4 +2300,95 @@ func Test_PreemptForAppOnReservedNode(t *testing.T) {
result :=
app.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first":
10}), true, 1*time.Second, &remaining, iterator, iterator, getNode)
assert.Assert(t, result != nil, "expected and allocation result back")
assert.Equal(t, result.ResultType, Reserved, "expected result type to
be Reserved")
+
+ // no victims exist on the node so the preemption itself never runs:
the high priority ask is not
+ // marked, and the reservation that was returned belongs to the low
priority ask that was released above
+ assert.Check(t, !askHigh.HasTriggeredPreemption(), "preemption should
not have been triggered")
+ assert.Equal(t, result.Request.GetAllocationKey(), aKey, "expected the
released ask to be reserved again")
+ select {
+ case released := <-releasedCh:
+ assert.Equal(t, released, 1, "expected 1 released reservation
to be reported")
+ case <-time.After(5 * time.Second):
+ t.Fatal("timed out waiting for the released reservation to be
reported")
+ }
+}
+
+// Test_PreemptReleasesReservationsOnSuccess checks the reservation release
accounting for the run that
+// does preempt. A stale low priority reservation is cancelled while the
working state is built, and the
+// preemption then succeeds on that same node. The release must be reported
exactly once, and it must not
+// also be carried on the allocation result: the partition decrements that
separately.
+func Test_PreemptReleasesReservationsOnSuccess(t *testing.T) {
+ appQueueMapping := NewAppQueueMapping()
+ node1 := newNode(nodeID1, map[string]resources.Quantity{"first": 5,
"pods": 1})
+ node2 := newNode(nodeID2, map[string]resources.Quantity{"first": 5,
"pods": 1})
+ iterator := getNodeIteratorFn(node1, node2)
+ rootQ, err := createRootQueue(map[string]string{"first": "10", "pods":
"2"})
+ assert.NilError(t, err)
+ parentQ, err := createManagedQueueGuaranteed(rootQ, "parent", true,
map[string]string{"first": "20"}, map[string]string{"first": "10"},
appQueueMapping)
+ assert.NilError(t, err)
+ childQ1, err := createManagedQueueGuaranteed(parentQ, "child1", false,
map[string]string{"first": "10"}, map[string]string{"first": "5"},
appQueueMapping)
+ assert.NilError(t, err)
+ childQ2, err := createManagedQueueGuaranteed(parentQ, "child2", false,
map[string]string{"first": "10"}, map[string]string{"first": "5"},
appQueueMapping)
+ assert.NilError(t, err)
+
+ alloc1, alloc2, err := creatApp1(childQ1, node1, node2,
map[string]resources.Quantity{"first": 5, "pods": 1}, appQueueMapping)
+ assert.NilError(t, err)
+
+ app2 := newApplication(appID2, "default", childQ2.QueuePath)
+ app2.SetQueue(childQ2)
+ childQ2.AddApplication(app2)
+ appQueueMapping.AddAppQueueMapping(app2.ApplicationID, childQ2)
+
+ releasedCh := make(chan int, 2)
+ app2.SetReservationReleasedCallback(func(released int) {
+ releasedCh <- released
+ })
+
+ // stale low priority ask holding a reservation on the node the
preemption will pick
+ askReserved := newAllocationAsk("alloc4", appID2,
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5, "pods":
1}))
+ askReserved.createTime = time.Now().Add(-90 * time.Minute)
+ assert.NilError(t, app2.AddAllocationAsk(askReserved), "ask addition to
app should not have failed")
+ assert.NilError(t, app2.Reserve(node2, askReserved), "reservation on
node should not have failed")
+ nodeRes := node2.GetReservations()
+ assert.Equal(t, len(nodeRes), 1, "expected 1 reservation")
+ // must be older than the reservation wait timeout for the cancellation
to trigger
+ nodeRes[0].createTime = nodeRes[0].createTime.Add(-90 * time.Minute)
+
+ // higher priority ask that drives the preemption
+ ask3 := newAllocationAskPriority("alloc3", appID2,
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5, "pods":
1}), 100)
+ assert.NilError(t, app2.AddAllocationAsk(ask3), "ask addition to app
should not have failed")
+
+ headRoom :=
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 10, "pods":
3})
+ preemptor := NewPreemptor(app2, headRoom, 30*time.Second, ask3,
iterator(), false)
+
+ // register predicate handler
+ preemptions := []mock.Preemption{
+ mock.NewPreemption(true, "alloc3", nodeID2, []string{"alloc2"},
0, 0),
+ }
+ plugin := mock.NewPreemptionPredicatePlugin(nil, nil, preemptions)
+ plugins.RegisterSchedulerPlugin(plugin)
+ defer plugins.UnregisterSchedulerPlugins()
+
+ result, ok := preemptor.TryPreemption()
+ assert.Assert(t, result != nil, "no result")
+ assert.Assert(t, ok, "no victims found")
+ assert.Equal(t, result.NodeID, nodeID2, "wrong node")
+ assert.Check(t, !alloc1.IsPreempted(), "alloc1 preempted")
+ assert.Check(t, alloc2.IsPreempted(), "alloc2 not preempted")
+ // the release is reported over the callback, not on the result: the
partition applies both and would
+ // otherwise decrement twice for the same cancellation
+ assert.Equal(t, result.CancelledReservations, 0, "release must not be
carried on the allocation result")
+ assert.Equal(t, len(node2.GetReservations()), 0, "reservation should
have been cancelled")
+
+ select {
+ case released := <-releasedCh:
+ assert.Equal(t, released, 1, "expected 1 released reservation
to be reported")
+ case <-time.After(5 * time.Second):
+ t.Fatal("timed out waiting for the released reservation to be
reported")
+ }
+ select {
+ case extra := <-releasedCh:
+ t.Fatalf("released reservation reported twice, second report
was %d", extra)
+ case <-time.After(100 * time.Millisecond):
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
