[YUNIKORN-3092] Reservations can permanently block nodes, leading to …

pkg/scheduler/objects/application.go

@@ -47,7 +47,9 @@ import (
 )
 
 var (
-	reservationDelay          = 2 * time.Second
+	reservationDelay       = 2 * time.Second
+	reservationWaitTimeout = 60 * time.Minute
+	// Make it configurable
 	completingTimeout         = 30 * time.Second
 	terminatedTimeout         = 3 * 24 * time.Hour
 	defaultPlaceholderTimeout = 15 * time.Minute

pkg/scheduler/objects/application_test.go

@@ -2248,6 +2248,26 @@ func TestTryAllocatePreemptQueue(t *testing.T) {
 }
 
 func TestTryAllocatePreemptNode(t *testing.T) {
+	iterator, getNode, _, ask3, _, app2, allocs := createPreemptNodeTestSetup(t)
+	preemptionAttemptsRemaining := 10
+
+	// preemption delay not yet passed, so preemption should fail
+	result3 := app2.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 18}), true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode)
+	assert.Assert(t, result3 == nil, "result3 expected")
+	assert.Assert(t, !allocs[1].IsPreempted(), "alloc1 should have been preempted")
+	assertAllocationLog(t, ask3)
+
+	// pass the time and try again
+	ask3.createTime = ask3.createTime.Add(-30 * time.Second)
+	result3 = app2.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 18}), true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode)
+	assert.Assert(t, result3 != nil, "result3 expected")
+	assert.Equal(t, Reserved, result3.ResultType, "expected reservation")
+	alloc3 := result3.Request
+	assert.Assert(t, alloc3 != nil, "alloc3 expected")
+	assert.Assert(t, allocs[0].IsPreempted(), "alloc1 should have been preempted")
+}
+
+func createPreemptNodeTestSetup(t *testing.T) (func() NodeIterator, func(NodeID string) *Node, *Queue, *Allocation, *Application, *Application, []*Allocation) {
 	node1 := newNode("node1", map[string]resources.Quantity{"first": 20})
 	node2 := newNode("node2", map[string]resources.Quantity{"first": 20})
 	nodeMap := map[string]*Node{"node1": node1, "node2": node2}
@@ -2330,18 +2350,124 @@ func TestTryAllocatePreemptNode(t *testing.T) {
 	err = node1.Reserve(app2, ask3)
 	assert.NilError(t, err)
 
-	// preemption delay not yet passed, so preemption should fail
-	result3 = app2.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 18}), true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode)
-	assert.Assert(t, result3 == nil, "result3 expected")
-	assert.Assert(t, !allocs[1].IsPreempted(), "alloc1 should have been preempted")
-	assertAllocationLog(t, ask3)
+	return iterator, getNode, childQ2, ask3, app1, app2, allocs
+}
+
+func createPreemptNodeWithReservationsTestSetup(t *testing.T) (func() NodeIterator, func(NodeID string) *Node, *Allocation, *Allocation, *Application, *Application, []*Allocation) {
+	iterator, getNode, childQ2, ask3, app1, _, allocs := createPreemptNodeTestSetup(t)
+
+	app3 := newApplication(appID3, "default", "root.parent.child2")
+	app3.SetQueue(childQ2)
+	childQ2.applications[appID3] = app3
+	ask4 := newAllocationAsk("alloc4", appID3, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}))
+	ask4.allowPreemptOther = true
+	ask4.priority = math.MaxInt32
+	err := app3.AddAllocationAsk(ask4)
+	assert.NilError(t, err)
+
+	return iterator, getNode, ask3, ask4, app1, app3, allocs
+}
+
+func TestTryAllocatePreemptNodeWithReservations(t *testing.T) {
+	iterator, getNode, _, ask4, _, app3, allocs := createPreemptNodeWithReservationsTestSetup(t)
+
+	preemptionAttemptsRemaining := 10
 
 	// pass the time and try again
-	ask3.createTime = ask3.createTime.Add(-30 * time.Second)
-	result3 = app2.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 18}), true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode)
+	ask4.createTime = ask4.createTime.Add(-30 * time.Second)
+	reservationWaitTimeout = -60 * time.Second
+	result3 := app3.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 18}), true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode)
 	assert.Assert(t, result3 != nil, "result3 expected")
 	assert.Equal(t, Reserved, result3.ResultType, "expected reservation")
