You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@yunikorn.apache.org by pb...@apache.org on 2023/07/11 12:21:29 UTC
[yunikorn-core] branch master updated: [YUNIKORN-1803] Add wrapper for queue events (#574)
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 5649c420 [YUNIKORN-1803] Add wrapper for queue events (#574)
5649c420 is described below
commit 5649c420ee1a00835c684b3d5145abe3e316d003
Author: Peter Bacsko <pb...@cloudera.com>
AuthorDate: Tue Jul 11 14:21:12 2023 +0200
[YUNIKORN-1803] Add wrapper for queue events (#574)
Closes: #574
Signed-off-by: Peter Bacsko <pb...@cloudera.com>
---
pkg/events/events.go | 2 +-
pkg/events/events_test.go | 2 +-
pkg/scheduler/objects/application_test.go | 92 +++++------
pkg/scheduler/objects/queue.go | 30 +++-
pkg/scheduler/objects/queue_events.go | 123 +++++++++++++++
pkg/scheduler/objects/queue_events_test.go | 243 +++++++++++++++++++++++++++++
pkg/scheduler/objects/queue_test.go | 83 ++++++++--
pkg/scheduler/partition_test.go | 33 ++++
8 files changed, 548 insertions(+), 60 deletions(-)
diff --git a/pkg/events/events.go b/pkg/events/events.go
index 92a88dfe..be6f72b8 100644
--- a/pkg/events/events.go
+++ b/pkg/events/events.go
@@ -51,6 +51,6 @@ func CreateNodeEventRecord(objectID, message, referenceID string, changeType si.
return createEventRecord(si.EventRecord_NODE, objectID, referenceID, message, changeType, changeDetail, resource)
}
-func CreateQueueEventRecord(objectID, referenceID, message string, changeType si.EventRecord_ChangeType, changeDetail si.EventRecord_ChangeDetail, resource *resources.Resource) *si.EventRecord {
+func CreateQueueEventRecord(objectID, message, referenceID string, changeType si.EventRecord_ChangeType, changeDetail si.EventRecord_ChangeDetail, resource *resources.Resource) *si.EventRecord {
return createEventRecord(si.EventRecord_QUEUE, objectID, referenceID, message, changeType, changeDetail, resource)
}
diff --git a/pkg/events/events_test.go b/pkg/events/events_test.go
index 20ff2cb5..dc86fcd1 100644
--- a/pkg/events/events_test.go
+++ b/pkg/events/events_test.go
@@ -56,6 +56,6 @@ func TestCreateEventRecordTypes(t *testing.T) {
record = CreateNodeEventRecord("node", "message", "ask", si.EventRecord_NONE, si.EventRecord_DETAILS_NONE, nil)
assert.Equal(t, record.Type, si.EventRecord_NODE)
- record = CreateQueueEventRecord("queue", "app", "message", si.EventRecord_NONE, si.EventRecord_DETAILS_NONE, nil)
+ record = CreateQueueEventRecord("queue", "message", "app", si.EventRecord_NONE, si.EventRecord_DETAILS_NONE, nil)
assert.Equal(t, record.Type, si.EventRecord_QUEUE)
}
diff --git a/pkg/scheduler/objects/application_test.go b/pkg/scheduler/objects/application_test.go
index a8734fd7..bebc844c 100644
--- a/pkg/scheduler/objects/application_test.go
+++ b/pkg/scheduler/objects/application_test.go
@@ -267,23 +267,25 @@ func TestAppAllocReservation(t *testing.T) {
}
// wait for events to be processed
+ noEvents := 0
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
+ noEvents = eventSystem.Store.CountStoredEvents()
+ return noEvents == 3
})
- assert.NilError(t, err, "the event should have been processed")
+ assert.NilError(t, err, "expected 3 events, got %d", noEvents)
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, 3, len(records))
+ allocAskRecord := records[1]
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]
+ allocAskCancelRecord := records[2]
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")
@@ -383,17 +385,19 @@ func TestAddAllocAsk(t *testing.T) {
}
// test add alloc ask event
+ noEvents := 0
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
+ noEvents = eventSystem.Store.CountStoredEvents()
+ return noEvents == 2
})
- assert.NilError(t, err, "the events should have been processed")
+ assert.NilError(t, err, "expected 2 events, got %d", noEvents)
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, 2, len(records))
+ record := records[1]
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")
@@ -1990,17 +1994,17 @@ func TestAskEvents(t *testing.T) {
noEvents := 0
err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
noEvents = eventSystem.Store.CountStoredEvents()
- return noEvents == 2
+ return noEvents == 3
})
- assert.NilError(t, err, "expected 2 events, got %d", noEvents)
+ assert.NilError(t, err, "expected 3 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, 3, len(records), "number of events")
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)
+ assert.Equal(t, si.EventRecord_ADD, records[1].EventChangeType)
+ assert.Equal(t, si.EventRecord_APP_REQUEST, records[1].EventChangeDetail)
+ assert.Equal(t, si.EventRecord_APP, records[2].Type)
+ assert.Equal(t, si.EventRecord_REMOVE, records[2].EventChangeType)
+ assert.Equal(t, si.EventRecord_REQUEST_CANCEL, records[2].EventChangeDetail)
ask2 := newAllocationAsk("alloc-2", appID1, res)
ask3 := newAllocationAsk("alloc-3", appID1, res)
@@ -2060,32 +2064,32 @@ func TestAllocationEvents(t *testing.T) { //nolint:funlen
noEvents := 0
err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
noEvents = eventSystem.Store.CountStoredEvents()
- return noEvents == 4
+ return noEvents == 5
})
- assert.NilError(t, err, "expected 4 events, got %d", noEvents)
+ assert.NilError(t, err, "expected 5 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, 5, len(records), "number of events")
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, "uuid-1", 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, si.EventRecord_ADD, records[2].EventChangeType)
+ assert.Equal(t, si.EventRecord_APP_ALLOC, records[2].EventChangeDetail)
+ assert.Equal(t, "uuid-2", 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, si.EventRecord_ALLOC_CANCEL, records[3].EventChangeDetail)
+ assert.Equal(t, "uuid-1", records[3].ReferenceID)
assert.Equal(t, "app-1", records[3].ObjectID)
+ assert.Equal(t, si.EventRecord_APP, records[4].Type)
+ assert.Equal(t, si.EventRecord_REMOVE, records[4].EventChangeType)
+ assert.Equal(t, si.EventRecord_ALLOC_REPLACED, records[4].EventChangeDetail)
+ assert.Equal(t, "uuid-2", records[4].ReferenceID)
+ assert.Equal(t, "app-1", records[4].ObjectID)
// add + replace
alloc1.placeholder = true
@@ -2181,14 +2185,14 @@ func TestPlaceholderLargerEvent(t *testing.T) {
noEvents := 0
err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
noEvents = eventSystem.Store.CountStoredEvents()
- return noEvents == 3
+ return noEvents == 4
})
- assert.NilError(t, err, "expected 3 events, got %d", noEvents)
+ assert.NilError(t, err, "expected 4 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)
+ assert.Equal(t, si.EventRecord_REQUEST, records[3].Type)
+ assert.Equal(t, si.EventRecord_NONE, records[3].EventChangeType)
+ assert.Equal(t, si.EventRecord_DETAILS_NONE, records[3].EventChangeDetail)
+ assert.Equal(t, "app-1", records[3].ReferenceID)
}
func TestAppDoesNotFitEvent(t *testing.T) {
@@ -2223,15 +2227,15 @@ func TestAppDoesNotFitEvent(t *testing.T) {
noEvents := 0
err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
noEvents = eventSystem.Store.CountStoredEvents()
- return noEvents == 1
+ return noEvents == 2
})
- assert.NilError(t, err, "expected 1 event, got %d", noEvents)
+ assert.NilError(t, err, "expected 2 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)
+ assert.Equal(t, si.EventRecord_REQUEST, records[1].Type)
+ assert.Equal(t, si.EventRecord_NONE, records[1].EventChangeType)
+ assert.Equal(t, si.EventRecord_DETAILS_NONE, records[1].EventChangeDetail)
+ assert.Equal(t, "app-1", records[1].ReferenceID)
+ assert.Equal(t, "alloc-0", records[1].ObjectID)
}
func (sa *Application) addPlaceholderDataWithLocking(ask *AllocationAsk) {
diff --git a/pkg/scheduler/objects/queue.go b/pkg/scheduler/objects/queue.go
index 25dedca4..2f4fe591 100644
--- a/pkg/scheduler/objects/queue.go
+++ b/pkg/scheduler/objects/queue.go
@@ -32,6 +32,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/log"
"github.com/apache/yunikorn-core/pkg/metrics"
"github.com/apache/yunikorn-core/pkg/scheduler/objects/template"
@@ -83,6 +84,7 @@ type Queue struct {
runningApps uint64
allocatingAcceptedApps map[string]bool
template *template.Template
+ queueEvents *queueEvents
sync.RWMutex
}
@@ -135,9 +137,10 @@ func NewConfiguredQueue(conf configs.QueueConfig, parent *Queue) (*Queue, error)
sq.mergeProperties(parent.getProperties(), conf.Properties)
}
sq.UpdateQueueProperties()
-
+ sq.queueEvents = newQueueEvents(sq, events.GetEventSystem())
log.Log(log.SchedQueue).Info("configured queue added to scheduler",
zap.String("queueName", sq.QueuePath))
+ sq.queueEvents.sendNewQueueEvent()
return sq, nil
}
@@ -169,8 +172,10 @@ func NewDynamicQueue(name string, leaf bool, parent *Queue) (*Queue, error) {
}
sq.UpdateQueueProperties()
+ sq.queueEvents = newQueueEvents(sq, events.GetEventSystem())
log.Log(log.SchedQueue).Info("dynamic queue added to scheduler",
zap.String("queueName", sq.QueuePath))
+ sq.queueEvents.sendNewQueueEvent()
return sq, nil
}
@@ -296,12 +301,17 @@ func (sq *Queue) applyConf(conf configs.QueueConfig) error {
sq.isManaged = true
}
+ prevLeaf := sq.isLeaf
sq.isLeaf = !conf.Parent
// Make sure the parent flag is set correctly: config might expect auto parent type creation
if len(conf.Queues) > 0 {
sq.isLeaf = false
}
+ if prevLeaf != sq.isLeaf && sq.queueEvents != nil {
+ sq.queueEvents.sendTypeChangedEvent()
+ }
+
if !sq.isLeaf {
if err = sq.setTemplate(conf.ChildTemplate); err != nil {
return nil
@@ -337,6 +347,9 @@ func (sq *Queue) setResources(resource configs.Resources) error {
}
if resources.StrictlyGreaterThanZero(maxResource) {
+ if !resources.Equals(sq.maxResource, maxResource) && sq.queueEvents != nil {
+ sq.queueEvents.sendMaxResourceChangedEvent()
+ }
sq.maxResource = maxResource
sq.updateMaxResourceMetrics()
} else {
@@ -344,6 +357,9 @@ func (sq *Queue) setResources(resource configs.Resources) error {
}
if resources.StrictlyGreaterThanZero(guaranteedResource) {
+ if !resources.Equals(sq.guaranteedResource, guaranteedResource) && sq.queueEvents != nil {
+ sq.queueEvents.sendGuaranteedResourceChangedEvent()
+ }
sq.guaranteedResource = guaranteedResource
sq.updateGuaranteedResourceMetrics()
} else {
@@ -658,8 +674,10 @@ func (sq *Queue) decPendingResource(delta *resources.Resource) {
func (sq *Queue) AddApplication(app *Application) {
sq.Lock()
defer sq.Unlock()
- sq.applications[app.ApplicationID] = app
- sq.appPriorities[app.ApplicationID] = app.GetAskMaxPriority()
+ appID := app.ApplicationID
+ sq.applications[appID] = app
+ sq.appPriorities[appID] = app.GetAskMaxPriority()
+ sq.queueEvents.sendNewApplicationEvent(appID)
// YUNIKORN-199: update the quota from the namespace
// get the tag with the quota
quota := app.GetTag(common.AppTagNamespaceResourceQuota)
@@ -700,6 +718,7 @@ func (sq *Queue) RemoveApplication(app *Application) {
zap.String("applicationID", appID))
return
}
+ sq.queueEvents.sendRemoveApplicationEvent(appID)
if appPending := app.GetPendingResource(); !resources.IsZero(appPending) {
sq.decPendingResource(appPending)
}
@@ -905,6 +924,7 @@ func (sq *Queue) RemoveQueue() bool {
log.Log(log.SchedQueue).Info("removing queue", zap.String("queue", sq.QueuePath))
// root is always managed and is the only queue with a nil parent: no need to guard
sq.parent.removeChildQueue(sq.Name)
+ sq.queueEvents.sendRemoveQueueEvent()
return true
}
@@ -1823,3 +1843,7 @@ func (sq *Queue) recalculatePriority() int32 {
sq.currentPriority = curr
return priorityValueByPolicy(sq.priorityPolicy, sq.priorityOffset, curr)
}
+
+func (sq *Queue) SendRemoveQueueEvent() {
+ sq.queueEvents.sendRemoveQueueEvent()
+}
diff --git a/pkg/scheduler/objects/queue_events.go b/pkg/scheduler/objects/queue_events.go
new file mode 100644
index 00000000..d79cabc5
--- /dev/null
+++ b/pkg/scheduler/objects/queue_events.go
@@ -0,0 +1,123 @@
+/*
+ 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 (
+ "github.com/apache/yunikorn-core/pkg/common"
+ "github.com/apache/yunikorn-core/pkg/events"
+ "github.com/apache/yunikorn-scheduler-interface/lib/go/si"
+)
+
+type queueEvents struct {
+ enabled bool
+ eventSystem events.EventSystem
+ queue *Queue
+}
+
+func (q *queueEvents) sendNewQueueEvent() {
+ if !q.enabled {
+ return
+ }
+
+ detail := si.EventRecord_QUEUE_DYNAMIC
+ if q.queue.IsManaged() {
+ detail = si.EventRecord_DETAILS_NONE
+ }
+ event := events.CreateQueueEventRecord(q.queue.QueuePath, common.Empty, common.Empty, si.EventRecord_ADD,
+ detail, nil)
+ q.eventSystem.AddEvent(event)
+}
+
+func (q *queueEvents) sendNewApplicationEvent(appID string) {
+ if !q.enabled {
+ return
+ }
+
+ event := events.CreateQueueEventRecord(q.queue.QueuePath, common.Empty, appID, si.EventRecord_ADD,
+ si.EventRecord_QUEUE_APP, nil)
+ q.eventSystem.AddEvent(event)
+}
+
+func (q *queueEvents) sendRemoveQueueEvent() {
+ if !q.enabled {
+ return
+ }
+
+ detail := si.EventRecord_QUEUE_DYNAMIC
+ if q.queue.IsManaged() {
+ detail = si.EventRecord_DETAILS_NONE
+ }
+
+ event := events.CreateQueueEventRecord(q.queue.QueuePath, common.Empty, common.Empty, si.EventRecord_REMOVE,
+ detail, nil)
+ q.eventSystem.AddEvent(event)
+}
+
+func (q *queueEvents) sendRemoveApplicationEvent(appID string) {
+ if !q.enabled {
+ return
+ }
+
+ event := events.CreateQueueEventRecord(q.queue.QueuePath, common.Empty, appID, si.EventRecord_REMOVE,
+ si.EventRecord_QUEUE_APP, nil)
+ q.eventSystem.AddEvent(event)
+}
+
+func (q *queueEvents) sendMaxResourceChangedEvent() {
+ if !q.enabled {
+ return
+ }
+
+ event := events.CreateQueueEventRecord(q.queue.QueuePath, common.Empty, common.Empty, si.EventRecord_SET,
+ si.EventRecord_QUEUE_MAX, q.queue.maxResource)
+ q.eventSystem.AddEvent(event)
+}
+
+func (q *queueEvents) sendGuaranteedResourceChangedEvent() {
+ if !q.enabled {
+ return
+ }
+
+ event := events.CreateQueueEventRecord(q.queue.QueuePath, common.Empty, common.Empty, si.EventRecord_SET,
+ si.EventRecord_QUEUE_GUARANTEED, q.queue.guaranteedResource)
+ q.eventSystem.AddEvent(event)
+}
+
+func (q *queueEvents) sendTypeChangedEvent() {
+ if !q.enabled {
+ return
+ }
+
+ message := "leaf queue: false"
+ if q.queue.isLeaf {
+ message = "leaf queue: true"
+ }
+
+ event := events.CreateQueueEventRecord(q.queue.QueuePath, message, common.Empty, si.EventRecord_SET,
+ si.EventRecord_QUEUE_TYPE, nil)
+ q.eventSystem.AddEvent(event)
+}
+
+func newQueueEvents(queue *Queue, evt events.EventSystem) *queueEvents {
+ return &queueEvents{
+ eventSystem: evt,
+ enabled: evt != nil,
+ queue: queue,
+ }
+}
diff --git a/pkg/scheduler/objects/queue_events_test.go b/pkg/scheduler/objects/queue_events_test.go
new file mode 100644
index 00000000..07a418ef
--- /dev/null
+++ b/pkg/scheduler/objects/queue_events_test.go
@@ -0,0 +1,243 @@
+/*
+ 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-core/pkg/common"
+ "github.com/apache/yunikorn-core/pkg/common/resources"
+ "github.com/apache/yunikorn-scheduler-interface/lib/go/si"
+)
+
+const (
+ testQueuePath = "root.test"
+)
+
+func TestNewQueueEvents(t *testing.T) {
+ queue := &Queue{
+ QueuePath: testQueuePath,
+ }
+
+ // not enabled
+ nq := newQueueEvents(queue, nil)
+ assert.Assert(t, nq.eventSystem == nil, "event system should be nil")
+ assert.Assert(t, !nq.enabled, "event system should be disabled")
+
+ // enabled
+ nq = newQueueEvents(queue, newEventSystemMock())
+ assert.Assert(t, nq.eventSystem != nil, "event system should not be nil")
+ assert.Assert(t, nq.enabled, "event system should be enabled")
+}
+
+func TestSendNewQueueEvent(t *testing.T) {
+ queue := &Queue{
+ QueuePath: testQueuePath,
+ isManaged: true,
+ }
+
+ // not enabled
+ nq := newQueueEvents(queue, nil)
+ nq.sendNewQueueEvent()
+
+ // enabled
+ mock := newEventSystemMock()
+ nq = newQueueEvents(queue, mock)
+ nq.sendNewQueueEvent()
+ assert.Equal(t, 1, len(mock.events), "event was not generated")
+ event := mock.events[0]
+ assert.Equal(t, si.EventRecord_QUEUE, event.Type)
+ assert.Equal(t, testQueuePath, event.ObjectID)
+ assert.Equal(t, common.Empty, event.ReferenceID)
+ assert.Equal(t, common.Empty, event.Message)
+ assert.Equal(t, si.EventRecord_ADD, event.EventChangeType)
+ assert.Equal(t, si.EventRecord_DETAILS_NONE, event.EventChangeDetail)
+ assert.Equal(t, 0, len(event.Resource.Resources))
+ mock = newEventSystemMock()
+ nq = newQueueEvents(&Queue{
+ QueuePath: testQueuePath,
+ isManaged: false,
+ }, mock)
+ nq.sendNewQueueEvent()
+ event = mock.events[0]
+ assert.Equal(t, si.EventRecord_QUEUE_DYNAMIC, event.EventChangeDetail)
+}
+
+func TestSendRemoveQueueEvent(t *testing.T) {
+ queue := &Queue{
+ QueuePath: testQueuePath,
+ isManaged: true,
+ }
+
+ // not enabled
+ nq := newQueueEvents(queue, nil)
+ nq.sendRemoveQueueEvent()
+
+ // enabled
+ mock := newEventSystemMock()
+ nq = newQueueEvents(queue, mock)
+ nq.sendRemoveQueueEvent()
+ assert.Equal(t, 1, len(mock.events), "event was not generated")
+ event := mock.events[0]
+ assert.Equal(t, si.EventRecord_QUEUE, event.Type)
+ assert.Equal(t, testQueuePath, event.ObjectID)
+ assert.Equal(t, common.Empty, event.ReferenceID)
+ assert.Equal(t, common.Empty, event.Message)
+ assert.Equal(t, si.EventRecord_REMOVE, event.EventChangeType)
+ assert.Equal(t, si.EventRecord_DETAILS_NONE, event.EventChangeDetail)
+ assert.Equal(t, 0, len(event.Resource.Resources))
+ mock = newEventSystemMock()
+ nq = newQueueEvents(&Queue{
+ QueuePath: testQueuePath,
+ isManaged: false,
+ }, mock)
+ nq.sendRemoveQueueEvent()
+ event = mock.events[0]
+ assert.Equal(t, si.EventRecord_QUEUE_DYNAMIC, event.EventChangeDetail)
+}
+
+func TestNewApplicationEvent(t *testing.T) {
+ queue := &Queue{
+ QueuePath: testQueuePath,
+ }
+
+ // not enabled
+ nq := newQueueEvents(queue, nil)
+ nq.sendNewApplicationEvent(appID0)
+
+ // enabled
+ mock := newEventSystemMock()
+ nq = newQueueEvents(queue, mock)
+ nq.sendNewApplicationEvent(appID0)
+ assert.Equal(t, 1, len(mock.events), "event was not generated")
+ event := mock.events[0]
+ assert.Equal(t, si.EventRecord_QUEUE, event.Type)
+ assert.Equal(t, testQueuePath, event.ObjectID)
+ assert.Equal(t, appID0, event.ReferenceID)
+ assert.Equal(t, common.Empty, event.Message)
+ assert.Equal(t, si.EventRecord_ADD, event.EventChangeType)
+ assert.Equal(t, si.EventRecord_QUEUE_APP, event.EventChangeDetail)
+ assert.Equal(t, 0, len(event.Resource.Resources))
+}
+
+func TestRemoveApplicationEvent(t *testing.T) {
+ queue := &Queue{
+ QueuePath: testQueuePath,
+ }
+
+ // not enabled
+ nq := newQueueEvents(queue, nil)
+ nq.sendRemoveApplicationEvent(appID0)
+
+ // enabled
+ mock := newEventSystemMock()
+ nq = newQueueEvents(queue, mock)
+ nq.sendRemoveApplicationEvent(appID0)
+ assert.Equal(t, 1, len(mock.events), "event was not generated")
+ event := mock.events[0]
+ assert.Equal(t, si.EventRecord_QUEUE, event.Type)
+ assert.Equal(t, testQueuePath, event.ObjectID)
+ assert.Equal(t, appID0, event.ReferenceID)
+ assert.Equal(t, common.Empty, event.Message)
+ assert.Equal(t, si.EventRecord_REMOVE, event.EventChangeType)
+ assert.Equal(t, si.EventRecord_QUEUE_APP, event.EventChangeDetail)
+ assert.Equal(t, 0, len(event.Resource.Resources))
+}
+
+func TestTypeChangedEvent(t *testing.T) {
+ queue := &Queue{
+ QueuePath: testQueuePath,
+ }
+
+ // not enabled
+ nq := newQueueEvents(queue, nil)
+ nq.sendTypeChangedEvent()
+
+ // enabled
+ mock := newEventSystemMock()
+ nq = newQueueEvents(queue, mock)
+ nq.sendTypeChangedEvent()
+ assert.Equal(t, 1, len(mock.events), "event was not generated")
+ event := mock.events[0]
+ assert.Equal(t, si.EventRecord_QUEUE, event.Type)
+ assert.Equal(t, testQueuePath, event.ObjectID)
+ assert.Equal(t, common.Empty, event.ReferenceID)
+ assert.Equal(t, "leaf queue: false", event.Message)
+ assert.Equal(t, si.EventRecord_SET, event.EventChangeType)
+ assert.Equal(t, si.EventRecord_QUEUE_TYPE, event.EventChangeDetail)
+ assert.Equal(t, 0, len(event.Resource.Resources))
+}
+
+func TestSendMaxResourceChangedEvent(t *testing.T) {
+ max := getTestResource()
+ queue := &Queue{
+ QueuePath: testQueuePath,
+ maxResource: max,
+ }
+
+ // not enabled
+ nq := newQueueEvents(queue, nil)
+ nq.sendMaxResourceChangedEvent()
+
+ // enabled
+ mock := newEventSystemMock()
+ nq = newQueueEvents(queue, mock)
+ nq.sendMaxResourceChangedEvent()
+ assert.Equal(t, 1, len(mock.events), "event was not generated")
+ event := mock.events[0]
+ assert.Equal(t, si.EventRecord_QUEUE, event.Type)
+ assert.Equal(t, testQueuePath, event.ObjectID)
+ assert.Equal(t, common.Empty, event.ReferenceID)
+ assert.Equal(t, common.Empty, event.Message)
+ assert.Equal(t, si.EventRecord_SET, event.EventChangeType)
+ assert.Equal(t, si.EventRecord_QUEUE_MAX, event.EventChangeDetail)
+ assert.Equal(t, 1, len(event.Resource.Resources))
+ protoRes := resources.NewResourceFromProto(event.Resource)
+ assert.DeepEqual(t, max, protoRes)
+}
+
+func TestSendGuaranteedResourceChangedEvent(t *testing.T) {
+ guaranteed := getTestResource()
+ queue := &Queue{
+ QueuePath: testQueuePath,
+ guaranteedResource: guaranteed,
+ }
+
+ // not enabled
+ nq := newQueueEvents(queue, nil)
+ nq.sendGuaranteedResourceChangedEvent()
+
+ // enabled
+ mock := newEventSystemMock()
+ nq = newQueueEvents(queue, mock)
+ nq.sendGuaranteedResourceChangedEvent()
+ assert.Equal(t, 1, len(mock.events), "event was not generated")
+ event := mock.events[0]
+ assert.Equal(t, si.EventRecord_QUEUE, event.Type)
+ assert.Equal(t, testQueuePath, event.ObjectID)
+ assert.Equal(t, common.Empty, event.ReferenceID)
+ assert.Equal(t, common.Empty, event.Message)
+ assert.Equal(t, si.EventRecord_SET, event.EventChangeType)
+ assert.Equal(t, si.EventRecord_QUEUE_GUARANTEED, event.EventChangeDetail)
+ assert.Equal(t, 1, len(event.Resource.Resources))
+ protoRes := resources.NewResourceFromProto(event.Resource)
+ assert.DeepEqual(t, guaranteed, protoRes)
+}
diff --git a/pkg/scheduler/objects/queue_test.go b/pkg/scheduler/objects/queue_test.go
index 6b9f3427..b23b8394 100644
--- a/pkg/scheduler/objects/queue_test.go
+++ b/pkg/scheduler/objects/queue_test.go
@@ -28,12 +28,15 @@ import (
"gotest.tools/v3/assert"
+ "github.com/apache/yunikorn-core/pkg/common"
"github.com/apache/yunikorn-core/pkg/common/configs"
"github.com/apache/yunikorn-core/pkg/common/resources"
+ "github.com/apache/yunikorn-core/pkg/events"
"github.com/apache/yunikorn-core/pkg/scheduler/objects/template"
"github.com/apache/yunikorn-core/pkg/scheduler/policies"
"github.com/apache/yunikorn-core/pkg/webservice/dao"
- "github.com/apache/yunikorn-scheduler-interface/lib/go/common"
+ siCommon "github.com/apache/yunikorn-scheduler-interface/lib/go/common"
+ "github.com/apache/yunikorn-scheduler-interface/lib/go/si"
)
// base test for creating a managed queue
@@ -344,7 +347,7 @@ func TestAddApplication(t *testing.T) {
assert.NilError(t, err, "failed to create managed leaf queue")
pending := resources.NewResourceFromMap(
map[string]resources.Quantity{
- common.Memory: 10,
+ siCommon.Memory: 10,
})
app := newApplication(appID1, "default", "root.parent.leaf")
app.pending = pending
@@ -388,7 +391,7 @@ func TestAddApplicationWithTag(t *testing.T) {
"first": 10,
})
tags := make(map[string]string)
- tags[common.AppTagNamespaceResourceQuota] = "{\"resources\":{\"first\":{\"value\":10}}}"
+ tags[siCommon.AppTagNamespaceResourceQuota] = "{\"resources\":{\"first\":{\"value\":10}}}"
// add apps again now with the tag set
app = newApplicationWithTags("app-3", "default", "root.leaf-man", tags)
leaf.AddApplication(app)
@@ -404,7 +407,7 @@ func TestAddApplicationWithTag(t *testing.T) {
}
// set to illegal limit (0 value)
- tags[common.AppTagNamespaceResourceQuota] = "{\"resources\":{\"first\":{\"value\":0}}}"
+ tags[siCommon.AppTagNamespaceResourceQuota] = "{\"resources\":{\"first\":{\"value\":0}}}"
app = newApplicationWithTags("app-4", "default", "root.leaf-un", tags)
leafUn.AddApplication(app)
assert.Equal(t, len(leaf.applications), 2, "Application was not added to the Dynamic queue as expected")
@@ -1780,7 +1783,7 @@ func TestFindQueueByAppID(t *testing.T) {
assert.NilError(t, err, "failed to create queue")
app := newApplication(appID1, "default", "root.parent.leaf")
- app.pending = resources.NewResourceFromMap(map[string]resources.Quantity{common.Memory: 10})
+ app.pending = resources.NewResourceFromMap(map[string]resources.Quantity{siCommon.Memory: 10})
leaf1.AddApplication(app)
// we should be able to find the queue from any other given the appID
@@ -1801,9 +1804,9 @@ func TestFindQueueByAppID(t *testing.T) {
// nolint: funlen
func TestFindEligiblePreemptionVictims(t *testing.T) {
- res := resources.NewResourceFromMap(map[string]resources.Quantity{common.Memory: 100})
- parentMax := map[string]string{common.Memory: "200"}
- parentGuar := map[string]string{common.Memory: "100"}
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{siCommon.Memory: 100})
+ parentMax := map[string]string{siCommon.Memory: "200"}
+ parentGuar := map[string]string{siCommon.Memory: "100"}
ask := createAllocationAsk("ask1", appID1, true, true, 0, res)
ask.pendingAskRepeat = 1
ask2 := createAllocationAsk("ask2", appID2, true, true, -1000, res)
@@ -1812,7 +1815,7 @@ func TestFindEligiblePreemptionVictims(t *testing.T) {
ask3 := createAllocationAsk("ask3", appID2, true, true, -1000, res)
ask3.pendingAskRepeat = 1
alloc3 := NewAllocation("alloc-3", nodeID1, instType1, ask3)
- root, err := createRootQueue(map[string]string{common.Memory: "1000"})
+ root, err := createRootQueue(map[string]string{siCommon.Memory: "1000"})
assert.NilError(t, err, "failed to create queue")
parent1, err := createManagedQueueGuaranteed(root, "parent1", true, parentMax, parentGuar)
assert.NilError(t, err, "failed to create queue")
@@ -1927,10 +1930,10 @@ func TestFindEligiblePreemptionVictims(t *testing.T) {
parent1.priorityPolicy = policies.DefaultPriorityPolicy
// increasing parent2 guaranteed resources should remove leaf2 victims
- parent2.guaranteedResource = resources.NewResourceFromMap(map[string]resources.Quantity{common.Memory: 200})
+ parent2.guaranteedResource = resources.NewResourceFromMap(map[string]resources.Quantity{siCommon.Memory: 200})
snapshot = leaf1.FindEligiblePreemptionVictims(leaf1.QueuePath, ask)
assert.Equal(t, 0, len(victims(snapshot)), "found victims")
- parent2.guaranteedResource = resources.NewResourceFromMap(map[string]resources.Quantity{common.Memory: 100})
+ parent2.guaranteedResource = resources.NewResourceFromMap(map[string]resources.Quantity{siCommon.Memory: 100})
}
func victims(snapshot map[string]*QueuePreemptionSnapshot) []*Allocation {
@@ -2462,3 +2465,61 @@ func TestQueue_setAllocatingAccepted(t *testing.T) {
var q *Queue
q.setAllocatingAccepted("")
}
+
+func TestQueueEvents(t *testing.T) {
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl) //nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+ queue, err := createRootQueue(nil)
+ queue.Name = "testQueue"
+ assert.NilError(t, err)
+
+ app := newApplication(appID0, "default", "root")
+ queue.AddApplication(app)
+ queue.RemoveApplication(app)
+ noEvents := 0
+ err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ noEvents = eventSystem.Store.CountStoredEvents()
+ return noEvents == 3
+ })
+ assert.NilError(t, err, "expected 2 events, got %d", noEvents)
+ records := eventSystem.Store.CollectEvents()
+ assert.Equal(t, 3, len(records), "number of events")
+ assert.Equal(t, si.EventRecord_QUEUE, records[1].Type)
+ assert.Equal(t, si.EventRecord_ADD, records[1].EventChangeType)
+ assert.Equal(t, si.EventRecord_QUEUE_APP, records[1].EventChangeDetail)
+ assert.Equal(t, si.EventRecord_QUEUE, records[2].Type)
+ assert.Equal(t, si.EventRecord_REMOVE, records[2].EventChangeType)
+ assert.Equal(t, si.EventRecord_QUEUE_APP, records[2].EventChangeDetail)
+
+ newConf := configs.QueueConfig{
+ Parent: false,
+ Name: "testQueue",
+ Resources: configs.Resources{
+ Guaranteed: map[string]string{
+ "memory": "1",
+ },
+ Max: map[string]string{
+ "memory": "5",
+ },
+ },
+ }
+ err = queue.ApplyConf(newConf)
+ assert.NilError(t, err)
+ 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, 3, len(records), "number of events")
+ assert.Equal(t, si.EventRecord_QUEUE, records[0].Type)
+ assert.Equal(t, si.EventRecord_SET, records[0].EventChangeType)
+ assert.Equal(t, si.EventRecord_QUEUE_TYPE, records[0].EventChangeDetail)
+ assert.Equal(t, si.EventRecord_QUEUE, records[1].Type)
+ assert.Equal(t, si.EventRecord_SET, records[1].EventChangeType)
+ assert.Equal(t, si.EventRecord_QUEUE_MAX, records[1].EventChangeDetail)
+ assert.Equal(t, si.EventRecord_QUEUE, records[2].Type)
+ assert.Equal(t, si.EventRecord_SET, records[2].EventChangeType)
+ assert.Equal(t, si.EventRecord_QUEUE_GUARANTEED, records[2].EventChangeDetail)
+}
diff --git a/pkg/scheduler/partition_test.go b/pkg/scheduler/partition_test.go
index 078a7a26..49bd067f 100644
--- a/pkg/scheduler/partition_test.go
+++ b/pkg/scheduler/partition_test.go
@@ -31,6 +31,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/plugins"
"github.com/apache/yunikorn-core/pkg/rmproxy/rmevent"
"github.com/apache/yunikorn-core/pkg/scheduler/objects"
@@ -3358,6 +3359,38 @@ func TestTryAllocateMaxRunning(t *testing.T) {
assert.Equal(t, alloc.GetAllocationKey(), allocID, "expected ask alloc-1 to be allocated")
}
+func TestNewQueueEvents(t *testing.T) {
+ events.CreateAndSetEventSystem()
+ eventSystem := events.GetEventSystem().(*events.EventSystemImpl) //nolint:errcheck
+ eventSystem.StartServiceWithPublisher(false)
+
+ partition, err := newBasePartition()
+ assert.NilError(t, err)
+ _, err = partition.createQueue("root.test", security.UserGroup{
+ User: "test",
+ })
+ assert.NilError(t, err)
+ 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_QUEUE, records[0].Type)
+ assert.Equal(t, si.EventRecord_ADD, records[0].EventChangeType)
+ assert.Equal(t, si.EventRecord_DETAILS_NONE, records[0].EventChangeDetail)
+ assert.Equal(t, "root", records[0].ObjectID)
+ assert.Equal(t, si.EventRecord_QUEUE, records[1].Type)
+ assert.Equal(t, si.EventRecord_ADD, records[1].EventChangeType)
+ assert.Equal(t, si.EventRecord_DETAILS_NONE, records[1].EventChangeDetail)
+ assert.Equal(t, "root.default", records[1].ObjectID)
+ assert.Equal(t, si.EventRecord_QUEUE, records[2].Type)
+ assert.Equal(t, si.EventRecord_ADD, records[2].EventChangeType)
+ assert.Equal(t, si.EventRecord_QUEUE_DYNAMIC, records[2].EventChangeDetail)
+ assert.Equal(t, "root.test", records[2].ObjectID)
+}
+
func TestUserHeadroom(t *testing.T) {
setupUGM()
partition, err := newConfiguredPartition()
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@yunikorn.apache.org
For additional commands, e-mail: issues-help@yunikorn.apache.org