You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@yunikorn.apache.org by wi...@apache.org on 2023/08/16 00:20:56 UTC
[yunikorn-core] branch master updated: [YUNIKORN-1888] Allow allocations to be added to existing nodes (#606)
This is an automated email from the ASF dual-hosted git repository.
wilfreds 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 a4b2ebb5 [YUNIKORN-1888] Allow allocations to be added to existing nodes (#606)
a4b2ebb5 is described below
commit a4b2ebb5748ac62a8b4ce3e8e6258dc7223e8eb8
Author: Craig Condit <cc...@apache.org>
AuthorDate: Wed Aug 16 10:20:35 2023 +1000
[YUNIKORN-1888] Allow allocations to be added to existing nodes (#606)
Adds the ability to assign existing allocations dynamically to an
already created node. This allows "late" events to be processed
properly.
Closes: #606
Signed-off-by: Wilfred Spiegelenburg <wi...@apache.org>
---
go.mod | 2 +-
go.sum | 4 ++--
pkg/rmproxy/rmevent/events.go | 5 +++++
pkg/rmproxy/rmproxy.go | 26 ++++++++++++++++++++--
pkg/scheduler/context.go | 52 +++++++++++++++++++++++++++++++++++++++++++
pkg/scheduler/partition.go | 1 +
6 files changed, 85 insertions(+), 5 deletions(-)
diff --git a/go.mod b/go.mod
index 14a5158c..456ca0f7 100644
--- a/go.mod
+++ b/go.mod
@@ -22,7 +22,7 @@ module github.com/apache/yunikorn-core
go 1.20
require (
- github.com/apache/yunikorn-scheduler-interface v0.0.0-20230731151735-8931af4dda26
+ github.com/apache/yunikorn-scheduler-interface v0.0.0-20230803134744-355d6a16e4f0
github.com/google/btree v1.1.2
github.com/google/go-cmp v0.5.9
github.com/google/uuid v1.3.0
diff --git a/go.sum b/go.sum
index f4f311d8..33bfdea4 100644
--- a/go.sum
+++ b/go.sum
@@ -40,8 +40,8 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuy
github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
-github.com/apache/yunikorn-scheduler-interface v0.0.0-20230731151735-8931af4dda26 h1:78Ow8OCXd7eSKN6tnpG27sdP2rUh//vcxFHaRDYc31Y=
-github.com/apache/yunikorn-scheduler-interface v0.0.0-20230731151735-8931af4dda26/go.mod h1:/n67iTTOytyVor6wETVjleqzsp/NxCUmPslHcTvJ+Nw=
+github.com/apache/yunikorn-scheduler-interface v0.0.0-20230803134744-355d6a16e4f0 h1:DAyG6TAEbBJUYw7qEz5sysOM0lrR2d3NjHB4bHBdk8A=
+github.com/apache/yunikorn-scheduler-interface v0.0.0-20230803134744-355d6a16e4f0/go.mod h1:/n67iTTOytyVor6wETVjleqzsp/NxCUmPslHcTvJ+Nw=
github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
diff --git a/pkg/rmproxy/rmevent/events.go b/pkg/rmproxy/rmevent/events.go
index e9f31f23..39ae125e 100644
--- a/pkg/rmproxy/rmevent/events.go
+++ b/pkg/rmproxy/rmevent/events.go
@@ -86,6 +86,11 @@ type RMRejectedAllocationAskEvent struct {
RejectedAllocationAsks []*si.RejectedAllocationAsk
}
+type RMRejectedAllocationEvent struct {
+ RmID string
+ RejectedAllocations []*si.RejectedAllocation
+}
+
type RMReleaseAllocationEvent struct {
RmID string
ReleasedAllocations []*si.AllocationRelease
diff --git a/pkg/rmproxy/rmproxy.go b/pkg/rmproxy/rmproxy.go
index 4ace8548..6a7501e2 100644
--- a/pkg/rmproxy/rmproxy.go
+++ b/pkg/rmproxy/rmproxy.go
@@ -179,7 +179,7 @@ func (rmp *RMProxy) processRMReleaseAllocationAskEvent(event *rmevent.RMReleaseA
rmp.triggerUpdateAllocation(event.RmID, response)
}
-func (rmp *RMProxy) processUpdatePartitionConfigsEvent(event *rmevent.RMRejectedAllocationAskEvent) {
+func (rmp *RMProxy) processRMRejectedAllocationAskEvent(event *rmevent.RMRejectedAllocationAskEvent) {
rmp.RLock()
defer rmp.RUnlock()
if len(event.RejectedAllocationAsks) == 0 {
@@ -192,6 +192,19 @@ func (rmp *RMProxy) processUpdatePartitionConfigsEvent(event *rmevent.RMRejected
metrics.GetSchedulerMetrics().AddRejectedContainers(len(event.RejectedAllocationAsks))
}
+func (rmp *RMProxy) processRMRejectedAllocationEvent(event *rmevent.RMRejectedAllocationEvent) {
+ rmp.RLock()
+ defer rmp.RUnlock()
+ if len(event.RejectedAllocations) == 0 {
+ return
+ }
+ response := &si.AllocationResponse{
+ RejectedAllocations: event.RejectedAllocations,
+ }
+ rmp.triggerUpdateAllocation(event.RmID, response)
+ metrics.GetSchedulerMetrics().AddRejectedContainers(len(event.RejectedAllocations))
+}
+
func (rmp *RMProxy) processRMNodeUpdateEvent(event *rmevent.RMNodeUpdateEvent) {
rmp.RLock()
defer rmp.RUnlock()
@@ -224,7 +237,9 @@ func (rmp *RMProxy) handleRMEvents() {
case *rmevent.RMReleaseAllocationEvent:
rmp.processRMReleaseAllocationEvent(v)
case *rmevent.RMRejectedAllocationAskEvent:
- rmp.processUpdatePartitionConfigsEvent(v)
+ rmp.processRMRejectedAllocationAskEvent(v)
+ case *rmevent.RMRejectedAllocationEvent:
+ rmp.processRMRejectedAllocationEvent(v)
case *rmevent.RMNodeUpdateEvent:
rmp.processRMNodeUpdateEvent(v)
case *rmevent.RMReleaseAllocationAskEvent:
@@ -294,6 +309,13 @@ func (rmp *RMProxy) UpdateAllocation(request *si.AllocationRequest) error {
return fmt.Errorf("received AllocationRequest, but RmID=\"%s\" not registered", request.RmID)
}
go func() {
+ // Update allocations
+ if len(request.Allocations) > 0 {
+ for _, alloc := range request.Allocations {
+ alloc.PartitionName = common.GetNormalizedPartitionName(alloc.PartitionName, request.RmID)
+ }
+ }
+
// Update asks
if len(request.Asks) > 0 {
for _, ask := range request.Asks {
diff --git a/pkg/scheduler/context.go b/pkg/scheduler/context.go
index 5c4b3ce8..c20a3e8c 100644
--- a/pkg/scheduler/context.go
+++ b/pkg/scheduler/context.go
@@ -747,11 +747,15 @@ func (cc *ClusterContext) updateNode(nodeInfo *si.NodeInfo) {
}
// Process an ask and allocation update request.
+// - Add new allocations for the application(s).
// - Add new asks and remove released asks for the application(s).
// - Release allocations for the application(s).
// Lock free call, all updates occur on the underlying application which is locked or via events.
func (cc *ClusterContext) handleRMUpdateAllocationEvent(event *rmevent.RMUpdateAllocationEvent) {
request := event.Request
+ if len(request.Allocations) != 0 {
+ cc.processAllocations(request)
+ }
if len(request.Asks) != 0 {
cc.processAsks(request)
}
@@ -765,6 +769,54 @@ func (cc *ClusterContext) handleRMUpdateAllocationEvent(event *rmevent.RMUpdateA
}
}
+func (cc *ClusterContext) processAllocations(request *si.AllocationRequest) {
+ // Send rejected allocations back to RM
+ rejectedAllocs := make([]*si.RejectedAllocation, 0)
+
+ // Send to scheduler
+ for _, siAlloc := range request.Allocations {
+ // try to get partition
+ partition := cc.GetPartition(siAlloc.PartitionName)
+ if partition == nil {
+ msg := fmt.Sprintf("Failed to find partition %s, for application %s and allocation %s", siAlloc.PartitionName, siAlloc.ApplicationID, siAlloc.AllocationKey)
+ log.Log(log.SchedContext).Error("Invalid allocation add requested by shim, partition not found",
+ zap.String("partition", siAlloc.PartitionName),
+ zap.String("nodeID", siAlloc.NodeID),
+ zap.String("applicationID", siAlloc.ApplicationID),
+ zap.String("allocationKey", siAlloc.AllocationKey))
+ rejectedAllocs = append(rejectedAllocs, &si.RejectedAllocation{
+ AllocationKey: siAlloc.AllocationKey,
+ ApplicationID: siAlloc.ApplicationID,
+ Reason: msg,
+ })
+ continue
+ }
+
+ alloc := objects.NewAllocationFromSI(siAlloc)
+ if err := partition.addAllocation(alloc); err != nil {
+ rejectedAllocs = append(rejectedAllocs, &si.RejectedAllocation{
+ AllocationKey: siAlloc.AllocationKey,
+ ApplicationID: siAlloc.ApplicationID,
+ Reason: err.Error(),
+ })
+ log.Log(log.SchedContext).Error("Invalid allocation add requested by shim",
+ zap.String("partition", siAlloc.PartitionName),
+ zap.String("nodeID", siAlloc.NodeID),
+ zap.String("applicationID", siAlloc.ApplicationID),
+ zap.String("allocationKey", siAlloc.AllocationKey),
+ zap.Error(err))
+ }
+ }
+
+ // Reject allocs returned to RM proxy for the apps and partitions not found
+ if len(rejectedAllocs) > 0 {
+ cc.rmEventHandler.HandleEvent(&rmevent.RMRejectedAllocationEvent{
+ RmID: request.RmID,
+ RejectedAllocations: rejectedAllocs,
+ })
+ }
+}
+
func (cc *ClusterContext) processAsks(request *si.AllocationRequest) {
// Send rejects back to RM
rejectedAsks := make([]*si.RejectedAllocationAsk, 0)
diff --git a/pkg/scheduler/partition.go b/pkg/scheduler/partition.go
index f19b64cd..101200de 100644
--- a/pkg/scheduler/partition.go
+++ b/pkg/scheduler/partition.go
@@ -1167,6 +1167,7 @@ func (pc *PartitionContext) addAllocation(alloc *objects.Allocation) error {
metrics.GetQueueMetrics(queue.GetQueuePath()).IncAllocatedContainer()
node.AddAllocation(alloc)
+ alloc.SetInstanceType(node.GetInstanceType())
app.RecoverAllocationAsk(alloc.GetAsk())
app.AddAllocation(alloc)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@yunikorn.apache.org
For additional commands, e-mail: issues-help@yunikorn.apache.org