-	alloc3 = result3.Request
+	alloc3 := result3.Request
+	assert.Assert(t, alloc3 != nil, "alloc3 expected")
+	assert.Assert(t, allocs[0].IsPreempted(), "alloc1 should have been preempted")
+}
+
+func TestTryAllocatePreemptNodeWithReservationsWithHighPriority(t *testing.T) {
+	iterator, getNode, _, ask4, _, app3, allocs := createPreemptNodeWithReservationsTestSetup(t)
+
+	// Make ask (preemptor) priority lower than the reserved asks (victims) so that preemption would not yield positive outcome
+	ask4.priority = -1
+
+	preemptionAttemptsRemaining := 10
+
+	// pass the time and try again
+	ask4.createTime = ask4.createTime.Add(-30 * time.Second)
+	reservationWaitTimeout = -60 * time.Second
+	result3 := app3.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 18}), true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode)
+	assert.Assert(t, result3 == nil, "result3 expected")
+
+	// Set higher priority than the reserved ask priority
+	ask4.priority = math.MaxInt32
+	ask4.preemptCheckTime = ask4.preemptCheckTime.Add(-30 * time.Second)
+	result4 := app3.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 18}), true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode)
+	assert.Assert(t, result4 != nil, "result3 expected")
+	assert.Equal(t, Reserved, result4.ResultType, "expected reservation")
+	alloc3 := result4.Request
+	assert.Assert(t, alloc3 != nil, "alloc3 expected")
+	assert.Assert(t, allocs[0].IsPreempted(), "alloc1 should have been preempted")
+}
+
+// TestTryAllocatePreemptNodeWithReservationsNotPossibleToCancel Ensures reservations cannot be cancelled because of the following constraints:
+// 1. Reservation Wait Time out not exceeded
+// 2. Reserved allocation has required node set
+// 3. Reserved allocation has already marked with "triggered preemption" flag
+// Preemption would start yielding results once above constraints has been resolved
+func TestTryAllocatePreemptNodeWithReservationsNotPossibleToCancel(t *testing.T) {
+	iterator, getNode, ask3, ask4, app1, app3, allocs := createPreemptNodeWithReservationsTestSetup(t)
+
+	// Make reservation ask (victim) as already "triggered preemption" so that it won't be considered for cancellation.
+	ask3.MarkTriggeredPreemption()
+
+	// Make reservation ask (victim) as daemon-set so that it won't be considered for cancellation.
+	ask5 := newAllocationAsk("alloc5", appID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}))
+	ask5.requiredNode = "node2"
+	err := app1.AddAllocationAsk(ask5)
+	assert.NilError(t, err)
+
+	preemptionAttemptsRemaining := 10
+
+	// on first attempt, should see a reservation on node2 since we're after the reservation timeout
+	var alloc11 *Allocation
+	ask5.createTime = ask5.createTime.Add(-10 * time.Second)
+	result1 := app1.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 18}), true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode)
+	assert.Assert(t, result1 != nil, "result1 expected")
+	alloc11 = result1.Request
+	assert.Equal(t, "alloc5", alloc11.allocationKey, "wrong node assignment")
+	assert.Assert(t, result1.Request != nil, "alloc1 expected")
+	assert.Equal(t, "node2", result1.NodeID, "wrong node assignment")
+	assert.Equal(t, Reserved, result1.ResultType, "expected reservation")
+	allocs = append(allocs, alloc11)
+	err = getNode("node2").Reserve(app1, ask5)
+	assert.NilError(t, err)
+
+	// Set higher priority than the reserved ask priority but no preemption because reserved ask waiting time not exceeded
+	ask4.priority = 1
+	result3 := app3.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 18}), true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode)
+	assert.Assert(t, result3 == nil, "result3 expected")
+
+	// Ensure reserved ask waiting time exceeds
+	// Both Node 1 & 2 has reservations, one allocation has required node set and another had marked for "triggered preemption" flag
+	// Still, preemption doesn't yield any positive outcome
+	ask4.createTime = ask4.createTime.Add(-30 * time.Second)
+	reservationWaitTimeout = -60 * time.Second
+	ask4.preemptCheckTime = ask4.preemptCheckTime.Add(-30 * time.Second)
+	result4 := app3.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 18}), true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode)
+	assert.Assert(t, result4 == nil, "result3 expected")
+
+	// Ensure reserved ask waiting time exceeds
+	// Ensure reserved allocation doesn't have required node set and not marked for "triggered preemption" flag
+	// Still, preemption doesn't yield any positive outcome
+	ask5.requiredNode = ""
+	ask3.preemptionTriggered = false
+	ask4.createTime = ask4.createTime.Add(-30 * time.Second)
+	reservationWaitTimeout = -60 * time.Second
+	ask4.preemptCheckTime = ask4.preemptCheckTime.Add(-30 * time.Second)
+	result5 := app3.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 18}), true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode)
+	assert.Assert(t, result5 != nil, "result3 expected")
+	assert.Equal(t, Reserved, result5.ResultType, "expected reservation")
+	alloc3 := result5.Request
 	assert.Assert(t, alloc3 != nil, "alloc3 expected")
 	assert.Assert(t, allocs[0].IsPreempted(), "alloc1 should have been preempted")
 }

