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 d765f3a4 [YUNIKORN-1801] add allocation events (#571)
d765f3a4 is described below
commit d765f3a4e1fff8607a27b2206be76e43df0cee2b
Author: Frank Yang <[email protected]>
AuthorDate: Thu Jun 22 15:40:22 2023 +0200
[YUNIKORN-1801] add allocation events (#571)
Signed-off-by: Frank Yang <[email protected]>
Closes: #571
Signed-off-by: Peter Bacsko <[email protected]>
---
pkg/scheduler/objects/application.go | 28 ++
pkg/scheduler/objects/application_events.go | 122 ++++++++
pkg/scheduler/objects/application_events_test.go | 345 +++++++++++++++++++++++
pkg/scheduler/objects/application_test.go | 123 +++++++-
pkg/scheduler/objects/events.go | 59 ----
pkg/scheduler/objects/events_test.go | 91 ------
pkg/scheduler/objects/utilities_test.go | 1 +
pkg/scheduler/partition_test.go | 183 +++++++++++-
8 files changed, 792 insertions(+), 160 deletions(-)
diff --git a/pkg/scheduler/objects/application.go
b/pkg/scheduler/objects/application.go
index 961f46cb..4de1764c 100644
--- a/pkg/scheduler/objects/application.go
+++ b/pkg/scheduler/objects/application.go
@@ -536,6 +536,7 @@ func (sa *Application) GetPendingResource()
*resources.Resource {
func (sa *Application) RemoveAllocationAsk(allocKey string) int {
sa.Lock()
defer sa.Unlock()
+ sa.sendRemoveAskEvent(allocKey)
return sa.removeAsksInternal(allocKey)
}
@@ -675,6 +676,7 @@ func (sa *Application) AddAllocationAsk(ask *AllocationAsk)
error {
zap.Stringer("pendingDelta", delta))
sa.sortedRequests.insert(ask)
+ sa.appEvents.sendNewAskEvent(ask)
return nil
}
@@ -1145,6 +1147,7 @@ func (sa *Application)
tryPlaceholderAllocate(nodeIterator func() NodeIterator,
log.Log(log.SchedApplication).Warn("ask
repeat update failed unexpectedly",
zap.Error(err))
}
+ sa.appEvents.sendRemoveAllocationEvent(ph,
si.TerminationType_PLACEHOLDER_REPLACED)
return alloc
}
}
@@ -1200,6 +1203,7 @@ func (sa *Application)
tryPlaceholderAllocate(nodeIterator func() NodeIterator,
log.Log(log.SchedApplication).Warn("ask repeat
update failed unexpectedly",
zap.Error(err))
}
+ sa.appEvents.sendRemoveAllocationEvent(phFit,
si.TerminationType_PLACEHOLDER_REPLACED)
return alloc
}
}
@@ -1476,6 +1480,8 @@ func (sa *Application) tryNode(node *Node, ask
*AllocationAsk) *Allocation {
}
// all is OK, last update for the app
sa.addAllocationInternal(alloc)
+
+ sa.appEvents.sendNewAllocationEvent(alloc)
// return allocation
return alloc
}
@@ -1770,6 +1776,11 @@ func (sa *Application) removeAllocationInternal(uuid
string, releaseType si.Term
}
}
delete(sa.allocations, uuid)
+ // If release count of placeholder is 0 and termination type is
PLACEHOLDER_REPLACED,
+ // the placeholder has already been replaced. No need to send remove
event.
+ if !(alloc.IsPlaceholder() && alloc.GetReleaseCount() == 0 &&
releaseType == si.TerminationType_PLACEHOLDER_REPLACED) {
+ sa.appEvents.sendRemoveAllocationEvent(alloc, releaseType)
+ }
return alloc
}
@@ -1900,6 +1911,7 @@ func (sa *Application) notifyRMAllocationReleased(rmID
string, released []*Alloc
TerminationType: terminationType,
Message: message,
})
+ sa.appEvents.sendRemoveAllocationEvent(alloc, terminationType)
}
sa.rmEventHandler.HandleEvent(releaseEvent)
// Wait from channel
@@ -1931,6 +1943,7 @@ func (sa *Application) notifyRMAllocationAskReleased(rmID
string, released []*Al
TerminationType: terminationType,
Message: message,
})
+ sa.appEvents.sendRemoveAskEvent(alloc, terminationType, false)
}
sa.rmEventHandler.HandleEvent(releaseEvent)
}
@@ -2009,3 +2022,18 @@ func (sa *Application)
SetTimedOutPlaceholder(taskGroupName string, timedOut int
sa.placeholderData[taskGroupName].TimedOut = timedOut
}
}
+
+func (sa *Application) sendRemoveAskEvent(allocKey string) {
+ if allocKey != "" {
+ // if allocKey is not empty, it means a specific ask is being
terminated
+ if ask := sa.requests[allocKey]; ask != nil {
+ sa.appEvents.sendRemoveAskEvent(ask,
si.TerminationType_STOPPED_BY_RM, false)
+ }
+ return
+ }
+
+ // if allocKey is empty, it means application is terminated
+ for _, ask := range sa.requests {
+ sa.appEvents.sendRemoveAskEvent(ask,
si.TerminationType_STOPPED_BY_RM, true)
+ }
+}
diff --git a/pkg/scheduler/objects/application_events.go
b/pkg/scheduler/objects/application_events.go
new file mode 100644
index 00000000..049ee626
--- /dev/null
+++ b/pkg/scheduler/objects/application_events.go
@@ -0,0 +1,122 @@
+/*
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+package objects
+
+import (
+ "fmt"
+
+ "github.com/apache/yunikorn-core/pkg/events"
+ "github.com/apache/yunikorn-scheduler-interface/lib/go/si"
+)
+
+type applicationEvents struct {
+ enabled bool
+ eventSystem events.EventSystem
+ app *Application
+}
+
+func (evt *applicationEvents) sendAppDoesNotFitEvent(request *AllocationAsk) {
+ if !evt.enabled {
+ return
+ }
+
+ message := fmt.Sprintf("Application %s does not fit into %s queue",
request.GetApplicationID(), evt.app.queuePath)
+ event := events.CreateRequestEventRecord(request.GetAllocationKey(),
request.GetApplicationID(), message, request.GetAllocatedResource())
+ evt.eventSystem.AddEvent(event)
+}
+
+func (evt *applicationEvents) sendPlaceholderLargerEvent(ph *Allocation,
request *AllocationAsk) {
+ if !evt.enabled {
+ 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().String(), ph.GetAllocatedResource().String(),
ph.GetAllocationKey())
+ event := events.CreateRequestEventRecord(ph.GetAllocationKey(),
evt.app.ApplicationID, message, request.GetAllocatedResource())
+ evt.eventSystem.AddEvent(event)
+}
+
+func (evt *applicationEvents) sendNewAllocationEvent(alloc *Allocation) {
+ if !evt.enabled {
+ return
+ }
+
+ event := events.CreateAppEventRecord(evt.app.ApplicationID, "",
alloc.GetUUID(), si.EventRecord_ADD, si.EventRecord_APP_ALLOC,
alloc.GetAllocatedResource())
+ evt.eventSystem.AddEvent(event)
+}
+
+func (evt *applicationEvents) sendNewAskEvent(request *AllocationAsk) {
+ if !evt.enabled {
+ return
+ }
+
+ event := events.CreateAppEventRecord(evt.app.ApplicationID, "",
request.GetAllocationKey(), si.EventRecord_ADD, si.EventRecord_APP_REQUEST,
request.GetAllocatedResource())
+ evt.eventSystem.AddEvent(event)
+}
+
+func (evt *applicationEvents) sendRemoveAllocationEvent(alloc *Allocation,
terminationType si.TerminationType) {
+ if !evt.enabled {
+ return
+ }
+
+ var eventChangeDetail si.EventRecord_ChangeDetail
+ switch terminationType {
+ case si.TerminationType_UNKNOWN_TERMINATION_TYPE:
+ eventChangeDetail = si.EventRecord_ALLOC_NODEREMOVED
+ case si.TerminationType_STOPPED_BY_RM:
+ eventChangeDetail = si.EventRecord_ALLOC_CANCEL
+ case si.TerminationType_TIMEOUT:
+ eventChangeDetail = si.EventRecord_ALLOC_TIMEOUT
+ case si.TerminationType_PREEMPTED_BY_SCHEDULER:
+ eventChangeDetail = si.EventRecord_ALLOC_PREEMPT
+ case si.TerminationType_PLACEHOLDER_REPLACED:
+ eventChangeDetail = si.EventRecord_ALLOC_REPLACED
+ }
+
+ event := events.CreateAppEventRecord(evt.app.ApplicationID, "",
alloc.GetUUID(), si.EventRecord_REMOVE, eventChangeDetail,
alloc.GetAllocatedResource())
+ evt.eventSystem.AddEvent(event)
+}
+
+func (evt *applicationEvents) sendRemoveAskEvent(request *AllocationAsk,
terminationType si.TerminationType, appRemoved bool) {
+ if !evt.enabled {
+ return
+ }
+
+ var eventChangeDetail si.EventRecord_ChangeDetail
+ switch terminationType {
+ case si.TerminationType_TIMEOUT:
+ eventChangeDetail = si.EventRecord_REQUEST_TIMEOUT
+ case si.TerminationType_STOPPED_BY_RM:
+ if appRemoved {
+ eventChangeDetail = si.EventRecord_REQUEST_CANCEL
+ } else {
+ eventChangeDetail = si.EventRecord_APP_REQUEST
+ }
+ }
+
+ event := events.CreateAppEventRecord(evt.app.ApplicationID, "",
request.GetAllocationKey(), si.EventRecord_REMOVE, eventChangeDetail,
request.GetAllocatedResource())
+ evt.eventSystem.AddEvent(event)
+}
+
+func newApplicationEvents(app *Application, evt events.EventSystem)
*applicationEvents {
+ return &applicationEvents{
+ eventSystem: evt,
+ enabled: evt != nil,
+ app: app,
+ }
+}
diff --git a/pkg/scheduler/objects/application_events_test.go
b/pkg/scheduler/objects/application_events_test.go
new file mode 100644
index 00000000..e8a284bb
--- /dev/null
+++ b/pkg/scheduler/objects/application_events_test.go
@@ -0,0 +1,345 @@
+/*
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+package objects
+
+import (
+ "testing"
+
+ "gotest.tools/v3/assert"
+
+ "github.com/apache/yunikorn-scheduler-interface/lib/go/si"
+)
+
+type EventSystemMock struct {
+ events []*si.EventRecord
+}
+
+func (m *EventSystemMock) AddEvent(event *si.EventRecord) {
+ m.events = append(m.events, event)
+}
+
+func (m *EventSystemMock) StartService() {}
+
+func (m *EventSystemMock) Stop() {}
+
+func newEventSystemMock() *EventSystemMock {
+ return &EventSystemMock{events: make([]*si.EventRecord, 0)}
+}
+
+func TestSendAppDoesNotFitEvent(t *testing.T) {
+ app := &Application{
+ 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.sendAppDoesNotFitEvent(&AllocationAsk{})
+
+ // enabled
+ mock := newEventSystemMock()
+ evt = newApplicationEvents(app, mock)
+ assert.Assert(t, evt.eventSystem != nil, "event system should not be
nil")
+ assert.Assert(t, evt.enabled, "event system should be enabled")
+ evt.sendAppDoesNotFitEvent(&AllocationAsk{
+ applicationID: appID0,
+ allocationKey: aKey,
+ })
+ assert.Equal(t, 1, len(mock.events), "event was not generated")
+}
+
+func TestSendPlaceholderLargerEvent(t *testing.T) {
+ app := &Application{
+ 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.sendPlaceholderLargerEvent(&Allocation{}, &AllocationAsk{})
+
+ // enabled
+ mock := newEventSystemMock()
+ evt = newApplicationEvents(app, mock)
+ assert.Assert(t, evt.eventSystem != nil, "event system should not be
nil")
+ assert.Assert(t, evt.enabled, "event system should be enabled")
+ evt.sendPlaceholderLargerEvent(&Allocation{
+ allocationKey: aKey,
+ }, &AllocationAsk{
+ applicationID: appID0,
+ allocationKey: aKey,
+ })
+ assert.Equal(t, 1, len(mock.events), "event was not generated")
+}
+
+func TestSendNewAllocationEvent(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.sendNewAllocationEvent(&Allocation{})
+
+ // enabled
+ mock := newEventSystemMock()
+ evt = newApplicationEvents(app, mock)
+ assert.Assert(t, evt.eventSystem != nil, "event system should not be
nil")
+ assert.Assert(t, evt.enabled, "event system should be enabled")
+ evt.sendNewAllocationEvent(&Allocation{
+ applicationID: appID0,
+ allocationKey: aKey,
+ uuid: aUUID,
+ })
+ assert.Equal(t, 1, len(mock.events), "event was not generated")
+ assert.Equal(t, si.EventRecord_APP, mock.events[0].Type, "event type is
not expected")
+ assert.Equal(t, si.EventRecord_ADD, mock.events[0].EventChangeType,
"event change type is not expected")
+ assert.Equal(t, si.EventRecord_APP_ALLOC,
mock.events[0].EventChangeDetail, "event change detail is not expected")
+ assert.Equal(t, appID0, mock.events[0].ObjectID, "event object id is
not expected")
+ assert.Equal(t, aUUID, mock.events[0].ReferenceID, "event reference id
is not expected")
+}
+
+func TestSendNewAskEvent(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.sendNewAskEvent(&AllocationAsk{})
+
+ // enabled
+ mock := newEventSystemMock()
+ evt = newApplicationEvents(app, mock)
+ assert.Assert(t, evt.eventSystem != nil, "event system should not be
nil")
+ assert.Assert(t, evt.enabled, "event system should be enabled")
+ evt.sendNewAskEvent(&AllocationAsk{
+ applicationID: appID0,
+ allocationKey: aKey,
+ })
+ assert.Equal(t, 1, len(mock.events), "event was not generated")
+ assert.Equal(t, si.EventRecord_APP, mock.events[0].Type, "event type is
not expected")
+ assert.Equal(t, si.EventRecord_ADD, mock.events[0].EventChangeType,
"event change type is not expected")
+ assert.Equal(t, si.EventRecord_APP_REQUEST,
mock.events[0].EventChangeDetail, "event change detail is not expected")
+ assert.Equal(t, appID0, mock.events[0].ObjectID, "event object id is
not expected")
+ assert.Equal(t, aKey, mock.events[0].ReferenceID, "event reference id
is not expected")
+}
+
+func TestSendRemoveAllocationEvent(t *testing.T) {
+ app := &Application{
+ ApplicationID: appID0,
+ queuePath: "root.test",
+ }
+
+ testCases := []struct {
+ name string
+ eventSystemMock *EventSystemMock
+ terminationType si.TerminationType
+ allocation *Allocation
+ expectedEventCnt int
+ expectedType si.EventRecord_Type
+ expectedChangeType si.EventRecord_ChangeType
+ expectedChangeDetail si.EventRecord_ChangeDetail
+ expectedObjectID string
+ expectedReferenceID string
+ }{
+ {
+ name: "disabled event system",
+ eventSystemMock: nil,
+ terminationType:
si.TerminationType_UNKNOWN_TERMINATION_TYPE,
+ allocation: &Allocation{},
+ },
+ {
+ name: "remove allocation cause of node
removal",
+ eventSystemMock: newEventSystemMock(),
+ terminationType:
si.TerminationType_UNKNOWN_TERMINATION_TYPE,
+ allocation: &Allocation{applicationID:
appID0, allocationKey: aKey, uuid: aUUID},
+ expectedEventCnt: 1,
+ expectedType: si.EventRecord_APP,
+ expectedChangeType: si.EventRecord_REMOVE,
+ expectedChangeDetail: si.EventRecord_ALLOC_NODEREMOVED,
+ expectedObjectID: appID0,
+ expectedReferenceID: aUUID,
+ },
+ {
+ name: "remove allocation cause of
resource manager cancel",
+ eventSystemMock: newEventSystemMock(),
+ terminationType: si.TerminationType_STOPPED_BY_RM,
+ allocation: &Allocation{applicationID:
appID0, allocationKey: aKey, uuid: aUUID},
+ expectedEventCnt: 1,
+ expectedType: si.EventRecord_APP,
+ expectedChangeType: si.EventRecord_REMOVE,
+ expectedChangeDetail: si.EventRecord_ALLOC_CANCEL,
+ expectedObjectID: appID0,
+ expectedReferenceID: aUUID,
+ },
+ {
+ name: "remove allocation cause of
timeout",
+ eventSystemMock: newEventSystemMock(),
+ terminationType: si.TerminationType_TIMEOUT,
+ allocation: &Allocation{applicationID:
appID0, allocationKey: aKey, uuid: aUUID},
+ expectedEventCnt: 1,
+ expectedType: si.EventRecord_APP,
+ expectedChangeType: si.EventRecord_REMOVE,
+ expectedChangeDetail: si.EventRecord_ALLOC_TIMEOUT,
+ expectedObjectID: appID0,
+ expectedReferenceID: aUUID,
+ },
+ {
+ name: "remove allocation cause of
preemption",
+ eventSystemMock: newEventSystemMock(),
+ terminationType:
si.TerminationType_PREEMPTED_BY_SCHEDULER,
+ allocation: &Allocation{applicationID:
appID0, allocationKey: aKey, uuid: aUUID},
+ expectedEventCnt: 1,
+ expectedType: si.EventRecord_APP,
+ expectedChangeType: si.EventRecord_REMOVE,
+ expectedChangeDetail: si.EventRecord_ALLOC_PREEMPT,
+ expectedObjectID: appID0,
+ expectedReferenceID: aUUID,
+ },
+ {
+ name: "remove allocation cause of
replacement",
+ eventSystemMock: newEventSystemMock(),
+ terminationType:
si.TerminationType_PLACEHOLDER_REPLACED,
+ allocation: &Allocation{applicationID:
appID0, allocationKey: aKey, uuid: aUUID},
+ expectedEventCnt: 1,
+ expectedType: si.EventRecord_APP,
+ expectedChangeType: si.EventRecord_REMOVE,
+ expectedChangeDetail: si.EventRecord_ALLOC_REPLACED,
+ expectedObjectID: appID0,
+ expectedReferenceID: aUUID,
+ },
+ }
+ for _, testCase := range testCases {
+ t.Run(testCase.name, func(t *testing.T) {
+ if testCase.eventSystemMock == nil {
+ 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.sendRemoveAllocationEvent(testCase.allocation, testCase.terminationType)
+ } else {
+ evt := newApplicationEvents(app,
testCase.eventSystemMock)
+ assert.Assert(t, evt.eventSystem != nil, "event
system should not be nil")
+ assert.Assert(t, evt.enabled, "event system
should be enabled")
+
evt.sendRemoveAllocationEvent(testCase.allocation, testCase.terminationType)
+ assert.Equal(t, testCase.expectedEventCnt,
len(testCase.eventSystemMock.events), "event was not generated")
+ assert.Equal(t, testCase.expectedType,
testCase.eventSystemMock.events[0].Type, "event type is not expected")
+ assert.Equal(t, testCase.expectedChangeType,
testCase.eventSystemMock.events[0].EventChangeType, "event change type is not
expected")
+ assert.Equal(t, testCase.expectedChangeDetail,
testCase.eventSystemMock.events[0].EventChangeDetail, "event change detail is
not expected")
+ assert.Equal(t, testCase.expectedObjectID,
testCase.eventSystemMock.events[0].ObjectID, "event object id is not expected")
+ assert.Equal(t, testCase.expectedReferenceID,
testCase.eventSystemMock.events[0].ReferenceID, "event reference id is not
expected")
+ }
+ })
+ }
+}
+
+func TestSendRemoveAskEvent(t *testing.T) {
+ app := &Application{
+ ApplicationID: appID0,
+ queuePath: "root.test",
+ }
+
+ testCases := []struct {
+ name string
+ eventSystemMock *EventSystemMock
+ terminationType si.TerminationType
+ appRemoved bool
+ allocationAsk *AllocationAsk
+ expectedEventCnt int
+ expectedType si.EventRecord_Type
+ expectedChangeType si.EventRecord_ChangeType
+ expectedChangeDetail si.EventRecord_ChangeDetail
+ expectedObjectID string
+ expectedReferenceID string
+ }{
+ {
+ name: "disabled event system",
+ eventSystemMock: nil,
+ terminationType:
si.TerminationType_UNKNOWN_TERMINATION_TYPE,
+ allocationAsk: &AllocationAsk{},
+ },
+ {
+ name: "remove allocation ask cause of
resource manager cancel",
+ eventSystemMock: newEventSystemMock(),
+ terminationType: si.TerminationType_STOPPED_BY_RM,
+ allocationAsk: &AllocationAsk{applicationID:
appID0, allocationKey: aKey},
+ appRemoved: false,
+ expectedEventCnt: 1,
+ expectedType: si.EventRecord_APP,
+ expectedChangeType: si.EventRecord_REMOVE,
+ expectedChangeDetail: si.EventRecord_APP_REQUEST,
+ expectedObjectID: appID0,
+ expectedReferenceID: aKey,
+ },
+ {
+ name: "remove allocation ask cause of
timeout",
+ eventSystemMock: newEventSystemMock(),
+ terminationType: si.TerminationType_TIMEOUT,
+ allocationAsk: &AllocationAsk{applicationID:
appID0, allocationKey: aKey},
+ expectedEventCnt: 1,
+ expectedType: si.EventRecord_APP,
+ expectedChangeType: si.EventRecord_REMOVE,
+ expectedChangeDetail: si.EventRecord_REQUEST_TIMEOUT,
+ expectedObjectID: appID0,
+ expectedReferenceID: aKey,
+ },
+ {
+ name: "remove allocation cause of
application removal",
+ eventSystemMock: newEventSystemMock(),
+ terminationType: si.TerminationType_STOPPED_BY_RM,
+ allocationAsk: &AllocationAsk{applicationID:
appID0, allocationKey: aKey},
+ appRemoved: true,
+ expectedEventCnt: 1,
+ expectedType: si.EventRecord_APP,
+ expectedChangeType: si.EventRecord_REMOVE,
+ expectedChangeDetail: si.EventRecord_REQUEST_CANCEL,
+ expectedObjectID: appID0,
+ expectedReferenceID: aKey,
+ },
+ }
+ for _, testCase := range testCases {
+ t.Run(testCase.name, func(t *testing.T) {
+ if testCase.eventSystemMock == nil {
+ 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.sendRemoveAskEvent(testCase.allocationAsk,
testCase.terminationType, testCase.appRemoved)
+ } else {
+ evt := newApplicationEvents(app,
testCase.eventSystemMock)
+ assert.Assert(t, evt.eventSystem != nil, "event
system should not be nil")
+ assert.Assert(t, evt.enabled, "event system
should be enabled")
+ evt.sendRemoveAskEvent(testCase.allocationAsk,
testCase.terminationType, testCase.appRemoved)
+ assert.Equal(t, testCase.expectedEventCnt,
len(testCase.eventSystemMock.events), "event was not generated")
+ assert.Equal(t, testCase.expectedType,
testCase.eventSystemMock.events[0].Type, "event type is not expected")
+ assert.Equal(t, testCase.expectedChangeType,
testCase.eventSystemMock.events[0].EventChangeType, "event change type is not
expected")
+ assert.Equal(t, testCase.expectedChangeDetail,
testCase.eventSystemMock.events[0].EventChangeDetail, "event change detail is
not expected")
+ assert.Equal(t, testCase.expectedObjectID,
testCase.eventSystemMock.events[0].ObjectID, "event object id is not expected")
+ assert.Equal(t, testCase.expectedReferenceID,
testCase.eventSystemMock.events[0].ReferenceID, "event reference id is not
expected")
+ }
+ })
+ }
+}
diff --git a/pkg/scheduler/objects/application_test.go
b/pkg/scheduler/objects/application_test.go
index e01f0856..a7aead35 100644
--- a/pkg/scheduler/objects/application_test.go
+++ b/pkg/scheduler/objects/application_test.go
@@ -30,6 +30,7 @@ import (
"github.com/apache/yunikorn-core/pkg/common/configs"
"github.com/apache/yunikorn-core/pkg/common/resources"
"github.com/apache/yunikorn-core/pkg/common/security"
+ "github.com/apache/yunikorn-core/pkg/events"
"github.com/apache/yunikorn-core/pkg/handler"
"github.com/apache/yunikorn-core/pkg/rmproxy"
"github.com/apache/yunikorn-core/pkg/rmproxy/rmevent"
@@ -199,6 +200,10 @@ func TestAppReservation(t *testing.T) {
// test multiple reservations from one allocation
func TestAppAllocReservation(t *testing.T) {
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl)
//nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+
app := newApplication(appID1, "default", "root.unknown")
if app == nil || app.ApplicationID != appID1 {
t.Fatalf("app create failed which should not have %v", app)
@@ -260,6 +265,30 @@ func TestAppAllocReservation(t *testing.T) {
if app.HasReserved() || node1.IsReserved() || node2.IsReserved() ||
reservedAsks != 2 {
t.Errorf("ask removal did not clean up all reservations,
reserved released = %d", reservedAsks)
}
+
+ // wait for events to be processed
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ fmt.Printf("checking event length: %d\n",
eventSystem.Store.CountStoredEvents())
+ return eventSystem.Store.CountStoredEvents() == 2
+ })
+ 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, 2, len(records), "expecting 2 events: 1 new alloc ask
and 1 alloc ask cancel")
+ allocAskRecord := records[0]
+ assert.Equal(t, si.EventRecord_APP, allocAskRecord.Type, "incorrect
event type, expect app")
+ assert.Equal(t, ask.applicationID, allocAskRecord.ObjectID, "incorrect
object ID, expected application ID")
+ assert.Equal(t, ask.allocationKey, allocAskRecord.ReferenceID,
"incorrect reference ID, expected alloc ask ID")
+ assert.Equal(t, si.EventRecord_ADD, allocAskRecord.EventChangeType,
"incorrect change type, expected add")
+ assert.Equal(t, si.EventRecord_APP_REQUEST,
allocAskRecord.EventChangeDetail, "incorrect change detail, expected new alloc
ask")
+ allocAskCancelRecord := records[1]
+ assert.Equal(t, si.EventRecord_APP, allocAskCancelRecord.Type,
"incorrect event type, expect app")
+ assert.Equal(t, ask.applicationID, allocAskCancelRecord.ObjectID,
"incorrect object ID, expected application ID")
+ assert.Equal(t, ask.allocationKey, allocAskCancelRecord.ReferenceID,
"incorrect reference ID, expected alloc ask ID")
+ assert.Equal(t, si.EventRecord_REMOVE,
allocAskCancelRecord.EventChangeType, "incorrect change type, expected remove")
+ assert.Equal(t, si.EventRecord_REQUEST_CANCEL,
allocAskCancelRecord.EventChangeDetail, "incorrect change detail, expected new
alloc ask")
}
// test update allocation repeat
@@ -311,6 +340,10 @@ func TestUpdateRepeat(t *testing.T) {
// test pending calculation and ask addition
func TestAddAllocAsk(t *testing.T) {
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl)
//nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+
app := newApplication(appID1, "default", "root.unknown")
if app == nil || app.ApplicationID != appID1 {
t.Fatalf("app create failed which should not have %v", app)
@@ -338,7 +371,7 @@ func TestAddAllocAsk(t *testing.T) {
t.Errorf("ask with zero repeat should not have been added to
app")
}
- // working cases
+ // add alloc ask
res =
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})
ask = newAllocationAskRepeat(aKey, appID1, res, 1)
err = app.AddAllocationAsk(ask)
@@ -348,6 +381,26 @@ func TestAddAllocAsk(t *testing.T) {
if !resources.Equals(res, pending) {
t.Errorf("pending resource not updated correctly, expected %v
but was: %v", res, pending)
}
+
+ // test add alloc ask event
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ fmt.Printf("checking event length: %d\n",
eventSystem.Store.CountStoredEvents())
+ return eventSystem.Store.CountStoredEvents() == 1
+ })
+ assert.NilError(t, err, "the events should have been processed")
+ records := eventSystem.Store.CollectEvents()
+ if records == nil {
+ t.Fatal("collecting eventChannel should return something")
+ }
+ assert.Equal(t, 1, len(records), "expecting add alloc ask event")
+ record := records[0]
+ assert.Equal(t, si.EventRecord_APP, record.Type, "incorrect event type,
expect app")
+ assert.Equal(t, appID1, record.ObjectID, "incorrect object ID, expected
application ID")
+ assert.Equal(t, aKey, record.ReferenceID, "incorrect reference ID,
expected placeholder alloc ID")
+ assert.Equal(t, si.EventRecord_ADD, record.EventChangeType, "incorrect
change type, expected add")
+ assert.Equal(t, si.EventRecord_APP_REQUEST, record.EventChangeDetail,
"incorrect change detail, expected app request")
+ eventSystem.Stop()
+
ask = newAllocationAskRepeat(aKey, appID1, res, 2)
err = app.AddAllocationAsk(ask)
assert.NilError(t, err, "ask should have been updated on app")
@@ -1178,6 +1231,10 @@ 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")
// state changes are not important
@@ -1234,6 +1291,25 @@ func TestReplaceAllocation(t *testing.T) {
assert.Equal(t, realAlloc.GetPlaceholderCreateTime(),
ph.GetCreateTime(), "real allocation's placeholder create time not updated as
expected: got %s, expected %s", realAlloc.GetPlaceholderCreateTime(),
ph.GetCreateTime())
assertUserGroupResource(t, getTestUserGroup(), res)
+ // wait for events to be processed
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ fmt.Printf("checking event length: %d\n",
eventSystem.Store.CountStoredEvents())
+ return eventSystem.Store.CountStoredEvents() == 1
+ })
+ 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, 1, len(records), "expecting alloc replaced event")
+ record := records[0]
+ assert.Equal(t, si.EventRecord_APP, record.Type, "incorrect event type,
expect app")
+ assert.Equal(t, alloc.applicationID, record.ObjectID, "incorrect object
ID, expected application ID")
+ assert.Equal(t, alloc.GetUUID(), record.ReferenceID, "incorrect
reference ID, expected placeholder alloc ID")
+ assert.Equal(t, si.EventRecord_REMOVE, record.EventChangeType,
"incorrect change type, expected remove")
+ assert.Equal(t, si.EventRecord_ALLOC_REPLACED,
record.EventChangeDetail, "incorrect change detail, expected alloc replaced")
+ eventSystem.Stop()
+
// add the placeholder back to the app, the failure test above changed
state and removed the ph
app.SetState(Running.String())
ph.ClearReleases()
@@ -1274,6 +1350,10 @@ func TestTimeoutPlaceholderAllocAsk(t *testing.T) {
func runTimeoutPlaceholderTest(t *testing.T, expectedState string,
gangSchedulingStyle string) {
setupUGM()
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl)
//nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+
// create a fake queue
queue, err := createRootQueue(nil)
assert.NilError(t, err, "queue create failed")
@@ -1352,10 +1432,33 @@ func runTimeoutPlaceholderTest(t *testing.T,
expectedState string, gangSchedulin
assert.Equal(t, len(log), 2, "wrong number of app events")
assert.Equal(t, log[0].ApplicationState, Accepted.String())
assert.Equal(t, log[1].ApplicationState, expectedState)
+
+ // wait for events to be processed
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ fmt.Printf("checking event length: %d\n",
eventSystem.Store.CountStoredEvents())
+ return eventSystem.Store.CountStoredEvents() == 4
+ })
+ 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, 4, len(records), "expecting 4 events: 1 new alloc ask,
1 alloc ask timeout, and 2 alloc timeout")
+ // new alloc sak and alloc timeout are tested in other cases, we only
test the alloc ask timeout
+ record := records[1]
+ assert.Equal(t, si.EventRecord_APP, record.Type, "incorrect event type,
expect app")
+ assert.Equal(t, phAsk.GetApplicationID(), record.ObjectID, "incorrect
object ID, expected application ID")
+ assert.Equal(t, phAsk.GetAllocationKey(), record.ReferenceID,
"incorrect reference ID, expected alloc ask ID")
+ assert.Equal(t, si.EventRecord_REMOVE, record.EventChangeType,
"incorrect change type, expected remove")
+ assert.Equal(t, si.EventRecord_REQUEST_TIMEOUT,
record.EventChangeDetail, "incorrect change detail, expected alloc ask timeout")
}
func TestTimeoutPlaceholderAllocReleased(t *testing.T) {
setupUGM()
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl)
//nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+
originalPhTimeout := defaultPlaceholderTimeout
defaultPlaceholderTimeout = 5 * time.Millisecond
defer func() { defaultPlaceholderTimeout = originalPhTimeout }()
@@ -1417,6 +1520,24 @@ func TestTimeoutPlaceholderAllocReleased(t *testing.T) {
assert.Equal(t, app.placeholderData[""].Replaced, int64(0))
assert.Equal(t, app.placeholderData[""].TimedOut, int64(1))
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res,
3))
+
+ // wait for events to be processed
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ fmt.Printf("checking event length: %d\n",
eventSystem.Store.CountStoredEvents())
+ return eventSystem.Store.CountStoredEvents() == 1
+ })
+ 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, 1, len(records), "expecting alloc timeout event")
+ record := records[0]
+ assert.Equal(t, si.EventRecord_APP, record.Type, "incorrect event type,
expect app")
+ assert.Equal(t, ph.applicationID, record.ObjectID, "incorrect object
ID, expected application ID")
+ assert.Equal(t, ph.GetUUID(), record.ReferenceID, "incorrect reference
ID, expected alloc ID")
+ assert.Equal(t, si.EventRecord_REMOVE, record.EventChangeType,
"incorrect change type, expected remove")
+ assert.Equal(t, si.EventRecord_ALLOC_TIMEOUT, record.EventChangeDetail,
"incorrect change detail, expected alloc timeout")
}
func TestTimeoutPlaceholderCompleting(t *testing.T) {
diff --git a/pkg/scheduler/objects/events.go b/pkg/scheduler/objects/events.go
deleted file mode 100644
index 60dcadf7..00000000
--- a/pkg/scheduler/objects/events.go
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- Licensed to the Apache Software Foundation (ASF) under one
- or more contributor license agreements. See the NOTICE file
- distributed with this work for additional information
- regarding copyright ownership. The ASF licenses this file
- to you under the Apache License, Version 2.0 (the
- "License"); you may not use this file except in compliance
- with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
-*/
-
-package objects
-
-import (
- "fmt"
-
- "github.com/apache/yunikorn-core/pkg/events"
-)
-
-type applicationEvents struct {
- enabled bool
- eventSystem events.EventSystem
- app *Application
-}
-
-func (evt *applicationEvents) sendAppDoesNotFitEvent(request *AllocationAsk) {
- if !evt.enabled {
- return
- }
-
- message := fmt.Sprintf("Application %s does not fit into %s queue",
request.GetApplicationID(), evt.app.queuePath)
- event := events.CreateRequestEventRecord(request.GetAllocationKey(),
request.GetApplicationID(), message, request.GetAllocatedResource())
- evt.eventSystem.AddEvent(event)
-}
-
-func (evt *applicationEvents) sendPlaceholderLargerEvent(ph *Allocation,
request *AllocationAsk) {
- if !evt.enabled {
- 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().String(), ph.GetAllocatedResource().String(),
ph.GetAllocationKey())
- event := events.CreateRequestEventRecord(ph.GetAllocationKey(),
evt.app.ApplicationID, message, request.GetAllocatedResource())
- evt.eventSystem.AddEvent(event)
-}
-
-func newApplicationEvents(app *Application, evt events.EventSystem)
*applicationEvents {
- return &applicationEvents{
- eventSystem: evt,
- enabled: evt != nil,
- app: app,
- }
-}
diff --git a/pkg/scheduler/objects/events_test.go
b/pkg/scheduler/objects/events_test.go
deleted file mode 100644
index 84ae5364..00000000
--- a/pkg/scheduler/objects/events_test.go
+++ /dev/null
@@ -1,91 +0,0 @@
-/*
- Licensed to the Apache Software Foundation (ASF) under one
- or more contributor license agreements. See the NOTICE file
- distributed with this work for additional information
- regarding copyright ownership. The ASF licenses this file
- to you under the Apache License, Version 2.0 (the
- "License"); you may not use this file except in compliance
- with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
-*/
-
-package objects
-
-import (
- "testing"
-
- "gotest.tools/v3/assert"
-
- "github.com/apache/yunikorn-scheduler-interface/lib/go/si"
-)
-
-type EventSystemMock struct {
- events []*si.EventRecord
-}
-
-func (m *EventSystemMock) AddEvent(event *si.EventRecord) {
- m.events = append(m.events, event)
-}
-
-func (m *EventSystemMock) StartService() {}
-
-func (m *EventSystemMock) Stop() {}
-
-func TestSendAppDoesNotFitEvent(t *testing.T) {
- app := &Application{
- 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.sendAppDoesNotFitEvent(&AllocationAsk{})
-
- // enabled
- mock := newEventSystemMock()
- evt = newApplicationEvents(app, mock)
- assert.Assert(t, evt.eventSystem != nil, "event system should not be
nil")
- assert.Assert(t, evt.enabled, "event system should be enabled")
- evt.sendAppDoesNotFitEvent(&AllocationAsk{
- applicationID: appID0,
- allocationKey: aKey,
- })
- assert.Equal(t, 1, len(mock.events), "event was not generated")
-}
-
-func TestSendPlaceholderLargerEvent(t *testing.T) {
- app := &Application{
- 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.sendPlaceholderLargerEvent(&Allocation{}, &AllocationAsk{})
-
- // enabled
- mock := newEventSystemMock()
- evt = newApplicationEvents(app, mock)
- assert.Assert(t, evt.eventSystem != nil, "event system should not be
nil")
- assert.Assert(t, evt.enabled, "event system should be enabled")
- evt.sendPlaceholderLargerEvent(&Allocation{
- allocationKey: aKey,
- }, &AllocationAsk{
- applicationID: appID0,
- allocationKey: aKey,
- })
- assert.Equal(t, 1, len(mock.events), "event was not generated")
-}
-
-func newEventSystemMock() *EventSystemMock {
- return &EventSystemMock{events: make([]*si.EventRecord, 0)}
-}
diff --git a/pkg/scheduler/objects/utilities_test.go
b/pkg/scheduler/objects/utilities_test.go
index 025989b1..8b391cdc 100644
--- a/pkg/scheduler/objects/utilities_test.go
+++ b/pkg/scheduler/objects/utilities_test.go
@@ -38,6 +38,7 @@ const (
appID1 = "app-1"
appID2 = "app-2"
aKey = "alloc-1"
+ aUUID = "alloc-uuid-1"
nodeID1 = "node-1"
instType1 = "itype-1"
)
diff --git a/pkg/scheduler/partition_test.go b/pkg/scheduler/partition_test.go
index 665acc1d..99210f11 100644
--- a/pkg/scheduler/partition_test.go
+++ b/pkg/scheduler/partition_test.go
@@ -308,6 +308,10 @@ func TestRemoveNode(t *testing.T) {
func TestRemoveNodeWithAllocations(t *testing.T) {
setupUGM()
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl)
//nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
@@ -347,6 +351,24 @@ func TestRemoveNodeWithAllocations(t *testing.T) {
assert.Equal(t, 0, len(confirmed), "node did not confirm correct
allocation")
assert.Equal(t, released[0].GetUUID(), allocUUID, "uuid returned by
release not the same as on allocation")
assertLimits(t, getTestUserGroup(), nil)
+
+ // wait for events to be processed
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ fmt.Printf("checking event length: %d\n",
eventSystem.Store.CountStoredEvents())
+ return eventSystem.Store.CountStoredEvents() == 1
+ })
+ 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, 1, len(records), "expecting alloc timeout event")
+ record := records[0]
+ assert.Equal(t, si.EventRecord_APP, record.Type, "incorrect event type,
expect app")
+ assert.Equal(t, appID1, record.ObjectID, "incorrect object ID, expected
application ID")
+ assert.Equal(t, allocUUID, record.ReferenceID, "incorrect reference ID,
expected alloc ID")
+ assert.Equal(t, si.EventRecord_REMOVE, record.EventChangeType,
"incorrect change type, expected remove")
+ assert.Equal(t, si.EventRecord_ALLOC_NODEREMOVED,
record.EventChangeDetail, "incorrect change detail, expected alloc node
removed")
}
// test with a replacement of a placeholder: placeholder and real on the same
node that gets removed
@@ -662,6 +684,10 @@ func TestPlaceholderDataWithNodeRemoval(t *testing.T) {
// ensure removed placeholders has been accounted under timed out in gang app
placeholder data
func TestPlaceholderDataWithRemoval(t *testing.T) {
setupUGM()
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl)
//nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
@@ -740,6 +766,25 @@ func TestPlaceholderDataWithRemoval(t *testing.T) {
releases, _ := partition.removeAllocation(release)
assert.Equal(t, 1, len(releases), "unexpected number of allocations
released")
assertPlaceholderData(t, gangApp, 7, 1)
+
+ // wait for events to be processed
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ fmt.Printf("checking event length: %d\n",
eventSystem.Store.CountStoredEvents())
+ return eventSystem.Store.CountStoredEvents() == 15
+ })
+ 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, 15, len(records), "expecting 15 events: 7 new alloc
ask, 7 new alloc, and 1 stopped by rm")
+ // new alloc ask and alloc events are tested in other cases, we only
want to test the stopped by rm event
+ record := records[14]
+ assert.Equal(t, si.EventRecord_APP, record.Type, "incorrect event type,
expect app")
+ assert.Equal(t, release.ApplicationID, record.ObjectID, "incorrect
object ID, expected application ID")
+ assert.Equal(t, release.UUID, record.ReferenceID, "incorrect reference
ID, expected alloc ID")
+ assert.Equal(t, si.EventRecord_REMOVE, record.EventChangeType,
"incorrect change type, expected remove")
+ assert.Equal(t, si.EventRecord_ALLOC_CANCEL, record.EventChangeDetail,
"incorrect change detail, expected alloc stopped by rm")
}
// check PlaceHolderData
@@ -1676,6 +1721,10 @@ func TestRequiredNodeAllocation(t *testing.T) {
func TestPreemption(t *testing.T) {
setupUGM()
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl)
//nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+
partition, _, app2, alloc1, alloc2 := setupPreemption(t)
res, err := resources.NewResourceFromConf(map[string]string{"vcore":
"5"})
@@ -1699,6 +1748,26 @@ func TestPreemption(t *testing.T) {
assert.Assert(t, !alloc1.IsPreempted(), "alloc-1 is preempted")
assert.Assert(t, alloc2.IsPreempted(), "alloc-2 is not preempted")
+ // wait for events to be processed
+ 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: 3 new alloc ask,
2 new alloc, and 1 alloc preempted event")
+ // new alloc ask and new alloc events are tested in other cases, we
only want to test the alloc preempted event
+ record := records[5]
+ assert.Equal(t, si.EventRecord_APP, record.Type, "incorrect event type,
expect app")
+ assert.Equal(t, ask3.GetApplicationID(), record.ObjectID, "incorrect
object ID, expected application ID")
+ assert.Equal(t, alloc2.GetUUID(), record.ReferenceID, "incorrect
reference ID, expected alloc ID")
+ assert.Equal(t, si.EventRecord_REMOVE, record.EventChangeType,
"incorrect change type, expected remove")
+ assert.Equal(t, si.EventRecord_ALLOC_PREEMPT, record.EventChangeDetail,
"incorrect change detail, expected alloc preempted")
+ eventSystem.Stop()
+
// allocation should still not do anything as we have not yet released
the preempted allocation
alloc = partition.tryAllocate()
if alloc != nil {
@@ -2626,7 +2695,8 @@ func TestPlaceholderSmallerThanReal(t *testing.T) {
// wait for events to be processed
err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
- return eventSystem.Store.CountStoredEvents() == 1
+ fmt.Printf("checking event length: %d\n",
eventSystem.Store.CountStoredEvents())
+ return eventSystem.Store.CountStoredEvents() == 4
})
assert.NilError(t, err, "the event should have been processed")
@@ -2634,12 +2704,30 @@ func TestPlaceholderSmallerThanReal(t *testing.T) {
if records == nil {
t.Fatal("collecting eventChannel should return something")
}
- assert.Equal(t, 1, len(records), "expecting one event for placeholder
mismatch")
- record := records[0]
- assert.Equal(t, si.EventRecord_REQUEST, record.Type, "incorrect event
type")
- assert.Equal(t, phID, record.ObjectID, "incorrect allocation ID,
expected placeholder alloc ID")
- assert.Equal(t, appID1, record.ReferenceID, "event should reference
application ID")
- assert.Assert(t, strings.Contains(record.Message, "Task group 'tg-1' in
application 'app-1'"), "unexpected message in record")
+ assert.Equal(t, 4, len(records), "expecting four events: ph-1 ask, ph-1
allocation, alloc-1 ask and placeholder mismatch")
+ phAskRecord := records[0]
+ assert.Equal(t, si.EventRecord_APP, phAskRecord.Type, "incorrect event
type, expect app")
+ assert.Equal(t, appID1, phAskRecord.ObjectID, "incorrect object ID,
expected application ID")
+ assert.Equal(t, phID, phAskRecord.ReferenceID, "incorrect reference ID,
expected placeholder alloc ID")
+ assert.Equal(t, si.EventRecord_ADD, phAskRecord.EventChangeType,
"incorrect change type, expected add")
+ assert.Equal(t, si.EventRecord_APP_REQUEST,
phAskRecord.EventChangeDetail, "incorrect change detail, expected app request")
+ phAllocRecord := records[1]
+ assert.Equal(t, si.EventRecord_APP, phAllocRecord.Type, "incorrect
event type, expect app")
+ assert.Equal(t, appID1, phAllocRecord.ObjectID, "incorrect object ID,
expected application ID")
+ assert.Equal(t, ph.GetUUID(), phAllocRecord.ReferenceID, "incorrect
reference ID, expected placeholder alloc UUID")
+ assert.Equal(t, si.EventRecord_ADD, phAllocRecord.EventChangeType,
"incorrect change type, expected add")
+ assert.Equal(t, si.EventRecord_APP_ALLOC,
phAllocRecord.EventChangeDetail, "incorrect change detail, expected app alloc")
+ allocAskRecord := records[2]
+ assert.Equal(t, si.EventRecord_APP, allocAskRecord.Type, "incorrect
event type, expect app")
+ assert.Equal(t, appID1, allocAskRecord.ObjectID, "incorrect object ID,
expected application ID")
+ assert.Equal(t, allocID, allocAskRecord.ReferenceID, "incorrect
reference ID, expected alloc ID")
+ assert.Equal(t, si.EventRecord_ADD, allocAskRecord.EventChangeType,
"incorrect change type, expected add")
+ assert.Equal(t, si.EventRecord_APP_REQUEST,
allocAskRecord.EventChangeDetail, "incorrect change detail, expected app alloc")
+ placeholderMismatchRecord := records[3]
+ assert.Equal(t, si.EventRecord_REQUEST, placeholderMismatchRecord.Type,
"incorrect event type")
+ assert.Equal(t, phID, placeholderMismatchRecord.ObjectID, "incorrect
allocation ID, expected placeholder alloc ID")
+ assert.Equal(t, appID1, placeholderMismatchRecord.ReferenceID, "event
should reference application ID")
+ assert.Assert(t, strings.Contains(placeholderMismatchRecord.Message,
"Task group 'tg-1' in application 'app-1'"), "unexpected message in record")
assertLimits(t, getTestUserGroup(), phRes)
// release placeholder: do what the context would do after the shim
processing
@@ -2713,7 +2801,7 @@ func TestPlaceholderSmallerMulti(t *testing.T) {
// wait for events to be processed
err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
fmt.Printf("checking event length: %d\n",
eventSystem.Store.CountStoredEvents())
- return eventSystem.Store.CountStoredEvents() == phCount
+ return eventSystem.Store.CountStoredEvents() == 16
})
assert.NilError(t, err, "the events should have been processed")
@@ -2721,7 +2809,7 @@ func TestPlaceholderSmallerMulti(t *testing.T) {
if records == nil {
t.Fatal("collecting eventChannel should return something")
}
- assert.Equal(t, phCount, len(records), "expecting %d events for
placeholder mismatch", phCount)
+ assert.Equal(t, 16, len(records), "expecting 16 events: 6 alloc ask, 5
alloc, and 5 placeholder mismatch")
assertLimits(t, getTestUserGroup(), tgRes)
// release placeholders: do what the context would do after the shim
processing
@@ -2811,6 +2899,10 @@ func TestPlaceholderBiggerThanReal(t *testing.T) {
func TestPlaceholderMatch(t *testing.T) {
setupUGM()
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl)
//nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
tgRes :=
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 10,
"second": 10})
@@ -2872,6 +2964,27 @@ func TestPlaceholderMatch(t *testing.T) {
assert.Equal(t, allocID2, alloc.GetAllocationKey(), "expected
allocation of alloc-2 to be returned")
assert.Equal(t, int64(1), app.GetAllPlaceholderData()[0].Count,
"placeholder data should show 1 available placeholder")
assert.Equal(t, int64(0), app.GetAllPlaceholderData()[0].Replaced,
"placeholder data should show no replacements yet")
+
+ // wait for events to be processed
+ 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: 3 new alloc ask,
2 new alloc , and 1 alloc replaced")
+ // new alloc sak and new alloc are tested in other cases, we only test
the alloc replaced
+ record := records[5]
+ assert.Equal(t, si.EventRecord_APP, record.Type, "incorrect event type,
expect app")
+ assert.Equal(t, alloc.GetApplicationID(), record.ObjectID, "incorrect
object ID, expected application ID")
+ assert.Equal(t, phUUID, record.ReferenceID, "incorrect reference ID,
expected alloc ID")
+ assert.Equal(t, si.EventRecord_REMOVE, record.EventChangeType,
"incorrect change type, expected remove")
+ assert.Equal(t, si.EventRecord_ALLOC_REPLACED,
record.EventChangeDetail, "incorrect change detail, expected alloc replaced")
+ eventSystem.Stop()
+
// release placeholder: do what the context would do after the shim
processing
release := &si.AllocationRelease{
PartitionName: "test",
@@ -3019,6 +3132,10 @@ func TestTryPlaceholderAllocate(t *testing.T) {
// The failure is triggered by the predicate plugin and is hidden in the alloc
handling
func TestFailReplacePlaceholder(t *testing.T) {
setupUGM()
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl)
//nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
if alloc := partition.tryPlaceholderAllocate(); alloc != nil {
@@ -3084,6 +3201,26 @@ func TestFailReplacePlaceholder(t *testing.T) {
t.Fatalf("node-2 allocation not updated as expected: got %s,
expected %s", node2.GetAllocatedResource(), res)
}
+ // wait for events to be processed
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ fmt.Printf("checking event length: %d\n",
eventSystem.Store.CountStoredEvents())
+ return eventSystem.Store.CountStoredEvents() == 4
+ })
+ 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, 4, len(records), "expecting 4 events: 2 new alloc ask,
1 new alloc, and 1 alloc replaced")
+ // new alloc ask and new alloc are tested in other cases, we only test
alloc replaced here
+ record := records[3]
+ assert.Equal(t, si.EventRecord_APP, record.Type, "incorrect event type,
expect app")
+ assert.Equal(t, alloc.GetApplicationID(), record.ObjectID, "incorrect
object ID, expected application ID")
+ assert.Equal(t, alloc.GetFirstRelease().GetUUID(), record.ReferenceID,
"incorrect reference ID, expected alloc ID")
+ assert.Equal(t, si.EventRecord_REMOVE, record.EventChangeType,
"incorrect change type, expected remove")
+ assert.Equal(t, si.EventRecord_ALLOC_REPLACED,
record.EventChangeDetail, "incorrect change detail, expected alloc replaced")
+ eventSystem.Stop()
+
phUUID := alloc.GetFirstRelease().GetUUID()
// placeholder is not released until confirmed by the shim
if !resources.Equals(app.GetPlaceholderResource(), res) {
@@ -3159,6 +3296,10 @@ func TestAddAllocationAsk(t *testing.T) {
func TestRemoveAllocationAsk(t *testing.T) {
setupUGM()
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl)
//nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
// add the app
@@ -3208,6 +3349,30 @@ func TestRemoveAllocationAsk(t *testing.T) {
partition.removeAllocationAsk(release)
assert.Assert(t, resources.IsZero(app.GetPendingResource()), "app
should not have pending asks")
assertLimits(t, getTestUserGroup(), nil)
+
+ // wait for events to be processed
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ fmt.Printf("checking event length: %d\n",
eventSystem.Store.CountStoredEvents())
+ return eventSystem.Store.CountStoredEvents() == 2
+ })
+ 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, 2, len(records), "expecting 2 events: 1 new alloc ask
and 1 alloc ask removal")
+ allocAskRecord := records[0]
+ assert.Equal(t, si.EventRecord_APP, allocAskRecord.Type, "incorrect
event type, expect app")
+ assert.Equal(t, ask.GetApplicationID(), allocAskRecord.ObjectID,
"incorrect object ID, expected application ID")
+ assert.Equal(t, ask.GetAllocationKey(), allocAskRecord.ReferenceID,
"incorrect reference ID, expected alloc ask ID")
+ assert.Equal(t, si.EventRecord_ADD, allocAskRecord.EventChangeType,
"incorrect change type, expected add")
+ assert.Equal(t, si.EventRecord_APP_REQUEST,
allocAskRecord.EventChangeDetail, "incorrect change detail, expected new alloc
ask")
+ allocAskCancelRecord := records[1]
+ assert.Equal(t, si.EventRecord_APP, allocAskCancelRecord.Type,
"incorrect event type, expect app")
+ assert.Equal(t, release.ApplicationID, allocAskCancelRecord.ObjectID,
"incorrect object ID, expected application ID")
+ assert.Equal(t, release.AllocationKey,
allocAskCancelRecord.ReferenceID, "incorrect reference ID, expected alloc ask
ID")
+ assert.Equal(t, si.EventRecord_REMOVE,
allocAskCancelRecord.EventChangeType, "incorrect change type, expected remove")
+ assert.Equal(t, si.EventRecord_APP_REQUEST,
allocAskCancelRecord.EventChangeDetail, "incorrect change detail, expected
alloc cancel")
}
func TestUpdateNodeSortingPolicy(t *testing.T) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
