You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@yunikorn.apache.org by pb...@apache.org on 2023/07/19 14:53:09 UTC
[yunikorn-core] branch master updated: [YUNIKORN-1800] Extend Application event wrapper with new events (#581)
This is an automated email from the ASF dual-hosted git repository.
pbacsko 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 7cf85473 [YUNIKORN-1800] Extend Application event wrapper with new events (#581)
7cf85473 is described below
commit 7cf8547375066993d0853f6e8e8d337195d67034
Author: Rainie Li <ra...@pinterest.com>
AuthorDate: Wed Jul 19 16:48:03 2023 +0200
[YUNIKORN-1800] Extend Application event wrapper with new events (#581)
Closes: #581
Signed-off-by: Peter Bacsko <pb...@cloudera.com>
---
pkg/scheduler/objects/application.go | 56 +++++-----
pkg/scheduler/objects/application_events.go | 32 ++++++
pkg/scheduler/objects/application_events_test.go | 122 ++++++++++++++++++++
pkg/scheduler/objects/application_state.go | 8 ++
pkg/scheduler/objects/application_state_test.go | 135 ++++++++++++++++-------
pkg/scheduler/objects/application_test.go | 62 +++++++----
pkg/scheduler/objects/queue.go | 1 +
pkg/scheduler/objects/queue_test.go | 25 +++--
8 files changed, 343 insertions(+), 98 deletions(-)
diff --git a/pkg/scheduler/objects/application.go b/pkg/scheduler/objects/application.go
index 01d6df01..8ebf645f 100644
--- a/pkg/scheduler/objects/application.go
+++ b/pkg/scheduler/objects/application.go
@@ -104,10 +104,11 @@ type Application struct {
placeholderData map[string]*PlaceholderData // track placeholder and gang related info
askMaxPriority int32 // highest priority value of outstanding asks
- rmEventHandler handler.EventHandler
- rmID string
- terminatedCallback func(appID string)
- appEvents *applicationEvents
+ rmEventHandler handler.EventHandler
+ rmID string
+ terminatedCallback func(appID string)
+ appEvents *applicationEvents
+ sendStateChangeEvents bool // whether to send state-change events or not (simplifies testing)
sync.RWMutex
}
@@ -158,27 +159,28 @@ func (sa *Application) GetApplicationSummary(rmID string) *ApplicationSummary {
func NewApplication(siApp *si.AddApplicationRequest, ugi security.UserGroup, eventHandler handler.EventHandler, rmID string) *Application {
app := &Application{
- ApplicationID: siApp.ApplicationID,
- Partition: siApp.PartitionName,
- SubmissionTime: time.Now(),
- queuePath: siApp.QueueName,
- tags: siApp.Tags,
- pending: resources.NewResource(),
- allocatedResource: resources.NewResource(),
- usedResource: resources.NewUsedResource(),
- maxAllocatedResource: resources.NewResource(),
- allocatedPlaceholder: resources.NewResource(),
- requests: make(map[string]*AllocationAsk),
- reservations: make(map[string]*reservation),
- allocations: make(map[string]*Allocation),
- stateMachine: NewAppState(),
- placeholderAsk: resources.NewResourceFromProto(siApp.PlaceholderAsk),
- startTime: time.Time{},
- finishedTime: time.Time{},
- rejectedMessage: "",
- stateLog: make([]*StateLogEntry, 0),
- askMaxPriority: configs.MinPriority,
- sortedRequests: sortedRequests{},
+ ApplicationID: siApp.ApplicationID,
+ Partition: siApp.PartitionName,
+ SubmissionTime: time.Now(),
+ queuePath: siApp.QueueName,
+ tags: siApp.Tags,
+ pending: resources.NewResource(),
+ allocatedResource: resources.NewResource(),
+ usedResource: resources.NewUsedResource(),
+ maxAllocatedResource: resources.NewResource(),
+ allocatedPlaceholder: resources.NewResource(),
+ requests: make(map[string]*AllocationAsk),
+ reservations: make(map[string]*reservation),
+ allocations: make(map[string]*Allocation),
+ stateMachine: NewAppState(),
+ placeholderAsk: resources.NewResourceFromProto(siApp.PlaceholderAsk),
+ startTime: time.Time{},
+ finishedTime: time.Time{},
+ rejectedMessage: "",
+ stateLog: make([]*StateLogEntry, 0),
+ askMaxPriority: configs.MinPriority,
+ sortedRequests: sortedRequests{},
+ sendStateChangeEvents: true,
}
placeholderTimeout := common.ConvertSITimeoutWithAdjustment(siApp, defaultPlaceholderTimeout)
gangSchedStyle := siApp.GetGangSchedulingStyle()
@@ -198,6 +200,7 @@ func NewApplication(siApp *si.AddApplicationRequest, ugi security.UserGroup, eve
app.rmEventHandler = eventHandler
app.rmID = rmID
app.appEvents = newApplicationEvents(app, events.GetEventSystem())
+ app.appEvents.sendNewApplicationEvent()
return app
}
@@ -304,6 +307,9 @@ func (sa *Application) HandleApplicationEventWithInfo(event applicationEvent, ev
if err != nil && err.Error() == noTransition {
return nil
}
+ if event == RejectApplication {
+ sa.appEvents.sendRejectApplicationEvent(eventInfo)
+ }
return err
}
diff --git a/pkg/scheduler/objects/application_events.go b/pkg/scheduler/objects/application_events.go
index 686c850f..9f98d6d4 100644
--- a/pkg/scheduler/objects/application_events.go
+++ b/pkg/scheduler/objects/application_events.go
@@ -102,6 +102,38 @@ func (evt *applicationEvents) sendRemoveAskEvent(request *AllocationAsk, detail
evt.eventSystem.AddEvent(event)
}
+func (evt *applicationEvents) sendNewApplicationEvent() {
+ if !evt.enabled {
+ return
+ }
+ event := events.CreateAppEventRecord(evt.app.ApplicationID, "", "", si.EventRecord_ADD, si.EventRecord_DETAILS_NONE, evt.app.allocatedResource)
+ evt.eventSystem.AddEvent(event)
+}
+
+func (evt *applicationEvents) sendRemoveApplicationEvent() {
+ if !evt.enabled {
+ return
+ }
+ event := events.CreateAppEventRecord(evt.app.ApplicationID, "", "", si.EventRecord_REMOVE, si.EventRecord_DETAILS_NONE, evt.app.allocatedResource)
+ evt.eventSystem.AddEvent(event)
+}
+
+func (evt *applicationEvents) sendRejectApplicationEvent(eventInfo string) {
+ if !evt.enabled {
+ return
+ }
+ event := events.CreateAppEventRecord(evt.app.ApplicationID, eventInfo, "", si.EventRecord_REMOVE, si.EventRecord_APP_REJECT, evt.app.allocatedResource)
+ evt.eventSystem.AddEvent(event)
+}
+
+func (evt *applicationEvents) sendStateChangeEvent(changeDetail si.EventRecord_ChangeDetail) {
+ if !evt.enabled || !evt.app.sendStateChangeEvents {
+ return
+ }
+ event := events.CreateAppEventRecord(evt.app.ApplicationID, "", "", si.EventRecord_SET, changeDetail, evt.app.allocatedResource)
+ evt.eventSystem.AddEvent(event)
+}
+
func newApplicationEvents(app *Application, evt events.EventSystem) *applicationEvents {
return &applicationEvents{
eventSystem: evt,
diff --git a/pkg/scheduler/objects/application_events_test.go b/pkg/scheduler/objects/application_events_test.go
index 0c2a3608..72270e2b 100644
--- a/pkg/scheduler/objects/application_events_test.go
+++ b/pkg/scheduler/objects/application_events_test.go
@@ -27,6 +27,27 @@ import (
"github.com/apache/yunikorn-scheduler-interface/lib/go/si"
)
+func isNewApplicationEvent(t *testing.T, app *Application, record *si.EventRecord) {
+ assert.Equal(t, si.EventRecord_APP, record.Type, "incorrect event type, expect app")
+ assert.Equal(t, app.ApplicationID, record.ObjectID, "incorrect object ID, expected application ID")
+ assert.Equal(t, si.EventRecord_ADD, record.EventChangeType, "incorrect change type, expected add")
+ assert.Equal(t, si.EventRecord_DETAILS_NONE, record.EventChangeDetail, "incorrect change detail, expected none")
+}
+
+func isRemoveApplicationEvent(t *testing.T, app *Application, record *si.EventRecord) {
+ assert.Equal(t, si.EventRecord_APP, record.Type, "incorrect event type, expect app")
+ assert.Equal(t, app.ApplicationID, record.ObjectID, "incorrect object ID, expected application ID")
+ assert.Equal(t, si.EventRecord_REMOVE, record.EventChangeType, "incorrect change type, expected remove")
+ assert.Equal(t, si.EventRecord_DETAILS_NONE, record.EventChangeDetail, "incorrect change detail, expected none")
+}
+
+func isStateChangeEvent(t *testing.T, app *Application, changeDetail si.EventRecord_ChangeDetail, record *si.EventRecord) {
+ assert.Equal(t, si.EventRecord_APP, record.Type, "incorrect event type, expect app")
+ assert.Equal(t, app.ApplicationID, record.ObjectID, "incorrect object ID, expected application ID")
+ assert.Equal(t, si.EventRecord_SET, record.EventChangeType, "incorrect change type, expected set")
+ assert.Equal(t, changeDetail, record.EventChangeDetail, "incorrect change detail")
+}
+
func TestSendAppDoesNotFitEvent(t *testing.T) {
app := &Application{
queuePath: "root.test",
@@ -283,3 +304,104 @@ func TestSendRemoveAskEvent(t *testing.T) {
assert.Equal(t, "alloc-1", event.ReferenceID)
assert.Equal(t, "", event.Message)
}
+
+func TestSendNewApplicationEvent(t *testing.T) {
+ app := &Application{
+ ApplicationID: appID0,
+ queuePath: "root.test",
+ }
+ // not enabled
+ evt := newApplicationEvents(app, nil)
+ assert.Assert(t, evt.eventSystem == nil, "event system should be nil")
+ assert.Assert(t, !evt.enabled, "event system should be disabled")
+ evt.sendNewApplicationEvent()
+
+ // enabled
+ mockEvents := newEventSystemMock()
+ appEvents := newApplicationEvents(app, mockEvents)
+
+ appEvents.sendNewApplicationEvent()
+ event := mockEvents.events[0]
+ assert.Equal(t, si.EventRecord_APP, event.Type)
+ assert.Equal(t, si.EventRecord_ADD, event.EventChangeType)
+ assert.Equal(t, si.EventRecord_DETAILS_NONE, event.EventChangeDetail)
+ assert.Equal(t, "app-0", event.ObjectID)
+ assert.Equal(t, "", event.ReferenceID)
+ assert.Equal(t, "", event.Message)
+}
+
+func TestSendRemoveApplicationEvent(t *testing.T) {
+ app := &Application{
+ ApplicationID: appID0,
+ queuePath: "root.test",
+ }
+ // not enabled
+ evt := newApplicationEvents(app, nil)
+ assert.Assert(t, evt.eventSystem == nil, "event system should be nil")
+ assert.Assert(t, !evt.enabled, "event system should be disabled")
+ evt.sendRemoveApplicationEvent()
+
+ // enabled
+ mockEvents := newEventSystemMock()
+ appEvents := newApplicationEvents(app, mockEvents)
+
+ appEvents.sendRemoveApplicationEvent()
+ event := mockEvents.events[0]
+ assert.Equal(t, si.EventRecord_APP, event.Type)
+ assert.Equal(t, si.EventRecord_REMOVE, event.EventChangeType)
+ assert.Equal(t, si.EventRecord_DETAILS_NONE, event.EventChangeDetail)
+ assert.Equal(t, "app-0", event.ObjectID)
+ assert.Equal(t, "", event.ReferenceID)
+ assert.Equal(t, "", event.Message)
+}
+
+func TestSendRejectApplicationEvent(t *testing.T) {
+ app := &Application{
+ ApplicationID: appID0,
+ queuePath: "root.test",
+ }
+ // not enabled
+ evt := newApplicationEvents(app, nil)
+ assert.Assert(t, evt.eventSystem == nil, "event system should be nil")
+ assert.Assert(t, !evt.enabled, "event system should be disabled")
+ evt.sendRejectApplicationEvent("ResourceReservationTimeout")
+
+ // enabled
+ mockEvents := newEventSystemMock()
+ appEvents := newApplicationEvents(app, mockEvents)
+
+ appEvents.sendRejectApplicationEvent("ResourceReservationTimeout")
+ event := mockEvents.events[0]
+ assert.Equal(t, si.EventRecord_APP, event.Type)
+ assert.Equal(t, si.EventRecord_REMOVE, event.EventChangeType)
+ assert.Equal(t, si.EventRecord_APP_REJECT, event.EventChangeDetail)
+ assert.Equal(t, "app-0", event.ObjectID)
+ assert.Equal(t, "", event.ReferenceID)
+ assert.Equal(t, "ResourceReservationTimeout", event.Message)
+}
+
+func TestSendStateChangeEvent(t *testing.T) {
+ app := &Application{
+ ApplicationID: appID0,
+ queuePath: "root.test",
+ sendStateChangeEvents: true,
+ }
+ // not enabled
+ evt := newApplicationEvents(app, nil)
+ assert.Assert(t, evt.eventSystem == nil, "event system should be nil")
+ assert.Assert(t, !evt.enabled, "event system should be disabled")
+ evt.sendStateChangeEvent(si.EventRecord_APP_RUNNING)
+
+ // enabled
+ mockEvents := newEventSystemMock()
+ appEvents := newApplicationEvents(app, mockEvents)
+
+ appEvents.sendStateChangeEvent(si.EventRecord_APP_RUNNING)
+ event := mockEvents.events[0]
+ assert.Equal(t, si.EventRecord_APP, event.Type)
+ assert.Equal(t, si.EventRecord_SET, event.EventChangeType)
+ assert.Equal(t, si.EventRecord_APP_RUNNING, event.EventChangeDetail)
+ assert.Equal(t, "app-0", event.ObjectID)
+ assert.Equal(t, "", event.ReferenceID)
+ assert.Equal(t, "", event.Message)
+}
diff --git a/pkg/scheduler/objects/application_state.go b/pkg/scheduler/objects/application_state.go
index f2f7151b..aef19048 100644
--- a/pkg/scheduler/objects/application_state.go
+++ b/pkg/scheduler/objects/application_state.go
@@ -28,6 +28,7 @@ import (
"github.com/apache/yunikorn-core/pkg/log"
"github.com/apache/yunikorn-core/pkg/metrics"
+ "github.com/apache/yunikorn-scheduler-interface/lib/go/si"
)
const noTransition = "no transition"
@@ -152,6 +153,7 @@ func NewAppState() *fsm.FSM {
app.setStateTimer(app.startTimeout, app.stateMachine.Current(), RunApplication)
metrics.GetQueueMetrics(app.queuePath).IncQueueApplicationsRunning()
metrics.GetSchedulerMetrics().IncTotalApplicationsRunning()
+ app.appEvents.sendStateChangeEvent(si.EventRecord_APP_STARTING)
},
fmt.Sprintf("enter_%s", Resuming.String()): func(_ context.Context, event *fsm.Event) {
app := event.Args[0].(*Application) //nolint:errcheck
@@ -161,11 +163,13 @@ func NewAppState() *fsm.FSM {
fmt.Sprintf("enter_%s", Completing.String()): func(_ context.Context, event *fsm.Event) {
app := event.Args[0].(*Application) //nolint:errcheck
app.setStateTimer(completingTimeout, app.stateMachine.Current(), CompleteApplication)
+ app.appEvents.sendStateChangeEvent(si.EventRecord_APP_COMPLETING)
},
fmt.Sprintf("leave_%s", New.String()): func(_ context.Context, event *fsm.Event) {
app := event.Args[0].(*Application) //nolint:errcheck
metrics.GetQueueMetrics(app.queuePath).IncQueueApplicationsAccepted()
metrics.GetSchedulerMetrics().IncTotalApplicationsAccepted()
+ app.appEvents.sendStateChangeEvent(si.EventRecord_APP_ACCEPTED)
},
fmt.Sprintf("enter_%s", Rejected.String()): func(_ context.Context, event *fsm.Event) {
app := event.Args[0].(*Application) //nolint:errcheck
@@ -190,6 +194,7 @@ func NewAppState() *fsm.FSM {
// account for going back into running state
if event.Dst == Running.String() {
app.queue.incRunningApps(app.ApplicationID)
+ app.appEvents.sendStateChangeEvent(si.EventRecord_APP_RUNNING)
}
},
fmt.Sprintf("enter_%s", Completed.String()): func(_ context.Context, event *fsm.Event) {
@@ -197,6 +202,7 @@ func NewAppState() *fsm.FSM {
metrics.GetSchedulerMetrics().IncTotalApplicationsCompleted()
metrics.GetQueueMetrics(app.queuePath).IncQueueApplicationsCompleted()
app.setStateTimer(terminatedTimeout, app.stateMachine.Current(), ExpireApplication)
+ app.appEvents.sendStateChangeEvent(si.EventRecord_APP_COMPLETED)
app.executeTerminatedCallback()
app.clearPlaceholderTimer()
app.cleanupAsks()
@@ -205,10 +211,12 @@ func NewAppState() *fsm.FSM {
app := event.Args[0].(*Application) //nolint:errcheck
metrics.GetQueueMetrics(app.queuePath).IncQueueApplicationsFailed()
metrics.GetSchedulerMetrics().IncTotalApplicationsFailed()
+ app.appEvents.sendStateChangeEvent(si.EventRecord_APP_FAILING)
},
fmt.Sprintf("enter_%s", Failed.String()): func(_ context.Context, event *fsm.Event) {
app := event.Args[0].(*Application) //nolint:errcheck
app.setStateTimer(terminatedTimeout, app.stateMachine.Current(), ExpireApplication)
+ app.appEvents.sendStateChangeEvent(si.EventRecord_APP_FAILED)
app.executeTerminatedCallback()
app.cleanupAsks()
},
diff --git a/pkg/scheduler/objects/application_state_test.go b/pkg/scheduler/objects/application_state_test.go
index 4c9f9a71..b9d27b26 100644
--- a/pkg/scheduler/objects/application_state_test.go
+++ b/pkg/scheduler/objects/application_state_test.go
@@ -19,12 +19,15 @@
package objects
import (
+ "fmt"
"testing"
"time"
"gotest.tools/v3/assert"
"github.com/apache/yunikorn-core/pkg/common"
+ "github.com/apache/yunikorn-core/pkg/events"
+ "github.com/apache/yunikorn-scheduler-interface/lib/go/si"
)
func TestAcceptStateTransition(t *testing.T) {
@@ -134,83 +137,83 @@ func TestRunStateTransition(t *testing.T) {
func TestCompletedStateTransition(t *testing.T) {
// complete only from completing
- appInfo := newApplication("app-00001", "default", "root.a")
- assert.Equal(t, appInfo.CurrentState(), New.String())
- err := appInfo.HandleApplicationEvent(RunApplication)
+ appInfo1 := newApplication("app-00001", "default", "root.a")
+ assert.Equal(t, appInfo1.CurrentState(), New.String())
+ err := appInfo1.HandleApplicationEvent(RunApplication)
assert.NilError(t, err, "no error expected new to accepted (completed test)")
- err = appInfo.HandleApplicationEvent(RunApplication)
+ err = appInfo1.HandleApplicationEvent(RunApplication)
assert.NilError(t, err, "no error expected accepted to starting (completed test)")
- err = appInfo.HandleApplicationEvent(RunApplication)
+ err = appInfo1.HandleApplicationEvent(RunApplication)
assert.NilError(t, err, "no error expected starting to running (completed test)")
- assert.Equal(t, appInfo.CurrentState(), Running.String())
+ assert.Equal(t, appInfo1.CurrentState(), Running.String())
// completed from run through completing
- err = appInfo.HandleApplicationEvent(CompleteApplication)
+ err = appInfo1.HandleApplicationEvent(CompleteApplication)
assert.NilError(t, err, "no error expected running to completing")
- assert.Equal(t, appInfo.CurrentState(), Completing.String())
- err = appInfo.HandleApplicationEvent(CompleteApplication)
+ assert.Equal(t, appInfo1.CurrentState(), Completing.String())
+ err = appInfo1.HandleApplicationEvent(CompleteApplication)
assert.NilError(t, err, "no error expected completing to completed")
- assert.Equal(t, appInfo.CurrentState(), Completed.String())
+ assert.Equal(t, appInfo1.CurrentState(), Completed.String())
// complete from run through completing
- appInfo = newApplication("app-00002", "default", "root.a")
- assert.Equal(t, appInfo.CurrentState(), New.String())
- err = appInfo.HandleApplicationEvent(RunApplication)
+ appInfo2 := newApplication("app-00002", "default", "root.a")
+ assert.Equal(t, appInfo2.CurrentState(), New.String())
+ err = appInfo2.HandleApplicationEvent(RunApplication)
assert.NilError(t, err, "no error expected new to accepted (completed test)")
- err = appInfo.HandleApplicationEvent(CompleteApplication)
+ err = appInfo2.HandleApplicationEvent(CompleteApplication)
assert.NilError(t, err, "no error expected accepted to completing (completed test)")
- assert.Equal(t, appInfo.CurrentState(), Completing.String())
+ assert.Equal(t, appInfo2.CurrentState(), Completing.String())
// completed from completing
- err = appInfo.HandleApplicationEvent(CompleteApplication)
+ err = appInfo2.HandleApplicationEvent(CompleteApplication)
assert.NilError(t, err, "no error expected completing to completed")
- assert.Equal(t, appInfo.CurrentState(), Completed.String())
+ assert.Equal(t, appInfo2.CurrentState(), Completed.String())
// completed to rejected: error expected
- err = appInfo.HandleApplicationEvent(RejectApplication)
+ err = appInfo2.HandleApplicationEvent(RejectApplication)
assert.Assert(t, err != nil, "error expected completed to rejected")
- assert.Equal(t, appInfo.CurrentState(), Completed.String())
+ assert.Equal(t, appInfo2.CurrentState(), Completed.String())
// completed to failing: error expected
- err = appInfo.HandleApplicationEvent(FailApplication)
+ err = appInfo2.HandleApplicationEvent(FailApplication)
assert.Assert(t, err != nil, "error expected completed to failing")
- assert.Equal(t, appInfo.CurrentState(), Completed.String())
+ assert.Equal(t, appInfo2.CurrentState(), Completed.String())
// completed fails from all completing
- appInfo = newApplication("app-00003", "default", "root.a")
- assert.Equal(t, appInfo.CurrentState(), New.String())
- err = appInfo.HandleApplicationEvent(CompleteApplication)
+ appInfo3 := newApplication("app-00003", "default", "root.a")
+ assert.Equal(t, appInfo3.CurrentState(), New.String())
+ err = appInfo3.HandleApplicationEvent(CompleteApplication)
assert.Assert(t, err != nil, "error expected new to completed")
- assert.Equal(t, appInfo.CurrentState(), New.String())
+ assert.Equal(t, appInfo3.CurrentState(), New.String())
}
func TestCompletingStateTransition(t *testing.T) {
- appInfo := newApplication("app-00001", "default", "root.a")
- assert.Equal(t, appInfo.CurrentState(), New.String())
- err := appInfo.HandleApplicationEvent(RunApplication)
+ appInfo1 := newApplication("app-00001", "default", "root.a")
+ assert.Equal(t, appInfo1.CurrentState(), New.String())
+ err := appInfo1.HandleApplicationEvent(RunApplication)
assert.NilError(t, err, "no error expected new to accepted (completing test)")
// accepted to completing and back again
- err = appInfo.HandleApplicationEvent(CompleteApplication)
+ err = appInfo1.HandleApplicationEvent(CompleteApplication)
assert.NilError(t, err, "no error expected accepted to completing")
- assert.Equal(t, appInfo.CurrentState(), Completing.String())
+ assert.Equal(t, appInfo1.CurrentState(), Completing.String())
// starting to completing
- appInfo = newApplication("app-00002", "default", "root.a")
- assert.Equal(t, appInfo.CurrentState(), New.String())
- err = appInfo.HandleApplicationEvent(RunApplication)
+ appInfo2 := newApplication("app-00002", "default", "root.a")
+ assert.Equal(t, appInfo2.CurrentState(), New.String())
+ err = appInfo2.HandleApplicationEvent(RunApplication)
assert.NilError(t, err, "no error expected new to accepted (completing test)")
- err = appInfo.HandleApplicationEvent(RunApplication)
+ err = appInfo2.HandleApplicationEvent(RunApplication)
assert.NilError(t, err, "no error expected accepted to starting")
- assert.Equal(t, appInfo.CurrentState(), Starting.String())
- err = appInfo.HandleApplicationEvent(CompleteApplication)
+ assert.Equal(t, appInfo2.CurrentState(), Starting.String())
+ err = appInfo2.HandleApplicationEvent(CompleteApplication)
assert.NilError(t, err, "no error expected starting to completing")
- assert.Equal(t, appInfo.CurrentState(), Completing.String())
+ assert.Equal(t, appInfo2.CurrentState(), Completing.String())
// completing to run and back again
- err = appInfo.HandleApplicationEvent(RunApplication)
+ err = appInfo2.HandleApplicationEvent(RunApplication)
assert.NilError(t, err, "no error expected starting to running (completing test)")
- err = appInfo.HandleApplicationEvent(CompleteApplication)
+ err = appInfo2.HandleApplicationEvent(CompleteApplication)
assert.NilError(t, err, "no error expected running to completing")
- assert.Equal(t, appInfo.CurrentState(), Completing.String())
+ assert.Equal(t, appInfo2.CurrentState(), Completing.String())
}
func TestFailedStateTransition(t *testing.T) {
@@ -235,3 +238,55 @@ func TestFailedStateTransition(t *testing.T) {
err = common.WaitFor(10*time.Microsecond, time.Millisecond*100, appInfo.IsFailed)
assert.NilError(t, err, "App should be in Failed state")
}
+
+func TestAppStateTransitionEvents(t *testing.T) {
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl) //nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+ // Accept only from new
+ appInfo := newApplication("app-00001", "default", "root.a")
+ assert.Equal(t, appInfo.CurrentState(), New.String())
+
+ // new to accepted
+ err := appInfo.HandleApplicationEvent(RunApplication)
+ assert.NilError(t, err, "no error expected new to accepted")
+ assert.Equal(t, appInfo.CurrentState(), Accepted.String())
+
+ // accepted to completing
+ err = appInfo.HandleApplicationEvent(CompleteApplication)
+ assert.NilError(t, err, "no error expected accepted to completing")
+ assert.Equal(t, appInfo.CurrentState(), Completing.String())
+
+ // completing to run
+ err = appInfo.HandleApplicationEvent(RunApplication)
+ assert.NilError(t, err, "no error expected starting to running (completing test)")
+
+ // run to failing
+ err = appInfo.HandleApplicationEvent(FailApplication)
+ assert.NilError(t, err, "no error expected new to failing")
+ err = common.WaitFor(10*time.Microsecond, time.Millisecond*100, appInfo.IsFailing)
+ assert.NilError(t, err, "App should be in Failing state")
+
+ // failing to failed
+ err = appInfo.HandleApplicationEvent(FailApplication)
+ assert.NilError(t, err, "no error expected failing to failed")
+ assert.Assert(t, appInfo.IsFailed(), "App should be in Failed state")
+
+ // Verify application events
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ fmt.Printf("checking event length: %d\n", eventSystem.Store.CountStoredEvents())
+ return eventSystem.Store.CountStoredEvents() == 6
+ })
+ assert.NilError(t, err, "the event should have been processed")
+ records := eventSystem.Store.CollectEvents()
+ if records == nil {
+ t.Fatal("collecting eventChannel should return something")
+ }
+ assert.Equal(t, 6, len(records), "expecting 6 events")
+ isNewApplicationEvent(t, appInfo, records[0])
+ isStateChangeEvent(t, appInfo, si.EventRecord_APP_ACCEPTED, records[1])
+ isStateChangeEvent(t, appInfo, si.EventRecord_APP_COMPLETING, records[2])
+ isStateChangeEvent(t, appInfo, si.EventRecord_APP_RUNNING, records[3])
+ isStateChangeEvent(t, appInfo, si.EventRecord_APP_FAILING, records[4])
+ isStateChangeEvent(t, appInfo, si.EventRecord_APP_FAILED, records[5])
+}
diff --git a/pkg/scheduler/objects/application_test.go b/pkg/scheduler/objects/application_test.go
index bf76da16..6e3f838d 100644
--- a/pkg/scheduler/objects/application_test.go
+++ b/pkg/scheduler/objects/application_test.go
@@ -200,11 +200,13 @@ func TestAppReservation(t *testing.T) {
// test multiple reservations from one allocation
func TestAppAllocReservation(t *testing.T) {
+ app := newApplication(appID1, "default", "root.unknown")
+ // Create event system after new application to avoid new application event.
events.CreateAndSetEventSystem()
eventSystem := events.GetEventSystem().(*events.EventSystemImpl) //nolint:errcheck
eventSystem.StartServiceWithPublisher(false)
-
- app := newApplication(appID1, "default", "root.unknown")
+ app.disableStateChangeEvents()
+ app.resetAppEvents()
if app == nil || app.ApplicationID != appID1 {
t.Fatalf("app create failed which should not have %v", app)
}
@@ -342,11 +344,13 @@ func TestUpdateRepeat(t *testing.T) {
// test pending calculation and ask addition
func TestAddAllocAsk(t *testing.T) {
+ app := newApplication(appID1, "default", "root.unknown")
+ // Create event system after new application to avoid new application event.
events.CreateAndSetEventSystem()
eventSystem := events.GetEventSystem().(*events.EventSystemImpl) //nolint:errcheck
eventSystem.StartServiceWithPublisher(false)
-
- app := newApplication(appID1, "default", "root.unknown")
+ app.disableStateChangeEvents()
+ app.resetAppEvents()
if app == nil || app.ApplicationID != appID1 {
t.Fatalf("app create failed which should not have %v", app)
}
@@ -1235,9 +1239,6 @@ func TestOnStatusChangeCalled(t *testing.T) {
func TestReplaceAllocation(t *testing.T) {
setupUGM()
- events.CreateAndSetEventSystem()
- eventSystem := events.GetEventSystem().(*events.EventSystemImpl) //nolint:errcheck
- eventSystem.StartServiceWithPublisher(false)
app := newApplication(appID1, "default", "root.a")
assert.Equal(t, New.String(), app.CurrentState(), "new app not in New state")
@@ -1417,9 +1418,6 @@ func runTimeoutPlaceholderTest(t *testing.T, expectedState string, gangSchedulin
func TestTimeoutPlaceholderAllocReleased(t *testing.T) {
setupUGM()
- events.CreateAndSetEventSystem()
- eventSystem := events.GetEventSystem().(*events.EventSystemImpl) //nolint:errcheck
- eventSystem.StartServiceWithPublisher(false)
originalPhTimeout := defaultPlaceholderTimeout
defaultPlaceholderTimeout = 5 * time.Millisecond
@@ -1973,11 +1971,13 @@ func TestMaxAskPriority(t *testing.T) {
}
func TestAskEvents(t *testing.T) {
+ app := newApplication(appID1, "default", "root.default")
+ // Create event system after new application to avoid new app event.
events.CreateAndSetEventSystem()
eventSystem := events.GetEventSystem().(*events.EventSystemImpl) //nolint:errcheck
eventSystem.StartServiceWithPublisher(false)
-
- app := newApplication(appID1, "default", "root.default")
+ app.disableStateChangeEvents()
+ app.resetAppEvents()
queue, err := createRootQueue(nil)
assert.NilError(t, err, "queue create failed")
app.queue = queue
@@ -2037,11 +2037,13 @@ func TestAskEvents(t *testing.T) {
}
func TestAllocationEvents(t *testing.T) { //nolint:funlen
+ app := newApplication(appID1, "default", "root.default")
+ // Create event system after new application to avoid new app event.
events.CreateAndSetEventSystem()
eventSystem := events.GetEventSystem().(*events.EventSystemImpl) //nolint:errcheck
eventSystem.StartServiceWithPublisher(false)
-
- app := newApplication(appID1, "default", "root.default")
+ app.disableStateChangeEvents()
+ app.resetAppEvents()
queue, err := createRootQueue(nil)
assert.NilError(t, err, "queue create failed")
app.queue = queue
@@ -2147,10 +2149,6 @@ func TestAllocationEvents(t *testing.T) { //nolint:funlen
}
func TestPlaceholderLargerEvent(t *testing.T) {
- events.CreateAndSetEventSystem()
- eventSystem := events.GetEventSystem().(*events.EventSystemImpl) //nolint:errcheck
- eventSystem.StartServiceWithPublisher(false)
-
resMap := map[string]string{"memory": "100", "vcores": "10"}
res, err := resources.NewResourceFromConf(resMap)
assert.NilError(t, err, "failed to create resource with error")
@@ -2159,6 +2157,12 @@ func TestPlaceholderLargerEvent(t *testing.T) {
assert.NilError(t, err, "failed to create resource with error")
app := newApplication(appID1, "default", "root.default")
+ // Create event system after new application to avoid new application event.
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl) //nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+ app.disableStateChangeEvents()
+ app.resetAppEvents()
queue, err := createRootQueue(nil)
assert.NilError(t, err, "queue create failed")
app.queue = queue
@@ -2192,10 +2196,6 @@ func TestPlaceholderLargerEvent(t *testing.T) {
}
func TestAppDoesNotFitEvent(t *testing.T) {
- events.CreateAndSetEventSystem()
- eventSystem := events.GetEventSystem().(*events.EventSystemImpl) //nolint:errcheck
- eventSystem.StartServiceWithPublisher(false)
-
resMap := map[string]string{"memory": "100", "vcores": "10"}
res, err := resources.NewResourceFromConf(resMap)
assert.NilError(t, err, "failed to create resource with error")
@@ -2204,6 +2204,12 @@ func TestAppDoesNotFitEvent(t *testing.T) {
assert.NilError(t, err, "failed to create resource with error")
ask := newAllocationAsk("alloc-0", "app-1", res)
app := newApplication(appID1, "default", "root.default")
+ // Create event system after new application to avoid new application event.
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl) //nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+ app.disableStateChangeEvents()
+ app.resetAppEvents()
queue, err := createRootQueue(nil)
assert.NilError(t, err, "queue create failed")
app.queue = queue
@@ -2257,3 +2263,15 @@ func (sa *Application) handleApplicationEventWithInfoLocking(event applicationEv
defer sa.Unlock()
return sa.HandleApplicationEventWithInfo(event, info)
}
+
+func (sa *Application) disableStateChangeEvents() {
+ sa.Lock()
+ defer sa.Unlock()
+ sa.sendStateChangeEvents = false
+}
+
+func (sa *Application) resetAppEvents() {
+ sa.Lock()
+ defer sa.Unlock()
+ sa.appEvents = newApplicationEvents(sa, events.GetEventSystem())
+}
diff --git a/pkg/scheduler/objects/queue.go b/pkg/scheduler/objects/queue.go
index e30d9bb6..55b94306 100644
--- a/pkg/scheduler/objects/queue.go
+++ b/pkg/scheduler/objects/queue.go
@@ -786,6 +786,7 @@ func (sq *Queue) RemoveApplication(app *Application) {
delete(sq.allocatingAcceptedApps, appID)
priority := sq.recalculatePriority()
sq.Unlock()
+ app.appEvents.sendRemoveApplicationEvent()
sq.parent.UpdateQueuePriority(sq.Name, priority)
diff --git a/pkg/scheduler/objects/queue_test.go b/pkg/scheduler/objects/queue_test.go
index 928ad054..6e1efb1b 100644
--- a/pkg/scheduler/objects/queue_test.go
+++ b/pkg/scheduler/objects/queue_test.go
@@ -39,6 +39,8 @@ import (
"github.com/apache/yunikorn-scheduler-interface/lib/go/si"
)
+const ZeroResource string = "{\"resources\":{\"first\":{\"value\":0}}}"
+
// base test for creating a managed queue
func TestQueueBasics(t *testing.T) {
// create the root
@@ -419,7 +421,7 @@ func TestAddApplicationWithTag(t *testing.T) {
}
// set max to illegal limit (0 value), but guarantee not 0
- tags[siCommon.AppTagNamespaceResourceQuota] = "{\"resources\":{\"first\":{\"value\":0}}}"
+ tags[siCommon.AppTagNamespaceResourceQuota] = ZeroResource
app = newApplicationWithTags("app-5", "default", "root.leaf-un", tags)
leafUn.AddApplication(app)
@@ -432,7 +434,7 @@ func TestAddApplicationWithTag(t *testing.T) {
}
// set guaranteed resource to illegal limit (0 value), but max resource not 0
- tags[siCommon.AppTagNamespaceResourceGuaranteed] = "{\"resources\":{\"first\":{\"value\":0}}}"
+ tags[siCommon.AppTagNamespaceResourceGuaranteed] = ZeroResource
tags[siCommon.AppTagNamespaceResourceQuota] = "{\"resources\":{\"first\":{\"value\":100}}}"
app = newApplicationWithTags("app-6", "default", "root.leaf-un", tags)
@@ -478,8 +480,8 @@ func TestAddApplicationWithTag(t *testing.T) {
}
// set both to 0 , the max resource and guaranteed resource will not update new value
- tags[siCommon.AppTagNamespaceResourceGuaranteed] = "{\"resources\":{\"first\":{\"value\":0}}}"
- tags[siCommon.AppTagNamespaceResourceQuota] = "{\"resources\":{\"first\":{\"value\":0}}}"
+ tags[siCommon.AppTagNamespaceResourceGuaranteed] = ZeroResource
+ tags[siCommon.AppTagNamespaceResourceQuota] = ZeroResource
app = newApplicationWithTags("app-9", "default", "root.leaf-un", tags)
leafUn.AddApplication(app)
assert.Equal(t, len(leafUn.applications), 7, "Application was not added to the Dynamic queue as expected")
@@ -2555,17 +2557,18 @@ func TestQueueEvents(t *testing.T) {
noEvents := 0
err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
noEvents = eventSystem.Store.CountStoredEvents()
- return noEvents == 3
+ return noEvents == 5
})
- assert.NilError(t, err, "expected 2 events, got %d", noEvents)
+ assert.NilError(t, err, "expected 5 events, got %d", noEvents)
records := eventSystem.Store.CollectEvents()
- assert.Equal(t, 3, len(records), "number of events")
- assert.Equal(t, si.EventRecord_QUEUE, records[1].Type)
- assert.Equal(t, si.EventRecord_ADD, records[1].EventChangeType)
- assert.Equal(t, si.EventRecord_QUEUE_APP, records[1].EventChangeDetail)
+ assert.Equal(t, 5, len(records), "number of events")
assert.Equal(t, si.EventRecord_QUEUE, records[2].Type)
- assert.Equal(t, si.EventRecord_REMOVE, records[2].EventChangeType)
+ assert.Equal(t, si.EventRecord_ADD, records[2].EventChangeType)
assert.Equal(t, si.EventRecord_QUEUE_APP, records[2].EventChangeDetail)
+ assert.Equal(t, si.EventRecord_QUEUE, records[3].Type)
+ assert.Equal(t, si.EventRecord_REMOVE, records[3].EventChangeType)
+ assert.Equal(t, si.EventRecord_QUEUE_APP, records[3].EventChangeDetail)
+ isRemoveApplicationEvent(t, app, records[4])
newConf := configs.QueueConfig{
Parent: false,
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@yunikorn.apache.org
For additional commands, e-mail: issues-help@yunikorn.apache.org