pkg/scheduler/objects/preemption.go

@@ -161,7 +161,35 @@ func (p *Preemptor) initWorkingState() {
 
 	// walk node iterator and track available resources per node
 	p.iterator.ForEachNode(func(node *Node) bool {
-		if !node.IsSchedulable() || (node.IsReserved() && !node.isReservedForAllocation(p.ask.GetAllocationKey())) || !node.FitInNode(p.ask.GetAllocatedResource()) {
+		hasOtherReservations := false
+		if node.IsReserved() && !node.isReservedForAllocation(p.ask.GetAllocationKey()) {
+			hasOtherReservations = true
+			for _, res := range node.GetReservations() {
+				// Is Allocation daemon set?
+				// Has this allocation already triggered preemption?
+				if res.alloc.requiredNode != "" || res.alloc.HasTriggeredPreemption() {
+					continue
+				}
+				createTime := res.alloc.GetCreateTime()
+				// Take reservation delay also into account
+				askAge := time.Since(createTime.Add(reservationWaitTimeout).Add(reservationDelay))
+
+				// Cancel reservation based on its priority and waiting time in reservation queue
+				if res.alloc.GetPriority() < p.ask.priority && askAge > reservationWaitTimeout {
+					num := res.app.UnReserve(res.node, res.alloc)
+					res.app.GetQueue().UnReserve(res.app.ApplicationID, num)
+					log.Log(log.SchedApplication).Info("Cancelled reservation to consider node for preemption",
+						zap.String("triggered by appID", p.application.ApplicationID),
+						zap.String("triggered by allocationKey", p.ask.allocationKey),
+						zap.String("affected application ID", res.appID),
+						zap.String("affected allocationKey", res.allocKey),
+						zap.String("node", res.nodeID),
+						zap.Int("reservations count", num))
+					hasOtherReservations = false
+				}
+			}
+		}
+		if !node.IsSchedulable() || hasOtherReservations || !node.FitInNode(p.ask.GetAllocatedResource()) {
 			// node is not available, remove any potential victims from consideration
 			delete(allocationsByNode, node.NodeID)
 		} else {