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 3917da66 [YUNIKORN-2567] Remove Application reference from
applicationEvents (#881)
3917da66 is described below
commit 3917da665fc023a86f0b66e234f972cc887dba3b
Author: Peter Bacsko <[email protected]>
AuthorDate: Thu May 30 21:02:19 2024 +0200
[YUNIKORN-2567] Remove Application reference from applicationEvents (#881)
Closes: #881
Signed-off-by: Peter Bacsko <[email protected]>
---
pkg/scheduler/objects/application.go | 12 +-
pkg/scheduler/objects/application_events.go | 47 ++++----
pkg/scheduler/objects/application_events_test.go | 138 +++++++----------------
pkg/scheduler/objects/application_state.go | 4 +-
pkg/scheduler/objects/application_test.go | 4 +-
pkg/scheduler/objects/queue.go | 2 +-
6 files changed, 77 insertions(+), 130 deletions(-)
diff --git a/pkg/scheduler/objects/application.go
b/pkg/scheduler/objects/application.go
index b76466be..80769855 100644
--- a/pkg/scheduler/objects/application.go
+++ b/pkg/scheduler/objects/application.go
@@ -187,8 +187,8 @@ func NewApplication(siApp *si.AddApplicationRequest, ugi
security.UserGroup, eve
app.user = ugi
app.rmEventHandler = eventHandler
app.rmID = rmID
- app.appEvents = newApplicationEvents(app, events.GetEventSystem())
- app.appEvents.sendNewApplicationEvent()
+ app.appEvents = newApplicationEvents(events.GetEventSystem())
+ app.appEvents.sendNewApplicationEvent(app.ApplicationID)
return app
}
@@ -2106,12 +2106,12 @@ func (sa *Application)
updateRunnableStatus(runnableInQueue, runnableByUserLimit
log.Log(log.SchedApplication).Info("Application is now
runnable in queue",
zap.String("appID", sa.ApplicationID),
zap.String("queue", sa.queuePath))
- sa.appEvents.sendAppRunnableInQueueEvent()
+
sa.appEvents.sendAppRunnableInQueueEvent(sa.ApplicationID)
} else {
log.Log(log.SchedApplication).Info("Maximum number of
running applications reached the queue limit",
zap.String("appID", sa.ApplicationID),
zap.String("queue", sa.queuePath))
- sa.appEvents.sendAppNotRunnableInQueueEvent()
+
sa.appEvents.sendAppNotRunnableInQueueEvent(sa.ApplicationID)
}
}
sa.runnableInQueue = runnableInQueue
@@ -2123,14 +2123,14 @@ func (sa *Application)
updateRunnableStatus(runnableInQueue, runnableByUserLimit
zap.String("queue", sa.queuePath),
zap.String("user", sa.user.User),
zap.Strings("groups", sa.user.Groups))
- sa.appEvents.sendAppRunnableQuotaEvent()
+ sa.appEvents.sendAppRunnableQuotaEvent(sa.ApplicationID)
} else {
log.Log(log.SchedApplication).Info("Maximum number of
running applications reached the user/group limit",
zap.String("appID", sa.ApplicationID),
zap.String("queue", sa.queuePath),
zap.String("user", sa.user.User),
zap.Strings("groups", sa.user.Groups))
- sa.appEvents.sendAppNotRunnableQuotaEvent()
+
sa.appEvents.sendAppNotRunnableQuotaEvent(sa.ApplicationID)
}
}
sa.runnableByUserLimit = runnableByUserLimit
diff --git a/pkg/scheduler/objects/application_events.go
b/pkg/scheduler/objects/application_events.go
index bb94ffbf..04fe51a0 100644
--- a/pkg/scheduler/objects/application_events.go
+++ b/pkg/scheduler/objects/application_events.go
@@ -20,7 +20,6 @@ package objects
import (
"fmt"
-
"github.com/apache/yunikorn-core/pkg/common"
"github.com/apache/yunikorn-core/pkg/events"
"github.com/apache/yunikorn-scheduler-interface/lib/go/si"
@@ -28,15 +27,14 @@ import (
type applicationEvents struct {
eventSystem events.EventSystem
- app *Application
}
func (evt *applicationEvents) sendPlaceholderLargerEvent(ph *Allocation,
request *AllocationAsk) {
if !evt.eventSystem.IsEventTrackingEnabled() {
return
}
- message := fmt.Sprintf("Task group '%s' in application '%s': allocation
resources '%s' are not matching placeholder '%s' allocation with ID '%s'",
ph.GetTaskGroup(), evt.app.ApplicationID, request.GetAllocatedResource(),
ph.GetAllocatedResource(), ph.GetAllocationKey())
- event := events.CreateRequestEventRecord(ph.GetAllocationKey(),
evt.app.ApplicationID, message, request.GetAllocatedResource())
+ message := fmt.Sprintf("Task group '%s' in application '%s': allocation
resources '%s' are not matching placeholder '%s' allocation with ID '%s'",
ph.GetTaskGroup(), ph.GetApplicationID(), request.GetAllocatedResource(),
ph.GetAllocatedResource(), ph.GetAllocationKey())
+ event := events.CreateRequestEventRecord(ph.GetAllocationKey(),
ph.GetApplicationID(), message, request.GetAllocatedResource())
evt.eventSystem.AddEvent(event)
}
@@ -44,7 +42,7 @@ func (evt *applicationEvents) sendNewAllocationEvent(alloc
*Allocation) {
if !evt.eventSystem.IsEventTrackingEnabled() {
return
}
- event := events.CreateAppEventRecord(evt.app.ApplicationID,
common.Empty, alloc.GetAllocationKey(), si.EventRecord_ADD,
si.EventRecord_APP_ALLOC, alloc.GetAllocatedResource())
+ event := events.CreateAppEventRecord(alloc.GetApplicationID(),
common.Empty, alloc.GetAllocationKey(), si.EventRecord_ADD,
si.EventRecord_APP_ALLOC, alloc.GetAllocatedResource())
evt.eventSystem.AddEvent(event)
}
@@ -52,7 +50,7 @@ func (evt *applicationEvents) sendNewAskEvent(request
*AllocationAsk) {
if !evt.eventSystem.IsEventTrackingEnabled() {
return
}
- event := events.CreateAppEventRecord(evt.app.ApplicationID,
common.Empty, request.GetAllocationKey(), si.EventRecord_ADD,
si.EventRecord_APP_REQUEST, request.GetAllocatedResource())
+ event := events.CreateAppEventRecord(request.GetApplicationID(),
common.Empty, request.GetAllocationKey(), si.EventRecord_ADD,
si.EventRecord_APP_REQUEST, request.GetAllocatedResource())
evt.eventSystem.AddEvent(event)
}
@@ -75,7 +73,7 @@ func (evt *applicationEvents) sendRemoveAllocationEvent(alloc
*Allocation, termi
eventChangeDetail = si.EventRecord_ALLOC_REPLACED
}
- event := events.CreateAppEventRecord(evt.app.ApplicationID,
common.Empty, alloc.GetAllocationKey(), si.EventRecord_REMOVE,
eventChangeDetail, alloc.GetAllocatedResource())
+ event := events.CreateAppEventRecord(alloc.GetApplicationID(),
common.Empty, alloc.GetAllocationKey(), si.EventRecord_REMOVE,
eventChangeDetail, alloc.GetAllocatedResource())
evt.eventSystem.AddEvent(event)
}
@@ -83,69 +81,68 @@ func (evt *applicationEvents) sendRemoveAskEvent(request
*AllocationAsk, detail
if !evt.eventSystem.IsEventTrackingEnabled() {
return
}
- event := events.CreateAppEventRecord(evt.app.ApplicationID,
common.Empty, request.GetAllocationKey(), si.EventRecord_REMOVE, detail,
request.GetAllocatedResource())
+ event := events.CreateAppEventRecord(request.GetApplicationID(),
common.Empty, request.GetAllocationKey(), si.EventRecord_REMOVE, detail,
request.GetAllocatedResource())
evt.eventSystem.AddEvent(event)
}
-func (evt *applicationEvents) sendNewApplicationEvent() {
+func (evt *applicationEvents) sendNewApplicationEvent(appID string) {
if !evt.eventSystem.IsEventTrackingEnabled() {
return
}
- event := events.CreateAppEventRecord(evt.app.ApplicationID,
common.Empty, common.Empty, si.EventRecord_ADD, si.EventRecord_APP_NEW,
evt.app.allocatedResource)
+ event := events.CreateAppEventRecord(appID, common.Empty, common.Empty,
si.EventRecord_ADD, si.EventRecord_APP_NEW, nil)
evt.eventSystem.AddEvent(event)
}
-func (evt *applicationEvents) sendRemoveApplicationEvent() {
+func (evt *applicationEvents) sendRemoveApplicationEvent(appID string) {
if !evt.eventSystem.IsEventTrackingEnabled() {
return
}
- event := events.CreateAppEventRecord(evt.app.ApplicationID,
common.Empty, common.Empty, si.EventRecord_REMOVE, si.EventRecord_DETAILS_NONE,
evt.app.allocatedResource)
+ event := events.CreateAppEventRecord(appID, common.Empty, common.Empty,
si.EventRecord_REMOVE, si.EventRecord_DETAILS_NONE, nil)
evt.eventSystem.AddEvent(event)
}
-func (evt *applicationEvents) sendStateChangeEvent(changeDetail
si.EventRecord_ChangeDetail, eventInfo string) {
- if !evt.eventSystem.IsEventTrackingEnabled() ||
!evt.app.sendStateChangeEvents {
+func (evt *applicationEvents) sendStateChangeEvent(appID string, changeDetail
si.EventRecord_ChangeDetail, eventInfo string) {
+ if !evt.eventSystem.IsEventTrackingEnabled() {
return
}
- event := events.CreateAppEventRecord(evt.app.ApplicationID, eventInfo,
common.Empty, si.EventRecord_SET, changeDetail, evt.app.allocatedResource)
+ event := events.CreateAppEventRecord(appID, eventInfo, common.Empty,
si.EventRecord_SET, changeDetail, nil)
evt.eventSystem.AddEvent(event)
}
-func (evt *applicationEvents) sendAppNotRunnableInQueueEvent() {
+func (evt *applicationEvents) sendAppNotRunnableInQueueEvent(appID string) {
if !evt.eventSystem.IsEventTrackingEnabled() {
return
}
- event := events.CreateAppEventRecord(evt.app.ApplicationID,
common.Empty, common.Empty, si.EventRecord_NONE,
si.EventRecord_APP_CANNOTRUN_QUEUE, nil)
+ event := events.CreateAppEventRecord(appID, common.Empty, common.Empty,
si.EventRecord_NONE, si.EventRecord_APP_CANNOTRUN_QUEUE, nil)
evt.eventSystem.AddEvent(event)
}
-func (evt *applicationEvents) sendAppRunnableInQueueEvent() {
+func (evt *applicationEvents) sendAppRunnableInQueueEvent(appID string) {
if !evt.eventSystem.IsEventTrackingEnabled() {
return
}
- event := events.CreateAppEventRecord(evt.app.ApplicationID,
common.Empty, common.Empty, si.EventRecord_NONE,
si.EventRecord_APP_RUNNABLE_QUEUE, nil)
+ event := events.CreateAppEventRecord(appID, common.Empty, common.Empty,
si.EventRecord_NONE, si.EventRecord_APP_RUNNABLE_QUEUE, nil)
evt.eventSystem.AddEvent(event)
}
-func (evt *applicationEvents) sendAppNotRunnableQuotaEvent() {
+func (evt *applicationEvents) sendAppNotRunnableQuotaEvent(appID string) {
if !evt.eventSystem.IsEventTrackingEnabled() {
return
}
- event := events.CreateAppEventRecord(evt.app.ApplicationID,
common.Empty, common.Empty, si.EventRecord_NONE,
si.EventRecord_APP_CANNOTRUN_QUOTA, nil)
+ event := events.CreateAppEventRecord(appID, common.Empty, common.Empty,
si.EventRecord_NONE, si.EventRecord_APP_CANNOTRUN_QUOTA, nil)
evt.eventSystem.AddEvent(event)
}
-func (evt *applicationEvents) sendAppRunnableQuotaEvent() {
+func (evt *applicationEvents) sendAppRunnableQuotaEvent(appID string) {
if !evt.eventSystem.IsEventTrackingEnabled() {
return
}
- event := events.CreateAppEventRecord(evt.app.ApplicationID,
common.Empty, common.Empty, si.EventRecord_NONE,
si.EventRecord_APP_RUNNABLE_QUOTA, nil)
+ event := events.CreateAppEventRecord(appID, common.Empty, common.Empty,
si.EventRecord_NONE, si.EventRecord_APP_RUNNABLE_QUOTA, nil)
evt.eventSystem.AddEvent(event)
}
-func newApplicationEvents(app *Application, evt events.EventSystem)
*applicationEvents {
+func newApplicationEvents(evt events.EventSystem) *applicationEvents {
return &applicationEvents{
eventSystem: evt,
- app: app,
}
}
diff --git a/pkg/scheduler/objects/application_events_test.go
b/pkg/scheduler/objects/application_events_test.go
index 8cb10628..94f6b40b 100644
--- a/pkg/scheduler/objects/application_events_test.go
+++ b/pkg/scheduler/objects/application_events_test.go
@@ -50,16 +50,13 @@ func isStateChangeEvent(t *testing.T, app *Application,
changeDetail si.EventRec
}
func TestSendPlaceholderLargerEvent(t *testing.T) {
- app := &Application{
- queuePath: "root.test",
- }
eventSystem := mock.NewEventSystemDisabled()
- appEvents := newApplicationEvents(app, eventSystem)
+ appEvents := newApplicationEvents(eventSystem)
appEvents.sendPlaceholderLargerEvent(&Allocation{}, &AllocationAsk{})
assert.Equal(t, 0, len(eventSystem.Events), "unexpected event")
eventSystem = mock.NewEventSystem()
- appEvents = newApplicationEvents(app, eventSystem)
+ appEvents = newApplicationEvents(eventSystem)
appEvents.sendPlaceholderLargerEvent(&Allocation{
allocationKey: aKey,
}, &AllocationAsk{
@@ -70,17 +67,13 @@ func TestSendPlaceholderLargerEvent(t *testing.T) {
}
func TestSendNewAllocationEvent(t *testing.T) {
- app := &Application{
- ApplicationID: appID0,
- queuePath: "root.test",
- }
eventSystem := mock.NewEventSystemDisabled()
- appEvents := newApplicationEvents(app, eventSystem)
+ appEvents := newApplicationEvents(eventSystem)
appEvents.sendNewAllocationEvent(&Allocation{})
assert.Equal(t, 0, len(eventSystem.Events), "unexpected event")
eventSystem = mock.NewEventSystem()
- appEvents = newApplicationEvents(app, eventSystem)
+ appEvents = newApplicationEvents(eventSystem)
assert.Assert(t, appEvents.eventSystem != nil, "event system should not
be nil")
appEvents.sendNewAllocationEvent(&Allocation{
applicationID: appID0,
@@ -96,17 +89,13 @@ func TestSendNewAllocationEvent(t *testing.T) {
}
func TestSendNewAskEvent(t *testing.T) {
- app := &Application{
- ApplicationID: appID0,
- queuePath: "root.test",
- }
eventSystem := mock.NewEventSystemDisabled()
- appEvents := newApplicationEvents(app, eventSystem)
+ appEvents := newApplicationEvents(eventSystem)
appEvents.sendNewAskEvent(&AllocationAsk{})
assert.Equal(t, 0, len(eventSystem.Events), "unexpected event")
eventSystem = mock.NewEventSystem()
- appEvents = newApplicationEvents(app, eventSystem)
+ appEvents = newApplicationEvents(eventSystem)
assert.Assert(t, appEvents.eventSystem != nil, "event system should not
be nil")
appEvents.sendNewAskEvent(&AllocationAsk{
applicationID: appID0,
@@ -122,12 +111,8 @@ func TestSendNewAskEvent(t *testing.T) {
}
func TestSendRemoveAllocationEvent(t *testing.T) {
- app := &Application{
- ApplicationID: appID0,
- queuePath: "root.test",
- }
eventSystem := mock.NewEventSystemDisabled()
- appEvents := newApplicationEvents(app, eventSystem)
+ appEvents := newApplicationEvents(eventSystem)
appEvents.sendRemoveAllocationEvent(&Allocation{},
si.TerminationType_STOPPED_BY_RM)
assert.Equal(t, 0, len(eventSystem.Events), "unexpected event")
@@ -207,11 +192,11 @@ func TestSendRemoveAllocationEvent(t *testing.T) {
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
if testCase.eventSystemMock == nil {
- appEvents := newApplicationEvents(app, nil)
+ appEvents := newApplicationEvents(nil)
assert.Assert(t, appEvents.eventSystem == nil,
"event system should be nil")
appEvents.sendRemoveAllocationEvent(testCase.allocation,
testCase.terminationType)
} else {
- appEvents := newApplicationEvents(app,
testCase.eventSystemMock)
+ appEvents :=
newApplicationEvents(testCase.eventSystemMock)
assert.Assert(t, appEvents.eventSystem != nil,
"event system should not be nil")
appEvents.sendRemoveAllocationEvent(testCase.allocation,
testCase.terminationType)
assert.Equal(t, testCase.expectedEventCnt,
len(testCase.eventSystemMock.Events), "event was not generated")
@@ -227,12 +212,8 @@ func TestSendRemoveAllocationEvent(t *testing.T) {
}
func TestSendRemoveAskEvent(t *testing.T) {
- app := &Application{
- ApplicationID: appID0,
- queuePath: "root.test",
- }
eventSystem := mock.NewEventSystemDisabled()
- appEvents := newApplicationEvents(app, eventSystem)
+ appEvents := newApplicationEvents(eventSystem)
appEvents.sendRemoveAskEvent(&AllocationAsk{},
si.EventRecord_REQUEST_CANCEL)
assert.Equal(t, 0, len(eventSystem.Events), "unexpected event")
@@ -240,7 +221,7 @@ func TestSendRemoveAskEvent(t *testing.T) {
applicationID: appID0,
allocationKey: aKey}
eventSystem = mock.NewEventSystem()
- appEvents = newApplicationEvents(app, eventSystem)
+ appEvents = newApplicationEvents(eventSystem)
appEvents.sendRemoveAskEvent(ask, si.EventRecord_REQUEST_CANCEL)
event := eventSystem.Events[0]
assert.Equal(t, si.EventRecord_APP, event.Type)
@@ -262,18 +243,14 @@ func TestSendRemoveAskEvent(t *testing.T) {
}
func TestSendNewApplicationEvent(t *testing.T) {
- app := &Application{
- ApplicationID: appID0,
- queuePath: "root.test",
- }
eventSystem := mock.NewEventSystemDisabled()
- appEvents := newApplicationEvents(app, eventSystem)
- appEvents.sendNewApplicationEvent()
+ appEvents := newApplicationEvents(eventSystem)
+ appEvents.sendNewApplicationEvent(appID0)
assert.Equal(t, 0, len(eventSystem.Events), "unexpected event")
mockEvents := mock.NewEventSystem()
- appEvents = newApplicationEvents(app, mockEvents)
- appEvents.sendNewApplicationEvent()
+ appEvents = newApplicationEvents(mockEvents)
+ appEvents.sendNewApplicationEvent(appID0)
event := mockEvents.Events[0]
assert.Equal(t, si.EventRecord_APP, event.Type)
assert.Equal(t, si.EventRecord_ADD, event.EventChangeType)
@@ -284,18 +261,14 @@ func TestSendNewApplicationEvent(t *testing.T) {
}
func TestSendRemoveApplicationEvent(t *testing.T) {
- app := &Application{
- ApplicationID: appID0,
- queuePath: "root.test",
- }
eventSystem := mock.NewEventSystemDisabled()
- appEvents := newApplicationEvents(app, eventSystem)
- appEvents.sendRemoveApplicationEvent()
+ appEvents := newApplicationEvents(eventSystem)
+ appEvents.sendRemoveApplicationEvent(appID0)
assert.Equal(t, 0, len(eventSystem.Events), "unexpected event")
eventSystem = mock.NewEventSystem()
- appEvents = newApplicationEvents(app, eventSystem)
- appEvents.sendRemoveApplicationEvent()
+ appEvents = newApplicationEvents(eventSystem)
+ appEvents.sendRemoveApplicationEvent(appID0)
event := eventSystem.Events[0]
assert.Equal(t, si.EventRecord_APP, event.Type)
assert.Equal(t, si.EventRecord_REMOVE, event.EventChangeType)
@@ -306,19 +279,14 @@ func TestSendRemoveApplicationEvent(t *testing.T) {
}
func TestSendStateChangeEvent(t *testing.T) {
- app := &Application{
- ApplicationID: appID0,
- queuePath: "root.test",
- sendStateChangeEvents: true,
- }
eventSystem := mock.NewEventSystemDisabled()
- appEvents := newApplicationEvents(app, eventSystem)
- appEvents.sendStateChangeEvent(si.EventRecord_APP_RUNNING, "")
+ appEvents := newApplicationEvents(eventSystem)
+ appEvents.sendStateChangeEvent(appID0, si.EventRecord_APP_RUNNING, "")
assert.Equal(t, 0, len(eventSystem.Events), "unexpected event")
eventSystem = mock.NewEventSystem()
- appEvents = newApplicationEvents(app, eventSystem)
- appEvents.sendStateChangeEvent(si.EventRecord_APP_RUNNING, "The
application is running")
+ appEvents = newApplicationEvents(eventSystem)
+ appEvents.sendStateChangeEvent(appID0, si.EventRecord_APP_RUNNING, "The
application is running")
event := eventSystem.Events[0]
assert.Equal(t, si.EventRecord_APP, event.Type)
assert.Equal(t, si.EventRecord_SET, event.EventChangeType)
@@ -328,13 +296,13 @@ func TestSendStateChangeEvent(t *testing.T) {
assert.Equal(t, "The application is running", event.Message)
eventSystem = mock.NewEventSystemDisabled()
- appEvents = newApplicationEvents(app, eventSystem)
- appEvents.sendStateChangeEvent(si.EventRecord_APP_RUNNING,
"ResourceReservationTimeout")
+ appEvents = newApplicationEvents(eventSystem)
+ appEvents.sendStateChangeEvent(appID0, si.EventRecord_APP_RUNNING,
"ResourceReservationTimeout")
assert.Equal(t, 0, len(eventSystem.Events), "unexpected event")
eventSystem = mock.NewEventSystem()
- appEvents = newApplicationEvents(app, eventSystem)
- appEvents.sendStateChangeEvent(si.EventRecord_APP_REJECT, "Failed to
add application to partition (placement rejected)")
+ appEvents = newApplicationEvents(eventSystem)
+ appEvents.sendStateChangeEvent(appID0, si.EventRecord_APP_REJECT,
"Failed to add application to partition (placement rejected)")
event = eventSystem.Events[0]
assert.Equal(t, si.EventRecord_APP, event.Type)
assert.Equal(t, si.EventRecord_SET, event.EventChangeType)
@@ -345,19 +313,14 @@ func TestSendStateChangeEvent(t *testing.T) {
}
func TestSendAppCannotRunInQueueEvent(t *testing.T) {
- app := &Application{
- ApplicationID: appID0,
- queuePath: "root.test",
- sendStateChangeEvents: true,
- }
eventSystem := mock.NewEventSystemDisabled()
- appEvents := newApplicationEvents(app, eventSystem)
- appEvents.sendAppNotRunnableInQueueEvent()
+ appEvents := newApplicationEvents(eventSystem)
+ appEvents.sendAppNotRunnableInQueueEvent(appID0)
assert.Equal(t, 0, len(eventSystem.Events), "unexpected event")
eventSystem = mock.NewEventSystem()
- appEvents = newApplicationEvents(app, eventSystem)
- appEvents.sendAppNotRunnableInQueueEvent()
+ appEvents = newApplicationEvents(eventSystem)
+ appEvents.sendAppNotRunnableInQueueEvent(appID0)
event := eventSystem.Events[0]
assert.Equal(t, si.EventRecord_APP, event.Type)
assert.Equal(t, si.EventRecord_NONE, event.EventChangeType)
@@ -368,19 +331,14 @@ func TestSendAppCannotRunInQueueEvent(t *testing.T) {
}
func TestSendAppCannotRunByQuotaEvent(t *testing.T) {
- app := &Application{
- ApplicationID: appID0,
- queuePath: "root.test",
- sendStateChangeEvents: true,
- }
eventSystem := mock.NewEventSystemDisabled()
- appEvents := newApplicationEvents(app, eventSystem)
- appEvents.sendAppNotRunnableQuotaEvent()
+ appEvents := newApplicationEvents(eventSystem)
+ appEvents.sendAppNotRunnableQuotaEvent(appID0)
assert.Equal(t, 0, len(eventSystem.Events), "unexpected event")
eventSystem = mock.NewEventSystem()
- appEvents = newApplicationEvents(app, eventSystem)
- appEvents.sendAppNotRunnableQuotaEvent()
+ appEvents = newApplicationEvents(eventSystem)
+ appEvents.sendAppNotRunnableQuotaEvent(appID0)
event := eventSystem.Events[0]
assert.Equal(t, si.EventRecord_APP, event.Type)
assert.Equal(t, si.EventRecord_NONE, event.EventChangeType)
@@ -391,19 +349,14 @@ func TestSendAppCannotRunByQuotaEvent(t *testing.T) {
}
func TestSendAppRunnableInQueueEvent(t *testing.T) {
- app := &Application{
- ApplicationID: appID0,
- queuePath: "root.test",
- sendStateChangeEvents: true,
- }
eventSystem := mock.NewEventSystemDisabled()
- appEvents := newApplicationEvents(app, eventSystem)
- appEvents.sendAppRunnableInQueueEvent()
+ appEvents := newApplicationEvents(eventSystem)
+ appEvents.sendAppRunnableInQueueEvent(appID0)
assert.Equal(t, 0, len(eventSystem.Events), "unexpected event")
eventSystem = mock.NewEventSystem()
- appEvents = newApplicationEvents(app, eventSystem)
- appEvents.sendAppRunnableInQueueEvent()
+ appEvents = newApplicationEvents(eventSystem)
+ appEvents.sendAppRunnableInQueueEvent(appID0)
event := eventSystem.Events[0]
assert.Equal(t, si.EventRecord_APP, event.Type)
assert.Equal(t, si.EventRecord_NONE, event.EventChangeType)
@@ -414,19 +367,14 @@ func TestSendAppRunnableInQueueEvent(t *testing.T) {
}
func TestSendAppRunnableByQuotaEvent(t *testing.T) {
- app := &Application{
- ApplicationID: appID0,
- queuePath: "root.test",
- sendStateChangeEvents: true,
- }
eventSystem := mock.NewEventSystemDisabled()
- appEvents := newApplicationEvents(app, eventSystem)
- appEvents.sendAppRunnableQuotaEvent()
+ appEvents := newApplicationEvents(eventSystem)
+ appEvents.sendAppRunnableQuotaEvent(appID0)
assert.Equal(t, 0, len(eventSystem.Events), "unexpected event")
eventSystem = mock.NewEventSystem()
- appEvents = newApplicationEvents(app, eventSystem)
- appEvents.sendAppRunnableQuotaEvent()
+ appEvents = newApplicationEvents(eventSystem)
+ appEvents.sendAppRunnableQuotaEvent(appID0)
event := eventSystem.Events[0]
assert.Equal(t, si.EventRecord_APP, event.Type)
assert.Equal(t, si.EventRecord_NONE, event.EventChangeType)
diff --git a/pkg/scheduler/objects/application_state.go
b/pkg/scheduler/objects/application_state.go
index a23f61da..f0d89a95 100644
--- a/pkg/scheduler/objects/application_state.go
+++ b/pkg/scheduler/objects/application_state.go
@@ -158,7 +158,9 @@ func NewAppState() *fsm.FSM {
zap.String("state", event.Dst))
return
}
-
app.appEvents.sendStateChangeEvent(eventDetails, eventInfo)
+ if app.sendStateChangeEvents {
+
app.appEvents.sendStateChangeEvent(app.ApplicationID, eventDetails, eventInfo)
+ }
},
"leave_state": func(_ context.Context, event
*fsm.Event) {
event.Args[0].(*Application).clearStateTimer()
//nolint:errcheck
diff --git a/pkg/scheduler/objects/application_test.go
b/pkg/scheduler/objects/application_test.go
index 82939359..04353a48 100644
--- a/pkg/scheduler/objects/application_test.go
+++ b/pkg/scheduler/objects/application_test.go
@@ -2619,7 +2619,7 @@ func TestUpdateRunnableStatus(t *testing.T) {
assert.Assert(t, app.runnableInQueue)
assert.Assert(t, app.runnableByUserLimit)
eventSystem := mock.NewEventSystem()
- app.appEvents = newApplicationEvents(app, eventSystem)
+ app.appEvents = newApplicationEvents(eventSystem)
// App runnable - no events
app.updateRunnableStatus(true, true)
@@ -2747,5 +2747,5 @@ func (sa *Application) disableStateChangeEvents() {
func (sa *Application) resetAppEvents() {
sa.Lock()
defer sa.Unlock()
- sa.appEvents = newApplicationEvents(sa, events.GetEventSystem())
+ sa.appEvents = newApplicationEvents(events.GetEventSystem())
}
diff --git a/pkg/scheduler/objects/queue.go b/pkg/scheduler/objects/queue.go
index 4d0e416d..4a168926 100644
--- a/pkg/scheduler/objects/queue.go
+++ b/pkg/scheduler/objects/queue.go
@@ -797,7 +797,7 @@ func (sq *Queue) RemoveApplication(app *Application) {
delete(sq.allocatingAcceptedApps, appID)
priority := sq.recalculatePriority()
sq.Unlock()
- app.appEvents.sendRemoveApplicationEvent()
+ app.appEvents.sendRemoveApplicationEvent(appID)
sq.parent.UpdateQueuePriority(sq.Name, priority)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
