You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@yunikorn.apache.org by cc...@apache.org on 2024/12/13 17:22:53 UTC
(yunikorn-core) branch master updated: [YUNIKORN-2976] Handle multiple require node allocations per node (#1001)
This is an automated email from the ASF dual-hosted git repository.
ccondit 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 b02e15ea [YUNIKORN-2976] Handle multiple require node allocations per node (#1001)
b02e15ea is described below
commit b02e15ea98429f34a3eb623f28f2911ea5a8e98d
Author: Wilfred Spiegelenburg <wi...@apache.org>
AuthorDate: Fri Dec 13 11:22:18 2024 -0600
[YUNIKORN-2976] Handle multiple require node allocations per node (#1001)
If an allocation requires a specific node the scheduler should not
consider any other node. We should allow multiple allocations that
require the same node to reserve the node at the same time. A required
node allocation must be placed on the node before anything else.
If other non required node reservations are made on a node remove the
existing reservations that do not require that node. Make sure that the
releases are tracked correctly in the partition.
After the repeat count removal reservations can be simplified:
- track reservations using the allocation key
- removed the composite key setup
- removed collection listener call on reserve or unreserve of a node
Closes: #1001
Signed-off-by: Craig Condit <cc...@apache.org>
---
pkg/common/errors.go | 27 +-
pkg/scheduler/objects/allocation.go | 20 +-
pkg/scheduler/objects/allocation_result.go | 9 +-
pkg/scheduler/objects/allocation_test.go | 1 -
pkg/scheduler/objects/application.go | 452 +++++++++++++-------------
pkg/scheduler/objects/application_test.go | 356 ++++++++++++++++----
pkg/scheduler/objects/node.go | 123 +++----
pkg/scheduler/objects/node_collection_test.go | 163 ++++++----
pkg/scheduler/objects/node_test.go | 163 ++++++----
pkg/scheduler/objects/preemption.go | 2 +-
pkg/scheduler/objects/reservation.go | 71 ++--
pkg/scheduler/objects/reservation_test.go | 91 +++---
pkg/scheduler/objects/utilities_test.go | 1 +
pkg/scheduler/partition.go | 32 +-
pkg/scheduler/partition_test.go | 92 ++++--
15 files changed, 929 insertions(+), 674 deletions(-)
diff --git a/pkg/common/errors.go b/pkg/common/errors.go
index f73c84b4..7e28426b 100644
--- a/pkg/common/errors.go
+++ b/pkg/common/errors.go
@@ -20,11 +20,24 @@ package common
import "errors"
-// InvalidQueueName returned when queue name is invalid
-var InvalidQueueName = errors.New("invalid queue name, max 64 characters consisting of alphanumeric characters and '-', '_', '#', '@', '/', ':' allowed")
+var (
+ // InvalidQueueName returned when queue name is invalid
+ InvalidQueueName = errors.New("invalid queue name, max 64 characters consisting of alphanumeric characters and '-', '_', '#', '@', '/', ':' allowed")
+ // ErrorReservingAlloc returned when an ask that is allocated tries to reserve a node.
+ ErrorReservingAlloc = errors.New("ask already allocated, no reservation allowed")
+ // ErrorDuplicateReserve returned when the same reservation already exists on the application
+ ErrorDuplicateReserve = errors.New("reservation already exists")
+ // ErrorNodeAlreadyReserved returned when the node is already reserved, failing the reservation
+ ErrorNodeAlreadyReserved = errors.New("node is already reserved")
+ // ErrorNodeNotFitReserve returned when the allocation does not fit on an empty node, failing the reservation
+ ErrorNodeNotFitReserve = errors.New("reservation does not fit on node")
+)
-const PreemptionPreconditionsFailed = "Preemption preconditions failed"
-const PreemptionDoesNotGuarantee = "Preemption queue guarantees check failed"
-const PreemptionShortfall = "Preemption helped but short of resources"
-const PreemptionDoesNotHelp = "Preemption does not help"
-const NoVictimForRequiredNode = "No fit on required node, preemption does not help"
+// Constant messages for AllocationLog entries
+const (
+ PreemptionPreconditionsFailed = "Preemption preconditions failed"
+ PreemptionDoesNotGuarantee = "Preemption queue guarantees check failed"
+ PreemptionShortfall = "Preemption helped but short of resources"
+ PreemptionDoesNotHelp = "Preemption does not help"
+ NoVictimForRequiredNode = "No fit on required node, preemption does not help"
+)
diff --git a/pkg/scheduler/objects/allocation.go b/pkg/scheduler/objects/allocation.go
index a4701cd6..0ef05222 100644
--- a/pkg/scheduler/objects/allocation.go
+++ b/pkg/scheduler/objects/allocation.go
@@ -48,7 +48,6 @@ type Allocation struct {
allowPreemptOther bool
originator bool
tags map[string]string
- resKeyWithoutNode string // the reservation key without node
foreign bool
preemptable bool
@@ -57,9 +56,8 @@ type Allocation struct {
allocLog map[string]*AllocationLogEntry
preemptionTriggered bool
preemptCheckTime time.Time
- schedulingAttempted bool // whether scheduler core has tried to schedule this allocation
- scaleUpTriggered bool // whether this aloocation has triggered autoscaling or not
- resKeyPerNode map[string]string // reservation key for a given node
+ schedulingAttempted bool // whether scheduler core has tried to schedule this allocation
+ scaleUpTriggered bool // whether this allocation has triggered autoscaling or not
allocatedResource *resources.Resource
askEvents *schedEvt.AskEvents
userQuotaCheckFailed bool
@@ -145,8 +143,6 @@ func NewAllocationFromSI(alloc *si.Allocation) *Allocation {
allowPreemptOther: alloc.PreemptionPolicy.GetAllowPreemptOther(),
originator: alloc.Originator,
allocLog: make(map[string]*AllocationLogEntry),
- resKeyPerNode: make(map[string]string),
- resKeyWithoutNode: reservationKeyWithoutNode(alloc.ApplicationID, alloc.AllocationKey),
askEvents: schedEvt.NewAskEvents(events.GetEventSystem()),
allocated: allocated,
nodeID: nodeID,
@@ -554,18 +550,6 @@ func (a *Allocation) HasTriggeredScaleUp() bool {
return a.scaleUpTriggered
}
-func (a *Allocation) setReservationKeyForNode(node, resKey string) {
- a.Lock()
- defer a.Unlock()
- a.resKeyPerNode[node] = resKey
-}
-
-func (a *Allocation) getReservationKeyForNode(node string) string {
- a.RLock()
- defer a.RUnlock()
- return a.resKeyPerNode[node]
-}
-
func (a *Allocation) setHeadroomCheckFailed(headroom *resources.Resource, queue string) {
a.Lock()
defer a.Unlock()
diff --git a/pkg/scheduler/objects/allocation_result.go b/pkg/scheduler/objects/allocation_result.go
index bcd56a86..99af95b7 100644
--- a/pkg/scheduler/objects/allocation_result.go
+++ b/pkg/scheduler/objects/allocation_result.go
@@ -36,10 +36,11 @@ func (art AllocationResultType) String() string {
}
type AllocationResult struct {
- ResultType AllocationResultType
- Request *Allocation
- NodeID string
- ReservedNodeID string
+ ResultType AllocationResultType
+ Request *Allocation
+ NodeID string
+ ReservedNodeID string
+ CancelledReservations int
}
func (ar *AllocationResult) String() string {
diff --git a/pkg/scheduler/objects/allocation_test.go b/pkg/scheduler/objects/allocation_test.go
index 62bf316a..6917acac 100644
--- a/pkg/scheduler/objects/allocation_test.go
+++ b/pkg/scheduler/objects/allocation_test.go
@@ -51,7 +51,6 @@ func TestNewAsk(t *testing.T) {
askStr := ask.String()
expected := "allocationKey ask-1, applicationID app-1, Resource map[first:10], Allocated false"
assert.Equal(t, askStr, expected, "Strings should have been equal")
- assert.Equal(t, "app-1|ask-1", ask.resKeyWithoutNode) //nolint:staticcheck
}
func TestAskAllocateDeallocate(t *testing.T) {
diff --git a/pkg/scheduler/objects/application.go b/pkg/scheduler/objects/application.go
index 60ad55ca..adff3817 100644
--- a/pkg/scheduler/objects/application.go
+++ b/pkg/scheduler/objects/application.go
@@ -452,41 +452,44 @@ func (sa *Application) timeoutPlaceholderProcessing() {
func (sa *Application) GetReservations() []string {
sa.RLock()
defer sa.RUnlock()
- keys := make([]string, 0)
+ keys := make([]string, len(sa.reservations))
+ var i int
for key := range sa.reservations {
- keys = append(keys, key)
+ keys[i] = key
+ i++
}
return keys
}
-// Return the allocation ask for the key, nil if not found
+// GetAllocationAsk returns the allocation alloc for the key, nil if not found
func (sa *Application) GetAllocationAsk(allocationKey string) *Allocation {
sa.RLock()
defer sa.RUnlock()
return sa.requests[allocationKey]
}
-// Return the allocated resources for this application
+// GetAllocatedResource returns the currently allocated resources for this application
func (sa *Application) GetAllocatedResource() *resources.Resource {
sa.RLock()
defer sa.RUnlock()
return sa.allocatedResource.Clone()
}
+// GetMaxAllocatedResource returns the peak of the allocated resources for this application
func (sa *Application) GetMaxAllocatedResource() *resources.Resource {
sa.RLock()
defer sa.RUnlock()
return sa.maxAllocatedResource.Clone()
}
-// Return the allocated placeholder resources for this application
+// GetPlaceholderResource returns the currently allocated placeholder resources for this application
func (sa *Application) GetPlaceholderResource() *resources.Resource {
sa.RLock()
defer sa.RUnlock()
return sa.allocatedPlaceholder.Clone()
}
-// Return the total placeholder ask for this application
+// GetPlaceholderAsk returns the total placeholder resource request for this application
// Is only set on app creation and used when app is added to a queue
func (sa *Application) GetPlaceholderAsk() *resources.Resource {
sa.RLock()
@@ -501,8 +504,8 @@ func (sa *Application) GetPendingResource() *resources.Resource {
return sa.pending
}
-// Remove one or more allocation asks from this application.
-// This also removes any reservations that are linked to the ask.
+// RemoveAllocationAsk removes one or more allocation asks from this application.
+// This also removes any reservations that are linked to the allocations.
// The return value is the number of reservations released
func (sa *Application) RemoveAllocationAsk(allocKey string) int {
sa.Lock()
@@ -517,23 +520,16 @@ func (sa *Application) removeAsksInternal(allocKey string, detail si.EventRecord
return 0
}
var deltaPendingResource *resources.Resource = nil
- // when allocation key not specified, cleanup all allocation ask
+ // when allocation key is not specified, cleanup all allocations
var toRelease int
if allocKey == "" {
// cleanup all reservations
- for key, reserve := range sa.reservations {
- releases, err := sa.unReserveInternal(reserve.node, reserve.ask)
- if err != nil {
- log.Log(log.SchedApplication).Warn("Removal of reservation failed while removing all allocation asks",
- zap.String("appID", sa.ApplicationID),
- zap.String("reservationKey", key),
- zap.Error(err))
- continue
- }
- // clean up the queue reservation (one at a time)
- sa.queue.UnReserve(sa.ApplicationID, releases)
+ for _, reserve := range sa.reservations {
+ releases := sa.unReserveInternal(reserve)
toRelease += releases
}
+ // clean up the queue reservation
+ sa.queue.UnReserve(sa.ApplicationID, toRelease)
// Cleanup total pending resource
deltaPendingResource = sa.pending
sa.pending = resources.NewResource()
@@ -546,16 +542,8 @@ func (sa *Application) removeAsksInternal(allocKey string, detail si.EventRecord
sa.queue.UpdateApplicationPriority(sa.ApplicationID, sa.askMaxPriority)
} else {
// cleanup the reservation for this allocation
- for _, key := range sa.GetAskReservations(allocKey) {
- reserve := sa.reservations[key]
- releases, err := sa.unReserveInternal(reserve.node, reserve.ask)
- if err != nil {
- log.Log(log.SchedApplication).Warn("Removal of reservation failed while removing allocation ask",
- zap.String("appID", sa.ApplicationID),
- zap.String("reservationKey", key),
- zap.Error(err))
- continue
- }
+ if reserve, ok := sa.reservations[allocKey]; ok {
+ releases := sa.unReserveInternal(reserve)
// clean up the queue reservation
sa.queue.UnReserve(sa.ApplicationID, releases)
toRelease += releases
@@ -576,7 +564,7 @@ func (sa *Application) removeAsksInternal(allocKey string, detail si.EventRecord
}
// clean up the queue pending resources
sa.queue.decPendingResource(deltaPendingResource)
- // Check if we need to change state based on the ask removal:
+ // Check if we need to change state based on the removal:
// 1) if pending is zero (no more asks left)
// 2) if confirmed allocations is zero (no real tasks running)
// Change the state to completing.
@@ -808,152 +796,145 @@ func (sa *Application) HasReserved() bool {
return len(sa.reservations) > 0
}
-// IsReservedOnNode returns true if and only if the node has been reserved by the application
-// An empty nodeID is never reserved.
-func (sa *Application) IsReservedOnNode(nodeID string) bool {
- if nodeID == "" {
- return false
- }
+// NodeReservedForAsk returns the nodeID that has been reserved by the application for the ask
+// An empty nodeID means the ask is not reserved. An empty askKey is never reserved.
+func (sa *Application) NodeReservedForAsk(askKey string) string {
sa.RLock()
defer sa.RUnlock()
- // make sure matches only for the whole nodeID
- separator := nodeID + "|"
- for key := range sa.reservations {
- if strings.HasPrefix(key, separator) {
- return true
- }
+ if reserved, ok := sa.reservations[askKey]; ok {
+ return reserved.nodeID
}
- return false
+ return ""
}
-// Reserve the application for this node and ask combination.
+// Reserve the application for this node and alloc combination.
// If the reservation fails the function returns false, if the reservation is made it returns true.
-// If the node and ask combination was already reserved for the application this is a noop and returns true.
+// If the node and alloc combination was already reserved for the application this is a noop and returns true.
func (sa *Application) Reserve(node *Node, ask *Allocation) error {
+ if node == nil || ask == nil {
+ return fmt.Errorf("reservation creation failed node or alloc are nil on appID %s", sa.ApplicationID)
+ }
sa.Lock()
defer sa.Unlock()
return sa.reserveInternal(node, ask)
}
-// Unlocked version for Reserve that really does the work.
+// reserveInternal is the unlocked version for Reserve that really does the work.
// Must only be called while holding the application lock.
func (sa *Application) reserveInternal(node *Node, ask *Allocation) error {
+ allocKey := ask.GetAllocationKey()
+ if sa.requests[allocKey] == nil {
+ log.Log(log.SchedApplication).Debug("alloc is not registered to this app",
+ zap.String("app", sa.ApplicationID),
+ zap.String("allocKey", allocKey))
+ return fmt.Errorf("reservation creation failed alloc %s not found on appID %s", allocKey, sa.ApplicationID)
+ }
// create the reservation (includes nil checks)
nodeReservation := newReservation(node, sa, ask, true)
if nodeReservation == nil {
log.Log(log.SchedApplication).Debug("reservation creation failed unexpectedly",
zap.String("app", sa.ApplicationID),
- zap.Any("node", node),
- zap.Any("ask", ask))
- return fmt.Errorf("reservation creation failed node or ask are nil on appID %s", sa.ApplicationID)
- }
- allocKey := ask.GetAllocationKey()
- if sa.requests[allocKey] == nil {
- log.Log(log.SchedApplication).Debug("ask is not registered to this app",
- zap.String("app", sa.ApplicationID),
- zap.String("allocKey", allocKey))
- return fmt.Errorf("reservation creation failed ask %s not found on appID %s", allocKey, sa.ApplicationID)
+ zap.Stringer("node", node),
+ zap.Stringer("alloc", ask))
+ return fmt.Errorf("reservation creation failed node or alloc are nil on appID %s", sa.ApplicationID)
}
- if !sa.canAskReserve(ask) {
- if ask.IsAllocated() {
- return fmt.Errorf("ask is already allocated")
- } else {
- return fmt.Errorf("ask is already reserved")
- }
+ // the alloc should not have reserved a node yet: do not allow multiple nodes to be reserved
+ if err := sa.canAllocationReserve(ask); err != nil {
+ return err
}
// check if we can reserve the node before reserving on the app
if err := node.Reserve(sa, ask); err != nil {
return err
}
- sa.reservations[nodeReservation.getKey()] = nodeReservation
+ sa.reservations[allocKey] = nodeReservation
log.Log(log.SchedApplication).Info("reservation added successfully",
zap.String("app", sa.ApplicationID),
zap.String("node", node.NodeID),
- zap.String("ask", allocKey))
+ zap.String("alloc", allocKey))
return nil
}
-// UnReserve the application for this node and ask combination.
-// This first removes the reservation from the node.
+// UnReserve the application for this node and alloc combination.
// If the reservation does not exist it returns 0 for reservations removed, if the reservation is removed it returns 1.
// The error is set if the reservation key cannot be removed from the app or node.
-func (sa *Application) UnReserve(node *Node, ask *Allocation) (int, error) {
+func (sa *Application) UnReserve(node *Node, ask *Allocation) int {
+ log.Log(log.SchedApplication).Info("unreserving allocation from application",
+ zap.String("appID", sa.ApplicationID),
+ zap.Stringer("node", node),
+ zap.Stringer("alloc", ask))
+ if node == nil || ask == nil {
+ return 0
+ }
sa.Lock()
defer sa.Unlock()
- return sa.unReserveInternal(node, ask)
+ reserve, ok := sa.reservations[ask.allocationKey]
+ if !ok {
+ log.Log(log.SchedApplication).Debug("reservation not found on application",
+ zap.String("appID", sa.ApplicationID),
+ zap.String("allocationKey", ask.allocationKey))
+ return 0
+ }
+ if reserve.nodeID != node.NodeID {
+ log.Log(log.SchedApplication).Warn("UnReserve: provided info not consistent with reservation",
+ zap.String("appID", sa.ApplicationID),
+ zap.String("node", reserve.nodeID),
+ zap.String("alloc", reserve.allocKey))
+ }
+ return sa.unReserveInternal(reserve)
}
// Unlocked version for UnReserve that really does the work.
+// This is idempotent and will not fail
// Must only be called while holding the application lock.
-func (sa *Application) unReserveInternal(node *Node, ask *Allocation) (int, error) {
- resKey := reservationKey(node, nil, ask)
- if resKey == "" {
- log.Log(log.SchedApplication).Debug("unreserve reservation key create failed unexpectedly",
- zap.String("appID", sa.ApplicationID),
- zap.Stringer("node", node),
- zap.Stringer("ask", ask))
- return 0, fmt.Errorf("reservation key failed node or ask are nil for appID %s", sa.ApplicationID)
+func (sa *Application) unReserveInternal(reserve *reservation) int {
+ // this should not happen
+ if reserve == nil {
+ return 0
}
// unReserve the node before removing from the app
- var num int
- var err error
- if num, err = node.unReserve(sa, ask); err != nil {
- return 0, err
- }
+ num := reserve.node.unReserve(reserve.alloc)
// if the unreserve worked on the node check the app
- if _, found := sa.reservations[resKey]; found {
+ if _, found := sa.reservations[reserve.allocKey]; found {
// worked on the node means either found or not but no error, log difference here
if num == 0 {
log.Log(log.SchedApplication).Info("reservation not found while removing from node, app has reservation",
zap.String("appID", sa.ApplicationID),
- zap.String("nodeID", node.NodeID),
- zap.String("ask", ask.GetAllocationKey()))
+ zap.String("nodeID", reserve.nodeID),
+ zap.String("alloc", reserve.allocKey))
}
- delete(sa.reservations, resKey)
- log.Log(log.SchedApplication).Info("reservation removed successfully", zap.String("node", node.NodeID),
- zap.String("app", ask.GetApplicationID()), zap.String("ask", ask.GetAllocationKey()))
- return 1, nil
+ delete(sa.reservations, reserve.allocKey)
+ log.Log(log.SchedApplication).Info("reservation removed successfully",
+ zap.String("appID", sa.ApplicationID),
+ zap.String("node", reserve.nodeID),
+ zap.String("alloc", reserve.allocKey))
+ return 1
}
// reservation was not found
log.Log(log.SchedApplication).Info("reservation not found while removing from app",
zap.String("appID", sa.ApplicationID),
- zap.String("nodeID", node.NodeID),
- zap.String("ask", ask.GetAllocationKey()),
+ zap.String("node", reserve.nodeID),
+ zap.String("alloc", reserve.allocKey),
zap.Int("nodeReservationsRemoved", num))
- return 0, nil
+ return 0
}
-// Return the allocation reservations on any node.
-// The returned array is 0 or more keys into the reservations map.
+// canAllocationReserve Check if the allocation has already been reserved. An alloc can never reserve more than one node.
// No locking must be called while holding the lock
-func (sa *Application) GetAskReservations(allocKey string) []string {
- reservationKeys := make([]string, 0)
- if allocKey == "" {
- return reservationKeys
- }
- for key := range sa.reservations {
- if strings.HasSuffix(key, allocKey) {
- reservationKeys = append(reservationKeys, key)
- }
- }
- return reservationKeys
-}
-
-// Check if the allocation has already been reserved. An ask can never reserve more than one node.
-// No locking must be called while holding the lock
-func (sa *Application) canAskReserve(ask *Allocation) bool {
- allocKey := ask.GetAllocationKey()
- if ask.IsAllocated() {
- log.Log(log.SchedApplication).Debug("ask already allocated, no reservation allowed",
- zap.String("askKey", allocKey))
- return false
- }
- if len(sa.GetAskReservations(allocKey)) > 0 {
+func (sa *Application) canAllocationReserve(alloc *Allocation) error {
+ allocKey := alloc.GetAllocationKey()
+ if alloc.IsAllocated() {
+ log.Log(log.SchedApplication).Debug("allocation is marked as allocated, no reservation allowed",
+ zap.String("allocationKey", allocKey))
+ return common.ErrorReservingAlloc
+ }
+ reserved := sa.reservations[allocKey]
+ if reserved != nil {
log.Log(log.SchedApplication).Debug("reservation already exists",
- zap.String("askKey", allocKey))
- return false
+ zap.String("allocKey", allocKey),
+ zap.String("nodeID", reserved.nodeID))
+ return common.ErrorDuplicateReserve
}
- return true
+ return nil
}
func (sa *Application) getOutstandingRequests(headRoom *resources.Resource, userHeadRoom *resources.Resource, total *[]*Allocation) {
@@ -1045,42 +1026,13 @@ func (sa *Application) tryAllocate(headRoom *resources.Resource, allowPreemption
requiredNode := request.GetRequiredNode()
// does request have any constraint to run on specific node?
if requiredNode != "" {
- // the iterator might not have the node we need as it could be reserved, or we have not added it yet
- node := getNodeFn(requiredNode)
- if node == nil {
- getRateLimitedAppLog().Info("required node is not found (could be transient)",
- zap.String("application ID", sa.ApplicationID),
- zap.String("allocationKey", request.GetAllocationKey()),
- zap.String("required node", requiredNode))
- return nil
- }
- // Are there any non daemon set reservations on specific required node?
- // Cancel those reservations to run daemon set pods
- reservations := node.GetReservations()
- if len(reservations) > 0 {
- if !sa.cancelReservations(reservations) {
- return nil
- }
- }
- // we don't care about predicate error messages here
- result, _ := sa.tryNode(node, request) //nolint:errcheck
+ result := sa.tryRequiredNode(request, getNodeFn)
if result != nil {
- // check if the node was reserved and we allocated after a release
- if _, ok := sa.reservations[reservationKey(node, nil, request)]; ok {
- log.Log(log.SchedApplication).Debug("allocation on required node after release",
- zap.String("appID", sa.ApplicationID),
- zap.String("nodeID", requiredNode),
- zap.String("allocationKey", request.GetAllocationKey()))
- result.ResultType = AllocatedReserved
- return result
- }
- log.Log(log.SchedApplication).Debug("allocation on required node is completed",
- zap.String("nodeID", node.NodeID),
- zap.String("allocationKey", request.GetAllocationKey()),
- zap.Stringer("resultType", result.ResultType))
return result
}
- return newReservedAllocationResult(node.NodeID, request)
+ // it did not allocate or reserve: should only happen if the node is not registered yet
+ // just continue with the next request
+ continue
}
iterator := nodeIterator()
@@ -1108,50 +1060,86 @@ func (sa *Application) tryAllocate(headRoom *resources.Resource, allowPreemption
return nil
}
-func (sa *Application) cancelReservations(reservations []*reservation) bool {
- for _, res := range reservations {
- // skip the node
- if res.ask.GetRequiredNode() != "" {
- log.Log(log.SchedApplication).Warn("reservation for ask with required node already exists on the node",
- zap.String("required node", res.node.NodeID),
- zap.String("existing ask reservation key", res.getKey()))
- return false
- }
+// tryRequiredNode tries to place the allocation in the specific node that is set as the required node in the allocation.
+// The first time the allocation is seen it will try to make the allocation on the node. If that does not work it will
+// always trigger the reservation of the node.
+func (sa *Application) tryRequiredNode(request *Allocation, getNodeFn func(string) *Node) *AllocationResult {
+ requiredNode := request.GetRequiredNode()
+ allocationKey := request.GetAllocationKey()
+ // the iterator might not have the node we need as it could be reserved, or we have not added it yet
+ node := getNodeFn(requiredNode)
+ if node == nil {
+ getRateLimitedAppLog().Info("required node is not found (could be transient)",
+ zap.String("application ID", sa.ApplicationID),
+ zap.String("allocationKey", allocationKey),
+ zap.String("required node", requiredNode))
+ return nil
}
- var err error
+ // Are there any reservations on this node that does not specifically require this node?
+ // Cancel any reservations to make room for the allocations that require the node
var num int
+ reservations := node.GetReservations()
+ if len(reservations) > 0 {
+ num = sa.cancelReservations(reservations)
+ }
+ _, thisReserved := sa.reservations[allocationKey]
+ // now try the request, we don't care about predicate error messages here
+ result, _ := sa.tryNode(node, request) //nolint:errcheck
+ if result != nil {
+ result.CancelledReservations = num
+ // check if the node was reserved and we allocated after a release
+ if thisReserved {
+ log.Log(log.SchedApplication).Debug("allocation on required node after release",
+ zap.String("appID", sa.ApplicationID),
+ zap.String("nodeID", requiredNode),
+ zap.String("allocationKey", allocationKey))
+ result.ResultType = AllocatedReserved
+ return result
+ }
+ log.Log(log.SchedApplication).Debug("allocation on required node is completed",
+ zap.String("nodeID", node.NodeID),
+ zap.String("allocationKey", allocationKey),
+ zap.Stringer("resultType", result.ResultType))
+ return result
+ }
+ // if this ask was already reserved we should not have deleted any reservations
+ // we also do not need to send back a reservation result and just return nil to check the next ask
+ if thisReserved {
+ return nil
+ }
+ result = newReservedAllocationResult(node.NodeID, request)
+ result.CancelledReservations = num
+ return result
+}
+
+// cancelReservations will cancel all non required node reservations for a node. The list of reservations passed in is
+// a copy of all reservations of a single node. This is called during the required node allocation cycle only.
+// The returned int value is used to update the partition counter of active reservations.
+func (sa *Application) cancelReservations(reservations []*reservation) int {
+ var released, num int
// un reserve all the apps that were reserved on the node
for _, res := range reservations {
- thisApp := res.app.ApplicationID == sa.ApplicationID
- if thisApp {
- num, err = sa.unReserveInternal(res.node, res.ask)
- } else {
- num, err = res.app.UnReserve(res.node, res.ask)
- }
- if err != nil {
- log.Log(log.SchedApplication).Warn("Unable to cancel reservations on node",
- zap.String("victim application ID", res.app.ApplicationID),
- zap.String("victim allocationKey", res.getKey()),
- zap.String("required node", res.node.NodeID),
- zap.Int("reservations count", num),
- zap.String("application ID", sa.ApplicationID))
- return false
- } else {
- log.Log(log.SchedApplication).Info("Cancelled reservation on required node",
- zap.String("affected application ID", res.app.ApplicationID),
- zap.String("affected allocationKey", res.getKey()),
- zap.String("required node", res.node.NodeID),
- zap.Int("reservations count", num),
- zap.String("application ID", sa.ApplicationID))
+ // cleanup if the reservation does not have this node as a requirement
+ if res.alloc.requiredNode != "" {
+ continue
}
- // remove the reservation of the queue
+ thisApp := res.app.ApplicationID == sa.ApplicationID
if thisApp {
+ num = sa.unReserveInternal(res)
sa.queue.UnReserve(sa.ApplicationID, num)
} else {
+ num = res.app.UnReserve(res.node, res.alloc)
res.app.GetQueue().UnReserve(res.app.ApplicationID, num)
}
+ log.Log(log.SchedApplication).Info("Cancelled reservation for required node allocation",
+ zap.String("triggered by appID", sa.ApplicationID),
+ zap.String("affected application ID", res.appID),
+ zap.String("affected allocationKey", res.allocKey),
+ zap.String("required node", res.nodeID),
+ zap.Int("reservations count", num))
+ released += num
}
- return true
+ return released
}
// tryPlaceholderAllocate tries to replace a placeholder that is allocated with a real allocation
@@ -1238,14 +1226,15 @@ func (sa *Application) tryPlaceholderAllocate(nodeIterator func() NodeIterator,
// pick the first fit and try all nodes if that fails give up
var allocResult *AllocationResult
if phFit != nil && reqFit != nil {
+ resKey := reqFit.GetAllocationKey()
iterator.ForEachNode(func(node *Node) bool {
if !node.IsSchedulable() {
- log.Log(log.SchedApplication).Debug("skipping node for placeholder ask as state is unschedulable",
- zap.String("allocationKey", reqFit.GetAllocationKey()),
+ log.Log(log.SchedApplication).Debug("skipping node for placeholder alloc as state is unschedulable",
+ zap.String("allocationKey", resKey),
zap.String("node", node.NodeID))
return true
}
- if !node.preAllocateCheck(reqFit.GetAllocatedResource(), reservationKey(nil, sa, reqFit)) {
+ if !node.preAllocateCheck(reqFit.GetAllocatedResource(), resKey) {
return true
}
// skip the node if conditions can not be satisfied
@@ -1257,7 +1246,7 @@ func (sa *Application) tryPlaceholderAllocate(nodeIterator func() NodeIterator,
if !node.TryAddAllocation(reqFit) {
log.Log(log.SchedApplication).Debug("Node update failed unexpectedly",
zap.String("applicationID", sa.ApplicationID),
- zap.Stringer("ask", reqFit),
+ zap.Stringer("alloc", reqFit),
zap.Stringer("placeholder", phFit))
return false
}
@@ -1266,7 +1255,7 @@ func (sa *Application) tryPlaceholderAllocate(nodeIterator func() NodeIterator,
log.Log(log.SchedApplication).Warn("allocation of ask failed unexpectedly",
zap.Error(err))
// unwind node allocation
- _ = node.RemoveAllocation(reqFit.GetAllocationKey())
+ _ = node.RemoveAllocation(resKey)
return false
}
// allocation worked: on a non placeholder node update resultType and return
@@ -1297,7 +1286,7 @@ func (sa *Application) checkHeadRooms(ask *Allocation, userHeadroom *resources.R
return userHeadroom.FitInMaxUndef(ask.GetAllocatedResource()) && headRoom.FitInMaxUndef(ask.GetAllocatedResource())
}
-// Try a reserved allocation of an outstanding reservation
+// tryReservedAllocate tries allocating an outstanding reservation
func (sa *Application) tryReservedAllocate(headRoom *resources.Resource, nodeIterator func() NodeIterator) *AllocationResult {
sa.Lock()
defer sa.Unlock()
@@ -1306,14 +1295,14 @@ func (sa *Application) tryReservedAllocate(headRoom *resources.Resource, nodeIte
// process all outstanding reservations and pick the first one that fits
for _, reserve := range sa.reservations {
- ask := sa.requests[reserve.askKey]
+ ask := sa.requests[reserve.allocKey]
// sanity check and cleanup if needed
if ask == nil || ask.IsAllocated() {
var unreserveAsk *Allocation
// if the ask was not found we need to construct one to unreserve
if ask == nil {
unreserveAsk = &Allocation{
- allocationKey: reserve.askKey,
+ allocationKey: reserve.allocKey,
applicationID: sa.ApplicationID,
allocLog: make(map[string]*AllocationLogEntry),
}
@@ -1348,17 +1337,17 @@ func (sa *Application) tryReservedAllocate(headRoom *resources.Resource, nodeIte
// try this on all other nodes
for _, reserve := range sa.reservations {
- // Other nodes cannot be tried if the ask has a required node
- ask := reserve.ask
- if ask.GetRequiredNode() != "" {
+ // Other nodes cannot be tried if a required node is requested
+ alloc := reserve.alloc
+ if alloc.GetRequiredNode() != "" {
continue
}
iterator := nodeIterator()
if iterator != nil {
- if !sa.checkHeadRooms(ask, userHeadroom, headRoom) {
+ if !sa.checkHeadRooms(alloc, userHeadroom, headRoom) {
continue
}
- result := sa.tryNodesNoReserve(ask, iterator, reserve.nodeID)
+ result := sa.tryNodesNoReserve(alloc, iterator, reserve.nodeID)
// have a candidate return it, including the node that was reserved
if result != nil {
return result
@@ -1413,8 +1402,8 @@ func (sa *Application) tryRequiredNodePreemption(reserve *reservation, ask *Allo
return false
}
-// Try all the nodes for a reserved request that have not been tried yet.
-// This should never result in a reservation as the ask is already reserved
+// tryNodesNoReserve tries all the nodes for a reserved request that have not been tried yet.
+// This should never result in a reservation as the allocation is already reserved
func (sa *Application) tryNodesNoReserve(ask *Allocation, iterator NodeIterator, reservedNode string) *AllocationResult {
var allocResult *AllocationResult
iterator.ForEachNode(func(node *Node) bool {
@@ -1449,14 +1438,13 @@ func (sa *Application) tryNodesNoReserve(ask *Allocation, iterator NodeIterator,
func (sa *Application) tryNodes(ask *Allocation, iterator NodeIterator) *AllocationResult {
var nodeToReserve *Node
scoreReserved := math.Inf(1)
- // check if the ask is reserved or not
+ // check if the alloc is reserved or not
allocKey := ask.GetAllocationKey()
- reservedAsks := sa.GetAskReservations(allocKey)
- allowReserve := !ask.IsAllocated() && len(reservedAsks) == 0
+ reserved := sa.reservations[allocKey]
var allocResult *AllocationResult
var predicateErrors map[string]int
iterator.ForEachNode(func(node *Node) bool {
- // skip the node if the node is not valid for the ask
+ // skip the node if the node is not schedulable
if !node.IsSchedulable() {
log.Log(log.SchedApplication).Debug("skipping node for ask as state is unschedulable",
zap.String("allocationKey", allocKey),
@@ -1478,28 +1466,23 @@ func (sa *Application) tryNodes(ask *Allocation, iterator NodeIterator) *Allocat
// allocation worked so return
if result != nil {
metrics.GetSchedulerMetrics().ObserveTryNodeLatency(tryNodeStart)
- // check if the node was reserved for this ask: if it is set the resultType and return
- // NOTE: this is a safeguard as reserved nodes should never be part of the iterator
- // but we have no locking
- if _, ok := sa.reservations[reservationKey(node, nil, ask)]; ok {
- log.Log(log.SchedApplication).Debug("allocate found reserved ask during non reserved allocate",
- zap.String("appID", sa.ApplicationID),
- zap.String("nodeID", node.NodeID),
- zap.String("allocationKey", allocKey))
- result.ResultType = AllocatedReserved
- allocResult = result
- return false
- }
- // we could also have a different node reserved for this ask if it has pick one of
- // the reserved nodes to unreserve (first one in the list)
- if len(reservedAsks) > 0 {
- nodeID := strings.TrimSuffix(reservedAsks[0], "|"+allocKey)
- log.Log(log.SchedApplication).Debug("allocate picking reserved ask during non reserved allocate",
- zap.String("appID", sa.ApplicationID),
- zap.String("nodeID", nodeID),
- zap.String("allocationKey", allocKey))
+ // check if the alloc had a reservation: if it has set the resultType and return
+ if reserved != nil {
+ if reserved.nodeID != node.NodeID {
+ // we have a different node reserved for this alloc
+ log.Log(log.SchedApplication).Debug("allocate picking reserved alloc during non reserved allocate",
+ zap.String("appID", sa.ApplicationID),
+ zap.String("reserved nodeID", reserved.nodeID),
+ zap.String("allocationKey", allocKey))
+ result.ReservedNodeID = reserved.nodeID
+ } else {
+ // NOTE: this is a safeguard as reserved nodes should never be part of the iterator
+ log.Log(log.SchedApplication).Debug("allocate found reserved alloc during non reserved allocate",
+ zap.String("appID", sa.ApplicationID),
+ zap.String("nodeID", node.NodeID),
+ zap.String("allocationKey", allocKey))
+ }
result.ResultType = AllocatedReserved
- result.ReservedNodeID = nodeID
allocResult = result
return false
}
@@ -1509,14 +1492,14 @@ func (sa *Application) tryNodes(ask *Allocation, iterator NodeIterator) *Allocat
}
// nothing allocated should we look at a reservation?
askAge := time.Since(ask.GetCreateTime())
- if allowReserve && askAge > reservationDelay {
+ if reserved == nil && askAge > reservationDelay {
log.Log(log.SchedApplication).Debug("app reservation check",
zap.String("allocationKey", allocKey),
zap.Time("createTime", ask.GetCreateTime()),
zap.Duration("askAge", askAge),
zap.Duration("reservationDelay", reservationDelay))
score := node.GetFitInScoreForAvailableResource(ask.GetAllocatedResource())
- // Record the so-far best node to reserve
+ // Record the best node so-far to reserve
if score < scoreReserved {
scoreReserved = score
nodeToReserve = node
@@ -1540,7 +1523,7 @@ func (sa *Application) tryNodes(ask *Allocation, iterator NodeIterator) *Allocat
zap.String("appID", sa.ApplicationID),
zap.String("nodeID", nodeToReserve.NodeID),
zap.String("allocationKey", allocKey),
- zap.Int("reservations", len(reservedAsks)))
+ zap.Int("reservations", len(sa.reservations)))
// skip the node if conditions can not be satisfied
if nodeToReserve.preReserveConditions(ask) != nil {
return nil
@@ -1552,11 +1535,12 @@ func (sa *Application) tryNodes(ask *Allocation, iterator NodeIterator) *Allocat
return nil
}
-// Try allocating on one specific node
+// tryNode tries allocating on one specific node
func (sa *Application) tryNode(node *Node, ask *Allocation) (*AllocationResult, error) {
toAllocate := ask.GetAllocatedResource()
+ allocationKey := ask.GetAllocationKey()
// create the key for the reservation
- if !node.preAllocateCheck(toAllocate, reservationKey(nil, sa, ask)) {
+ if !node.preAllocateCheck(toAllocate, allocationKey) {
// skip schedule onto node
return nil, nil
}
@@ -1571,13 +1555,13 @@ func (sa *Application) tryNode(node *Node, ask *Allocation) (*AllocationResult,
log.Log(log.SchedApplication).DPanic("queue update failed unexpectedly",
zap.Error(err))
// revert the node update
- node.RemoveAllocation(ask.GetAllocationKey())
+ node.RemoveAllocation(allocationKey)
return nil, nil
}
- // mark this ask as allocated
+ // mark this alloc as allocated
_, err := sa.allocateAsk(ask)
if err != nil {
- log.Log(log.SchedApplication).Warn("allocation of ask failed unexpectedly",
+ log.Log(log.SchedApplication).Warn("allocation of alloc failed unexpectedly",
zap.Error(err))
}
// all is OK, last update for the app
diff --git a/pkg/scheduler/objects/application_test.go b/pkg/scheduler/objects/application_test.go
index 5b7eed87..c6cfca9c 100644
--- a/pkg/scheduler/objects/application_test.go
+++ b/pkg/scheduler/objects/application_test.go
@@ -234,11 +234,8 @@ func TestAppReservation(t *testing.T) {
if app.HasReserved() {
t.Fatal("new app should not have reservations")
}
- if app.IsReservedOnNode("") {
- t.Error("app should not have reservations for empty node ID")
- }
- if app.IsReservedOnNode("unknown") {
- t.Error("new app should not have reservations for unknown node")
+ if app.NodeReservedForAsk("") != "" {
+ t.Error("app should not have reservations for empty ask")
}
queue, err := createRootQueue(nil)
@@ -270,33 +267,21 @@ func TestAppReservation(t *testing.T) {
// reserve that works
err = app.Reserve(node, ask)
assert.NilError(t, err, "reservation should not have failed")
- if app.IsReservedOnNode("") {
- t.Errorf("app should not have reservations for empty node ID")
- }
- if app.IsReservedOnNode("unknown") {
- t.Error("app should not have reservations for unknown node")
+ if app.NodeReservedForAsk("") != "" {
+ t.Error("app should not have reservations for empty ask")
}
- if app.HasReserved() && !app.IsReservedOnNode(nodeID1) {
+ if app.HasReserved() && app.NodeReservedForAsk(aKey) != nodeID1 {
t.Errorf("app should have reservations for node %s", nodeID1)
}
- // node name similarity check: chop of the last char to make sure we check the full name
- similar := nodeID1[:len(nodeID1)-1]
- if app.HasReserved() && app.IsReservedOnNode(similar) {
- t.Errorf("similar app should not have reservations for node %s", similar)
- }
-
// reserve the same reservation
err = app.Reserve(node, ask)
if err == nil {
t.Errorf("reservation should have failed")
}
- // unreserve unknown node/ask
- _, err = app.UnReserve(nil, nil)
- if err == nil {
- t.Errorf("illegal reservation release but did not fail")
- }
+ // unreserve unknown node/alloc
+ assert.Equal(t, app.UnReserve(nil, nil), 0, "illegal reservation release should have returned 0")
// 2nd reservation for app
ask2 := newAllocationAsk("alloc-2", appID1, res)
@@ -309,20 +294,16 @@ func TestAppReservation(t *testing.T) {
}
err = app.Reserve(node2, ask2)
assert.NilError(t, err, "reservation of 2nd node should not have failed")
- _, err = app.UnReserve(node2, ask2)
- assert.NilError(t, err, "remove of reservation of 2nd node should not have failed")
+ assert.Equal(t, app.UnReserve(node2, ask2), 1, "remove of reservation of 2nd node should not have failed")
// unreserve the same should fail
- _, err = app.UnReserve(node2, ask2)
- assert.NilError(t, err, "remove twice of reservation of 2nd node should have failed")
+ assert.Equal(t, app.UnReserve(node2, ask2), 0, "remove twice of reservation of 2nd node should return 0")
// failure case: remove reservation from node, app still needs cleanup
var num int
- num, err = node.unReserve(app, ask)
- assert.NilError(t, err, "un-reserve on node should not have failed")
+ num = node.unReserve(ask)
assert.Equal(t, num, 1, "un-reserve on node should have removed reservation")
- num, err = app.UnReserve(node, ask)
- assert.NilError(t, err, "app has reservation should not have failed")
+ num = app.UnReserve(node, ask)
assert.Equal(t, num, 1, "un-reserve on app should have removed reservation from app")
}
@@ -338,12 +319,9 @@ func TestAppAllocReservation(t *testing.T) {
if app == nil || app.ApplicationID != appID1 {
t.Fatalf("app create failed which should not have %v", app)
}
- if app.HasReserved() {
- t.Fatal("new app should not have reservations")
- }
- if len(app.GetAskReservations("")) != 0 {
- t.Fatal("new app should not have reservation for empty allocKey")
- }
+ assert.Assert(t, !app.HasReserved(), "new app should not have reservations")
+ assert.Equal(t, len(app.GetReservations()), 0, "new app should not have reservation for empty allocKey")
+
queue, err := createRootQueue(nil)
assert.NilError(t, err, "queue create failed")
app.queue = queue
@@ -360,46 +338,44 @@ func TestAppAllocReservation(t *testing.T) {
assert.NilError(t, err, "ask2 should have been added to app")
err = app.Reserve(node1, ask)
assert.NilError(t, err, "reservation should not have failed")
- if len(app.GetAskReservations("")) != 0 {
+ if app.reservations[""] != nil {
t.Fatal("app should not have reservation for empty allocKey")
}
- nodeKey1 := nodeID1 + "|" + aKey
- askReserved := app.GetAskReservations(aKey)
- if len(askReserved) != 1 || askReserved[0] != nodeKey1 {
- t.Errorf("app should have reservations for %s on %s and has not", aKey, nodeID1)
+ allocReserved := app.reservations[aKey]
+ if allocReserved == nil || allocReserved.nodeID != nodeID1 {
+ t.Fatalf("app should have reservations for %s on %s and has not", aKey, nodeID1)
}
+ assert.Equal(t, len(app.GetReservations()), 1, "app should have 1 reservation")
- nodeID2 := "node-2"
node2 := newNode(nodeID2, map[string]resources.Quantity{"first": 10})
err = app.Reserve(node2, ask2)
assert.NilError(t, err, "reservation should not have failed: error %v", err)
- nodeKey2 := nodeID2 + "|" + aKey2
- askReserved = app.GetAskReservations(aKey2)
- if len(askReserved) != 1 && askReserved[0] != nodeKey2 {
- t.Errorf("app should have reservations for %s on %s and has not", aKey, nodeID2)
+ allocReserved = app.reservations[aKey2]
+ if allocReserved == nil || allocReserved.nodeID != nodeID2 {
+ t.Fatalf("app should have reservations for %s on %s and has not", aKey, nodeID2)
}
+ assert.Equal(t, len(app.GetReservations()), 2, "app should have 2 reservation")
// check duplicate reserve: nothing should change
- if app.canAskReserve(ask) {
- t.Error("ask has already reserved, reserve check should have failed")
- }
+ assert.Assert(t, app.canAllocationReserve(ask) != nil, "alloc has already reserved, reserve check should have failed")
node3 := newNode("node-3", map[string]resources.Quantity{"first": 10})
err = app.Reserve(node3, ask)
if err == nil {
- t.Errorf("reservation should have failed")
+ t.Fatal("reservation should have failed")
}
- askReserved = app.GetAskReservations(aKey)
- if len(askReserved) != 1 && askReserved[0] != nodeKey1 {
- t.Errorf("app should have reservations for node %s and has not: %v", nodeID1, askReserved)
+ allocReserved = app.reservations[aKey]
+ if allocReserved == nil || allocReserved.nodeID != nodeID1 {
+ t.Fatalf("app should have reservations for node %s and has not: %v", nodeID1, allocReserved)
}
- askReserved = app.GetAskReservations(aKey2)
- if len(askReserved) != 1 && askReserved[0] != nodeKey2 {
- t.Errorf("app should have reservations for node %s and has not: %v", nodeID2, askReserved)
+ allocReserved = app.reservations[aKey2]
+ if allocReserved == nil || allocReserved.nodeID != nodeID2 {
+ t.Fatalf("app should have reservations for node %s and has not: %v", nodeID2, allocReserved)
}
+ assert.Equal(t, len(app.GetReservations()), 2, "app should have 2 reservation")
// clean up all asks and reservations
- reservedAsks := app.RemoveAllocationAsk("")
- if app.HasReserved() || node1.IsReserved() || node2.IsReserved() || reservedAsks != 2 {
- t.Errorf("ask removal did not clean up all reservations, reserved released = %d", reservedAsks)
+ reservedRelease := app.RemoveAllocationAsk("")
+ if app.HasReserved() || node1.IsReserved() || node2.IsReserved() || reservedRelease != 2 {
+ t.Fatalf("ask removal did not clean up all reservations, reserved released = %d", reservedRelease)
}
}
@@ -495,7 +471,6 @@ func TestAddAllocAsk(t *testing.T) {
// test add alloc ask event
noEvents := uint64(0)
err = common.WaitForCondition(10*time.Millisecond, time.Second, func() bool {
- fmt.Printf("checking event length: %d\n", eventSystem.Store.CountStoredEvents())
noEvents = eventSystem.Store.CountStoredEvents()
return noEvents == 2
})
@@ -652,7 +627,7 @@ func TestRemoveReservedAllocAsk(t *testing.T) {
node := newNode(nodeID1, map[string]resources.Quantity{"first": 10})
err = app.Reserve(node, ask2)
assert.NilError(t, err, "reservation should not have failed")
- if len(app.GetAskReservations(allocKey)) != 1 || !node.IsReserved() {
+ if app.reservations[allocKey] == nil || !node.IsReserved() {
t.Fatalf("app should have reservation for %v on node", allocKey)
}
before := app.GetPendingResource().Clone()
@@ -671,11 +646,10 @@ func TestRemoveReservedAllocAsk(t *testing.T) {
err = app.Reserve(node, ask2)
assert.NilError(t, err, "reservation should not have failed: error %v", err)
- if len(app.GetAskReservations(allocKey)) != 1 || !node.IsReserved() {
+ if app.reservations[allocKey] == nil || !node.IsReserved() {
t.Fatalf("app should have reservation for %v on node", allocKey)
}
- var num int
- num, err = node.unReserve(app, ask2)
+ num := node.unReserve(ask2)
assert.NilError(t, err, "un-reserve on node should not have failed")
assert.Equal(t, num, 1, "un-reserve on node should have removed reservation")
@@ -1954,6 +1928,235 @@ func TestCanReplace(t *testing.T) {
}
}
+func TestTryRequiredNode(t *testing.T) {
+ node := newNode(nodeID1, map[string]resources.Quantity{"first": 5})
+ nodeMap := map[string]*Node{nodeID1: node}
+ getNode := func(nodeID string) *Node {
+ return nodeMap[nodeID]
+ }
+ resMap := map[string]string{"first": "5"}
+ rootQ, err := createRootQueue(resMap)
+ assert.NilError(t, err, "unexpected error when creating root queue")
+ var childQ *Queue
+ childQ, err = createManagedQueue(rootQ, "child", false, resMap)
+ assert.NilError(t, err, "unexpected error when creating child queue")
+
+ app := newApplication(appID1, "default", "root.child")
+ app.SetQueue(childQ)
+ childQ.applications[appID1] = app
+ allocRes := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 3})
+ alloc := newAllocation(aKey, nodeID1, allocRes)
+ app.AddAllocation(alloc)
+ node.AddAllocation(alloc)
+
+ ask := newAllocationAsk(aKey2, appID1, allocRes)
+ err = app.AddAllocationAsk(ask)
+ assert.NilError(t, err, "unexpected error when adding an ask")
+ err = app.Reserve(node, ask)
+ assert.NilError(t, err, "unexpected error when reserving ask")
+
+ // get a small enough allocation that fits after cancelling the reservation
+ allocRes = resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1})
+ ask = newAllocationAsk(aKey3, appID1, allocRes)
+ ask.requiredNode = nodeID1
+ err = app.AddAllocationAsk(ask)
+ assert.NilError(t, err, "unexpected error when adding an ask")
+ result := app.tryRequiredNode(ask, getNode)
+ assert.Assert(t, result != nil, "alloc expected")
+ assert.Assert(t, result.Request == ask, "alloc expected for the ask")
+ assert.Equal(t, result.ResultType, Allocated, "expected allocated result")
+ assert.Equal(t, result.CancelledReservations, 1, "expected 1 cancelled reservation")
+ assert.Equal(t, nodeID1, result.NodeID, "wrong node")
+ // the non required node one should be cancelled
+ assert.Assert(t, !node.isReservedForAllocation(aKey2), "expecting no reservation for alloc-2 on node")
+ assert.Equal(t, app.NodeReservedForAsk(aKey2), "", "expecting no reservation for alloc-2 on node-1")
+}
+
+func TestTryRequiredNodeReserved(t *testing.T) {
+ node := newNode(nodeID1, map[string]resources.Quantity{"first": 5})
+ nodeMap := map[string]*Node{nodeID1: node}
+ getNode := func(nodeID string) *Node {
+ return nodeMap[nodeID]
+ }
+ resMap := map[string]string{"first": "5"}
+ rootQ, err := createRootQueue(resMap)
+ assert.NilError(t, err, "unexpected error when creating root queue")
+ var childQ *Queue
+ childQ, err = createManagedQueue(rootQ, "child", false, resMap)
+ assert.NilError(t, err, "unexpected error when creating child queue")
+
+ app := newApplication(appID1, "default", "root.child")
+ app.SetQueue(childQ)
+ childQ.applications[appID1] = app
+ allocRes := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 2})
+ ask := newAllocationAsk(aKey2, appID1, allocRes)
+ ask.requiredNode = nodeID1
+ err = app.AddAllocationAsk(ask)
+ assert.NilError(t, err, "unexpected error when adding an ask")
+ err = app.Reserve(node, ask)
+ assert.NilError(t, err, "unexpected error when reserving ask")
+
+ result := app.tryRequiredNode(ask, getNode)
+ assert.Assert(t, result != nil, "alloc expected")
+ assert.Assert(t, result.Request == ask, "alloc expected for the ask")
+ assert.Equal(t, result.ResultType, AllocatedReserved, "expected allocated reserved result")
+ assert.Equal(t, result.CancelledReservations, 0, "expected no cancelled reservation")
+ assert.Equal(t, nodeID1, result.NodeID, "wrong node")
+}
+
+func TestTryRequiredNodeReserve(t *testing.T) {
+ node := newNode(nodeID1, map[string]resources.Quantity{"first": 5})
+ nodeMap := map[string]*Node{nodeID1: node}
+ getNode := func(nodeID string) *Node {
+ return nodeMap[nodeID]
+ }
+ resMap := map[string]string{"first": "5"}
+ rootQ, err := createRootQueue(resMap)
+ assert.NilError(t, err, "unexpected error when creating root queue")
+ var childQ *Queue
+ childQ, err = createManagedQueue(rootQ, "child", false, resMap)
+ assert.NilError(t, err, "unexpected error when creating child queue")
+
+ app := newApplication(appID1, "default", "root.child")
+ app.SetQueue(childQ)
+ childQ.applications[appID1] = app
+ allocRes := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 3})
+ alloc := newAllocation(aKey, nodeID1, allocRes)
+ app.AddAllocation(alloc)
+ node.AddAllocation(alloc)
+
+ ask := newAllocationAsk(aKey2, appID1, allocRes)
+ ask.requiredNode = nodeID1
+ err = app.AddAllocationAsk(ask)
+ assert.NilError(t, err, "unexpected error when adding an ask")
+
+ result := app.tryRequiredNode(ask, getNode)
+ assert.Assert(t, result != nil, "alloc expected")
+ assert.Assert(t, result.Request == ask, "alloc expected for the ask")
+ assert.Equal(t, result.ResultType, Reserved, "expected reserved result")
+ assert.Equal(t, nodeID1, result.NodeID, "wrong node")
+}
+
+func TestTryRequiredNodeCancel(t *testing.T) {
+ node := newNode(nodeID1, map[string]resources.Quantity{"first": 5})
+ nodeMap := map[string]*Node{nodeID1: node}
+ getNode := func(nodeID string) *Node {
+ return nodeMap[nodeID]
+ }
+ resMap := map[string]string{"first": "5"}
+ rootQ, err := createRootQueue(resMap)
+ assert.NilError(t, err, "unexpected error when creating root queue")
+ var childQ *Queue
+ childQ, err = createManagedQueue(rootQ, "child", false, resMap)
+ assert.NilError(t, err, "unexpected error when creating child queue")
+
+ app := newApplication(appID1, "default", "root.child")
+ app.SetQueue(childQ)
+ childQ.applications[appID1] = app
+ allocRes := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 3})
+ alloc := newAllocation(aKey, nodeID1, allocRes)
+ app.AddAllocation(alloc)
+ node.AddAllocation(alloc)
+
+ ask := newAllocationAsk(aKey2, appID1, allocRes)
+ err = app.AddAllocationAsk(ask)
+ assert.NilError(t, err, "adding new allocation to app failed unexpected")
+ err = app.reserveInternal(node, ask)
+ assert.NilError(t, err, "reserving new allocation on app/node failed unexpected")
+ assert.Assert(t, node.isReservedForAllocation(aKey2), "expecting alloc reservation on node")
+ assert.Equal(t, app.NodeReservedForAsk(aKey2), nodeID1, "expecting app reservation on node")
+
+ ask = newAllocationAsk(aKey3, appID1, allocRes)
+ ask.requiredNode = nodeID1
+ err = app.AddAllocationAsk(ask)
+ assert.NilError(t, err, "adding new allocation to app failed unexpected")
+ result := app.tryRequiredNode(ask, getNode)
+ assert.Assert(t, result != nil, "alloc expected")
+ assert.Assert(t, result.Request == ask, "alloc expected for the ask")
+ assert.Equal(t, result.ResultType, Reserved, "expected allocated result")
+ assert.Equal(t, result.CancelledReservations, 1, "expected 1 cancelled reservation")
+ assert.Equal(t, nodeID1, result.NodeID, "wrong node")
+ assert.Assert(t, !node.isReservedForAllocation(aKey2), "expecting no reservation for alloc-2 on node")
+ assert.Equal(t, app.NodeReservedForAsk(aKey2), "", "expecting no reservation for alloc-2 on app")
+}
+
+func TestTryRequiredNodeAdd(t *testing.T) {
+ node := newNode(nodeID1, map[string]resources.Quantity{"first": 5})
+ nodeMap := map[string]*Node{nodeID1: node}
+ getNode := func(nodeID string) *Node {
+ return nodeMap[nodeID]
+ }
+ rootQ, err := createRootQueue(map[string]string{"first": "5"})
+ assert.NilError(t, err, "unexpected error when creating root queue")
+ var childQ *Queue
+ childQ, err = createManagedQueue(rootQ, "child", false, map[string]string{"first": "5"})
+ assert.NilError(t, err, "unexpected error when creating child queue")
+
+ app := newApplication(appID1, "default", "root.child")
+ app.SetQueue(childQ)
+ childQ.applications[appID1] = app
+ allocRes := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 3})
+ alloc := newAllocation(aKey, nodeID1, allocRes)
+ app.AddAllocation(alloc)
+ node.AddAllocation(alloc)
+
+ ask := newAllocationAsk(aKey2, appID1, allocRes)
+ ask.requiredNode = nodeID1
+ err = app.AddAllocationAsk(ask)
+ assert.NilError(t, err, "adding new allocation to app failed unexpected")
+
+ result := app.tryRequiredNode(ask, getNode)
+ assert.Assert(t, result != nil, "alloc expected")
+ assert.Assert(t, result.Request == ask, "alloc expected for the ask")
+ assert.Equal(t, result.ResultType, Reserved, "expected reserved result")
+ assert.Equal(t, nodeID1, result.NodeID, "wrong node")
+
+ // finish processing do what the context would do
+ err = app.reserveInternal(node, ask)
+ assert.NilError(t, err, "reservation processing failed unexpected")
+
+ ask = newAllocationAsk(aKey3, appID1, allocRes)
+ ask.requiredNode = nodeID1
+ err = app.AddAllocationAsk(ask)
+ assert.NilError(t, err, "adding new allocation to app failed unexpected")
+ result = app.tryRequiredNode(ask, getNode)
+ assert.Assert(t, result != nil, "alloc expected")
+ assert.Assert(t, result.Request == ask, "alloc expected for the ask")
+ assert.Equal(t, result.ResultType, Reserved, "expected allocated result")
+ assert.Equal(t, result.CancelledReservations, 0, "expected no cancelled reservation")
+ assert.Equal(t, nodeID1, result.NodeID, "wrong node")
+ assert.Assert(t, node.isReservedForAllocation(aKey2), "expecting reservation for alloc-2 on node")
+ assert.Equal(t, app.NodeReservedForAsk(aKey2), nodeID1, "expecting reservation for alloc-2 on app")
+ assert.Assert(t, !node.isReservedForAllocation(aKey3), "expecting no reservation for alloc-3 on node")
+ assert.Equal(t, app.NodeReservedForAsk(aKey3), "", "expecting no reservation for alloc-3 on app")
+}
+
+func TestTryRequiredNodeExists(t *testing.T) {
+ node := newNode(nodeID1, map[string]resources.Quantity{"first": 5})
+ nodeMap := map[string]*Node{nodeID1: node}
+ getNode := func(nodeID string) *Node {
+ return nodeMap[nodeID]
+ }
+ rootQ, err := createRootQueue(map[string]string{"first": "5"})
+ assert.NilError(t, err, "unexpected error when creating root queue")
+ var childQ *Queue
+ childQ, err = createManagedQueue(rootQ, "child", false, map[string]string{"first": "5"})
+ assert.NilError(t, err, "unexpected error when creating child queue")
+
+ app := newApplication(appID1, "default", "root.child")
+ app.SetQueue(childQ)
+ childQ.applications[appID1] = app
+ allocRes := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 3})
+
+ ask := newAllocationAsk(aKey2, appID1, allocRes)
+ ask.requiredNode = nodeID2
+ err = app.AddAllocationAsk(ask)
+ assert.NilError(t, err, "adding new allocation to app failed unexpected")
+
+ result := app.tryRequiredNode(ask, getNode)
+ assert.Assert(t, result == nil, "alloc not expected")
+}
+
func TestTryAllocateNoRequests(t *testing.T) {
node := newNode("node1", map[string]resources.Quantity{"first": 5})
nodeMap := map[string]*Node{"node1": node}
@@ -3210,3 +3413,28 @@ type mockAppEventHandler struct {
func (m mockAppEventHandler) HandleEvent(ev interface{}) {
m.callback(ev)
}
+
+func TestApplication_canAllocationReserve(t *testing.T) {
+ res := resources.NewResource()
+ tests := []struct {
+ name string
+ alloc *Allocation
+ wantErr bool
+ }{
+ {"new", newAllocationWithKey(aKey, appID1, "", res), false},
+ {"allocated", newAllocationWithKey(aKey2, appID1, nodeID1, res), true},
+ {"duplicate", newAllocationWithKey(aKey3, appID1, "", res), true},
+ }
+ app := newApplication(appID0, "default", "root.unknown")
+ app.reservations[aKey3] = &reservation{
+ nodeID: nodeID1,
+ allocKey: aKey,
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ if err := app.canAllocationReserve(tt.alloc); (err != nil) != tt.wantErr {
+ t.Errorf("canAllocationReserve() error = %v, wantErr %v", err, tt.wantErr)
+ }
+ })
+ }
+}
diff --git a/pkg/scheduler/objects/node.go b/pkg/scheduler/objects/node.go
index ae5971f4..2047ff44 100644
--- a/pkg/scheduler/objects/node.go
+++ b/pkg/scheduler/objects/node.go
@@ -20,17 +20,17 @@ package objects
import (
"fmt"
- "strings"
"go.uber.org/zap"
+ "github.com/apache/yunikorn-core/pkg/common"
"github.com/apache/yunikorn-core/pkg/common/resources"
"github.com/apache/yunikorn-core/pkg/events"
"github.com/apache/yunikorn-core/pkg/locking"
"github.com/apache/yunikorn-core/pkg/log"
"github.com/apache/yunikorn-core/pkg/plugins"
schedEvt "github.com/apache/yunikorn-core/pkg/scheduler/objects/events"
- "github.com/apache/yunikorn-scheduler-interface/lib/go/common"
+ siCommon "github.com/apache/yunikorn-scheduler-interface/lib/go/common"
"github.com/apache/yunikorn-scheduler-interface/lib/go/si"
)
@@ -110,9 +110,9 @@ func (sn *Node) initializeAttribute(newAttributes map[string]string) {
sn.attributes = map[string]string{}
}
- sn.Hostname = sn.attributes[common.HostName]
- sn.Rackname = sn.attributes[common.RackName]
- sn.Partition = sn.attributes[common.NodePartition]
+ sn.Hostname = sn.attributes[siCommon.HostName]
+ sn.Rackname = sn.attributes[siCommon.RackName]
+ sn.Partition = sn.attributes[siCommon.NodePartition]
}
// Get an attribute by name. The most used attributes can be directly accessed via the
@@ -129,7 +129,7 @@ func (sn *Node) GetAttributes() map[string]string {
// Get InstanceType of this node.
// This is a lock free call because all attributes are considered read only
func (sn *Node) GetInstanceType() string {
- itype := sn.GetAttribute(common.InstanceType)
+ itype := sn.GetAttribute(siCommon.InstanceType)
if itype != "" {
return itype
}
@@ -521,7 +521,7 @@ func (sn *Node) preConditions(ask *Allocation, allocate bool) error {
// preAllocateCheck checks if the node should be considered as a possible node to allocate on.
// No updates are made this only performs a pre allocate checks
-func (sn *Node) preAllocateCheck(res *resources.Resource, resKey string) bool {
+func (sn *Node) preAllocateCheck(res *resources.Resource, allocationKey string) bool {
// cannot allocate zero or negative resource
if !resources.StrictlyGreaterThanZero(res) {
log.Log(log.SchedNode).Debug("pre alloc check: requested resource is zero",
@@ -530,10 +530,10 @@ func (sn *Node) preAllocateCheck(res *resources.Resource, resKey string) bool {
}
// check if the node is reserved for this app/alloc
if sn.IsReserved() {
- if !sn.isReservedForApp(resKey) {
- log.Log(log.SchedNode).Debug("pre alloc check: node reserved for different app or ask",
+ if !sn.isReservedForAllocation(allocationKey) {
+ log.Log(log.SchedNode).Debug("pre alloc check: node reserved for different alloc",
zap.String("nodeID", sn.NodeID),
- zap.String("resKey", resKey))
+ zap.String("allocationKey", allocationKey))
return false
}
}
@@ -544,95 +544,98 @@ func (sn *Node) preAllocateCheck(res *resources.Resource, resKey string) bool {
return sn.availableResource.FitIn(res)
}
-// Return if the node has been reserved by any application
+// IsReserved returns true if the node has been reserved for an allocation
func (sn *Node) IsReserved() bool {
sn.RLock()
defer sn.RUnlock()
return len(sn.reservations) > 0
}
-// isReservedForApp returns true if and only if the node has been reserved by the application
-// NOTE: a return value of false does not mean the node is not reserved by a different app
-func (sn *Node) isReservedForApp(key string) bool {
+// isReservedForAllocation returns true if and only if the node has been reserved by this allocation
+// NOTE: a return value of false does not mean the node is not reserved by a different allocation, use IsReserved
+// to test if the node has any reservation.
+func (sn *Node) isReservedForAllocation(key string) bool {
if key == "" {
return false
}
sn.RLock()
defer sn.RUnlock()
- if strings.Contains(key, "|") {
- return sn.reservations[key] != nil
- }
- // make sure matches only for the whole appID
- separator := key + "|"
- for resKey := range sn.reservations {
- if strings.HasPrefix(resKey, separator) {
- return true
- }
- }
- return false
+ return sn.reservations[key] != nil
}
-// Reserve the node for this application and ask combination, if not reserved yet.
+// Reserve the node for this application and alloc combination.
// The reservation is checked against the node resources.
-// If the reservation fails the function returns false, if the reservation is made it returns true.
+// If the reservation fails the function returns an error, if the reservation is made it returns nil.
func (sn *Node) Reserve(app *Application, ask *Allocation) error {
- defer sn.notifyListeners()
sn.Lock()
defer sn.Unlock()
- if len(sn.reservations) > 0 {
- return fmt.Errorf("node is already reserved, nodeID %s", sn.NodeID)
- }
appReservation := newReservation(sn, app, ask, false)
- // this should really not happen just guard against panic
- // either app or ask are nil
+ // this should really not happen just guard against panic either app or alloc are nil
if appReservation == nil {
log.Log(log.SchedNode).Debug("reservation creation failed unexpectedly",
zap.String("nodeID", sn.NodeID),
- zap.Any("app", app),
- zap.Any("ask", ask))
- return fmt.Errorf("reservation creation failed app or ask are nil on nodeID %s", sn.NodeID)
+ zap.Stringer("app", app),
+ zap.Stringer("alloc", ask))
+ return fmt.Errorf("reservation creation failed either app or alloc are nil on nodeID %s", sn.NodeID)
+ }
+ reqNode := ask.requiredNode != ""
+ if !reqNode && len(sn.reservations) > 0 {
+ log.Log(log.SchedNode).Warn("normal reservation on already reserved node",
+ zap.String("nodeID", sn.NodeID),
+ zap.String("new app", appReservation.appID),
+ zap.String("new alloc", appReservation.allocKey))
+ return common.ErrorNodeAlreadyReserved
+ }
+ // allow multiple required node reservations on the same node
+ if reqNode {
+ // make sure all other reservations are for required nodes
+ for _, reserved := range sn.reservations {
+ if reserved.alloc.requiredNode == "" {
+ log.Log(log.SchedNode).Warn("trying to add normal reservation to node with required node reservation",
+ zap.String("nodeID", sn.NodeID),
+ zap.String("existing app", reserved.appID),
+ zap.String("existing alloc", reserved.allocKey),
+ zap.String("new app", appReservation.appID),
+ zap.String("new alloc", appReservation.allocKey))
+ return fmt.Errorf("normal reservation: required node reservation present, nodeID %s", sn.NodeID)
+ }
+ }
}
// reservation must fit on the empty node
if !sn.totalResource.FitIn(ask.GetAllocatedResource()) {
log.Log(log.SchedNode).Debug("reservation does not fit on the node",
zap.String("nodeID", sn.NodeID),
zap.String("appID", app.ApplicationID),
- zap.String("ask", ask.GetAllocationKey()),
- zap.Stringer("allocationAsk", ask.GetAllocatedResource()))
- return fmt.Errorf("reservation does not fit on node %s, appID %s, ask %s", sn.NodeID, app.ApplicationID, ask.GetAllocatedResource().String())
+ zap.String("alloc", ask.GetAllocationKey()),
+ zap.Stringer("requested resources", ask.GetAllocatedResource()))
+ return common.ErrorNodeNotFitReserve
}
- sn.reservations[appReservation.getKey()] = appReservation
+ sn.reservations[ask.allocationKey] = appReservation
sn.nodeEvents.SendReservedEvent(sn.NodeID, ask.GetAllocatedResource(), ask.GetAllocationKey())
// reservation added successfully
return nil
}
-// unReserve the node for this application and ask combination
-// If the reservation does not exist it returns 0 for reservations removed, if the reservation is removed it returns 1.
-// The error is set if the reservation key cannot be generated.
-func (sn *Node) unReserve(app *Application, ask *Allocation) (int, error) {
- defer sn.notifyListeners()
+// unReserve the node for this application and alloc combination
+// If the reservation does not exist or alloc is nil it returns 0 for reservations removed,
+// if the reservation is removed it returns 1.
+func (sn *Node) unReserve(alloc *Allocation) int {
+ if alloc == nil {
+ return 0
+ }
sn.Lock()
defer sn.Unlock()
- resKey := reservationKey(nil, app, ask)
- if resKey == "" {
- log.Log(log.SchedNode).Debug("unreserve reservation key create failed unexpectedly",
- zap.String("nodeID", sn.NodeID),
- zap.Any("app", app),
- zap.Any("ask", ask))
- return 0, fmt.Errorf("reservation key failed app or ask are nil on nodeID %s", sn.NodeID)
- }
- if _, ok := sn.reservations[resKey]; ok {
- delete(sn.reservations, resKey)
- sn.nodeEvents.SendUnreservedEvent(sn.NodeID, ask.GetAllocatedResource(), ask.GetAllocationKey())
- return 1, nil
+ if _, ok := sn.reservations[alloc.allocationKey]; ok {
+ delete(sn.reservations, alloc.allocationKey)
+ sn.nodeEvents.SendUnreservedEvent(sn.NodeID, alloc.GetAllocatedResource(), alloc.GetAllocationKey())
+ return 1
}
// reservation was not found
log.Log(log.SchedNode).Debug("reservation not found while removing from node",
zap.String("nodeID", sn.NodeID),
- zap.String("appID", app.ApplicationID),
- zap.String("ask", ask.GetAllocationKey()))
- return 0, nil
+ zap.String("alloc", alloc.GetAllocationKey()),
+ zap.String("appID", alloc.GetApplicationID()))
+ return 0
}
// GetReservations returns all reservation made on this node
diff --git a/pkg/scheduler/objects/node_collection_test.go b/pkg/scheduler/objects/node_collection_test.go
index 30d73e67..a43c49cb 100644
--- a/pkg/scheduler/objects/node_collection_test.go
+++ b/pkg/scheduler/objects/node_collection_test.go
@@ -163,13 +163,8 @@ func TestSetNodeSortingPolicy(t *testing.T) {
node := newNode(nodesInfo[id].nodeID, map[string]resources.Quantity{"vcore": resources.Quantity(defaultCapicity[0]), "memory": resources.Quantity(defaultCapicity[1])})
res := resources.NewResourceFromMap(map[string]resources.Quantity{"vcore": resources.Quantity(nodesInfo[id].allocatedVcore), "memory": resources.Quantity(nodesInfo[id].allocatedMem)})
alloc := newAllocation(fmt.Sprintf("test-app-%d", id+1), fmt.Sprintf("test-%d", id+1), res)
- if ok := node.TryAddAllocation(alloc); !ok {
- t.Error("Allocation error happen in node.")
- }
-
- if err := nc.AddNode(node); err != nil {
- t.Errorf("AddNode error:%s", err.Error())
- }
+ assert.Assert(t, node.TryAddAllocation(alloc), "Allocation error happened on node")
+ assert.NilError(t, nc.AddNode(node), "Adding node to collection failed")
}
conf := configs.PartitionConfig{
@@ -196,9 +191,8 @@ func TestSetNodeSortingPolicy(t *testing.T) {
}
nc.SetNodeSortingPolicy(NewNodeSortingPolicy(conf.NodeSortPolicy.Type, conf.NodeSortPolicy.ResourceWeights))
- iter := nc.GetNodeIterator()
id := 0
- iter.ForEachNode(func(node *Node) bool {
+ nc.GetNodeIterator().ForEachNode(func(node *Node) bool {
assert.Equal(t, node.NodeID, tt.nodesOrder[id], "%s: NodeID wanted %s, but it got %s.", nc.GetNodeSortingPolicy().PolicyType().String(), tt.nodesOrder[id], node.NodeID)
id++
return true
@@ -231,9 +225,7 @@ func TestGetNodeSortingPolicy(t *testing.T) {
alloc := newAllocation(fmt.Sprintf("test-app-%d", id+1), fmt.Sprintf("test-%d", id), res)
node.AddAllocation(alloc)
- if err := nc.AddNode(node); err != nil {
- t.Errorf("AddNode error:%s", err.Error())
- }
+ assert.NilError(t, nc.AddNode(node), "Adding node to collection failed")
}
conf := configs.PartitionConfig{
@@ -260,21 +252,16 @@ func TestGetNodeSortingPolicy(t *testing.T) {
}
nc.SetNodeSortingPolicy(NewNodeSortingPolicy(conf.NodeSortPolicy.Type, conf.NodeSortPolicy.ResourceWeights))
- if ans := nc.GetNodeSortingPolicy().PolicyType().String(); ans != tt.want {
- t.Errorf("got %s, want %s", ans, tt.want)
- }
+ assert.Equal(t, nc.GetNodeSortingPolicy().PolicyType().String(), tt.want, "expected sort policy not set")
- // Checking thes nodes order in iterator is after setting node policy with Default weight{vcore:1, memory:1}.
- iter := nc.GetNodeIterator()
+ // Checking the nodes order in iterator is after setting node policy with Default weight{vcore:1, memory:1}.
index := 0
- iter.ForEachNode(func(node *Node) bool {
+ nc.GetNodeIterator().ForEachNode(func(node *Node) bool {
if index >= len(tt.exceptNodeOrder) {
t.Error("Wrong length of nodes in node iterator.")
}
- if node.NodeID != tt.exceptNodeOrder[index] {
- t.Errorf("Policy: %s, got %s, want %s", nc.GetNodeSortingPolicy().PolicyType().String(), node.NodeID, tt.exceptNodeOrder[index])
- }
+ assert.Equal(t, node.NodeID, tt.exceptNodeOrder[index], "Policy: %s, node order wrong", nc.GetNodeSortingPolicy().PolicyType().String())
index++
return true
})
@@ -292,21 +279,15 @@ func TestGetFullNodeIterator(t *testing.T) {
allocName := fmt.Sprintf("alloc-%02d", i)
app := newApplication(appName, "default", "root.test")
ask := newAllocationAsk(allocName, appName, resources.NewResourceFromMap(map[string]resources.Quantity{"vcore": resources.Quantity(i)}))
- if err := node.Reserve(app, ask); err != nil {
- t.Error("Reserving failed.")
- }
+ assert.NilError(t, node.Reserve(app, ask), "Reserving failed.")
} else {
res := resources.NewResourceFromMap(map[string]resources.Quantity{"vcore": resources.Quantity(i)})
alloc := newAllocation(fmt.Sprintf("test-app-%d", i), fmt.Sprintf("test-%d", i), res)
- if ok := node.TryAddAllocation(alloc); !ok {
- t.Error("Allocation error in node.")
- }
- }
- if err := nc.AddNode(node); err != nil {
- t.Error("Adding another node into BC failed.")
+ assert.Assert(t, node.TryAddAllocation(alloc), "Adding allocation to node failed unexpectedly")
}
+ assert.NilError(t, nc.AddNode(node), "Adding another node into BC failed.")
}
- nodes := make([]*Node, 0)
+ var nodes []*Node
nc.GetFullNodeIterator().ForEachNode(func(node *Node) bool {
nodes = append(nodes, node)
return true
@@ -328,6 +309,9 @@ func TestGetNodeIterator(t *testing.T) {
{"Some nodes are reserved", []bool{false, true, false, true}, []int{1, 3}},
}
+ // Check order of available nodes
+ nsp := []string{policies.FairnessPolicy.String(), policies.BinPackingPolicy.String()}
+
for _, tt := range tests {
t.Run("There are reserved nodes in an instance of node collection.", func(t *testing.T) {
nc := NewNodeCollection("test")
@@ -338,47 +322,29 @@ func TestGetNodeIterator(t *testing.T) {
node := newNode(nodeName, map[string]resources.Quantity{"vcore": resources.Quantity(10)})
if tt.reserved[i-1] {
appName := fmt.Sprintf("app-%02d", i)
- allocName := fmt.Sprintf("alloc-%02d", i)
app := newApplication(appName, "default", "root.test")
- ask := newAllocationAsk(allocName, appName, resources.NewResourceFromMap(map[string]resources.Quantity{"vcore": resources.Quantity(i)}))
- if err := node.Reserve(app, ask); err != nil {
- t.Error("Reserving failed.")
- }
+ ask := newAllocationAsk(fmt.Sprintf("alloc-%02d", i), appName, resources.NewResourceFromMap(map[string]resources.Quantity{"vcore": resources.Quantity(i)}))
+ assert.NilError(t, node.Reserve(app, ask), "Reserving failed.")
} else {
res := resources.NewResourceFromMap(map[string]resources.Quantity{"vcore": resources.Quantity(i)})
alloc := newAllocation(fmt.Sprintf("test-app-%d", i), fmt.Sprintf("test-%d", i), res)
- if ok := node.TryAddAllocation(alloc); !ok {
- t.Error("Allocation error happen in node.")
- }
- }
-
- if err := nc.AddNode(node); err != nil {
- t.Error("Adding another node into BC failed.")
+ assert.Assert(t, node.TryAddAllocation(alloc), "Adding allocation to node failed unexpectedly")
}
+ assert.NilError(t, nc.AddNode(node), "Adding another node into BC failed.")
}
- // Check order of avialble nodes
- NodeSortingPolicy := []string{policies.FairnessPolicy.String(), policies.BinPackingPolicy.String()}
-
// Fair policy
- nc.SetNodeSortingPolicy(NewNodeSortingPolicy(NodeSortingPolicy[0], nil))
- iter := nc.GetNodeIterator()
- if ans := nc.GetNodeSortingPolicy().PolicyType().String(); ans != NodeSortingPolicy[0] {
- t.Errorf("got %s, want %s", ans, NodeSortingPolicy[0])
- }
- index := 0
- iter.ForEachNode(func(node *Node) bool {
- fmt.Println(node.NodeID)
- return true
- })
+ nc.SetNodeSortingPolicy(NewNodeSortingPolicy(nsp[0], nil))
+ assert.Equal(t, nc.GetNodeSortingPolicy().PolicyType().String(), nsp[0], "expected sort policy not set")
- iter.ForEachNode(func(node *Node) bool {
+ index := 0
+ nc.GetNodeIterator().ForEachNode(func(node *Node) bool {
if index >= len(tt.wantWithFair) {
t.Errorf("Want length of nodes: %d, Get length of nodes: %d", index, len(tt.wantWithFair))
}
if want := fmt.Sprintf("node-%d", tt.wantWithFair[index]); node.NodeID != want {
- t.Errorf("%s with %s, Want %s, got %s.", tt.name, NodeSortingPolicy[0], want, node.NodeID)
+ t.Errorf("%s with %s, Want %s, got %s.", tt.name, nsp[0], want, node.NodeID)
}
index++
@@ -386,26 +352,87 @@ func TestGetNodeIterator(t *testing.T) {
})
// Binpacking policy
- nc.SetNodeSortingPolicy(NewNodeSortingPolicy(NodeSortingPolicy[1], nil))
- if ans := nc.GetNodeSortingPolicy().PolicyType().String(); ans != NodeSortingPolicy[1] {
- t.Errorf("got %s, want %s", ans, NodeSortingPolicy[1])
- }
+ nc.SetNodeSortingPolicy(NewNodeSortingPolicy(nsp[1], nil))
+ assert.Equal(t, nc.GetNodeSortingPolicy().PolicyType().String(), nsp[1], "expected sort policy not set")
- iter = nc.GetNodeIterator()
- DescreasingIndex := len(tt.wantWithFair) - 1
+ decIndex := len(tt.wantWithFair) - 1
index = 0
- iter.ForEachNode(func(node *Node) bool {
+ nc.GetNodeIterator().ForEachNode(func(node *Node) bool {
if index >= len(tt.wantWithFair) {
t.Errorf("Want length of nodes: %d, Get length of nodes: %d", index, len(tt.wantWithFair))
}
- if want := fmt.Sprintf("node-%d", tt.wantWithFair[DescreasingIndex]); node.NodeID != want {
- t.Errorf("%s with %s, want %s, got %s.", tt.name, NodeSortingPolicy[1], want, node.NodeID)
+ if want := fmt.Sprintf("node-%d", tt.wantWithFair[decIndex]); node.NodeID != want {
+ t.Errorf("%s with %s, want %s, got %s.", tt.name, nsp[1], want, node.NodeID)
}
index++
- DescreasingIndex--
+ decIndex--
return true
})
})
}
}
+
+// TestNodeIteratorReserveUpdate reservation add or remove should not need a node collection update make sure it works.
+// YUNIKORN-2976 removed the listener notify in Reserve and unReserve
+func TestNodeIteratorReserveUpdate(t *testing.T) {
+ nc := NewNodeCollection("test")
+ count := 3
+ for i := 0; i < count; i++ {
+ node := newNode(fmt.Sprintf("node-%d", i), map[string]resources.Quantity{"some": resources.Quantity(10)})
+ assert.NilError(t, nc.AddNode(node), "Adding another node into BC failed.")
+ }
+ // first check: both iterators return all nodes
+ allNodes := make([]*Node, 0)
+ nc.GetFullNodeIterator().ForEachNode(func(node *Node) bool {
+ allNodes = append(allNodes, node)
+ return true
+ })
+ assert.Equal(t, len(allNodes), count, "wrong length")
+
+ var itNodes []*Node
+ nc.GetNodeIterator().ForEachNode(func(node *Node) bool {
+ itNodes = append(itNodes, node)
+ return true
+ })
+ assert.Equal(t, len(itNodes), count, "wrong length")
+
+ // add reservation to all nodes
+ app := newApplication(appID0, "default", "root.test")
+ for i, node := range allNodes {
+ ask := newAllocationAsk(fmt.Sprintf("ask-%d", i), appID0, resources.NewResourceFromMap(map[string]resources.Quantity{"some": resources.Quantity(5)}))
+ app.AddAllocation(ask)
+ assert.NilError(t, node.Reserve(app, ask), "Reserving failed.")
+ }
+
+ // full iterator returns all nodes
+ itNodes = nil
+ nc.GetFullNodeIterator().ForEachNode(func(node *Node) bool {
+ itNodes = append(itNodes, node)
+ return true
+ })
+ assert.Equal(t, len(itNodes), count, "wrong length")
+
+ // filtered iterator returns NO nodes
+ itNodes = nil
+ nc.GetNodeIterator().ForEachNode(func(node *Node) bool {
+ itNodes = append(itNodes, node)
+ return true
+ })
+ assert.Equal(t, len(itNodes), 0, "wrong length")
+
+ // run over initial list of nodes and remove reservations.
+ // only one reservation so just pick that one
+ for _, node := range allNodes {
+ alloc := node.GetReservations()[0].alloc
+ assert.Equal(t, node.unReserve(alloc), 1, "unReserve should have returned a single removal")
+ assert.Assert(t, !node.IsReserved(), "node should not have been reserved")
+ }
+ // filtered iterator returns all nodes again
+ itNodes = nil
+ nc.GetNodeIterator().ForEachNode(func(node *Node) bool {
+ itNodes = append(itNodes, node)
+ return true
+ })
+ assert.Equal(t, len(itNodes), count, "wrong length")
+}
diff --git a/pkg/scheduler/objects/node_test.go b/pkg/scheduler/objects/node_test.go
index 27ad614d..6a725ac8 100644
--- a/pkg/scheduler/objects/node_test.go
+++ b/pkg/scheduler/objects/node_test.go
@@ -105,10 +105,9 @@ func TestCheckConditions(t *testing.T) {
}
func TestPreAllocateCheck(t *testing.T) {
- nodeID := nodeID1
resNode := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 10, "second": 1})
- node := newNode(nodeID, resNode.Resources)
- if node == nil || node.NodeID != nodeID {
+ node := newNode(nodeID1, resNode.Resources)
+ if node == nil || node.NodeID != nodeID1 {
t.Fatalf("node create failed which should not have %v", node)
}
@@ -132,24 +131,20 @@ func TestPreAllocateCheck(t *testing.T) {
assert.Assert(t, !node.preAllocateCheck(resOther, ""), "unknown resource type should not have fitted on node")
// set allocated resource
- alloc := newAllocation(appID1, nodeID, resSmall)
+ alloc := newAllocation(appID1, nodeID1, resSmall)
node.AddAllocation(alloc)
assert.Assert(t, node.preAllocateCheck(resSmall, ""), "small resource should have fitted in available allocation")
assert.Assert(t, !node.preAllocateCheck(resNode, ""), "node resource should not have fitted in available allocation")
// check if we can allocate on a reserved node
- q := map[string]resources.Quantity{"first": 0}
- res := resources.NewResourceFromMap(q)
- ask := newAllocationAsk(aKey, appID1, res)
+ ask := newAllocationAsk(aKey, appID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 0}))
app := newApplication(appID1, "default", "root.unknown")
// standalone reservation unreserve returns false as app is not reserved
reserve := newReservation(node, app, ask, false)
- node.reservations[reserve.getKey()] = reserve
- assert.Assert(t, !node.preAllocateCheck(resSmall, "app-2"), "node was reserved for different app but check passed")
- assert.Assert(t, !node.preAllocateCheck(resSmall, "app-1|alloc-2"), "node was reserved for this app but not the alloc and check passed")
- assert.Assert(t, node.preAllocateCheck(resSmall, appID1), "node was reserved for this app but check did not pass check")
- assert.Assert(t, node.preAllocateCheck(resSmall, "app-1|alloc-1"), "node was reserved for this app/alloc but check did not pass check")
+ node.reservations[reserve.allocKey] = reserve
+ assert.Assert(t, !node.preAllocateCheck(resSmall, aKey2), "node was reserved for different app but check passed")
+ assert.Assert(t, node.preAllocateCheck(resSmall, aKey), "node was reserved for this app/alloc but check did not pass check")
// Check if we can allocate on non scheduling node
node.SetSchedulable(false)
@@ -190,20 +185,14 @@ func TestNodeReservation(t *testing.T) {
if node == nil || node.NodeID != nodeID1 {
t.Fatalf("node create failed which should not have %v", node)
}
- if node.IsReserved() {
- t.Fatal("new node should not have reservations")
- }
- if node.isReservedForApp("") {
- t.Error("new node should not have reservations for empty key")
- }
- if node.isReservedForApp("unknown") {
- t.Error("new node should not have reservations for unknown key")
- }
+ assert.Assert(t, !node.IsReserved(), "new node should not have reservations")
+ assert.Assert(t, !node.isReservedForAllocation(""), "new node should not have reservations for empty key")
+ assert.Assert(t, !node.isReservedForAllocation("unknown"), "new node should not have reservations for unknown key")
// reserve illegal request
err := node.Reserve(nil, nil)
if err == nil {
- t.Errorf("illegal reservation requested but did not fail: error %v", err)
+ t.Fatal("illegal reservation requested but did not fail")
}
// too large for node
@@ -212,7 +201,7 @@ func TestNodeReservation(t *testing.T) {
app := newApplication(appID1, "default", "root.unknown")
err = node.Reserve(app, ask)
if err == nil {
- t.Errorf("requested reservation does not fit in node resource but did not fail: error %v", err)
+ t.Fatal("requested reservation does not fit in node resource but did not fail")
}
// resource type not available on node
@@ -221,7 +210,7 @@ func TestNodeReservation(t *testing.T) {
app = newApplication(appID1, "default", "root.unknown")
err = node.Reserve(app, ask)
if err == nil {
- t.Errorf("requested reservation does not match node resource types but did not fail: error %v", err)
+ t.Fatal("requested reservation does not match node resource types but did not fail")
}
res = resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})
@@ -230,39 +219,29 @@ func TestNodeReservation(t *testing.T) {
// reserve that works
err = node.Reserve(app, ask)
assert.NilError(t, err, "reservation should not have failed")
- if node.isReservedForApp("") {
- t.Error("node should not have reservations for empty key")
- }
- if node.isReservedForApp("unknown") {
- t.Errorf("node should not have reservations for unknown key")
- }
- if node.IsReserved() && !node.isReservedForApp(appID1) {
- t.Errorf("node should have reservations for app-1")
- }
+ assert.Assert(t, !node.isReservedForAllocation(""), "node should not have reservations for empty key")
+ assert.Assert(t, !node.isReservedForAllocation("unknown"), "node should not have reservations for unknown key")
+ assert.Assert(t, node.IsReserved(), "node should have been reserved")
+ assert.Assert(t, node.isReservedForAllocation(aKey), "node should have reservations for alloc-1")
// 2nd reservation on node
- err = node.Reserve(nil, nil)
+ app2 := newApplication(appID2, "default", "root.unknown")
+ ask2 := newAllocationAsk("alloc-2", appID2, res)
+ err = node.Reserve(app2, ask2)
if err == nil {
- t.Errorf("reservation requested on already reserved node: error %v", err)
+ t.Fatal("reservation requested on already reserved node")
}
- // unreserve different app
- _, err = node.unReserve(nil, nil)
- if err == nil {
- t.Errorf("illegal reservation release but did not fail: error %v", err)
- }
- ask2 := newAllocationAsk("alloc-2", appID2, res)
- app2 := newApplication(appID2, "default", "root.unknown")
- var num int
- num, err = node.unReserve(app2, ask2)
- assert.NilError(t, err, "un-reserve different app should have failed without error")
- assert.Equal(t, num, 0, "un-reserve different app should have failed without releases")
- num, err = node.unReserve(app, ask)
- assert.NilError(t, err, "un-reserve should not have failed")
+ // unreserve different alloc
+ num := node.unReserve(nil)
+ assert.Equal(t, num, 0, "un-reserve different alloc should have failed without releases")
+ num = node.unReserve(ask2)
+ assert.Equal(t, num, 0, "un-reserve different alloc should have failed without releases")
+ num = node.unReserve(ask)
assert.Equal(t, num, 1, "un-reserve app should have released ")
}
-func TestIsReservedForApp(t *testing.T) {
+func TestRequiredNodeAfterReservation(t *testing.T) {
node := newNode(nodeID1, map[string]resources.Quantity{"first": 10})
if node == nil || node.NodeID != nodeID1 {
t.Fatalf("node create failed which should not have %v", node)
@@ -270,33 +249,76 @@ func TestIsReservedForApp(t *testing.T) {
if node.IsReserved() {
t.Fatal("new node should not have reservations")
}
-
- // check if we can allocate on a reserved node
- q := map[string]resources.Quantity{"first": 0}
- res := resources.NewResourceFromMap(q)
- ask := newAllocationAsk(aKey, appID1, res)
+ // normal node reservation
+ allocRes := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})
+ ask := newAllocationAsk(aKey, appID1, allocRes)
app := newApplication(appID1, "default", "root.unknown")
+ err := node.Reserve(app, ask)
+ assert.NilError(t, err, "normal node reservation should not have failed")
+ assert.Assert(t, node.IsReserved(), "node should have been reserved")
+ assert.Assert(t, node.isReservedForAllocation(aKey), "node should have reservations for alloc-1")
- // standalone reservation unreserve returns false as app is not reserved
- reserve := newReservation(node, app, ask, false)
- node.reservations[reserve.getKey()] = reserve
- if node.isReservedForApp("app-2") {
- t.Error("node was reserved for different app but check passed ")
+ ask2 := newAllocationAsk(aKey2, appID1, allocRes)
+ ask2.requiredNode = nodeID1
+ app = newApplication(appID1, "default", "root.unknown")
+ // required node as 2nd reservation
+ err = node.Reserve(app, ask2)
+ if err == nil {
+ t.Fatalf("adding required node reservation should have failed")
}
- if node.isReservedForApp("app-1|alloc-2") {
- t.Error("node was reserved for this app but not the alloc and check passed ")
+}
+
+func TestMultiRequiredNodeReservation(t *testing.T) {
+ node := newNode(nodeID1, map[string]resources.Quantity{"first": 10})
+ if node == nil || node.NodeID != nodeID1 {
+ t.Fatalf("node create failed which should not have %v", node)
}
- if !node.isReservedForApp(appID1) {
- t.Error("node was reserved for this app but check did not passed ")
+ if node.IsReserved() {
+ t.Fatal("new node should not have reservations")
}
- if !node.isReservedForApp("app-1|alloc-1") {
- t.Error("node was reserved for this app/alloc but check did not passed ")
+
+ // required node reservation
+ allocRes := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})
+ ask := newAllocationAsk(aKey, appID1, allocRes)
+ ask.requiredNode = nodeID1
+ app := newApplication(appID1, "default", "root.unknown")
+ err := node.Reserve(app, ask)
+ assert.NilError(t, err, "required node reservation should not have failed")
+ assert.Assert(t, node.IsReserved(), "node should have been reserved")
+ assert.Assert(t, node.isReservedForAllocation(aKey), "node should have reservations for alloc-1")
+
+ ask2 := newAllocationAsk(aKey2, appID1, allocRes)
+ app = newApplication(appID1, "default", "root.unknown")
+ // non required node as 2nd reservation
+ err = node.Reserve(app, ask2)
+ if err == nil {
+ t.Fatalf("adding to required node reservation should have failed")
}
- // app name similarity check: chop of the last char to make sure we check the full name
- similar := appID1[:len(appID1)-1]
- if node.isReservedForApp(similar) {
- t.Errorf("similar app should not have reservations on node %s", similar)
+
+ // required node as 2nd reservation on node
+ ask2.requiredNode = nodeID1
+ err = node.Reserve(app, ask2)
+ assert.NilError(t, err, "required node reservation should not have failed")
+ assert.Assert(t, node.IsReserved(), "node should have been reserved")
+ assert.Assert(t, node.isReservedForAllocation(aKey2), "node should have reservations for alloc-2")
+}
+
+func TestIsReservedForAllocation(t *testing.T) {
+ node := newNode(nodeID1, map[string]resources.Quantity{"first": 10})
+ if node == nil || node.NodeID != nodeID1 {
+ t.Fatalf("node create failed which should not have %v", node)
}
+ assert.Assert(t, !node.IsReserved(), "new node should not have reservations")
+
+ // check if we can allocate on a reserved node
+ ask := newAllocationAsk(aKey, appID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 0}))
+ app := newApplication(appID1, "default", "root.unknown")
+
+ // standalone reservation unreserve returns false as app is not reserved
+ reserve := newReservation(node, app, ask, false)
+ node.reservations[reserve.allocKey] = reserve
+ assert.Assert(t, !node.isReservedForAllocation(aKey2), "node was reserved for different alloc but check passed ")
+ assert.Assert(t, node.isReservedForAllocation(aKey), "node was reserved for this alloc but check did not passed ")
}
func TestAttributes(t *testing.T) {
@@ -858,8 +880,7 @@ func TestNodeEvents(t *testing.T) {
assert.Equal(t, si.EventRecord_NODE_RESERVATION, event.EventChangeDetail)
mockEvents.Reset()
- _, err = node.unReserve(app, ask)
- assert.NilError(t, err, "could not unreserve")
+ assert.Equal(t, node.unReserve(ask), 1, "expected the reservation to be removed")
event = mockEvents.Events[0]
assert.Equal(t, si.EventRecord_NODE, event.Type)
assert.Equal(t, si.EventRecord_REMOVE, event.EventChangeType)
diff --git a/pkg/scheduler/objects/preemption.go b/pkg/scheduler/objects/preemption.go
index 10cacea2..c70aee4b 100644
--- a/pkg/scheduler/objects/preemption.go
+++ b/pkg/scheduler/objects/preemption.go
@@ -162,7 +162,7 @@ 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.isReservedForApp(reservationKey(nil, p.application, p.ask))) || !node.FitInNode(p.ask.GetAllocatedResource()) {
+ if !node.IsSchedulable() || (node.IsReserved() && !node.isReservedForAllocation(p.ask.GetAllocationKey())) || !node.FitInNode(p.ask.GetAllocatedResource()) {
// node is not available, remove any potential victims from consideration
delete(allocationsByNode, node.NodeID)
} else {
diff --git a/pkg/scheduler/objects/reservation.go b/pkg/scheduler/objects/reservation.go
index 5c7f672b..3001372d 100644
--- a/pkg/scheduler/objects/reservation.go
+++ b/pkg/scheduler/objects/reservation.go
@@ -21,37 +21,36 @@ package objects
import (
"go.uber.org/zap"
- "github.com/apache/yunikorn-core/pkg/common"
"github.com/apache/yunikorn-core/pkg/log"
)
type reservation struct {
- nodeID string
- appID string
- askKey string
- // these references must ONLY be used for ask, node and application removal otherwise
+ appID string
+ nodeID string
+ allocKey string
+ // these references must ONLY be used for alloc, node and application removal otherwise
// the reservations cannot be removed and scheduling might be impacted.
- app *Application
- node *Node
- ask *Allocation
+ app *Application
+ node *Node
+ alloc *Allocation
}
// The reservation inside the scheduler. A reservation object is never mutated and does not use locking.
// The key depends on where the reservation was made (node or app).
// appBased must be true for a reservation for an app and false for a reservation on a node
-func newReservation(node *Node, app *Application, ask *Allocation, appBased bool) *reservation {
- if ask == nil || app == nil || node == nil {
+func newReservation(node *Node, app *Application, alloc *Allocation, appBased bool) *reservation {
+ if alloc == nil || app == nil || node == nil {
log.Log(log.SchedReservation).Warn("Illegal reservation requested: one input is nil",
zap.Stringer("node", node),
zap.Stringer("app", app),
- zap.Stringer("ask", ask))
+ zap.Stringer("alloc", alloc))
return nil
}
res := &reservation{
- askKey: ask.GetAllocationKey(),
- ask: ask,
- app: app,
- node: node,
+ allocKey: alloc.GetAllocationKey(),
+ alloc: alloc,
+ app: app,
+ node: node,
}
if appBased {
res.nodeID = node.NodeID
@@ -61,51 +60,21 @@ func newReservation(node *Node, app *Application, ask *Allocation, appBased bool
return res
}
-func reservationKey(node *Node, app *Application, ask *Allocation) string {
- if ask == nil || (app == nil && node == nil) || (app != nil && node != nil) {
- log.Log(log.SchedReservation).Warn("Illegal reservation key requested",
- zap.Any("node", node),
- zap.Any("app", app),
- zap.Any("ask", ask))
- return ""
- }
- if node == nil {
- return ask.resKeyWithoutNode
- }
- key := ask.getReservationKeyForNode(node.NodeID)
- if key != common.Empty {
- return key
- }
-
- key = node.NodeID + "|" + ask.GetAllocationKey()
- ask.setReservationKeyForNode(node.NodeID, key)
- return key
-}
-
-func reservationKeyWithoutNode(appID, allocKey string) string {
- return appID + "|" + allocKey
-}
-
-// Return the reservation key
-func (r *reservation) getKey() string {
- if r.nodeID == "" {
- return r.appID + "|" + r.askKey
- }
- return r.nodeID + "|" + r.askKey
-}
-
func (r *reservation) String() string {
+ if r == nil {
+ return "nil reservation"
+ }
if r.nodeID == "" {
- return r.node.NodeID + " -> " + r.appID + "|" + r.askKey
+ return r.node.NodeID + " -> " + r.appID + "|" + r.allocKey
}
- return r.app.ApplicationID + " -> " + r.nodeID + "|" + r.askKey
+ return r.app.ApplicationID + " -> " + r.nodeID + "|" + r.allocKey
}
// GetObjects returns the objects that created the reservation.
// None of the returned values will be nil unless the reservation itself is nil
func (r *reservation) GetObjects() (*Node, *Application, *Allocation) {
if r != nil {
- return r.node, r.app, r.ask
+ return r.node, r.app, r.alloc
}
return nil, nil, nil
}
diff --git a/pkg/scheduler/objects/reservation_test.go b/pkg/scheduler/objects/reservation_test.go
index 617ee55a..69716332 100644
--- a/pkg/scheduler/objects/reservation_test.go
+++ b/pkg/scheduler/objects/reservation_test.go
@@ -33,62 +33,49 @@ func TestNewReservation(t *testing.T) {
app := newApplication("app-1", "default", "root.unknown")
node := newNodeRes("node-1", res)
- // check the basics (failures)
- reserve := newReservation(nil, nil, nil, true)
- if reserve != nil {
- t.Errorf("reservation with nil objects should have returned nil: %v", reserve)
+ tests := []struct {
+ name string
+ node *Node
+ app *Application
+ ask *Allocation
+ appBased bool
+ expected *reservation
+ }{
+ {"nil input", nil, nil, nil, true, nil},
+ {"nil alloc", node, app, nil, true, nil},
+ {"nil app", node, nil, ask, true, nil},
+ {"nil node", nil, app, ask, true, nil},
+ {"node based", node, app, ask, false, &reservation{"app-1", "", "alloc-1", app, node, ask}},
+ {"app based", node, app, ask, true, &reservation{"", "node-1", "alloc-1", app, node, ask}},
}
- reserve = newReservation(node, app, nil, true)
- if reserve != nil {
- t.Errorf("reservation with nil ask set should have returned nil: '%v'", reserve)
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ reserve := newReservation(tt.node, tt.app, tt.ask, tt.appBased)
+ if tt.expected == nil {
+ assert.Equal(t, tt.expected, reserve, "unexpected reservation")
+ } else {
+ assert.Equal(t, reserve.appID, tt.expected.appID, "incorrect appID")
+ assert.Equal(t, reserve.nodeID, tt.expected.nodeID, "incorrect node ID")
+ assert.Equal(t, reserve.allocKey, tt.expected.allocKey, "incorrect alloc key")
+ if tt.appBased {
+ assert.Equal(t, reserve.String(), "app-1 -> node-1|alloc-1", "incorrect string form")
+ } else {
+ assert.Equal(t, reserve.String(), "node-1 -> app-1|alloc-1", "incorrect string form")
+ }
+ }
+ })
}
- reserve = newReservation(node, nil, ask, true)
- if reserve != nil {
- t.Errorf("reservation with nil app set should have returned nil: '%v'", reserve)
- }
- reserve = newReservation(nil, app, ask, true)
- if reserve != nil {
- t.Errorf("reservation with nil node set should have returned nil: '%v'", reserve)
- }
-
- // working cases
- reserve = newReservation(node, app, ask, true)
- if reserve == nil {
- t.Fatalf("reservation with all objects set should have returned nil: %v", reserve)
- }
- assert.Equal(t, reserve.getKey(), "node-1|alloc-1", "incorrect node reservation key")
- assert.Equal(t, reserve.String(), "app-1 -> node-1|alloc-1", "incorrect string form")
-
- reserve = newReservation(node, app, ask, false)
- if reserve == nil {
- t.Fatalf("reservation with all objects set should have returned nil: %v", reserve)
- }
- assert.Equal(t, reserve.getKey(), "app-1|alloc-1", "incorrect app reservation key")
- assert.Equal(t, reserve.String(), "node-1 -> app-1|alloc-1", "incorrect string form")
}
-func TestReservationKey(t *testing.T) {
- // create the input objects
- res := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1})
- ask := newAllocationAsk("alloc-1", "app-1", res)
- app := newApplication("app-1", "default", "root.unknown")
- node := newNodeRes("node-1", res)
-
- // check the basics
- reserve := reservationKey(nil, nil, nil)
- assert.Equal(t, reserve, "", "reservation with nil objects should have empty key")
- reserve = reservationKey(node, app, nil)
- assert.Equal(t, reserve, "", "reservation with nil ask set should have empty key")
- reserve = reservationKey(node, app, ask)
- assert.Equal(t, reserve, "", "reservation with all objects set should have empty key")
-
- // other cases
- reserve = reservationKey(node, nil, ask)
- assert.Equal(t, reserve, "node-1|alloc-1", "incorrect node reservation key")
- assert.Equal(t, "node-1|alloc-1", ask.resKeyPerNode["node-1"])
- assert.Equal(t, 1, len(ask.resKeyPerNode))
- reserve = reservationKey(nil, app, ask)
- assert.Equal(t, reserve, "app-1|alloc-1", "incorrect app reservation key")
+func TestReservationString(t *testing.T) {
+ var nilReserve *reservation
+ defer func() {
+ if r := recover(); r != nil {
+ t.Fatal("panic on nil reservation in object test")
+ }
+ }()
+ str := nilReserve.String()
+ assert.Equal(t, "nil reservation", str, "nil reservation did not return correct string")
}
func TestGetObjects(t *testing.T) {
diff --git a/pkg/scheduler/objects/utilities_test.go b/pkg/scheduler/objects/utilities_test.go
index 97be13c2..20015a87 100644
--- a/pkg/scheduler/objects/utilities_test.go
+++ b/pkg/scheduler/objects/utilities_test.go
@@ -45,6 +45,7 @@ const (
appID3 = "app-3"
aKey = "alloc-1"
aKey2 = "alloc-2"
+ aKey3 = "alloc-3"
nodeID1 = "node-1"
nodeID2 = "node-2"
instType1 = "itype-1"
diff --git a/pkg/scheduler/partition.go b/pkg/scheduler/partition.go
index 7db7e4fe..1375d185 100644
--- a/pkg/scheduler/partition.go
+++ b/pkg/scheduler/partition.go
@@ -891,6 +891,9 @@ func (pc *PartitionContext) allocate(result *objects.AllocationResult) *objects.
return nil
}
+ // reservations were cancelled during the processing
+ pc.decReservationCount(result.CancelledReservations)
+
// reservation
if result.ResultType == objects.Reserved {
pc.reserve(app, targetNode, result.Request)
@@ -949,12 +952,23 @@ func (pc *PartitionContext) allocate(result *objects.AllocationResult) *objects.
// Lock free call this must be called holding the context lock
func (pc *PartitionContext) reserve(app *objects.Application, node *objects.Node, ask *objects.Allocation) {
appID := app.ApplicationID
- // app has node already reserved cannot reserve again
- if app.IsReservedOnNode(node.NodeID) {
- log.Log(log.SchedPartition).Info("Application is already reserved on node",
+ // check if ask has reserved already, cannot have multiple reservations for one ask
+ nodeID := app.NodeReservedForAsk(ask.GetAllocationKey())
+ // We should not see a reservation for this ask yet
+ // sanity check the node that is reserved: same node just be done
+ // different node: fix it, unreserve the old node and reserve the new one
+ // this is all to safeguard the system it should never happen.
+ if nodeID != "" {
+ // same nodeID we do not need to do anything
+ if nodeID == node.NodeID {
+ return
+ }
+ log.Log(log.SchedPartition).Warn("ask is already reserved on different node, fixing reservations",
zap.String("appID", appID),
- zap.String("nodeID", node.NodeID))
- return
+ zap.String("allocationKey", ask.GetAllocationKey()),
+ zap.String("reserved nodeID", nodeID),
+ zap.String("new nodeID", node.NodeID))
+ pc.unReserve(app, pc.nodes.GetNode(nodeID), ask)
}
// all ok, add the reservation to the app, this will also reserve the node
if err := app.Reserve(node, ask); err != nil {
@@ -978,13 +992,7 @@ func (pc *PartitionContext) reserve(app *objects.Application, node *objects.Node
// NOTE: this is a lock free call. It must NOT be called holding the PartitionContext lock.
func (pc *PartitionContext) unReserve(app *objects.Application, node *objects.Node, ask *objects.Allocation) {
// remove the reservation of the app, this will also unReserve the node
- var err error
- var num int
- if num, err = app.UnReserve(node, ask); err != nil {
- log.Log(log.SchedPartition).Info("Failed to unreserve, error during allocate on the app",
- zap.Error(err))
- return
- }
+ num := app.UnReserve(node, ask)
// remove the reservation of the queue
appID := app.ApplicationID
app.GetQueue().UnReserve(appID, num)
diff --git a/pkg/scheduler/partition_test.go b/pkg/scheduler/partition_test.go
index 75e529eb..673c3948 100644
--- a/pkg/scheduler/partition_test.go
+++ b/pkg/scheduler/partition_test.go
@@ -484,7 +484,7 @@ func TestPlaceholderDataWithPlaceholderPreemption(t *testing.T) {
}
// check if updated (must be after allocate call)
assert.Equal(t, 1, len(app2.GetReservations()), "app reservation should have been updated")
- assert.Equal(t, 1, len(app2.GetAskReservations(allocKey2)), "ask should have been reserved")
+ assert.Equal(t, app2.NodeReservedForAsk(allocKey2), nodeID2, "ask should have been reserved")
// try through reserved scheduling cycle this should trigger preemption
result = partition.tryReservedAllocate()
@@ -1626,6 +1626,10 @@ func TestRequiredNodeReservation(t *testing.T) {
if result := partition.tryAllocate(); result != nil {
t.Fatalf("empty cluster allocate returned allocation: %s", result)
}
+ node := partition.nodes.GetNode(nodeID1)
+ if node == nil {
+ t.Fatal("node-1 should have been created")
+ }
app := newApplication(appID1, "default", "root.parent.sub-leaf")
res, err := resources.NewResourceFromConf(map[string]string{"vcore": "8"})
@@ -1661,7 +1665,12 @@ func TestRequiredNodeReservation(t *testing.T) {
}
// check if updated (must be after allocate call)
assert.Equal(t, 1, len(app.GetReservations()), "app should have one reserved ask")
- assert.Equal(t, 1, len(app.GetAskReservations(allocKey2)), "ask should have been reserved")
+ assert.Equal(t, app.NodeReservedForAsk(allocKey2), nodeID1, "ask should have been reserved on node-1")
+ assert.Assert(t, node.IsReserved(), "node should have been reserved")
+ reservations := node.GetReservations()
+ assert.Equal(t, len(reservations), 1, "node should have two reservations")
+ _, _, resAsk := reservations[0].GetObjects()
+ assert.Equal(t, resAsk.GetAllocationKey(), allocKey2, "alloc-2 should have been reserved on the node")
assertLimits(t, getTestUserGroup(), res)
// allocation that fits on the node should not be allocated
@@ -1669,7 +1678,7 @@ func TestRequiredNodeReservation(t *testing.T) {
res2, err = resources.NewResourceFromConf(map[string]string{"vcore": "1"})
assert.NilError(t, err, "failed to create resource")
- ask3 := newAllocationAsk("alloc-3", appID1, res2)
+ ask3 := newAllocationAsk(allocKey3, appID1, res2)
ask3.SetRequiredNode(nodeID1)
err = app.AddAllocationAsk(ask3)
assert.NilError(t, err, "failed to add ask alloc-3 to app-1")
@@ -1678,13 +1687,24 @@ func TestRequiredNodeReservation(t *testing.T) {
t.Fatal("allocation attempt should not have returned an allocation")
}
- // reservation count remains same as last try allocate should have failed to find a reservation
- assert.Equal(t, 1, len(app.GetReservations()), "ask should not have been reserved, count changed")
+ // reservation count should be updated as tryAllocate should have added a reservation for the 3rd ask
+ assert.Equal(t, 2, len(app.GetReservations()), "ask should not have been reserved, count changed")
+ assert.Equal(t, app.NodeReservedForAsk(allocKey2), nodeID1, "alloc-2 should have been reserved on node-1")
+ assert.Equal(t, app.NodeReservedForAsk(allocKey3), nodeID1, "alloc-3 should have been reserved on node-1")
+ assert.Assert(t, node.IsReserved(), "node should have been reserved")
+ reservations = node.GetReservations()
+ assert.Equal(t, len(reservations), 2, "node should have two reservations")
+ _, _, resAsk = reservations[0].GetObjects()
+ _, _, resAsk2 := reservations[1].GetObjects()
+ if !((resAsk.GetAllocationKey() == allocKey3 && resAsk2.GetAllocationKey() == allocKey2) ||
+ (resAsk2.GetAllocationKey() == allocKey3 && resAsk.GetAllocationKey() == allocKey2)) {
+ t.Fatal("missing reservation on the node")
+ }
assertLimits(t, getTestUserGroup(), res)
}
// allocate ask request with required node having non daemon set reservations
-func TestRequiredNodeCancelNonDSReservations(t *testing.T) {
+func TestRequiredNodeCancelOtherReservations(t *testing.T) {
partition := createQueuesNodes(t)
if partition == nil {
t.Fatal("partition create failed")
@@ -1733,8 +1753,9 @@ func TestRequiredNodeCancelNonDSReservations(t *testing.T) {
t.Fatal("2nd allocation did not return the correct allocation")
}
// check if updated (must be after allocate call)
- assert.Equal(t, 1, len(app.GetReservations()), "ask should have been reserved")
+ assert.Equal(t, 1, len(app.GetReservations()), "allocation should have been reserved")
assert.Equal(t, 1, len(app.GetQueue().GetReservedApps()), "queue reserved apps should be 1")
+ assert.Equal(t, 1, partition.reservations, "partition reservations should be 1")
res1, err := resources.NewResourceFromConf(map[string]string{"vcore": "1"})
assert.NilError(t, err, "failed to create resource")
@@ -1757,8 +1778,9 @@ func TestRequiredNodeCancelNonDSReservations(t *testing.T) {
assert.Equal(t, objects.Allocated, result.ResultType, "allocation result type should have been allocated")
// earlier app (app1) reservation count should be zero
- assert.Equal(t, 0, len(app.GetReservations()), "ask should have been reserved")
+ assert.Equal(t, 0, len(app.GetReservations()), "allocation should not have been reserved")
assert.Equal(t, 0, len(app.GetQueue().GetReservedApps()), "queue reserved apps should be 0")
+ assert.Equal(t, 0, partition.reservations, "partition reservations should be 0")
}
// allocate ask request with required node having daemon set reservations
@@ -1787,11 +1809,11 @@ func TestRequiredNodeCancelDSReservations(t *testing.T) {
err = partition.AddApplication(app)
assert.NilError(t, err, "failed to add app-1 to partition")
- ask := newAllocationAsk("alloc-1", appID1, res)
+ ask := newAllocationAsk(allocKey, appID1, res)
ask.SetRequiredNode(nodeID1)
err = app.AddAllocationAsk(ask)
assert.NilError(t, err, "failed to add ask 1 to app")
- ask = newAllocationAsk("alloc-2", appID1, res)
+ ask = newAllocationAsk(allocKey2, appID1, res)
ask.SetRequiredNode(nodeID1)
err = app.AddAllocationAsk(ask)
assert.NilError(t, err, "failed to add ask 2 to app")
@@ -1815,6 +1837,7 @@ func TestRequiredNodeCancelDSReservations(t *testing.T) {
// check if updated (must be after allocate call)
assert.Equal(t, 1, len(app.GetReservations()), "ask should have been reserved")
assert.Equal(t, 1, len(app.GetQueue().GetReservedApps()), "queue reserved apps should be 1")
+ assert.Equal(t, 1, partition.reservations, "partition reservations should be 1")
res1, err := resources.NewResourceFromConf(map[string]string{"vcore": "1"})
assert.NilError(t, err, "failed to create resource")
@@ -1825,18 +1848,24 @@ func TestRequiredNodeCancelDSReservations(t *testing.T) {
assert.NilError(t, err, "failed to add app-2 to partition")
// required node set on ask
- ask2 := newAllocationAsk("alloc-2", appID2, res1)
+ ask2 := newAllocationAsk(allocKey3, appID2, res1)
ask2.SetRequiredNode(nodeID1)
err = app1.AddAllocationAsk(ask2)
- assert.NilError(t, err, "failed to add ask alloc-2 to app-1")
+ assert.NilError(t, err, "failed to add ask alloc-3 to app-1")
+ // this is a reservation handled before we get here and we get a nil
result = partition.tryAllocate()
if result != nil {
t.Fatal("3rd allocation should not return allocation")
}
- // still reservation count is 1
- assert.Equal(t, 1, len(app.GetReservations()), "ask should have been reserved")
- assert.Equal(t, 1, len(app.GetQueue().GetReservedApps()), "queue reserved apps should be 1")
+ // check the reservation count etc for each app
+ assert.Equal(t, app.NodeReservedForAsk(allocKey2), nodeID1, "alloc-2 should be reserved on node-1")
+ assert.Equal(t, 1, len(app.GetReservations()), "only one alloc on app-1 should have been reserved")
+ assert.Equal(t, app1.NodeReservedForAsk(allocKey3), nodeID1, "alloc-3 should be reserved on node-1")
+ assert.Equal(t, 1, len(app1.GetReservations()), "alloc-3 on app-2 should have been reserved")
+ // both apps run in the same queue so just pick one app to get the queue
+ assert.Equal(t, 2, len(app.GetQueue().GetReservedApps()), "queue reserved apps should be 2")
+ assert.Equal(t, 2, partition.reservations, "partition reservations should be 2")
}
func TestRequiredNodeNotExist(t *testing.T) {
@@ -2015,15 +2044,16 @@ func TestPreemption(t *testing.T) {
// Preemption followed by a normal allocation
func TestPreemptionForRequiredNodeNormalAlloc(t *testing.T) {
setupUGM()
- // setup the partition so we can try the real allocation
+ // set up the partition so we can try the real allocation
partition, app := setupPreemptionForRequiredNode(t)
// now try the allocation again: the normal path
result := partition.tryAllocate()
- if result != nil {
- t.Fatal("allocations should not have returned a result")
+ if result == nil || result.Request == nil {
+ t.Fatal("allocation should have returned a result")
}
// check if updated (must be after allocate call)
- assert.Equal(t, 1, len(app.GetReservations()), "ask should have been reserved")
+ assert.Equal(t, 0, len(app.GetReservations()), "no allocations on app should have been reserved")
+ assert.Equal(t, 0, partition.reservations, "no allocations on partition should have been reserved")
}
// Preemption followed by a reserved allocation
@@ -2234,8 +2264,8 @@ func setupPreemptionForRequiredNode(t *testing.T) (*PartitionContext, *objects.A
t.Fatal("allocation attempt should not have returned an allocation")
}
// check if updated (must be after allocate call)
- assert.Equal(t, 1, len(app.GetReservations()), "ask should have been reserved")
- assert.Equal(t, 1, len(app.GetAskReservations(allocKey2)), "ask should have been reserved")
+ assert.Equal(t, 1, len(app.GetReservations()), "application should have been reserved")
+ assert.Equal(t, app.NodeReservedForAsk(allocKey2), nodeID1, "allocation should have been reserved on node-1")
assertUserGroupResourceMaxLimits(t, getTestUserGroup(), resources.NewResourceFromMap(map[string]resources.Quantity{"vcore": 8000}), getExpectedQueuesLimitsForPreemptionWithRequiredNode())
// try through reserved scheduling cycle this should trigger preemption
@@ -2395,7 +2425,7 @@ func TestTryAllocateReserve(t *testing.T) {
err = app.AddAllocationAsk(newAllocationAsk("alloc-1", appID1, res))
assert.NilError(t, err, "failed to add ask alloc-1 to app")
- ask := newAllocationAsk("alloc-2", appID1, res)
+ ask := newAllocationAsk(allocKey2, appID1, res)
err = app.AddAllocationAsk(ask)
assert.NilError(t, err, "failed to add ask alloc-2 to app")
node2 := partition.GetNode(nodeID2)
@@ -2403,8 +2433,8 @@ func TestTryAllocateReserve(t *testing.T) {
t.Fatal("expected node-2 to be returned got nil")
}
partition.reserve(app, node2, ask)
- if !app.IsReservedOnNode(node2.NodeID) || len(app.GetAskReservations("alloc-2")) == 0 {
- t.Fatalf("reservation failure for ask2 and node2")
+ if app.NodeReservedForAsk(allocKey2) != nodeID2 {
+ t.Fatalf("reservation failure for alloc-2 and node-2")
}
// first allocation should be app-1 and alloc-2
@@ -2416,11 +2446,11 @@ func TestTryAllocateReserve(t *testing.T) {
assert.Equal(t, result.ReservedNodeID, "", "node should not be set for allocated from reserved")
assert.Check(t, !result.Request.HasRelease(), "released allocation should not be present")
assert.Equal(t, result.Request.GetApplicationID(), appID1, "expected application app-1 to be allocated")
- assert.Equal(t, result.Request.GetAllocationKey(), "alloc-2", "expected ask alloc-2 to be allocated")
+ assert.Equal(t, result.Request.GetAllocationKey(), allocKey2, "expected ask alloc-2 to be allocated")
assertLimits(t, getTestUserGroup(), resources.NewResourceFromMap(map[string]resources.Quantity{"vcore": 1000}))
// reservations should have been removed: it is in progress
- if app.IsReservedOnNode(node2.NodeID) || len(app.GetAskReservations("alloc-2")) != 0 {
+ if app.NodeReservedForAsk(allocKey2) != "" {
t.Fatalf("reservation removal failure for ask2 and node2")
}
@@ -2462,11 +2492,11 @@ func TestTryAllocateWithReserved(t *testing.T) {
err = partition.AddApplication(app)
assert.NilError(t, err, "failed to add app-1 to partition")
- ask := newAllocationAsk("alloc-1", appID1, res)
+ ask := newAllocationAsk(allocKey, appID1, res)
err = app.AddAllocationAsk(ask)
assert.NilError(t, err, "failed to add ask alloc-1 to app")
- ask2 := newAllocationAsk("alloc-2", appID1, res)
+ ask2 := newAllocationAsk(allocKey2, appID1, res)
err = app.AddAllocationAsk(ask2)
assert.NilError(t, err, "failed to add ask alloc-2 to app")
@@ -2476,8 +2506,8 @@ func TestTryAllocateWithReserved(t *testing.T) {
t.Fatal("expected node-2 to be returned got nil")
}
partition.reserve(app, node2, ask)
- if !app.IsReservedOnNode(node2.NodeID) || len(app.GetAskReservations("alloc-1")) == 0 {
- t.Fatal("reservation failure for ask and node2")
+ if app.NodeReservedForAsk(allocKey) != nodeID2 {
+ t.Fatal("reservation failure for alloc-1 and node-2")
}
result := partition.tryAllocate()
if result == nil || result.Request == nil {
@@ -2486,7 +2516,7 @@ func TestTryAllocateWithReserved(t *testing.T) {
assert.Equal(t, objects.AllocatedReserved, result.ResultType, "expected reserved allocation to be returned")
assert.Equal(t, "", result.ReservedNodeID, "reserved node should be reset after processing")
assert.Equal(t, 0, len(node2.GetReservationKeys()), "reservation should have been removed from node")
- assert.Equal(t, false, app.IsReservedOnNode(node2.NodeID), "reservation cleanup for ask on app failed")
+ assert.Equal(t, "", app.NodeReservedForAsk(allocKey), "reservation cleanup for ask on app failed")
assertLimits(t, getTestUserGroup(), resources.NewResourceFromMap(map[string]resources.Quantity{"vcore": 5000}))
// node2 is unreserved now so the next one should allocate on the 2nd node (fair sharing)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@yunikorn.apache.org
For additional commands, e-mail: issues-help@yunikorn.apache.org