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/06/29 20:52:55 UTC
[yunikorn-core] branch master updated: [YUNIKORN-1846] Rewrite event verification in application_test.go (#586)
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 6915f50d [YUNIKORN-1846] Rewrite event verification in application_test.go (#586)
6915f50d is described below
commit 6915f50d1ae39e765a042e2a3b136b3e53225db1
Author: Peter Bacsko <pb...@cloudera.com>
AuthorDate: Thu Jun 29 22:52:42 2023 +0200
[YUNIKORN-1846] Rewrite event verification in application_test.go (#586)
Closes: #586
Signed-off-by: Peter Bacsko <pb...@cloudera.com>
---
pkg/scheduler/objects/application.go | 41 +--
pkg/scheduler/objects/application_events.go | 16 +-
pkg/scheduler/objects/application_events_test.go | 110 +++-----
pkg/scheduler/objects/application_test.go | 322 ++++++++++++++++++-----
4 files changed, 305 insertions(+), 184 deletions(-)
diff --git a/pkg/scheduler/objects/application.go b/pkg/scheduler/objects/application.go
index 4ac248d6..0b150b7f 100644
--- a/pkg/scheduler/objects/application.go
+++ b/pkg/scheduler/objects/application.go
@@ -464,7 +464,7 @@ func (sa *Application) timeoutPlaceholderProcessing() {
zap.Error(err))
}
sa.notifyRMAllocationAskReleased(sa.rmID, sa.getAllRequestsInternal(), si.TerminationType_TIMEOUT, "releasing placeholders asks on placeholder timeout")
- sa.removeAsksInternal("")
+ sa.removeAsksInternal("", si.EventRecord_REQUEST_TIMEOUT)
// all allocations are placeholders but GetAllAllocations is locked and cannot be used
sa.notifyRMAllocationReleased(sa.rmID, sa.getPlaceholderAllocations(), si.TerminationType_TIMEOUT, "releasing allocated placeholders on placeholder timeout")
// we are in an accepted or new state so nothing can be replaced yet: mark everything as timedout
@@ -536,12 +536,11 @@ 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)
+ return sa.removeAsksInternal(allocKey, si.EventRecord_REQUEST_CANCEL)
}
// unlocked version of the allocation ask removal
-func (sa *Application) removeAsksInternal(allocKey string) int {
+func (sa *Application) removeAsksInternal(allocKey string, detail si.EventRecord_ChangeDetail) int {
// shortcut no need to do anything
if len(sa.requests) == 0 {
return 0
@@ -567,6 +566,9 @@ func (sa *Application) removeAsksInternal(allocKey string) int {
// Cleanup total pending resource
deltaPendingResource = sa.pending
sa.pending = resources.NewResource()
+ for _, ask := range sa.requests {
+ sa.appEvents.sendRemoveAskEvent(ask, detail)
+ }
sa.requests = make(map[string]*AllocationAsk)
sa.sortedRequests = sortedRequests{}
sa.askMaxPriority = configs.MinPriority
@@ -592,6 +594,7 @@ func (sa *Application) removeAsksInternal(allocKey string) int {
sa.pending = resources.Sub(sa.pending, deltaPendingResource)
delete(sa.requests, allocKey)
sa.sortedRequests.remove(ask)
+ sa.appEvents.sendRemoveAskEvent(ask, detail)
if priority := ask.GetPriority(); priority >= sa.askMaxPriority {
sa.updateAskMaxPriority()
}
@@ -1147,7 +1150,6 @@ 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
}
}
@@ -1203,7 +1205,6 @@ 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
}
}
@@ -1480,9 +1481,6 @@ 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
}
return nil
@@ -1631,6 +1629,7 @@ func (sa *Application) addAllocationInternal(info *Allocation) {
sa.allocatedResource = resources.Add(sa.allocatedResource, info.GetAllocatedResource())
sa.maxAllocatedResource = resources.ComponentWiseMax(sa.allocatedResource, sa.maxAllocatedResource)
}
+ sa.appEvents.sendNewAllocationEvent(info)
sa.allocations[info.GetUUID()] = info
}
@@ -1764,11 +1763,7 @@ 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)
- }
+ sa.appEvents.sendRemoveAllocationEvent(alloc, releaseType)
return alloc
}
@@ -1802,6 +1797,7 @@ func (sa *Application) RemoveAllAllocations() []*Allocation {
allocationsToRelease = append(allocationsToRelease, alloc)
// Aggregate the resources used by this alloc to the application's user resource tracker
sa.updateUsedResource(alloc)
+ sa.appEvents.sendRemoveAllocationEvent(alloc, si.TerminationType_STOPPED_BY_RM)
}
if resources.IsZero(sa.pending) {
@@ -1899,7 +1895,6 @@ 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,7 +1926,6 @@ func (sa *Application) notifyRMAllocationAskReleased(rmID string, released []*Al
TerminationType: terminationType,
Message: message,
})
- sa.appEvents.sendRemoveAskEvent(alloc, terminationType, false)
}
sa.rmEventHandler.HandleEvent(releaseEvent)
}
@@ -2010,18 +2004,3 @@ 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
index 3a8510bb..686c850f 100644
--- a/pkg/scheduler/objects/application_events.go
+++ b/pkg/scheduler/objects/application_events.go
@@ -93,24 +93,12 @@ func (evt *applicationEvents) sendRemoveAllocationEvent(alloc *Allocation, termi
evt.eventSystem.AddEvent(event)
}
-func (evt *applicationEvents) sendRemoveAskEvent(request *AllocationAsk, terminationType si.TerminationType, appRemoved bool) {
+func (evt *applicationEvents) sendRemoveAskEvent(request *AllocationAsk, detail si.EventRecord_ChangeDetail) {
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, common.Empty, request.GetAllocationKey(), si.EventRecord_REMOVE, eventChangeDetail, request.GetAllocatedResource())
+ event := events.CreateAppEventRecord(evt.app.ApplicationID, "", request.GetAllocationKey(), si.EventRecord_REMOVE, detail, request.GetAllocatedResource())
evt.eventSystem.AddEvent(event)
}
diff --git a/pkg/scheduler/objects/application_events_test.go b/pkg/scheduler/objects/application_events_test.go
index d7706d97..0c2a3608 100644
--- a/pkg/scheduler/objects/application_events_test.go
+++ b/pkg/scheduler/objects/application_events_test.go
@@ -250,84 +250,36 @@ func TestSendRemoveAskEvent(t *testing.T) {
ApplicationID: appID0,
queuePath: "root.test",
}
+ ask := &AllocationAsk{
+ applicationID: appID0,
+ allocationKey: aKey}
- 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")
- }
- })
- }
+ // 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.sendRemoveAskEvent(ask, si.EventRecord_REQUEST_CANCEL)
+
+ // enabled
+ mockEvents := newEventSystemMock()
+ appEvents := newApplicationEvents(app, mockEvents)
+
+ appEvents.sendRemoveAskEvent(ask, si.EventRecord_REQUEST_CANCEL)
+ 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_REQUEST_CANCEL, event.EventChangeDetail)
+ assert.Equal(t, "app-0", event.ObjectID)
+ assert.Equal(t, "alloc-1", event.ReferenceID)
+ assert.Equal(t, "", event.Message)
+
+ mockEvents.Reset()
+ appEvents.sendRemoveAskEvent(ask, si.EventRecord_REQUEST_TIMEOUT)
+ 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_REQUEST_TIMEOUT, event.EventChangeDetail)
+ assert.Equal(t, "app-0", event.ObjectID)
+ assert.Equal(t, "alloc-1", event.ReferenceID)
+ assert.Equal(t, "", event.Message)
}
diff --git a/pkg/scheduler/objects/application_test.go b/pkg/scheduler/objects/application_test.go
index a7aead35..a8734fd7 100644
--- a/pkg/scheduler/objects/application_test.go
+++ b/pkg/scheduler/objects/application_test.go
@@ -1291,25 +1291,6 @@ 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()
@@ -1350,10 +1331,6 @@ 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")
@@ -1432,25 +1409,6 @@ 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) {
@@ -1520,24 +1478,6 @@ 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) {
@@ -2032,6 +1972,268 @@ func TestMaxAskPriority(t *testing.T) {
assert.Equal(t, app.GetAskMaxPriority(), int32(15), "wrong priority after updating p=15 repeat to 1")
}
+func TestAskEvents(t *testing.T) {
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl) //nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+
+ app := newApplication(appID1, "default", "root.default")
+ queue, err := createRootQueue(nil)
+ assert.NilError(t, err, "queue create failed")
+ app.queue = queue
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})
+ ask := newAllocationAsk(aKey, appID1, res)
+
+ err = app.AddAllocationAsk(ask)
+ assert.NilError(t, err)
+ app.RemoveAllocationAsk(ask.allocationKey)
+ noEvents := 0
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ noEvents = eventSystem.Store.CountStoredEvents()
+ return noEvents == 2
+ })
+ assert.NilError(t, err, "expected 2 events, got %d", noEvents)
+ records := eventSystem.Store.CollectEvents()
+ assert.Equal(t, 2, len(records), "number of events")
+ assert.Equal(t, si.EventRecord_APP, records[0].Type)
+ assert.Equal(t, si.EventRecord_ADD, records[0].EventChangeType)
+ assert.Equal(t, si.EventRecord_APP_REQUEST, records[0].EventChangeDetail)
+ assert.Equal(t, si.EventRecord_APP, records[1].Type)
+ assert.Equal(t, si.EventRecord_REMOVE, records[1].EventChangeType)
+ assert.Equal(t, si.EventRecord_REQUEST_CANCEL, records[1].EventChangeDetail)
+
+ ask2 := newAllocationAsk("alloc-2", appID1, res)
+ ask3 := newAllocationAsk("alloc-3", appID1, res)
+ err = app.AddAllocationAsk(ask)
+ assert.NilError(t, err)
+ err = app.AddAllocationAsk(ask2)
+ assert.NilError(t, err)
+ err = app.AddAllocationAsk(ask3)
+ assert.NilError(t, err)
+ app.removeAsksInternal("", si.EventRecord_REQUEST_TIMEOUT)
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ noEvents = eventSystem.Store.CountStoredEvents()
+ return noEvents == 6
+ })
+ assert.NilError(t, err, "expected 6 events, got %d", noEvents)
+ records = eventSystem.Store.CollectEvents()
+ refIdsRemoved := make(map[string]int) // order can change due to map iteration
+ assert.Equal(t, 6, len(records), "number of events")
+ assert.Equal(t, si.EventRecord_APP, records[3].Type)
+ assert.Equal(t, si.EventRecord_REMOVE, records[3].EventChangeType)
+ assert.Equal(t, si.EventRecord_REQUEST_TIMEOUT, records[3].EventChangeDetail)
+ refIdsRemoved[records[3].ReferenceID]++
+ assert.Equal(t, si.EventRecord_APP, records[4].Type)
+ assert.Equal(t, si.EventRecord_REMOVE, records[4].EventChangeType)
+ assert.Equal(t, si.EventRecord_REQUEST_TIMEOUT, records[4].EventChangeDetail)
+ refIdsRemoved[records[4].ReferenceID]++
+ assert.Equal(t, si.EventRecord_APP, records[5].Type)
+ assert.Equal(t, si.EventRecord_REMOVE, records[5].EventChangeType)
+ assert.Equal(t, si.EventRecord_REQUEST_TIMEOUT, records[5].EventChangeDetail)
+ refIdsRemoved[records[5].ReferenceID]++
+ assert.Equal(t, 1, refIdsRemoved["alloc-1"])
+ assert.Equal(t, 1, refIdsRemoved["alloc-2"])
+ assert.Equal(t, 1, refIdsRemoved["alloc-3"])
+}
+
+func TestAllocationEvents(t *testing.T) { //nolint:funlen
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl) //nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+
+ app := newApplication(appID1, "default", "root.default")
+ queue, err := createRootQueue(nil)
+ assert.NilError(t, err, "queue create failed")
+ app.queue = queue
+
+ resMap := map[string]string{"memory": "100", "vcores": "10"}
+ res, err := resources.NewResourceFromConf(resMap)
+ assert.NilError(t, err, "failed to create resource with error")
+ alloc1 := newAllocation(appID1, "uuid-1", nodeID1, "root.a", res)
+ alloc2 := newAllocation(appID1, "uuid-2", nodeID1, "root.a", res)
+
+ // add + remove
+ app.AddAllocation(alloc1)
+ app.AddAllocation(alloc2)
+ app.RemoveAllocation("uuid-1", si.TerminationType_STOPPED_BY_RM)
+ app.RemoveAllocation("uuid-2", si.TerminationType_PLACEHOLDER_REPLACED)
+ noEvents := 0
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ noEvents = eventSystem.Store.CountStoredEvents()
+ return noEvents == 4
+ })
+ assert.NilError(t, err, "expected 4 events, got %d", noEvents)
+ records := eventSystem.Store.CollectEvents()
+
+ assert.Equal(t, 4, len(records), "number of events")
+ assert.Equal(t, si.EventRecord_APP, records[0].Type)
+ assert.Equal(t, si.EventRecord_ADD, records[0].EventChangeType)
+ assert.Equal(t, si.EventRecord_APP_ALLOC, records[0].EventChangeDetail)
+ assert.Equal(t, "uuid-1", records[0].ReferenceID)
+ assert.Equal(t, "app-1", records[0].ObjectID)
+ assert.Equal(t, si.EventRecord_APP, records[1].Type)
+ assert.Equal(t, si.EventRecord_ADD, records[1].EventChangeType)
+ assert.Equal(t, si.EventRecord_APP_ALLOC, records[1].EventChangeDetail)
+ assert.Equal(t, "uuid-2", records[1].ReferenceID)
+ assert.Equal(t, "app-1", records[1].ObjectID)
+ assert.Equal(t, si.EventRecord_APP, records[2].Type)
+ assert.Equal(t, si.EventRecord_REMOVE, records[2].EventChangeType)
+ assert.Equal(t, si.EventRecord_ALLOC_CANCEL, records[2].EventChangeDetail)
+ assert.Equal(t, "uuid-1", records[2].ReferenceID)
+ assert.Equal(t, "app-1", records[2].ObjectID)
+ assert.Equal(t, si.EventRecord_APP, records[3].Type)
+ assert.Equal(t, si.EventRecord_REMOVE, records[3].EventChangeType)
+ assert.Equal(t, si.EventRecord_ALLOC_REPLACED, records[3].EventChangeDetail)
+ assert.Equal(t, "uuid-2", records[3].ReferenceID)
+ assert.Equal(t, "app-1", records[3].ObjectID)
+
+ // add + replace
+ alloc1.placeholder = true
+ app.AddAllocation(alloc1)
+ app.ReplaceAllocation("uuid-1")
+ noEvents = 0
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ noEvents = eventSystem.Store.CountStoredEvents()
+ return noEvents == 2
+ })
+ assert.NilError(t, err, "expected 2 events, got %d", noEvents)
+ records = eventSystem.Store.CollectEvents()
+ assert.Equal(t, 2, len(records), "number of events")
+ assert.Equal(t, si.EventRecord_APP, records[0].Type)
+ assert.Equal(t, si.EventRecord_ADD, records[0].EventChangeType)
+ assert.Equal(t, si.EventRecord_APP_ALLOC, records[0].EventChangeDetail)
+ assert.Equal(t, "uuid-1", records[0].ReferenceID)
+ assert.Equal(t, "app-1", records[0].ObjectID)
+ assert.Equal(t, si.EventRecord_APP, records[1].Type)
+ assert.Equal(t, si.EventRecord_REMOVE, records[1].EventChangeType)
+ assert.Equal(t, si.EventRecord_ALLOC_REPLACED, records[1].EventChangeDetail)
+ assert.Equal(t, "uuid-1", records[0].ReferenceID)
+ assert.Equal(t, "app-1", records[0].ObjectID)
+
+ // add + remove all
+ app.AddAllocation(alloc1)
+ app.AddAllocation(alloc2)
+ app.RemoveAllAllocations()
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ noEvents = eventSystem.Store.CountStoredEvents()
+ return noEvents == 4
+ })
+ assert.NilError(t, err, "expected 4 events, got %d", noEvents)
+ records = eventSystem.Store.CollectEvents()
+ refIdsRemoved := make(map[string]int) // order can change due to map iteration
+ assert.Equal(t, 4, len(records), "number of events")
+ assert.Equal(t, si.EventRecord_APP, records[0].Type)
+ assert.Equal(t, si.EventRecord_ADD, records[0].EventChangeType)
+ assert.Equal(t, si.EventRecord_APP_ALLOC, records[0].EventChangeDetail)
+ assert.Equal(t, "uuid-1", records[0].ReferenceID)
+ assert.Equal(t, "app-1", records[0].ObjectID)
+ assert.Equal(t, si.EventRecord_APP, records[1].Type)
+ assert.Equal(t, si.EventRecord_ADD, records[1].EventChangeType)
+ assert.Equal(t, si.EventRecord_APP_ALLOC, records[1].EventChangeDetail)
+ assert.Equal(t, "uuid-2", records[1].ReferenceID)
+ assert.Equal(t, "app-1", records[1].ObjectID)
+ assert.Equal(t, si.EventRecord_APP, records[2].Type)
+ assert.Equal(t, si.EventRecord_REMOVE, records[2].EventChangeType)
+ assert.Equal(t, si.EventRecord_ALLOC_CANCEL, records[2].EventChangeDetail)
+ refIdsRemoved[records[2].ReferenceID]++
+ assert.Equal(t, "app-1", records[2].ObjectID)
+ assert.Equal(t, si.EventRecord_APP, records[3].Type)
+ assert.Equal(t, si.EventRecord_REMOVE, records[3].EventChangeType)
+ assert.Equal(t, si.EventRecord_ALLOC_CANCEL, records[3].EventChangeDetail)
+ refIdsRemoved[records[3].ReferenceID]++
+ assert.Equal(t, "app-1", records[3].ObjectID)
+ assert.Equal(t, 1, refIdsRemoved["uuid-1"])
+ assert.Equal(t, 1, refIdsRemoved["uuid-2"])
+}
+
+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")
+ smallerResMap := map[string]string{"memory": "50", "vcores": "10"}
+ smallerRes, err := resources.NewResourceFromConf(smallerResMap)
+ assert.NilError(t, err, "failed to create resource with error")
+
+ app := newApplication(appID1, "default", "root.default")
+ queue, err := createRootQueue(nil)
+ assert.NilError(t, err, "queue create failed")
+ app.queue = queue
+ // smallerRes < res in the same task group, so delta will be -50 memory
+ alloc1 := newAllocation(appID1, "uuid-1", nodeID1, "root.a", smallerRes)
+ alloc1.placeholder = true
+ alloc1.taskGroupName = "testGroup"
+ app.AddAllocation(alloc1)
+ ask := newAllocationAsk("alloc-0", "app-1", res)
+ ask.taskGroupName = "testGroup"
+ err = app.AddAllocationAsk(ask)
+ assert.NilError(t, err)
+
+ app.tryPlaceholderAllocate(func() NodeIterator {
+ return nil
+ }, func(s string) *Node {
+ return nil
+ })
+
+ noEvents := 0
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ noEvents = eventSystem.Store.CountStoredEvents()
+ return noEvents == 3
+ })
+ assert.NilError(t, err, "expected 3 events, got %d", noEvents)
+ records := eventSystem.Store.CollectEvents()
+ assert.Equal(t, si.EventRecord_REQUEST, records[2].Type)
+ assert.Equal(t, si.EventRecord_NONE, records[2].EventChangeType)
+ assert.Equal(t, si.EventRecord_DETAILS_NONE, records[2].EventChangeDetail)
+ assert.Equal(t, "app-1", records[2].ReferenceID)
+}
+
+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")
+ headroomMap := map[string]string{"memory": "0", "vcores": "0"}
+ headroom, err := resources.NewResourceFromConf(headroomMap)
+ assert.NilError(t, err, "failed to create resource with error")
+ ask := newAllocationAsk("alloc-0", "app-1", res)
+ app := newApplication(appID1, "default", "root.default")
+ queue, err := createRootQueue(nil)
+ assert.NilError(t, err, "queue create failed")
+ app.queue = queue
+ sr := sortedRequests{}
+ sr.insert(ask)
+ app.sortedRequests = sr
+ attempts := 0
+
+ app.tryAllocate(headroom, time.Second, &attempts, func() NodeIterator {
+ return nil
+ }, func() NodeIterator {
+ return nil
+ }, func(s string) *Node {
+ return nil
+ })
+
+ noEvents := 0
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ noEvents = eventSystem.Store.CountStoredEvents()
+ return noEvents == 1
+ })
+ assert.NilError(t, err, "expected 1 event, got %d", noEvents)
+ records := eventSystem.Store.CollectEvents()
+ assert.Equal(t, si.EventRecord_REQUEST, records[0].Type)
+ assert.Equal(t, si.EventRecord_NONE, records[0].EventChangeType)
+ assert.Equal(t, si.EventRecord_DETAILS_NONE, records[0].EventChangeDetail)
+ assert.Equal(t, "app-1", records[0].ReferenceID)
+ assert.Equal(t, "alloc-0", records[0].ObjectID)
+}
+
func (sa *Application) addPlaceholderDataWithLocking(ask *AllocationAsk) {
sa.Lock()
defer sa.Unlock()
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@yunikorn.apache.org
For additional commands, e-mail: issues-help@yunikorn.apache.org