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 99f8d532 [YUNIKORN-3332]Prevent timed out Placeholder creation (#1054)
99f8d532 is described below
commit 99f8d532872ea87a8592d5395446b9e8428de7b8
Author: Ompal singh <[email protected]>
AuthorDate: Thu Aug 20 17:06:22 2026 +1000
[YUNIKORN-3332]Prevent timed out Placeholder creation (#1054)
On restart we should not create placeholders that are already timed-out.
Creating a placeholder that is already timed out happens if:
* Stop YuniKorn
* Create a pod with gang definition and a short placeholder timeout
* Start YuniKorn
Those placeholders time-out on allocation which can lead to unexpected
behaviour.
Closes: #1054
Signed-off-by: Wilfred Spiegelenburg <[email protected]>
---
pkg/cache/application.go | 49 ++++++++-
pkg/cache/application_test.go | 235 ++++++++++++++++++++++++++++++++++++++++++
2 files changed, 279 insertions(+), 5 deletions(-)
diff --git a/pkg/cache/application.go b/pkg/cache/application.go
index 44d0dc55..eaa4af89 100644
--- a/pkg/cache/application.go
+++ b/pkg/cache/application.go
@@ -22,7 +22,9 @@ import (
"context"
"fmt"
"sort"
+ "strconv"
"strings"
+ "time"
"github.com/looplab/fsm"
"go.uber.org/zap"
@@ -37,6 +39,7 @@ import (
"github.com/apache/yunikorn-k8shim/pkg/locking"
"github.com/apache/yunikorn-k8shim/pkg/log"
"github.com/apache/yunikorn-scheduler-interface/lib/go/api"
+ siCommon "github.com/apache/yunikorn-scheduler-interface/lib/go/common"
"github.com/apache/yunikorn-scheduler-interface/lib/go/si"
)
@@ -510,19 +513,55 @@ func (app *Application) onResuming() {
}
}
+// isPlaceholderTimeoutElapsed returns true when the configured placeholder
timeout has already
+// elapsed since the application creation time. Used on restart to avoid
creating placeholder
+// pods that would immediately time out.
+func (app *Application) isPlaceholderTimeoutElapsed() bool {
+ if app.placeholderTimeoutInSec <= 0 {
+ return false
+ }
+ creationTimeTag :=
app.tags[siCommon.DomainYuniKorn+siCommon.CreationTime]
+ if creationTimeTag == "" {
+ return false
+ }
+ createdAtSec, err := strconv.ParseInt(creationTimeTag, 10, 64)
+ if err != nil {
+ return false
+ }
+ elapsed := time.Since(time.Unix(createdAtSec, 0))
+ return elapsed >= time.Duration(app.placeholderTimeoutInSec)*time.Second
+}
+
// onReserving triggered when entering the reserving state.
// During normal operation this creates all the placeholders. During recovery
this call could cause the application
// in the shim and core to progress to the next state.
func (app *Application) onReserving() {
// if any placeholder already exist during recovery we might need to
send
// an event to trigger Application state change in the core
- if len(app.getPlaceHolderTasks()) > 0 {
+ switch {
+ case len(app.getPlaceHolderTasks()) > 0:
ev := NewUpdateApplicationReservationEvent(app.applicationID)
dispatcher.Dispatch(ev)
- } else if app.originatingTask != nil {
- // not recovery or no placeholders created yet add an event to
the pod
-
events.GetRecorder().Eventf(app.originatingTask.GetTaskPod().DeepCopy(), nil,
v1.EventTypeNormal, "GangScheduling",
- "CreatingPlaceholders", "Application %s creating
placeholders", app.applicationID)
+ case app.isPlaceholderTimeoutElapsed():
+ log.Log(log.ShimCacheApplication).Info("placeholder timeout
exceeded on restart, not creating placeholders",
+ zap.String("appID", app.applicationID),
+ zap.String("schedulingStyle", app.schedulingStyle))
+ if app.originatingTask != nil {
+
events.GetRecorder().Eventf(app.originatingTask.GetTaskPod().DeepCopy(), nil,
v1.EventTypeWarning, "GangScheduling",
+ "PlaceholderTimeout", "Application %s
placeholder timeout exceeded, not creating placeholders", app.applicationID)
+ }
+ if app.schedulingStyle ==
constants.SchedulingPolicyStyleParamValues["Hard"] {
+
dispatcher.Dispatch(NewFailApplicationEvent(app.applicationID,
constants.ApplicationInsufficientResourcesFailure))
+ } else {
+
dispatcher.Dispatch(NewRunApplicationEvent(app.applicationID))
+ }
+ return
+ default:
+ if app.originatingTask != nil {
+ // not recovery or no placeholders created yet add an
event to the pod
+
events.GetRecorder().Eventf(app.originatingTask.GetTaskPod().DeepCopy(), nil,
v1.EventTypeNormal, "GangScheduling",
+ "CreatingPlaceholders", "Application %s
creating placeholders", app.applicationID)
+ }
}
go func() {
diff --git a/pkg/cache/application_test.go b/pkg/cache/application_test.go
index 8b224ae2..5aa0e078 100644
--- a/pkg/cache/application_test.go
+++ b/pkg/cache/application_test.go
@@ -21,6 +21,7 @@ package cache
import (
"fmt"
"sort"
+ "strconv"
"strings"
"testing"
"time"
@@ -867,6 +868,240 @@ func TestTryReservePostRestart(t *testing.T) {
assert.Equal(t, createdPods.count(), 0)
}
+func TestIsPlaceholderTimeoutElapsed(t *testing.T) {
+ app := NewApplication("app00001", "root.default", "test-user",
+ testGroups, map[string]string{}, newMockSchedulerAPI())
+ creationTag := siCommon.DomainYuniKorn + siCommon.CreationTime
+
+ tests := []struct {
+ name string
+ timeout int64
+ tag string
+ setTag bool
+ elapsed bool
+ }{
+ {
+ name: "no timeout configured",
+ timeout: 0,
+ tag:
strconv.FormatInt(time.Now().Add(-60*time.Second).Unix(), 10),
+ setTag: true,
+ elapsed: false,
+ },
+ {
+ name: "missing creation time tag",
+ timeout: 10,
+ setTag: false,
+ elapsed: false,
+ },
+ {
+ name: "invalid creation time tag",
+ timeout: 10,
+ tag: "not-a-number",
+ setTag: true,
+ elapsed: false,
+ },
+ {
+ name: "timeout not yet elapsed",
+ timeout: 60,
+ tag: strconv.FormatInt(time.Now().Unix(), 10),
+ setTag: true,
+ elapsed: false,
+ },
+ {
+ name: "timeout elapsed",
+ timeout: 10,
+ tag:
strconv.FormatInt(time.Now().Add(-60*time.Second).Unix(), 10),
+ setTag: true,
+ elapsed: true,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ app.SetPlaceholderTimeout(tt.timeout)
+ if tt.setTag {
+ app.tags[creationTag] = tt.tag
+ } else {
+ delete(app.tags, creationTag)
+ }
+ assert.Equal(t, tt.elapsed,
app.isPlaceholderTimeoutElapsed())
+ })
+ }
+}
+
+func TestOnReservingCreatesPlaceholdersWhenTimeoutNotElapsed(t *testing.T) {
+ context := initContextForTest()
+ dispatcher.RegisterEventHandler("TestAppHandler",
dispatcher.EventTypeApp, context.ApplicationEventHandler())
+ dispatcher.Start()
+ defer dispatcher.Stop()
+
+ createdPods := newThreadSafePodsMap()
+ mockedAPIProvider := client.NewMockedAPIProvider(false)
+ mockedAPIProvider.MockCreateFn(func(pod *v1.Pod) (*v1.Pod, error) {
+ createdPods.add(pod)
+ return pod, nil
+ })
+ mgr := NewPlaceholderManager(mockedAPIProvider.GetAPIs())
+ mgr.Start()
+ defer mgr.Stop()
+
+ app := NewApplication("app00001", "root.abc", "test-user",
+ testGroups, map[string]string{},
mockedAPIProvider.GetAPIs().SchedulerAPI)
+ context.addApplicationToContext(app)
+ app.setTaskGroups([]TaskGroup{
+ {
+ Name: "test-group-1",
+ MinMember: 1,
+ MinResource: map[string]resource.Quantity{
+ v1.ResourceCPU.String():
resource.MustParse("500m"),
+ v1.ResourceMemory.String():
resource.MustParse("500Mi"),
+ },
+ },
+ })
+ app.SetPlaceholderTimeout(300)
+ app.tags[siCommon.DomainYuniKorn+siCommon.CreationTime] =
strconv.FormatInt(time.Now().Unix(), 10)
+
+ err := app.handle(NewSubmitApplicationEvent(app.applicationID))
+ assert.NilError(t, err)
+ err = app.handle(NewSimpleApplicationEvent(app.GetApplicationID(),
AcceptApplication))
+ assert.NilError(t, err)
+ err = app.handle(NewSimpleApplicationEvent(app.applicationID,
TryReserve))
+ assert.NilError(t, err)
+
+ assertAppState(t, app, ApplicationStates().Reserving, 3*time.Second)
+ err = utils.WaitForCondition(func() bool {
+ return createdPods.count() == 1
+ }, 100*time.Millisecond, 3*time.Second)
+ assert.NilError(t, err, "placeholders should be created when timeout
has not elapsed")
+}
+
+func TestOnReservingWithExistingPlaceholders(t *testing.T) {
+ context := initContextForTest()
+ dispatcher.RegisterEventHandler("TestAppHandler",
dispatcher.EventTypeApp, context.ApplicationEventHandler())
+ dispatcher.Start()
+ defer dispatcher.Stop()
+
+ createdPods := newThreadSafePodsMap()
+ mockedAPIProvider := client.NewMockedAPIProvider(false)
+ mockedAPIProvider.MockCreateFn(func(pod *v1.Pod) (*v1.Pod, error) {
+ createdPods.add(pod)
+ return pod, nil
+ })
+ mgr := NewPlaceholderManager(mockedAPIProvider.GetAPIs())
+ mgr.Start()
+ defer mgr.Stop()
+
+ app := NewApplication("app00001", "root.abc", "test-user",
+ testGroups, map[string]string{},
mockedAPIProvider.GetAPIs().SchedulerAPI)
+ context.addApplicationToContext(app)
+ app.setTaskGroups([]TaskGroup{
+ {
+ Name: "test-group-1",
+ MinMember: 2,
+ MinResource: map[string]resource.Quantity{
+ v1.ResourceCPU.String():
resource.MustParse("500m"),
+ v1.ResourceMemory.String():
resource.MustParse("500Mi"),
+ },
+ },
+ })
+ app.SetPlaceholderTimeout(10)
+
app.setSchedulingStyle(constants.SchedulingPolicyStyleParamValues["Hard"])
+ app.tags[siCommon.DomainYuniKorn+siCommon.CreationTime] =
strconv.FormatInt(
+ time.Now().Add(-60*time.Second).Unix(), 10)
+
+ existingPlaceholder := NewTaskPlaceholder("placeholder-01", app,
context, &v1.Pod{
+ ObjectMeta: apis.ObjectMeta{
+ Name: "placeholder-pod-01",
+ UID: "UID-placeholder-01",
+ },
+ })
+ existingPlaceholder.setTaskGroupName("test-group-1")
+ app.addTask(existingPlaceholder)
+
+ err := app.handle(NewSubmitApplicationEvent(app.applicationID))
+ assert.NilError(t, err)
+ err = app.handle(NewSimpleApplicationEvent(app.GetApplicationID(),
AcceptApplication))
+ assert.NilError(t, err)
+ err = app.handle(NewSimpleApplicationEvent(app.applicationID,
TryReserve))
+ assert.NilError(t, err)
+
+ assertAppState(t, app, ApplicationStates().Reserving, 3*time.Second)
+ err = utils.WaitForCondition(func() bool {
+ return createdPods.count() == 1
+ }, 100*time.Millisecond, 3*time.Second)
+ assert.NilError(t, err, "missing placeholders should still be created
during recovery")
+}
+
+func TestOnReservingSkipsTimedOutPlaceholders(t *testing.T) {
+ tests := []struct {
+ name string
+ schedulingStyle string
+ expectedState string
+ }{
+ {
+ name: "soft",
+ schedulingStyle:
constants.SchedulingPolicyStyleParamDefault,
+ expectedState: ApplicationStates().Running,
+ },
+ {
+ name: "hard",
+ schedulingStyle:
constants.SchedulingPolicyStyleParamValues["Hard"],
+ expectedState: ApplicationStates().Failing,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ context := initContextForTest()
+ dispatcher.RegisterEventHandler("TestAppHandler",
dispatcher.EventTypeApp, context.ApplicationEventHandler())
+ dispatcher.Start()
+ defer dispatcher.Stop()
+
+ createdPods := newThreadSafePodsMap()
+ mockedAPIProvider := client.NewMockedAPIProvider(false)
+ mockedAPIProvider.MockCreateFn(func(pod *v1.Pod)
(*v1.Pod, error) {
+ createdPods.add(pod)
+ return pod, nil
+ })
+ mgr :=
NewPlaceholderManager(mockedAPIProvider.GetAPIs())
+ mgr.Start()
+ defer mgr.Stop()
+
+ app := NewApplication("app00001", "root.abc",
"test-user",
+ testGroups, map[string]string{},
mockedAPIProvider.GetAPIs().SchedulerAPI)
+ context.addApplicationToContext(app)
+ app.setTaskGroups([]TaskGroup{
+ {
+ Name: "test-group-1",
+ MinMember: 1,
+ MinResource:
map[string]resource.Quantity{
+ v1.ResourceCPU.String():
resource.MustParse("500m"),
+ v1.ResourceMemory.String():
resource.MustParse("500Mi"),
+ },
+ },
+ })
+ app.SetPlaceholderTimeout(1)
+ app.setSchedulingStyle(tt.schedulingStyle)
+ app.tags[siCommon.DomainYuniKorn+siCommon.CreationTime]
= strconv.FormatInt(
+ time.Now().Add(-60*time.Second).Unix(), 10)
+
+ err :=
app.handle(NewSubmitApplicationEvent(app.applicationID))
+ assert.NilError(t, err)
+ err =
app.handle(NewSimpleApplicationEvent(app.GetApplicationID(), AcceptApplication))
+ assert.NilError(t, err)
+
+ err =
app.handle(NewSimpleApplicationEvent(app.applicationID, TryReserve))
+ assert.NilError(t, err)
+
+ assertAppState(t, app, tt.expectedState, 3*time.Second)
+ err = utils.WaitForCondition(func() bool {
+ return createdPods.count() == 0
+ }, 100*time.Millisecond, time.Second)
+ assert.NilError(t, err, "placeholders should not be
created when timeout already elapsed")
+ })
+ }
+}
+
func TestTriggerAppSubmission(t *testing.T) {
// Trigger app submission should be successful if the app is in New
state
mockScheduler := newMockSchedulerAPI()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
