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-k8shim.git
The following commit(s) were added to refs/heads/master by this push:
new f4ee41b1 [YUNIKORN-2884] handling volume and pod binding failure
retries (#1062)
f4ee41b1 is described below
commit f4ee41b134454da421731e22a3f4af4c0360c55c
Author: Aditya Maheshwari <[email protected]>
AuthorDate: Wed Aug 19 14:59:13 2026 +1000
[YUNIKORN-2884] handling volume and pod binding failure retries (#1062)
Handle volume bind and bind failures on the k8shim side and retry when
logical::
* volume bind timeout fails directly without retry
* volume API call failures are retried with backoff
* bind API call failures are retried with backoff
When retries are exhausted revert the pod to pending in the k8shim and
core.
Closes: #1062
Signed-off-by: Wilfred Spiegelenburg <[email protected]>
---
go.mod | 3 +-
go.sum | 6 +-
pkg/cache/task.go | 139 ++++++++++++++++-----
pkg/cache/task_test.go | 230 +++++++++++++++++++++++++++++++++++
pkg/common/test/volumebinder_mock.go | 11 ++
pkg/shim/scheduler_test.go | 19 +--
6 files changed, 365 insertions(+), 43 deletions(-)
diff --git a/go.mod b/go.mod
index 349efd5e..6085beec 100644
--- a/go.mod
+++ b/go.mod
@@ -21,7 +21,7 @@ module github.com/apache/yunikorn-k8shim
go 1.26.0
require (
- github.com/apache/yunikorn-core v0.0.0-20260728123950-900bd603df82
+ github.com/apache/yunikorn-core v0.0.0-20260810093905-7dc1287c839f
github.com/apache/yunikorn-scheduler-interface
v0.0.0-20260727092410-674338955bdf
github.com/google/go-cmp v0.7.0
github.com/google/uuid v1.6.0
@@ -92,6 +92,7 @@ require (
github.com/prometheus/procfs v0.19.2 // indirect
github.com/spf13/cobra v1.10.2 // indirect
github.com/spf13/pflag v1.0.9 // indirect
+ github.com/tidwall/btree v1.8.1 // indirect
github.com/x448/float16 v0.8.4 // indirect
github.com/xlab/treeprint v1.2.0 // indirect
go.opentelemetry.io/otel v1.43.0 // indirect
diff --git a/go.sum b/go.sum
index 974ceb67..a999efb9 100644
--- a/go.sum
+++ b/go.sum
@@ -10,8 +10,8 @@ github.com/alexbrainman/sspi
v0.0.0-20250919150558-7d374ff0d59e h1:4dAU9FXIyQktp
github.com/alexbrainman/sspi v0.0.0-20250919150558-7d374ff0d59e/go.mod
h1:cEWa1LVoE5KvSD9ONXsZrj0z6KqySlCCNKHlLzbqAt4=
github.com/antlr4-go/antlr/v4 v4.13.1
h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ=
github.com/antlr4-go/antlr/v4 v4.13.1/go.mod
h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw=
-github.com/apache/yunikorn-core v0.0.0-20260728123950-900bd603df82
h1:uvaD7WchXd9nLkXArGjdul1OQfxXO+ukvI0VJJ0hAwk=
-github.com/apache/yunikorn-core v0.0.0-20260728123950-900bd603df82/go.mod
h1:Z/8pYK5VeDk9Q74vTCm3FnCVm1DtciaGYjl5/P7kO+M=
+github.com/apache/yunikorn-core v0.0.0-20260810093905-7dc1287c839f
h1:Rxkc5C4zNgdZOnYep0j60wb9XUdDrFsdvLCuDOZ4atg=
+github.com/apache/yunikorn-core v0.0.0-20260810093905-7dc1287c839f/go.mod
h1:431ZW8b1pAkJinnDmwpmsywHKfxikvSh5PX4DhWY4zo=
github.com/apache/yunikorn-scheduler-interface
v0.0.0-20260727092410-674338955bdf
h1:IXEpAeqZgCXODJtvB6Ib5CCcpRyvF0mqcicz80rA3Cc=
github.com/apache/yunikorn-scheduler-interface
v0.0.0-20260727092410-674338955bdf/go.mod
h1:qb739Bdm82PH7gsfEYabulGF90xKGNQ1hWmf197rDfw=
github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5
h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio=
@@ -187,6 +187,8 @@ github.com/stretchr/testify v1.8.0/go.mod
h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
github.com/stretchr/testify v1.8.1/go.mod
h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
github.com/stretchr/testify v1.11.1
h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
github.com/stretchr/testify v1.11.1/go.mod
h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+github.com/tidwall/btree v1.8.1 h1:27ehoXvm5AG/g+1VxLS1SD3vRhp/H7LuEfwNvddEdmA=
+github.com/tidwall/btree v1.8.1/go.mod
h1:jBbTdUWhSZClZWoDg54VnvV7/54modSOzDN7VXftj1A=
github.com/tidwall/gjson v1.18.0
h1:FIDeeyB800efLX89e5a8Y0BNH+LOngJyGrIWxG2FKQY=
github.com/tidwall/gjson v1.18.0/go.mod
h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
diff --git a/pkg/cache/task.go b/pkg/cache/task.go
index 795648e3..2519b69e 100644
--- a/pkg/cache/task.go
+++ b/pkg/cache/task.go
@@ -27,6 +27,8 @@ import (
"github.com/looplab/fsm"
"go.uber.org/zap"
v1 "k8s.io/api/core/v1"
+ "k8s.io/apimachinery/pkg/util/wait"
+ "k8s.io/client-go/util/retry"
podutil "k8s.io/kubernetes/pkg/api/v1/pod"
"github.com/apache/yunikorn-k8shim/pkg/common"
@@ -39,6 +41,15 @@ import (
"github.com/apache/yunikorn-scheduler-interface/lib/go/si"
)
+// retryBackoff controls the retry schedule for binding pod volumes and
binding the
+// pod to a node. Declared as a package variable so tests can shorten it.
+var retryBackoff = wait.Backoff{
+ Steps: 8, // total time is 1 + 2 + 4 + 8 + 16 + 30 + 30 + 30 = 121
seconds
+ Duration: time.Second,
+ Factor: 2,
+ Cap: 30 * time.Second,
+}
+
type Task struct {
taskID string
alias string
@@ -343,46 +354,78 @@ func (task *Task) postTaskPending() {
// This routine binds the pod to the allocated node.
// It calls K8s api to bind a pod to the assigned node, this may need some
time,
// so we do a delay binding, background process, to avoid blocking main
process.
-// The result of the binding is tracked and failures are properly handled.
-// If successful, we move task to next state BOUND, otherwise we fail the task
+// Volume binding and pod binding are retried with a backoff; if they
ultimately fail
+// the allocation is rolled back to a pending ask so the core can re-schedule
the task
+// on a different node. On success we move the task to the next state BOUND.
func (task *Task) postTaskAllocated() {
- go func() {
- // we need to obtain task's lock first,
- // this ensures no other threads modifying task state at the
time being
- task.lock.Lock()
- defer task.lock.Unlock()
-
- // post a message to indicate the pod gets its allocation
- events.GetRecorder().Eventf(task.pod.DeepCopy(),
- nil, v1.EventTypeNormal, "Scheduled", "Scheduled",
- "Successfully assigned %s to node %s", task.alias,
task.nodeName)
+ // Snapshot the fields needed for binding before launching the
goroutine without re-acquiring the lock.
+ // We already hold task.lock (via task.handle()) during state
transitions.
+ pod := task.pod
+ alias := task.alias
+ nodeName := task.nodeName
+ allocationKey := task.allocationKey
+ go func(pod *v1.Pod, alias, nodeName, allocationKey string) {
+ // this lock is to protect the task from being modified while
we are binding the pod to node
+ // once all task related operations are done, release the lock
+ // This is important specially while calling
rollbackAllocation, which needs to acquire the context lock, so we cannot hold
the task lock while calling it.
+ task.lock.Lock()
// before binding pod to node, first bind volumes to pod
log.Log(log.ShimCacheTask).Debug("bind pod volumes",
- zap.String("podName", task.pod.Name),
- zap.String("podUID", string(task.pod.UID)))
- if err := task.context.bindPodVolumes(task.pod); err != nil {
- log.Log(log.ShimCacheTask).Error("bind volumes to pod
failed", zap.String("taskID", task.taskID), zap.Error(err))
- task.failWithEvent(fmt.Sprintf("bind volumes to pod
failed, name: %s, %s", task.alias, err.Error()), "PodVolumesBindFailure")
+ zap.String("podName", pod.Name),
+ zap.String("podUID", string(pod.UID)))
+ if err := retry.OnError(retryBackoff, func(err error) bool {
+ if strings.HasPrefix(err.Error(), "binding volumes:") {
+ log.Log(log.ShimCacheTask).Warn("bind volumes
to pod failed due to volume binding error, stopping retries",
+ zap.String("taskID", task.taskID),
zap.Error(err))
+ return false
+ }
+ log.Log(log.ShimCacheTask).Error("bind volumes to pod
failed, retrying",
+ zap.String("taskID", task.taskID),
zap.Error(err))
+ return true
+ }, func() error {
+ return task.context.bindPodVolumes(pod)
+ }); err != nil {
+ log.Log(log.ShimCacheTask).Error("bind volumes to pod
failed after retries",
+ zap.String("taskID", task.taskID),
zap.Error(err))
+ // release task lock before calling
rescheduleOnBindFailure to avoid deadlock
+ task.lock.Unlock()
+ task.rescheduleOnBindFailure(allocationKey, nodeName,
"PodVolumesBindFailure",
+ fmt.Sprintf("Failed to bind volumes for %s on
node %s, it will be retried", alias, nodeName))
return
}
log.Log(log.ShimCacheTask).Debug("bind pod",
- zap.String("podName", task.pod.Name),
- zap.String("podUID", string(task.pod.UID)))
+ zap.String("podName", pod.Name),
+ zap.String("podUID", string(pod.UID)))
- if err :=
task.context.apiProvider.GetAPIs().KubeClient.Bind(task.pod, task.nodeName);
err != nil {
- log.Log(log.ShimCacheTask).Error("bind pod to node
failed", zap.String("taskID", task.taskID), zap.Error(err))
- task.failWithEvent(fmt.Sprintf("bind pod to node
failed, name: %s, %s", task.alias, err.Error()), "PodBindFailure")
+ if err := retry.OnError(retryBackoff, func(err error) bool {
+ log.Log(log.ShimCacheTask).Error("bind pod to node
failed, retrying",
+ zap.String("taskID", task.taskID),
zap.Error(err))
+ return true
+ }, func() error {
+ return
task.context.apiProvider.GetAPIs().KubeClient.Bind(pod, nodeName)
+ }); err != nil {
+ log.Log(log.ShimCacheTask).Error("bind pod to node
failed after retries",
+ zap.String("taskID", task.taskID),
zap.Error(err))
+ task.lock.Unlock()
+ task.rescheduleOnBindFailure(allocationKey, nodeName,
"PodBindFailure",
+ fmt.Sprintf("Failed to bind %s to node %s, it
will be retried", alias, nodeName))
return
}
- log.Log(log.ShimCacheTask).Info("successfully bound pod",
zap.String("podName", task.pod.Name))
- dispatcher.Dispatch(NewBindTaskEvent(task.applicationID,
task.taskID))
- events.GetRecorder().Eventf(task.pod.DeepCopy(), nil,
- v1.EventTypeNormal, "PodBindSuccessful",
"PodBindSuccessful",
- "Pod %s is successfully bound to node %s", task.alias,
task.nodeName)
+ // post a message to indicate the pod gets its allocation
+ events.GetRecorder().Eventf(pod.DeepCopy(),
+ nil, v1.EventTypeNormal, "Scheduled", "Scheduled",
+ "Successfully assigned %s to node %s", alias, nodeName)
+ log.Log(log.ShimCacheTask).Info("successfully bound pod",
zap.String("podName", pod.Name))
task.schedulingState = TaskSchedAllocated
- }()
+
+ dispatcher.Dispatch(NewBindTaskEvent(task.applicationID,
task.taskID))
+ task.lock.Unlock()
+ events.GetRecorder().Eventf(pod.DeepCopy(), nil,
+ v1.EventTypeNormal, "PodBindSuccessful",
"PodBindSuccessful",
+ "Pod %s is successfully bound to node %s", alias,
nodeName)
+ }(pod, alias, nodeName, allocationKey)
}
// beforeTaskAllocated is called before handling the TaskAllocated event.
@@ -627,21 +670,26 @@ func (task *Task) failWithEvent(errorMessage,
actionReason string) {
// pending ask so it can be re-scheduled on a different node.
// Must be called without holding the task lock.
func (task *Task) rollbackOnAssumePodFailure(allocationKey, nodeID string) {
- // Read fields needed for event posting and release request.
- // Clear stale node assignment under write lock so the task is clean
for the next allocation.
task.lock.Lock()
- podCopy := task.pod.DeepCopy()
alias := task.alias
+ podCopy := task.pod.DeepCopy()
appID := task.applicationID
partition := task.application.partition
task.allocationKey = ""
task.nodeName = ""
task.lock.Unlock()
+ task.rollbackAllocation(podCopy, appID, partition, allocationKey,
nodeID, "AssumePodFailed",
+ fmt.Sprintf("Node assignment failed for %s on node %s, it will
be retried", alias, nodeID))
+}
+
+// rollbackAllocation resets the task allocation state and notifies the core
to move
+// the allocation back to a pending ask so it can be re-scheduled on a
different node.
+// Must be called without holding the task lock.
+func (task *Task) rollbackAllocation(podCopy *v1.Pod, appID, partition,
allocationKey, nodeID, eventReason, eventMsg string) {
// Post a warning event so operators can see the retry via kubectl
describe pod.
events.GetRecorder().Eventf(podCopy, nil,
- v1.EventTypeWarning, "AssumePodFailed", "AssumePodFailed",
- "Node assignment failed for %s on node %s, it will be retried",
alias, nodeID)
+ v1.EventTypeWarning, eventReason, eventReason, eventMsg)
// Revert any PV/PVC assumptions made by the volume binder. Idempotent:
safe to call
// even if AssumePodVolumes was never reached or already cleaned up
internally.
@@ -671,6 +719,31 @@ func (task *Task)
rollbackOnAssumePodFailure(allocationKey, nodeID string) {
zap.String("allocationKey", allocationKey))
}
+// rescheduleOnBindFailure is called when volume or pod binding fails after
all retries.
+// Move the task back to Scheduling before rolling back the allocation, so a
+// subsequent TaskAllocated event from the core (which requires Scheduling) is
accepted.
+// Must be called without holding the task lock as rollbackAllocation uses
context lock.
+func (task *Task) rescheduleOnBindFailure(allocationKey, nodeID, eventReason,
eventMsg string) {
+ task.lock.Lock()
+ // Move the task back to Scheduling before releasing to the core, so
the re-delivered
+ // allocation (valid only from the Scheduling state) is accepted by the
state machine.
+ if task.sm.Current() != TaskStates().Scheduling {
+ log.Log(log.ShimCacheTask).Info("task is not in Scheduling
state, moving it back to Scheduling",
+ zap.String("appID", task.applicationID),
+ zap.String("taskID", task.taskID),
+ zap.String("currentState", task.sm.Current()))
+ task.sm.SetState(TaskStates().Scheduling)
+ }
+ podCopy := task.pod.DeepCopy()
+ appID := task.applicationID
+ partition := task.application.partition
+ task.allocationKey = ""
+ task.nodeName = ""
+ task.lock.Unlock()
+
+ task.rollbackAllocation(podCopy, appID, partition, allocationKey,
nodeID, eventReason, eventMsg)
+}
+
func (task *Task) SetTaskPod(pod *v1.Pod) {
task.lock.Lock()
defer task.lock.Unlock()
diff --git a/pkg/cache/task_test.go b/pkg/cache/task_test.go
index 98184b2b..f04f1751 100644
--- a/pkg/cache/task_test.go
+++ b/pkg/cache/task_test.go
@@ -30,12 +30,17 @@ import (
schedulingv1 "k8s.io/api/scheduling/v1"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/apimachinery/pkg/types"
+ "k8s.io/apimachinery/pkg/util/wait"
k8sEvents "k8s.io/client-go/tools/events"
+ "k8s.io/kubernetes/pkg/scheduler/framework/plugins/volumebinding"
"github.com/apache/yunikorn-k8shim/pkg/client"
"github.com/apache/yunikorn-k8shim/pkg/common/constants"
"github.com/apache/yunikorn-k8shim/pkg/common/events"
+ "github.com/apache/yunikorn-k8shim/pkg/common/test"
"github.com/apache/yunikorn-k8shim/pkg/common/utils"
+ "github.com/apache/yunikorn-k8shim/pkg/dispatcher"
"github.com/apache/yunikorn-k8shim/pkg/locking"
"github.com/apache/yunikorn-scheduler-interface/lib/go/si"
)
@@ -1023,3 +1028,228 @@ func TestRollbackOnAssumePodFailure_NilSchedulerAPI(t
*testing.T) {
assert.Equal(t, "", task.GetAllocationKey(), "allocationKey should be
cleared even with nil SchedulerAPI")
assert.Equal(t, "", task.GetNodeName(), "nodeName should be cleared
even with nil SchedulerAPI")
}
+
+// newBindTestPod returns a minimal pod usable for exercising the bind path.
+func newBindTestPod(name, uid string) *v1.Pod {
+ return &v1.Pod{
+ TypeMeta: metav1.TypeMeta{Kind: "Pod", APIVersion: "v1"},
+ ObjectMeta: metav1.ObjectMeta{Name: name, UID: types.UID(uid)},
+ }
+}
+
+// setShortBindBackoff shrinks the bind retry backoff so tests do not wait for
the
+// production schedule, and returns a restore function.
+func setShortBindBackoff(steps int) func() {
+ original := retryBackoff
+ retryBackoff = wait.Backoff{Steps: steps, Duration: time.Millisecond}
+ return func() { retryBackoff = original }
+}
+
+// TestRescheduleOnBindFailure_ClearsStateAndReschedules verifies that a bind
failure
+// clears the allocation, moves the task back to Scheduling, and sends a
+// SCHEDULING_FAILED_ON_RM release so the core can re-schedule on a different
node.
+func TestRescheduleOnBindFailure_ClearsStateAndReschedules(t *testing.T) {
+ mockedContext, apiProvider := initContextAndAPIProviderForTest()
+ recorder := k8sEvents.NewFakeRecorder(1024)
+ events.SetRecorder(recorder)
+ defer events.SetRecorder(events.NewMockedRecorder())
+
+ var rollbackSent atomic.Bool
+ apiProvider.MockSchedulerAPIUpdateAllocationFn(func(request
*si.AllocationRequest) error {
+ if request.Releases != nil {
+ for _, rel := range
request.Releases.AllocationsToRelease {
+ if rel.TerminationType ==
si.TerminationType_SCHEDULING_FAILED_ON_RM &&
+ rel.AllocationKey == taskUID1 {
+ rollbackSent.Store(true)
+ }
+ }
+ }
+ return nil
+ })
+
+ task := newRollbackTask(mockedContext, taskUID1, fakeNodeName)
+ task.sm.SetState(TaskStates().Allocated)
+ task.rescheduleOnBindFailure(taskUID1, fakeNodeName, "PodBindFailure",
"bind failed, it will be retried")
+
+ assert.Equal(t, TaskStates().Scheduling, task.GetTaskState(), "task
should be back in Scheduling after bind failure")
+ assert.Equal(t, "", task.GetAllocationKey(), "allocationKey should be
cleared")
+ assert.Equal(t, "", task.GetNodeName(), "nodeName should be cleared")
+ assert.Assert(t, rollbackSent.Load(), "SCHEDULING_FAILED_ON_RM release
request should be sent")
+ assert.Assert(t, len(recorder.Events) >= 1, "a bind failure event
should be posted")
+}
+
+// TestPostTaskAllocated_BindRetrySucceeds verifies that a transient pod bind
failure is
+// retried and, once it succeeds, the task stays Allocated without being
rolled back.
+func TestPostTaskAllocated_BindRetrySucceeds(t *testing.T) {
+ mockedContext, apiProvider := initContextAndAPIProviderForTest()
+ events.SetRecorder(events.NewMockedRecorder())
+ defer events.SetRecorder(events.NewMockedRecorder())
+ defer setShortBindBackoff(5)()
+
+ var bindCalls atomic.Int32
+ apiProvider.MockBindFn(func(_ *v1.Pod, _ string) error {
+ if bindCalls.Add(1) < 3 {
+ return fmt.Errorf("transient bind error")
+ }
+ return nil
+ })
+
+ app := NewApplication(appID1, queueNameA, testUser, testGroups,
map[string]string{},
+ apiProvider.GetAPIs().SchedulerAPI)
+ task := NewTask(taskUID1, app, mockedContext,
newBindTestPod("bind-retry-pod", "bind-retry-uid"))
+ task.sm.SetState(TaskStates().Scheduling)
+
+ err := task.handle(NewAllocateTaskEvent(app.applicationID, task.taskID,
taskUID1, fakeNodeName))
+ assert.NilError(t, err, "failed to handle AllocateTask event")
+
+ err = utils.WaitForCondition(func() bool {
+ return bindCalls.Load() == 3 && task.GetTaskSchedulingState()
== TaskSchedAllocated
+ }, 10*time.Millisecond, 3*time.Second)
+ assert.NilError(t, err, "pod bind did not succeed after retries")
+ assert.Equal(t, TaskStates().Allocated, task.GetTaskState(), "task
should remain Allocated after a successful bind")
+ assert.Equal(t, fakeNodeName, task.GetNodeName(), "node assignment
should be kept after a successful bind")
+}
+
+// TestPostTaskAllocated_BindExhaustedReschedules verifies that when pod
binding keeps
+// failing, the allocation is rolled back and the task returns to Scheduling.
+func TestPostTaskAllocated_BindExhaustedReschedules(t *testing.T) {
+ mockedContext, apiProvider := initContextAndAPIProviderForTest()
+ events.SetRecorder(events.NewMockedRecorder())
+ defer events.SetRecorder(events.NewMockedRecorder())
+ defer setShortBindBackoff(2)()
+
+ apiProvider.MockBindFn(func(_ *v1.Pod, _ string) error {
+ return fmt.Errorf("permanent bind error")
+ })
+
+ var rollbackSent atomic.Bool
+ apiProvider.MockSchedulerAPIUpdateAllocationFn(func(request
*si.AllocationRequest) error {
+ if request.Releases != nil {
+ for _, rel := range
request.Releases.AllocationsToRelease {
+ if rel.TerminationType ==
si.TerminationType_SCHEDULING_FAILED_ON_RM {
+ rollbackSent.Store(true)
+ }
+ }
+ }
+ return nil
+ })
+
+ app := NewApplication(appID1, queueNameA, testUser, testGroups,
map[string]string{},
+ apiProvider.GetAPIs().SchedulerAPI)
+ task := NewTask(taskUID1, app, mockedContext,
newBindTestPod("bind-fail-pod", "bind-fail-uid"))
+ task.sm.SetState(TaskStates().Scheduling)
+
+ err := task.handle(NewAllocateTaskEvent(app.applicationID, task.taskID,
taskUID1, fakeNodeName))
+ assert.NilError(t, err, "failed to handle AllocateTask event")
+
+ err = utils.WaitForCondition(func() bool {
+ return task.GetTaskState() == TaskStates().Scheduling
+ }, 10*time.Millisecond, 3*time.Second)
+ assert.NilError(t, err, "task should be moved back to Scheduling after
bind exhaustion")
+ assert.Equal(t, "", task.GetAllocationKey(), "allocationKey should be
cleared after rollback")
+ assert.Equal(t, "", task.GetNodeName(), "nodeName should be cleared
after rollback")
+ assert.Assert(t, rollbackSent.Load(), "SCHEDULING_FAILED_ON_RM release
request should be sent")
+}
+
+// TestPostTaskAllocated_VolumeBindExhaustedReschedules verifies that when
volume binding
+// keeps failing, the allocation is rolled back and the task returns to
Scheduling.
+func TestPostTaskAllocated_VolumeBindExhaustedReschedules(t *testing.T) {
+ binder := test.NewVolumeBinderMock()
+ binder.SetAllBound(false)
+ binder.SetPodVolumes(&volumebinding.PodVolumes{})
+ binder.EnableBindPodVolumesError("permanent volume bind error")
+ mockedContext := initAssumePodTest(binder)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+ defer setShortBindBackoff(2)()
+
+ apiProvider := mockedContext.apiProvider.(*client.MockedAPIProvider)
//nolint:errcheck
+ var rollbackSent atomic.Bool
+ apiProvider.MockSchedulerAPIUpdateAllocationFn(func(request
*si.AllocationRequest) error {
+ if request.Releases != nil {
+ for _, rel := range
request.Releases.AllocationsToRelease {
+ if rel.TerminationType ==
si.TerminationType_SCHEDULING_FAILED_ON_RM {
+ rollbackSent.Store(true)
+ }
+ }
+ }
+ return nil
+ })
+
+ // assume the pod so it is present in the scheduler cache with volumes
not fully bound
+ err := mockedContext.AssumePod(pod1UID, fakeNodeName)
+ assert.NilError(t, err, "failed to assume pod")
+
+ app := NewApplication(appID, queue, testUser, testGroups,
map[string]string{},
+ apiProvider.GetAPIs().SchedulerAPI)
+ task := NewTask(pod1UID, app, mockedContext, newBindTestPod(podName1,
pod1UID))
+ task.sm.SetState(TaskStates().Scheduling)
+
+ err = task.handle(NewAllocateTaskEvent(app.applicationID, task.taskID,
pod1UID, fakeNodeName))
+ assert.NilError(t, err, "failed to handle AllocateTask event")
+
+ err = utils.WaitForCondition(func() bool {
+ return task.GetTaskState() == TaskStates().Scheduling
+ }, 10*time.Millisecond, 3*time.Second)
+ assert.NilError(t, err, "task should be moved back to Scheduling after
volume bind exhaustion")
+ assert.Equal(t, "", task.GetAllocationKey(), "allocationKey should be
cleared after rollback")
+ assert.Equal(t, "", task.GetNodeName(), "nodeName should be cleared
after rollback")
+ assert.Assert(t, rollbackSent.Load(), "SCHEDULING_FAILED_ON_RM release
request should be sent")
+}
+
+// TestPostTaskAllocated_VolumeBindNoRetryOnBindingVolumesError verifies that
when volume binding
+// fails with a "binding volumes:" error, retries are stopped immediately and
the allocation is rolled back.
+func TestPostTaskAllocated_VolumeBindNoRetryOnBindingVolumesError(t
*testing.T) {
+ binder := test.NewVolumeBinderMock()
+ binder.SetAllBound(false)
+ binder.SetPodVolumes(&volumebinding.PodVolumes{})
+ binder.EnableBindPodVolumesError("binding volumes: timed out waiting
for volume binding")
+ mockedContext := initAssumePodTest(binder)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ // Using a long backoff to prove it doesn't wait for retries
+ origBackoff := retryBackoff
+ retryBackoff = wait.Backoff{
+ Steps: 5,
+ Duration: 10 * time.Second,
+ Factor: 2,
+ }
+ defer func() {
+ retryBackoff = origBackoff
+ }()
+
+ apiProvider := mockedContext.apiProvider.(*client.MockedAPIProvider)
//nolint:errcheck
+ var rollbackSent atomic.Bool
+ apiProvider.MockSchedulerAPIUpdateAllocationFn(func(request
*si.AllocationRequest) error {
+ if request.Releases != nil {
+ for _, rel := range
request.Releases.AllocationsToRelease {
+ if rel.TerminationType ==
si.TerminationType_SCHEDULING_FAILED_ON_RM {
+ rollbackSent.Store(true)
+ }
+ }
+ }
+ return nil
+ })
+
+ // assume the pod so it is present in the scheduler cache with volumes
not fully bound
+ err := mockedContext.AssumePod(pod1UID, fakeNodeName)
+ assert.NilError(t, err, "failed to assume pod")
+
+ app := NewApplication(appID, queue, testUser, testGroups,
map[string]string{},
+ apiProvider.GetAPIs().SchedulerAPI)
+ task := NewTask(pod1UID, app, mockedContext, newBindTestPod(podName1,
pod1UID))
+ task.sm.SetState(TaskStates().Scheduling)
+
+ err = task.handle(NewAllocateTaskEvent(app.applicationID, task.taskID,
pod1UID, fakeNodeName))
+ assert.NilError(t, err, "failed to handle AllocateTask event")
+
+ err = utils.WaitForCondition(func() bool {
+ return task.GetTaskState() == TaskStates().Scheduling
+ }, 10*time.Millisecond, 3*time.Second)
+ assert.NilError(t, err, "task should be moved back to Scheduling
without retrying for binding volumes error")
+ assert.Equal(t, int32(1), binder.GetBindCount(), "bindPodVolumes should
only be called once when error starts with 'binding volumes:'")
+ assert.Equal(t, "", task.GetAllocationKey(), "allocationKey should be
cleared after rollback")
+ assert.Equal(t, "", task.GetNodeName(), "nodeName should be cleared
after rollback")
+ assert.Assert(t, rollbackSent.Load(), "SCHEDULING_FAILED_ON_RM release
request should be sent")
+}
diff --git a/pkg/common/test/volumebinder_mock.go
b/pkg/common/test/volumebinder_mock.go
index db2e6479..607d24e4 100644
--- a/pkg/common/test/volumebinder_mock.go
+++ b/pkg/common/test/volumebinder_mock.go
@@ -21,6 +21,7 @@ package test
import (
"context"
"errors"
+ "sync/atomic"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/util/sets"
@@ -41,6 +42,7 @@ type VolumeBinderMock struct {
podVolumes *volumebinding.PodVolumes
allBound bool
revertCalledCount int
+ bindCount atomic.Int32
}
func NewVolumeBinderMock() *VolumeBinderMock {
@@ -94,9 +96,14 @@ func (v *VolumeBinderMock) SetPodVolumes(podVolumes
*volumebinding.PodVolumes) {
}
func (v *VolumeBinderMock) BindPodVolumes(_ context.Context, _ *v1.Pod, _
*volumebinding.PodVolumes) error {
+ v.bindCount.Add(1)
return v.bindError
}
+func (v *VolumeBinderMock) GetBindCount() int32 {
+ return v.bindCount.Load()
+}
+
func (v *VolumeBinderMock) EnableVolumeClaimsError(message string) {
v.volumeClaimError = errors.New(message)
}
@@ -105,6 +112,10 @@ func (v *VolumeBinderMock)
EnableFindPodVolumesError(message string) {
v.findPodVolumesError = errors.New(message)
}
+func (v *VolumeBinderMock) EnableBindPodVolumesError(message string) {
+ v.bindError = errors.New(message)
+}
+
func (v *VolumeBinderMock) SetConflictReasons(reasons ...string) {
var conflicts []volumebinding.ConflictReason
for _, r := range reasons {
diff --git a/pkg/shim/scheduler_test.go b/pkg/shim/scheduler_test.go
index 10b68d42..145cabc1 100644
--- a/pkg/shim/scheduler_test.go
+++ b/pkg/shim/scheduler_test.go
@@ -20,6 +20,7 @@ package shim
import (
"fmt"
+ "sync/atomic"
"testing"
"time"
@@ -184,10 +185,13 @@ func TestTaskFailures(t *testing.T) {
assert.NilError(t, cluster.start(), "failed to start cluster")
defer cluster.stop()
- // mock pod bind failures
+ // mock a transient pod bind failure: task0001 fails its first two bind
attempts and
+ // then succeeds. With the bind retry mechanism the task must recover
and bind rather
+ // than failing.
+ var task1BindAttempts atomic.Int32
cluster.apiProvider.MockBindFn(func(pod *v1.Pod, hostID string) error {
- if pod.Name == "task0001" {
- return fmt.Errorf("mocked error when binding the pod")
+ if pod.Name == "task0001" && task1BindAttempts.Add(1) <= 2 {
+ return fmt.Errorf("mocked transient error when binding
the pod")
}
return nil
})
@@ -216,15 +220,16 @@ func TestTaskFailures(t *testing.T) {
cluster.AddPod(task2)
// wait for scheduling app and tasks
- // verify app state
+ // both tasks must end up bound: task0002 directly, task0001 after its
bind is retried
cluster.waitAndAssertApplicationState(t, "app0001",
cache.ApplicationStates().Running)
- cluster.waitAndAssertTaskState(t, "app0001", "task0001",
cache.TaskStates().Failed)
+ cluster.waitAndAssertTaskState(t, "app0001", "task0001",
cache.TaskStates().Bound)
cluster.waitAndAssertTaskState(t, "app0001", "task0002",
cache.TaskStates().Bound)
- // one task get bound, one ask failed, so we are expecting only 1
allocation in the scheduler
+ // both tasks get bound, so we are expecting 2 allocations in the
scheduler
err = cluster.waitAndVerifySchedulerAllocations("root.a",
- "[mycluster]default", "app0001", 1)
+ "[mycluster]default", "app0001", 2)
assert.NilError(t, err, "number of allocations is not expected, error")
+ assert.Assert(t, task1BindAttempts.Load() >= 3, "task0001 bind should
have been retried before succeeding")
}
// simulate PVC error during Context.AssumePod() call
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
