This is an automated email from the ASF dual-hosted git repository.
wilfreds 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 5ddcbbbd [YUNIKORN-2527] Reset state of configured queue on re-add
(#831)
5ddcbbbd is described below
commit 5ddcbbbddd38019f47e22456b1c6a797edd7fe73
Author: Wilfred Spiegelenburg <[email protected]>
AuthorDate: Wed Apr 3 22:39:49 2024 +1100
[YUNIKORN-2527] Reset state of configured queue on re-add (#831)
When a queue is removed and re-added the configured queue status is not
reset. If that happens within the cleanup cycle, the cleanup still gets
triggered. The queue is removed causing an inconsistency between the
config and the queues available in the core.
The object state needs to allow for moving from Draining back to Active.
Updated tests to cover the new state transition.
Closes: #831
Signed-off-by: Wilfred Spiegelenburg <[email protected]>
---
pkg/scheduler/objects/object_state.go | 2 +-
pkg/scheduler/objects/object_state_test.go | 10 ++---
pkg/scheduler/objects/queue.go | 8 ++++
pkg/scheduler/objects/queue_test.go | 33 +++++++++++++++-
pkg/scheduler/partition_test.go | 61 ++++++++++++++++++++++++++++++
5 files changed, 106 insertions(+), 8 deletions(-)
diff --git a/pkg/scheduler/objects/object_state.go
b/pkg/scheduler/objects/object_state.go
index 7c8a678b..3afbe595 100644
--- a/pkg/scheduler/objects/object_state.go
+++ b/pkg/scheduler/objects/object_state.go
@@ -68,7 +68,7 @@ func NewObjectState() *fsm.FSM {
Dst: Draining.String(),
}, {
Name: Start.String(),
- Src: []string{Active.String(),
Stopped.String()},
+ Src: []string{Active.String(),
Stopped.String(), Draining.String()},
Dst: Active.String(),
}, {
Name: Stop.String(),
diff --git a/pkg/scheduler/objects/object_state_test.go
b/pkg/scheduler/objects/object_state_test.go
index 20cab62d..dbe0180c 100644
--- a/pkg/scheduler/objects/object_state_test.go
+++ b/pkg/scheduler/objects/object_state_test.go
@@ -50,15 +50,15 @@ func TestStateTransition(t *testing.T) {
assert.Assert(t, err == nil)
assert.Equal(t, stateMachine.Current(), Draining.String())
- // start on draining not allowed
- err = stateMachine.Event(context.Background(), Start.String(),
"test_object")
- assert.Assert(t, err != nil)
- assert.Equal(t, stateMachine.Current(), Draining.String())
-
// stop on draining not allowed
err = stateMachine.Event(context.Background(), Stop.String(),
"test_object")
assert.Assert(t, err != nil)
assert.Equal(t, stateMachine.Current(), Draining.String())
+
+ // draining to active
+ err = stateMachine.Event(context.Background(), Start.String(),
"test_object")
+ assert.Assert(t, err == nil)
+ assert.Equal(t, stateMachine.Current(), Active.String())
}
func TestTransitionToSelf(t *testing.T) {
diff --git a/pkg/scheduler/objects/queue.go b/pkg/scheduler/objects/queue.go
index 59a3d371..9f272284 100644
--- a/pkg/scheduler/objects/queue.go
+++ b/pkg/scheduler/objects/queue.go
@@ -328,6 +328,14 @@ func (sq *Queue) applyConf(conf configs.QueueConfig) error
{
sq.isManaged = true
}
+ // if the queue is marked for removal reverse that state
+ if !sq.IsRunning() {
+ err = sq.handleQueueEvent(Start)
+ if err != nil {
+ log.Log(log.SchedQueue).Info("managed queue state
change failed",
+ zap.String("queue", sq.QueuePath))
+ }
+ }
prevLeaf := sq.isLeaf
sq.isLeaf = !conf.Parent
// Make sure the parent flag is set correctly: config might expect auto
parent type creation
diff --git a/pkg/scheduler/objects/queue_test.go
b/pkg/scheduler/objects/queue_test.go
index ff48eadd..f9111d65 100644
--- a/pkg/scheduler/objects/queue_test.go
+++ b/pkg/scheduler/objects/queue_test.go
@@ -632,8 +632,8 @@ func TestQueueStates(t *testing.T) {
t.Errorf("leaf queue is not marked draining: %v", err)
}
err = leaf.handleQueueEvent(Start)
- if err == nil || !leaf.IsDraining() {
- t.Errorf("leaf queue changed state which should not happen:
%v", err)
+ if err != nil || !leaf.IsRunning() {
+ t.Errorf("leaf queue is not marked running: %v", err)
}
}
@@ -2210,6 +2210,35 @@ func TestNewConfiguredQueue(t *testing.T) {
assert.Assert(t, childNonLeaf.maxResource == nil)
}
+func TestResetRunningState(t *testing.T) {
+ emptyConf := configs.QueueConfig{
+ Name: "not_used",
+ }
+ // create the root
+ root, err := createRootQueue(nil)
+ assert.NilError(t, err, "queue create failed")
+ // single parent under root
+ var parent *Queue
+ parent, err = createManagedQueue(root, "parent", true, nil)
+ assert.NilError(t, err, "failed to create parent queue")
+ var leaf *Queue
+ leaf, err = createManagedQueue(parent, "leaf", false, nil)
+ assert.NilError(t, err, "failed to create leaf queue")
+ if len(parent.children) == 0 {
+ t.Error("leaf queue is not added to the parent queue")
+ }
+ parent.MarkQueueForRemoval()
+ assert.Assert(t, parent.IsDraining(), "parent should be marked as
draining")
+ assert.Assert(t, leaf.IsDraining(), "leaf should be marked as draining")
+ err = parent.applyConf(emptyConf)
+ assert.NilError(t, err, "failed to update parent")
+ assert.Assert(t, parent.IsRunning(), "parent should be running again")
+ assert.Assert(t, leaf.IsDraining(), "leaf should still be marked as
draining")
+ err = leaf.applyConf(emptyConf)
+ assert.NilError(t, err, "failed to update leaf")
+ assert.Assert(t, leaf.IsRunning(), "leaf should be running again")
+}
+
func TestNewRecoveryQueue(t *testing.T) {
var err error
if _, err = NewRecoveryQueue(nil); err == nil {
diff --git a/pkg/scheduler/partition_test.go b/pkg/scheduler/partition_test.go
index 90159e79..c42d784e 100644
--- a/pkg/scheduler/partition_test.go
+++ b/pkg/scheduler/partition_test.go
@@ -1426,6 +1426,67 @@ func TestUpdateQueues(t *testing.T) {
assertUpdateQueues(t, "both", map[string]string{})
}
+func TestReAddQueues(t *testing.T) {
+ conf := []configs.QueueConfig{
+ {
+ Name: "parent",
+ Parent: true,
+ Queues: []configs.QueueConfig{
+ {
+ Name: "leaf",
+ Parent: false,
+ Queues: nil,
+ },
+ },
+ },
+ }
+
+ confDefault := []configs.QueueConfig{
+ {
+ Name: "default",
+ Parent: false,
+ Queues: nil,
+ },
+ {
+ Name: "parent",
+ Parent: true,
+ Queues: nil,
+ },
+ }
+
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ // There is a queue setup as the config must be valid when we run
+ root := partition.GetQueue("root")
+ if root == nil {
+ t.Error("root queue not found in partition")
+ }
+ def := partition.GetQueue(defQueue)
+ if def == nil {
+ t.Fatal("default queue should exist")
+ }
+ err = partition.updateQueues(conf, root)
+ assert.NilError(t, err, "queue update from config failed")
+ leaf := partition.GetQueue("root.parent.leaf")
+ if leaf == nil {
+ t.Fatal("leaf queue should be created")
+ }
+ assert.Assert(t, def.IsDraining(), "'root.default' queue should have
been marked for removal")
+ err = partition.updateQueues(confDefault, root)
+ assert.NilError(t, err, "queue update from config default failed")
+ assert.Assert(t, def.IsRunning(), "'root.default' queue should have
been marked running again")
+ assert.Assert(t, leaf.IsDraining(), "'root.parent.leaf' queue should
have been marked for removal")
+ partition.partitionManager.cleanQueues(root)
+ leaf = partition.GetQueue("root.parent.leaf")
+ if leaf != nil {
+ t.Fatal("leaf queue should have been cleaned up")
+ }
+ def = partition.GetQueue(defQueue)
+ if def == nil {
+ t.Fatal("default queue should still exist")
+ }
+}
+
func TestGetApplication(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
