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 81d69749 [YUNIKORN-2581] Expose running placement rules in REST (#857)
81d69749 is described below
commit 81d697494d972d01c01135dd981e1ec2fc89f57d
Author: Wilfred Spiegelenburg <wilfr...@apache.org>
AuthorDate: Fri May 31 11:26:57 2024 +1000
[YUNIKORN-2581] Expose running placement rules in REST (#857)
Add a new REST endpoint under the partition that returns the currently
active placement rules:
/ws/v1/partition/:partition/placementrules
The call retrieves the active set from the placement manager. This will
include the recovery rule. The rule interface is extended with a new
private method ruleDAO() that must be implemented by all rules and
returns the RuleDAO object.
Add the currently active set of placement rules to the state dump. This
handles multiple partitions. It wraps the list of RuleDAO objects in a
RuleDAOInfo per partition.
Closes: #857
Signed-off-by: Wilfred Spiegelenburg <wilfr...@apache.org>
---
go.mod | 1 +
go.sum | 2 +
pkg/scheduler/partition.go | 9 ++-
pkg/scheduler/placement/filter.go | 47 ++++++++++++-
pkg/scheduler/placement/filter_test.go | 65 +++++++++++++-----
pkg/scheduler/placement/fixed_rule.go | 19 +++++
pkg/scheduler/placement/fixed_rule_test.go | 38 ++++++++++
pkg/scheduler/placement/placement.go | 23 +++++--
pkg/scheduler/placement/placement_test.go | 51 ++++++++++++--
pkg/scheduler/placement/provided_rule.go | 17 +++++
pkg/scheduler/placement/provided_rule_test.go | 33 +++++++++
pkg/scheduler/placement/recovery_rule.go | 10 +++
pkg/scheduler/placement/recovery_rule_test.go | 9 +++
pkg/scheduler/placement/rule.go | 44 ++++++++----
pkg/scheduler/placement/rule_test.go | 9 +++
pkg/scheduler/placement/tag_rule.go | 18 +++++
pkg/scheduler/placement/tag_rule_test.go | 38 ++++++++++
pkg/scheduler/placement/testrule.go | 17 +++++
pkg/scheduler/placement/user_rule.go | 17 +++++
pkg/scheduler/placement/user_rule_test.go | 33 +++++++++
pkg/webservice/dao/rule_info.go | 39 +++++++++++
pkg/webservice/handlers.go | 32 +++++++++
pkg/webservice/handlers_test.go | 99 ++++++++++++++++++++++++---
pkg/webservice/routes.go | 6 ++
pkg/webservice/state_dump.go | 2 +
25 files changed, 623 insertions(+), 55 deletions(-)
diff --git a/go.mod b/go.mod
index 65b2c5a8..5f180926 100644
--- a/go.mod
+++ b/go.mod
@@ -33,6 +33,7 @@ require (
github.com/prometheus/common v0.45.0
github.com/sasha-s/go-deadlock v0.3.1
go.uber.org/zap v1.26.0
+ golang.org/x/exp v0.0.0-20240409090435-93d18d7e34b8
golang.org/x/net v0.21.0
golang.org/x/time v0.5.0
google.golang.org/grpc v1.58.3
diff --git a/go.sum b/go.sum
index 35ee67e4..f04fef47 100644
--- a/go.sum
+++ b/go.sum
@@ -50,6 +50,8 @@ go.uber.org/multierr v1.10.0
h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ=
go.uber.org/multierr v1.10.0/go.mod
h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
go.uber.org/zap v1.26.0 h1:sI7k6L95XOKS281NhVKOFCUNIvv9e0w4BF8N3u+tCRo=
go.uber.org/zap v1.26.0/go.mod h1:dtElttAiwGvoJ/vj4IwHBS/gXsEu/pZ50mUIRWuG0so=
+golang.org/x/exp v0.0.0-20240409090435-93d18d7e34b8
h1:ESSUROHIBHg7USnszlcdmjBEwdMj9VUvU+OPk4yl2mc=
+golang.org/x/exp v0.0.0-20240409090435-93d18d7e34b8/go.mod
h1:/lliqkxwWAhPjf5oSOIJup2XcqJaw8RGS6k3TGEc7GI=
golang.org/x/net v0.23.0 h1:7EYJ93RZ9vYSZAIb2x3lnuvqO5zneoD6IvWjuhfxjTs=
golang.org/x/net v0.23.0/go.mod h1:JKghWKKOSdJwpW2GEx0Ja7fmaKnMsbu+MWVZTokSYmg=
golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4=
diff --git a/pkg/scheduler/partition.go b/pkg/scheduler/partition.go
index 5662bd75..fc2c5404 100644
--- a/pkg/scheduler/partition.go
+++ b/pkg/scheduler/partition.go
@@ -481,14 +481,19 @@ func (pc *PartitionContext) getQueueInternal(name string)
*objects.Queue {
return queue
}
-// Get the queue info for the whole queue structure to pass to the webservice
+// GetPartitionQueues builds the queue info for the whole queue structure to
pass to the webservice
func (pc *PartitionContext) GetPartitionQueues() dao.PartitionQueueDAOInfo {
partitionQueueDAOInfo := pc.root.GetPartitionQueueDAOInfo(true)
partitionQueueDAOInfo.Partition =
common.GetPartitionNameWithoutClusterID(pc.Name)
return partitionQueueDAOInfo
}
-// Create the recovery queue.
+// GetPlacementRules returns the current active rule set as dao to expose to
the webservice
+func (pc *PartitionContext) GetPlacementRules() []*dao.RuleDAO {
+ return pc.getPlacementManager().GetRulesDAO()
+}
+
+// createRecoveryQueue creates the recovery queue to add to the hierarchy
func (pc *PartitionContext) createRecoveryQueue() (*objects.Queue, error) {
return objects.NewRecoveryQueue(pc.root)
}
diff --git a/pkg/scheduler/placement/filter.go
b/pkg/scheduler/placement/filter.go
index 72f8181f..df94a002 100644
--- a/pkg/scheduler/placement/filter.go
+++ b/pkg/scheduler/placement/filter.go
@@ -22,10 +22,17 @@ import (
"regexp"
"go.uber.org/zap"
+ "golang.org/x/exp/maps"
"github.com/apache/yunikorn-core/pkg/common/configs"
"github.com/apache/yunikorn-core/pkg/common/security"
"github.com/apache/yunikorn-core/pkg/log"
+ "github.com/apache/yunikorn-core/pkg/webservice/dao"
+)
+
+const (
+ filterAllow = "allow"
+ filterDeny = "deny"
)
type Filter struct {
@@ -51,7 +58,7 @@ func (filter Filter) allowUser(userObj security.UserGroup)
bool {
// if we have found the user in the list stop looking and return
if filteredUser {
log.Log(log.Config).Debug("Filter matched user getName",
zap.String("user", user))
- return filteredUser && filter.allow
+ return filter.allow
}
// not in the user list, check the groups in the list
// walk over the list first match is taken
@@ -59,7 +66,7 @@ func (filter Filter) allowUser(userObj security.UserGroup)
bool {
filteredUser = filter.filterGroup(group)
if filteredUser {
log.Log(log.Config).Debug("Filter matched user group",
zap.String("group", group))
- return filteredUser && filter.allow
+ return filter.allow
}
}
return !filter.allow
@@ -85,6 +92,40 @@ func (filter Filter) filterGroup(group string) bool {
return filter.groupList[group]
}
+// filterDAO returns the DAO object for the filter.
+// Returns nil if the filter is considered "empty"
+func (filter Filter) filterDAO() *dao.FilterDAO {
+ // do not render an empty filter in the DAO
+ if filter.empty {
+ return nil
+ }
+ ft := filterAllow
+ if !filter.allow {
+ ft = filterDeny
+ }
+ var userList, groupList []string
+ if len(filter.userList) != 0 {
+ userList = maps.Keys(filter.userList)
+ }
+ if len(filter.groupList) != 0 {
+ groupList = maps.Keys(filter.groupList)
+ }
+ var userExp, groupExp string
+ if filter.userExp != nil {
+ userExp = filter.userExp.String()
+ }
+ if filter.groupExp != nil {
+ groupExp = filter.groupExp.String()
+ }
+ return &dao.FilterDAO{
+ Type: ft,
+ UserList: userList,
+ GroupList: groupList,
+ UserExp: userExp,
+ GroupExp: groupExp,
+ }
+}
+
// Create a new filter based on the checked config
// There should be no errors as the config is syntax checked before we get to
this point.
func newFilter(conf configs.Filter) Filter {
@@ -94,7 +135,7 @@ func newFilter(conf configs.Filter) Filter {
empty: true,
}
// type can only be '' , allow or deny.
- filter.allow = conf.Type != "deny"
+ filter.allow = conf.Type != filterDeny
var err error
// create the user list or regexp
diff --git a/pkg/scheduler/placement/filter_test.go
b/pkg/scheduler/placement/filter_test.go
index 9c2011ae..99afc9bb 100644
--- a/pkg/scheduler/placement/filter_test.go
+++ b/pkg/scheduler/placement/filter_test.go
@@ -19,16 +19,19 @@
package placement
import (
+ "reflect"
+ "regexp"
"testing"
"github.com/apache/yunikorn-core/pkg/common/configs"
"github.com/apache/yunikorn-core/pkg/common/security"
+ "github.com/apache/yunikorn-core/pkg/webservice/dao"
)
func TestNewFilterLists(t *testing.T) {
// test simple no user or group: allow
conf := configs.Filter{}
- conf.Type = "allow"
+ conf.Type = filterAllow
filter := newFilter(conf)
if !filter.allow {
@@ -46,7 +49,7 @@ func TestNewFilterLists(t *testing.T) {
// test simple no user or group: deny
conf = configs.Filter{}
- conf.Type = "deny"
+ conf.Type = filterDeny
filter = newFilter(conf)
if filter.allow {
@@ -64,7 +67,7 @@ func TestNewFilterLists(t *testing.T) {
// test simple empty lists
conf = configs.Filter{}
- conf.Type = "allow"
+ conf.Type = filterAllow
conf.Users = []string{}
conf.Groups = []string{}
@@ -124,7 +127,7 @@ func TestNewFilterLists(t *testing.T) {
func TestNewFilterExpressions(t *testing.T) {
// test expression
conf := configs.Filter{}
- conf.Type = "allow"
+ conf.Type = filterAllow
conf.Users = []string{"user*"}
conf.Groups = []string{"group[1-9]"}
@@ -271,7 +274,7 @@ func TestAllowUser(t *testing.T) {
}
// test deny user list
conf := configs.Filter{}
- conf.Type = "deny"
+ conf.Type = filterDeny
conf.Users = []string{"user1"}
filter := newFilter(conf)
@@ -286,7 +289,7 @@ func TestAllowUser(t *testing.T) {
// test allow user list
conf = configs.Filter{}
- conf.Type = "allow"
+ conf.Type = filterAllow
conf.Users = []string{"user1"}
filter = newFilter(conf)
@@ -301,7 +304,7 @@ func TestAllowUser(t *testing.T) {
// test deny user exp
conf = configs.Filter{}
- conf.Type = "deny"
+ conf.Type = filterDeny
conf.Users = []string{"user[0-9]"}
filter = newFilter(conf)
@@ -316,7 +319,7 @@ func TestAllowUser(t *testing.T) {
// test allow user exp
conf = configs.Filter{}
- conf.Type = "allow"
+ conf.Type = filterAllow
conf.Users = []string{"user[0-9]"}
filter = newFilter(conf)
@@ -340,7 +343,7 @@ func TestAllowGroup(t *testing.T) {
// test deny group list
conf := configs.Filter{}
- conf.Type = "deny"
+ conf.Type = filterDeny
conf.Groups = []string{"group1"}
filter := newFilter(conf)
@@ -355,7 +358,7 @@ func TestAllowGroup(t *testing.T) {
// test allow group list
conf = configs.Filter{}
- conf.Type = "allow"
+ conf.Type = filterAllow
conf.Groups = []string{"group1"}
filter = newFilter(conf)
@@ -370,7 +373,7 @@ func TestAllowGroup(t *testing.T) {
// test deny group exp
conf = configs.Filter{}
- conf.Type = "deny"
+ conf.Type = filterDeny
conf.Groups = []string{"group[0-9]"}
filter = newFilter(conf)
@@ -385,7 +388,7 @@ func TestAllowGroup(t *testing.T) {
// test allow group exp
conf = configs.Filter{}
- conf.Type = "allow"
+ conf.Type = filterAllow
conf.Groups = []string{"group[0-9]"}
filter = newFilter(conf)
@@ -409,7 +412,7 @@ func TestAllowSecondaryGroup(t *testing.T) {
// test deny group list
conf := configs.Filter{}
- conf.Type = "deny"
+ conf.Type = filterDeny
conf.Groups = []string{"group2"}
filter := newFilter(conf)
@@ -420,7 +423,7 @@ func TestAllowSecondaryGroup(t *testing.T) {
// test allow group list
conf = configs.Filter{}
- conf.Type = "allow"
+ conf.Type = filterAllow
conf.Groups = []string{"group1", "group2"}
filter = newFilter(conf)
@@ -431,7 +434,7 @@ func TestAllowSecondaryGroup(t *testing.T) {
// test deny group exp
conf = configs.Filter{}
- conf.Type = "deny"
+ conf.Type = filterDeny
conf.Groups = []string{"group[0-9]"}
filter = newFilter(conf)
@@ -442,7 +445,7 @@ func TestAllowSecondaryGroup(t *testing.T) {
// test allow group exp
conf = configs.Filter{}
- conf.Type = "allow"
+ conf.Type = filterAllow
conf.Groups = []string{"group[0-9]"}
filter = newFilter(conf)
@@ -469,7 +472,7 @@ func TestAllowNoLists(t *testing.T) {
}
// test default allow behaviour (no filter)
conf = configs.Filter{}
- conf.Type = "allow"
+ conf.Type = filterAllow
filter = newFilter(conf)
if !filter.allowUser(userObj) {
@@ -477,10 +480,36 @@ func TestAllowNoLists(t *testing.T) {
}
// test default deny behaviour (no filter)
conf = configs.Filter{}
- conf.Type = "deny"
+ conf.Type = filterDeny
filter = newFilter(conf)
if filter.allowUser(userObj) {
t.Error("deny filter type only did not deny user")
}
}
+
+func TestFilter_filterDAO(t *testing.T) {
+ // filters are tested also from each rule in different combinations
+ // this does the outliers and cases that should not happen
+ reg := regexp.MustCompile("^.*$")
+ tests := []struct {
+ name string
+ filter Filter
+ want *dao.FilterDAO
+ }{
+ {"empty", Filter{empty: true}, nil},
+ {"empty", Filter{}, &dao.FilterDAO{Type: filterDeny}},
+ {
+ "everything",
+ Filter{allow: true, userList: map[string]bool{"user":
true}, groupList: map[string]bool{"group": true}, userExp: reg, groupExp: reg},
+ &dao.FilterDAO{Type: filterAllow, UserList:
[]string{"user"}, GroupList: []string{"group"}, UserExp: "^.*$", GroupExp:
"^.*$"},
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ if got := tt.filter.filterDAO();
!reflect.DeepEqual(got, tt.want) {
+ t.Errorf("filterDAO() = %v, want %v", got,
tt.want)
+ }
+ })
+ }
+}
diff --git a/pkg/scheduler/placement/fixed_rule.go
b/pkg/scheduler/placement/fixed_rule.go
index b62abd51..3e2e9216 100644
--- a/pkg/scheduler/placement/fixed_rule.go
+++ b/pkg/scheduler/placement/fixed_rule.go
@@ -20,6 +20,7 @@ package placement
import (
"fmt"
+ "strconv"
"strings"
"go.uber.org/zap"
@@ -28,6 +29,7 @@ import (
"github.com/apache/yunikorn-core/pkg/log"
"github.com/apache/yunikorn-core/pkg/scheduler/objects"
"github.com/apache/yunikorn-core/pkg/scheduler/placement/types"
+ "github.com/apache/yunikorn-core/pkg/webservice/dao"
)
// A rule to place an application based on the queue in the configuration.
@@ -44,6 +46,23 @@ func (fr *fixedRule) getName() string {
return types.Fixed
}
+func (fr *fixedRule) ruleDAO() *dao.RuleDAO {
+ var pDAO *dao.RuleDAO
+ if fr.parent != nil {
+ pDAO = fr.parent.ruleDAO()
+ }
+ return &dao.RuleDAO{
+ Name: fr.getName(),
+ Parameters: map[string]string{
+ "queue": fr.queue,
+ "create": strconv.FormatBool(fr.create),
+ "qualified": strconv.FormatBool(fr.qualified),
+ },
+ ParentRule: pDAO,
+ Filter: fr.filter.filterDAO(),
+ }
+}
+
func (fr *fixedRule) initialise(conf configs.PlacementRule) error {
fr.queue = normalise(conf.Value)
if fr.queue == "" {
diff --git a/pkg/scheduler/placement/fixed_rule_test.go
b/pkg/scheduler/placement/fixed_rule_test.go
index 5067f93e..c1d1e6cc 100644
--- a/pkg/scheduler/placement/fixed_rule_test.go
+++ b/pkg/scheduler/placement/fixed_rule_test.go
@@ -25,6 +25,7 @@ import (
"github.com/apache/yunikorn-core/pkg/common/configs"
"github.com/apache/yunikorn-core/pkg/common/security"
+ "github.com/apache/yunikorn-core/pkg/webservice/dao"
)
func TestFixedRule(t *testing.T) {
@@ -248,3 +249,40 @@ func TestFixedRuleParent(t *testing.T) {
t.Errorf("fixed rule with parent declared as leaf should have
failed '%s', error %v", queue, err)
}
}
+
+func Test_fixedRule_ruleDAO(t *testing.T) {
+ tests := []struct {
+ name string
+ conf configs.PlacementRule
+ want *dao.RuleDAO
+ }{
+ {
+ "base",
+ configs.PlacementRule{Name: "fixed", Value: "default"},
+ &dao.RuleDAO{Name: "fixed", Parameters:
map[string]string{"queue": "default", "qualified": "false", "create": "false"}},
+ },
+ {
+ "qualified",
+ configs.PlacementRule{Name: "fixed", Value:
"root.default"},
+ &dao.RuleDAO{Name: "fixed", Parameters:
map[string]string{"queue": "root.default", "qualified": "true", "create":
"false"}},
+ },
+ {
+ "parent",
+ configs.PlacementRule{Name: "fixed", Value: "default",
Create: true, Parent: &configs.PlacementRule{Name: "test", Create: true}},
+ &dao.RuleDAO{Name: "fixed", Parameters:
map[string]string{"queue": "default", "qualified": "false", "create": "true"},
ParentRule: &dao.RuleDAO{Name: "test", Parameters: map[string]string{"create":
"true"}}},
+ },
+ {
+ "filter",
+ configs.PlacementRule{Name: "fixed", Value: "default",
Create: true, Filter: configs.Filter{Type: filterAllow, Groups:
[]string{"group1", "group2"}}},
+ &dao.RuleDAO{Name: "fixed", Parameters:
map[string]string{"queue": "default", "qualified": "false", "create": "true"},
Filter: &dao.FilterDAO{Type: filterAllow, GroupList: []string{"group1",
"group2"}}},
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ ur, err := newRule(tt.conf)
+ assert.NilError(t, err, "setting up the rule failed")
+ ruleDAO := ur.ruleDAO()
+ assert.DeepEqual(t, tt.want, ruleDAO)
+ })
+ }
+}
diff --git a/pkg/scheduler/placement/placement.go
b/pkg/scheduler/placement/placement.go
index 4aad4fce..196ccfd1 100644
--- a/pkg/scheduler/placement/placement.go
+++ b/pkg/scheduler/placement/placement.go
@@ -30,6 +30,7 @@ import (
"github.com/apache/yunikorn-core/pkg/log"
"github.com/apache/yunikorn-core/pkg/scheduler/objects"
"github.com/apache/yunikorn-core/pkg/scheduler/placement/types"
+ "github.com/apache/yunikorn-core/pkg/webservice/dao"
)
// RejectedError is the standard error returned if placement has failed
@@ -52,10 +53,19 @@ func NewPlacementManager(rules []configs.PlacementRule,
queueFunc func(string) *
return m
}
-// Update the rules for an active placement manager
-// Note that this will only be called when the manager is created earlier and
the config is updated.
+// GetRulesDAO returns a list of RuleDAO objects of the configured rules
+func (m *AppPlacementManager) GetRulesDAO() []*dao.RuleDAO {
+ m.RLock()
+ defer m.RUnlock()
+ info := make([]*dao.RuleDAO, len(m.rules))
+ for i, r := range m.rules {
+ info[i] = r.ruleDAO()
+ }
+ return info
+}
+
+// UpdateRules sets the rules for an active placement manager
func (m *AppPlacementManager) UpdateRules(rules []configs.PlacementRule) error
{
- log.Log(log.Config).Info("Building new rule list for placement manager")
if err := m.initialise(rules); err != nil {
log.Log(log.Config).Info("Placement manager rules not
reloaded", zap.Error(err))
return err
@@ -63,7 +73,7 @@ func (m *AppPlacementManager) UpdateRules(rules
[]configs.PlacementRule) error {
return nil
}
-// Initialise the rules from a parsed config.
+// initialise the rules from a parsed config.
func (m *AppPlacementManager) initialise(rules []configs.PlacementRule) error {
log.Log(log.Config).Info("Building new rule list for placement manager")
// build temp list from new config
@@ -85,6 +95,9 @@ func (m *AppPlacementManager) initialise(rules
[]configs.PlacementRule) error {
return nil
}
+// PlaceApplication executes the rules for the passed in application.
+// On success the queueName of the application is set to the queue the
application wil run in.
+// On failure the queueName is set to "" and an error is returned.
func (m *AppPlacementManager) PlaceApplication(app *objects.Application) error
{
m.RLock()
defer m.RUnlock()
@@ -174,7 +187,7 @@ func (m *AppPlacementManager) PlaceApplication(app
*objects.Application) error {
return nil
}
-// Build the rule set based on the config.
+// buildRules builds a new rule set based on the config.
// If the rule set is correct and can be used the new set is returned.
// If any error is encountered a nil array is returned and the error set
func buildRules(rules []configs.PlacementRule) ([]rule, error) {
diff --git a/pkg/scheduler/placement/placement_test.go
b/pkg/scheduler/placement/placement_test.go
index b70efc07..14fe6ace 100644
--- a/pkg/scheduler/placement/placement_test.go
+++ b/pkg/scheduler/placement/placement_test.go
@@ -35,7 +35,7 @@ import (
func TestManagerNew(t *testing.T) {
// basic info without rules, manager should not init
man := NewPlacementManager(nil, queueFunc)
- assert.Equal(t, 2, len(man.rules), "wrong rule count for empty
placement manager")
+ assert.Equal(t, 2, len(man.rules), "wrong rule count for new placement
manager, no config")
assert.Equal(t, types.Provided, man.rules[0].getName(), "wrong name for
implicit provided rule")
assert.Equal(t, types.Recovery, man.rules[1].getName(), "wrong name for
implicit recovery rule")
}
@@ -43,13 +43,21 @@ func TestManagerNew(t *testing.T) {
func TestManagerInit(t *testing.T) {
// basic info without rules, manager should implicitly init
man := NewPlacementManager(nil, queueFunc)
- assert.Equal(t, 2, len(man.rules), "wrong rule count for nil placement
manager")
+ assert.Equal(t, 2, len(man.rules), "wrong rule count for nil rules
config")
+ ruleDAOs := man.GetRulesDAO()
+ assert.Equal(t, 2, len(ruleDAOs), "wrong DAO count for nil rules
config")
+ assert.Equal(t, ruleDAOs[0].Name, types.Provided, "expected provided
rule as first rule")
+ assert.Equal(t, ruleDAOs[1].Name, types.Recovery, "expected recovery
rule as second rule")
// try to init with empty list should do the same
var rules []configs.PlacementRule
err := man.initialise(rules)
assert.NilError(t, err, "Failed to initialize empty placement rules")
- assert.Equal(t, 2, len(man.rules), "wrong rule count for empty
placement manager")
+ assert.Equal(t, 2, len(man.rules), "wrong rule count for empty rules
config")
+ ruleDAOs = man.GetRulesDAO()
+ assert.Equal(t, 2, len(ruleDAOs), "wrong DAO count for empty rules
config")
+ assert.Equal(t, ruleDAOs[0].Name, types.Provided, "expected provided
rule as first rule")
+ assert.Equal(t, ruleDAOs[1].Name, types.Recovery, "expected recovery
rule as second rule")
rules = []configs.PlacementRule{
{Name: "unknown"},
@@ -58,6 +66,11 @@ func TestManagerInit(t *testing.T) {
if err == nil {
t.Error("initialise with 'unknown' rule list should have
failed")
}
+ // rules should not have changed
+ ruleDAOs = man.GetRulesDAO()
+ assert.Equal(t, 2, len(ruleDAOs), "unexpected change: wrong DAO count
for failed reinit")
+ assert.Equal(t, ruleDAOs[0].Name, types.Provided, "expected provided
rule as first rule")
+ assert.Equal(t, ruleDAOs[1].Name, types.Recovery, "expected recovery
rule as second rule")
// init the manager with one rule
rules = []configs.PlacementRule{
@@ -65,17 +78,29 @@ func TestManagerInit(t *testing.T) {
}
err = man.initialise(rules)
assert.NilError(t, err, "failed to init existing manager")
+ ruleDAOs = man.GetRulesDAO()
+ assert.Equal(t, 2, len(ruleDAOs), "wrong DAO count for manager with
test rule")
+ assert.Equal(t, ruleDAOs[0].Name, types.Test, "expected test rule as
first rule")
+ assert.Equal(t, ruleDAOs[1].Name, types.Recovery, "expected recovery
rule as second rule")
// update the manager: remove rules implicit state is reverted
rules = []configs.PlacementRule{}
err = man.initialise(rules)
assert.NilError(t, err, "Failed to re-initialize empty placement rules")
- assert.Equal(t, 2, len(man.rules), "wrong rule count for newly empty
placement manager")
+ assert.Equal(t, 2, len(man.rules), "wrong rule count for empty rules
config")
+ ruleDAOs = man.GetRulesDAO()
+ assert.Equal(t, 2, len(ruleDAOs), "wrong DAO count for empty rules
list")
+ assert.Equal(t, ruleDAOs[0].Name, types.Provided, "expected provided
rule as first rule")
+ assert.Equal(t, ruleDAOs[1].Name, types.Recovery, "expected recovery
rule as second rule")
// check if we handle a nil list
err = man.initialise(nil)
assert.NilError(t, err, "Failed to re-initialize nil placement rules")
- assert.Equal(t, 2, len(man.rules), "wrong rule count for nil placement
manager")
+ assert.Equal(t, 2, len(man.rules), "wrong rule count for nil rules
config")
+ ruleDAOs = man.GetRulesDAO()
+ assert.Equal(t, 2, len(ruleDAOs), "wrong DAO count for nil rules
config")
+ assert.Equal(t, ruleDAOs[0].Name, types.Provided, "expected provided
rule as first rule")
+ assert.Equal(t, ruleDAOs[1].Name, types.Recovery, "expected recovery
rule as second rule")
}
func TestManagerUpdate(t *testing.T) {
@@ -87,17 +112,29 @@ func TestManagerUpdate(t *testing.T) {
}
err := man.UpdateRules(rules)
assert.NilError(t, err, "failed to update existing manager")
+ ruleDAOs := man.GetRulesDAO()
+ assert.Equal(t, 2, len(ruleDAOs), "wrong DAO count for manager with
test rule")
+ assert.Equal(t, ruleDAOs[0].Name, types.Test, "expected test rule as
first rule")
+ assert.Equal(t, ruleDAOs[1].Name, types.Recovery, "expected recovery
rule as second rule")
// update the manager: remove rules init state is reverted
rules = []configs.PlacementRule{}
err = man.UpdateRules(rules)
assert.NilError(t, err, "Failed to re-initialize empty placement rules")
- assert.Equal(t, 2, len(man.rules), "wrong rule count for newly empty
placement manager")
+ assert.Equal(t, 2, len(man.rules), "wrong rule count for empty rules
config")
+ ruleDAOs = man.GetRulesDAO()
+ assert.Equal(t, 2, len(ruleDAOs), "wrong DAO count for empty rules
config")
+ assert.Equal(t, ruleDAOs[0].Name, types.Provided, "expected provided
rule as first rule")
+ assert.Equal(t, ruleDAOs[1].Name, types.Recovery, "expected recovery
rule as second rule")
// check if we handle a nil list
err = man.UpdateRules(nil)
assert.NilError(t, err, "Failed to re-initialize nil placement rules")
- assert.Equal(t, 2, len(man.rules), "wrong rule count for nil placement
manager")
+ assert.Equal(t, 2, len(man.rules), "wrong rule count for nil rules
config")
+ ruleDAOs = man.GetRulesDAO()
+ assert.Equal(t, 2, len(ruleDAOs), "wrong DAO count for nil rules
config")
+ assert.Equal(t, ruleDAOs[0].Name, types.Provided, "expected provided
rule as first rule")
+ assert.Equal(t, ruleDAOs[1].Name, types.Recovery, "expected recovery
rule as second rule")
}
func TestManagerBuildRule(t *testing.T) {
diff --git a/pkg/scheduler/placement/provided_rule.go
b/pkg/scheduler/placement/provided_rule.go
index 27974a78..834ff83c 100644
--- a/pkg/scheduler/placement/provided_rule.go
+++ b/pkg/scheduler/placement/provided_rule.go
@@ -20,6 +20,7 @@ package placement
import (
"fmt"
+ "strconv"
"strings"
"go.uber.org/zap"
@@ -28,6 +29,7 @@ import (
"github.com/apache/yunikorn-core/pkg/log"
"github.com/apache/yunikorn-core/pkg/scheduler/objects"
"github.com/apache/yunikorn-core/pkg/scheduler/placement/types"
+ "github.com/apache/yunikorn-core/pkg/webservice/dao"
)
// A rule to place an application based on the queue provided by the user on
submission.
@@ -42,6 +44,21 @@ func (pr *providedRule) getName() string {
return types.Provided
}
+func (pr *providedRule) ruleDAO() *dao.RuleDAO {
+ var pDAO *dao.RuleDAO
+ if pr.parent != nil {
+ pDAO = pr.parent.ruleDAO()
+ }
+ return &dao.RuleDAO{
+ Name: pr.getName(),
+ Parameters: map[string]string{
+ "create": strconv.FormatBool(pr.create),
+ },
+ ParentRule: pDAO,
+ Filter: pr.filter.filterDAO(),
+ }
+}
+
func (pr *providedRule) initialise(conf configs.PlacementRule) error {
pr.create = conf.Create
pr.filter = newFilter(conf.Filter)
diff --git a/pkg/scheduler/placement/provided_rule_test.go
b/pkg/scheduler/placement/provided_rule_test.go
index 5077c834..1a14d8f1 100644
--- a/pkg/scheduler/placement/provided_rule_test.go
+++ b/pkg/scheduler/placement/provided_rule_test.go
@@ -25,6 +25,7 @@ import (
"github.com/apache/yunikorn-core/pkg/common/configs"
"github.com/apache/yunikorn-core/pkg/common/security"
+ "github.com/apache/yunikorn-core/pkg/webservice/dao"
)
func TestProvidedRulePlace(t *testing.T) {
@@ -207,3 +208,35 @@ func TestProvidedRuleParent(t *testing.T) {
t.Errorf("provided rule placed app in incorrect queue '%s', err
%v", queue, err)
}
}
+
+func Test_providedRule_ruleDAO(t *testing.T) {
+ tests := []struct {
+ name string
+ conf configs.PlacementRule
+ want *dao.RuleDAO
+ }{
+ {
+ "base",
+ configs.PlacementRule{Name: "provided"},
+ &dao.RuleDAO{Name: "provided", Parameters:
map[string]string{"create": "false"}},
+ },
+ {
+ "parent",
+ configs.PlacementRule{Name: "provided", Create: true,
Parent: &configs.PlacementRule{Name: "test", Create: true}},
+ &dao.RuleDAO{Name: "provided", Parameters:
map[string]string{"create": "true"}, ParentRule: &dao.RuleDAO{Name: "test",
Parameters: map[string]string{"create": "true"}}},
+ },
+ {
+ "filter",
+ configs.PlacementRule{Name: "provided", Create: false,
Filter: configs.Filter{Type: filterDeny, Users: []string{"john*"}}},
+ &dao.RuleDAO{Name: "provided", Parameters:
map[string]string{"create": "false"}, Filter: &dao.FilterDAO{Type: filterDeny,
UserExp: "john*"}},
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ ur, err := newRule(tt.conf)
+ assert.NilError(t, err, "setting up the rule failed")
+ ruleDAO := ur.ruleDAO()
+ assert.DeepEqual(t, tt.want, ruleDAO)
+ })
+ }
+}
diff --git a/pkg/scheduler/placement/recovery_rule.go
b/pkg/scheduler/placement/recovery_rule.go
index 4e2101b8..f1f6d725 100644
--- a/pkg/scheduler/placement/recovery_rule.go
+++ b/pkg/scheduler/placement/recovery_rule.go
@@ -26,6 +26,7 @@ import (
"github.com/apache/yunikorn-core/pkg/log"
"github.com/apache/yunikorn-core/pkg/scheduler/objects"
"github.com/apache/yunikorn-core/pkg/scheduler/placement/types"
+ "github.com/apache/yunikorn-core/pkg/webservice/dao"
)
// A rule to place an application into the recovery queue if no other rules
matched and application submission is forced.
@@ -39,6 +40,15 @@ func (rr *recoveryRule) getName() string {
return types.Recovery
}
+func (rr *recoveryRule) ruleDAO() *dao.RuleDAO {
+ return &dao.RuleDAO{
+ Name: rr.getName(),
+ Parameters: map[string]string{
+ "queue": common.RecoveryQueueFull,
+ },
+ }
+}
+
func (rr *recoveryRule) initialise(_ configs.PlacementRule) error {
// no configuration needed for the recovery rule
return nil
diff --git a/pkg/scheduler/placement/recovery_rule_test.go
b/pkg/scheduler/placement/recovery_rule_test.go
index 75f005da..61c0d443 100644
--- a/pkg/scheduler/placement/recovery_rule_test.go
+++ b/pkg/scheduler/placement/recovery_rule_test.go
@@ -26,6 +26,7 @@ import (
"github.com/apache/yunikorn-core/pkg/common"
"github.com/apache/yunikorn-core/pkg/common/configs"
"github.com/apache/yunikorn-core/pkg/common/security"
+ "github.com/apache/yunikorn-core/pkg/webservice/dao"
siCommon "github.com/apache/yunikorn-scheduler-interface/lib/go/common"
)
@@ -75,3 +76,11 @@ partitions:
t.Errorf("recovery rule did not place forced application into
recovery queue, resolved queue '%s', err %v ", queue, err)
}
}
+
+func Test_recoveryRule_ruleDAO(t *testing.T) {
+ // nothing should be set as everything is ignored by the rule
+ rrDAO := &dao.RuleDAO{Name: "recovery", Parameters:
map[string]string{"queue": common.RecoveryQueueFull}}
+ rr := &recoveryRule{}
+ ruleDAO := rr.ruleDAO()
+ assert.DeepEqual(t, rrDAO, ruleDAO)
+}
diff --git a/pkg/scheduler/placement/rule.go b/pkg/scheduler/placement/rule.go
index 1816a52d..cdc542e1 100644
--- a/pkg/scheduler/placement/rule.go
+++ b/pkg/scheduler/placement/rule.go
@@ -28,6 +28,7 @@ import (
"github.com/apache/yunikorn-core/pkg/log"
"github.com/apache/yunikorn-core/pkg/scheduler/objects"
"github.com/apache/yunikorn-core/pkg/scheduler/placement/types"
+ "github.com/apache/yunikorn-core/pkg/webservice/dao"
)
// Interface that all placement rules need to implement.
@@ -48,6 +49,10 @@ type rule interface {
// Return the parent rule.
// This method is implemented in the basicRule which each rule must be
based on.
getParent() rule
+
+ // Returns the rule in a form that can be exposed via the REST api
+ // This method is implemented in the basicRule which each rule must be
based on.
+ ruleDAO() *dao.RuleDAO
}
// Basic structure that every placement rule uses.
@@ -61,13 +66,13 @@ type basicRule struct {
filter Filter
}
-// Get the parent rule used in testing only.
+// getParent gets the parent rule used in testing only.
// Should not be implemented in rules.
func (r *basicRule) getParent() rule {
return r.parent
}
-// Return the name if not overwritten by the rule.
+// getName returns the name if not overwritten by the rule.
// Marked as nolint as rules should override this.
//
//nolint:unused
@@ -75,41 +80,56 @@ func (r *basicRule) getName() string {
return "unnamed rule"
}
-// Create a new rule based on the getName of the rule requested. The rule is
initialised with the configuration and can
-// be used directly.
+// ruleDAO returns the RuleDAO object if not overwritten by the rule.
+// Marked as nolint as rules should override this.
+//
+//nolint:unused
+func (r *basicRule) ruleDAO() *dao.RuleDAO {
+ return &dao.RuleDAO{
+ Name: r.getName(),
+ }
+}
+
+// newRule creates a new rule based on the getName of the rule requested. The
rule is initialised with the configuration
+// and can be used directly.
+// Note that the recoveryRule should not be added to the list as it is an
internal rule only that should not be part of
+// the configuration.
func newRule(conf configs.PlacementRule) (rule, error) {
// create the rule from the config
- var newRule rule
+ var r rule
var err error
// create the new rule fail if the name is unknown
switch normalise(conf.Name) {
// rule that uses the user's name as the queue
case types.User:
- newRule = &userRule{}
+ r = &userRule{}
// rule that uses a fixed queue name
case types.Fixed:
- newRule = &fixedRule{}
+ r = &fixedRule{}
// rule that uses the queue provided on submit
case types.Provided:
- newRule = &providedRule{}
+ r = &providedRule{}
// rule that uses a tag from the application (like namespace)
case types.Tag:
- newRule = &tagRule{}
+ r = &tagRule{}
+ // recovery rule must not be specified in the config
+ case types.Recovery:
+ return nil, fmt.Errorf("recovery rule cannot be part of the
config, failing placement rule config")
// test rule not to be used outside of testing code
case types.Test:
- newRule = &testRule{}
+ r = &testRule{}
default:
return nil, fmt.Errorf("unknown rule name specified %s, failing
placement rule config", conf.Name)
}
// initialise the rule: do not expect the rule to log errors
- err = newRule.initialise(conf)
+ err = r.initialise(conf)
if err != nil {
log.Log(log.Config).Error("Rule init failed", zap.Error(err))
return nil, err
}
log.Log(log.Config).Debug("New rule created", zap.Any("ruleConf", conf))
- return newRule, nil
+ return r, nil
}
// Normalise the rule name from the config.
diff --git a/pkg/scheduler/placement/rule_test.go
b/pkg/scheduler/placement/rule_test.go
index f4e741c8..3df65c3d 100644
--- a/pkg/scheduler/placement/rule_test.go
+++ b/pkg/scheduler/placement/rule_test.go
@@ -52,6 +52,15 @@ func TestNewRule(t *testing.T) {
if err != nil || nr == nil {
t.Errorf("new normalised newRule build failed which should not,
newRule 'nil' , err: %v, ", err)
}
+
+ // test deny recovery rule name
+ conf = configs.PlacementRule{
+ Name: "recovery",
+ }
+ nr, err = newRule(conf)
+ if err == nil || nr != nil {
+ t.Errorf("new newRule did not fail with recovery rule name, err
'nil' , newRule: %v, ", nr)
+ }
}
// Test for a basic test rule.
diff --git a/pkg/scheduler/placement/tag_rule.go
b/pkg/scheduler/placement/tag_rule.go
index 95f38ea3..6bc54b42 100644
--- a/pkg/scheduler/placement/tag_rule.go
+++ b/pkg/scheduler/placement/tag_rule.go
@@ -20,6 +20,7 @@ package placement
import (
"fmt"
+ "strconv"
"strings"
"go.uber.org/zap"
@@ -28,6 +29,7 @@ import (
"github.com/apache/yunikorn-core/pkg/log"
"github.com/apache/yunikorn-core/pkg/scheduler/objects"
"github.com/apache/yunikorn-core/pkg/scheduler/placement/types"
+ "github.com/apache/yunikorn-core/pkg/webservice/dao"
)
// A rule to place an application based on the a tag on the application.
@@ -43,6 +45,22 @@ func (tr *tagRule) getName() string {
return types.Tag
}
+func (tr *tagRule) ruleDAO() *dao.RuleDAO {
+ var pDAO *dao.RuleDAO
+ if tr.parent != nil {
+ pDAO = tr.parent.ruleDAO()
+ }
+ return &dao.RuleDAO{
+ Name: tr.getName(),
+ Parameters: map[string]string{
+ "tagName": tr.tagName,
+ "create": strconv.FormatBool(tr.create),
+ },
+ ParentRule: pDAO,
+ Filter: tr.filter.filterDAO(),
+ }
+}
+
func (tr *tagRule) initialise(conf configs.PlacementRule) error {
tr.tagName = normalise(conf.Value)
if tr.tagName == "" {
diff --git a/pkg/scheduler/placement/tag_rule_test.go
b/pkg/scheduler/placement/tag_rule_test.go
index 6eb8e79e..527ea736 100644
--- a/pkg/scheduler/placement/tag_rule_test.go
+++ b/pkg/scheduler/placement/tag_rule_test.go
@@ -26,6 +26,7 @@ import (
"github.com/apache/yunikorn-core/pkg/common"
"github.com/apache/yunikorn-core/pkg/common/configs"
"github.com/apache/yunikorn-core/pkg/common/security"
+ "github.com/apache/yunikorn-core/pkg/webservice/dao"
)
func TestTagRule(t *testing.T) {
@@ -251,3 +252,40 @@ func TestTagRuleParent(t *testing.T) {
t.Errorf("tag rule placed app in incorrect queue '%s', err %v",
queue, err)
}
}
+
+func Test_tagRule_ruleDAO(t *testing.T) {
+ tests := []struct {
+ name string
+ conf configs.PlacementRule
+ want *dao.RuleDAO
+ }{
+ {
+ "base",
+ configs.PlacementRule{Name: "tag", Value: "namespace"},
+ &dao.RuleDAO{Name: "tag", Parameters:
map[string]string{"tagName": "namespace", "create": "false"}},
+ },
+ {
+ "mixedcase",
+ configs.PlacementRule{Name: "tag", Value:
"MixedCaseTag"},
+ &dao.RuleDAO{Name: "tag", Parameters:
map[string]string{"tagName": "mixedcasetag", "create": "false"}},
+ },
+ {
+ "parent",
+ configs.PlacementRule{Name: "tag", Value: "one",
Create: true, Parent: &configs.PlacementRule{Name: "test", Create: true}},
+ &dao.RuleDAO{Name: "tag", Parameters:
map[string]string{"tagName": "one", "create": "true"}, ParentRule:
&dao.RuleDAO{Name: "test", Parameters: map[string]string{"create": "true"}}},
+ },
+ {
+ "filter",
+ configs.PlacementRule{Name: "tag", Value: "two",
Create: true, Filter: configs.Filter{Type: filterDeny, Groups:
[]string{"group[0-9]"}}},
+ &dao.RuleDAO{Name: "tag", Parameters:
map[string]string{"tagName": "two", "create": "true"}, Filter:
&dao.FilterDAO{Type: filterDeny, GroupExp: "group[0-9]"}},
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ ur, err := newRule(tt.conf)
+ assert.NilError(t, err, "setting up the rule failed")
+ ruleDAO := ur.ruleDAO()
+ assert.DeepEqual(t, tt.want, ruleDAO)
+ })
+ }
+}
diff --git a/pkg/scheduler/placement/testrule.go
b/pkg/scheduler/placement/testrule.go
index fce27c68..f06a145e 100644
--- a/pkg/scheduler/placement/testrule.go
+++ b/pkg/scheduler/placement/testrule.go
@@ -20,10 +20,12 @@ package placement
import (
"fmt"
+ "strconv"
"github.com/apache/yunikorn-core/pkg/common/configs"
"github.com/apache/yunikorn-core/pkg/scheduler/objects"
"github.com/apache/yunikorn-core/pkg/scheduler/placement/types"
+ "github.com/apache/yunikorn-core/pkg/webservice/dao"
)
// A simple test rule to place an application based on a nil application.
@@ -36,6 +38,21 @@ func (tr *testRule) getName() string {
return types.Test
}
+func (tr *testRule) ruleDAO() *dao.RuleDAO {
+ var pDAO *dao.RuleDAO
+ if tr.parent != nil {
+ pDAO = tr.parent.ruleDAO()
+ }
+ return &dao.RuleDAO{
+ Name: types.Test,
+ Parameters: map[string]string{
+ "create": strconv.FormatBool(tr.create),
+ },
+ ParentRule: pDAO,
+ Filter: tr.filter.filterDAO(),
+ }
+}
+
// Simple init for the test rule: allow everything as per a normal rule.
func (tr *testRule) initialise(conf configs.PlacementRule) error {
tr.create = conf.Create
diff --git a/pkg/scheduler/placement/user_rule.go
b/pkg/scheduler/placement/user_rule.go
index 2c195e4e..1e1f914d 100644
--- a/pkg/scheduler/placement/user_rule.go
+++ b/pkg/scheduler/placement/user_rule.go
@@ -20,6 +20,7 @@ package placement
import (
"fmt"
+ "strconv"
"strings"
"go.uber.org/zap"
@@ -28,6 +29,7 @@ import (
"github.com/apache/yunikorn-core/pkg/log"
"github.com/apache/yunikorn-core/pkg/scheduler/objects"
"github.com/apache/yunikorn-core/pkg/scheduler/placement/types"
+ "github.com/apache/yunikorn-core/pkg/webservice/dao"
)
// A rule to place an application based on the user name of the submitting
user.
@@ -39,6 +41,21 @@ func (ur *userRule) getName() string {
return types.User
}
+func (ur *userRule) ruleDAO() *dao.RuleDAO {
+ var pDAO *dao.RuleDAO
+ if ur.parent != nil {
+ pDAO = ur.parent.ruleDAO()
+ }
+ return &dao.RuleDAO{
+ Name: ur.getName(),
+ Parameters: map[string]string{
+ "create": strconv.FormatBool(ur.create),
+ },
+ ParentRule: pDAO,
+ Filter: ur.filter.filterDAO(),
+ }
+}
+
func (ur *userRule) initialise(conf configs.PlacementRule) error {
ur.create = conf.Create
ur.filter = newFilter(conf.Filter)
diff --git a/pkg/scheduler/placement/user_rule_test.go
b/pkg/scheduler/placement/user_rule_test.go
index 9e7126e3..62b099e8 100644
--- a/pkg/scheduler/placement/user_rule_test.go
+++ b/pkg/scheduler/placement/user_rule_test.go
@@ -25,6 +25,7 @@ import (
"github.com/apache/yunikorn-core/pkg/common/configs"
"github.com/apache/yunikorn-core/pkg/common/security"
+ "github.com/apache/yunikorn-core/pkg/webservice/dao"
)
func TestUserRulePlace(t *testing.T) {
@@ -219,3 +220,35 @@ func TestUserRuleParent(t *testing.T) {
t.Errorf("user rule placed app in incorrect queue '%s', err
%v", queue, err)
}
}
+
+func Test_userRule_ruleDAO(t *testing.T) {
+ tests := []struct {
+ name string
+ conf configs.PlacementRule
+ want *dao.RuleDAO
+ }{
+ {
+ "base",
+ configs.PlacementRule{Name: "user"},
+ &dao.RuleDAO{Name: "user", Parameters:
map[string]string{"create": "false"}},
+ },
+ {
+ "parent",
+ configs.PlacementRule{Name: "user", Create: true,
Parent: &configs.PlacementRule{Name: "test", Create: true}},
+ &dao.RuleDAO{Name: "user", Parameters:
map[string]string{"create": "true"}, ParentRule: &dao.RuleDAO{Name: "test",
Parameters: map[string]string{"create": "true"}}},
+ },
+ {
+ "filter",
+ configs.PlacementRule{Name: "user", Create: true,
Filter: configs.Filter{Type: filterDeny, Users: []string{"user"}}},
+ &dao.RuleDAO{Name: "user", Parameters:
map[string]string{"create": "true"}, Filter: &dao.FilterDAO{Type: filterDeny,
UserList: []string{"user"}}},
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ ur, err := newRule(tt.conf)
+ assert.NilError(t, err, "setting up the rule failed")
+ ruleDAO := ur.ruleDAO()
+ assert.DeepEqual(t, tt.want, ruleDAO)
+ })
+ }
+}
diff --git a/pkg/webservice/dao/rule_info.go b/pkg/webservice/dao/rule_info.go
new file mode 100644
index 00000000..9e852864
--- /dev/null
+++ b/pkg/webservice/dao/rule_info.go
@@ -0,0 +1,39 @@
+/*
+ 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 dao
+
+type RuleDAOInfo struct {
+ Partition string `json:"partition"` // no omitempty, partition name
should not be empty
+ Rules []*RuleDAO `json:"rules,omitempty"`
+}
+
+type FilterDAO struct {
+ Type string `json:"type"` // no omitempty, type must exist
+ UserList []string `json:"userList,omitempty"`
+ GroupList []string `json:"groupList,omitempty"`
+ UserExp string `json:"userExp,omitempty"`
+ GroupExp string `json:"groupExp,omitempty"`
+}
+
+type RuleDAO struct {
+ Name string `json:"name"` // no omitempty, name must
exist
+ Parameters map[string]string `json:"parameters,omitempty"`
+ Filter *FilterDAO `json:"filter,omitempty"`
+ ParentRule *RuleDAO `json:"parentRule,omitempty"`
+}
diff --git a/pkg/webservice/handlers.go b/pkg/webservice/handlers.go
index 5ccc31a5..5e37c80f 100644
--- a/pkg/webservice/handlers.go
+++ b/pkg/webservice/handlers.go
@@ -836,6 +836,25 @@ func getApplication(w http.ResponseWriter, r
*http.Request) {
}
}
+func getPartitionRules(w http.ResponseWriter, r *http.Request) {
+ writeHeaders(w)
+ vars := httprouter.ParamsFromContext(r.Context())
+ if vars == nil {
+ buildJSONErrorResponse(w, MissingParamsName,
http.StatusBadRequest)
+ return
+ }
+ partition := vars.ByName("partition")
+ partitionContext :=
schedulerContext.GetPartitionWithoutClusterID(partition)
+ if partitionContext == nil {
+ buildJSONErrorResponse(w, PartitionDoesNotExists,
http.StatusNotFound)
+ return
+ }
+ rulesDao := partitionContext.GetPlacementRules()
+ if err := json.NewEncoder(w).Encode(rulesDao); err != nil {
+ buildJSONErrorResponse(w, err.Error(),
http.StatusInternalServerError)
+ }
+}
+
func getPartitionInfoDAO(lists map[string]*scheduler.PartitionContext)
[]*dao.PartitionInfo {
result := make([]*dao.PartitionInfo, 0, len(lists))
@@ -942,6 +961,19 @@ func getApplicationsDAO(lists
map[string]*scheduler.PartitionContext) []*dao.App
return result
}
+func getPlacementRulesDAO(lists map[string]*scheduler.PartitionContext)
[]*dao.RuleDAOInfo {
+ result := make([]*dao.RuleDAOInfo, 0, len(lists))
+
+ for _, partition := range lists {
+ result = append(result, &dao.RuleDAOInfo{
+ Partition:
common.GetPartitionNameWithoutClusterID(partition.Name),
+ Rules: partition.GetPlacementRules(),
+ })
+ }
+
+ return result
+}
+
func getPartitionQueuesDAO(lists map[string]*scheduler.PartitionContext)
[]dao.PartitionQueueDAOInfo {
result := make([]dao.PartitionQueueDAOInfo, 0, len(lists))
diff --git a/pkg/webservice/handlers_test.go b/pkg/webservice/handlers_test.go
index 1b16ba3a..8bb60695 100644
--- a/pkg/webservice/handlers_test.go
+++ b/pkg/webservice/handlers_test.go
@@ -43,6 +43,7 @@ import (
"github.com/apache/yunikorn-core/pkg/metrics/history"
"github.com/apache/yunikorn-core/pkg/scheduler"
"github.com/apache/yunikorn-core/pkg/scheduler/objects"
+ "github.com/apache/yunikorn-core/pkg/scheduler/placement/types"
"github.com/apache/yunikorn-core/pkg/scheduler/policies"
"github.com/apache/yunikorn-core/pkg/scheduler/ugm"
"github.com/apache/yunikorn-core/pkg/webservice/dao"
@@ -50,9 +51,12 @@ import (
"github.com/apache/yunikorn-scheduler-interface/lib/go/si"
)
-const unmarshalError = "Failed to unmarshal error response from response body"
-const statusCodeError = "Incorrect Status code"
-const jsonMessageError = "JSON error message is incorrect"
+const (
+ unmarshalError = "Failed to unmarshal error response from response
body"
+ statusCodeError = "Incorrect Status code"
+ jsonMessageError = "JSON error message is incorrect"
+ httpRequestError = "HTTP request creation failed"
+)
const partitionNameWithoutClusterID = "default"
const normalizedPartitionName = "[rm-123]default"
@@ -67,6 +71,7 @@ partitions:
first: "some value with spaces"
second: somethingElse
`
+
const updatedConf = `
partitions:
- name: default
@@ -127,6 +132,7 @@ partitions:
name: default
submitacl: "*"
`
+
const configTwoLevelQueues = `
partitions:
-
@@ -247,6 +253,27 @@ partitions:
cpu: "200"
`
+const placementRuleConfig = `
+partitions:
+ - name: default
+ placementrules:
+ - name: user
+ - name: tag
+ value: namespace
+ create: true
+ parent:
+ name: fixed
+ value: root.namespaces
+ - name: fixed
+ value: root.default
+ queues:
+ - name: root
+ parent: true
+ submitacl: '*'
+ queues:
+ - name: default
+`
+
const rmID = "rm-123"
const policyGroup = "default-policy-group"
const queueName = "root.default"
@@ -1852,7 +1879,8 @@ func TestFullStateDumpPath(t *testing.T) {
var aggregated AggregatedStateInfo
err = json.Unmarshal(resp.outputBytes, &aggregated)
assert.NilError(t, err)
- verifyStateDumpJSON(t, &aggregated)
+ // default config has only one partition
+ verifyStateDumpJSON(t, &aggregated, 1)
}
func TestSpecificUserAndGroupResourceUsage(t *testing.T) {
@@ -2523,15 +2551,18 @@ func clearUserManager() {
userManager.ClearGroupTrackers()
}
-func verifyStateDumpJSON(t *testing.T, aggregated *AggregatedStateInfo) {
+func verifyStateDumpJSON(t *testing.T, aggregated *AggregatedStateInfo,
partitionCount int) {
assert.Check(t, aggregated.Timestamp != 0)
- assert.Check(t, len(aggregated.Partitions) > 0)
+ assert.Check(t, len(aggregated.Partitions) == partitionCount,
"incorrect partition count")
assert.Check(t, len(aggregated.Nodes) > 0)
assert.Check(t, len(aggregated.ClusterInfo) > 0)
- assert.Check(t, len(aggregated.Queues) > 0)
+ assert.Check(t, len(aggregated.Queues) == 1, "should only have root
queue")
assert.Check(t, len(aggregated.LogLevel) > 0)
- assert.Check(t, len(aggregated.Config.SchedulerConfig.Partitions) > 0)
+ assert.Check(t, len(aggregated.Config.SchedulerConfig.Partitions) ==
partitionCount, "incorrect partition count")
assert.Check(t, len(aggregated.Config.Extra) > 0)
+ assert.Check(t, aggregated.RMDiagnostics["empty"] != nil, "expected no
RM registered for diagnostics")
+ assert.Check(t, len(aggregated.PlacementRules) == partitionCount,
"incorrect partition count")
+ assert.Check(t, len(aggregated.PlacementRules[0].Rules) == 2,
"incorrect rule count")
}
func TestCheckHealthStatusNotFound(t *testing.T) {
@@ -2598,6 +2629,58 @@ func runHealthCheckTest(t *testing.T, expected
*dao.SchedulerHealthDAOInfo) {
}
}
+func TestGetPartitionRuleHandler(t *testing.T) {
+ setup(t, configDefault, 1)
+
+ NewWebApp(schedulerContext, nil)
+
+ // test partition not exists
+ req, err := http.NewRequest("GET",
"/ws/v1/partition/default/placementrules", strings.NewReader(""))
+ assert.NilError(t, err, httpRequestError)
+ req = req.WithContext(context.WithValue(req.Context(),
httprouter.ParamsKey, httprouter.Params{httprouter.Param{Key: "partition",
Value: "notexists"}}))
+ resp := &MockResponseWriter{}
+ getPartitionRules(resp, req)
+ assertPartitionNotExists(t, resp)
+
+ // test params name missing
+ req, err = http.NewRequest("GET",
"/ws/v1/partition/default/placementrules", strings.NewReader(""))
+ assert.NilError(t, err, httpRequestError)
+ resp = &MockResponseWriter{}
+ getPartitionRules(resp, req)
+ assertParamsMissing(t, resp)
+
+ // default config without rules defined
+ req, err = http.NewRequest("GET",
"/ws/v1/partition/default/placementrules", strings.NewReader(""))
+ assert.NilError(t, err, httpRequestError)
+ req = req.WithContext(context.WithValue(req.Context(),
httprouter.ParamsKey, httprouter.Params{httprouter.Param{Key: "partition",
Value: partitionNameWithoutClusterID}}))
+ var partitionRules []*dao.RuleDAO
+ resp = &MockResponseWriter{}
+ getPartitionRules(resp, req)
+ err = json.Unmarshal(resp.outputBytes, &partitionRules)
+ assert.NilError(t, err, unmarshalError)
+ // assert content: default is provided and recovery
+ assert.Equal(t, len(partitionRules), 2)
+ assert.Equal(t, partitionRules[0].Name, types.Provided)
+ assert.Equal(t, partitionRules[1].Name, types.Recovery)
+
+ // change the config: 3 rules, expect recovery also
+ err = schedulerContext.UpdateRMSchedulerConfig(rmID,
[]byte(placementRuleConfig))
+ assert.NilError(t, err, "Error when updating clusterInfo from config")
+ req, err = http.NewRequest("GET",
"/ws/v1/partition/default/placementrules", strings.NewReader(""))
+ assert.NilError(t, err, httpRequestError)
+ req = req.WithContext(context.WithValue(req.Context(),
httprouter.ParamsKey, httprouter.Params{httprouter.Param{Key: "partition",
Value: partitionNameWithoutClusterID}}))
+ resp = &MockResponseWriter{}
+ getPartitionRules(resp, req)
+ err = json.Unmarshal(resp.outputBytes, &partitionRules)
+ assert.NilError(t, err, unmarshalError)
+ assert.Equal(t, len(partitionRules), 4)
+ assert.Equal(t, partitionRules[0].Name, types.User)
+ assert.Equal(t, partitionRules[1].Name, types.Tag)
+ assert.Equal(t, partitionRules[1].ParentRule.Name, types.Fixed)
+ assert.Equal(t, partitionRules[2].Name, types.Fixed)
+ assert.Equal(t, partitionRules[3].Name, types.Recovery)
+}
+
type ResponseRecorderWithDeadline struct {
*httptest.ResponseRecorder
setWriteFails bool
diff --git a/pkg/webservice/routes.go b/pkg/webservice/routes.go
index f3653b25..3eeef978 100644
--- a/pkg/webservice/routes.go
+++ b/pkg/webservice/routes.go
@@ -92,6 +92,12 @@ var webRoutes = routes{
"/ws/v1/partitions",
getPartitions,
},
+ route{
+ "Scheduler",
+ "GET",
+ "/ws/v1/partition/:partition/placementrules",
+ getPartitionRules,
+ },
route{
"Scheduler",
"GET",
diff --git a/pkg/webservice/state_dump.go b/pkg/webservice/state_dump.go
index bb3f98f3..24f769e6 100644
--- a/pkg/webservice/state_dump.go
+++ b/pkg/webservice/state_dump.go
@@ -49,6 +49,7 @@ type AggregatedStateInfo struct {
RMDiagnostics map[string]interface{}
`json:"rmDiagnostics,omitempty"`
LogLevel string
`json:"logLevel,omitempty"`
Config *dao.ConfigDAOInfo
`json:"config,omitempty"`
+ PlacementRules []*dao.RuleDAOInfo
`json:"placementRules,omitempty"`
EventStreams []events.EventStreamData
`json:"eventStreams,omitempty"`
}
@@ -79,6 +80,7 @@ func doStateDump(w io.Writer) error {
RMDiagnostics: getResourceManagerDiagnostics(),
LogLevel: zapConfig.Level.Level().String(),
Config: getClusterConfigDAO(),
+ PlacementRules: getPlacementRulesDAO(partitionContext),
EventStreams: events.GetEventSystem().GetEventStreams(),
}
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@yunikorn.apache.org
For additional commands, e-mail: issues-h...@yunikorn.apache.org
