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 2026/10/08 02:58:12 UTC
(yunikorn-core) branch master updated: [YUNIKORN-3218] Handle race condition during application timeout completion (#1173)
This is an automated email from the ASF dual-hosted git repository.
wilfred-s 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 8763b6f1 [YUNIKORN-3218] Handle race condition during application timeout completion (#1173)
8763b6f1 is described below
commit 8763b6f1957827a3f00a6ebf8e1b85afddb09a2c
Author: Aditya Maheshwari <ad...@gmail.com>
AuthorDate: Thu Oct 8 13:27:25 2026 +1100
[YUNIKORN-3218] Handle race condition during application timeout completion (#1173)
Finalising an application happens in two steps. This results in a small
window when the application is still seen on the partition but no longer
connected to a queue. If an allocation is added during that window the
allocation addition will work but the allocation is never scheduled. The
pod will stay pending forever.
Close the gap by reviving the application if the queue has been removed
and the application is found. Stop the removal of the application from
the partition if the application was revived.
The full fix requires changes on the k8shim side to not prematurely
remove the application from the cache.
Track revived applications via the introduction of a new metric.
Closes: #1173
Signed-off-by: Wilfred Spiegelenburg <wi...@apache.org>
---
pkg/metrics/queue.go | 28 +-
pkg/metrics/queue_test.go | 22 ++
pkg/metrics/scheduler.go | 25 ++
pkg/metrics/scheduler_test.go | 26 ++
pkg/scheduler/health_checker_test.go | 5 +-
pkg/scheduler/objects/application.go | 115 ++++++-
pkg/scheduler/objects/application_property_test.go | 30 +-
pkg/scheduler/objects/application_revive_test.go | 284 +++++++++++++++++
pkg/scheduler/objects/application_state.go | 25 ++
pkg/scheduler/objects/application_test.go | 193 ++++++++----
pkg/scheduler/objects/node_collection_test.go | 3 +-
pkg/scheduler/objects/preemption_test.go | 185 +++++++----
pkg/scheduler/objects/preemption_utilities_test.go | 5 +-
pkg/scheduler/objects/queue_test.go | 18 +-
pkg/scheduler/objects/quota_preemptor_test.go | 8 +-
pkg/scheduler/partition.go | 187 ++++++++++-
pkg/scheduler/partition_revive_test.go | 342 +++++++++++++++++++++
pkg/scheduler/partition_test.go | 3 +-
pkg/webservice/handlers_test.go | 40 +--
19 files changed, 1354 insertions(+), 190 deletions(-)
diff --git a/pkg/metrics/queue.go b/pkg/metrics/queue.go
index 7506ad80..5c32a1cd 100644
--- a/pkg/metrics/queue.go
+++ b/pkg/metrics/queue.go
@@ -41,6 +41,7 @@ const (
AppCompleting = "completing"
AppCompleted = "completed"
AppExpired = "expired"
+ AppRevived = "revived"
ContainerReleased = "released"
ContainerAllocated = "allocated"
@@ -75,7 +76,7 @@ func InitQueueMetrics(name string) *QueueMetrics {
Namespace: Namespace,
Name: "queue_app",
ConstLabels: prometheus.Labels{"queue": name},
- Help: "Queue application metrics. State of the application includes `new`, `accepted`, `rejected`, `running`, `failing`, `failed`, `resuming`, `completing`, `completed`.",
+ Help: "Queue application metrics. State of the application includes `new`, `accepted`, `rejected`, `running`, `failing`, `failed`, `resuming`, `completing`, `completed`, `revived`.",
}, []string{"state"})
q.containerMetrics = prometheus.NewCounterVec(
@@ -306,6 +307,12 @@ func (m *QueueMetrics) IncQueueApplicationsCompleted() {
m.incQueueApplications(AppCompleted)
}
+// DecQueueApplicationsCompleted reverses IncQueueApplicationsCompleted when a completed application
+// is revived.
+func (m *QueueMetrics) DecQueueApplicationsCompleted() {
+ m.decQueueApplications(AppCompleted)
+}
+
func (m *QueueMetrics) GetQueueApplicationsCompleted() (int, error) {
metricDto := &dto.Metric{}
err := m.appMetrics.WithLabelValues(AppCompleted).Write(metricDto)
@@ -315,6 +322,25 @@ func (m *QueueMetrics) GetQueueApplicationsCompleted() (int, error) {
return -1, err
}
+// IncQueueApplicationsRevived counts an application brought back out of the Completed state.
+func (m *QueueMetrics) IncQueueApplicationsRevived() {
+ m.incQueueApplications(AppRevived)
+}
+
+// DecQueueApplicationsRevived is called when a revived application completes again.
+func (m *QueueMetrics) DecQueueApplicationsRevived() {
+ m.decQueueApplications(AppRevived)
+}
+
+func (m *QueueMetrics) GetQueueApplicationsRevived() (int, error) {
+ metricDto := &dto.Metric{}
+ err := m.appMetrics.WithLabelValues(AppRevived).Write(metricDto)
+ if err == nil {
+ return int(*metricDto.Gauge.Value), nil
+ }
+ return -1, err
+}
+
func (m *QueueMetrics) IncAllocatedContainer() {
m.containerMetrics.WithLabelValues(ContainerAllocated).Inc()
}
diff --git a/pkg/metrics/queue_test.go b/pkg/metrics/queue_test.go
index 30fb7834..84cc46b8 100644
--- a/pkg/metrics/queue_test.go
+++ b/pkg/metrics/queue_test.go
@@ -168,6 +168,28 @@ func TestApplicationsCompleted(t *testing.T) {
curr, err := qm.GetQueueApplicationsCompleted()
assert.NilError(t, err)
assert.Equal(t, 1, curr)
+
+ qm.DecQueueApplicationsCompleted()
+ curr, err = qm.GetQueueApplicationsCompleted()
+ assert.NilError(t, err)
+ assert.Equal(t, 0, curr)
+}
+
+func TestApplicationsRevived(t *testing.T) {
+ qm = getQueueMetrics()
+ defer unregisterQueueMetrics()
+
+ qm.IncQueueApplicationsRevived()
+ verifyAppMetrics(t, "revived")
+
+ curr, err := qm.GetQueueApplicationsRevived()
+ assert.NilError(t, err)
+ assert.Equal(t, 1, curr)
+
+ qm.DecQueueApplicationsRevived()
+ curr, err = qm.GetQueueApplicationsRevived()
+ assert.NilError(t, err)
+ assert.Equal(t, 0, curr)
}
func TestAllocatedContainers(t *testing.T) {
diff --git a/pkg/metrics/scheduler.go b/pkg/metrics/scheduler.go
index abf48644..ac9de124 100644
--- a/pkg/metrics/scheduler.go
+++ b/pkg/metrics/scheduler.go
@@ -371,6 +371,12 @@ func (m *SchedulerMetrics) IncTotalApplicationsCompleted() {
m.application.WithLabelValues(AppCompleted).Inc()
}
+// DecTotalApplicationsCompleted reverses IncTotalApplicationsCompleted when a completed
+// application is revived.
+func (m *SchedulerMetrics) DecTotalApplicationsCompleted() {
+ m.application.WithLabelValues(AppCompleted).Dec()
+}
+
func (m *SchedulerMetrics) GetTotalApplicationsCompleted() (int, error) {
metricDto := &dto.Metric{}
err := m.application.WithLabelValues(AppCompleted).Write(metricDto)
@@ -380,6 +386,25 @@ func (m *SchedulerMetrics) GetTotalApplicationsCompleted() (int, error) {
return -1, err
}
+// IncTotalApplicationsRevived counts an application brought back out of the Completed state.
+func (m *SchedulerMetrics) IncTotalApplicationsRevived() {
+ m.application.WithLabelValues(AppRevived).Inc()
+}
+
+// DecTotalApplicationsRevived is called when a revived application completes again.
+func (m *SchedulerMetrics) DecTotalApplicationsRevived() {
+ m.application.WithLabelValues(AppRevived).Dec()
+}
+
+func (m *SchedulerMetrics) GetTotalApplicationsRevived() (int, error) {
+ metricDto := &dto.Metric{}
+ err := m.application.WithLabelValues(AppRevived).Write(metricDto)
+ if err == nil {
+ return int(*metricDto.Gauge.Value), nil
+ }
+ return -1, err
+}
+
func (m *SchedulerMetrics) IncActiveNodes() {
m.node.WithLabelValues(NodeActive).Inc()
}
diff --git a/pkg/metrics/scheduler_test.go b/pkg/metrics/scheduler_test.go
index c290a78b..0dde612c 100644
--- a/pkg/metrics/scheduler_test.go
+++ b/pkg/metrics/scheduler_test.go
@@ -158,6 +158,32 @@ func TestSchedulerApplicationsCompleted(t *testing.T) {
curr, err := sm.GetTotalApplicationsCompleted()
assert.NilError(t, err)
assert.Equal(t, curr, 1)
+
+ sm.DecTotalApplicationsCompleted()
+ verifyMetric(t, 0, "completed", "yunikorn_scheduler_application_total", dto.MetricType_GAUGE, "state")
+
+ curr, err = sm.GetTotalApplicationsCompleted()
+ assert.NilError(t, err)
+ assert.Equal(t, curr, 0)
+}
+
+func TestSchedulerApplicationsRevived(t *testing.T) {
+ sm = getSchedulerMetrics(t)
+ defer unregisterMetrics()
+
+ sm.IncTotalApplicationsRevived()
+ verifyMetric(t, 1, "revived", "yunikorn_scheduler_application_total", dto.MetricType_GAUGE, "state")
+
+ curr, err := sm.GetTotalApplicationsRevived()
+ assert.NilError(t, err)
+ assert.Equal(t, curr, 1)
+
+ sm.DecTotalApplicationsRevived()
+ verifyMetric(t, 0, "revived", "yunikorn_scheduler_application_total", dto.MetricType_GAUGE, "state")
+
+ curr, err = sm.GetTotalApplicationsRevived()
+ assert.NilError(t, err)
+ assert.Equal(t, curr, 0)
}
func TestSchedulerApplicationsFailed(t *testing.T) {
diff --git a/pkg/scheduler/health_checker_test.go b/pkg/scheduler/health_checker_test.go
index e4017828..dd9d0432 100644
--- a/pkg/scheduler/health_checker_test.go
+++ b/pkg/scheduler/health_checker_test.go
@@ -41,7 +41,7 @@ partitions:
queues:
- name: default
resources:
- guaranteed:
+ guaranteed:
memory: 100000
vcore: 10000
max:
@@ -214,7 +214,8 @@ func TestGetSchedulerHealthStatusContext(t *testing.T) {
// add the allocation to the app as well
part := schedulerContext.partitions[partName]
app := newApplication("appID", partName, "root.default")
- app.AddAllocation(alloc)
+ err = app.AddAllocation(alloc)
+ assert.NilError(t, err)
err = part.AddApplication(app)
assert.NilError(t, err, "Could not add application")
healthInfo = GetSchedulerHealthStatus(schedulerMetrics, schedulerContext)
diff --git a/pkg/scheduler/objects/application.go b/pkg/scheduler/objects/application.go
index 7071ed97..4b0c1a94 100644
--- a/pkg/scheduler/objects/application.go
+++ b/pkg/scheduler/objects/application.go
@@ -25,6 +25,7 @@ import (
"strconv"
"strings"
"sync"
+ "sync/atomic"
"time"
"github.com/looplab/fsm"
@@ -121,6 +122,7 @@ type Application struct {
runnableInQueue bool // whether the application is runnable/schedulable in the queue. Default is true.
runnableByUserLimit bool // whether the application is runnable/schedulable based on user/group quota. Default is true.
backoffDeadline time.Time // no scheduling from this application until this deadline
+ revived atomic.Bool // brought back out of Completed and not completed again since (read lock free)
rmEventHandler handler.EventHandler
rmID string
@@ -259,6 +261,62 @@ func (sa *Application) IsResuming() bool {
return sa.stateMachine.Is(Resuming.String())
}
+// IsRevived returns true if the application was brought back out of Completed and has not completed
+// again since. The flag is read lock free so it can be consulted while the partition lock is held.
+func (sa *Application) IsRevived() bool {
+ return sa.revived.Load()
+}
+
+// IsTerminated returns true if the application is in a state it can only leave by being revived.
+func (sa *Application) IsTerminated() bool {
+ return sa.IsCompleted() || sa.IsFailed() || sa.IsRejected() || sa.IsExpired()
+}
+
+// ensureRunnable puts the application back into the running lifecycle so an ask or allocation can be
+// added to it. The application can finish between the partition looking it up and the add landing
+// here, so the check and the state change have to happen together under the application lock.
+// Progressing a New application is left to the callers that already do it.
+// Call with sa.Lock() held.
+func (sa *Application) ensureRunnable() error {
+ switch {
+ case sa.IsCompleting(), sa.IsCompleted():
+ // leave_state cancels the pending completion or expiry timer on the way out
+ if err := sa.HandleApplicationEvent(RunApplication); err != nil {
+ return fmt.Errorf("failed to revive application %s from state %s: %w", sa.ApplicationID, sa.CurrentState(), err)
+ }
+ case sa.IsFailing(), sa.IsFailed(), sa.IsRejected(), sa.IsExpired():
+ return fmt.Errorf("application %s is in terminal state %s and cannot accept new work", sa.ApplicationID, sa.CurrentState())
+ }
+ return nil
+}
+
+// restoreAfterCompletion rebuilds the state that enter_Completed and LogAppSummary tore down, so a
+// revived application can track asks and resources again. Call with sa.Lock() held.
+func (sa *Application) restoreAfterCompletion() {
+ // AggregateTrackedResource does not guard a nil receiver, and cleanupTrackedResource nils these
+ if sa.usedResource == nil {
+ sa.usedResource = resources.NewTrackedResource()
+ }
+ if sa.preemptedResource == nil {
+ sa.preemptedResource = resources.NewTrackedResource()
+ }
+ if sa.placeholderResource == nil {
+ sa.placeholderResource = resources.NewTrackedResource()
+ }
+ if sa.requests == nil {
+ sa.requests = make(map[string]*Allocation)
+ }
+ if sa.pendingPriorities == nil {
+ sa.pendingPriorities = make(map[int32]int)
+ }
+ if sa.allocations == nil {
+ sa.allocations = make(map[string]*Allocation)
+ }
+ // cleanupAsks nils the slice; the requests map it mirrors is empty at this point
+ sa.sortedRequests = sortedRequests{}
+ sa.finishedTime = time.Time{}
+}
+
// HandleApplicationEvent handles the state event for the application.
// The application lock is expected to be held.
func (sa *Application) HandleApplicationEvent(event applicationEvent) error {
@@ -678,16 +736,17 @@ func (sa *Application) AddAllocationAsk(ask *Allocation) error {
if _, tracked := sa.requests[ask.GetAllocationKey()]; tracked {
return fmt.Errorf("ask %s is already tracked on app %s", ask.GetAllocationKey(), sa.ApplicationID)
}
+ // must run before any accounting: a terminated app has no queue to charge the ask to
+ if err := sa.ensureRunnable(); err != nil {
+ return err
+ }
if ask.createTime.Before(sa.submissionTime) {
sa.submissionTime = ask.createTime
}
delta := ask.GetAllocatedResource().Clone()
- // Check if we need to change state based on the ask added, there are two cases:
- // 1) first ask added on a new app: state is New
- // 2) all asks and allocation have been removed: state is Completing
- // Move the state and get it scheduling (again)
- if sa.stateMachine.Is(New.String()) || sa.stateMachine.Is(Completing.String()) {
+ // first ask added on a new app: state is New. Move the state and get it scheduling.
+ if sa.IsNew() {
if err := sa.HandleApplicationEvent(RunApplication); err != nil {
log.Log(log.SchedApplication).Debug("Application state change failed while adding new ask",
zap.String("currentState", sa.CurrentState()),
@@ -776,11 +835,16 @@ func (sa *Application) UpdateAllocationResources(alloc *Allocation, isQuotaPreem
// Add the ask when a node allocation is recovered.
// Safeguarded against a nil but the recovery generates the ask and should never be nil.
-func (sa *Application) RecoverAllocationAsk(alloc *Allocation) {
+// An application in a terminal state refuses the ask and returns an error.
+func (sa *Application) RecoverAllocationAsk(alloc *Allocation) error {
sa.Lock()
defer sa.Unlock()
if alloc == nil {
- return
+ return nil
+ }
+
+ if err := sa.ensureRunnable(); err != nil {
+ return err
}
sa.addAllocationAskInternal(alloc)
@@ -792,6 +856,7 @@ func (sa *Application) RecoverAllocationAsk(alloc *Allocation) {
zap.Error(err))
}
}
+ return nil
}
// addToPriorities records that an ask at priority p just became pending (added, or
@@ -1962,6 +2027,35 @@ func (sa *Application) UnSetQueue() {
sa.finishedTime = time.Now()
}
+// RestoreQueue re-attaches a queue to an application being revived out of Completed. Unlike SetQueue
+// it does not count the application as newly submitted. The revived flag is raised here, together
+// with the queue, so a concurrent moveTerminatedApp cannot tear the application down in the window
+// between restoring the queue and the state machine leaving Completed.
+func (sa *Application) RestoreQueue(queue *Queue) {
+ sa.Lock()
+ defer sa.Unlock()
+ sa.queue = queue
+ sa.finishedTime = time.Time{}
+ sa.revived.Store(true)
+}
+
+// DetachQueueForCompletion clears the queue link for an application that is being moved to the
+// completed list. It returns the detached queue and true only while the application is still
+// terminated and has not been revived; otherwise it leaves the application untouched and returns
+// nil, false. The caller removes the application from the returned queue outside the application
+// lock to keep the queue-before-application lock order.
+func (sa *Application) DetachQueueForCompletion() (*Queue, bool) {
+ sa.Lock()
+ defer sa.Unlock()
+ if sa.revived.Load() || !sa.IsTerminated() {
+ return nil, false
+ }
+ queue := sa.queue
+ sa.queue = nil
+ sa.finishedTime = time.Now()
+ return queue, true
+}
+
func (sa *Application) StartTime() time.Time {
sa.RLock()
defer sa.RUnlock()
@@ -2017,10 +2111,15 @@ func (sa *Application) getAllRequestsInternal() []*Allocation {
}
// Add a new Allocation to the application
-func (sa *Application) AddAllocation(alloc *Allocation) {
+// An application in a terminal state refuses the allocation and returns an error.
+func (sa *Application) AddAllocation(alloc *Allocation) error {
sa.Lock()
defer sa.Unlock()
+ if err := sa.ensureRunnable(); err != nil {
+ return err
+ }
sa.addAllocationInternal(Allocated, alloc)
+ return nil
}
// Add the Allocation to the application
diff --git a/pkg/scheduler/objects/application_property_test.go b/pkg/scheduler/objects/application_property_test.go
index 8e58c407..dd216c1d 100644
--- a/pkg/scheduler/objects/application_property_test.go
+++ b/pkg/scheduler/objects/application_property_test.go
@@ -215,13 +215,17 @@ func runPropertyFuzz(t *testing.T, seed int64) int { //nolint:funlen
nextCreationTime++
alloc := newFuzzAsk(key, appID, "", res, false, priority, "recovered-node", nextCreationTime)
assert.Assert(t, alloc != nil, "NewAllocationFromSI unexpectedly returned nil")
- app.RecoverAllocationAsk(alloc)
- // a recovered ask is already allocated (NodeID set => allocated=true in
- // NewAllocationFromSI) so addAllocationAskInternal must NOT add it to the pending
- // histogram: record it only as allocated, never as pending. This is the exact invariant
- // this fuzz operation is probing.
- keyPriority[key] = priority
- allocatedKeys[key] = true
+ // RecoverAllocationAsk refuses an app in a terminal state (Failing/Failed/Rejected/
+ // Expired) rather than stranding the ask on it, and case 5 below deliberately drives
+ // the app to Failed, so the model may only record the key when the product took it.
+ if recoverErr := app.RecoverAllocationAsk(alloc); recoverErr == nil {
+ // a recovered ask is already allocated (NodeID set => allocated=true in
+ // NewAllocationFromSI) so addAllocationAskInternal must NOT add it to the pending
+ // histogram: record it only as allocated, never as pending. This is the exact invariant
+ // this fuzz operation is probing.
+ keyPriority[key] = priority
+ allocatedKeys[key] = true
+ }
case 5: // occasionally clear-all or fail-and-cleanup, otherwise another RemoveAllocationAsk-ish no-op
switch rng.Intn(50) {
@@ -304,11 +308,13 @@ func runPropertyFuzz(t *testing.T, seed int64) int { //nolint:funlen
}); ok {
ask := app.GetAllocationAsk(key)
assert.Assert(t, ask != nil, "seed=%d step=%d: allocated key %s missing from sa.requests", seed, i, key)
- app.AddAllocation(ask)
- // AddAllocation touches sa.allocations/allocatedResource and the app state only: it
- // leaves sa.requests, the pending histogram and sortedRequests alone, so the pending
- // and allocated reference sets are unchanged by design.
- confirmedKeys[key] = true
+ // AddAllocation refuses a terminal app for the same reason RecoverAllocationAsk does
+ if addErr := app.AddAllocation(ask); addErr == nil {
+ // AddAllocation touches sa.allocations/allocatedResource and the app state only: it
+ // leaves sa.requests, the pending histogram and sortedRequests alone, so the pending
+ // and allocated reference sets are unchanged by design.
+ confirmedKeys[key] = true
+ }
}
case 7: // RollbackAllocation - revert a confirmed allocation back to a pending ask
diff --git a/pkg/scheduler/objects/application_revive_test.go b/pkg/scheduler/objects/application_revive_test.go
new file mode 100644
index 00000000..f9055127
--- /dev/null
+++ b/pkg/scheduler/objects/application_revive_test.go
@@ -0,0 +1,284 @@
+/*
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+package objects
+
+import (
+ "testing"
+
+ "gotest.tools/v3/assert"
+
+ "github.com/apache/yunikorn-core/pkg/common/resources"
+ "github.com/apache/yunikorn-core/pkg/metrics"
+)
+
+// completedApp returns an application that has genuinely run through the state machine into
+// Completed, so the enter_Completed teardown (cleanupAsks, timers, metrics) has really happened.
+// A non-empty rmID also runs LogAppSummary, which is what nils the tracked resources.
+func completedApp(t *testing.T, rmID string) *Application {
+ t.Helper()
+ app := newApplication(appID1, "default", "root.a")
+ queue, err := createRootQueue(nil)
+ assert.NilError(t, err, "queue create failed")
+ app.queue = queue
+
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 10})
+ assert.NilError(t, app.AddAllocationAsk(newAllocationAsk("seed-ask", appID1, res)), "seed ask add failed")
+ assert.Equal(t, app.CurrentState(), Accepted.String())
+
+ // dropping the only ask leaves no pending and no allocations, which moves the app to Completing
+ app.RemoveAllocationAsk("seed-ask")
+ assert.Equal(t, app.CurrentState(), Completing.String())
+
+ app.Lock()
+ err = app.HandleApplicationEvent(CompleteApplication)
+ app.Unlock()
+ assert.NilError(t, err, "move to Completed failed")
+ assert.Equal(t, app.CurrentState(), Completed.String())
+
+ if rmID != "" {
+ app.LogAppSummary(rmID)
+ }
+ return app
+}
+
+func TestAddAllocationAskRevivesCompletedApp(t *testing.T) {
+ setupUGM()
+ app := completedApp(t, "rm-revive")
+ assert.Assert(t, !app.IsRevived(), "app should not be flagged revived while completed")
+
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5})
+ assert.NilError(t, app.AddAllocationAsk(newAllocationAsk("late-ask", appID1, res)), "late ask should revive the app")
+
+ assert.Equal(t, app.CurrentState(), Running.String(), "completed app should be revived to Running")
+ assert.Assert(t, app.IsRevived(), "revived flag should be set")
+ assert.Assert(t, app.GetAllocationAsk("late-ask") != nil, "ask should be tracked after revival")
+ assert.Assert(t, resources.Equals(app.GetPendingResource(), res), "pending should hold the late ask")
+ // the terminatedTimeout -> ExpireApplication timer must not survive the revival
+ app.RLock()
+ timer := app.stateTimer
+ app.RUnlock()
+ assert.Assert(t, timer == nil, "expiry timer should be cleared on revival")
+}
+
+func TestRecoverAllocationAskRevivesCompletedApp(t *testing.T) {
+ setupUGM()
+ app := completedApp(t, "rm-revive")
+
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5})
+ alloc := newAllocation(appID1, nodeID1, res)
+ assert.NilError(t, app.RecoverAllocationAsk(alloc), "recover should revive the app")
+
+ assert.Equal(t, app.CurrentState(), Running.String())
+ assert.Assert(t, app.GetAllocationAsk(alloc.GetAllocationKey()) != nil, "recovered ask should be tracked")
+}
+
+func TestAddAllocationRevivesCompletedApp(t *testing.T) {
+ setupUGM()
+ app := completedApp(t, "rm-revive")
+
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5})
+ assert.NilError(t, app.AddAllocation(newAllocation(appID1, nodeID1, res)), "add allocation should revive the app")
+
+ assert.Equal(t, app.CurrentState(), Running.String())
+ assert.Assert(t, resources.Equals(app.GetAllocatedResource(), res), "allocated resource should be tracked")
+}
+
+// A revived app has to aggregate tracked resources again: LogAppSummary nils them on completion and
+// AggregateTrackedResource does not guard a nil receiver.
+func TestRevivedAppTracksCompletedResourceWithoutPanic(t *testing.T) {
+ setupUGM()
+ app := completedApp(t, "rm-revive")
+ app.RLock()
+ nilled := app.usedResource == nil
+ app.RUnlock()
+ assert.Assert(t, nilled, "LogAppSummary should have nilled the tracker")
+
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5})
+ alloc := newAllocation(appID1, nodeID1, res)
+ assert.NilError(t, app.AddAllocation(alloc))
+
+ app.RLock()
+ restored := app.usedResource != nil && app.preemptedResource != nil && app.placeholderResource != nil
+ app.RUnlock()
+ assert.Assert(t, restored, "tracked resources should be rebuilt on revival")
+
+ // would panic on a nil TrackedResource without the revival restore
+ assert.Assert(t, app.RemoveAllocation(alloc.GetAllocationKey(), 0) != nil, "allocation should be removed")
+}
+
+func TestRestoreQueueClearsFinishedTime(t *testing.T) {
+ setupUGM()
+ app := completedApp(t, "")
+ app.UnSetQueue()
+ assert.Assert(t, !app.FinishedTime().IsZero(), "finished time should be set after completion")
+
+ queue, err := createRootQueue(nil)
+ assert.NilError(t, err, "queue create failed")
+ app.RestoreQueue(queue)
+ assert.Assert(t, app.FinishedTime().IsZero(), "finished time should be cleared when the queue is restored")
+ assert.Assert(t, app.GetQueue() != nil, "queue should be re-attached")
+}
+
+func TestAddPathsRejectTerminalApps(t *testing.T) {
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5})
+
+ for _, state := range []applicationState{Failing, Failed, Rejected, Expired} {
+ t.Run(state.String(), func(t *testing.T) {
+ setupUGM()
+ app := newApplication(appID1, "default", "root.a")
+ queue, err := createRootQueue(nil)
+ assert.NilError(t, err, "queue create failed")
+ app.queue = queue
+ app.SetState(state.String())
+
+ assert.ErrorContains(t, app.AddAllocationAsk(newAllocationAsk("ask", appID1, res)), "cannot accept new work")
+ assert.Assert(t, app.GetAllocationAsk("ask") == nil, "ask must not be tracked on a terminal app")
+
+ alloc := newAllocation(appID1, nodeID1, res)
+ assert.ErrorContains(t, app.RecoverAllocationAsk(alloc), "cannot accept new work")
+ assert.ErrorContains(t, app.AddAllocation(alloc), "cannot accept new work")
+ assert.Assert(t, resources.IsZero(app.GetAllocatedResource()), "nothing should be allocated")
+ assert.Equal(t, app.CurrentState(), state.String(), "state must be unchanged")
+ })
+ }
+}
+
+// Completing is the pre-existing grace period: it must still revive, but it is not a completed-app
+// revival and must not be counted as one.
+func TestCompletingStillRevivesWithoutReviveFlag(t *testing.T) {
+ setupUGM()
+ app := newApplication(appID1, "default", "root.a")
+ queue, err := createRootQueue(nil)
+ assert.NilError(t, err, "queue create failed")
+ app.queue = queue
+
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 10})
+ assert.NilError(t, app.AddAllocationAsk(newAllocationAsk("ask-1", appID1, res)))
+ app.RemoveAllocationAsk("ask-1")
+ assert.Equal(t, app.CurrentState(), Completing.String())
+
+ assert.NilError(t, app.AddAllocationAsk(newAllocationAsk("ask-2", appID1, res)))
+ assert.Equal(t, app.CurrentState(), Running.String())
+ assert.Assert(t, !app.IsRevived(), "Completing is not a completed-app revival")
+}
+
+// The completed gauges come back down on revival and the revived gauge unwinds when the application
+// completes again. The monotonic queue_app_total counter is deliberately left alone.
+func TestReviveMetricsBalance(t *testing.T) {
+ setupUGM()
+ metrics.Reset()
+ app := completedApp(t, "")
+
+ completed, err := metrics.GetSchedulerMetrics().GetTotalApplicationsCompleted()
+ assert.NilError(t, err)
+ assert.Equal(t, completed, 1, "completed gauge should count the finished app")
+
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5})
+ assert.NilError(t, app.AddAllocationAsk(newAllocationAsk("late-ask", appID1, res)))
+
+ completed, err = metrics.GetSchedulerMetrics().GetTotalApplicationsCompleted()
+ assert.NilError(t, err)
+ assert.Equal(t, completed, 0, "completed gauge should be given back on revival")
+ revived, err := metrics.GetSchedulerMetrics().GetTotalApplicationsRevived()
+ assert.NilError(t, err)
+ assert.Equal(t, revived, 1, "revived gauge should count the revival")
+
+ // complete the app a second time: the revived gauge unwinds, completed goes back up
+ app.RemoveAllocationAsk("late-ask")
+ app.Lock()
+ err = app.HandleApplicationEvent(CompleteApplication)
+ app.Unlock()
+ assert.NilError(t, err)
+ assert.Equal(t, app.CurrentState(), Completed.String())
+
+ revived, err = metrics.GetSchedulerMetrics().GetTotalApplicationsRevived()
+ assert.NilError(t, err)
+ assert.Equal(t, revived, 0, "revived gauge should unwind when the app completes again")
+ completed, err = metrics.GetSchedulerMetrics().GetTotalApplicationsCompleted()
+ assert.NilError(t, err)
+ assert.Equal(t, completed, 1, "completed gauge should count the app again")
+ assert.Assert(t, !app.IsRevived(), "revived flag should be cleared on re-completion")
+}
+
+// Expiry is still terminal: leaving Completed for Expired must not hand the completed gauge back.
+func TestExpireDoesNotCountAsRevival(t *testing.T) {
+ setupUGM()
+ metrics.Reset()
+ app := completedApp(t, "")
+
+ app.Lock()
+ err := app.HandleApplicationEvent(ExpireApplication)
+ app.Unlock()
+ assert.NilError(t, err)
+ assert.Equal(t, app.CurrentState(), Expired.String())
+
+ completed, err := metrics.GetSchedulerMetrics().GetTotalApplicationsCompleted()
+ assert.NilError(t, err)
+ assert.Equal(t, completed, 1, "expiry must not return the completed gauge")
+ revived, err := metrics.GetSchedulerMetrics().GetTotalApplicationsRevived()
+ assert.NilError(t, err)
+ assert.Equal(t, revived, 0, "expiry is not a revival")
+}
+
+// A genuinely completed application that has not been revived hands back its queue so the partition
+// can file it on the completed list.
+func TestDetachQueueForCompletion(t *testing.T) {
+ setupUGM()
+ app := completedApp(t, "rm-detach")
+ assert.Assert(t, app.GetQueue() != nil, "completed app should still hold its queue")
+
+ queue, ok := app.DetachQueueForCompletion()
+ assert.Assert(t, ok, "a completed, non-revived app must be detachable")
+ assert.Assert(t, queue != nil, "the detached queue should be returned")
+ assert.Assert(t, app.GetQueue() == nil, "the queue link should be cleared")
+ assert.Assert(t, !app.FinishedTime().IsZero(), "the finished time should be stamped")
+}
+
+// RestoreQueue raises the revived flag together with the queue, so DetachQueueForCompletion must
+// refuse to tear the application down even while it still shows a terminal state.
+func TestDetachQueueForCompletionSkipsRevived(t *testing.T) {
+ setupUGM()
+ app := completedApp(t, "rm-detach")
+ queue := app.GetQueue()
+ app.RestoreQueue(queue)
+ assert.Assert(t, app.IsRevived(), "RestoreQueue must raise the revived flag")
+
+ detached, ok := app.DetachQueueForCompletion()
+ assert.Assert(t, !ok, "a revived app must not be detached")
+ assert.Assert(t, detached == nil, "no queue should be returned for a revived app")
+ assert.Assert(t, app.GetQueue() != nil, "the revived app must keep its queue")
+}
+
+// A non-terminated (still running) application must never be detached by the completion cleanup.
+func TestDetachQueueForCompletionSkipsRunning(t *testing.T) {
+ setupUGM()
+ app := newApplication(appID1, "default", "root.a")
+ queue, err := createRootQueue(nil)
+ assert.NilError(t, err, "queue create failed")
+ app.queue = queue
+
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5})
+ assert.NilError(t, app.AddAllocationAsk(newAllocationAsk("ask", appID1, res)))
+ assert.Assert(t, !app.IsTerminated(), "app must be in a non-terminal state")
+
+ detached, ok := app.DetachQueueForCompletion()
+ assert.Assert(t, !ok, "a non-terminated app must not be detached")
+ assert.Assert(t, detached == nil, "no queue should be returned for an active app")
+ assert.Assert(t, app.GetQueue() != nil, "the active app must keep its queue")
+}
diff --git a/pkg/scheduler/objects/application_state.go b/pkg/scheduler/objects/application_state.go
index db85f908..708a4f3e 100644
--- a/pkg/scheduler/objects/application_state.go
+++ b/pkg/scheduler/objects/application_state.go
@@ -105,6 +105,11 @@ func eventDesc() fsm.Events {
Name: RunApplication.String(),
Src: []string{Accepted.String(), Running.String(), Completing.String()},
Dst: Running.String(),
+ }, {
+ // revival: the shim sent new work for an application the core had already completed
+ Name: RunApplication.String(),
+ Src: []string{Completed.String()},
+ Dst: Running.String(),
}, {
Name: CompleteApplication.String(),
Src: []string{Accepted.String(), Running.String()},
@@ -253,11 +258,31 @@ func callbacks() fsm.Callbacks {
qm := metrics.GetQueueMetrics(app.queuePath)
qm.IncQueueApplicationsCompleted()
qm.IncQueueApplicationsCompletedTotal()
+ if app.revived.Load() {
+ app.revived.Store(false)
+ qm.DecQueueApplicationsRevived()
+ metrics.GetSchedulerMetrics().DecTotalApplicationsRevived()
+ }
app.setStateTimer(terminatedTimeout, app.stateMachine.Current(), ExpireApplication)
app.executeTerminatedCallback()
app.clearPlaceholderTimer()
app.cleanupAsks()
},
+ fmt.Sprintf("leave_%s", Completed.String()): func(_ context.Context, event *fsm.Event) {
+ app := event.Args[0].(*Application) //nolint:errcheck
+ // Expired is still a terminal end state: only a revival back into the running
+ // lifecycle hands the completed gauges back.
+ if event.Dst == Expired.String() {
+ return
+ }
+ qm := metrics.GetQueueMetrics(app.queuePath)
+ qm.DecQueueApplicationsCompleted()
+ metrics.GetSchedulerMetrics().DecTotalApplicationsCompleted()
+ app.revived.Store(true)
+ qm.IncQueueApplicationsRevived()
+ metrics.GetSchedulerMetrics().IncTotalApplicationsRevived()
+ app.restoreAfterCompletion()
+ },
fmt.Sprintf("enter_%s", Failed.String()): func(_ context.Context, event *fsm.Event) {
app := event.Args[0].(*Application) //nolint:errcheck
metrics.GetSchedulerMetrics().IncTotalApplicationsFailed()
diff --git a/pkg/scheduler/objects/application_test.go b/pkg/scheduler/objects/application_test.go
index a19012d0..42dc2801 100644
--- a/pkg/scheduler/objects/application_test.go
+++ b/pkg/scheduler/objects/application_test.go
@@ -577,25 +577,29 @@ func TestRecoverAllocAsk(t *testing.T) {
app.queue = queue
// failure cases
- app.RecoverAllocationAsk(nil)
+ err1 := app.RecoverAllocationAsk(nil)
+ assert.NilError(t, err1)
assert.Equal(t, len(app.requests), 0, "nil ask should not be added")
res := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})
ask := newAllocationAsk(aKey, appID1, res)
- app.RecoverAllocationAsk(ask)
+ err2 := app.RecoverAllocationAsk(ask)
+ assert.NilError(t, err2)
assert.Equal(t, len(app.requests), 1, "ask should have been added")
assert.Assert(t, app.IsAccepted(), "Application should be in accepted state")
assertUserGroupResource(t, getTestUserGroup(), nil)
ask = newAllocationAsk("ask-2", appID1, res)
- app.RecoverAllocationAsk(ask)
+ err3 := app.RecoverAllocationAsk(ask)
+ assert.NilError(t, err3)
assert.Equal(t, len(app.requests), 2, "ask should have been added, total should be 2")
assert.Assert(t, app.IsAccepted(), "Application should have stayed in accepted state")
assertUserGroupResource(t, getTestUserGroup(), nil)
assert.Equal(t, 0, len(app.placeholderData))
ask = newAllocationAskTG("ask-3", appID1, "testGroup", res)
- app.RecoverAllocationAsk(ask)
+ err4 := app.RecoverAllocationAsk(ask)
+ assert.NilError(t, err4)
assertPlaceholderData(t, app, "testGroup", 1, 0, 0, res)
}
@@ -769,7 +773,8 @@ func TestRemovePlaceholderAllocationWithNoRealAllocation(t *testing.T) {
res := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})
allocInfo := newAllocationWithKey(aKey, appID1, nodeID1, res)
allocInfo.placeholder = true
- app.AddAllocation(allocInfo)
+ err1 := app.AddAllocation(allocInfo)
+ assert.NilError(t, err1)
err := app.handleApplicationEventWithLocking(RunApplication)
assert.NilError(t, err, "no error expected new to accepted")
@@ -819,7 +824,8 @@ func TestStateChangeOnUpdate(t *testing.T) {
assert.Assert(t, app.IsAccepted(), "Application did not change to accepted state: %s", app.CurrentState())
// add an alloc
allocInfo := newAllocationWithKey(askID, appID1, nodeID1, res)
- app.AddAllocation(allocInfo)
+ err1 := app.AddAllocation(allocInfo)
+ assert.NilError(t, err1)
// app should be running
assert.Assert(t, app.IsRunning(), "Application did not return running state after alloc: %s", app.CurrentState())
assertUserGroupResource(t, getTestUserGroup(), res)
@@ -882,7 +888,8 @@ func TestStateChangeOnPlaceholderAdd(t *testing.T) {
assert.Assert(t, app.IsAccepted(), "Application did not change to accepted state: %s", app.CurrentState())
// add an alloc based on the placeholder ask
allocInfo := newAllocationAll(askID, appID1, nodeID1, tg1, res, true, 0)
- app.AddAllocation(allocInfo)
+ err1 := app.AddAllocation(allocInfo)
+ assert.NilError(t, err1)
// app should be in the same state as it was before as it is a placeholder allocation
assert.Assert(t, app.IsAccepted(), "Application did not return accepted state after alloc: %s", app.CurrentState())
assert.Assert(t, resources.Equals(app.GetPlaceholderResource(), res), "placeholder allocation not set as expected")
@@ -919,7 +926,8 @@ func TestAllocations(t *testing.T) {
res, err := resources.NewResourceFromConf(resMap)
assert.NilError(t, err, "failed to create resource with error")
alloc := newAllocation(appID1, nodeID1, res)
- app.AddAllocation(alloc)
+ err1 := app.AddAllocation(alloc)
+ assert.NilError(t, err1)
if !resources.Equals(app.allocatedResource, res) {
t.Errorf("allocated resources is not updated correctly: %v", app.allocatedResource)
}
@@ -930,10 +938,12 @@ func TestAllocations(t *testing.T) {
// add more allocations to test the removals
alloc = newAllocation(appID1, nodeID1, res)
- app.AddAllocation(alloc)
+ err2 := app.AddAllocation(alloc)
+ assert.NilError(t, err2)
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 2))
alloc = newAllocation(appID1, nodeID1, res)
- app.AddAllocation(alloc)
+ err3 := app.AddAllocation(alloc)
+ assert.NilError(t, err3)
allocs = app.GetAllAllocations()
assert.Equal(t, len(allocs), 3)
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 3))
@@ -968,7 +978,8 @@ func TestRemoveAllAllocationsWithPendingAsks(t *testing.T) {
// 1. Add an allocation: UGM tracks 100m/10v
alloc := newAllocation(appID1, nodeID1, res)
- app.AddAllocation(alloc)
+ err1 := app.AddAllocation(alloc)
+ assert.NilError(t, err1)
assertUserGroupResource(t, getTestUserGroup(), res)
// 2. Add a pending ask so that app.pending is non-zero
@@ -1017,7 +1028,8 @@ func TestGangAllocChange(t *testing.T) {
assert.NilError(t, err, "failed to create resource with error")
alloc := newAllocation(appID1, nodeID1, res)
alloc.placeholder = true
- app.AddAllocation(alloc)
+ err1 := app.AddAllocation(alloc)
+ assert.NilError(t, err1)
assert.Assert(t, resources.Equals(app.allocatedPlaceholder, res), "allocated placeholders resources is not updated correctly: %s", app.allocatedPlaceholder.String())
assert.Equal(t, len(app.GetAllAllocations()), 1)
assert.Assert(t, app.IsAccepted(), "app should still be in accepted state")
@@ -1026,7 +1038,8 @@ func TestGangAllocChange(t *testing.T) {
// add second placeholder this should trigger state update
alloc = newAllocation(appID1, nodeID1, res)
alloc.placeholder = true
- app.AddAllocation(alloc)
+ err2 := app.AddAllocation(alloc)
+ assert.NilError(t, err2)
assert.Assert(t, resources.Equals(app.allocatedPlaceholder, totalPH), "allocated placeholders resources is not updated correctly: %s", app.allocatedPlaceholder.String())
assert.Equal(t, len(app.GetAllAllocations()), 2)
assert.Assert(t, app.IsRunning(), "app should have changed to running state")
@@ -1034,14 +1047,16 @@ func TestGangAllocChange(t *testing.T) {
// add a real alloc this should NOT trigger state update
alloc = newAllocation(appID1, nodeID1, res)
- app.AddAllocation(alloc)
+ err3 := app.AddAllocation(alloc)
+ assert.NilError(t, err3)
assert.Equal(t, len(app.GetAllAllocations()), 3)
assert.Assert(t, app.IsRunning(), "app should still be in running state")
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 3))
// add a second real alloc this should NOT trigger state update
alloc = newAllocation(appID1, nodeID1, res)
- app.AddAllocation(alloc)
+ err4 := app.AddAllocation(alloc)
+ assert.NilError(t, err4)
assert.Equal(t, len(app.GetAllAllocations()), 4)
assert.Assert(t, app.IsRunning(), "app should still be in running state")
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 4))
@@ -1062,7 +1077,8 @@ func TestAllocChange(t *testing.T) {
assert.NilError(t, err, "failed to create resource with error")
alloc := newAllocation(appID1, nodeID1, res)
// adding a normal allocation should change the state
- app.AddAllocation(alloc)
+ err1 := app.AddAllocation(alloc)
+ assert.NilError(t, err1)
assert.Assert(t, resources.Equals(app.allocatedResource, res), "allocated resources is not updated correctly: %s", app.allocatedResource.String())
assert.Equal(t, len(app.GetAllAllocations()), 1)
assert.Assert(t, app.IsRunning(), "app should be in running state")
@@ -1070,7 +1086,8 @@ func TestAllocChange(t *testing.T) {
// add a second real alloc this should trigger state update
alloc = newAllocation(appID1, nodeID1, res)
- app.AddAllocation(alloc)
+ err2 := app.AddAllocation(alloc)
+ assert.NilError(t, err2)
assert.Equal(t, len(app.GetAllAllocations()), 2)
assert.Assert(t, app.IsRunning(), "app should have changed to running` state")
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 2))
@@ -1137,8 +1154,10 @@ func TestUpdateAllocationResourceAllocated(t *testing.T) {
assert.NilError(t, err, "failed to create resource with error")
alloc1 := newAllocationWithKey(alloc, appID1, nodeID1, res)
queue.IncAllocatedResource(res, false)
- app.RecoverAllocationAsk(alloc1)
- app.AddAllocation(alloc1)
+ err1 := app.RecoverAllocationAsk(alloc1)
+ assert.NilError(t, err1)
+ err2 := app.AddAllocation(alloc1)
+ assert.NilError(t, err2)
assert.Check(t, resources.Equals(res, queue.GetAllocatedResource()), "resources not on queue")
// check nil alloc update
@@ -1262,7 +1281,8 @@ func TestResourceUsageAggregation(t *testing.T) {
alloc.SetInstanceType(instType1)
// Mock the time to be 3 seconds before
alloc.SetBindTime(time.Now().Add(-3 * time.Second))
- app.AddAllocation(alloc)
+ err1 := app.AddAllocation(alloc)
+ assert.NilError(t, err1)
if !resources.Equals(app.allocatedResource, res) {
t.Errorf("allocated resources is not updated correctly: %v", app.allocatedResource)
@@ -1285,7 +1305,8 @@ func TestResourceUsageAggregation(t *testing.T) {
// Mock the time to be 3 seconds before
alloc.SetBindTime(time.Now().Add(-3 * time.Second))
- app.AddAllocation(alloc)
+ err2 := app.AddAllocation(alloc)
+ assert.NilError(t, err2)
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 2))
// remove one of the 2
@@ -1300,7 +1321,8 @@ func TestResourceUsageAggregation(t *testing.T) {
alloc = newAllocation(appID1, nodeID1, res)
alloc.SetInstanceType(instType1)
- app.AddAllocation(alloc)
+ err3 := app.AddAllocation(alloc)
+ assert.NilError(t, err3)
allocs = app.GetAllAllocations()
assert.Equal(t, len(allocs), 2)
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 2))
@@ -1425,7 +1447,8 @@ func TestReplaceAllocation(t *testing.T) {
assert.NilError(t, err, "failed to create resource with error")
ph := newPlaceholderAlloc(appID1, nodeID1, res, "tg")
// add the placeholder to the app
- app.AddAllocation(ph)
+ err1 := app.AddAllocation(ph)
+ assert.NilError(t, err1)
// add PlaceholderData
app.addPlaceholderData(ph)
assertPlaceholderData(t, app, "tg", 1, 0, 0, res)
@@ -1447,7 +1470,8 @@ func TestReplaceAllocation(t *testing.T) {
// add the placeholder back to the app, the failure test above changed state and removed the ph
app.SetState(Running.String())
- app.AddAllocation(ph)
+ err2 := app.AddAllocation(ph)
+ assert.NilError(t, err2)
app.addPlaceholderData(ph)
assertPlaceholderData(t, app, "tg", 2, 0, 1, res)
assertUserGroupResource(t, getTestUserGroup(), res)
@@ -1466,7 +1490,8 @@ func TestReplaceAllocation(t *testing.T) {
// add the placeholder back to the app, the failure test above changed state and removed the ph
app.SetState(Running.String())
ph.ClearRelease()
- app.AddAllocation(ph)
+ err3 := app.AddAllocation(ph)
+ assert.NilError(t, err3)
app.addPlaceholderData(ph)
assertPlaceholderData(t, app, "tg", 3, 0, 2, res)
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 2))
@@ -1487,14 +1512,17 @@ func TestReplaceAllocationTracking(t *testing.T) {
ph1.SetInstanceType(instType1)
ph2.SetInstanceType(instType1)
ph3.SetInstanceType(instType1)
- app.AddAllocation(ph1)
+ err1 := app.AddAllocation(ph1)
+ assert.NilError(t, err1)
assert.NilError(t, err, "could not add ask")
app.addPlaceholderData(ph1)
assert.Equal(t, true, app.HasPlaceholderAllocation())
- app.AddAllocation(ph2)
+ err2 := app.AddAllocation(ph2)
+ assert.NilError(t, err2)
assert.NilError(t, err, "could not add ask")
app.addPlaceholderData(ph2)
- app.AddAllocation(ph3)
+ err3 := app.AddAllocation(ph3)
+ assert.NilError(t, err3)
assert.NilError(t, err, "could not add ask")
app.addPlaceholderData(ph3)
@@ -1563,14 +1591,16 @@ func runTimeoutPlaceholderTest(t *testing.T, expectedState string, gangSchedulin
// add the placeholder to the app
ph1 := newPlaceholderAlloc(appID1, nodeID1, res, tg2)
- app.AddAllocation(ph1)
+ err1 := app.AddAllocation(ph1)
+ assert.NilError(t, err1)
app.addPlaceholderDataWithLocking(ph1)
assertPlaceholderData(t, app, tg2, 1, 0, 0, res)
assertUserGroupResource(t, getTestUserGroup(), res)
assert.Assert(t, app.getPlaceholderTimer() != nil, "Placeholder timer should be initiated after the first placeholder allocation")
// add a second one to check the filter
ph2 := newPlaceholderAlloc(appID1, nodeID1, res, tg2)
- app.AddAllocation(ph2)
+ err2 := app.AddAllocation(ph2)
+ assert.NilError(t, err2)
app.addPlaceholderDataWithLocking(ph2)
assertPlaceholderData(t, app, tg2, 2, 0, 0, res)
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 2))
@@ -1644,20 +1674,23 @@ func TestTimeoutPlaceholderAllocReleased(t *testing.T) {
phReleased := newPlaceholderAlloc(appID1, nodeID1, res, tg1)
err = phReleased.SetReleased(true)
assert.NilError(t, err, "Unexpected error when releasing placeholder")
- app.AddAllocation(phReleased)
+ err1 := app.AddAllocation(phReleased)
+ assert.NilError(t, err1)
app.addPlaceholderDataWithLocking(phReleased)
assertPlaceholderData(t, app, tg1, 1, 0, 0, res)
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 1))
assert.Assert(t, app.getPlaceholderTimer() != nil, "Placeholder timer should be initiated after the first placeholder allocation")
ph := newPlaceholderAlloc(appID1, nodeID1, res, tg1)
- app.AddAllocation(ph)
+ err2 := app.AddAllocation(ph)
+ assert.NilError(t, err2)
app.addPlaceholderDataWithLocking(ph)
assertPlaceholderData(t, app, tg1, 2, 0, 0, res)
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 2))
alloc := newAllocation(appID1, nodeID1, res)
- app.AddAllocation(alloc)
+ err3 := app.AddAllocation(alloc)
+ assert.NilError(t, err3)
assert.Assert(t, app.IsRunning(), "App should be in running state after the first allocation")
err = common.WaitForCondition(10*time.Millisecond, 1*time.Second, func() bool {
return app.getPlaceholderTimer() == nil
@@ -1706,7 +1739,7 @@ func TestTimeoutPlaceholderProcessingDoesNotWaitForRMReply(t *testing.T) {
res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 100, "vcores": 10})
ph := newPlaceholderAlloc(appID1, nodeID1, res, tg1)
- app.AddAllocation(ph)
+ assert.NilError(t, app.AddAllocation(ph))
app.addPlaceholderDataWithLocking(ph)
app.SetState(Running.String())
app.clearPlaceholderTimer()
@@ -1745,7 +1778,7 @@ func TestTimeoutStateTimerDoesNotWaitForRMReply(t *testing.T) {
app.SetState(Accepted.String())
res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 100})
ph := newPlaceholderAlloc(appID1, nodeID1, res, tg1)
- app.AddAllocation(ph)
+ assert.NilError(t, app.AddAllocation(ph))
app.addPlaceholderDataWithLocking(ph)
app.clearPlaceholderTimer()
app.SetState(Completing.String())
@@ -1794,20 +1827,23 @@ func TestTimeoutPlaceholderAllocPreempted(t *testing.T) {
phReleased := newPlaceholderAlloc(appID1, nodeID1, res, tg1)
err = phReleased.SetReleased(true)
assert.NilError(t, err, "Unexpected error when releasing placeholder")
- app.AddAllocation(phReleased)
+ err1 := app.AddAllocation(phReleased)
+ assert.NilError(t, err1)
app.addPlaceholderDataWithLocking(phReleased)
assertPlaceholderData(t, app, tg1, 1, 0, 0, res)
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 1))
assert.Assert(t, app.getPlaceholderTimer() != nil, "Placeholder timer should be initiated after the first placeholder allocation")
ph := newPlaceholderAlloc(appID1, nodeID1, res, tg1)
- app.AddAllocation(ph)
+ err2 := app.AddAllocation(ph)
+ assert.NilError(t, err2)
app.addPlaceholderDataWithLocking(ph)
assertPlaceholderData(t, app, tg1, 2, 0, 0, res)
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 2))
alloc := newAllocation(appID1, nodeID1, res)
- app.AddAllocation(alloc)
+ err3 := app.AddAllocation(alloc)
+ assert.NilError(t, err3)
assert.Assert(t, app.IsRunning(), "App should be in running state after the first allocation")
err = common.WaitForCondition(10*time.Millisecond, 1*time.Second, func() bool {
// Preempt the placeholder in the meantime
@@ -1868,14 +1904,16 @@ func TestTimeoutPlaceholderCompleting(t *testing.T) {
assert.NilError(t, err, "Unexpected error when creating resource from map")
// add the placeholder to the app
ph := newPlaceholderAlloc(appID1, nodeID1, res, tg1)
- app.AddAllocation(ph)
+ err1 := app.AddAllocation(ph)
+ assert.NilError(t, err1)
assert.Assert(t, app.getPlaceholderTimer() != nil, "Placeholder timer should be initiated after the first placeholder allocation")
app.addPlaceholderDataWithLocking(ph)
assertPlaceholderData(t, app, tg1, 1, 0, 0, res)
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 1))
// add a real allocation as well
alloc := newAllocation(appID1, nodeID1, res)
- app.AddAllocation(alloc)
+ err2 := app.AddAllocation(alloc)
+ assert.NilError(t, err2)
// move on to running
app.SetState(Running.String())
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 2))
@@ -1922,14 +1960,16 @@ func TestTimeoutPlaceholderCompletingWithPreemptedPh(t *testing.T) {
// add the placeholder to the app
tg := "tg-1"
ph := newPlaceholderAlloc(appID1, nodeID1, res, tg)
- app.AddAllocation(ph)
+ err1 := app.AddAllocation(ph)
+ assert.NilError(t, err1)
assert.Assert(t, app.getPlaceholderTimer() != nil, "Placeholder timer should be initiated after the first placeholder allocation")
app.addPlaceholderDataWithLocking(ph)
assertPlaceholderData(t, app, tg, 1, 0, 0, res)
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 1))
// add a real allocation as well
alloc := newAllocation(appID1, nodeID1, res)
- app.AddAllocation(alloc)
+ err2 := app.AddAllocation(alloc)
+ assert.NilError(t, err2)
// move on to running
app.SetState(Running.String())
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 2))
@@ -1983,14 +2023,16 @@ func TestAppTimersAfterAppRemoval(t *testing.T) {
// add the placeholder to the app
tg := "tg-1"
ph := newPlaceholderAlloc(appID1, nodeID1, res, tg)
- app.AddAllocation(ph)
+ err1 := app.AddAllocation(ph)
+ assert.NilError(t, err1)
assert.Assert(t, app.getPlaceholderTimer() != nil, "Placeholder timer should be initiated after the first placeholder allocation")
app.addPlaceholderDataWithLocking(ph)
assertPlaceholderData(t, app, tg, 1, 0, 0, res)
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 1))
// add a real allocation as well
alloc := newAllocation(appID1, nodeID1, res)
- app.AddAllocation(alloc)
+ err2 := app.AddAllocation(alloc)
+ assert.NilError(t, err2)
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res, 2))
// move on to running
app.SetState(Running.String())
@@ -2193,7 +2235,8 @@ func TestTryRequiredNode(t *testing.T) {
childQ.applications[appID1] = app
allocRes := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 3})
alloc := newAllocation(aKey, nodeID1, allocRes)
- app.AddAllocation(alloc)
+ err1 := app.AddAllocation(alloc)
+ assert.NilError(t, err1)
node.AddAllocation(alloc)
ask := newAllocationAsk(aKey2, appID1, allocRes)
@@ -2269,7 +2312,8 @@ func TestTryRequiredNodeReserve(t *testing.T) {
childQ.applications[appID1] = app
allocRes := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 3})
alloc := newAllocation(aKey, nodeID1, allocRes)
- app.AddAllocation(alloc)
+ err1 := app.AddAllocation(alloc)
+ assert.NilError(t, err1)
node.AddAllocation(alloc)
ask := newAllocationAsk(aKey2, appID1, allocRes)
@@ -2302,7 +2346,8 @@ func TestTryRequiredNodeCancel(t *testing.T) {
childQ.applications[appID1] = app
allocRes := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 3})
alloc := newAllocation(aKey, nodeID1, allocRes)
- app.AddAllocation(alloc)
+ err1 := app.AddAllocation(alloc)
+ assert.NilError(t, err1)
node.AddAllocation(alloc)
ask := newAllocationAsk(aKey2, appID1, allocRes)
@@ -2344,7 +2389,8 @@ func TestTryRequiredNodeAdd(t *testing.T) {
childQ.applications[appID1] = app
allocRes := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 3})
alloc := newAllocation(aKey, nodeID1, allocRes)
- app.AddAllocation(alloc)
+ err1 := app.AddAllocation(alloc)
+ assert.NilError(t, err1)
node.AddAllocation(alloc)
ask := newAllocationAsk(aKey2, appID1, allocRes)
@@ -2933,7 +2979,8 @@ func TestRollbackAllocationAskNotTracked(t *testing.T) {
_, err = app.AllocateAsk(aKey)
assert.NilError(t, err, "ask should have been allocated")
// confirm the allocation so it lands in sa.allocations, which is what RollbackAllocation looks in
- app.AddAllocation(ask)
+ err1 := app.AddAllocation(ask)
+ assert.NilError(t, err1)
// wipe the asks: sa.allocations (and so the rollback target) survives this
app.RemoveAllocationAsk("")
@@ -3037,8 +3084,10 @@ func TestAllocationEvents(t *testing.T) { //nolint:funlen
alloc2 := newAllocation(appID1, nodeID1, res)
// add + remove
- app.AddAllocation(alloc1)
- app.AddAllocation(alloc2)
+ err1 := app.AddAllocation(alloc1)
+ assert.NilError(t, err1)
+ err2 := app.AddAllocation(alloc2)
+ assert.NilError(t, err2)
app.RemoveAllocation(alloc1.GetAllocationKey(), si.TerminationType_STOPPED_BY_RM)
app.RemoveAllocation(alloc2.GetAllocationKey(), si.TerminationType_PLACEHOLDER_REPLACED)
noEvents := uint64(0)
@@ -3073,7 +3122,8 @@ func TestAllocationEvents(t *testing.T) { //nolint:funlen
// add + replace
alloc1.placeholder = true
- app.AddAllocation(alloc1)
+ err3 := app.AddAllocation(alloc1)
+ assert.NilError(t, err3)
app.ReplaceAllocation(alloc1.GetAllocationKey())
noEvents = 0
err = common.WaitForCondition(10*time.Millisecond, time.Second, func() bool {
@@ -3095,8 +3145,10 @@ func TestAllocationEvents(t *testing.T) { //nolint:funlen
assert.Equal(t, "app-1", records[0].ObjectID)
// add + remove all
- app.AddAllocation(alloc1)
- app.AddAllocation(alloc2)
+ err4 := app.AddAllocation(alloc1)
+ assert.NilError(t, err4)
+ err5 := app.AddAllocation(alloc2)
+ assert.NilError(t, err5)
app.RemoveAllAllocations()
err = common.WaitForCondition(10*time.Millisecond, time.Second, func() bool {
noEvents = eventSystem.Store.CountStoredEvents()
@@ -3152,7 +3204,8 @@ func TestPlaceholderLargerEvent(t *testing.T) {
alloc1 := newAllocation(appID1, nodeID1, smallerRes)
alloc1.placeholder = true
alloc1.taskGroupName = "testGroup"
- app.AddAllocation(alloc1)
+ err1 := app.AddAllocation(alloc1)
+ assert.NilError(t, err1)
ask := newAllocationAsk("alloc-0", "app-1", res)
ask.taskGroupName = "testGroup"
err = app.AddAllocationAsk(ask)
@@ -4161,7 +4214,8 @@ func TestTryPlaceHolderAllocateSmallerRequest(t *testing.T) {
res := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})
ph := newPlaceholderAlloc(appID0, nodeID1, res, tg1)
- app.AddAllocation(ph)
+ err1 := app.AddAllocation(ph)
+ assert.NilError(t, err1)
app.addPlaceholderData(ph)
assertPlaceholderData(t, app, tg1, 1, 0, 0, res)
@@ -4205,7 +4259,8 @@ func TestTryPlaceHolderAllocateLargerRequest(t *testing.T) {
res := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})
ph := newPlaceholderAlloc(appID0, nodeID1, res, tg1)
- app.AddAllocation(ph)
+ err1 := app.AddAllocation(ph)
+ assert.NilError(t, err1)
app.addPlaceholderData(ph)
assertPlaceholderData(t, app, tg1, 1, 0, 0, res)
@@ -4254,7 +4309,8 @@ func TestTryPlaceHolderAllocateDifferentTaskGroups(t *testing.T) {
res := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})
ph := newPlaceholderAlloc(appID0, nodeID1, res, tg1)
- app.AddAllocation(ph)
+ err1 := app.AddAllocation(ph)
+ assert.NilError(t, err1)
app.addPlaceholderData(ph)
assertPlaceholderData(t, app, tg1, 1, 0, 0, res)
@@ -4306,7 +4362,8 @@ func TestTryPlaceHolderAllocateDifferentNodes(t *testing.T) {
res := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})
ph := newPlaceholderAlloc(appID0, nodeID1, res, tg1)
- app.AddAllocation(ph)
+ err1 := app.AddAllocation(ph)
+ assert.NilError(t, err1)
app.addPlaceholderData(ph)
assertPlaceholderData(t, app, tg1, 1, 0, 0, res)
plugins.RegisterSchedulerPlugin(tt.mockPlugin)
@@ -4388,7 +4445,8 @@ func TestTryPlaceHolderAllocateRevertsOnPreemptedPlaceholder(t *testing.T) {
res := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})
ph := newPlaceholderAlloc(appID0, nodeID1, res, tg1)
- app.AddAllocation(ph)
+ err1 := app.AddAllocation(ph)
+ assert.NilError(t, err1)
app.addPlaceholderData(ph)
assertPlaceholderData(t, app, tg1, 1, 0, 0, res)
@@ -4511,12 +4569,14 @@ func TestAppSubmissionTime(t *testing.T) {
// allocations
alloc1 := newAllocation(appID1, nodeID1, res)
alloc1.createTime = time.Unix(0, 60)
- app.AddAllocation(alloc1)
+ err1 := app.AddAllocation(alloc1)
+ assert.NilError(t, err1)
assert.Equal(t, app.submissionTime, time.Unix(0, 50), "app submission time is not set properly")
alloc2 := newAllocation(appID1, nodeID1, res)
alloc2.createTime = time.Unix(0, 30)
- app.AddAllocation(alloc2)
+ err2 := app.AddAllocation(alloc2)
+ assert.NilError(t, err2)
assert.Equal(t, app.submissionTime, time.Unix(0, 30), "app submission time is not set properly")
}
@@ -4683,7 +4743,8 @@ func TestRollbackAllocationFromRunning(t *testing.T) {
assert.Assert(t, resources.Equals(delta, res), "AllocateAsk delta should equal res, got %v", delta)
ask.SetNodeID(nodeID1)
- app.AddAllocation(ask)
+ err1 := app.AddAllocation(ask)
+ assert.NilError(t, err1)
assert.Assert(t, app.IsRunning(), "app should be in Running state after AddAllocation")
assert.Assert(t, resources.IsZero(app.GetPendingResource()), "pending should be zero before rollback")
assert.Assert(t, resources.Equals(app.GetAllocatedResource(), res), "allocated should equal res before rollback")
@@ -4730,7 +4791,8 @@ func TestRollbackAllocationPartial(t *testing.T) {
_, err = app.AllocateAsk(aKey)
assert.NilError(t, err, "AllocateAsk for ask1 should succeed")
ask1.SetNodeID(nodeID1)
- app.AddAllocation(ask1)
+ err1 := app.AddAllocation(ask1)
+ assert.NilError(t, err1)
// Allocate second ask.
ask2 := newAllocationAsk(aKey2, appID1, res)
@@ -4739,7 +4801,8 @@ func TestRollbackAllocationPartial(t *testing.T) {
_, err = app.AllocateAsk(aKey2)
assert.NilError(t, err, "AllocateAsk for ask2 should succeed")
ask2.SetNodeID(nodeID2)
- app.AddAllocation(ask2)
+ err2 := app.AddAllocation(ask2)
+ assert.NilError(t, err2)
doubleRes := resources.Multiply(res, 2)
assert.Assert(t, app.IsRunning(), "app should be Running with two allocations")
diff --git a/pkg/scheduler/objects/node_collection_test.go b/pkg/scheduler/objects/node_collection_test.go
index 6b2db5ca..9d362631 100644
--- a/pkg/scheduler/objects/node_collection_test.go
+++ b/pkg/scheduler/objects/node_collection_test.go
@@ -405,7 +405,8 @@ func TestNodeIteratorReserveUpdate(t *testing.T) {
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)
+ err := app.AddAllocation(ask)
+ assert.NilError(t, err)
assert.NilError(t, node.Reserve(app, ask), "Reserving failed.")
}
diff --git a/pkg/scheduler/objects/preemption_test.go b/pkg/scheduler/objects/preemption_test.go
index 26ffc5c5..c001ea65 100644
--- a/pkg/scheduler/objects/preemption_test.go
+++ b/pkg/scheduler/objects/preemption_test.go
@@ -77,7 +77,9 @@ func creatApp1WithTwoDifferentAllocations(
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(app1Rec))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ if err := app1.AddAllocation(alloc1); err != nil {
+ return nil, nil, err
+ }
if !node1.TryAddAllocation(alloc1) {
return nil, nil, fmt.Errorf("node alloc1 failed")
}
@@ -85,7 +87,9 @@ func creatApp1WithTwoDifferentAllocations(
if node2 != nil {
alloc2 = newAllocationWithKey("alloc2", appID1, nodeID2, resources.NewResourceFromMap(app2Rec))
alloc2.createTime = ask2.createTime
- app1.AddAllocation(alloc2)
+ if err := app1.AddAllocation(alloc2); err != nil {
+ return nil, nil, err
+ }
if !node2.TryAddAllocation(alloc2) {
return nil, nil, fmt.Errorf("node alloc2 failed")
}
@@ -650,13 +654,15 @@ func TestTryPreemptionOnNodeWithOGParentAndUGPreemptor(t *testing.T) {
if i%2 == 0 {
alloc1 := newAllocationWithKey(ask1.allocationKey, appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err1 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err1)
assert.Check(t, node1.TryAddAllocation(alloc1), "node alloc1 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask1.GetAllocatedResource()))
} else {
alloc1 := newAllocationWithKey(ask1.allocationKey, appID1, nodeID2, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err2 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err2)
assert.Check(t, node2.TryAddAllocation(alloc1), "node alloc1 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask1.GetAllocatedResource()))
}
@@ -899,7 +905,8 @@ func TestTryPreemption_VictimReleased_InsufficientResource(t *testing.T) {
alloc4 := newAllocationWithKey("alloc4", appID3, nodeID2, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}))
alloc4.createTime = ask4.createTime
- app3.AddAllocation(alloc4)
+ err1 := app3.AddAllocation(alloc4)
+ assert.NilError(t, err1)
assert.Check(t, node2.TryAddAllocation(alloc4), "node alloc2 failed")
assert.NilError(t, childQ3.TryIncAllocatedResource(ask4.GetAllocatedResource()))
@@ -963,12 +970,14 @@ func TestTryPreemption_VictimsAvailableOnDifferentNodes(t *testing.T) {
assert.NilError(t, app1.AddAllocationAsk(ask2))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 4, "pods": 1}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err1 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err1)
appQueueMapping.AddAppQueueMapping(app1.ApplicationID, childQ1)
assert.Check(t, node1.TryAddAllocation(alloc1), "node alloc1 failed")
alloc2 := newAllocationWithKey("alloc2", appID1, nodeID2, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 2, "pods": 1}))
alloc2.createTime = ask2.createTime
- app1.AddAllocation(alloc2)
+ err2 := app1.AddAllocation(alloc2)
+ assert.NilError(t, err2)
appQueueMapping.AddAppQueueMapping(app1.ApplicationID, childQ1)
assert.Check(t, node2.TryAddAllocation(alloc2), "node alloc2 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask1.GetAllocatedResource()))
@@ -1040,7 +1049,8 @@ func TestTryPreemption_OnQueue_VictimsOnDifferentNodes(t *testing.T) {
alloc4 := newAllocationWithKey("alloc4", appID3, nodeID2, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}))
alloc4.createTime = ask4.createTime
- app3.AddAllocation(alloc4)
+ err1 := app3.AddAllocation(alloc4)
+ assert.NilError(t, err1)
assert.Check(t, node2.TryAddAllocation(alloc4), "node alloc2 failed")
assert.NilError(t, childQ3.TryIncAllocatedResource(ask4.GetAllocatedResource()))
@@ -1108,11 +1118,13 @@ func TestTryPreemption_OnQueue_VictimsAvailable_LowerPriority(t *testing.T) {
assert.NilError(t, app1.AddAllocationAsk(ask2))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err1 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err1)
assert.Check(t, node1.TryAddAllocation(alloc1), "node alloc1 failed")
alloc2 := newAllocationWithKey("alloc2", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}))
alloc2.createTime = ask2.createTime
- app1.AddAllocation(alloc2)
+ err2 := app1.AddAllocation(alloc2)
+ assert.NilError(t, err2)
assert.Check(t, node1.TryAddAllocation(alloc2), "node alloc2 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask1.GetAllocatedResource()))
@@ -1129,7 +1141,8 @@ func TestTryPreemption_OnQueue_VictimsAvailable_LowerPriority(t *testing.T) {
alloc4 := newAllocationAll("alloc4", appID3, nodeID2, "", resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}), false, 1000)
alloc4.createTime = ask4.createTime
- app3.AddAllocation(alloc4)
+ err3 := app3.AddAllocation(alloc4)
+ assert.NilError(t, err3)
assert.Check(t, node2.TryAddAllocation(alloc4), "node alloc2 failed")
assert.NilError(t, childQ3.TryIncAllocatedResource(ask4.GetAllocatedResource()))
@@ -1198,11 +1211,13 @@ func TestTryPreemption_AskResTypesDifferent_GuaranteedSetOnPreemptorSide(t *test
assert.NilError(t, app1.AddAllocationAsk(ask2))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1, "mem": 200}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err1 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err1)
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
alloc2 := newAllocationWithKey("alloc2", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1, "mem": 200}))
alloc2.createTime = ask2.createTime
- app1.AddAllocation(alloc2)
+ err2 := app1.AddAllocation(alloc2)
+ assert.NilError(t, err2)
assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
assert.NilError(t, childQ2.TryIncAllocatedResource(ask1.GetAllocatedResource()))
assert.NilError(t, childQ2.TryIncAllocatedResource(ask2.GetAllocatedResource()))
@@ -1271,11 +1286,13 @@ func TestTryPreemption_OnNode_AskResTypesDifferent_GuaranteedSetOnPreemptorSide(
assert.NilError(t, app1.AddAllocationAsk(ask2))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1, "mem": 200}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err1 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err1)
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
alloc2 := newAllocationWithKey("alloc2", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1, "mem": 200}))
alloc2.createTime = ask2.createTime
- app1.AddAllocation(alloc2)
+ err2 := app1.AddAllocation(alloc2)
+ assert.NilError(t, err2)
assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
assert.NilError(t, childQ2.TryIncAllocatedResource(ask1.GetAllocatedResource()))
assert.NilError(t, childQ2.TryIncAllocatedResource(ask2.GetAllocatedResource()))
@@ -1344,7 +1361,8 @@ func TestTryPreemption_AskResTypesDifferent_GuaranteedSetOnVictimAndPreemptorSid
assert.NilError(t, app1.AddAllocationAsk(askN))
allocN := newAllocationWithKey(askN.allocationKey, appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"mem": 100}))
allocN.createTime = askN.createTime
- app1.AddAllocation(allocN)
+ err1 := app1.AddAllocation(allocN)
+ assert.NilError(t, err1)
assert.Check(t, node.TryAddAllocation(allocN), "node alloc1 failed")
}
@@ -1359,15 +1377,18 @@ func TestTryPreemption_AskResTypesDifferent_GuaranteedSetOnVictimAndPreemptorSid
assert.NilError(t, app1.AddAllocationAsk(ask3))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1, "mem": 100}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err2 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err2)
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
alloc2 := newAllocationWithKey("alloc2", appID2, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1, "mem": 100}))
alloc2.createTime = ask2.createTime
- app2.AddAllocation(alloc2)
+ err3 := app2.AddAllocation(alloc2)
+ assert.NilError(t, err3)
assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
alloc3 := newAllocationWithKey("alloc3", appID2, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1, "mem": 100}))
alloc3.createTime = ask3.createTime
- app3.AddAllocation(alloc3)
+ err4 := app3.AddAllocation(alloc3)
+ assert.NilError(t, err4)
assert.Check(t, node.TryAddAllocation(alloc3), "node alloc3 failed")
for i := 5; i < 8; i++ {
@@ -1440,7 +1461,8 @@ func TestTryPreemption_OnNode_AskResTypesDifferent_GuaranteedSetOnVictimAndPreem
assert.NilError(t, app1.AddAllocationAsk(askN))
allocN := newAllocationWithKey(askN.allocationKey, appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"mem": 100}))
allocN.createTime = askN.createTime
- app1.AddAllocation(allocN)
+ err1 := app1.AddAllocation(allocN)
+ assert.NilError(t, err1)
assert.Check(t, node.TryAddAllocation(allocN), "node alloc1 failed")
}
@@ -1455,15 +1477,18 @@ func TestTryPreemption_OnNode_AskResTypesDifferent_GuaranteedSetOnVictimAndPreem
assert.NilError(t, app1.AddAllocationAsk(ask3))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1, "mem": 100}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err2 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err2)
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
alloc2 := newAllocationWithKey("alloc2", appID2, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1, "mem": 100}))
alloc2.createTime = ask2.createTime
- app2.AddAllocation(alloc2)
+ err3 := app2.AddAllocation(alloc2)
+ assert.NilError(t, err3)
assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
alloc3 := newAllocationWithKey("alloc3", appID2, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1, "mem": 100}))
alloc3.createTime = ask3.createTime
- app3.AddAllocation(alloc3)
+ err4 := app3.AddAllocation(alloc3)
+ assert.NilError(t, err4)
assert.Check(t, node.TryAddAllocation(alloc3), "node alloc3 failed")
for i := 5; i < 8; i++ {
@@ -1568,7 +1593,8 @@ func TestTryPreemption_AskResTypesSame_GuaranteedSetOnPreemptorSide(t *testing.T
assert.NilError(t, app1.AddAllocationAsk(askN))
allocN := newAllocationWithKey(askN.allocationKey, appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"gpu": 100}))
allocN.createTime = askN.createTime
- app1.AddAllocation(allocN)
+ err1 := app1.AddAllocation(allocN)
+ assert.NilError(t, err1)
assert.Check(t, node.TryAddAllocation(allocN), "node alloc1 failed")
}
@@ -1583,15 +1609,18 @@ func TestTryPreemption_AskResTypesSame_GuaranteedSetOnPreemptorSide(t *testing.T
assert.NilError(t, app1.AddAllocationAsk(ask3))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1, "pods": 1}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err2 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err2)
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
alloc2 := newAllocationWithKey("alloc2", appID2, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1, "pods": 1}))
alloc2.createTime = ask2.createTime
- app2.AddAllocation(alloc2)
+ err3 := app2.AddAllocation(alloc2)
+ assert.NilError(t, err3)
assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
alloc3 := newAllocationWithKey("alloc3", appID2, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1, "pods": 1}))
alloc3.createTime = ask3.createTime
- app3.AddAllocation(alloc3)
+ err4 := app3.AddAllocation(alloc3)
+ assert.NilError(t, err4)
assert.Check(t, node.TryAddAllocation(alloc3), "node alloc3 failed")
for i := 5; i < 8; i++ {
@@ -1671,7 +1700,8 @@ func TestTryPreemption_OnNode_AskResTypesSame_GuaranteedSetOnPreemptorSide(t *te
assert.NilError(t, app1.AddAllocationAsk(askN))
allocN := newAllocationWithKey(askN.allocationKey, appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"gpu": 100}))
allocN.createTime = askN.createTime
- app1.AddAllocation(allocN)
+ err1 := app1.AddAllocation(allocN)
+ assert.NilError(t, err1)
assert.Check(t, node.TryAddAllocation(allocN), "node alloc1 failed")
}
@@ -1686,15 +1716,18 @@ func TestTryPreemption_OnNode_AskResTypesSame_GuaranteedSetOnPreemptorSide(t *te
assert.NilError(t, app1.AddAllocationAsk(ask3))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err2 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err2)
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
alloc2 := newAllocationWithKey("alloc2", appID2, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1}))
alloc2.createTime = ask2.createTime
- app2.AddAllocation(alloc2)
+ err3 := app2.AddAllocation(alloc2)
+ assert.NilError(t, err3)
assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
alloc3 := newAllocationWithKey("alloc3", appID2, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1}))
alloc3.createTime = ask3.createTime
- app3.AddAllocation(alloc3)
+ err4 := app3.AddAllocation(alloc3)
+ assert.NilError(t, err4)
assert.Check(t, node.TryAddAllocation(alloc3), "node alloc3 failed")
for i := 5; i < 8; i++ {
@@ -1770,7 +1803,8 @@ func TestTryPreemption_AskResTypesSame_GuaranteedSetOnVictimAndPreemptorSides(t
assert.NilError(t, app1.AddAllocationAsk(askN))
allocN := newAllocationWithKey(askN.allocationKey, appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"gpu": 100}))
allocN.createTime = askN.createTime
- app1.AddAllocation(allocN)
+ err1 := app1.AddAllocation(allocN)
+ assert.NilError(t, err1)
assert.Check(t, node.TryAddAllocation(allocN), "node alloc1 failed")
}
@@ -1785,15 +1819,18 @@ func TestTryPreemption_AskResTypesSame_GuaranteedSetOnVictimAndPreemptorSides(t
assert.NilError(t, app1.AddAllocationAsk(ask3))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err2 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err2)
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
alloc2 := newAllocationWithKey("alloc2", appID2, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1}))
alloc2.createTime = ask2.createTime
- app2.AddAllocation(alloc2)
+ err3 := app2.AddAllocation(alloc2)
+ assert.NilError(t, err3)
assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
alloc3 := newAllocationWithKey("alloc3", appID2, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1}))
alloc3.createTime = ask3.createTime
- app3.AddAllocation(alloc3)
+ err4 := app3.AddAllocation(alloc3)
+ assert.NilError(t, err4)
assert.Check(t, node.TryAddAllocation(alloc3), "node alloc3 failed")
for i := 5; i < 8; i++ {
@@ -1867,7 +1904,8 @@ func TestTryPreemption_OnNode_AskResTypesSame_GuaranteedSetOnVictimAndPreemptorS
assert.NilError(t, app1.AddAllocationAsk(askN))
allocN := newAllocationWithKey(askN.allocationKey, appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"gpu": 100}))
allocN.createTime = askN.createTime
- app1.AddAllocation(allocN)
+ err1 := app1.AddAllocation(allocN)
+ assert.NilError(t, err1)
assert.Check(t, node.TryAddAllocation(allocN), "node alloc1 failed")
}
@@ -1882,15 +1920,18 @@ func TestTryPreemption_OnNode_AskResTypesSame_GuaranteedSetOnVictimAndPreemptorS
assert.NilError(t, app1.AddAllocationAsk(ask3))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err2 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err2)
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
alloc2 := newAllocationWithKey("alloc2", appID2, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1}))
alloc2.createTime = ask2.createTime
- app2.AddAllocation(alloc2)
+ err3 := app2.AddAllocation(alloc2)
+ assert.NilError(t, err3)
assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
alloc3 := newAllocationWithKey("alloc3", appID2, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1}))
alloc3.createTime = ask3.createTime
- app3.AddAllocation(alloc3)
+ err4 := app3.AddAllocation(alloc3)
+ assert.NilError(t, err4)
assert.Check(t, node.TryAddAllocation(alloc3), "node alloc3 failed")
for i := 5; i < 8; i++ {
@@ -1966,7 +2007,8 @@ func TestTryPreemption_OnNode_UGParent_With_UGPreemptorChild_GNotSetOnVictimChil
assert.NilError(t, app1.AddAllocationAsk(ask1))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err1 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err1)
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask1.GetAllocatedResource()))
@@ -1979,7 +2021,8 @@ func TestTryPreemption_OnNode_UGParent_With_UGPreemptorChild_GNotSetOnVictimChil
assert.NilError(t, app2.AddAllocationAsk(ask2))
alloc2 := newAllocationWithKey("alloc2", appID2, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1}))
alloc2.createTime = ask2.createTime
- app2.AddAllocation(alloc2)
+ err2 := app2.AddAllocation(alloc2)
+ assert.NilError(t, err2)
assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask2.GetAllocatedResource()))
@@ -2047,7 +2090,8 @@ func TestTryPreemption_OnNode_UGParent_With_GNotSetOnBothChilds(t *testing.T) {
assert.NilError(t, app1.AddAllocationAsk(ask1))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err1 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err1)
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask1.GetAllocatedResource()))
@@ -2060,7 +2104,8 @@ func TestTryPreemption_OnNode_UGParent_With_GNotSetOnBothChilds(t *testing.T) {
assert.NilError(t, app2.AddAllocationAsk(ask2))
alloc2 := newAllocationWithKey("alloc2", appID2, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1}))
alloc2.createTime = ask2.createTime
- app2.AddAllocation(alloc2)
+ err2 := app2.AddAllocation(alloc2)
+ assert.NilError(t, err2)
assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask2.GetAllocatedResource()))
@@ -2122,7 +2167,8 @@ func TestTryPreemption_OnNode_UGParent_With_UGPreemptorChild_OGVictimChild_As_Si
assert.NilError(t, app1.AddAllocationAsk(ask1))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err1 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err1)
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask1.GetAllocatedResource()))
@@ -2135,7 +2181,8 @@ func TestTryPreemption_OnNode_UGParent_With_UGPreemptorChild_OGVictimChild_As_Si
assert.NilError(t, app2.AddAllocationAsk(ask2))
alloc2 := newAllocationWithKey("alloc2", appID2, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"vcores": 1}))
alloc2.createTime = ask2.createTime
- app2.AddAllocation(alloc2)
+ err2 := app2.AddAllocation(alloc2)
+ assert.NilError(t, err2)
assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask2.GetAllocatedResource()))
@@ -2296,7 +2343,8 @@ func TestTryPreemption_AskQueue_With_OG_And_UG_ResTypes(t *testing.T) {
alloc3 := newAllocationWithKey("alloc3", appID2, nodeID2, alloc3Res)
alloc3.createTime = ask3.createTime
- app2.AddAllocation(alloc3)
+ err1 := app2.AddAllocation(alloc3)
+ assert.NilError(t, err1)
if !node2.TryAddAllocation(alloc3) {
t.Fatal("node alloc3 failed")
}
@@ -2369,7 +2417,8 @@ func TestTryPreemption_AskQueue_Under_DiffParent_With_OG_And_UG_ResTypes(t *test
alloc3 := newAllocationWithKey("alloc3", appID2, nodeID2, alloc3Res)
alloc3.createTime = ask3.createTime
- app2.AddAllocation(alloc3)
+ err1 := app2.AddAllocation(alloc3)
+ assert.NilError(t, err1)
if !node2.TryAddAllocation(alloc3) {
t.Fatal("node alloc3 failed")
}
@@ -2565,7 +2614,8 @@ func TestTryPreemption_NodeAvailableDeficit(t *testing.T) {
ask1 := newAllocationAsk("alloc1", appID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 2}))
assert.NilError(t, app1.AddAllocationAsk(ask1))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 2}))
- app1.AddAllocation(alloc1)
+ err1 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err1)
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask1.GetAllocatedResource()))
@@ -2716,7 +2766,7 @@ func TestTryPreemption_ResidualShortfall(t *testing.T) {
ask := newAllocationAskPriority(alloc.allocKey, alloc.appID, resources.NewResourceFromMap(map[string]resources.Quantity{"first": resources.Quantity(alloc.quantity)}), alloc.priority)
assert.NilError(t, app.AddAllocationAsk(ask))
allocation := newAllocationAll(alloc.allocKey, alloc.appID, alloc.nodeID, "", resources.NewResourceFromMap(map[string]resources.Quantity{"first": resources.Quantity(alloc.quantity)}), false, alloc.priority)
- app.AddAllocation(allocation)
+ assert.NilError(t, app.AddAllocation(allocation))
assert.Check(t, nodesMap[alloc.nodeID].TryAddAllocation(allocation), "TryAddAllocation failed for %s on %s", alloc.allocKey, alloc.nodeID)
assert.NilError(t, childQ.TryIncAllocatedResource(ask.GetAllocatedResource()))
@@ -2779,7 +2829,8 @@ func TestTryPreemption_PredicateVictimsNotTruncated(t *testing.T) {
assert.NilError(t, app1.AddAllocationAsk(ask1))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err1 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err1)
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask1.GetAllocatedResource()))
@@ -2789,7 +2840,8 @@ func TestTryPreemption_PredicateVictimsNotTruncated(t *testing.T) {
assert.NilError(t, app1.AddAllocationAsk(ask2))
alloc2 := newAllocationWithKey("alloc2", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}))
alloc2.createTime = ask2.createTime
- app1.AddAllocation(alloc2)
+ err2 := app1.AddAllocation(alloc2)
+ assert.NilError(t, err2)
assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask2.GetAllocatedResource()))
@@ -2843,7 +2895,8 @@ func TestTryPreemption_PrematureVictimLoopTermination(t *testing.T) {
assert.NilError(t, app1.AddAllocationAsk(ask2))
alloc2 := newAllocationWithKey("alloc2", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 2}))
alloc2.createTime = ask2.createTime
- app1.AddAllocation(alloc2)
+ err1 := app1.AddAllocation(alloc2)
+ assert.NilError(t, err1)
assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask2.GetAllocatedResource()))
@@ -2852,7 +2905,8 @@ func TestTryPreemption_PrematureVictimLoopTermination(t *testing.T) {
assert.NilError(t, app1.AddAllocationAsk(ask1))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 8}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ err2 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err2)
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask1.GetAllocatedResource()))
@@ -2887,7 +2941,7 @@ func TestTryPreemption_PrematureVictimLoopTermination(t *testing.T) {
// victim's resource size exceeds the ask queue's remaining guaranteed headroom.
func TestTryPreemption_DecoupleAskQueueQuotaFromVictimSize(t *testing.T) {
appQueueMapping := NewAppQueueMapping()
- node := newNode(nodeID1, map[string]resources.Quantity{"first": 8})
+ node := newNode(nodeID1, map[string]resources.Quantity{"first": 10})
iterator := getNodeIteratorFn(node)
rootQ, err := createRootQueue(map[string]string{"first": "20"})
assert.NilError(t, err)
@@ -2903,11 +2957,26 @@ func TestTryPreemption_DecoupleAskQueueQuotaFromVictimSize(t *testing.T) {
childQ1.AddApplication(app1)
appQueueMapping.AddAppQueueMapping(app1.ApplicationID, childQ1)
+ // alloc2 is older (-10s) and fits in askQueue's remaining guaranteed resource (2)
+ ask2 := newAllocationAsk("alloc2", appID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 2}))
+ ask2.createTime = time.Now().Add(-10 * time.Second)
+ assert.NilError(t, app1.AddAllocationAsk(ask2))
+ alloc2 := newAllocationWithKey("alloc2", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 2}))
+ alloc2.createTime = ask2.createTime
+ err1 := app1.AddAllocation(alloc2)
+ assert.NilError(t, err1)
+ assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
+ assert.NilError(t, childQ1.TryIncAllocatedResource(ask2.GetAllocatedResource()))
+
+ // alloc1 is newer (-5s), so it sorts before alloc2, but is too large (8) to fit in askQueue's remaining headroom (2)
// alloc1 is 8 cores on the node (the only victim on this node)
ask1 := newAllocationAsk("alloc1", appID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 8}))
+ ask1.createTime = time.Now().Add(-5 * time.Second)
assert.NilError(t, app1.AddAllocationAsk(ask1))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 8}))
- app1.AddAllocation(alloc1)
+ alloc1.createTime = ask1.createTime
+ err2 := app1.AddAllocation(alloc1)
+ assert.NilError(t, err2)
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask1.GetAllocatedResource()))
@@ -2990,7 +3059,7 @@ func TestTryPreemption_PredicateTailVictimsNotTruncated(t *testing.T) {
assert.NilError(t, app1.AddAllocationAsk(ask1))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1}))
alloc1.createTime = ask1.createTime
- app1.AddAllocation(alloc1)
+ assert.NilError(t, app1.AddAllocation(alloc1))
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask1.GetAllocatedResource()))
@@ -3000,7 +3069,7 @@ func TestTryPreemption_PredicateTailVictimsNotTruncated(t *testing.T) {
assert.NilError(t, app1.AddAllocationAsk(ask2))
alloc2 := newAllocationWithKey("alloc2", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1}))
alloc2.createTime = ask2.createTime
- app1.AddAllocation(alloc2)
+ assert.NilError(t, app1.AddAllocation(alloc2))
assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask2.GetAllocatedResource()))
@@ -3068,7 +3137,7 @@ func TestTryPreemption_DecoupleAskQueueQuotaFromVictimSize_AdditionalVictims(t *
ask1 := newAllocationAsk("alloc1", appID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 8}))
assert.NilError(t, app1.AddAllocationAsk(ask1))
alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 8}))
- app1.AddAllocation(alloc1)
+ assert.NilError(t, app1.AddAllocation(alloc1))
assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
assert.NilError(t, childQ1.TryIncAllocatedResource(ask1.GetAllocatedResource()))
diff --git a/pkg/scheduler/objects/preemption_utilities_test.go b/pkg/scheduler/objects/preemption_utilities_test.go
index 054c2f70..e41a9de3 100644
--- a/pkg/scheduler/objects/preemption_utilities_test.go
+++ b/pkg/scheduler/objects/preemption_utilities_test.go
@@ -137,7 +137,7 @@ func removeAllocationAsks(node *Node, asks []*Allocation) {
}
}
-func assignAllocationsToQueue(allocations []*Allocation, queue *Queue) {
+func assignAllocationsToQueue(t *testing.T, allocations []*Allocation, queue *Queue) {
for _, allocation := range allocations {
var app *Application
var ok bool
@@ -148,7 +148,8 @@ func assignAllocationsToQueue(allocations []*Allocation, queue *Queue) {
} else {
app = queue.applications[allocation.applicationID]
}
- app.AddAllocation(allocation)
+ err := app.AddAllocation(allocation)
+ assert.NilError(t, err)
queue.IncAllocatedResource(allocation.GetAllocatedResource(), false)
}
}
diff --git a/pkg/scheduler/objects/queue_test.go b/pkg/scheduler/objects/queue_test.go
index d1a0ca6f..31468b79 100644
--- a/pkg/scheduler/objects/queue_test.go
+++ b/pkg/scheduler/objects/queue_test.go
@@ -760,7 +760,8 @@ func TestSortAppsWithPlaceholderAllocations(t *testing.T) {
alloc := newAllocation(appID1, "node-0", res)
alloc.placeholder = true
// adding a placeholder allocation & pending request to "app1"
- app1.AddAllocation(alloc)
+ err = app1.AddAllocation(alloc)
+ assert.NilError(t, err)
err = app1.AddAllocationAsk(newAllocationAsk("ask-0", appID1, res))
assert.NilError(t, err, "could not add ask")
phApps := leaf.sortApplications(true)
@@ -769,7 +770,8 @@ func TestSortAppsWithPlaceholderAllocations(t *testing.T) {
// adding a placeholder allocation & pending request to "app2"
alloc2 := newAllocation(appID2, "node-1", res)
alloc2.placeholder = true
- app2.AddAllocation(alloc2)
+ err = app2.AddAllocation(alloc2)
+ assert.NilError(t, err)
err = app2.AddAllocationAsk(newAllocationAsk("ask-0", appID1, res))
assert.NilError(t, err, "could not add ask")
phApps = leaf.sortApplications(true)
@@ -2256,8 +2258,10 @@ func TestFindEligiblePreemptionVictims(t *testing.T) {
assert.NilError(t, err, "failed to add ask")
err = app2.AddAllocationAsk(ask3)
assert.NilError(t, err, "failed to add ask")
- app2.AddAllocation(alloc2)
- app2.AddAllocation(alloc3)
+ err = app2.AddAllocation(alloc2)
+ assert.NilError(t, err)
+ err = app2.AddAllocation(alloc3)
+ assert.NilError(t, err)
err = leaf2.TryIncAllocatedResource(alloc2.GetAllocatedResource())
assert.NilError(t, err, "failed to inc allocated resources")
err = leaf2.TryIncAllocatedResource(alloc3.GetAllocatedResource())
@@ -2364,7 +2368,8 @@ func TestFindEligiblePreemptionVictims(t *testing.T) {
// recreate alloc2 to restore non-prempted state
app2.RemoveAllocation(alloc2.GetAllocationKey(), si.TerminationType_STOPPED_BY_RM)
alloc2 = createAllocation("ask2", appID2, nodeID1, true, true, -1000, false, res)
- app2.AddAllocation(alloc2)
+ err = app2.AddAllocation(alloc2)
+ assert.NilError(t, err)
// setting priority offset on parent2 queue should remove leaf2 victims
parent2.priorityOffset = 1001
@@ -3362,7 +3367,8 @@ func TestQueueRunningAppsForSingleAllocationApp(t *testing.T) {
assert.NilError(t, err, "failed to add ask")
alloc := newAllocationWithKey("ask-1", appID1, nodeID1, res)
- app.AddAllocation(alloc)
+ err = app.AddAllocation(alloc)
+ assert.NilError(t, err)
assert.Equal(t, app.CurrentState(), Running.String(), "app state should be running")
assert.Equal(t, leaf.runningApps, uint64(1), "leaf should have 1 app running")
diff --git a/pkg/scheduler/objects/quota_preemptor_test.go b/pkg/scheduler/objects/quota_preemptor_test.go
index f3853a49..03f5db0d 100644
--- a/pkg/scheduler/objects/quota_preemptor_test.go
+++ b/pkg/scheduler/objects/quota_preemptor_test.go
@@ -112,7 +112,7 @@ func TestQuotaChangeFilterVictims(t *testing.T) {
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
asks := prepareAllocationAsks(t, node)
- assignAllocationsToQueue(asks, leaf)
+ assignAllocationsToQueue(t, asks, leaf)
if tc.irrelevantAllocations[0] {
asks[0].SetRequiredNode("node2")
asks[1].SetRequiredNode("node2")
@@ -227,7 +227,7 @@ func TestQuotaChangeTryPreemption(t *testing.T) {
leaf.maxResource = tc.oldMax
leaf.guaranteedResource = tc.guaranteed
asks := tc.victims
- assignAllocationsToQueue(asks, leaf)
+ assignAllocationsToQueue(t, asks, leaf)
leaf.maxResource = tc.newMax
leaf.guaranteedResource = tc.guaranteed
preemptor := NewQuotaPreemptor(tc.queue)
@@ -356,7 +356,7 @@ func TestQuotaChangeTryPreemptionWithDifferentResTypes(t *testing.T) {
leaf.maxResource = tc.oldMax
leaf.guaranteedResource = tc.guaranteed
asks := v.allocs
- assignAllocationsToQueue(asks, leaf)
+ assignAllocationsToQueue(t, asks, leaf)
leaf.maxResource = tc.newMax
leaf.guaranteedResource = tc.guaranteed
preemptor := NewQuotaPreemptor(tc.queue)
@@ -609,7 +609,7 @@ func TestQuotaChangeTryPreemptionForParentQueue(t *testing.T) {
t.Run(tc.name, func(t *testing.T) {
tc.queue.maxResource = tc.oldMax
for q, v := range tc.victims {
- assignAllocationsToQueue(v, q)
+ assignAllocationsToQueue(t, v, q)
}
tc.queue.maxResource = tc.newMax
tc.queue.guaranteedResource = tc.newMax
diff --git a/pkg/scheduler/partition.go b/pkg/scheduler/partition.go
index c162e530..209a3a0e 100644
--- a/pkg/scheduler/partition.go
+++ b/pkg/scheduler/partition.go
@@ -486,6 +486,140 @@ func (pc *PartitionContext) getRejectedApplication(appID string) *objects.Applic
return pc.rejectedApplications[appID]
}
+// getOrReviveApplication returns the application to add work to, bringing it back from the completed
+// list if the shim sent work for it after the core had already finished it. Returns nil when no
+// revivable application exists.
+// NOTE: this is a lock free call. It must NOT be called holding the PartitionContext lock.
+func (pc *PartitionContext) getOrReviveApplication(appID string) *objects.Application {
+ if app := pc.getApplication(appID); app != nil {
+ // moveTerminatedApp unsets the queue before it takes the partition lock, so an app can be
+ // visible here with no queue attached
+ if app.GetQueue() == nil && !pc.restoreAppQueue(app) {
+ return nil
+ }
+ return app
+ }
+ return pc.reviveCompletedApplication(appID)
+}
+
+// reviveCompletedApplication moves the most recently completed generation of appID back into the
+// active application list and re-attaches its queue.
+// NOTE: this is a lock free call. It must NOT be called holding the PartitionContext lock.
+func (pc *PartitionContext) reviveCompletedApplication(appID string) *objects.Application {
+ app := pc.takeCompletedApplication(appID)
+ if app == nil {
+ return nil
+ }
+ if !pc.restoreAppQueue(app) {
+ // no queue to run in: hand it back to the completed list so it is not lost
+ pc.moveTerminatedApp(appID)
+ return nil
+ }
+ log.Log(log.SchedPartition).Info("reviving completed application for new work from the shim",
+ zap.String("partitionName", pc.Name),
+ zap.String("appID", appID),
+ zap.String("queue", app.GetQueuePath()))
+ return app
+}
+
+// takeCompletedApplication removes the latest completed generation of appID from the completed list
+// and returns it. Expired applications are past the point of no return and are left alone.
+func (pc *PartitionContext) takeCompletedApplication(appID string) *objects.Application {
+ pc.Lock()
+ defer pc.Unlock()
+ // completed apps are re-keyed as appID + a negative unix timestamp, so the smallest suffix is
+ // the most recent generation. Compare keys to avoid taking an application lock under this one.
+ var latestKey string
+ latestStamp := int64(math.MaxInt64)
+ for key, completed := range pc.completedApplications {
+ if completed.ApplicationID != appID || completed.IsExpired() {
+ continue
+ }
+ stamp, err := strconv.ParseInt(strings.TrimPrefix(key, appID), 10, 64)
+ if err != nil {
+ continue
+ }
+ if stamp < latestStamp {
+ latestStamp = stamp
+ latestKey = key
+ }
+ }
+ if latestKey == "" {
+ return nil
+ }
+ app := pc.completedApplications[latestKey]
+ delete(pc.completedApplications, latestKey)
+ pc.applications[appID] = app
+ return app
+}
+
+// restoreAppQueue re-attaches a leaf queue to an application that lost it on completion, recreating
+// the queue when it was cleaned up in the meantime. Returns false when no queue can be resolved.
+// NOTE: this is a lock free call. It must NOT be called holding the PartitionContext lock.
+func (pc *PartitionContext) restoreAppQueue(app *objects.Application) bool {
+ queue := pc.resolveRevivedQueue(app)
+ if queue == nil {
+ return false
+ }
+ app.RestoreQueue(queue)
+ queue.AddApplication(app)
+ // A moveTerminatedApp running concurrently can file this app as completed while we revive it; put
+ // it back on the active list and drop that completed generation so the maps stay consistent.
+ pc.ensureAppActive(app)
+ return true
+}
+
+// ensureAppActive makes the partition tracking consistent for a freshly revived application: it puts
+// the application back on the active list and drops any completed generation that still points at the
+// same application, covering a moveTerminatedApp that filed it as completed while the revival was in
+// flight.
+func (pc *PartitionContext) ensureAppActive(app *objects.Application) {
+ pc.Lock()
+ defer pc.Unlock()
+ appID := app.ApplicationID
+ pc.applications[appID] = app
+ for key, completed := range pc.completedApplications {
+ if completed == app {
+ delete(pc.completedApplications, key)
+ }
+ }
+}
+
+// resolveRevivedQueue resolves (recreating when needed) the leaf queue for a reviving application and
+// records the app-to-queue mapping. Returns nil when no leaf queue can be resolved.
+// NOTE: this call takes the PartitionContext lock. It must NOT be called holding the lock.
+func (pc *PartitionContext) resolveRevivedQueue(app *objects.Application) *objects.Queue {
+ pc.Lock()
+ defer pc.Unlock()
+ queueName := app.GetQueuePath()
+ queue := pc.getQueueInternal(queueName)
+ if queue == nil {
+ var err error
+ if common.IsRecoveryQueue(queueName) {
+ queue, err = pc.createRecoveryQueue()
+ } else {
+ queue, err = pc.createQueue(queueName, app.GetUser())
+ }
+ if err != nil {
+ log.Log(log.SchedPartition).Warn("failed to recreate queue while reviving application",
+ zap.String("partitionName", pc.Name),
+ zap.String("appID", app.ApplicationID),
+ zap.String("queue", queueName),
+ zap.Error(err))
+ return nil
+ }
+ }
+ if !queue.IsLeafQueue() {
+ log.Log(log.SchedPartition).Warn("queue is no longer a leaf while reviving application",
+ zap.String("partitionName", pc.Name),
+ zap.String("appID", app.ApplicationID),
+ zap.String("queue", queueName))
+ return nil
+ }
+ pc.appQueueMapping.AddAppQueueMapping(app.ApplicationID, queue)
+ return queue
+}
+
// GetQueue returns queue from the structure based on the fully qualified name.
// Wrapper around the unlocked version getQueueInternal()
// Visible by tests
@@ -1196,13 +1330,12 @@ func (pc *PartitionContext) UpdateAllocation(alloc *objects.Allocation) (bool, b
return pc.handleForeignAllocation(allocationKey, applicationID, nodeID, node, alloc)
}
- // find application
- app := pc.getApplication(alloc.GetApplicationID())
+ // find application, reviving it if the shim sent work after the core completed it
+ app := pc.getOrReviveApplication(applicationID)
if app == nil {
metrics.GetSchedulerMetrics().IncSchedulingError()
return false, false, fmt.Errorf("failed to find application %s", applicationID)
}
- queue := app.GetQueue()
// find node if one is specified
allocated := alloc.IsAllocated()
@@ -1257,15 +1390,23 @@ func (pc *PartitionContext) UpdateAllocation(alloc *objects.Allocation) (bool, b
zap.String("appID", applicationID),
zap.String("allocationKey", allocationKey))
+ // pins the application in a runnable state before anything is charged to it
+ if err := app.RecoverAllocationAsk(alloc); err != nil {
+ metrics.GetSchedulerMetrics().IncSchedulingError()
+ return false, false, err
+ }
+ if err := app.AddAllocation(alloc); err != nil {
+ metrics.GetSchedulerMetrics().IncSchedulingError()
+ return false, false, err
+ }
+
// Increase the queue resource usage at any cost even if allocation put up the usage over the max resources.
// Quota preemption delay (only if set) used before restart cannot be adjusted based on the lost time after restart.
// So, override the quota preemption time with the configured delay again so that preemption would be triggerred once the delay expires.
- queue.IncAllocatedResource(res, pc.IsQuotaPreemptionEnabled())
- metrics.GetQueueMetrics(queue.GetQueuePath()).IncAllocatedContainer()
+ app.GetQueue().IncAllocatedResource(res, pc.IsQuotaPreemptionEnabled())
+ metrics.GetQueueMetrics(app.GetQueuePath()).IncAllocatedContainer()
node.AddAllocation(alloc)
alloc.SetInstanceType(node.GetInstanceType())
- app.RecoverAllocationAsk(alloc)
- app.AddAllocation(alloc)
pc.updateAllocationCount(1)
if alloc.IsPlaceholder() {
pc.incPhAllocationCount()
@@ -1328,16 +1469,19 @@ func (pc *PartitionContext) UpdateAllocation(alloc *objects.Allocation) (bool, b
zap.Error(err))
return false, false, err
}
+ if err := app.AddAllocation(existing); err != nil {
+ metrics.GetSchedulerMetrics().IncSchedulingError()
+ return false, false, err
+ }
// Increase the queue resource usage at any cost even if ask accommodated earlier based on the old max resources causes usage overflow based on the current max resources now.
// In case quota preemption set but not completed, usage would be brought down as part of enforcement through preemption when the already set delay expires.
// In case quota preemption set and completed already, configured delay would be set again and
// usage would be brought down when the newly set delay expires.
- queue.IncAllocatedResource(alloc.GetAllocatedResource(), pc.IsQuotaPreemptionEnabled())
- metrics.GetQueueMetrics(queue.GetQueuePath()).IncAllocatedContainer()
+ app.GetQueue().IncAllocatedResource(alloc.GetAllocatedResource(), pc.IsQuotaPreemptionEnabled())
+ metrics.GetQueueMetrics(app.GetQueuePath()).IncAllocatedContainer()
node.AddAllocation(existing)
existing.SetInstanceType(node.GetInstanceType())
- app.AddAllocation(existing)
pc.updateAllocationCount(1)
if existing.IsPlaceholder() {
pc.incPhAllocationCount()
@@ -1757,7 +1901,19 @@ func (pc *PartitionContext) moveTerminatedApp(appID string) {
zap.String("appID", appID))
return
}
- app.UnSetQueue()
+ // enter_Completed dispatches this on its own goroutine, so the app can have been revived in the
+ // meantime: moving it now would strip the queue off a running application. Detach the queue and
+ // decide to complete in a single step under the application lock so it cannot race the revival.
+ queue, ok := app.DetachQueueForCompletion()
+ if !ok {
+ log.Log(log.SchedPartition).Info("Application was revived before cleanup, not removing it",
+ zap.String("appID", appID),
+ zap.String("app status", app.CurrentState()))
+ return
+ }
+ if queue != nil {
+ queue.RemoveApplication(app)
+ }
// new ID as completedApplications map key, use negative value to get a divider
newID := appID + strconv.FormatInt(-(time.Now()).Unix(), 10)
log.Log(log.SchedPartition).Info("Removing terminated application from the application list",
@@ -1766,6 +1922,15 @@ func (pc *PartitionContext) moveTerminatedApp(appID string) {
app.LogAppSummary(pc.RmID)
pc.Lock()
defer pc.Unlock()
+ // A revival can start the moment the queue is detached above (the shim sees an app with no queue)
+ // and re-attach a queue through restoreAppQueue/ensureAppActive. The revived flag is read lock
+ // free, so it is safe to check while holding the partition lock: leave a revived app active and
+ // let the revival own the active-list entry.
+ if app.IsRevived() {
+ log.Log(log.SchedPartition).Info("Application revived during cleanup, leaving it on the active list",
+ zap.String("appID", appID))
+ return
+ }
delete(pc.applications, appID)
pc.completedApplications[newID] = app
}
diff --git a/pkg/scheduler/partition_revive_test.go b/pkg/scheduler/partition_revive_test.go
new file mode 100644
index 00000000..504c54cb
--- /dev/null
+++ b/pkg/scheduler/partition_revive_test.go
@@ -0,0 +1,342 @@
+/*
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+package scheduler
+
+import (
+ "testing"
+
+ "gotest.tools/v3/assert"
+
+ "github.com/apache/yunikorn-core/pkg/common"
+ "github.com/apache/yunikorn-core/pkg/common/resources"
+ "github.com/apache/yunikorn-core/pkg/scheduler/objects"
+)
+
+// completedPartitionApp adds an application and drives it through moveTerminatedApp, so it is off
+// pc.applications and has no queue attached, exactly as enter_Completed leaves it.
+func completedPartitionApp(t *testing.T, partition *PartitionContext) *objects.Application {
+ t.Helper()
+ app := newApplication(appID1, "default", defQueue)
+ assert.NilError(t, partition.AddApplication(app), "add application failed")
+ app.SetState(objects.Completed.String())
+ // this is what enter_Completed dispatches on its own goroutine
+ partition.moveTerminatedApp(appID1)
+
+ assert.Assert(t, partition.getApplication(appID1) == nil, "app should be off the active list")
+ assert.Assert(t, app.GetQueue() == nil, "app should have lost its queue")
+ assert.Equal(t, len(partition.GetCompletedApplications()), 1)
+ return app
+}
+
+func TestUpdateAllocationRevivesCompletedApplication(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ completedPartitionApp(t, partition)
+
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5, "vcores": 5})
+ askCreated, _, err := partition.UpdateAllocation(newAllocationAsk("late-ask", appID1, res))
+ assert.NilError(t, err, "late ask should revive the completed application")
+ assert.Assert(t, askCreated, "a new request should have been created")
+
+ revived := partition.getApplication(appID1)
+ assert.Assert(t, revived != nil, "app should be back on the active list")
+ assert.Equal(t, revived.CurrentState(), objects.Running.String())
+ assert.Equal(t, len(partition.GetCompletedApplications()), 0, "app should be off the completed list")
+
+ queue := revived.GetQueue()
+ assert.Assert(t, queue != nil, "queue should be re-attached")
+ assert.Equal(t, queue.QueuePath, defQueue)
+ assert.Assert(t, resources.Equals(queue.GetPendingResource(), res), "queue must track the new ask")
+}
+
+func TestUpdateAllocationRevivesCompletedApplicationWithNode(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ completedPartitionApp(t, partition)
+
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5, "vcores": 5})
+ node := newNodeMaxResource(nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 100, "vcores": 100}))
+ assert.NilError(t, partition.AddNode(node), "add node failed")
+
+ _, allocCreated, err := partition.UpdateAllocation(newAllocation("late-alloc", appID1, nodeID1, res))
+ assert.NilError(t, err, "allocation should revive the completed application")
+ assert.Assert(t, allocCreated, "an allocation should have been created")
+
+ revived := partition.getApplication(appID1)
+ assert.Assert(t, revived != nil, "app should be back on the active list")
+ assert.Equal(t, revived.CurrentState(), objects.Running.String())
+ assert.Equal(t, partition.GetTotalAllocationCount(), 1)
+ assert.Assert(t, resources.Equals(revived.GetQueue().GetAllocatedResource(), res), "queue must track the allocation")
+ assert.Assert(t, node.GetAllocation("late-alloc") != nil, "node must hold the allocation")
+}
+
+// The app is still on the active list but moveTerminatedApp already unset its queue: the revival has
+// to re-attach one rather than charge usage to a nil queue.
+func TestUpdateAllocationRestoresQueueForUnqueuedApp(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ app := newApplication(appID1, "default", defQueue)
+ assert.NilError(t, partition.AddApplication(app), "add application failed")
+ app.SetState(objects.Completed.String())
+ app.UnSetQueue()
+ assert.Assert(t, app.GetQueue() == nil, "app should have lost its queue")
+ assert.Assert(t, partition.getApplication(appID1) != nil, "app is still on the active list")
+
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5, "vcores": 5})
+ _, _, err = partition.UpdateAllocation(newAllocationAsk("late-ask", appID1, res))
+ assert.NilError(t, err, "late ask should be accepted")
+
+ assert.Assert(t, app.GetQueue() != nil, "queue should be re-attached")
+ assert.Equal(t, app.CurrentState(), objects.Running.String())
+ assert.Assert(t, resources.Equals(app.GetQueue().GetPendingResource(), res), "queue must track the new ask")
+}
+
+func TestUpdateAllocationUnknownAppStillFails(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5, "vcores": 5})
+ _, _, err = partition.UpdateAllocation(newAllocationAsk("ask", "no-such-app", res))
+ assert.ErrorContains(t, err, "failed to find application", "an unknown app must still be rejected")
+}
+
+// An expired application is past the point of no return and must not be revived.
+func TestUpdateAllocationDoesNotReviveExpiredApplication(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ app := newApplication(appID1, "default", defQueue)
+ assert.NilError(t, partition.AddApplication(app), "add application failed")
+ app.SetState(objects.Expired.String())
+ partition.moveTerminatedApp(appID1)
+
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5, "vcores": 5})
+ _, _, err = partition.UpdateAllocation(newAllocationAsk("late-ask", appID1, res))
+ assert.ErrorContains(t, err, "failed to find application", "expired apps must not be revived")
+ assert.Equal(t, len(partition.GetCompletedApplications()), 1, "expired app should stay on the completed list")
+}
+
+// enter_Completed dispatches moveTerminatedApp on its own goroutine. If the app is revived before
+// that goroutine runs, the cleanup must not strip the queue off the now-running application.
+func TestMoveTerminatedAppSkipsRevivedApplication(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ app := newApplication(appID1, "default", defQueue)
+ assert.NilError(t, partition.AddApplication(app), "add application failed")
+ app.SetState(objects.Completed.String())
+
+ // the shim wins the race and revives the app before the cleanup callback runs
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5, "vcores": 5})
+ _, _, err = partition.UpdateAllocation(newAllocationAsk("late-ask", appID1, res))
+ assert.NilError(t, err, "late ask should revive the app")
+ assert.Equal(t, app.CurrentState(), objects.Running.String())
+
+ // the in-flight cleanup now lands
+ partition.moveTerminatedApp(appID1)
+
+ assert.Assert(t, partition.getApplication(appID1) == app, "revived app must stay on the active list")
+ assert.Assert(t, app.GetQueue() != nil, "revived app must keep its queue")
+ assert.Equal(t, len(partition.GetCompletedApplications()), 0, "revived app must not be filed as completed")
+}
+
+// Only the most recent completed generation is revived. The completed list is keyed by
+// appID + a negative unix timestamp, so it is seeded directly here: moveTerminatedApp would give
+// two completions in the same second the same key and one would overwrite the other.
+func TestReviveTakesLatestCompletedGeneration(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+
+ older := newApplication(appID1, "default", defQueue)
+ older.SetState(objects.Completed.String())
+ newer := newApplication(appID1, "default", defQueue)
+ newer.SetState(objects.Completed.String())
+
+ partition.Lock()
+ partition.completedApplications[appID1+"-100"] = older
+ partition.completedApplications[appID1+"-200"] = newer
+ partition.Unlock()
+
+ taken := partition.takeCompletedApplication(appID1)
+ assert.Assert(t, taken == newer, "the most recent generation should be taken")
+ assert.Equal(t, len(partition.GetCompletedApplications()), 1, "the older generation should be left alone")
+ assert.Assert(t, partition.getApplication(appID1) == newer, "the taken generation should be on the active list")
+}
+
+// moveTerminatedApp must not tear down an application that a revival already re-attached a queue to,
+// even while the application still shows a terminal state. RestoreQueue raises the revived flag
+// together with the queue: this is exactly the narrow window between the cleanup goroutine passing
+// its terminal-state check and actually ripping the queue off.
+func TestMoveTerminatedAppSkipsAppRevivedMidFlight(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ app := newApplication(appID1, "default", defQueue)
+ assert.NilError(t, partition.AddApplication(app), "add application failed")
+ queue := app.GetQueue()
+ app.SetState(objects.Completed.String())
+
+ // a revival re-attaches the queue and raises the revived flag; the app has not left Completed yet
+ app.RestoreQueue(queue)
+ assert.Assert(t, app.IsRevived(), "revived flag should be set")
+
+ // the in-flight cleanup now lands
+ partition.moveTerminatedApp(appID1)
+
+ assert.Assert(t, partition.getApplication(appID1) == app, "revived app must stay on the active list")
+ assert.Assert(t, app.GetQueue() != nil, "revived app must keep its queue")
+ assert.Equal(t, len(partition.GetCompletedApplications()), 0, "revived app must not be filed as completed")
+}
+
+// The cleanup can win the race and file the app as completed before the revival restores its queue.
+// restoreAppQueue has to reclaim it: put it back on the active list and drop the completed generation.
+func TestRestoreAppQueueReclaimsCompletedApp(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ app := newApplication(appID1, "default", defQueue)
+ assert.NilError(t, partition.AddApplication(app), "add application failed")
+ app.SetState(objects.Completed.String())
+
+ // cleanup wins first: the app is filed as completed and loses its queue
+ partition.moveTerminatedApp(appID1)
+ assert.Assert(t, partition.getApplication(appID1) == nil, "app should be off the active list")
+ assert.Assert(t, app.GetQueue() == nil, "app should have lost its queue")
+ assert.Equal(t, len(partition.GetCompletedApplications()), 1)
+
+ // the revival that still holds the app pointer now restores its queue
+ assert.Assert(t, partition.restoreAppQueue(app), "queue should be restored")
+
+ assert.Assert(t, partition.getApplication(appID1) == app, "app must be back on the active list")
+ assert.Assert(t, app.GetQueue() != nil, "queue must be re-attached")
+ assert.Assert(t, app.IsRevived(), "revived flag should be set")
+ assert.Equal(t, len(partition.GetCompletedApplications()), 0, "the completed generation must be dropped")
+}
+
+// The leaf queue can be cleaned up while the app sits on the completed list. resolveRevivedQueue must
+// recreate it dynamically so the revived app has somewhere to run.
+func TestResolveRevivedQueueRecreatesDynamicQueue(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ app := newApplication(appID1, "default", "root.recreated")
+
+ queue := partition.resolveRevivedQueue(app)
+ assert.Assert(t, queue != nil, "a dynamic leaf queue should be recreated")
+ assert.Equal(t, queue.QueuePath, "root.recreated")
+ assert.Assert(t, queue.IsLeafQueue(), "recreated queue must be a leaf")
+}
+
+// A revived application that ran in the recovery queue must have it recreated through the recovery
+// queue path rather than the regular dynamic queue path.
+func TestResolveRevivedQueueRecreatesRecoveryQueue(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ app := newApplication(appID1, "default", common.RecoveryQueueFull)
+
+ queue := partition.resolveRevivedQueue(app)
+ assert.Assert(t, queue != nil, "the recovery queue should be recreated")
+ assert.Assert(t, queue.IsLeafQueue(), "recovery queue must be a leaf")
+}
+
+// The queue can no longer be recreated (its parent is now a leaf): resolveRevivedQueue must give up.
+func TestResolveRevivedQueueCreateFails(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ // root.default is a leaf, so a child under it cannot be created
+ app := newApplication(appID1, "default", "root.default.sub")
+
+ queue := partition.resolveRevivedQueue(app)
+ assert.Assert(t, queue == nil, "queue under a leaf parent must not be created")
+}
+
+// The queue path now resolves to a non-leaf queue: the app cannot run there so revival must fail.
+func TestResolveRevivedQueueRejectsNonLeaf(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ app := newApplication(appID1, "default", "root")
+
+ queue := partition.resolveRevivedQueue(app)
+ assert.Assert(t, queue == nil, "a non-leaf queue must be rejected")
+}
+
+// restoreAppQueue returns false when no leaf queue can be resolved.
+func TestRestoreAppQueueFailsWithoutLeaf(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ app := newApplication(appID1, "default", "root")
+
+ assert.Assert(t, !partition.restoreAppQueue(app), "restore must fail when no leaf queue resolves")
+}
+
+// getOrReviveApplication returns nil when an active app lost its queue and it cannot be restored.
+func TestGetOrReviveApplicationRestoreFails(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ // non-leaf queue path means the queue cannot be restored
+ app := newApplication(appID1, "default", "root")
+ partition.Lock()
+ partition.applications[appID1] = app
+ partition.Unlock()
+ assert.Assert(t, app.GetQueue() == nil, "app must have no queue attached")
+
+ assert.Assert(t, partition.getOrReviveApplication(appID1) == nil, "revival must fail when the queue cannot be restored")
+}
+
+// reviveCompletedApplication hands the app back to the completed list when its queue cannot be restored.
+func TestReviveCompletedApplicationQueueGoneReturnsToCompleted(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ // non-leaf queue path means the queue cannot be restored
+ app := newApplication(appID1, "default", "root")
+ app.SetState(objects.Completed.String())
+ partition.Lock()
+ partition.completedApplications[appID1+"-100"] = app
+ partition.Unlock()
+
+ assert.Assert(t, partition.reviveCompletedApplication(appID1) == nil, "revival must fail when the queue cannot be restored")
+ assert.Assert(t, partition.getApplication(appID1) == nil, "app must not stay on the active list")
+ assert.Equal(t, len(partition.GetCompletedApplications()), 1, "app must be handed back to the completed list")
+}
+
+// Completed entries whose key suffix is not a timestamp are ignored rather than crashing the scan.
+func TestTakeCompletedApplicationSkipsUnparsableKey(t *testing.T) {
+ setupUGM()
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ app := newApplication(appID1, "default", defQueue)
+ app.SetState(objects.Completed.String())
+ partition.Lock()
+ partition.completedApplications[appID1+"notanumber"] = app
+ partition.Unlock()
+
+ assert.Assert(t, partition.takeCompletedApplication(appID1) == nil, "an unparsable key must be skipped")
+ assert.Equal(t, len(partition.GetCompletedApplications()), 1, "the skipped entry must be left alone")
+}
diff --git a/pkg/scheduler/partition_test.go b/pkg/scheduler/partition_test.go
index 40dcfbd1..a63b61b7 100644
--- a/pkg/scheduler/partition_test.go
+++ b/pkg/scheduler/partition_test.go
@@ -5446,7 +5446,8 @@ func TestAppSchedulingOrderFIFO(t *testing.T) {
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1}), 0, false, map[string]string{
siCommon.CreationTime: "30", // older than app2Ask2 (alloc-3)
})
- app1.AddAllocation(allocExisting)
+ err = app1.AddAllocation(allocExisting)
+ assert.NilError(t, err)
// should schedule from app-1
alloc = partition.tryAllocate()
diff --git a/pkg/webservice/handlers_test.go b/pkg/webservice/handlers_test.go
index d8a5f183..59b5f232 100644
--- a/pkg/webservice/handlers_test.go
+++ b/pkg/webservice/handlers_test.go
@@ -115,35 +115,35 @@ partitions:
`
const configMultiPartitions = `
-partitions:
+partitions:
- name: gpu
usergroupresolver:
type: test
preemption:
enabled: false
- queues:
+ queues:
- name: root
- name: default
usergroupresolver:
type: ""
nodesortpolicy:
type: fair
- queues:
+ queues:
- name: root
- queues:
+ queues:
- name: default
submitacl: "*"
`
const configTwoLevelQueues = `
-partitions:
+partitions:
- name: gpu
- queues:
+ queues:
- name: root
- name: default
- nodesortpolicy:
+ nodesortpolicy:
type: binpacking
- queues:
+ queues:
- name: root
properties:
application.sort.policy: fifo
@@ -156,24 +156,24 @@ partitions:
memory: 400000
max:
memory: 600000
- queues:
+ queues:
- name: a
- queues:
+ queues:
- name: a1
properties:
application.sort.policy: fifo
- resources:
- guaranteed:
+ resources:
+ guaranteed:
memory: 500000
vcore: 50000
- max:
+ max:
memory: 800000
vcore: 80000
- resources:
- guaranteed:
+ resources:
+ guaranteed:
memory: 500000
vcore: 50000
- max:
+ max:
memory: 800000
vcore: 80000
`
@@ -992,10 +992,11 @@ func TestPartitions(t *testing.T) { //nolint:funlen
assert.Equal(t, cs["default"].Applications["total"], 7)
assert.Equal(t, cs["default"].Applications[objects.New.String()], 1)
assert.Equal(t, cs["default"].Applications[objects.Accepted.String()], 1)
- assert.Equal(t, cs["default"].Applications[objects.Running.String()], 1)
+ // app-5 was Completed and then received alloc-1, which revives it into Running
+ assert.Equal(t, cs["default"].Applications[objects.Running.String()], 2)
assert.Equal(t, cs["default"].Applications[objects.Completing.String()], 1)
assert.Equal(t, cs["default"].Applications[objects.Rejected.String()], 1)
- assert.Equal(t, cs["default"].Applications[objects.Completed.String()], 1)
+ assert.Equal(t, cs["default"].Applications[objects.Completed.String()], 0)
assert.Equal(t, cs["default"].Applications[objects.Failed.String()], 1)
assert.DeepEqual(t, cs["default"].Capacity.Capacity, map[string]int64{"memory": 1000, "vcore": 1000})
assert.DeepEqual(t, cs["default"].Capacity.UsedCapacity, map[string]int64{"memory": 300, "vcore": 700})
@@ -2840,7 +2841,8 @@ func prepareUserAndGroupContext(t *testing.T, config string) {
// add an alloc
allocInfo := newAlloc(ask.GetAllocationKey(), ask.GetApplicationID(), "node-1", ask.GetAllocatedResource())
- app.AddAllocation(allocInfo)
+ err = app.AddAllocation(allocInfo)
+ assert.NilError(t, err)
assert.Assert(t, app.IsRunning(), "Application did not return running state after alloc: %s", app.CurrentState())
NewWebApp(schedulerContext.Load(), nil)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@yunikorn.apache.org
For additional commands, e-mail: issues-help@yunikorn.apache.org