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 dee60d15 [YUNIKORN-2012] group quota is not maintained (#668)
dee60d15 is described below
commit dee60d15ee776d3c9890cacc5538ffa3cbf236fa
Author: Manikandan R <[email protected]>
AuthorDate: Fri Nov 3 12:11:05 2023 +1100
[YUNIKORN-2012] group quota is not maintained (#668)
In certain cases it looks like group quota is not considered during
scheduling. This is caused by incorrectly maintaining the quotas in the
user group manager. The quota was reset while it should not causing the
quota to never be checked.
Further test code added in YUNIKORN-2024i via #669
Closes: #668
Signed-off-by: Wilfred Spiegelenburg <[email protected]>
---
pkg/scheduler/partition_test.go | 1 +
pkg/scheduler/ugm/group_tracker.go | 3 +-
pkg/scheduler/ugm/manager.go | 254 +++++++++++++++++++++++--------------
pkg/scheduler/ugm/manager_test.go | 119 ++++++++++++-----
pkg/scheduler/ugm/queue_tracker.go | 33 ++++-
pkg/scheduler/ugm/user_tracker.go | 3 +-
6 files changed, 275 insertions(+), 138 deletions(-)
diff --git a/pkg/scheduler/partition_test.go b/pkg/scheduler/partition_test.go
index 186fa315..aa867f0f 100644
--- a/pkg/scheduler/partition_test.go
+++ b/pkg/scheduler/partition_test.go
@@ -45,6 +45,7 @@ func setupUGM() {
userManager := ugm.GetUserManager()
userManager.ClearUserTrackers()
userManager.ClearGroupTrackers()
+ userManager.ClearConfigLimits()
}
func setupNode(t *testing.T, nodeID string, partition *PartitionContext,
nodeRes *resources.Resource) *objects.Node {
diff --git a/pkg/scheduler/ugm/group_tracker.go
b/pkg/scheduler/ugm/group_tracker.go
index f6f316f0..62304216 100644
--- a/pkg/scheduler/ugm/group_tracker.go
+++ b/pkg/scheduler/ugm/group_tracker.go
@@ -118,11 +118,10 @@ func (gt *GroupTracker) UnlinkQT(queuePath string) bool {
return gt.queueTracker.UnlinkQT(strings.Split(queuePath, configs.DOT))
}
-// canBeRemoved Does "root" queue has any child queue trackers? Is there any
running applications in "root" qt?
func (gt *GroupTracker) canBeRemoved() bool {
gt.RLock()
defer gt.RUnlock()
- return len(gt.queueTracker.childQueueTrackers) == 0 &&
len(gt.queueTracker.runningApplications) == 0
+ return gt.queueTracker.canBeRemoved()
}
func (gt *GroupTracker) getName() string {
diff --git a/pkg/scheduler/ugm/manager.go b/pkg/scheduler/ugm/manager.go
index fd40aafe..0f431bc7 100644
--- a/pkg/scheduler/ugm/manager.go
+++ b/pkg/scheduler/ugm/manager.go
@@ -39,9 +39,11 @@ var m *Manager
type Manager struct {
userTrackers map[string]*UserTracker
groupTrackers map[string]*GroupTracker
- userWildCardLimitsConfig map[string]*LimitConfig // Hold limits
settings of user '*'
- groupWildCardLimitsConfig map[string]*LimitConfig // Hold limits
settings of group '*'
- configuredGroups map[string][]string // Hold groups for
all configured queue paths.
+ userWildCardLimitsConfig map[string]*LimitConfig // Hold
limits settings of user '*'
+ groupWildCardLimitsConfig map[string]*LimitConfig // Hold
limits settings of group '*'
+ configuredGroups map[string][]string // Hold
groups for all configured queue paths.
+ userLimits map[string]map[string]*LimitConfig // Holds
queue path * user limit config
+ groupLimits map[string]map[string]*LimitConfig // Holds
queue path * group limit config
sync.RWMutex
}
@@ -312,19 +314,29 @@ func (m *Manager) isGroupRemovable(gt *GroupTracker) bool
{
}
func (m *Manager) UpdateConfig(config configs.QueueConfig, queuePath string)
error {
- m.Lock()
- defer m.Unlock()
+ userWildCardLimitsConfig := make(map[string]*LimitConfig)
+ groupWildCardLimitsConfig := make(map[string]*LimitConfig)
+ configuredGroups := make(map[string][]string)
- m.userWildCardLimitsConfig = make(map[string]*LimitConfig)
- m.groupWildCardLimitsConfig = make(map[string]*LimitConfig)
- m.configuredGroups = make(map[string][]string)
- return m.internalProcessConfig(config, queuePath)
+ userLimits := make(map[string]map[string]*LimitConfig) // Holds queue
path * user limit config
+ groupLimits := make(map[string]map[string]*LimitConfig) // Holds queue
path * group limit config
+
+ // as and when parse new configs, store them in temporary maps
+ if err := m.internalProcessConfig(config, queuePath, userLimits,
groupLimits, userWildCardLimitsConfig, groupWildCardLimitsConfig,
configuredGroups); err != nil {
+ return err
+ }
+
+ // compare existing config with new configs stored in above temporary
maps
+ m.clearEarlierSetLimits(userLimits, groupLimits)
+
+ // switch over - replace the existing config with new configs
+ m.replaceLimitConfigs(userLimits, groupLimits,
userWildCardLimitsConfig, groupWildCardLimitsConfig, configuredGroups)
+
+ return nil
}
-func (m *Manager) internalProcessConfig(cur configs.QueueConfig, queuePath
string) error {
- // Holds user and group for which limits have been configured with
specific queue path
- userLimits := make(map[string]bool)
- groupLimits := make(map[string]bool)
+func (m *Manager) internalProcessConfig(cur configs.QueueConfig, queuePath
string, newUserLimits map[string]map[string]*LimitConfig, newGroupLimits
map[string]map[string]*LimitConfig,
+ newUserWildCardLimitsConfig map[string]*LimitConfig,
newGroupWildCardLimitsConfig map[string]*LimitConfig, newConfiguredGroups
map[string][]string) error {
// Traverse limits of specific queue path
for _, limit := range cur.Limits {
var maxResource *resources.Resource
@@ -348,12 +360,16 @@ func (m *Manager) internalProcessConfig(cur
configs.QueueConfig, queuePath strin
zap.Uint64("max application",
limit.MaxApplications),
zap.Any("max resources", limit.MaxResources))
if user == common.Wildcard {
- m.userWildCardLimitsConfig[queuePath] =
limitConfig
+ newUserWildCardLimitsConfig[queuePath] =
limitConfig
continue
}
- if err := m.processUserConfig(user, limitConfig,
queuePath, userLimits); err != nil {
+ if err := m.setUserLimits(user, limitConfig,
queuePath); err != nil {
return err
}
+ if _, ok := newUserLimits[queuePath]; !ok {
+ newUserLimits[queuePath] =
make(map[string]*LimitConfig)
+ }
+ newUserLimits[queuePath][user] = limitConfig
}
for _, group := range limit.Groups {
if group == common.Empty {
@@ -365,24 +381,24 @@ func (m *Manager) internalProcessConfig(cur
configs.QueueConfig, queuePath strin
zap.String("queue path", queuePath),
zap.Uint64("max application",
limit.MaxApplications),
zap.Any("max resources", limit.MaxResources))
- if err := m.processGroupConfig(group, limitConfig,
queuePath, groupLimits); err != nil {
+ if err := m.setGroupLimits(group, limitConfig,
queuePath); err != nil {
return err
}
+ if _, ok := newGroupLimits[queuePath]; !ok {
+ newGroupLimits[queuePath] =
make(map[string]*LimitConfig)
+ }
+ newGroupLimits[queuePath][group] = limitConfig
if group == common.Wildcard {
- m.groupWildCardLimitsConfig[queuePath] =
limitConfig
+ newGroupWildCardLimitsConfig[queuePath] =
limitConfig
} else {
- m.configuredGroups[queuePath] =
append(m.configuredGroups[queuePath], group)
+ newConfiguredGroups[queuePath] =
append(newConfiguredGroups[queuePath], group)
}
}
}
- if err := m.clearEarlierSetLimits(userLimits, groupLimits, queuePath);
err != nil {
- return err
- }
-
if len(cur.Queues) > 0 {
for _, child := range cur.Queues {
childQueuePath := queuePath + configs.DOT + child.Name
- if err := m.internalProcessConfig(child,
childQueuePath); err != nil {
+ if err := m.internalProcessConfig(child,
childQueuePath, newUserLimits, newGroupLimits, newUserWildCardLimitsConfig,
newGroupWildCardLimitsConfig, newConfiguredGroups); err != nil {
return err
}
}
@@ -390,102 +406,132 @@ func (m *Manager) internalProcessConfig(cur
configs.QueueConfig, queuePath strin
return nil
}
-func (m *Manager) processUserConfig(user string, limitConfig *LimitConfig,
queuePath string, userLimits map[string]bool) error {
- if err := m.setUserLimits(user, limitConfig, queuePath); err != nil {
- return err
- }
- userLimits[user] = true
- return nil
-}
+// clearEarlierSetLimits Clear already configured limits of users and groups
for which limits have been configured before but not now
+func (m *Manager) clearEarlierSetLimits(newUserLimits
map[string]map[string]*LimitConfig, newGroupLimits
map[string]map[string]*LimitConfig) {
+ // Clear already configured limits of group for which limits have been
configured before but not now
+ m.clearEarlierSetGroupLimits(newGroupLimits)
-func (m *Manager) processGroupConfig(group string, limitConfig *LimitConfig,
queuePath string, groupLimits map[string]bool) error {
- if err := m.setGroupLimits(group, limitConfig, queuePath); err != nil {
- return err
- }
- groupLimits[group] = true
- return nil
+ // Clear already configured limits of user for which limits have been
configured before but not now
+ m.clearEarlierSetUserLimits(newUserLimits)
}
-// clearEarlierSetLimits Clear already configured limits of users and groups
for which limits have been configured before but not now
-func (m *Manager) clearEarlierSetLimits(userLimits map[string]bool,
groupLimits map[string]bool, queuePath string) error {
- // Clear already configured limits of group for which limits have been
configured before but not now
- for _, gt := range m.groupTrackers {
- appUsersMap := m.clearEarlierSetGroupLimits(gt, queuePath,
groupLimits)
- if len(appUsersMap) > 0 {
- for app, user := range appUsersMap {
- ut := m.userTrackers[user]
- ut.setGroupForApp(app, nil)
+// clearEarlierSetUserLimits Traverse new user config and decide whether
earlier usage needs to be cleared or not
+// by comparing with the existing config. Reset earlier usage only config set
earlier but not now
+func (m *Manager) clearEarlierSetUserLimits(newUserLimits
map[string]map[string]*LimitConfig) {
+ m.RLock()
+ defer m.RUnlock()
+ for queuePath, limitConfig := range m.userLimits {
+ // Is queue path exists?
+ if newUserLimit, ok := newUserLimits[queuePath]; !ok {
+ for u := range limitConfig {
+ if ut, utExists := m.userTrackers[u]; utExists {
+ m.resetUserEarlierUsage(ut, queuePath)
+ }
+ }
+ } else {
+ // Queue path exists. Is user exists?
+ for u := range limitConfig {
+ if _, ulExists := newUserLimit[u]; !ulExists {
+ if ut, utExists := m.userTrackers[u];
utExists {
+ m.resetUserEarlierUsage(ut,
queuePath)
+ }
+ }
}
}
}
-
- // Clear already configured limits of user for which limits have been
configured before but not now
- for _, ut := range m.userTrackers {
- m.clearEarlierSetUserLimits(ut, queuePath, userLimits)
- }
- return nil
}
-func (m *Manager) clearEarlierSetUserLimits(ut *UserTracker, queuePath string,
userLimits map[string]bool) {
+// resetUserEarlierUsage Clear or reset earlier usage only when user already
tracked for the queue path.
+// Reset the max apps and max resources to default, unlink the end leaf queue
of queue path from its immediate parent and
+// eventually remove user tracker object itself from ugm if it can be removed.
+func (m *Manager) resetUserEarlierUsage(ut *UserTracker, queuePath string) {
// Is this user already tracked for the queue path?
if ut.IsQueuePathTrackedCompletely(queuePath) {
- u := ut.userName
- // Is there any limit config set for user in the current
configuration? If not, then clear those old limit settings
- if _, ok := userLimits[u]; !ok {
- log.Log(log.SchedUGM).Debug("Need to clear earlier set
configs for user",
- zap.String("user", u),
- zap.String("queue path", queuePath))
- // Is there any running applications in end queue of
this queue path? If not, then remove the linkage between end queue and its
immediate parent
- if ut.IsUnlinkRequired(queuePath) {
- ut.UnlinkQT(queuePath)
- } else {
- ut.setLimits(queuePath,
resources.NewResource(), 0)
- log.Log(log.SchedUGM).Debug("Cleared earlier
set limit configs for user",
- zap.String("user", u),
- zap.String("queue path", queuePath))
+ log.Log(log.SchedUGM).Debug("Need to clear earlier set configs
for user",
+ zap.String("user", ut.userName),
+ zap.String("queue path", queuePath))
+ ut.setLimits(queuePath, resources.NewResource(), 0)
+ // Is there any running applications in end queue of this queue
path? If not, then remove the linkage between end queue and its immediate parent
+ if ut.IsUnlinkRequired(queuePath) {
+ ut.UnlinkQT(queuePath)
+ }
+ log.Log(log.SchedUGM).Debug("Cleared earlier set limit configs
for user",
+ zap.String("user", ut.userName),
+ zap.String("queue path", queuePath))
+ if ut.canBeRemoved() {
+ delete(m.userTrackers, ut.userName)
+ }
+ }
+}
+
+// clearEarlierSetGroupLimits Traverse new group config and decide whether
earlier usage needs to be cleared or not
+// by comparing with the existing config. Reset earlier usage only config set
earlier but not now
+func (m *Manager) clearEarlierSetGroupLimits(newGroupLimits
map[string]map[string]*LimitConfig) {
+ m.RLock()
+ defer m.RUnlock()
+ for queuePath, limitConfig := range m.groupLimits {
+ // Is queue path exists?
+ if newGroupLimit, ok := newGroupLimits[queuePath]; !ok {
+ for g := range limitConfig {
+ if gt, gtExists := m.groupTrackers[g]; gtExists
{
+ m.resetGroupEarlierUsage(gt, queuePath)
+ }
}
- // Does "root" queue has any child queue trackers? At
some point during this whole traversal, root might
- // not have any child queue trackers. When the
situation comes, remove the linkage between the user and
- // its root queue tracker
- if ut.canBeRemoved() {
- delete(m.userTrackers, ut.userName)
+ } else {
+ // Queue path exists. Is group exists?
+ for g := range limitConfig {
+ if _, glExists := newGroupLimit[g]; !glExists {
+ if gt, gtExists := m.groupTrackers[g];
gtExists {
+ m.resetGroupEarlierUsage(gt,
queuePath)
+ }
+ }
}
}
}
}
-func (m *Manager) clearEarlierSetGroupLimits(gt *GroupTracker, queuePath
string, groupLimits map[string]bool) map[string]string {
- appUsersMap := make(map[string]string)
- // Is this group already tracked for the queue path?
+// resetGroupEarlierUsage Clear or reset earlier usage only when group already
tracked for the queue path.
+// Decrease the group usage and collect the list of applications for which
user app group linkage needs to be broken.
+// Reset the max apps and max resources to default, unlink the end leaf queue
of queue path from its immediate parent and
+// eventually remove group tracker object itself from ugm if it can be removed.
+func (m *Manager) resetGroupEarlierUsage(gt *GroupTracker, queuePath string) {
if gt.IsQueuePathTrackedCompletely(queuePath) {
- g := gt.groupName
- // Is there any limit config set for group in the current
configuration? If not, then clear those old limit settings
- if ok := groupLimits[g]; !ok {
- log.Log(log.SchedUGM).Debug("Need to clear earlier set
configs for group",
- zap.String("group", g),
- zap.String("queue path", queuePath))
- appUsersMap =
gt.decreaseAllTrackedResourceUsage(queuePath)
- // Is there any running applications in end queue of
this queue path? If not, then remove the linkage between end queue and its
immediate parent
- if gt.IsUnlinkRequired(queuePath) {
- gt.UnlinkQT(queuePath)
- } else {
- gt.setLimits(queuePath,
resources.NewResource(), 0)
- log.Log(log.SchedUGM).Debug("Cleared earlier
set limit configs for group",
- zap.String("group", g),
- zap.String("queue path", queuePath))
- }
- // Does "root" queue has any child queue trackers? At
some point during this whole traversal, root might
- // not have any child queue trackers. When the
situation comes, remove the linkage between the group and
- // its root queue tracker
- if gt.canBeRemoved() {
- delete(m.groupTrackers, gt.groupName)
- }
+ log.Log(log.SchedUGM).Debug("Need to clear earlier set configs
for group",
+ zap.String("group", gt.groupName),
+ zap.String("queue path", queuePath))
+ appUsersMap := gt.decreaseAllTrackedResourceUsage(queuePath)
+ for app, u := range appUsersMap {
+ ut := m.userTrackers[u]
+ delete(ut.appGroupTrackers, app)
+ }
+ gt.setLimits(queuePath, resources.NewResource(), 0)
+ // Is there any running applications in end queue of this queue
path? If not, then remove the linkage between end queue and its immediate parent
+ if gt.IsUnlinkRequired(queuePath) {
+ gt.UnlinkQT(queuePath)
+ }
+ log.Log(log.SchedUGM).Debug("Cleared earlier set limit configs
for group",
+ zap.String("group", gt.groupName),
+ zap.String("queue path", queuePath))
+ if gt.canBeRemoved() {
+ delete(m.groupTrackers, gt.groupName)
}
}
- return appUsersMap
+}
+
+func (m *Manager) replaceLimitConfigs(newUserLimits
map[string]map[string]*LimitConfig, newGroupLimits
map[string]map[string]*LimitConfig,
+ newUserWildCardLimitsConfig map[string]*LimitConfig,
newGroupWildCardLimitsConfig map[string]*LimitConfig, newConfiguredGroups
map[string][]string) {
+ m.Lock()
+ defer m.Unlock()
+ m.userLimits = newUserLimits
+ m.groupLimits = newGroupLimits
+ m.userWildCardLimitsConfig = newUserWildCardLimitsConfig
+ m.groupWildCardLimitsConfig = newGroupWildCardLimitsConfig
+ m.configuredGroups = newConfiguredGroups
}
func (m *Manager) setUserLimits(user string, limitConfig *LimitConfig,
queuePath string) error {
+ m.Lock()
+ defer m.Unlock()
log.Log(log.SchedUGM).Debug("Setting user limits",
zap.String("user", user),
zap.String("queue path", queuePath),
@@ -504,6 +550,8 @@ func (m *Manager) setUserLimits(user string, limitConfig
*LimitConfig, queuePath
}
func (m *Manager) setGroupLimits(group string, limitConfig *LimitConfig,
queuePath string) error {
+ m.Lock()
+ defer m.Unlock()
log.Log(log.SchedUGM).Debug("Setting group limits",
zap.String("group", group),
zap.String("queue path", queuePath),
@@ -620,8 +668,20 @@ func (m *Manager) ClearUserTrackers() {
m.userTrackers = make(map[string]*UserTracker)
}
+// ClearGroupTrackers only for tests
func (m *Manager) ClearGroupTrackers() {
m.Lock()
defer m.Unlock()
m.groupTrackers = make(map[string]*GroupTracker)
}
+
+// ClearConfigLimits only for tests
+func (m *Manager) ClearConfigLimits() {
+ m.Lock()
+ defer m.Unlock()
+ m.userWildCardLimitsConfig = make(map[string]*LimitConfig)
+ m.groupWildCardLimitsConfig = make(map[string]*LimitConfig)
+ m.configuredGroups = make(map[string][]string)
+ m.userLimits = make(map[string]map[string]*LimitConfig)
+ m.groupLimits = make(map[string]map[string]*LimitConfig)
+}
diff --git a/pkg/scheduler/ugm/manager_test.go
b/pkg/scheduler/ugm/manager_test.go
index 4a0c8365..7ce5e72b 100644
--- a/pkg/scheduler/ugm/manager_test.go
+++ b/pkg/scheduler/ugm/manager_test.go
@@ -59,6 +59,7 @@ func TestUserManagerOnceInitialization(t *testing.T) {
}
func TestGetGroup(t *testing.T) {
+ setupUGM()
user := security.UserGroup{User: "test", Groups: []string{"test",
"test1"}}
manager := GetUserManager()
@@ -368,15 +369,11 @@ func TestUpdateConfigWithWildCardUsersAndGroups(t
*testing.T) {
// should run as user 'user' setting is map[memory:60 vcores:60] and
total usage of "root.parent" is map[memory:50 vcores:50]
increased := manager.IncreaseTrackedResource(queuePath1, TestApp2,
usage, user)
- if !increased {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, user)
- }
+ assert.Equal(t, increased, true)
// should not run as user 'user' setting is map[memory:60 vcores:60]
and total usage of "root.parent" is map[memory:60 vcores:60]
increased = manager.IncreaseTrackedResource(queuePath1, TestApp3,
usage, user)
- if increased {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, user)
- }
+ assert.Equal(t, increased, false)
// configure max resource for root.parent to allow one more application
to run through wild card user settings (not through specific user)
// configure limits for user2 only. However, user1 should not be
cleared as it has running applications
@@ -390,29 +387,22 @@ func TestUpdateConfigWithWildCardUsersAndGroups(t
*testing.T) {
// user1 still should be able to run app as wild card user '*' setting
is map[memory:70 vcores:70] for "root.parent" and
// total usage of "root.parent" is map[memory:60 vcores:60]
increased = manager.IncreaseTrackedResource(queuePath1, TestApp2,
usage, user)
- if !increased {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp2, user)
- }
+ assert.Equal(t, increased, true)
// user1 should not be able to run app as wild card user '*' setting is
map[memory:70 vcores:70] for "root.parent"
// and total usage of "root.parent" is map[memory:70 vcores:70]
increased = manager.IncreaseTrackedResource(queuePath1, TestApp3,
usage, user)
- if increased {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp3, user)
- }
+ assert.Equal(t, increased, false)
// configure max resource for group1 * root.parent (map[memory:70
vcores:70]) higher than wild card group * root.parent settings (map[memory:10
vcores:10])
// ensure group's specific settings has been used for enforcement
checks as specific limits always has higher precedence when compared to wild
card group limit settings
+ // group1 quota (map[memory:70 vcores:70]) has been removed from
root.parent.leaf.
+ // so resource usage has been decreased for the same group and that too
for entire queue hierarchy (root->parent->leaf)
+ // since group1 quota has been configured for root.parent, resource
usage would be increased from
+ // the place where it has been left.
conf = createUpdateConfigWithWildCardUsersAndGroups(user1.User,
user1.Groups[0], "*", "*", "10", "10")
assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
- // since group1 quota (map[memory:70 vcores:70]) has reached before,
further group increase is not allowed.
- // Hence, user's tracked usage would be reverted
- increased = manager.IncreaseTrackedResource(queuePath1, TestApp1,
usage, user1)
- if increased {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, user1)
- }
-
// configure max resource for user2 * root.parent (map[memory:70
vcores:70]) higher than wild card user * root.parent settings (map[memory:10
vcores:10])
// ensure user's specific settings has been used for enforcement checks
as specific limits always has higher precedence when compared to wild card user
limit settings
conf = createUpdateConfigWithWildCardUsersAndGroups(user1.User, "",
"*", "", "10", "10")
@@ -421,16 +411,12 @@ func TestUpdateConfigWithWildCardUsersAndGroups(t
*testing.T) {
// can be allowed to run upto resource usage map[memory:70 vcores:70]
for i := 1; i <= 7; i++ {
increased = manager.IncreaseTrackedResource(queuePath1,
TestApp1, usage, user1)
- if !increased {
- t.Fatalf("unable to increase tracked resource:
queuepath %s, app %s, res %v", queuePath1, TestApp1, user1)
- }
+ assert.Equal(t, increased, true)
}
// user2 should not be able to run app as user2 max limit is
map[memory:70 vcores:70] and usage so far is map[memory:70 vcores:70]
increased = manager.IncreaseTrackedResource(queuePath1, TestApp1,
usage, user1)
- if increased {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, user1)
- }
+ assert.Equal(t, increased, false)
user3 := security.UserGroup{User: "user3", Groups: []string{"group3"}}
conf = createUpdateConfigWithWildCardUsersAndGroups(user1.User,
user1.Groups[0], "", "*", "10", "10")
@@ -438,9 +424,7 @@ func TestUpdateConfigWithWildCardUsersAndGroups(t
*testing.T) {
// user3 should be able to run app as group3 uses wild card group limit
settings map[memory:10 vcores:10]
increased = manager.IncreaseTrackedResource(queuePath1, TestApp1,
usage, user3)
- if !increased {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, user)
- }
+ assert.Equal(t, increased, true)
// user4 (though belongs to different group, group4) should not be able
to run app as group4 also
// uses wild card group limit settings map[memory:10 vcores:10]
@@ -453,16 +437,12 @@ func TestUpdateConfigWithWildCardUsersAndGroups(t
*testing.T) {
// Since app is TestApp1, gt of "*" would be used as it is already
mapped. group4 won't be used
increased = manager.IncreaseTrackedResource(queuePath1, TestApp1,
usage, user4)
- if increased {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, user1)
- }
+ assert.Equal(t, increased, false)
// Now group4 would be used as user4 is running TestApp2 for the first
time. So can be allowed to run upto resource usage map[memory:70 vcores:70]
for i := 1; i <= 7; i++ {
increased = manager.IncreaseTrackedResource(queuePath1,
TestApp2, usage, user4)
- if !increased {
- t.Fatalf("unable to increase tracked resource:
queuepath %s, app %s, res %v", queuePath1, TestApp1, user1)
- }
+ assert.Equal(t, increased, true)
}
// user4 should not be able to run app as user4 max limit is
map[memory:70 vcores:70] and usage so far is map[memory:70 vcores:70]
@@ -540,6 +520,28 @@ func TestUpdateConfigClearEarlierSetLimits(t *testing.T) {
assert.Equal(t, len(manager.groupWildCardLimitsConfig), 0)
}
+func TestUpdateConfigClearEarlierSetGroupLimits(t *testing.T) {
+ setupUGM()
+ user := security.UserGroup{User: "user1", Groups: []string{"group1"}}
+ conf := createConfigWithGroupOnly(user.Groups[0], 50, 5)
+
+ manager := GetUserManager()
+ assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
+
+ usage, err := resources.NewResourceFromConf(map[string]string{"memory":
"25", "vcores": "25"})
+ if err != nil {
+ t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage)
+ }
+ cQueue := "root.parent.leaf"
+ for i := 1; i <= 2; i++ {
+ increased := manager.IncreaseTrackedResource(cQueue, TestApp1,
usage, user)
+ assert.Equal(t, increased, true, "unable to increase tracked
resource: queuepath "+cQueue+", app "+TestApp1+", res "+usage.String())
+ }
+ assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
+ increased := manager.IncreaseTrackedResource(cQueue, TestApp1, usage,
user)
+ assert.Equal(t, increased, false, "unable to increase tracked resource:
queuepath "+cQueue+", app "+TestApp1+", res "+usage.String())
+}
+
func TestSetMaxLimitsForRemovedUsers(t *testing.T) {
setupUGM()
// Queue setup:
@@ -653,8 +655,12 @@ func TestUserGroupHeadroom(t *testing.T) {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage)
}
// ensure group headroom returned when there is no limit settings
configured for user
+ // group1 quota (map[memory:70 vcores:70]) has been removed from
root.parent.leaf.
+ // so resource usage has been decreased for the same group and that too
for entire queue hierarchy (root->parent->leaf)
+ // since group1 quota has been configured for root.parent, resource
usage would be increased from
+ // the place where it has been left. so there is no usage after the
recent config change, entire group's quota would be returned as headroom.
headroom = manager.Headroom("root.parent", TestApp1, user)
- assert.Equal(t, resources.Equals(headroom, resources.Sub(usage1,
usage)), true)
+ assert.Equal(t, resources.Equals(headroom, usage1), true)
}
func TestDecreaseTrackedResourceForGroupTracker(t *testing.T) {
@@ -1304,6 +1310,48 @@ func createConfigWithDifferentGroups(user string, group
string, resourceKey stri
return conf
}
+func createConfigWithGroupOnly(group string, mem int, maxApps uint64)
configs.PartitionConfig {
+ conf := configs.PartitionConfig{
+ Name: "test",
+ Queues: []configs.QueueConfig{
+ {
+ Name: "root",
+ Parent: true,
+ SubmitACL: "*",
+ Queues: []configs.QueueConfig{
+ {
+ Name: "parent",
+ Parent: true,
+ SubmitACL: "*",
+ Queues: []configs.QueueConfig{
+ {
+ Name:
"leaf",
+ Parent:
false,
+ SubmitACL: "*",
+ Queues: nil,
+ },
+ },
+ Limits: []configs.Limit{
+ {
+ Limit: "parent
queue limit",
+ Groups:
[]string{
+ group,
+ },
+ MaxResources:
map[string]string{
+
"memory": strconv.Itoa(mem),
+
"vcores": strconv.Itoa(mem),
+ },
+
MaxApplications: maxApps,
+ },
+ },
+ },
+ },
+ },
+ },
+ }
+ return conf
+}
+
func createConfigWithoutLimits() configs.PartitionConfig {
conf := configs.PartitionConfig{
Name: "test",
@@ -1330,6 +1378,7 @@ func setupUGM() {
manager := GetUserManager()
manager.ClearUserTrackers()
manager.ClearGroupTrackers()
+ manager.ClearConfigLimits()
}
func assertUGM(t *testing.T, userGroup security.UserGroup, expected
*resources.Resource, usersCount int) {
diff --git a/pkg/scheduler/ugm/queue_tracker.go
b/pkg/scheduler/ugm/queue_tracker.go
index deef4a74..39124c25 100644
--- a/pkg/scheduler/ugm/queue_tracker.go
+++ b/pkg/scheduler/ugm/queue_tracker.go
@@ -250,6 +250,7 @@ func (qt *QueueTracker) headroom(hierarchy []string)
*resources.Resource {
}
childHeadroom =
qt.childQueueTrackers[childName].headroom(hierarchy[1:])
}
+
// arrived at the leaf or on the way out: check against current max if
set
if !resources.Equals(resources.NewResource(), qt.maxResources) {
headroom = qt.maxResources.Clone()
@@ -349,6 +350,10 @@ func (qt *QueueTracker) UnlinkQT(hierarchy []string) bool {
if qt.childQueueTrackers[childName] != nil {
if
qt.childQueueTrackers[childName].UnlinkQT(hierarchy[1:]) {
delete(qt.childQueueTrackers, childName)
+ // returning false, so that it comes out when
end queue detach itself from its immediate parent.
+ // i.e., once leaf detached from root.parent
for root.parent.leaf queue path.
+ // otherwise, detachment continues all the way
upto the root, even parent from root. which is not needed.
+ return false
}
}
} else if len(hierarchy) <= 1 {
@@ -389,12 +394,10 @@ func (qt *QueueTracker)
decreaseTrackedResourceUsageDownwards(hierarchy []string
}
}
}
-
if len(qt.runningApplications) > 0 && qt.resourceUsage !=
resources.NewResource() {
qt.resourceUsage = resources.NewResource()
qt.runningApplications = make(map[string]bool)
}
-
return removedApplications
}
@@ -456,3 +459,29 @@ func (qt *QueueTracker) canRunApp(hierarchy []string,
applicationID string, trac
}
return true
}
+
+// canBeRemoved Start from root and reach all levels of queue hierarchy to
confirm whether corresponding queue tracker
+// object can be removed from ugm or not. Based on running applications,
resource usage, child queue trackers, max running apps, max resources etc
+// it decides the removal. It returns false the moment it sees any unexpected
values for any queue in any levels.
+func (qt *QueueTracker) canBeRemoved() bool {
+ for _, childQT := range qt.childQueueTrackers {
+ // quick check to avoid further traversal
+ if childQT.canBeRemovedInternal() {
+ if !childQT.canBeRemoved() {
+ return false
+ }
+ } else {
+ return false
+ }
+ }
+ // reached leaf queues, no more to traverse
+ return qt.canBeRemovedInternal()
+}
+
+func (qt *QueueTracker) canBeRemovedInternal() bool {
+ if len(qt.runningApplications) == 0 &&
resources.Equals(resources.NewResource(), qt.resourceUsage) &&
len(qt.childQueueTrackers) == 0 &&
+ qt.maxRunningApps == 0 &&
resources.Equals(resources.NewResource(), qt.maxResources) {
+ return true
+ }
+ return false
+}
diff --git a/pkg/scheduler/ugm/user_tracker.go
b/pkg/scheduler/ugm/user_tracker.go
index 5639253d..5e842d58 100644
--- a/pkg/scheduler/ugm/user_tracker.go
+++ b/pkg/scheduler/ugm/user_tracker.go
@@ -164,11 +164,10 @@ func (ut *UserTracker) UnlinkQT(queuePath string) bool {
return ut.queueTracker.UnlinkQT(strings.Split(queuePath, configs.DOT))
}
-// canBeRemoved Does "root" queue has any child queue trackers? Is there any
running applications in "root" qt?
func (ut *UserTracker) canBeRemoved() bool {
ut.RLock()
defer ut.RUnlock()
- return len(ut.queueTracker.childQueueTrackers) == 0 &&
len(ut.queueTracker.runningApplications) == 0
+ return ut.queueTracker.canBeRemoved()
}
func (ut *UserTracker) canRunApp(queuePath, applicationID string) bool {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
