This is an automated email from the ASF dual-hosted git repository.
mani 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 148f221a [YUNIKORN-1610] Enforcement changes for User Based Quota(#563)
148f221a is described below
commit 148f221ab4f17d10e2e20b52fc2c48991d5b2b65
Author: Manikandan R <[email protected]>
AuthorDate: Mon Jun 26 12:19:33 2023 +0530
[YUNIKORN-1610] Enforcement changes for User Based Quota(#563)
Closes: #563
Signed-off-by: Manikandan R <[email protected]>
---
pkg/common/configs/config_test.go | 2 +
pkg/common/configs/configvalidator.go | 3 +-
.../ugm/tracker.go => common/constants.go} | 21 +-
pkg/common/security/acl.go | 19 +-
pkg/common/security/acl_test.go | 16 +-
pkg/scheduler/objects/application.go | 16 +-
pkg/scheduler/partition_test.go | 21 +-
pkg/scheduler/ugm/group_tracker.go | 41 +-
pkg/scheduler/ugm/group_tracker_test.go | 97 ++--
pkg/scheduler/ugm/manager.go | 578 ++++++++-------------
pkg/scheduler/ugm/manager_test.go | 557 +++++++++++++-------
pkg/scheduler/ugm/queue_tracker.go | 283 +++++++---
pkg/scheduler/ugm/queue_tracker_test.go | 144 +++--
pkg/scheduler/ugm/tracker.go | 4 +-
pkg/scheduler/ugm/user_tracker.go | 41 +-
pkg/scheduler/ugm/user_tracker_test.go | 84 ++-
pkg/scheduler/utilities_test.go | 20 +-
17 files changed, 1090 insertions(+), 857 deletions(-)
diff --git a/pkg/common/configs/config_test.go
b/pkg/common/configs/config_test.go
index ca67a47a..437d73a4 100644
--- a/pkg/common/configs/config_test.go
+++ b/pkg/common/configs/config_test.go
@@ -1848,6 +1848,7 @@ partitions:
`
return fmt.Sprintf(data, leafQueueMaxResource, leafQueueMaxResource,
leafQueueMaxApps, userMaxResource)
}
+
func TestUserLimitsWithHierarchicalQueue(t *testing.T) {
// Make sure parent queue user max apps should not less than the child
queue max resource
// validate the config and check after the update
@@ -2050,6 +2051,7 @@ partitions:
`
return fmt.Sprintf(data, leafQueueMaxResources, leafQueueMaxResources,
leafQueueMaxApps, groupMaxResources)
}
+
func TestGroupLimitsWithHierarchicalQueue(t *testing.T) {
// Make sure parent queue user max apps should not less than the child
queue max resource
// validate the config and check after the update
diff --git a/pkg/common/configs/configvalidator.go
b/pkg/common/configs/configvalidator.go
index ab7dc550..b4c25e45 100644
--- a/pkg/common/configs/configvalidator.go
+++ b/pkg/common/configs/configvalidator.go
@@ -31,7 +31,6 @@ import (
"github.com/apache/yunikorn-core/pkg/common"
"github.com/apache/yunikorn-core/pkg/common/resources"
- "github.com/apache/yunikorn-core/pkg/common/security"
"github.com/apache/yunikorn-core/pkg/log"
"github.com/apache/yunikorn-core/pkg/scheduler/placement/types"
"github.com/apache/yunikorn-core/pkg/scheduler/policies"
@@ -99,7 +98,7 @@ func checkACL(acl string) error {
// trim any white space
acl = strings.TrimSpace(acl)
// handle special cases: deny and wildcard
- if len(acl) == 0 || acl == security.WildCard {
+ if len(acl) == 0 || acl == common.Wildcard {
return nil
}
diff --git a/pkg/scheduler/ugm/tracker.go b/pkg/common/constants.go
similarity index 52%
copy from pkg/scheduler/ugm/tracker.go
copy to pkg/common/constants.go
index 85f3c56d..f7dc90d0 100644
--- a/pkg/scheduler/ugm/tracker.go
+++ b/pkg/common/constants.go
@@ -16,21 +16,10 @@
limitations under the License.
*/
-package ugm
+package common
-import (
- "github.com/apache/yunikorn-core/pkg/common/resources"
- "github.com/apache/yunikorn-core/pkg/common/security"
+const (
+ Wildcard = "*"
+ Separator = ","
+ Space = " "
)
-
-// Tracker Defines a set of interfaces to track and retrieve the user group
resource usage
-type Tracker interface {
- GetUserResources(user security.UserGroup) *resources.Resource
- GetGroupResources(group string) *resources.Resource
-
- GetUsersResources() []*UserTracker
- GetGroupsResources() []*GroupTracker
-
- IncreaseTrackedResource(queuePath, applicationID string, usage
*resources.Resource, user security.UserGroup) error
- DecreaseTrackedResource(queuePath, applicationID string, usage
*resources.Resource, user security.UserGroup, removeApp bool) error
-}
diff --git a/pkg/common/security/acl.go b/pkg/common/security/acl.go
index 3d83a5ad..e1ec919f 100644
--- a/pkg/common/security/acl.go
+++ b/pkg/common/security/acl.go
@@ -25,15 +25,10 @@ import (
"go.uber.org/zap"
+ "github.com/apache/yunikorn-core/pkg/common"
"github.com/apache/yunikorn-core/pkg/log"
)
-const (
- WildCard = "*"
- Separator = ","
- Space = " "
-)
-
// User and group regexp, must allow at least what we allow in the config
checks
// See configs.UserNameRegExp and configs.GroupRegExp in the config validator.
var userNameRegExp = regexp.MustCompile("^[_a-zA-Z][a-zA-Z0-9_.@-]*[$]?$")
@@ -48,7 +43,7 @@ type ACL struct {
// the ACL allows all access, set the flag
func (a *ACL) setAllAllowed(part string) {
part = strings.TrimSpace(part)
- a.allAllowed = part == WildCard
+ a.allAllowed = part == common.Wildcard
}
// set the user list in the ACL, invalid user names are ignored
@@ -59,7 +54,7 @@ func (a *ACL) setUsers(userList []string) {
return
}
// special case if the user list is just the wildcard
- if len(userList) == 1 && userList[0] == WildCard {
+ if len(userList) == 1 && userList[0] == common.Wildcard {
log.Log(log.Security).Info("user list is wildcard, allowing all
access")
a.allAllowed = true
return
@@ -92,7 +87,7 @@ func (a *ACL) setGroups(groupList []string) {
log.Log(log.Security).Info("ignoring group list in ACL:
wildcard set")
return
}
- if len(groupList) == 1 && groupList[0] == WildCard {
+ if len(groupList) == 1 && groupList[0] == common.Wildcard {
log.Log(log.Security).Info("group list is wildcard, allowing
all access")
a.users = make(map[string]bool)
a.allAllowed = true
@@ -122,7 +117,7 @@ func NewACL(aclStr string) (ACL, error) {
}
// before trimming check
// should have no more than two groups defined
- fields := strings.Split(aclStr, Space)
+ fields := strings.Split(aclStr, common.Space)
if len(fields) > 2 {
return acl, fmt.Errorf("multiple spaces found in ACL: '%s'",
aclStr)
}
@@ -133,9 +128,9 @@ func NewACL(aclStr string) (ACL, error) {
return acl, nil
}
// parse users and groups
- acl.setUsers(strings.Split(fields[0], Separator))
+ acl.setUsers(strings.Split(fields[0], common.Separator))
if len(fields) == 2 {
- acl.setGroups(strings.Split(fields[1], Separator))
+ acl.setGroups(strings.Split(fields[1], common.Separator))
}
return acl, nil
}
diff --git a/pkg/common/security/acl_test.go b/pkg/common/security/acl_test.go
index db2c4993..b57649d4 100644
--- a/pkg/common/security/acl_test.go
+++ b/pkg/common/security/acl_test.go
@@ -22,6 +22,8 @@ import (
"errors"
"fmt"
"testing"
+
+ "github.com/apache/yunikorn-core/pkg/common"
)
func IsSameACL(got, expected ACL) error {
@@ -111,23 +113,23 @@ func TestACLCreate(t *testing.T) {
ACL{users: make(map[string]bool), groups:
map[string]bool{"group1": true, "group2": true}, allAllowed: false},
},
{
- "* group1,group2",
+ common.Wildcard + " group1,group2",
ACL{users: make(map[string]bool), groups:
make(map[string]bool), allAllowed: true},
},
{
- "user1,user2 *",
+ "user1,user2 " + common.Wildcard,
ACL{users: make(map[string]bool), groups:
make(map[string]bool), allAllowed: true},
},
{
- "*",
+ common.Wildcard,
ACL{users: make(map[string]bool), allAllowed: true},
},
{
- "* ",
+ common.Wildcard + " ",
ACL{users: make(map[string]bool), groups:
make(map[string]bool), allAllowed: true},
},
{
- " *",
+ " " + common.Wildcard,
ACL{users: make(map[string]bool), groups:
make(map[string]bool), allAllowed: true},
},
{
@@ -210,12 +212,12 @@ func TestACLAccess(t *testing.T) {
false,
},
{
- "*",
+ common.Wildcard,
UserGroup{User: "", Groups: nil},
true,
},
{
- "*",
+ common.Wildcard,
UserGroup{User: "user1", Groups: []string{"group1"}},
true,
},
diff --git a/pkg/scheduler/objects/application.go
b/pkg/scheduler/objects/application.go
index 4de1764c..4ac248d6 100644
--- a/pkg/scheduler/objects/application.go
+++ b/pkg/scheduler/objects/application.go
@@ -1637,25 +1637,13 @@ func (sa *Application) addAllocationInternal(info
*Allocation) {
// Increase user resource usage
// No locking must be called while holding the lock
func (sa *Application) incUserResourceUsage(resource *resources.Resource) {
- if err := ugm.GetUserManager().IncreaseTrackedResource(sa.queuePath,
sa.ApplicationID, resource, sa.user); err != nil {
- log.Log(log.SchedApplication).Error("Unable to track the user
resource usage",
- zap.String("application id", sa.ApplicationID),
- zap.String("user", sa.user.User),
- zap.String("currentState", sa.stateMachine.Current()),
- zap.Error(err))
- }
+ ugm.GetUserManager().IncreaseTrackedResource(sa.queuePath,
sa.ApplicationID, resource, sa.user)
}
// Decrease user resource usage
// No locking must be called while holding the lock
func (sa *Application) decUserResourceUsage(resource *resources.Resource,
removeApp bool) {
- if err := ugm.GetUserManager().DecreaseTrackedResource(sa.queuePath,
sa.ApplicationID, resource, sa.user, removeApp); err != nil {
- log.Log(log.SchedApplication).Error("Unable to track the user
resource usage",
- zap.String("application id", sa.ApplicationID),
- zap.String("user", sa.user.User),
- zap.String("currentState", sa.stateMachine.Current()),
- zap.Error(err))
- }
+ ugm.GetUserManager().DecreaseTrackedResource(sa.queuePath,
sa.ApplicationID, resource, sa.user, removeApp)
}
// When the resource allocated with this allocation is to be removed,
diff --git a/pkg/scheduler/partition_test.go b/pkg/scheduler/partition_test.go
index 99210f11..5a2995d2 100644
--- a/pkg/scheduler/partition_test.go
+++ b/pkg/scheduler/partition_test.go
@@ -107,7 +107,7 @@ func TestNewPartition(t *testing.T) {
"memory": "10",
"vcores": "10",
},
- MaxApplications: 1,
+ MaxApplications: 2,
},
},
},
@@ -350,7 +350,7 @@ func TestRemoveNodeWithAllocations(t *testing.T) {
assert.Equal(t, 1, len(released), "node did not release correct
allocation")
assert.Equal(t, 0, len(confirmed), "node did not confirm correct
allocation")
assert.Equal(t, released[0].GetUUID(), allocUUID, "uuid returned by
release not the same as on allocation")
- assertLimits(t, getTestUserGroup(), nil)
+ assertLimits(t, getTestUserGroup(), resources.Zero)
// wait for events to be processed
err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
@@ -1084,7 +1084,7 @@ func TestRemoveApp(t *testing.T) {
allocs = partition.removeApplication("will_not_remove")
assert.Equal(t, 1, len(allocs), "existing application with allocations
returned unexpected allocations %v", allocs)
- assertLimits(t, getTestUserGroup(), nil)
+ assertLimits(t, getTestUserGroup(), resources.Zero)
}
func TestRemoveAppAllocs(t *testing.T) {
@@ -1145,7 +1145,7 @@ func TestRemoveAppAllocs(t *testing.T) {
allocs, _ = partition.removeAllocation(release)
assert.Equal(t, 1, len(allocs), "removal request for existing
allocation returned wrong allocations: %v", allocs)
assert.Equal(t, 0, partition.GetTotalAllocationCount(), "removal
requests did not remove all allocations: %v", partition.allocations)
- assertLimits(t, getTestUserGroup(), nil)
+ assertLimits(t, getTestUserGroup(), resources.Zero)
}
// Dynamic queue creation based on the name from the rules
@@ -1375,7 +1375,7 @@ func TestTryAllocate(t *testing.T) {
expectedQueuesMaxLimits["root.leaf"] = make(map[string]interface{})
expectedQueuesMaxLimits["root"][maxresources] =
resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 10,
"vcores": 10})
expectedQueuesMaxLimits["root.leaf"][maxresources] =
resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5,
"vcores": 5})
- expectedQueuesMaxLimits["root"][maxapplications] = uint64(2)
+ expectedQueuesMaxLimits["root"][maxapplications] = uint64(10)
expectedQueuesMaxLimits["root.leaf"][maxapplications] = uint64(1)
assertUserGroupResourceMaxLimits(t, getTestUserGroup(), nil,
expectedQueuesMaxLimits)
@@ -1891,10 +1891,9 @@ func getExpectedQueuesLimitsForPreemption()
map[string]map[string]interface{} {
expectedQueuesMaxLimits["root"][maxresources] =
resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 10,
"vcores": 10})
expectedQueuesMaxLimits["root.parent"][maxresources] =
resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5,
"vcores": 5})
expectedQueuesMaxLimits["root.parent.leaf1"][maxresources] =
expectedQueuesMaxLimits["root.parent"][maxresources]
- expectedQueuesMaxLimits["root"][maxapplications] = uint64(2)
- expectedQueuesMaxLimits["root.parent"][maxapplications] = uint64(2)
- expectedQueuesMaxLimits["root.parent.leaf1"][maxapplications] =
uint64(1)
- expectedQueuesMaxLimits["root.parent.leaf1"][maxapplications] =
uint64(1)
+ expectedQueuesMaxLimits["root"][maxapplications] = uint64(10)
+ expectedQueuesMaxLimits["root.parent"][maxapplications] = uint64(8)
+ expectedQueuesMaxLimits["root.parent.leaf1"][maxapplications] =
uint64(8)
return expectedQueuesMaxLimits
}
@@ -1908,9 +1907,9 @@ func
getExpectedQueuesLimitsForPreemptionWithRequiredNode() map[string]map[strin
expectedQueuesMaxLimits["root.leaf"][maxresources] =
resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5,
"vcores": 5})
expectedQueuesMaxLimits["root.parent"][maxresources] =
expectedQueuesMaxLimits["root.leaf"][maxresources]
expectedQueuesMaxLimits["root.parent.sub-leaf"][maxresources] =
resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 3,
"vcores": 3})
- expectedQueuesMaxLimits["root"][maxapplications] = uint64(2)
+ expectedQueuesMaxLimits["root"][maxapplications] = uint64(10)
expectedQueuesMaxLimits["root.leaf"][maxapplications] = uint64(1)
- expectedQueuesMaxLimits["root.parent"][maxapplications] = uint64(2)
+ expectedQueuesMaxLimits["root.parent"][maxapplications] = uint64(8)
expectedQueuesMaxLimits["root.parent.sub-leaf"][maxapplications] =
uint64(2)
return expectedQueuesMaxLimits
}
diff --git a/pkg/scheduler/ugm/group_tracker.go
b/pkg/scheduler/ugm/group_tracker.go
index 6a4f99b2..48287a44 100644
--- a/pkg/scheduler/ugm/group_tracker.go
+++ b/pkg/scheduler/ugm/group_tracker.go
@@ -43,14 +43,14 @@ func newGroupTracker(group string) *GroupTracker {
return groupTracker
}
-func (gt *GroupTracker) increaseTrackedResource(queuePath, applicationID
string, usage *resources.Resource) error {
+func (gt *GroupTracker) increaseTrackedResource(queuePath, applicationID
string, usage *resources.Resource) bool {
gt.Lock()
defer gt.Unlock()
gt.applications[applicationID] = true
- return gt.queueTracker.increaseTrackedResource(queuePath,
applicationID, usage)
+ return gt.queueTracker.increaseTrackedResource(queuePath,
applicationID, group, usage)
}
-func (gt *GroupTracker) decreaseTrackedResource(queuePath, applicationID
string, usage *resources.Resource, removeApp bool) (bool, error) {
+func (gt *GroupTracker) decreaseTrackedResource(queuePath, applicationID
string, usage *resources.Resource, removeApp bool) (bool, bool) {
gt.Lock()
defer gt.Unlock()
if removeApp {
@@ -65,16 +65,10 @@ func (gt *GroupTracker) getTrackedApplications()
map[string]bool {
return gt.applications
}
-func (gt *GroupTracker) setMaxApplications(count uint64, queuePath string)
error {
+func (gt *GroupTracker) setLimits(queuePath string, resource
*resources.Resource, maxApps uint64) {
gt.Lock()
defer gt.Unlock()
- return gt.queueTracker.setMaxApplications(count, queuePath)
-}
-
-func (gt *GroupTracker) setMaxResources(resource *resources.Resource,
queuePath string) error {
- gt.Lock()
- defer gt.Unlock()
- return gt.queueTracker.setMaxResources(resource, queuePath)
+ gt.queueTracker.setLimit(queuePath, resource, maxApps)
}
func (gt *GroupTracker) GetGroupResourceUsageDAOInfo()
*dao.GroupResourceUsageDAOInfo {
@@ -90,3 +84,28 @@ func (gt *GroupTracker) GetGroupResourceUsageDAOInfo()
*dao.GroupResourceUsageDA
groupResourceUsage.Queues = gt.queueTracker.getResourceUsageDAOInfo("")
return groupResourceUsage
}
+
+func (gt *GroupTracker) IsQueuePathTrackedCompletely(queuePath string) bool {
+ gt.RLock()
+ defer gt.RUnlock()
+ return gt.queueTracker.IsQueuePathTrackedCompletely(queuePath)
+}
+
+func (gt *GroupTracker) IsUnlinkRequired(queuePath string) bool {
+ gt.RLock()
+ defer gt.RUnlock()
+ return gt.queueTracker.IsUnlinkRequired(queuePath)
+}
+
+func (gt *GroupTracker) UnlinkQT(queuePath string) bool {
+ gt.RLock()
+ defer gt.RUnlock()
+ return gt.queueTracker.UnlinkQT(queuePath)
+}
+
+// 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
+}
diff --git a/pkg/scheduler/ugm/group_tracker_test.go
b/pkg/scheduler/ugm/group_tracker_test.go
index f9608b21..7d0ce280 100644
--- a/pkg/scheduler/ugm/group_tracker_test.go
+++ b/pkg/scheduler/ugm/group_tracker_test.go
@@ -32,6 +32,7 @@ func TestGTIncreaseTrackedResource(t *testing.T) {
// root->parent->child1->child12
// root->parent->child2
// root->parent->child12 (similar name like above leaf queue, but it is
being treated differently as similar names are allowed)
+ GetUserManager()
user := &security.UserGroup{User: "test", Groups: []string{"test"}}
groupTracker := newGroupTracker(user.User)
@@ -39,36 +40,36 @@ func TestGTIncreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage1)
}
- err = groupTracker.increaseTrackedResource(queuePath1, TestApp1, usage1)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage1, err)
+ result := groupTracker.increaseTrackedResource(queuePath1, TestApp1,
usage1)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, usage1)
}
usage2, err := resources.NewResourceFromConf(map[string]string{"mem":
"20M", "vcore": "20"})
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage2)
}
- err = groupTracker.increaseTrackedResource(queuePath2, TestApp2, usage2)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath2, TestApp2, usage2, err)
+ result = groupTracker.increaseTrackedResource(queuePath2, TestApp2,
usage2)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath2, TestApp2, usage2)
}
usage3, err := resources.NewResourceFromConf(map[string]string{"mem":
"30M", "vcore": "30"})
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage3)
}
- err = groupTracker.increaseTrackedResource(queuePath3, TestApp3, usage3)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath3, TestApp3, usage3, err)
+ result = groupTracker.increaseTrackedResource(queuePath3, TestApp3,
usage3)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath3, TestApp3, usage3)
}
usage4, err := resources.NewResourceFromConf(map[string]string{"mem":
"20M", "vcore": "20"})
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage3)
}
- err = groupTracker.increaseTrackedResource(queuePath4, TestApp4, usage4)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath4, TestApp4, usage4, err)
+ result = groupTracker.increaseTrackedResource(queuePath4, TestApp4,
usage4)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath4, TestApp4, usage4)
}
actualResources := getGroupResource(groupTracker)
@@ -91,9 +92,9 @@ func TestGTDecreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage1)
}
- err = groupTracker.increaseTrackedResource(queuePath1, TestApp1, usage1)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage1, err)
+ result := groupTracker.increaseTrackedResource(queuePath1, TestApp1,
usage1)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, usage1)
}
assert.Equal(t, 1, len(groupTracker.getTrackedApplications()))
@@ -101,9 +102,9 @@ func TestGTDecreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage2)
}
- err = groupTracker.increaseTrackedResource(queuePath2, TestApp2, usage2)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath2, TestApp2, usage2, err)
+ result = groupTracker.increaseTrackedResource(queuePath2, TestApp2,
usage2)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath2, TestApp2, usage2)
}
actualResources := getGroupResource(groupTracker)
@@ -117,14 +118,14 @@ func TestGTDecreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage3)
}
- removeQT, err := groupTracker.decreaseTrackedResource(queuePath1,
TestApp1, usage3, false)
- if err != nil {
+ removeQT, decreased := groupTracker.decreaseTrackedResource(queuePath1,
TestApp1, usage3, false)
+ if !decreased {
t.Fatalf("unable to decrease tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage3, err)
}
assert.Equal(t, removeQT, false, "wrong remove queue tracker value")
- removeQT, err = groupTracker.decreaseTrackedResource(queuePath2,
TestApp2, usage3, false)
- if err != nil {
+ removeQT, decreased = groupTracker.decreaseTrackedResource(queuePath2,
TestApp2, usage3, false)
+ if !decreased {
t.Fatalf("unable to decrease tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath2, TestApp2, usage3, err)
}
assert.Equal(t, removeQT, false, "wrong remove queue tracker value")
@@ -141,8 +142,8 @@ func TestGTDecreaseTrackedResource(t *testing.T) {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage3)
}
- removeQT, err = groupTracker.decreaseTrackedResource(queuePath1,
TestApp1, usage4, true)
- if err != nil {
+ removeQT, decreased = groupTracker.decreaseTrackedResource(queuePath1,
TestApp1, usage4, true)
+ if !decreased {
t.Fatalf("unable to decrease tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage1, err)
}
assert.Equal(t, 1, len(groupTracker.getTrackedApplications()))
@@ -153,8 +154,8 @@ func TestGTDecreaseTrackedResource(t *testing.T) {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage3)
}
- removeQT, err = groupTracker.decreaseTrackedResource(queuePath2,
TestApp2, usage5, true)
- if err != nil {
+ removeQT, decreased = groupTracker.decreaseTrackedResource(queuePath2,
TestApp2, usage5, true)
+ if !decreased {
t.Fatalf("unable to decrease tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath2, TestApp2, usage2, err)
}
assert.Equal(t, 0, len(groupTracker.getTrackedApplications()))
@@ -170,39 +171,25 @@ func TestGTSetMaxLimits(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage1)
}
- err = groupTracker.increaseTrackedResource(queuePath1, TestApp1, usage1)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage1, err)
- }
-
- setMaxAppsErr := groupTracker.setMaxApplications(1, queuePath1)
- assert.NilError(t, setMaxAppsErr)
-
- setMaxResourcesErr := groupTracker.setMaxResources(usage1, queuePath1)
- assert.NilError(t, setMaxResourcesErr)
-
- setParentMaxAppsErr := groupTracker.setMaxApplications(1, "root.parent")
- assert.NilError(t, setParentMaxAppsErr)
-
- setParentMaxResourcesErr := groupTracker.setMaxResources(usage1,
"root.parent")
- assert.NilError(t, setParentMaxResourcesErr)
- err = groupTracker.increaseTrackedResource(queuePath1, TestApp2, usage1)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage1, err)
+ result := groupTracker.increaseTrackedResource(queuePath1, TestApp1,
usage1)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, usage1)
}
- setMaxAppsErr1 := groupTracker.setMaxApplications(1, queuePath1)
- assert.Error(t, setMaxAppsErr1, "current running applications is
greater than config max applications for "+queuePath1)
-
- setMaxResourcesErr1 := groupTracker.setMaxResources(usage1, queuePath1)
- assert.Error(t, setMaxResourcesErr1, "current resource usage is greater
than config max resource for "+queuePath1)
-
- setParentMaxAppsErr1 := groupTracker.setMaxApplications(1,
"root.parent")
- assert.Error(t, setParentMaxAppsErr1, "current running applications is
greater than config max applications for root.parent")
+ groupTracker.setLimits(queuePath1, resources.Multiply(usage1, 5), 5)
+ groupTracker.setLimits("root.parent", resources.Multiply(usage1, 10),
10)
- setParentMaxResourcesErr1 := groupTracker.setMaxResources(usage1,
"root.parent")
- assert.Error(t, setParentMaxResourcesErr1, "current resource usage is
greater than config max resource for root.parent")
+ result = groupTracker.increaseTrackedResource(queuePath1, TestApp2,
usage1)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp2, usage1)
+ }
+ result = groupTracker.increaseTrackedResource(queuePath1, TestApp3,
usage1)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp2, usage1)
+ }
+ groupTracker.setLimits(queuePath1, usage1, 1)
+ groupTracker.setLimits("root.parent", usage1, 1)
}
func getGroupResource(gt *GroupTracker) map[string]*resources.Resource {
diff --git a/pkg/scheduler/ugm/manager.go b/pkg/scheduler/ugm/manager.go
index be5b5679..5b6a41f5 100644
--- a/pkg/scheduler/ugm/manager.go
+++ b/pkg/scheduler/ugm/manager.go
@@ -24,6 +24,7 @@ import (
"go.uber.org/zap"
+ "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/common/security"
@@ -33,25 +34,22 @@ import (
var once sync.Once
var m *Manager
-const maxresources = "maxresources"
-const maxapplications = "maxapplications"
-
// Manager implements tracker. A User Group Manager to track the usage for
both user and groups.
// Holds object of both user and group trackers
type Manager struct {
- userTrackers map[string]*UserTracker
- groupTrackers map[string]*GroupTracker
- userLimitsConfig map[string]map[string]map[string]interface{} // Hold
limits settings of user * queue path
- groupLimitsConfig map[string]map[string]map[string]interface{} // Hold
limits settings of group * queue path
+ 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 '*'
sync.RWMutex
}
func newManager() *Manager {
manager := &Manager{
- userTrackers: make(map[string]*UserTracker),
- groupTrackers: make(map[string]*GroupTracker),
- userLimitsConfig:
make(map[string]map[string]map[string]interface{}),
- groupLimitsConfig:
make(map[string]map[string]map[string]interface{}),
+ userTrackers: make(map[string]*UserTracker),
+ groupTrackers: make(map[string]*GroupTracker),
+ userWildCardLimitsConfig: make(map[string]*LimitConfig),
+ groupWildCardLimitsConfig: make(map[string]*LimitConfig),
}
return manager
}
@@ -63,192 +61,134 @@ func GetUserManager() *Manager {
return m
}
+// LimitConfig Holds limit settings of wild card user/group
+type LimitConfig struct {
+ maxResources *resources.Resource
+ maxApplications uint64
+}
+
// IncreaseTrackedResource Increase the resource usage for the given user
group and queue path combination.
// As and when every allocation or asks requests fulfilled on application,
corresponding user and group
// resource usage would be increased against specific application.
-func (m *Manager) IncreaseTrackedResource(queuePath string, applicationID
string, usage *resources.Resource, user security.UserGroup) error {
- log.Log(log.SchedUGM).Debug("Increasing resource usage",
zap.String("user", user.User),
+func (m *Manager) IncreaseTrackedResource(queuePath, applicationID string,
usage *resources.Resource, user security.UserGroup) bool {
+ log.Log(log.SchedUGM).Debug("Increasing resource usage",
+ zap.String("user", user.User),
+ zap.String("group", user.Groups[0]),
zap.String("queue path", queuePath),
zap.String("application", applicationID),
zap.Stringer("resource", usage))
if queuePath == "" || applicationID == "" || usage == nil || user.User
== "" {
- log.Log(log.SchedUGM).Error("Mandatory parameters are missing
to increase the resource usage",
+ log.Log(log.SchedUGM).Debug("Mandatory parameters are missing
to increase the resource usage",
zap.String("user", user.User),
zap.String("queue path", queuePath),
zap.String("application", applicationID),
zap.Stringer("resource", usage))
- return fmt.Errorf("mandatory parameters are missing. queuepath:
%s, application id: %s, resource usage: %s, user: %s",
- queuePath, applicationID, usage.String(), user.User)
- }
- m.Lock()
- defer m.Unlock()
- var userTracker *UserTracker
- if m.userTrackers[user.User] == nil {
- userTracker = newUserTracker(user.User)
-
- // Set the limits for all configured queue paths of the user
- for configQueuePath, config := range
m.userLimitsConfig[user.User] {
- log.Log(log.SchedUGM).Debug("Setting the limit max
applications settings.",
- zap.String("user", user.User),
- zap.String("queue path", configQueuePath))
- maxApps, ok := config[maxapplications].(uint64)
- if !ok {
- log.Log(log.SchedUGM).Warn("Problem in setting
the limit max applications settings. Unable to cast the value from interface to
uint64",
- zap.String("user", user.User),
- zap.String("queue path",
configQueuePath),
- zap.Uint64("limit max applications",
maxApps))
- return fmt.Errorf("unable to set the max
applications. user: %s, queuepath : %s, applicationid: %s",
- user.User, configQueuePath,
applicationID)
- }
- err := userTracker.setMaxApplications(maxApps,
configQueuePath)
- if err != nil {
- log.Log(log.SchedUGM).Warn("Problem in setting
the limit max applications settings.",
- zap.String("user", user.User),
- zap.String("queue path",
configQueuePath),
- zap.Uint64("limit max applications",
maxApps),
- zap.Error(err))
- return fmt.Errorf("unable to set the max
applications. user: %s, queuepath : %s, applicationid: %s, usage: %s, reason:
%w",
- user.User, configQueuePath,
applicationID, usage.String(), err)
- }
- maxResources, ok :=
config[maxresources].(map[string]string)
- if !ok {
- log.Log(log.SchedUGM).Warn("Problem in setting
the limit max resources settings. Unable to cast the value from interface to
resource",
- zap.String("user", user.User),
- zap.String("queue path",
configQueuePath),
- zap.Any("limit max resources",
maxResources))
- return fmt.Errorf("unable to set the max
resources. user: %s, queuepath : %s, applicationid: %s",
- user.User, configQueuePath,
applicationID)
- }
- resource, resourceErr :=
resources.NewResourceFromConf(maxResources)
- if resourceErr != nil {
- log.Log(log.SchedUGM).Warn("Problem in setting
the limit max resources settings.",
- zap.String("user", user.User),
- zap.String("queue path",
configQueuePath),
- zap.Any("limit max resources",
maxResources),
- zap.Error(resourceErr))
- return fmt.Errorf("unable to set the max
resources. user: %s, queuepath : %s, applicationid: %s, usage: %s, reason: %w",
- user.User, configQueuePath,
applicationID, usage.String(), resourceErr)
- }
- setMaxResourcesErr :=
userTracker.setMaxResources(resource, configQueuePath)
- if setMaxResourcesErr != nil {
- log.Log(log.SchedUGM).Warn("Problem in setting
the limit max resources settings.",
- zap.String("user", user.User),
- zap.String("queue path",
configQueuePath),
- zap.Any("limit max resources",
maxResources),
- zap.Error(setMaxResourcesErr))
- return fmt.Errorf("unable to set the max
resources. user: %s, queuepath : %s, applicationid: %s, usage: %s, reason: %w",
- user.User, configQueuePath,
applicationID, usage.String(), setMaxResourcesErr)
- }
- }
- m.userTrackers[user.User] = userTracker
- } else {
- userTracker = m.userTrackers[user.User]
+ return false
}
- err := userTracker.increaseTrackedResource(queuePath, applicationID,
usage)
- if err != nil {
- log.Log(log.SchedUGM).Error("Problem in increasing the user
resource usage",
- zap.String("user", user.User),
- zap.String("queue path", queuePath),
- zap.String("application", applicationID),
- zap.Stringer("resource", usage),
- zap.String("err message", err.Error()))
- return err
+ userTracker := m.getUserTracker(user.User, true)
+ log.Log(log.SchedUGM).Debug("Increasing resource usage for user",
+ zap.String("user", user.User),
+ zap.String("queue path", queuePath),
+ zap.String("application", applicationID),
+ zap.Stringer("resource", usage))
+ increased := userTracker.increaseTrackedResource(queuePath,
applicationID, usage)
+ if !increased {
+ return increased
}
- if err = m.ensureGroupTrackerForApp(queuePath, applicationID, user);
err != nil {
- return err
+ if err := m.ensureGroupTrackerForApp(queuePath, applicationID, user);
err != nil {
+ return false
}
group, err := m.getGroup(user)
if err != nil {
- return err
+ return false
}
- groupTracker := m.groupTrackers[group]
- if groupTracker != nil {
- err = groupTracker.increaseTrackedResource(queuePath,
applicationID, usage)
- if err != nil {
- log.Log(log.SchedUGM).Error("Problem in increasing the
group resource usage",
- zap.String("user", user.User),
- zap.String("group", group),
- zap.String("queue path", queuePath),
- zap.String("application", applicationID),
- zap.Stringer("resource", usage),
- zap.String("err message", err.Error()))
- return err
- }
+ groupTracker := m.getGroupTracker(group, true)
+ log.Log(log.SchedUGM).Debug("Increasing resource usage for group",
+ zap.String("group", group),
+ zap.String("queue path", queuePath),
+ zap.String("application", applicationID),
+ zap.Stringer("resource", usage))
+ increased = groupTracker.increaseTrackedResource(queuePath,
applicationID, usage)
+ if !increased {
+ return increased
}
- return nil
+ return true
}
// DecreaseTrackedResource Decrease the resource usage for the given user
group and queue path combination.
// As and when every allocation or asks release happens, corresponding user
and group
// resource usage would be decreased against specific application. When the
final asks release happens, removeApp should be set to true and
// application itself would be removed from the tracker and no more usage
would be tracked further for that specific application.
-func (m *Manager) DecreaseTrackedResource(queuePath string, applicationID
string, usage *resources.Resource, user security.UserGroup, removeApp bool)
error {
+func (m *Manager) DecreaseTrackedResource(queuePath, applicationID string,
usage *resources.Resource, user security.UserGroup, removeApp bool) bool {
log.Log(log.SchedUGM).Debug("Decreasing resource usage",
zap.String("user", user.User),
zap.String("queue path", queuePath),
zap.String("application", applicationID),
zap.Stringer("resource", usage),
zap.Bool("removeApp", removeApp))
if queuePath == "" || applicationID == "" || usage == nil || user.User
== "" {
- log.Log(log.SchedUGM).Error("Mandatory parameters are missing
to decrease the resource usage",
+ log.Log(log.SchedUGM).Debug("Mandatory parameters are missing
to decrease the resource usage",
zap.String("user", user.User),
zap.String("queue path", queuePath),
zap.String("application", applicationID),
zap.Stringer("resource", usage),
zap.Bool("removeApp", removeApp))
- return fmt.Errorf("mandatory parameters are missing. queuepath:
%s, application id: %s, resource usage: %s, user: %s",
- queuePath, applicationID, usage.String(), user.User)
+ return false
}
- m.Lock()
- defer m.Unlock()
- userTracker := m.userTrackers[user.User]
- if userTracker != nil {
- removeQT, err := userTracker.decreaseTrackedResource(queuePath,
applicationID, usage, removeApp)
- if err != nil {
- log.Log(log.SchedUGM).Error("Problem in decreasing the
user resource usage",
- zap.String("user", user.User),
- zap.String("queue path", queuePath),
- zap.String("application", applicationID),
- zap.Stringer("resource", usage),
- zap.Bool("removeApp", removeApp),
- zap.String("err message", err.Error()))
- return err
- }
- if removeApp && removeQT {
- delete(m.userTrackers, user.User)
- }
- } else {
+
+ userTracker := m.getUserTracker(user.User, false)
+ if userTracker == nil {
log.Log(log.SchedUGM).Error("user tracker must be available in
userTrackers map",
zap.String("user", user.User))
- return fmt.Errorf("user tracker for %s is missing in
userTrackers map", user.User)
+ return false
+ }
+ log.Log(log.SchedUGM).Debug("Decreasing resource usage for user",
+ zap.String("user", user.User),
+ zap.String("queue path", queuePath),
+ zap.String("application", applicationID),
+ zap.Stringer("resource", usage),
+ zap.Bool("removeApp", removeApp))
+ removeQT, decreased := userTracker.decreaseTrackedResource(queuePath,
applicationID, usage, removeApp)
+ if !decreased {
+ return decreased
+ }
+ if removeApp && removeQT {
+ log.Log(log.SchedUGM).Debug("Removing user from manager",
+ zap.String("user", user.User),
+ zap.String("queue path", queuePath),
+ zap.String("application", applicationID),
+ zap.Bool("removeApp", removeApp))
+ delete(m.userTrackers, user.User)
}
group, err := m.getGroup(user)
if err != nil {
- return err
+ return false
}
- groupTracker := m.groupTrackers[group]
- if groupTracker != nil {
- removeQT, err :=
groupTracker.decreaseTrackedResource(queuePath, applicationID, usage, removeApp)
- if err != nil {
- log.Log(log.SchedUGM).Error("Problem in decreasing the
group resource usage",
- zap.String("user", user.User),
- zap.String("group", group),
- zap.String("queue path", queuePath),
- zap.String("application", applicationID),
- zap.Stringer("resource", usage),
- zap.Bool("removeApp", removeApp),
- zap.String("err message", err.Error()))
- return err
- }
- if removeApp && removeQT {
- delete(m.groupTrackers, group)
- }
- } else {
+ groupTracker := m.getGroupTracker(group, false)
+ if groupTracker == nil {
log.Log(log.SchedUGM).Error("appGroupTrackers tracker must be
available in groupTrackers map",
zap.String("appGroupTrackers", group))
- return fmt.Errorf("appGroupTrackers tracker for %s is missing
in groupTrackers map", group)
+ return false
}
- return nil
+ log.Log(log.SchedUGM).Debug("Decreasing resource usage for group",
+ zap.String("group", group),
+ zap.String("queue path", queuePath),
+ zap.String("application", applicationID),
+ zap.Stringer("resource", usage),
+ zap.Bool("removeApp", removeApp))
+ removeQT, decreased = groupTracker.decreaseTrackedResource(queuePath,
applicationID, usage, removeApp)
+ if !decreased {
+ return decreased
+ }
+ if removeApp && removeQT {
+ log.Log(log.SchedUGM).Debug("Removing group from manager",
+ zap.String("group", group),
+ zap.String("queue path", queuePath),
+ zap.String("application", applicationID),
+ zap.Bool("removeApp", removeApp))
+ delete(m.groupTrackers, group)
+ }
+ return true
}
func (m *Manager) GetUserResources(user security.UserGroup)
*resources.Resource {
@@ -310,6 +250,8 @@ func (m *Manager) GetGroupTracker(group string)
*GroupTracker {
}
func (m *Manager) ensureGroupTrackerForApp(queuePath string, applicationID
string, user security.UserGroup) error {
+ m.Lock()
+ defer m.Unlock()
userTracker := m.userTrackers[user.User]
if !userTracker.hasGroupForApp(applicationID) {
var groupTracker *GroupTracker
@@ -318,66 +260,12 @@ func (m *Manager) ensureGroupTrackerForApp(queuePath
string, applicationID strin
return err
}
if m.groupTrackers[group] == nil {
- log.Log(log.SchedUGM).Debug("Group tracker does not
exist. Creating group tracker object and linking the same with application",
+ log.Log(log.SchedUGM).Debug("Group tracker doesn't
exists. Creating group tracker",
zap.String("application", applicationID),
zap.String("queue path", queuePath),
zap.String("user", user.User),
zap.String("group", group))
groupTracker = newGroupTracker(group)
-
- // Set the limits for all configured queue paths of the
group
- for configQueuePath, config := range
m.groupLimitsConfig[group] {
- log.Log(log.SchedUGM).Debug("Setting the limit
max applications settings.",
- zap.String("group", group),
- zap.String("queue path",
configQueuePath))
- maxApps, ok := config[maxapplications].(uint64)
- if !ok {
- log.Log(log.SchedUGM).Warn("Problem in
setting the limit max applications settings. Unable to cast the value from
interface to uint64",
- zap.String("group", group),
- zap.String("queue path",
configQueuePath),
- zap.Uint64("limit max
applications", maxApps))
- return fmt.Errorf("unable to set the
max applications. group: %s, queuepath : %s, applicationid: %s",
- group, configQueuePath,
applicationID)
- }
- if setMaxApplicationsErr :=
groupTracker.setMaxApplications(maxApps, configQueuePath);
setMaxApplicationsErr != nil {
- log.Log(log.SchedUGM).Warn("Problem in
setting the limit max applications settings.",
- zap.String("group", group),
- zap.String("queue path",
configQueuePath),
- zap.Uint64("limit max
applications", maxApps),
-
zap.Error(setMaxApplicationsErr))
- return fmt.Errorf("unable to set the
max applications. group: %s, queuepath : %s, applicationid: %s, reason: %w",
- group, configQueuePath,
applicationID, setMaxApplicationsErr)
- }
-
- maxResources, ok :=
config[maxresources].(map[string]string)
- if !ok {
- log.Log(log.SchedUGM).Warn("Problem in
setting the limit max resources settings. Unable to cast the value from
interface to resource",
- zap.String("group", group),
- zap.String("queue path",
configQueuePath),
- zap.Any("limit max resources",
maxResources))
- return fmt.Errorf("unable to set the
max resources. group: %s, queuepath : %s, applicationid: %s",
- group, configQueuePath,
applicationID)
- }
- resource, resourceErr :=
resources.NewResourceFromConf(maxResources)
- if resourceErr != nil {
- log.Log(log.SchedUGM).Warn("Problem in
setting the limit max resources settings.",
- zap.String("group", group),
- zap.String("queue path",
configQueuePath),
- zap.Any("limit max resources",
maxResources),
- zap.Error(resourceErr))
- return fmt.Errorf("unable to set the
max resources. group: %s, queuepath : %s, applicationid: %s, reason: %w",
- group, configQueuePath,
applicationID, resourceErr)
- }
- if setMaxResourcesErr :=
groupTracker.setMaxResources(resource, configQueuePath); setMaxResourcesErr !=
nil {
- log.Log(log.SchedUGM).Warn("Problem in
setting the limit max resources settings.",
- zap.String("group", group),
- zap.String("queue path",
configQueuePath),
- zap.Any("limit max resources",
maxResources),
- zap.Error(setMaxResourcesErr))
- return fmt.Errorf("unable to set the
max resources. group: %s, queuepath : %s, applicationid: %s, reason: %w",
- group, configQueuePath,
applicationID, setMaxResourcesErr)
- }
- }
m.groupTrackers[group] = groupTracker
} else {
log.Log(log.SchedUGM).Debug("Group tracker already
exists and linking (reusing) the same with application",
@@ -419,20 +307,27 @@ func (m *Manager) UpdateConfig(config
configs.QueueConfig, queuePath string) err
m.Lock()
defer m.Unlock()
- // clear the local limit config maps before processing the limit config
- m.userLimitsConfig = make(map[string]map[string]map[string]interface{})
- m.groupLimitsConfig = make(map[string]map[string]map[string]interface{})
+ m.userWildCardLimitsConfig = make(map[string]*LimitConfig)
+ m.groupWildCardLimitsConfig = make(map[string]*LimitConfig)
return m.internalProcessConfig(config, queuePath)
}
func (m *Manager) internalProcessConfig(cur configs.QueueConfig, queuePath
string) error {
// Holds user and group for which limits have been configured with
specific queue path
- userGroupLimits := make(map[string]bool)
+ userLimits := make(map[string]bool)
+ groupLimits := make(map[string]bool)
// Traverse limits of specific queue path
for _, limit := range cur.Limits {
- tempLimitsMap := make(map[string]interface{})
- tempLimitsMap[maxresources] = limit.MaxResources
- tempLimitsMap[maxapplications] = limit.MaxApplications
+ var maxResource *resources.Resource
+ var err error
+ if maxResource, err =
resources.NewResourceFromConf(limit.MaxResources); err != nil {
+ log.Log(log.SchedUGM).Warn("Problem in using the limit
max resources settings.",
+ zap.String("queue path", queuePath),
+ zap.Any("limit max resources",
limit.MaxResources),
+ zap.Error(err))
+ return fmt.Errorf("problem in using the max resources
settings for queuepath: %s. reason: %w", queuePath, err)
+ }
+ limitConfig := &LimitConfig{maxResources: maxResource,
maxApplications: limit.MaxApplications}
for _, user := range limit.Users {
log.Log(log.SchedUGM).Debug("Processing user limits
configuration",
zap.String("user", user),
@@ -440,9 +335,11 @@ 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))
- tempUserMap := make(map[string]map[string]interface{})
- tempUserMap[queuePath] = tempLimitsMap
- if err := m.processUserConfig(user, limit, queuePath,
userGroupLimits, tempLimitsMap, tempUserMap); err != nil {
+ if user == common.Wildcard {
+ m.userWildCardLimitsConfig[queuePath] =
limitConfig
+ continue
+ }
+ if err := m.processUserConfig(user, limitConfig,
queuePath, userLimits); err != nil {
return err
}
}
@@ -453,14 +350,16 @@ 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))
- tempGroupMap := make(map[string]map[string]interface{})
- tempGroupMap[queuePath] = tempLimitsMap
- if err := m.processGroupConfig(group, limit, queuePath,
userGroupLimits, tempLimitsMap, tempGroupMap); err != nil {
+ if group == common.Wildcard {
+ m.groupWildCardLimitsConfig[queuePath] =
limitConfig
+ continue
+ }
+ if err := m.processGroupConfig(group, limitConfig,
queuePath, groupLimits); err != nil {
return err
}
}
}
- if err := m.clearEarlierSetLimits(userGroupLimits, queuePath); err !=
nil {
+ if err := m.clearEarlierSetLimits(userLimits, groupLimits, queuePath);
err != nil {
return err
}
@@ -475,113 +374,111 @@ func (m *Manager) internalProcessConfig(cur
configs.QueueConfig, queuePath strin
return nil
}
-func (m *Manager) processUserConfig(user string, limit configs.Limit,
queuePath string, userGroupLimits map[string]bool, tempLimitsMap
map[string]interface{}, tempUserMap map[string]map[string]interface{}) error {
+func (m *Manager) processUserConfig(user string, limitConfig *LimitConfig,
queuePath string, userLimits map[string]bool) error {
if user == "*" {
// traverse all tracked users
for u, ut := range m.userTrackers {
// Is this user already tracked for the queue path?
- if m.IsQueuePathTrackedCompletely(ut.queueTracker,
queuePath) {
+ if ut.IsQueuePathTrackedCompletely(queuePath) {
log.Log(log.SchedUGM).Debug("Processing wild
card user limits configuration for all existing users",
zap.String("user", u),
- zap.String("limit", limit.Limit),
zap.String("queue path", queuePath),
- zap.Uint64("max application",
limit.MaxApplications),
- zap.Any("max resources",
limit.MaxResources))
-
- // creates an entry for the user being
processed always as it has been cleaned before
- if _, ok := m.userLimitsConfig[u]; ok {
- m.userLimitsConfig[u][queuePath] =
tempLimitsMap
- } else {
- m.userLimitsConfig[u] = tempUserMap
- }
- if err := m.setUserLimits(u, limit, queuePath);
err != nil {
+ zap.Uint64("max application",
limitConfig.maxApplications),
+ zap.Any("max resources",
limitConfig.maxResources))
+ if err := m.setUserLimits(u, limitConfig,
queuePath); err != nil {
return err
}
- userGroupLimits[u] = true
+ userLimits[u] = true
}
}
- } else {
- // creates an entry for the user being processed always as it
has been cleaned before
- if _, ok := m.userLimitsConfig[user]; ok {
- m.userLimitsConfig[user][queuePath] = tempLimitsMap
- } else {
- m.userLimitsConfig[user] = tempUserMap
- }
- if err := m.setUserLimits(user, limit, queuePath); err != nil {
+ } else if user != "" {
+ if err := m.setUserLimits(user, limitConfig, queuePath); err !=
nil {
return err
}
- userGroupLimits[user] = true
+ userLimits[user] = true
}
return nil
}
-func (m *Manager) processGroupConfig(group string, limit configs.Limit,
queuePath string, userGroupLimits map[string]bool, tempLimitsMap
map[string]interface{}, tempGroupMap map[string]map[string]interface{}) error {
+func (m *Manager) processGroupConfig(group string, limitConfig *LimitConfig,
queuePath string, groupLimits map[string]bool) error {
if group == "*" {
// traverse all tracked groups
for g, gt := range m.groupTrackers {
// Is this group already tracked for the queue path?
- if m.IsQueuePathTrackedCompletely(gt.queueTracker,
queuePath) {
+ if gt.IsQueuePathTrackedCompletely(queuePath) {
log.Log(log.SchedUGM).Debug("Processing wild
card user limits configuration for all existing groups",
zap.String("group", g),
- zap.String("limit", limit.Limit),
zap.String("queue path", queuePath),
- zap.Uint64("max application",
limit.MaxApplications),
- zap.Any("max resources",
limit.MaxResources))
- // creates an entry for the group being
processed always as it has been cleaned before
- if _, ok := m.groupLimitsConfig[g]; ok {
- m.groupLimitsConfig[g][queuePath] =
tempLimitsMap
- } else {
- m.groupLimitsConfig[g] = tempGroupMap
- }
- if err := m.setGroupLimits(g, limit,
queuePath); err != nil {
+ zap.Uint64("max application",
limitConfig.maxApplications),
+ zap.Any("max resources",
limitConfig.maxResources))
+ if err := m.setGroupLimits(g, limitConfig,
queuePath); err != nil {
return err
}
- userGroupLimits[g] = true
+ groupLimits[g] = true
}
}
- } else {
- // creates an entry for the group being processed always as it
has been cleaned before
- if _, ok := m.groupLimitsConfig[group]; ok {
- m.groupLimitsConfig[group][queuePath] = tempLimitsMap
- } else {
- m.groupLimitsConfig[group] = tempGroupMap
- }
- if err := m.setGroupLimits(group, limit, queuePath); err != nil
{
+ } else if group != "" {
+ if err := m.setGroupLimits(group, limitConfig, queuePath); err
!= nil {
return err
}
- userGroupLimits[group] = true
+ groupLimits[group] = true
}
return nil
}
-func (m *Manager) clearEarlierSetLimits(userGroupLimits map[string]bool,
queuePath string) error {
- // Clear already configured limits of user for which limits have been
configured before but not now through #cur
+// 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 user for which limits have been
configured before but not now
for u, ut := range m.userTrackers {
// Is this user already tracked for the queue path?
- if m.IsQueuePathTrackedCompletely(ut.queueTracker, queuePath) {
- if _, ok := userGroupLimits[u]; !ok {
- err :=
ut.setMaxResources(resources.NewResource(), queuePath)
- if err != nil {
- return err
+ if ut.IsQueuePathTrackedCompletely(queuePath) {
+ // 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))
}
- err = ut.setMaxApplications(0, queuePath)
- if err != nil {
- return err
+ // 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)
}
}
}
}
- // Clear already configured limits of group for which limits have been
configured before but not now through #cur
+ // Clear already configured limits of group for which limits have been
configured before but not now
for g, gt := range m.groupTrackers {
// Is this group already tracked for the queue path?
- if m.IsQueuePathTrackedCompletely(gt.queueTracker, queuePath) {
- if _, ok := userGroupLimits[g]; !ok {
- if err :=
gt.setMaxResources(resources.NewResource(), queuePath); err != nil {
- return err
+ if gt.IsQueuePathTrackedCompletely(queuePath) {
+ // 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))
+ // 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))
}
- if err := gt.setMaxApplications(0, queuePath);
err != nil {
- return err
+ // 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)
}
}
}
@@ -589,13 +486,12 @@ func (m *Manager) clearEarlierSetLimits(userGroupLimits
map[string]bool, queuePa
return nil
}
-func (m *Manager) setUserLimits(user string, limit configs.Limit, queuePath
string) error {
+func (m *Manager) setUserLimits(user string, limitConfig *LimitConfig,
queuePath string) error {
log.Log(log.SchedUGM).Debug("Setting user limits",
zap.String("user", user),
- zap.String("limit", limit.Limit),
zap.String("queue path", queuePath),
- zap.Uint64("max application", limit.MaxApplications),
- zap.Any("max resources", limit.MaxResources))
+ zap.Uint64("max application", limitConfig.maxApplications),
+ zap.Any("max resources", limitConfig.maxResources))
userTracker, ok := m.userTrackers[user]
if !ok {
log.Log(log.SchedUGM).Debug("User tracker does not exist.
Creating user tracker object to set the limit configuration",
@@ -604,42 +500,16 @@ func (m *Manager) setUserLimits(user string, limit
configs.Limit, queuePath stri
userTracker = newUserTracker(user)
m.userTrackers[user] = userTracker
}
- if err := userTracker.setMaxApplications(limit.MaxApplications,
queuePath); err != nil {
- log.Log(log.SchedUGM).Warn("Problem in setting the limit max
applications settings.",
- zap.String("user", user),
- zap.String("queue path", queuePath),
- zap.Uint64("limit max applications",
limit.MaxApplications),
- zap.Error(err))
- return fmt.Errorf("unable to set the limit for user %s because
%w", user, err)
- }
-
- if resource, err := resources.NewResourceFromConf(limit.MaxResources);
err == nil {
- if err = userTracker.setMaxResources(resource, queuePath); err
!= nil {
- log.Log(log.SchedUGM).Warn("Problem in setting the
limit max resources settings.",
- zap.String("user", user),
- zap.String("queue path", queuePath),
- zap.Any("limit max resources",
limit.MaxResources),
- zap.Error(err))
- return fmt.Errorf("unable to set the limit for user %s
because %w", user, err)
- }
- } else {
- log.Log(log.SchedUGM).Warn("Problem in using the limit max
resources settings.",
- zap.String("user", user),
- zap.String("queue path", queuePath),
- zap.Any("limit max resources", limit.MaxResources),
- zap.Error(err))
- return fmt.Errorf("unable to set the limit for user %s because
%w", user, err)
- }
+ userTracker.setLimits(queuePath, limitConfig.maxResources,
limitConfig.maxApplications)
return nil
}
-func (m *Manager) setGroupLimits(group string, limit configs.Limit, queuePath
string) error {
+func (m *Manager) setGroupLimits(group string, limitConfig *LimitConfig,
queuePath string) error {
log.Log(log.SchedUGM).Debug("Setting group limits",
zap.String("group", group),
- zap.String("limit", limit.Limit),
zap.String("queue path", queuePath),
- zap.Uint64("max application", limit.MaxApplications),
- zap.Any("max resources", limit.MaxResources))
+ zap.Uint64("max application", limitConfig.maxApplications),
+ zap.Any("max resources", limitConfig.maxResources))
groupTracker, ok := m.groupTrackers[group]
if !ok {
log.Log(log.SchedUGM).Debug("Group tracker does not exist.
Creating group tracker object to set the limit configuration",
@@ -648,46 +518,58 @@ func (m *Manager) setGroupLimits(group string, limit
configs.Limit, queuePath st
groupTracker = newGroupTracker(group)
m.groupTrackers[group] = groupTracker
}
- if err := groupTracker.setMaxApplications(limit.MaxApplications,
queuePath); err != nil {
- log.Log(log.SchedUGM).Warn("Problem in setting the limit max
applications settings.",
- zap.String("group", group),
- zap.String("queue path", queuePath),
- zap.Uint64("limit max applications",
limit.MaxApplications),
- zap.Error(err))
- return fmt.Errorf("unable to set the limit for group %s because
%w", group, err)
+ groupTracker.setLimits(queuePath, limitConfig.maxResources,
limitConfig.maxApplications)
+ return nil
+}
+
+func (m *Manager) getUserTracker(user string, createIfNotPresent bool)
*UserTracker {
+ m.Lock()
+ defer m.Unlock()
+ if ut, ok := m.userTrackers[user]; ok {
+ return ut
+ }
+ if createIfNotPresent {
+ log.Log(log.SchedUGM).Debug("User tracker doesn't exists.
Creating user tracker.",
+ zap.String("user", user))
+ userTracker := newUserTracker(user)
+ m.userTrackers[user] = userTracker
+ return userTracker
}
+ return nil
+}
- if resource, err := resources.NewResourceFromConf(limit.MaxResources);
err == nil {
- if err = groupTracker.setMaxResources(resource, queuePath); err
!= nil {
- log.Log(log.SchedUGM).Warn("Problem in setting the
limit max resources settings.",
- zap.String("group", group),
- zap.String("queue path", queuePath),
- zap.Any("limit max resources",
limit.MaxResources),
- zap.Error(err))
- return fmt.Errorf("unable to set the limit for group %s
because %w", group, err)
- }
- } else {
- log.Log(log.SchedUGM).Warn("Problem in using the limit max
resources settings.",
- zap.String("group", group),
- zap.String("queue path", queuePath),
- zap.Any("limit max resources", limit.MaxResources),
- zap.Error(err))
- return fmt.Errorf("unable to set the limit for group %s because
%w", group, err)
+func (m *Manager) getGroupTracker(group string, createIfNotPresent bool)
*GroupTracker {
+ m.Lock()
+ defer m.Unlock()
+ if gt, ok := m.groupTrackers[group]; ok {
+ return gt
+ }
+ if createIfNotPresent {
+ log.Log(log.SchedUGM).Debug("Group tracker doesn't exists.
Creating group tracker.",
+ zap.String("group", group))
+ groupTracker := newGroupTracker(group)
+ m.groupTrackers[group] = groupTracker
+ return groupTracker
}
return nil
}
-func (m *Manager) IsQueuePathTrackedCompletely(qt *QueueTracker, queuePath
string) bool {
- if queuePath == configs.RootQueue || queuePath == qt.queueName {
- return true
+func (m *Manager) getUserWildCardLimitsConfig(queuePath string) *LimitConfig {
+ m.RLock()
+ defer m.RUnlock()
+ if config, ok := m.userWildCardLimitsConfig[queuePath]; ok {
+ return config
}
- childQueuePath, immediateChildQueueName := getChildQueuePath(queuePath)
- if immediateChildQueueName != "" {
- if childUt, ok :=
qt.childQueueTrackers[immediateChildQueueName]; ok {
- return m.IsQueuePathTrackedCompletely(childUt,
childQueuePath)
- }
+ return nil
+}
+
+func (m *Manager) getGroupWildCardLimitsConfig(queuePath string) *LimitConfig {
+ m.RLock()
+ defer m.RUnlock()
+ if config, ok := m.groupWildCardLimitsConfig[queuePath]; ok {
+ return config
}
- return false
+ return nil
}
// ClearUserTrackers only for tests
diff --git a/pkg/scheduler/ugm/manager_test.go
b/pkg/scheduler/ugm/manager_test.go
index c8765553..c0bae9dd 100644
--- a/pkg/scheduler/ugm/manager_test.go
+++ b/pkg/scheduler/ugm/manager_test.go
@@ -59,12 +59,14 @@ func TestAddRemoveUserAndGroups(t *testing.T) {
}
manager := GetUserManager()
- err = manager.IncreaseTrackedResource("", "", usage1, user)
- assert.Error(t, err, "mandatory parameters are missing. queuepath: ,
application id: , resource usage: "+usage1.String()+", user: "+user.User)
+ increased := manager.IncreaseTrackedResource("", "", usage1, user)
+ if increased {
+ t.Errorf("mandatory parameters are missing. queuepath: ,
application id: , resource usage: %s, user: %s", usage1.String(), user.User)
+ }
- err = manager.IncreaseTrackedResource(queuePath1, TestApp1, usage1,
user)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage1, err)
+ increased = manager.IncreaseTrackedResource(queuePath1, TestApp1,
usage1, user)
+ if !increased {
+ t.Errorf("unable to increase tracked resource. queuepath: %s,
application id: %s, resource usage: %s, user: %s", queuePath1, TestApp1,
usage1.String(), user.User)
}
userTrackers := manager.GetUsersResources()
@@ -77,9 +79,9 @@ func TestAddRemoveUserAndGroups(t *testing.T) {
assert.Equal(t, user.User, manager.GetUserTracker(user.User).userName)
assert.Equal(t, user.Groups[0],
manager.GetGroupTracker(user.Groups[0]).groupName)
- err = manager.IncreaseTrackedResource(queuePath1, TestApp1, usage1,
user)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage1, err)
+ increased = manager.IncreaseTrackedResource(queuePath1, TestApp1,
usage1, user)
+ if !increased {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, usage1)
}
assertUGM(t, user, resources.Multiply(usage1, 2), 1)
@@ -88,9 +90,9 @@ func TestAddRemoveUserAndGroups(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage2)
}
- err = manager.IncreaseTrackedResource(queuePath2, TestApp2, usage2,
user1)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath2, TestApp2, usage2, err)
+ increased = manager.IncreaseTrackedResource(queuePath2, TestApp2,
usage2, user1)
+ if !increased {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath2, TestApp2, usage2)
}
assertUGM(t, user1, usage2, 2)
assert.Equal(t, user.User, manager.GetUserTracker(user.User).userName)
@@ -98,43 +100,43 @@ func TestAddRemoveUserAndGroups(t *testing.T) {
assert.Equal(t, user1.User, manager.GetUserTracker(user1.User).userName)
assert.Equal(t, user1.Groups[0],
manager.GetGroupTracker(user1.Groups[0]).groupName)
- assert.Equal(t, true,
manager.IsQueuePathTrackedCompletely(manager.GetUserTracker(user.User).queueTracker,
queuePath1))
- assert.Equal(t, true,
manager.IsQueuePathTrackedCompletely(manager.GetUserTracker(user1.User).queueTracker,
queuePath2))
- assert.Equal(t, false,
manager.IsQueuePathTrackedCompletely(manager.GetUserTracker(user1.User).queueTracker,
queuePath1))
- assert.Equal(t, false,
manager.IsQueuePathTrackedCompletely(manager.GetUserTracker(user.User).queueTracker,
queuePath2))
- assert.Equal(t, false,
manager.IsQueuePathTrackedCompletely(manager.GetUserTracker(user.User).queueTracker,
queuePath3))
- assert.Equal(t, false,
manager.IsQueuePathTrackedCompletely(manager.GetUserTracker(user.User).queueTracker,
queuePath4))
+ assert.Equal(t, true,
manager.GetUserTracker(user.User).queueTracker.IsQueuePathTrackedCompletely(queuePath1))
+ assert.Equal(t, true,
manager.GetUserTracker(user1.User).queueTracker.IsQueuePathTrackedCompletely(queuePath2))
+ assert.Equal(t, false,
manager.GetUserTracker(user1.User).queueTracker.IsQueuePathTrackedCompletely(queuePath1))
+ assert.Equal(t, false,
manager.GetUserTracker(user.User).queueTracker.IsQueuePathTrackedCompletely(queuePath2))
+ assert.Equal(t, false,
manager.GetUserTracker(user.User).queueTracker.IsQueuePathTrackedCompletely(queuePath3))
+ assert.Equal(t, false,
manager.GetUserTracker(user.User).queueTracker.IsQueuePathTrackedCompletely(queuePath4))
- assert.Equal(t, true,
manager.IsQueuePathTrackedCompletely(manager.GetUserTracker(user.Groups[0]).queueTracker,
queuePath1))
- assert.Equal(t, true,
manager.IsQueuePathTrackedCompletely(manager.GetUserTracker(user1.Groups[0]).queueTracker,
queuePath2))
- assert.Equal(t, false,
manager.IsQueuePathTrackedCompletely(manager.GetUserTracker(user1.Groups[0]).queueTracker,
queuePath1))
- assert.Equal(t, false,
manager.IsQueuePathTrackedCompletely(manager.GetUserTracker(user.Groups[0]).queueTracker,
queuePath2))
- assert.Equal(t, false,
manager.IsQueuePathTrackedCompletely(manager.GetUserTracker(user.Groups[0]).queueTracker,
queuePath3))
- assert.Equal(t, false,
manager.IsQueuePathTrackedCompletely(manager.GetUserTracker(user.Groups[0]).queueTracker,
queuePath4))
+ assert.Equal(t, true,
manager.GetUserTracker(user.Groups[0]).queueTracker.IsQueuePathTrackedCompletely(queuePath1))
+ assert.Equal(t, true,
manager.GetUserTracker(user1.Groups[0]).queueTracker.IsQueuePathTrackedCompletely(queuePath2))
+ assert.Equal(t, false,
manager.GetUserTracker(user1.Groups[0]).queueTracker.IsQueuePathTrackedCompletely(queuePath1))
+ assert.Equal(t, false,
manager.GetUserTracker(user.Groups[0]).queueTracker.IsQueuePathTrackedCompletely(queuePath2))
+ assert.Equal(t, false,
manager.GetUserTracker(user.Groups[0]).queueTracker.IsQueuePathTrackedCompletely(queuePath3))
+ assert.Equal(t, false,
manager.GetUserTracker(user.Groups[0]).queueTracker.IsQueuePathTrackedCompletely(queuePath4))
usage3, err := resources.NewResourceFromConf(map[string]string{"mem":
"5M", "vcore": "5"})
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage3)
}
- err = manager.DecreaseTrackedResource("", "", usage1, user, false)
- assert.Error(t, err, "mandatory parameters are missing. queuepath: ,
application id: , resource usage: "+usage1.String()+", user: "+user.User)
+ decreased := manager.DecreaseTrackedResource("", "", usage1, user,
false)
+ assert.Equal(t, decreased, false)
- err = manager.DecreaseTrackedResource(queuePath1, TestApp1, usage3,
user, false)
- if err != nil {
+ decreased = manager.DecreaseTrackedResource(queuePath1, TestApp1,
usage3, user, false)
+ if !decreased {
t.Fatalf("unable to decrease tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage3, err)
}
assertUGM(t, user, usage1, 2)
- err = manager.DecreaseTrackedResource(queuePath1, TestApp1, usage3,
user, true)
- if err != nil {
+ decreased = manager.DecreaseTrackedResource(queuePath1, TestApp1,
usage3, user, true)
+ if !decreased {
t.Fatalf("unable to decrease tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage3, err)
}
assert.Equal(t, 1, len(manager.GetUsersResources()), "userTrackers
count should be 1")
assert.Equal(t, 1, len(manager.GetGroupsResources()), "groupTrackers
count should be 1")
- err = manager.DecreaseTrackedResource(queuePath2, TestApp2, usage2,
user1, true)
- if err != nil {
+ decreased = manager.DecreaseTrackedResource(queuePath2, TestApp2,
usage2, user1, true)
+ if !decreased {
t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath2, TestApp2, usage2, err)
}
assert.Equal(t, 0, len(manager.GetUsersResources()), "userTrackers
count should be 0")
@@ -144,124 +146,301 @@ func TestAddRemoveUserAndGroups(t *testing.T) {
assert.Assert(t, manager.GetGroupTracker(user.Groups[0]) == nil)
}
-func TestIncreaseUserResourceWithInvalidConfig(t *testing.T) {
+func TestUpdateConfig(t *testing.T) {
setupUGM()
// Queue setup:
// root->parent
user := security.UserGroup{User: "user1", Groups: []string{"group1"}}
manager := GetUserManager()
- expectedResource, err :=
resources.NewResourceFromConf(map[string]string{"memory": "5", "vcores": "5"})
+ expectedResource, err :=
resources.NewResourceFromConf(map[string]string{"memory": "50", "vcores": "50"})
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, expectedResource)
}
- m.userLimitsConfig[user.User] = make(map[string]map[string]interface{})
- m.userLimitsConfig[user.User][queuePath1] = make(map[string]interface{})
- m.userLimitsConfig[user.User][queuePath1][maxapplications] = -2
- assert.Error(t, manager.IncreaseTrackedResource(queuePath1, TestApp1,
expectedResource, user), "unable to set the max applications. user:
"+user.User+", queuepath : "+queuePath1+", applicationid: "+TestApp1)
-
- m.userLimitsConfig[user.User][queuePath1][maxapplications] = uint64(2)
- m.userLimitsConfig[user.User][queuePath1][maxresources] =
make(map[string]interface{})
- assert.Error(t, manager.IncreaseTrackedResource(queuePath1, TestApp1,
expectedResource, user), "unable to set the max resources. user: "+user.User+",
queuepath : "+queuePath1+", applicationid: "+TestApp1)
-
- m.userLimitsConfig[user.User][queuePath1][maxapplications] = uint64(2)
- m.userLimitsConfig[user.User][queuePath1][maxresources] =
map[string]string{"invalid": "-5", "vcores": "5"}
- assert.Error(t, manager.IncreaseTrackedResource(queuePath1, TestApp1,
expectedResource, user), "unable to set the max resources. user: "+user.User+",
queuepath : "+queuePath1+", applicationid: "+TestApp1+", usage: map[memory:5
vcores:5], reason: invalid quantity")
-
- m.userLimitsConfig[user.User][queuePath1][maxapplications] = uint64(2)
- m.userLimitsConfig[user.User][queuePath1][maxresources] =
map[string]string{"memory": "5", "vcores": "5"}
- m.groupLimitsConfig[user.Groups[0]] =
make(map[string]map[string]interface{})
- m.groupLimitsConfig[user.Groups[0]][queuePath1] =
make(map[string]interface{})
- m.groupLimitsConfig[user.Groups[0]][queuePath1][maxapplications] = -2
- assert.Error(t, manager.IncreaseTrackedResource(queuePath1, TestApp1,
expectedResource, user), "unable to set the max applications. group:
"+user.Groups[0]+", queuepath : "+queuePath1+", applicationid: "+TestApp1)
-
- m.userLimitsConfig[user.User][queuePath1][maxapplications] = uint64(2)
- m.userLimitsConfig[user.User][queuePath1][maxresources] =
map[string]string{"memory": "5", "vcores": "5"}
- m.groupLimitsConfig[user.Groups[0]][queuePath1][maxapplications] =
uint64(2)
- m.groupLimitsConfig[user.Groups[0]][queuePath1][maxresources] =
make(map[string]interface{})
- assert.Error(t, manager.IncreaseTrackedResource(queuePath1, TestApp1,
expectedResource, user), "unable to set the max resources. group:
"+user.Groups[0]+", queuepath : "+queuePath1+", applicationid: "+TestApp1)
-
- m.userLimitsConfig[user.User][queuePath1][maxapplications] = uint64(2)
- m.userLimitsConfig[user.User][queuePath1][maxresources] =
map[string]string{"memory": "5", "vcores": "5"}
- m.groupLimitsConfig[user.Groups[0]][queuePath1][maxapplications] =
uint64(2)
- m.groupLimitsConfig[user.Groups[0]][queuePath1][maxresources] =
map[string]string{"invalid": "-5", "vcores": "5"}
- assert.Error(t, manager.IncreaseTrackedResource(queuePath1, TestApp1,
expectedResource, user), "unable to set the max resources. group:
"+user.Groups[0]+", queuepath : "+queuePath1+", applicationid: "+TestApp1+",
reason: invalid quantity")
+ conf := createConfig(user.User, user.Groups[0], "memory", "50", 50, 5)
+ assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
+
+ usage, err := resources.NewResourceFromConf(map[string]string{"memory":
"10", "vcores": "10"})
+ if err != nil {
+ t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage)
+ }
+ assertMaxLimits(t, user, expectedResource, 5)
+
+ for i := 1; i <= 5; i++ {
+ increased := manager.IncreaseTrackedResource(queuePath1,
TestApp1, usage, user)
+ if !increased {
+ t.Fatalf("unable to increase tracked resource:
queuepath %s, app %s, res %v, error %t", queuePath1, TestApp1, usage, err)
+ }
+ }
+ // configure max resource for root.parent lesser than current resource
usage. should be allowed to set but user cannot be allowed to do any activity
further
+ conf = createConfig(user.User, user.Groups[0], "memory", "50", 40, 4)
+ err = manager.UpdateConfig(conf.Queues[0], "root")
+ assert.NilError(t, err)
+ increased := manager.IncreaseTrackedResource(queuePath1, TestApp1,
usage, user)
+ if increased {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, user)
+ }
+
+ // configure max resource for root and parent to allow one more
application to run
+ conf = createConfig(user.User, user.Groups[0], "memory", "50", 60, 6)
+ err = manager.UpdateConfig(conf.Queues[0], "root")
+ assert.NilError(t, err, "unable to set the limit for user user1 because
current resource usage is greater than config max resource for root.parent")
+
+ increased = manager.IncreaseTrackedResource(queuePath1, TestApp1,
usage, user)
+ if !increased {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, user)
+ }
+
+ // configure max resource for root lesser than current resource usage.
should be allowed to set but user cannot be allowed to do any activity further
+ conf = createConfig(user.User, user.Groups[0], "memory", "50", 10, 10)
+ err = manager.UpdateConfig(conf.Queues[0], "root")
+ assert.NilError(t, err)
+ increased = manager.IncreaseTrackedResource(queuePath1, TestApp1,
usage, user)
+ if increased {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, user)
+ }
}
-func TestUpdateConfig(t *testing.T) {
+func TestUpdateConfigWithWildCardUsersAndGroups(t *testing.T) {
setupUGM()
// Queue setup:
// root->parent
user := security.UserGroup{User: "user1", Groups: []string{"group1"}}
+ conf := createUpdateConfig(user.User, user.Groups[0])
manager := GetUserManager()
+ assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
- expectedResource, err :=
resources.NewResourceFromConf(map[string]string{"memory": "5", "vcores": "5"})
+ expectedResource, err :=
resources.NewResourceFromConf(map[string]string{"memory": "50", "vcores": "50"})
+ if err != nil {
+ t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, expectedResource)
+ }
+ usage, err := resources.NewResourceFromConf(map[string]string{"memory":
"10", "vcores": "10"})
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, expectedResource)
}
+ assertMaxLimits(t, user, expectedResource, 5)
+
+ for i := 1; i <= 5; i++ {
+ increased := manager.IncreaseTrackedResource(queuePath1,
TestApp1, usage, user)
+ if !increased {
+ t.Fatalf("unable to increase tracked resource:
queuepath %s, app %s, res %v", queuePath1, TestApp1, expectedResource)
+ }
+ }
- conf := createConfig(user.User, user.Groups[0], "memory", "-10")
- assert.Error(t, manager.UpdateConfig(conf.Queues[0], "root"), "unable
to set the limit for user user1 because invalid quantity")
- conf = createConfig(user.User, user.Groups[0], "invalid", "invalidate")
- assert.Error(t, manager.UpdateConfig(conf.Queues[0], "root"), "unable
to set the limit for user user1 because invalid quantity")
- conf = createUpdateConfig(user.User, user.Groups[0])
+ // configure max resource for root.parent to allow one more application
to run
+ conf = createConfig(user.User, user.Groups[0], "memory", "50", 60, 6)
+ err = manager.UpdateConfig(conf.Queues[0], "root")
+ assert.NilError(t, err, "unable to set the limit for user user1 because
current resource usage is greater than config max resource for root.parent")
- manager = GetUserManager()
+ // 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, TestApp1,
usage, user)
+ if !increased {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, user)
+ }
+
+ // 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, TestApp1,
usage, user)
+ if increased {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, user)
+ }
+
+ // configure max resource for root.parent to allow one more application
to run through wild card user settings (not through specific user)
+ user1 := security.UserGroup{User: "user2", Groups: []string{"group2"}}
+ conf = createUpdateConfigWithWildCardUsersAndGroups(user1.User,
user1.Groups[0], "*", "*", "70", "70")
assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
- assertMaxLimits(t, user, expectedResource, 1)
+ // should run as wild card user '*' setting is map[memory:70 vcores:70]
and total usage of "root.parent" is map[memory:60 vcores:60]
+ increased = manager.IncreaseTrackedResource(queuePath1, TestApp1,
usage, user)
+ if !increased {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, user)
+ }
- for i := 1; i <= 2; i++ {
- err = manager.IncreaseTrackedResource(queuePath1, TestApp1,
expectedResource, user)
- if err != nil {
- t.Fatalf("unable to increase tracked resource:
queuepath %s, app %s, res %v, error %t", queuePath1, TestApp1,
expectedResource, err)
+ // should not run as wild card user '*' setting is map[memory:70
vcores:70] and total usage of "root.parent" is map[memory:70 vcores:70]
+ increased = manager.IncreaseTrackedResource(queuePath1, TestApp1,
usage, user)
+ if increased {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, user)
+ }
+
+ // configure max resource for root.parent to allow one more application
to run through wild card group settings (not through specific group)
+ // also wild card user limit settings not set
+ conf = createUpdateConfigWithWildCardUsersAndGroups(user1.User,
user1.Groups[0], "", "*", "80", "80")
+ assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
+
+ // should run as wild card group '*' setting is map[memory:80
vcores:80] and total usage of "root.parent" is map[memory:70 vcores:70]
+ increased = manager.IncreaseTrackedResource(queuePath1, TestApp1,
usage, user)
+ if !increased {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, user)
+ }
+
+ // should not run as wild card group '*' setting is map[memory:80
vcores:80] and total usage of "root.parent" is map[memory:80 vcores:80]
+ increased = manager.IncreaseTrackedResource(queuePath1, TestApp1,
usage, user)
+ if increased {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, user)
+ }
+
+ // should run as wild card group '*' setting is map[memory:80
vcores:80] and total usage of "root.parent" 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)
+ }
+
+ // 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 overrides the wild card user limit
settings
+ conf = createUpdateConfigWithWildCardUsersAndGroups(user1.User,
user1.Groups[0], "", "*", "10", "10")
+ assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
+
+ // can be allowed to run upto resource usage map[memory:70 vcores:70]
+ for i := 1; i <= 6; 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, user)
}
}
- assert.Error(t, manager.UpdateConfig(conf.Queues[0], "root"), "unable
to set the limit for user user1 because current resource usage is greater than
config max resource for root.parent")
- err = manager.IncreaseTrackedResource(queuePath1, TestApp1,
expectedResource, user)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, expectedResource, err)
+ // should not run 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, user)
}
- assert.Error(t, manager.UpdateConfig(conf.Queues[0], "root"), "unable
to set the limit for user user1 because current resource usage is greater than
config max resource for root")
}
-func TestUpdateConfigWithWildCardUsersAndGroups(t *testing.T) {
+func TestUpdateConfigClearEarlierSetLimits(t *testing.T) {
setupUGM()
// Queue setup:
// root->parent
user := security.UserGroup{User: "user1", Groups: []string{"group1"}}
conf := createUpdateConfig(user.User, user.Groups[0])
+
manager := GetUserManager()
assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
- expectedResource, err :=
resources.NewResourceFromConf(map[string]string{"memory": "5", "vcores": "5"})
+ expectedResource, err :=
resources.NewResourceFromConf(map[string]string{"memory": "50", "vcores": "50"})
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, expectedResource)
}
- assertMaxLimits(t, user, expectedResource, 1)
+ assertMaxLimits(t, user, expectedResource, 5)
- for i := 1; i <= 2; i++ {
- err = manager.IncreaseTrackedResource(queuePath1, TestApp1,
expectedResource, user)
- if err != nil {
- t.Fatalf("unable to increase tracked resource:
queuepath %s, app %s, res %v, error %t", queuePath1, TestApp1,
expectedResource, err)
- }
+ // create config user2 * root.parent with [50, 50] and maxapps as 5
(twice for root), but not user1. so user1 should not be there as it doesn't
have any running applications
+ user1 := security.UserGroup{User: "user2", Groups: []string{"group2"}}
+ conf = createUpdateConfig(user1.User, user1.Groups[0])
+ assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
+
+ expectedResource, err =
resources.NewResourceFromConf(map[string]string{"memory": "50", "vcores": "50"})
+ if err != nil {
+ t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, expectedResource)
}
- assert.Error(t, manager.UpdateConfig(conf.Queues[0], "root"), "unable
to set the limit for user user1 because current resource usage is greater than
config max resource for root.parent")
- err = manager.IncreaseTrackedResource(queuePath1, TestApp1,
expectedResource, user)
+ // ensure user1 has been removed from local maps
+ assert.Equal(t, manager.GetUserTracker(user.User) == nil, true)
+ assert.Equal(t, manager.GetGroupTracker(user.Groups[0]) == nil, true)
+ assertMaxLimits(t, user1, expectedResource, 5)
+
+ // override user2 * root.parent config with [60, 60] and maxapps as 6
(twice for root)
+ conf = createConfig(user1.User, user1.Groups[0], "memory", "10", 60, 6)
+ assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
+
+ expectedResource, err =
resources.NewResourceFromConf(map[string]string{"memory": "60", "vcores": "60"})
if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, expectedResource, err)
+ t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, expectedResource)
}
- assert.Error(t, manager.UpdateConfig(conf.Queues[0], "root"), "unable
to set the limit for user user1 because current resource usage is greater than
config max resource for root")
+ assert.Equal(t, manager.GetUserTracker(user.User) == nil, true)
+ assert.Equal(t, manager.GetGroupTracker(user.Groups[0]) == nil, true)
+ assertMaxLimits(t, user1, expectedResource, 6)
- user1 := security.UserGroup{User: "user2", Groups: []string{"group2"}}
- conf = createUpdateConfigWithWildCardUsersAndGroups(user1.User,
user1.Groups[0])
+ expectedResource, err =
resources.NewResourceFromConf(map[string]string{"memory": "70", "vcores": "70"})
+ if err != nil {
+ t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, expectedResource)
+ }
+
+ // override user2 * root.parent maxapps as [70, 70] and maxapps as 10
(twice for root)
+ // and wild card settings
+ conf = createUpdateConfigWithWildCardUsersAndGroups(user1.User,
user1.Groups[0], "*", "*", "10", "10")
+ assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
+ assert.Equal(t, manager.GetUserTracker(user.User) == nil, true)
+ assert.Equal(t, manager.GetGroupTracker(user.Groups[0]) == nil, true)
+ assertMaxLimits(t, user1, expectedResource, 10)
+ assertWildCardLimits(t, m.userWildCardLimitsConfig, expectedResource)
+ assertWildCardLimits(t, m.groupWildCardLimitsConfig, expectedResource)
+
+ print("ggg")
+ // config without limits - should clear all earlier set configs
+ conf = createConfigWithoutLimits()
+ assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
+ assert.Equal(t, manager.GetUserTracker(user.User) == nil, true)
+ assert.Equal(t, manager.GetGroupTracker(user.Groups[0]) == nil, true)
+ assert.Equal(t, manager.GetUserTracker(user1.User) == nil, true)
+ assert.Equal(t, manager.GetGroupTracker(user1.Groups[0]) == nil, true)
+ assert.Equal(t, len(manager.userWildCardLimitsConfig), 0)
+ assert.Equal(t, len(manager.groupWildCardLimitsConfig), 0)
+}
+
+func TestSetMaxLimitsForRemovedUsers(t *testing.T) {
+ setupUGM()
+ // Queue setup:
+ // root->parent
+ user := security.UserGroup{User: "user1", Groups: []string{"group1"}}
+ conf := createUpdateConfig(user.User, user.Groups[0])
+ manager := GetUserManager()
+ assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
+
+ expectedResource, err :=
resources.NewResourceFromConf(map[string]string{"memory": "50", "vcores": "50"})
+ if err != nil {
+ t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, expectedResource)
+ }
+ usage, err := resources.NewResourceFromConf(map[string]string{"memory":
"10", "vcores": "10"})
+ if err != nil {
+ t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage)
+ }
+ assertMaxLimits(t, user, expectedResource, 5)
+
+ for i := 1; i <= 2; i++ {
+ increased := manager.IncreaseTrackedResource(queuePath1,
TestApp1, usage, user)
+ assert.Equal(t, increased, true, "unable to increase tracked
resource: queuepath "+queuePath1+", app "+TestApp1+", res "+usage.String())
+ }
+ assert.Equal(t, manager.GetUserTracker(user.User) != nil, true)
+ assert.Equal(t, manager.GetGroupTracker(user.Groups[0]) != nil, true)
+
+ decreased := manager.DecreaseTrackedResource(queuePath1, TestApp1,
usage, user, false)
+ assert.Equal(t, decreased, true)
+
+ decreased = manager.DecreaseTrackedResource(queuePath1, TestApp1,
usage, user, true)
+ assert.Equal(t, decreased, true)
+ assert.Equal(t, manager.GetUserTracker(user.User) != nil, true)
+ assert.Equal(t, manager.GetGroupTracker(user.Groups[0]) != nil, true)
+
+ increased := manager.IncreaseTrackedResource("root.parent.leaf",
TestApp1, usage, user)
+ assert.Equal(t, increased, true, "unable to increase tracked resource:
queuepath root.parent.leaf, app "+TestApp1+", res "+usage.String())
+
+ increased = manager.IncreaseTrackedResource("root.parent.leaf",
TestApp1, usage, user)
+ assert.Equal(t, increased, false, "unable to increase tracked resource:
queuepath root.parent.leaf, app "+TestApp1+", res "+usage.String())
+
+ assert.Equal(t, manager.GetUserTracker(user.User) != nil, true)
+ assert.Equal(t, manager.GetGroupTracker(user.Groups[0]) != nil, true)
+
+ decreased = manager.DecreaseTrackedResource("root.parent.leaf",
TestApp1, usage, user, true)
+ assert.Equal(t, decreased, true, "unable to decrease tracked resource:
queuepath root.parent.leaf, app "+TestApp1+", res "+usage.String())
+
+ conf = createConfigWithoutLimits()
assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
+
+ for i := 1; i <= 2; i++ {
+ increased := manager.IncreaseTrackedResource(queuePath1,
TestApp1, usage, user)
+ assert.Equal(t, increased, true, "unable to increase tracked
resource: queuepath "+queuePath1+", app "+TestApp1+", res "+usage.String())
+ }
+ assert.Equal(t, manager.GetUserTracker(user.User) != nil, true)
+ assert.Equal(t, manager.GetGroupTracker(user.Groups[0]) != nil, true)
+
+ decreased = manager.DecreaseTrackedResource(queuePath1, TestApp1,
usage, user, false)
+ assert.Equal(t, decreased, true)
+
+ decreased = manager.DecreaseTrackedResource(queuePath1, TestApp1,
usage, user, true)
+ assert.Equal(t, decreased, true)
+ assert.Equal(t, manager.GetUserTracker(user.User) == nil, true)
+ assert.Equal(t, manager.GetGroupTracker(user.Groups[0]) == nil, true)
}
-func createUpdateConfigWithWildCardUsersAndGroups(user string, group string)
configs.PartitionConfig {
+func createUpdateConfigWithWildCardUsersAndGroups(user string, group string,
wildUser string, wildGroup string, memory string, vcores string)
configs.PartitionConfig {
conf := configs.PartitionConfig{
Name: "test",
Queues: []configs.QueueConfig{
@@ -277,16 +456,30 @@ func createUpdateConfigWithWildCardUsersAndGroups(user
string, group string) con
Queues: nil,
Limits: []configs.Limit{
{
- Limit: "root
queue limit",
+ Limit: "parent
queue limit for specific user",
+ Users: []string{
+ user,
+ },
+ Groups:
[]string{
+ group,
+ },
+ MaxResources:
map[string]string{
+
"memory": "70",
+
"vcores": "70",
+ },
+
MaxApplications: 10,
+ },
+ {
+ Limit: "parent
queue limit for wild card user",
Users: []string{
- user,
"*",
+
wildUser,
},
Groups:
[]string{
- group,
"*",
+
wildGroup,
},
MaxResources:
map[string]string{
-
"memory": "50",
-
"vcores": "50",
+
"memory": memory,
+
"vcores": vcores,
},
MaxApplications: 10,
},
@@ -297,14 +490,14 @@ func createUpdateConfigWithWildCardUsersAndGroups(user
string, group string) con
{
Limit: "root queue limit",
Users: []string{
- user, "*",
+ user, wildUser,
},
Groups: []string{
- group, "*",
+ group, wildGroup,
},
MaxResources: map[string]string{
- "memory": "100",
- "vcores": "100",
+ "memory": "140",
+ "vcores": "140",
},
MaxApplications: 20,
},
@@ -316,10 +509,10 @@ func createUpdateConfigWithWildCardUsersAndGroups(user
string, group string) con
}
func createUpdateConfig(user string, group string) configs.PartitionConfig {
- return createConfig(user, group, "memory", "10")
+ return createConfig(user, group, "memory", "10", 50, 5)
}
-func createConfig(user string, group string, resourceKey string, resourceValue
string) configs.PartitionConfig {
+func createConfig(user string, group string, resourceKey string, resourceValue
string, mem int, maxApps uint64) configs.PartitionConfig {
conf := configs.PartitionConfig{
Name: "test",
Queues: []configs.QueueConfig{
@@ -332,10 +525,33 @@ func createConfig(user string, group string, resourceKey
string, resourceValue s
Name: "parent",
Parent: true,
SubmitACL: "*",
- Queues: nil,
+ Queues: []configs.QueueConfig{
+ {
+ Name:
"leaf",
+ Parent:
false,
+ SubmitACL: "*",
+ Queues: nil,
+ Limits:
[]configs.Limit{
+ {
+
Limit: "leaf queue limit",
+
Users: []string{
+
user,
+
},
+
Groups: []string{
+
group,
+
},
+
MaxResources: map[string]string{
+
resourceKey: resourceValue,
+
"vcores": "10",
+
},
+
MaxApplications: maxApps,
+ },
+ },
+ },
+ },
Limits: []configs.Limit{
{
- Limit: "root
queue limit",
+ Limit: "parent
queue limit",
Users: []string{
user,
},
@@ -343,10 +559,10 @@ func createConfig(user string, group string, resourceKey
string, resourceValue s
group,
},
MaxResources:
map[string]string{
-
"memory": "5",
-
"vcores": "5",
+
"memory": strconv.Itoa(mem),
+
"vcores": strconv.Itoa(mem),
},
-
MaxApplications: 1,
+
MaxApplications: maxApps,
},
},
},
@@ -361,10 +577,10 @@ func createConfig(user string, group string, resourceKey
string, resourceValue s
group,
},
MaxResources: map[string]string{
- resourceKey:
resourceValue,
- "vcores": "10",
+ "memory":
strconv.Itoa(mem * 2),
+ "vcores":
strconv.Itoa(mem * 2),
},
- MaxApplications: 2,
+ MaxApplications: maxApps * 2,
},
},
},
@@ -373,76 +589,26 @@ func createConfig(user string, group string, resourceKey
string, resourceValue s
return conf
}
-func TestUpdateConfigClearEarlierSetLimits(t *testing.T) {
- setupUGM()
- // Queue setup:
- // root->parent
- user := security.UserGroup{User: "user1", Groups: []string{"group1"}}
- conf := createUpdateConfig(user.User, user.Groups[0])
-
- manager := GetUserManager()
- assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
-
- expectedResource, err :=
resources.NewResourceFromConf(map[string]string{"memory": "5", "vcores": "5"})
- if err != nil {
- t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, expectedResource)
- }
- assertMaxLimits(t, user, expectedResource, 1)
-
- user1 := security.UserGroup{User: "user2", Groups: []string{"group2"}}
- conf = createUpdateConfig(user1.User, user1.Groups[0])
- assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
-
- expectedResource, err =
resources.NewResourceFromConf(map[string]string{"memory": "5", "vcores": "5"})
- if err != nil {
- t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, expectedResource)
- }
- assertMaxLimits(t, user, resources.NewResource(), 0)
- assertMaxLimits(t, user1, expectedResource, 1)
-}
-
-func TestSetMaxLimitsForRemovedUsers(t *testing.T) {
- setupUGM()
- // Queue setup:
- // root->parent
- user := security.UserGroup{User: "user1", Groups: []string{"group1"}}
- conf := createUpdateConfig(user.User, user.Groups[0])
- manager := GetUserManager()
- assert.NilError(t, manager.UpdateConfig(conf.Queues[0], "root"))
-
- expectedResource, err :=
resources.NewResourceFromConf(map[string]string{"memory": "5", "vcores": "5"})
- if err != nil {
- t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, expectedResource)
- }
- assertMaxLimits(t, user, expectedResource, 1)
-
- for i := 1; i <= 2; i++ {
- err = manager.IncreaseTrackedResource(queuePath1, TestApp1,
expectedResource, user)
- if err != nil {
- t.Fatalf("unable to increase tracked resource:
queuepath %s, app %s, res %v, error %t", queuePath1, TestApp1,
expectedResource, err)
- }
- }
- assert.Equal(t, manager.GetUserTracker(user.User) != nil, true)
- assert.Equal(t, manager.GetGroupTracker(user.Groups[0]) != nil, true)
-
- err = manager.DecreaseTrackedResource(queuePath1, TestApp1,
expectedResource, user, false)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, expectedResource, err)
- }
- err = manager.DecreaseTrackedResource(queuePath1, TestApp1,
expectedResource, user, true)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, expectedResource, err)
- }
- assert.Equal(t, manager.GetUserTracker(user.User) == nil, true)
- assert.Equal(t, manager.GetGroupTracker(user.Groups[0]) == nil, true)
-
- err = manager.IncreaseTrackedResource(queuePath1, TestApp1,
expectedResource, user)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, expectedResource, err)
+func createConfigWithoutLimits() configs.PartitionConfig {
+ conf := configs.PartitionConfig{
+ Name: "test",
+ Queues: []configs.QueueConfig{
+ {
+ Name: "root",
+ Parent: true,
+ SubmitACL: "*",
+ Queues: []configs.QueueConfig{
+ {
+ Name: "parent",
+ Parent: true,
+ SubmitACL: "*",
+ Queues: nil,
+ },
+ },
+ },
+ },
}
- assert.Equal(t, manager.GetUserTracker(user.User) != nil, true)
- assert.Equal(t, manager.GetGroupTracker(user.Groups[0]) != nil, true)
- assertMaxLimits(t, user, expectedResource, 1)
+ return conf
}
func setupUGM() {
@@ -465,10 +631,23 @@ func assertMaxLimits(t *testing.T, userGroup
security.UserGroup, expectedResourc
manager := GetUserManager()
assert.Equal(t,
manager.GetUserTracker(userGroup.User).queueTracker.maxRunningApps,
uint64(expectedMaxApps*2))
assert.Equal(t,
manager.GetGroupTracker(userGroup.Groups[0]).queueTracker.maxRunningApps,
uint64(expectedMaxApps*2))
- assert.Equal(t,
resources.Equals(manager.GetUserTracker(userGroup.User).queueTracker.maxResourceUsage,
resources.Multiply(expectedResource, 2)), true)
- assert.Equal(t,
resources.Equals(manager.GetGroupTracker(userGroup.Groups[0]).queueTracker.maxResourceUsage,
resources.Multiply(expectedResource, 2)), true)
+ assert.Equal(t,
resources.Equals(manager.GetUserTracker(userGroup.User).queueTracker.maxResources,
resources.Multiply(expectedResource, 2)), true)
+ assert.Equal(t,
resources.Equals(manager.GetGroupTracker(userGroup.Groups[0]).queueTracker.maxResources,
resources.Multiply(expectedResource, 2)), true)
assert.Equal(t,
manager.GetUserTracker(userGroup.User).queueTracker.childQueueTrackers["parent"].maxRunningApps,
uint64(expectedMaxApps))
assert.Equal(t,
manager.GetGroupTracker(userGroup.Groups[0]).queueTracker.childQueueTrackers["parent"].maxRunningApps,
uint64(expectedMaxApps))
- assert.Equal(t,
resources.Equals(manager.GetUserTracker(userGroup.User).queueTracker.childQueueTrackers["parent"].maxResourceUsage,
expectedResource), true)
- assert.Equal(t,
resources.Equals(manager.GetGroupTracker(userGroup.Groups[0]).queueTracker.childQueueTrackers["parent"].maxResourceUsage,
expectedResource), true)
+ assert.Equal(t,
resources.Equals(manager.GetUserTracker(userGroup.User).queueTracker.childQueueTrackers["parent"].maxResources,
expectedResource), true)
+ assert.Equal(t,
resources.Equals(manager.GetGroupTracker(userGroup.Groups[0]).queueTracker.childQueueTrackers["parent"].maxResources,
expectedResource), true)
+}
+
+func assertWildCardLimits(t *testing.T, limitsConfig map[string]*LimitConfig,
expectedResource *resources.Resource) {
+ assert.Equal(t, limitsConfig["root"].maxApplications, uint64(20))
+ assert.Equal(t, limitsConfig["root.parent"].maxApplications, uint64(10))
+ expResource := limitsConfig["root"].maxResources
+ assert.Equal(t, resources.Equals(expResource,
resources.Multiply(expectedResource, 2)), true)
+ expResource = limitsConfig["root.parent"].maxResources
+ configuredResource, err :=
resources.NewResourceFromConf(map[string]string{"memory": "10", "vcores": "10"})
+ if err != nil {
+ t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, configuredResource)
+ }
+ assert.Equal(t, resources.Equals(expResource, configuredResource), true)
}
diff --git a/pkg/scheduler/ugm/queue_tracker.go
b/pkg/scheduler/ugm/queue_tracker.go
index 415ab3aa..9ca5e1c7 100644
--- a/pkg/scheduler/ugm/queue_tracker.go
+++ b/pkg/scheduler/ugm/queue_tracker.go
@@ -19,8 +19,6 @@
package ugm
import (
- "fmt"
-
"go.uber.org/zap"
"github.com/apache/yunikorn-core/pkg/common/configs"
@@ -31,73 +29,171 @@ import (
type QueueTracker struct {
queueName string
+ queuePath string
resourceUsage *resources.Resource
runningApplications map[string]bool
- maxResourceUsage *resources.Resource
+ maxResources *resources.Resource
maxRunningApps uint64
childQueueTrackers map[string]*QueueTracker
}
func newRootQueueTracker() *QueueTracker {
- return newQueueTracker(configs.RootQueue)
+ qt := newQueueTracker("", configs.RootQueue)
+ return qt
}
-func newQueueTracker(queueName string) *QueueTracker {
- log.Log(log.SchedUGM).Debug("Creating queue tracker object for queue",
- zap.String("queue", queueName))
+func newQueueTracker(queuePath string, queueName string) *QueueTracker {
+ qp := queueName
+ if queuePath != "" {
+ qp = queuePath + "." + queueName
+ }
queueTracker := &QueueTracker{
queueName: queueName,
+ queuePath: qp,
resourceUsage: resources.NewResource(),
runningApplications: make(map[string]bool),
+ maxResources: resources.NewResource(),
+ maxRunningApps: 0,
childQueueTrackers: make(map[string]*QueueTracker),
}
+ log.Log(log.SchedUGM).Debug("Created queue tracker object for queue",
+ zap.String("queue", queueName))
return queueTracker
}
-func (qt *QueueTracker) increaseTrackedResource(queuePath string,
applicationID string, usage *resources.Resource) error {
+type trackingType int
+
+const (
+ none trackingType = iota
+ user
+ group
+)
+
+func (qt *QueueTracker) increaseTrackedResource(queuePath string,
applicationID string, trackType trackingType, usage *resources.Resource) bool {
log.Log(log.SchedUGM).Debug("Increasing resource usage",
+ zap.Int("tracking type", int(trackType)),
zap.String("queue path", queuePath),
zap.String("application", applicationID),
zap.Stringer("resource", usage))
- if queuePath == "" || applicationID == "" || usage == nil {
- return fmt.Errorf("mandatory parameters are missing. queuepath:
%s, application id: %s, resource usage: %s",
- queuePath, applicationID, usage.String())
+ finalResourceUsage := qt.resourceUsage.Clone()
+ finalResourceUsage.AddTo(usage)
+ wildCardQuotaExceeded := false
+ existingApp := qt.runningApplications[applicationID]
+
+ // apply user/group specific limit settings set if configured,
otherwise use wild card limit settings
+ if qt.maxRunningApps != 0 && !resources.Equals(resources.NewResource(),
qt.maxResources) {
+ log.Log(log.SchedUGM).Debug("applying enforcement checks using
limit settings of specific user/group",
+ zap.Int("tracking type", int(trackType)),
+ zap.String("queue path", queuePath),
+ zap.Bool("existing app", existingApp),
+ zap.Uint64("max running apps", qt.maxRunningApps),
+ zap.String("max resources", qt.maxResources.String()))
+ if (!existingApp && len(qt.runningApplications)+1 >
int(qt.maxRunningApps)) ||
+ resources.StrictlyGreaterThan(finalResourceUsage,
qt.maxResources) {
+ log.Log(log.SchedUGM).Warn("Unable to increase resource
usage as allowing new application to run would exceed either configured max
applications or max resources limit of specific user/group",
+ zap.Int("tracking type", int(trackType)),
+ zap.String("queue path", queuePath),
+ zap.Bool("existing app", existingApp),
+ zap.Int("current running applications",
len(qt.runningApplications)),
+ zap.Uint64("max running applications",
qt.maxRunningApps),
+ zap.String("current resource usage",
qt.resourceUsage.String()),
+ zap.String("max resource usage",
qt.maxResources.String()))
+ return false
+ }
+ }
+
+ // Try wild card settings
+ if qt.maxRunningApps == 0 && resources.Equals(resources.NewResource(),
qt.maxResources) {
+ // Is there any wild card settings? Do we need to apply
enforcement checks using wild card limit settings?
+ var config *LimitConfig
+ if trackType == user {
+ config = m.getUserWildCardLimitsConfig(qt.queuePath)
+ } else if trackType == group {
+ config = m.getGroupWildCardLimitsConfig(qt.queuePath)
+ }
+ if config != nil {
+ log.Log(log.SchedUGM).Debug("applying enforcement
checks using limit settings of wild card user/group",
+ zap.Int("tracking type", int(trackType)),
+ zap.String("queue path", queuePath),
+ zap.Bool("existing app", existingApp),
+ zap.Uint64("wild card max running apps",
config.maxApplications),
+ zap.String("wild card max resources",
config.maxResources.String()),
+ zap.Bool("wild card quota exceeded",
wildCardQuotaExceeded))
+ wildCardQuotaExceeded = (config.maxApplications != 0 &&
!existingApp && len(qt.runningApplications)+1 > int(config.maxApplications)) ||
+ (!resources.Equals(resources.NewResource(),
config.maxResources) && resources.StrictlyGreaterThan(finalResourceUsage,
config.maxResources))
+ if wildCardQuotaExceeded {
+ log.Log(log.SchedUGM).Warn("Unable to increase
resource usage as allowing new application to run would exceed either
configured max applications or max resources limit of wild card user/group",
+ zap.Int("tracking type",
int(trackType)),
+ zap.String("queue path", queuePath),
+ zap.Bool("existing app", existingApp),
+ zap.Int("current running applications",
len(qt.runningApplications)),
+ zap.Uint64("max running applications",
config.maxApplications),
+ zap.String("current resource usage",
qt.resourceUsage.String()),
+ zap.String("max resource usage",
config.maxResources.String()))
+ return false
+ }
+ }
+ }
+
+ childQueuePath, immediateChildQueueName := getChildQueuePath(queuePath)
+ if childQueuePath != "" {
+ if qt.childQueueTrackers[immediateChildQueueName] == nil {
+ qt.childQueueTrackers[immediateChildQueueName] =
newQueueTracker(qt.queuePath, immediateChildQueueName)
+ }
+ allowed :=
qt.childQueueTrackers[immediateChildQueueName].increaseTrackedResource(childQueuePath,
applicationID, trackType, usage)
+ if !allowed {
+ return allowed
+ }
}
+
qt.resourceUsage.AddTo(usage)
qt.runningApplications[applicationID] = true
log.Log(log.SchedUGM).Debug("Successfully increased resource usage",
+ zap.Int("tracking type", int(trackType)),
zap.String("queue path", queuePath),
zap.String("application", applicationID),
+ zap.Bool("existing app", existingApp),
zap.Stringer("resource", usage),
+ zap.Uint64("max running applications", qt.maxRunningApps),
+ zap.String("max resource usage", qt.maxResources.String()),
zap.Stringer("total resource after increasing",
qt.resourceUsage),
zap.Int("total applications after increasing",
len(qt.runningApplications)))
-
- childQueuePath, immediateChildQueueName := getChildQueuePath(queuePath)
- if childQueuePath != "" {
- if qt.childQueueTrackers[immediateChildQueueName] == nil {
- qt.childQueueTrackers[immediateChildQueueName] =
newQueueTracker(immediateChildQueueName)
- }
- err :=
qt.childQueueTrackers[immediateChildQueueName].increaseTrackedResource(childQueuePath,
applicationID, usage)
- if err != nil {
- return err
- }
- }
- return nil
+ return true
}
-func (qt *QueueTracker) decreaseTrackedResource(queuePath string,
applicationID string, usage *resources.Resource, removeApp bool) (bool, error) {
+func (qt *QueueTracker) decreaseTrackedResource(queuePath string,
applicationID string, usage *resources.Resource, removeApp bool) (bool, bool) {
log.Log(log.SchedUGM).Debug("Decreasing resource usage",
zap.String("queue path", queuePath),
zap.String("application", applicationID),
zap.Stringer("resource", usage),
zap.Bool("removeApp", removeApp))
- if queuePath == "" || usage == nil {
- return false, fmt.Errorf("mandatory parameters are missing.
queuepath: %s, application id: %s, resource usage: %s",
- queuePath, applicationID, usage.String())
+ childQueuePath, immediateChildQueueName := getChildQueuePath(queuePath)
+ if childQueuePath != "" {
+ if qt.childQueueTrackers[immediateChildQueueName] == nil {
+ log.Log(log.SchedUGM).Error("Child queueTracker tracker
must be available in child queues map",
+ zap.String("child queueTracker name",
immediateChildQueueName))
+ return false, false
+ }
+ removeQT, decreased :=
qt.childQueueTrackers[immediateChildQueueName].decreaseTrackedResource(childQueuePath,
applicationID, usage, removeApp)
+ if !decreased {
+ return false, decreased
+ }
+ if removeQT {
+ log.Log(log.SchedUGM).Debug("Removed queue tracker
linkage from its parent",
+ zap.String("queue path ", queuePath),
+ zap.String("removed queue name",
immediateChildQueueName),
+ zap.String("parent queue", qt.queueName))
+ delete(qt.childQueueTrackers, immediateChildQueueName)
+ }
}
+
qt.resourceUsage.SubFrom(usage)
if removeApp {
+ log.Log(log.SchedUGM).Debug("Removed application from running
applications",
+ zap.String("application", applicationID),
+ zap.String("queue path", queuePath),
+ zap.String("queue name", qt.queueName))
delete(qt.runningApplications, applicationID)
}
log.Log(log.SchedUGM).Debug("Successfully decreased resource usage",
@@ -107,26 +203,13 @@ func (qt *QueueTracker) decreaseTrackedResource(queuePath
string, applicationID
zap.Stringer("total resource after decreasing",
qt.resourceUsage),
zap.Int("total applications after decreasing",
len(qt.runningApplications)))
- childQueuePath, immediateChildQueueName := getChildQueuePath(queuePath)
- if childQueuePath != "" {
- if qt.childQueueTrackers[immediateChildQueueName] != nil {
- removeQT, err :=
qt.childQueueTrackers[immediateChildQueueName].decreaseTrackedResource(childQueuePath,
applicationID, usage, removeApp)
- if err != nil {
- return false, err
- }
- if removeQT {
- delete(qt.childQueueTrackers,
immediateChildQueueName)
- }
- } else {
- log.Log(log.SchedUGM).Error("Child queueTracker tracker
must be available in child queues map",
- zap.String("child queueTracker name",
immediateChildQueueName))
- return false, fmt.Errorf("child queueTracker tracker
for %s is missing in child queues map", immediateChildQueueName)
- }
- }
-
// Determine if the queue tracker should be removed
- removeQT := len(qt.childQueueTrackers) == 0 &&
len(qt.runningApplications) == 0 && resources.IsZero(qt.resourceUsage)
- return removeQT, nil
+ removeQT := len(qt.childQueueTrackers) == 0 &&
len(qt.runningApplications) == 0 && resources.IsZero(qt.resourceUsage) &&
+ qt.maxRunningApps == 0 &&
resources.Equals(resources.NewResource(), qt.maxResources)
+ log.Log(log.SchedUGM).Debug("Remove queue tracker",
+ zap.String("queue path ", queuePath),
+ zap.Bool("remove QT", removeQT))
+ return removeQT, true
}
func (qt *QueueTracker) getChildQueueTracker(queuePath string) *QueueTracker {
@@ -137,7 +220,7 @@ func (qt *QueueTracker) getChildQueueTracker(queuePath
string) *QueueTracker {
for childQueuePath != "" {
if childQueueTracker != nil {
if len(childQueueTracker.childQueueTrackers) ==
0 || childQueueTracker.childQueueTrackers[immediateChildQueueName] == nil {
- newChildQt :=
newQueueTracker(immediateChildQueueName)
+ newChildQt :=
newQueueTracker(qt.queuePath, immediateChildQueueName)
childQueueTracker.childQueueTrackers[immediateChildQueueName] = newChildQt
childQueueTracker = newChildQt
} else {
@@ -150,40 +233,14 @@ func (qt *QueueTracker) getChildQueueTracker(queuePath
string) *QueueTracker {
return childQueueTracker
}
-func (qt *QueueTracker) setMaxApplications(count uint64, queuePath string)
error {
- log.Log(log.SchedUGM).Debug("Setting max applications",
+func (qt *QueueTracker) setLimit(queuePath string, maxResource
*resources.Resource, maxApps uint64) {
+ log.Log(log.SchedUGM).Debug("Setting limits",
zap.String("queue path", queuePath),
- zap.Uint64("max applications", count))
+ zap.Uint64("max applications", maxApps),
+ zap.String("max resources", maxResource.String()))
childQueueTracker := qt.getChildQueueTracker(queuePath)
- if childQueueTracker.maxRunningApps != 0 && count != 0 &&
len(childQueueTracker.runningApplications) > int(count) {
- log.Log(log.SchedUGM).Warn("Current running applications is
greater than config max applications",
- zap.String("queue path", queuePath),
- zap.Uint64("current max applications",
childQueueTracker.maxRunningApps),
- zap.Int("total running applications",
len(childQueueTracker.runningApplications)),
- zap.Uint64("config max applications", count))
- return fmt.Errorf("current running applications is greater than
config max applications for %s", queuePath)
- } else {
- childQueueTracker.maxRunningApps = count
- }
- return nil
-}
-
-func (qt *QueueTracker) setMaxResources(resource *resources.Resource,
queuePath string) error {
- log.Log(log.SchedUGM).Debug("Setting max resources",
- zap.String("queue path", queuePath),
- zap.String("max resources", resource.String()))
- childQueueTracker := qt.getChildQueueTracker(queuePath)
- if (!resources.Equals(childQueueTracker.maxResourceUsage,
resources.NewResource()) && !resources.Equals(resource,
resources.NewResource())) &&
resources.StrictlyGreaterThan(childQueueTracker.resourceUsage, resource) {
- log.Log(log.SchedUGM).Warn("Current resource usage is greater
than config max resource",
- zap.String("queue path", queuePath),
- zap.String("current max resource usage",
childQueueTracker.maxResourceUsage.String()),
- zap.String("total resource usage",
childQueueTracker.resourceUsage.String()),
- zap.String("config max resources", resource.String()))
- return fmt.Errorf("current resource usage is greater than
config max resource for %s", queuePath)
- } else {
- childQueueTracker.maxResourceUsage = resource
- }
- return nil
+ childQueueTracker.maxRunningApps = maxApps
+ childQueueTracker.maxResources = maxResource
}
func (qt *QueueTracker) getResourceUsageDAOInfo(parentQueuePath string)
*dao.ResourceUsageDAOInfo {
@@ -201,7 +258,7 @@ func (qt *QueueTracker)
getResourceUsageDAOInfo(parentQueuePath string) *dao.Res
for app := range qt.runningApplications {
usage.RunningApplications = append(usage.RunningApplications,
app)
}
- usage.MaxResources = qt.maxResourceUsage
+ usage.MaxResources = qt.maxResources
usage.MaxApplications = qt.maxRunningApps
for _, cqt := range qt.childQueueTrackers {
childUsage := cqt.getResourceUsageDAOInfo(fullQueuePath)
@@ -209,3 +266,69 @@ func (qt *QueueTracker)
getResourceUsageDAOInfo(parentQueuePath string) *dao.Res
}
return usage
}
+
+// IsQueuePathTrackedCompletely Traverse queue path upto the end queue through
its linkage
+// to confirm entire queuePath has been tracked completely or not
+func (qt *QueueTracker) IsQueuePathTrackedCompletely(queuePath string) bool {
+ if queuePath == configs.RootQueue || queuePath == qt.queueName {
+ return true
+ }
+ childQueuePath, immediateChildQueueName := getChildQueuePath(queuePath)
+ if immediateChildQueueName != "" {
+ if childUt, ok :=
qt.childQueueTrackers[immediateChildQueueName]; ok {
+ return
childUt.IsQueuePathTrackedCompletely(childQueuePath)
+ }
+ }
+ return false
+}
+
+// IsUnlinkRequired Traverse queue path upto the leaf queue and decide whether
+// linkage needs to be removed or not based on the running applications.
+// If there are any running applications in end leaf queue, we should remove
the linkage between
+// the leaf and its parent queue using UnlinkQT method. Otherwise, we should
leave as it is.
+func (qt *QueueTracker) IsUnlinkRequired(queuePath string) bool {
+ childQueuePath, immediateChildQueueName := getChildQueuePath(queuePath)
+ if immediateChildQueueName != "" {
+ if childUt, ok :=
qt.childQueueTrackers[immediateChildQueueName]; ok {
+ return childUt.IsUnlinkRequired(childQueuePath)
+ }
+ }
+ if queuePath == configs.RootQueue || queuePath == qt.queueName {
+ if len(qt.runningApplications) == 0 {
+ log.Log(log.SchedUGM).Debug("Is Unlink Required?",
+ zap.String("queue path", queuePath),
+ zap.Int("no. of applications",
len(qt.runningApplications)))
+ return true
+ }
+ }
+ return false
+}
+
+// UnlinkQT Traverse queue path upto the end queue. If end queue has any more
child queue trackers,
+// then goes upto each child queue and removes the linkage with its immediate
parent
+func (qt *QueueTracker) UnlinkQT(queuePath string) bool {
+ log.Log(log.SchedUGM).Debug("Unlinking current queue tracker from its
parent",
+ zap.String("current queue ", qt.queueName),
+ zap.String("queue path", queuePath),
+ zap.Int("no. of child queue trackers",
len(qt.childQueueTrackers)))
+ childQueuePath, immediateChildQueueName := getChildQueuePath(queuePath)
+
+ if childQueuePath == "" && len(qt.childQueueTrackers) > 0 {
+ for qName := range qt.childQueueTrackers {
+ qt.UnlinkQT(qt.queueName + configs.DOT + qName)
+ }
+ }
+
+ if childQueuePath != "" {
+ if qt.childQueueTrackers[immediateChildQueueName] != nil {
+ unlink :=
qt.childQueueTrackers[immediateChildQueueName].UnlinkQT(childQueuePath)
+ if unlink {
+ delete(qt.childQueueTrackers,
immediateChildQueueName)
+ }
+ }
+ }
+ if len(qt.runningApplications) == 0 && len(qt.childQueueTrackers) == 0 {
+ return true
+ }
+ return false
+}
diff --git a/pkg/scheduler/ugm/queue_tracker_test.go
b/pkg/scheduler/ugm/queue_tracker_test.go
index 1a79444e..ea88355a 100644
--- a/pkg/scheduler/ugm/queue_tracker_test.go
+++ b/pkg/scheduler/ugm/queue_tracker_test.go
@@ -31,45 +31,43 @@ func TestQTIncreaseTrackedResource(t *testing.T) {
// root->parent->child1->child12
// root->parent->child2
// root->parent->child12 (similar name like above leaf queue, but it is
being treated differently as similar names are allowed)
- queueTracker := newQueueTracker("root")
+ GetUserManager()
+ queueTracker := newQueueTracker("", "root")
usage1, err := resources.NewResourceFromConf(map[string]string{"mem":
"10M", "vcore": "10"})
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage1)
}
- err = queueTracker.increaseTrackedResource("", "", usage1)
- assert.Error(t, err, "mandatory parameters are missing. queuepath: ,
application id: , resource usage: "+usage1.String())
-
- err = queueTracker.increaseTrackedResource(queuePath1, TestApp1, usage1)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage1, err)
+ result := queueTracker.increaseTrackedResource(queuePath1, TestApp1,
user, usage1)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, usage1)
}
usage2, err := resources.NewResourceFromConf(map[string]string{"mem":
"20M", "vcore": "20"})
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage2)
}
- err = queueTracker.increaseTrackedResource(queuePath2, TestApp2, usage2)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath2, TestApp2, usage2, err)
+ result = queueTracker.increaseTrackedResource(queuePath2, TestApp2,
user, usage2)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath2, TestApp2, usage2)
}
usage3, err := resources.NewResourceFromConf(map[string]string{"mem":
"30M", "vcore": "30"})
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage3)
}
- err = queueTracker.increaseTrackedResource(queuePath3, TestApp3, usage3)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath3, TestApp3, usage3, err)
+ result = queueTracker.increaseTrackedResource(queuePath3, TestApp3,
user, usage3)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath3, TestApp3, usage3)
}
usage4, err := resources.NewResourceFromConf(map[string]string{"mem":
"20M", "vcore": "20"})
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage3)
}
- err = queueTracker.increaseTrackedResource(queuePath4, TestApp4, usage4)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath4, TestApp4, usage4, err)
+ result = queueTracker.increaseTrackedResource(queuePath4, TestApp4,
user, usage4)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath4, TestApp4, usage4)
}
actualResources := getQTResource(queueTracker)
@@ -86,15 +84,16 @@ func TestQTDecreaseTrackedResource(t *testing.T) {
// Queue setup:
// root->parent->child1
// root->parent->child2
- queueTracker := newQueueTracker("root")
+ GetUserManager()
+ queueTracker := newQueueTracker("", "root")
usage1, err := resources.NewResourceFromConf(map[string]string{"mem":
"70M", "vcore": "70"})
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage1)
}
- err = queueTracker.increaseTrackedResource(queuePath1, TestApp1, usage1)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage1, err)
+ result := queueTracker.increaseTrackedResource(queuePath1, TestApp1,
user, usage1)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, usage1)
}
assert.Equal(t, 1, len(queueTracker.runningApplications))
@@ -102,9 +101,9 @@ func TestQTDecreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage2)
}
- err = queueTracker.increaseTrackedResource(queuePath2, TestApp2, usage2)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath2, TestApp2, usage2, err)
+ result = queueTracker.increaseTrackedResource(queuePath2, TestApp2,
user, usage2)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath2, TestApp2, usage2)
}
actualResources := getQTResource(queueTracker)
@@ -119,17 +118,14 @@ func TestQTDecreaseTrackedResource(t *testing.T) {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage3)
}
- err = queueTracker.increaseTrackedResource("", "", usage3)
- assert.Error(t, err, "mandatory parameters are missing. queuepath: ,
application id: , resource usage: "+usage3.String())
-
- removeQT, err := queueTracker.decreaseTrackedResource(queuePath1,
TestApp1, usage3, false)
- if err != nil {
+ removeQT, decreased := queueTracker.decreaseTrackedResource(queuePath1,
TestApp1, usage3, false)
+ if !decreased {
t.Fatalf("unable to decrease tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage3, err)
}
assert.Equal(t, removeQT, false, "wrong remove queue tracker value")
- removeQT, err = queueTracker.decreaseTrackedResource(queuePath2,
TestApp2, usage3, false)
- if err != nil {
+ removeQT, decreased = queueTracker.decreaseTrackedResource(queuePath2,
TestApp2, usage3, false)
+ if !decreased {
t.Fatalf("unable to decrease tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath2, TestApp2, usage3, err)
}
actualResources1 := getQTResource(queueTracker)
@@ -145,8 +141,8 @@ func TestQTDecreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage3)
}
- removeQT, err = queueTracker.decreaseTrackedResource(queuePath1,
TestApp1, usage4, true)
- if err != nil {
+ removeQT, decreased = queueTracker.decreaseTrackedResource(queuePath1,
TestApp1, usage4, true)
+ if !decreased {
t.Fatalf("unable to decrease tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage1, err)
}
assert.Equal(t, 1, len(queueTracker.runningApplications))
@@ -158,8 +154,8 @@ func TestQTDecreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage5)
}
- removeQT, err = queueTracker.decreaseTrackedResource(queuePath2,
TestApp2, usage5, true)
- if err != nil {
+ removeQT, decreased = queueTracker.decreaseTrackedResource(queuePath2,
TestApp2, usage5, true)
+ if !decreased {
t.Fatalf("unable to decrease tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath2, TestApp2, usage2, err)
}
assert.Equal(t, 0, len(queueTracker.runningApplications))
@@ -168,9 +164,9 @@ func TestQTDecreaseTrackedResource(t *testing.T) {
assert.Equal(t, removeQT, true, "wrong remove queue tracker value")
// Test parent queueTracker has not zero usage, but child queueTrackers
has all deleted
- err = queueTracker.increaseTrackedResource(queuePath1, TestApp1, usage1)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage1, err)
+ result = queueTracker.increaseTrackedResource(queuePath1, TestApp1,
user, usage1)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, usage1)
}
assert.Equal(t, 1, len(queueTracker.runningApplications))
@@ -178,9 +174,79 @@ func TestQTDecreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage2)
}
- err = queueTracker.increaseTrackedResource("root.parent", TestApp2,
usage2)
+ result = queueTracker.increaseTrackedResource("root.parent", TestApp2,
user, usage2)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", "root.parent", TestApp2, usage2)
+ }
+}
+
+func TestQTQuotaEnforcement(t *testing.T) {
+ // Queue setup:
+ // root. max apps - 6 , max res - 60M, 60cores
+ // root-> parent. max apps - 5 , max res - 50M, 50cores
+ // root->parent->child1. max apps - 2 , max res - 20M, 20cores
+ // root->parent->child1->child12. config not set
+ // root->parent->child2. max apps - 2 , max res - 20M, 20cores
+ // root->parent->child12 (similar name like above leaf queue, but it is
being treated differently as similar names are allowed). config not set
+ GetUserManager()
+ queueTracker := newQueueTracker("", "root")
+
+ usage1, err := resources.NewResourceFromConf(map[string]string{"mem":
"10M", "vcore": "10"})
if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", "root.parent", TestApp2, usage2, err)
+ t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage1)
+ }
+
+ queueTracker.maxResources = resources.Multiply(usage1, 6)
+ queueTracker.maxRunningApps = 6
+
+ parentQueueTracker := newQueueTracker("root", "parent")
+ parentQueueTracker.maxResources = resources.Multiply(usage1, 5)
+ parentQueueTracker.maxRunningApps = 5
+ queueTracker.childQueueTrackers["parent"] = parentQueueTracker
+
+ child1QueueTracker := newQueueTracker("root.parent", "child1")
+ child1QueueTracker.maxResources = resources.Multiply(usage1, 2)
+ child1QueueTracker.maxRunningApps = 2
+ parentQueueTracker.childQueueTrackers["child1"] = child1QueueTracker
+
+ child2QueueTracker := newQueueTracker("root.parent.child2", "child2")
+ child2QueueTracker.maxResources = resources.Multiply(usage1, 2)
+ child2QueueTracker.maxRunningApps = 2
+ parentQueueTracker.childQueueTrackers["child2"] = child2QueueTracker
+
+ result := queueTracker.increaseTrackedResource(queuePath1, TestApp1,
user, usage1)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, usage1)
+ }
+
+ result = queueTracker.increaseTrackedResource(queuePath2, TestApp2,
user, usage1)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath2, TestApp2, usage1)
+ }
+
+ result = queueTracker.increaseTrackedResource(queuePath2, TestApp2,
user, usage1)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath2, TestApp2, usage1)
+ }
+
+ result = queueTracker.increaseTrackedResource(queuePath2, TestApp3,
user, usage1)
+ if result {
+ t.Fatalf("Increasing resource usage should fail as child2's
resource usage exceeded configured max resources limit. queuepath %s, app %s,
res %v", queuePath2, TestApp3, usage1)
+ }
+
+ result = queueTracker.increaseTrackedResource(queuePath3, TestApp3,
user, usage1)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath3, TestApp3, usage1)
+ }
+
+ result = queueTracker.increaseTrackedResource(queuePath4, TestApp4,
user, usage1)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath4, TestApp4, usage1)
+ }
+
+ result = queueTracker.increaseTrackedResource(queuePath4, TestApp4,
user, usage1)
+ if result {
+ t.Fatalf("Increasing resource usage should fail as parent's
resource usage exceeded configured max resources limit. queuepath %s, app %s,
res %v", queuePath4, TestApp4, usage1)
}
}
diff --git a/pkg/scheduler/ugm/tracker.go b/pkg/scheduler/ugm/tracker.go
index 85f3c56d..ac2bf7af 100644
--- a/pkg/scheduler/ugm/tracker.go
+++ b/pkg/scheduler/ugm/tracker.go
@@ -31,6 +31,6 @@ type Tracker interface {
GetUsersResources() []*UserTracker
GetGroupsResources() []*GroupTracker
- IncreaseTrackedResource(queuePath, applicationID string, usage
*resources.Resource, user security.UserGroup) error
- DecreaseTrackedResource(queuePath, applicationID string, usage
*resources.Resource, user security.UserGroup, removeApp bool) error
+ IncreaseTrackedResource(queuePath, applicationID string, usage
*resources.Resource, user security.UserGroup) bool
+ DecreaseTrackedResource(queuePath, applicationID string, usage
*resources.Resource, user security.UserGroup, removeApp bool) bool
}
diff --git a/pkg/scheduler/ugm/user_tracker.go
b/pkg/scheduler/ugm/user_tracker.go
index 85d508d8..eb95824a 100644
--- a/pkg/scheduler/ugm/user_tracker.go
+++ b/pkg/scheduler/ugm/user_tracker.go
@@ -48,13 +48,13 @@ func newUserTracker(user string) *UserTracker {
return userTracker
}
-func (ut *UserTracker) increaseTrackedResource(queuePath, applicationID
string, usage *resources.Resource) error {
+func (ut *UserTracker) increaseTrackedResource(queuePath, applicationID
string, usage *resources.Resource) bool {
ut.Lock()
defer ut.Unlock()
- return ut.queueTracker.increaseTrackedResource(queuePath,
applicationID, usage)
+ return ut.queueTracker.increaseTrackedResource(queuePath,
applicationID, user, usage)
}
-func (ut *UserTracker) decreaseTrackedResource(queuePath, applicationID
string, usage *resources.Resource, removeApp bool) (bool, error) {
+func (ut *UserTracker) decreaseTrackedResource(queuePath, applicationID
string, usage *resources.Resource, removeApp bool) (bool, bool) {
ut.Lock()
defer ut.Unlock()
if removeApp {
@@ -83,16 +83,10 @@ func (ut *UserTracker) getTrackedApplications()
map[string]*GroupTracker {
return ut.appGroupTrackers
}
-func (ut *UserTracker) setMaxApplications(count uint64, queuePath string)
error {
+func (ut *UserTracker) setLimits(queuePath string, resource
*resources.Resource, maxApps uint64) {
ut.Lock()
defer ut.Unlock()
- return ut.queueTracker.setMaxApplications(count, queuePath)
-}
-
-func (ut *UserTracker) setMaxResources(resource *resources.Resource, queuePath
string) error {
- ut.Lock()
- defer ut.Unlock()
- return ut.queueTracker.setMaxResources(resource, queuePath)
+ ut.queueTracker.setLimit(queuePath, resource, maxApps)
}
func (ut *UserTracker) GetUserResourceUsageDAOInfo()
*dao.UserResourceUsageDAOInfo {
@@ -108,3 +102,28 @@ func (ut *UserTracker) GetUserResourceUsageDAOInfo()
*dao.UserResourceUsageDAOIn
userResourceUsage.Queues = ut.queueTracker.getResourceUsageDAOInfo("")
return userResourceUsage
}
+
+func (ut *UserTracker) IsQueuePathTrackedCompletely(queuePath string) bool {
+ ut.RLock()
+ defer ut.RUnlock()
+ return ut.queueTracker.IsQueuePathTrackedCompletely(queuePath)
+}
+
+func (ut *UserTracker) IsUnlinkRequired(queuePath string) bool {
+ ut.RLock()
+ defer ut.RUnlock()
+ return ut.queueTracker.IsUnlinkRequired(queuePath)
+}
+
+func (ut *UserTracker) UnlinkQT(queuePath string) bool {
+ ut.RLock()
+ defer ut.RUnlock()
+ return ut.queueTracker.UnlinkQT(queuePath)
+}
+
+// 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
+}
diff --git a/pkg/scheduler/ugm/user_tracker_test.go
b/pkg/scheduler/ugm/user_tracker_test.go
index d8e5485d..bf9cc143 100644
--- a/pkg/scheduler/ugm/user_tracker_test.go
+++ b/pkg/scheduler/ugm/user_tracker_test.go
@@ -50,9 +50,9 @@ func TestIncreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage1)
}
- err = userTracker.increaseTrackedResource(queuePath1, TestApp1, usage1)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage1, err)
+ result := userTracker.increaseTrackedResource(queuePath1, TestApp1,
usage1)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, usage1)
}
groupTracker := newGroupTracker(user.User)
userTracker.setGroupForApp(TestApp1, groupTracker)
@@ -61,8 +61,8 @@ func TestIncreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage2)
}
- err = userTracker.increaseTrackedResource(queuePath2, TestApp2, usage2)
- if err != nil {
+ result = userTracker.increaseTrackedResource(queuePath2, TestApp2,
usage2)
+ if !result {
t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath2, TestApp2, usage2, err)
}
userTracker.setGroupForApp(TestApp2, groupTracker)
@@ -71,8 +71,8 @@ func TestIncreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage3)
}
- err = userTracker.increaseTrackedResource(queuePath3, TestApp3, usage3)
- if err != nil {
+ result = userTracker.increaseTrackedResource(queuePath3, TestApp3,
usage3)
+ if !result {
t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath3, TestApp3, usage3, err)
}
userTracker.setGroupForApp(TestApp3, groupTracker)
@@ -81,8 +81,8 @@ func TestIncreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage3)
}
- err = userTracker.increaseTrackedResource(queuePath4, TestApp4, usage4)
- if err != nil {
+ result = userTracker.increaseTrackedResource(queuePath4, TestApp4,
usage4)
+ if !result {
t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath4, TestApp4, usage4, err)
}
userTracker.setGroupForApp(TestApp4, groupTracker)
@@ -111,8 +111,8 @@ func TestDecreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage1)
}
- err = userTracker.increaseTrackedResource(queuePath1, TestApp1, usage1)
- if err != nil {
+ result := userTracker.increaseTrackedResource(queuePath1, TestApp1,
usage1)
+ if !result {
t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage1, err)
}
groupTracker := newGroupTracker(user.User)
@@ -123,8 +123,8 @@ func TestDecreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage2)
}
- err = userTracker.increaseTrackedResource(queuePath2, TestApp2, usage2)
- if err != nil {
+ result = userTracker.increaseTrackedResource(queuePath2, TestApp2,
usage2)
+ if !result {
t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath2, TestApp2, usage2, err)
}
userTracker.setGroupForApp(TestApp2, groupTracker)
@@ -140,14 +140,14 @@ func TestDecreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage3)
}
- removeQT, err := userTracker.decreaseTrackedResource(queuePath1,
TestApp1, usage3, false)
- if err != nil {
+ removeQT, decreased := userTracker.decreaseTrackedResource(queuePath1,
TestApp1, usage3, false)
+ if !decreased {
t.Fatalf("unable to decrease tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage3, err)
}
assert.Equal(t, removeQT, false, "wrong remove queue tracker value")
- removeQT, err = userTracker.decreaseTrackedResource(queuePath2,
TestApp2, usage3, false)
- if err != nil {
+ removeQT, decreased = userTracker.decreaseTrackedResource(queuePath2,
TestApp2, usage3, false)
+ if !decreased {
t.Fatalf("unable to decrease tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath2, TestApp2, usage3, err)
}
actualResources1 := getUserResource(userTracker)
@@ -163,8 +163,8 @@ func TestDecreaseTrackedResource(t *testing.T) {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage3)
}
- removeQT, err = userTracker.decreaseTrackedResource(queuePath1,
TestApp1, usage4, true)
- if err != nil {
+ removeQT, decreased = userTracker.decreaseTrackedResource(queuePath1,
TestApp1, usage4, true)
+ if !decreased {
t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage1, err)
}
assert.Equal(t, 1, len(userTracker.getTrackedApplications()))
@@ -174,8 +174,8 @@ func TestDecreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage5)
}
- removeQT, err = userTracker.decreaseTrackedResource(queuePath2,
TestApp2, usage5, true)
- if err != nil {
+ removeQT, decreased = userTracker.decreaseTrackedResource(queuePath2,
TestApp2, usage5, true)
+ if !decreased {
t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath2, TestApp2, usage2, err)
}
assert.Equal(t, 0, len(userTracker.getTrackedApplications()))
@@ -191,41 +191,25 @@ func TestSetMaxLimits(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource:
error %t, res %v", err, usage1)
}
- err = userTracker.increaseTrackedResource(queuePath1, TestApp1, usage1)
- if err != nil {
+ result := userTracker.increaseTrackedResource(queuePath1, TestApp1,
usage1)
+ if !result {
t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage1, err)
}
- groupTracker := newGroupTracker(user.User)
- userTracker.setGroupForApp(TestApp1, groupTracker)
-
- setMaxAppsErr := userTracker.setMaxApplications(1, queuePath1)
- assert.NilError(t, setMaxAppsErr)
- setMaxResourcesErr := userTracker.setMaxResources(usage1, queuePath1)
- assert.NilError(t, setMaxResourcesErr)
+ userTracker.setLimits(queuePath1, resources.Multiply(usage1, 5), 5)
+ userTracker.setLimits("root.parent", resources.Multiply(usage1, 10), 10)
- setParentMaxAppsErr := userTracker.setMaxApplications(1, "root.parent")
- assert.NilError(t, setParentMaxAppsErr)
-
- setParentMaxResourcesErr := userTracker.setMaxResources(usage1,
"root.parent")
- assert.NilError(t, setParentMaxResourcesErr)
-
- err = userTracker.increaseTrackedResource(queuePath1, TestApp2, usage1)
- if err != nil {
- t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v, error %t", queuePath1, TestApp1, usage1, err)
+ result = userTracker.increaseTrackedResource(queuePath1, TestApp1,
usage1)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp1, usage1)
}
- setMaxAppsErr1 := userTracker.setMaxApplications(1, queuePath1)
- assert.Error(t, setMaxAppsErr1, "current running applications is
greater than config max applications for "+queuePath1)
-
- setMaxResourcesErr1 := userTracker.setMaxResources(usage1, queuePath1)
- assert.Error(t, setMaxResourcesErr1, "current resource usage is greater
than config max resource for "+queuePath1)
-
- setParentMaxAppsErr1 := userTracker.setMaxApplications(1, "root.parent")
- assert.Error(t, setParentMaxAppsErr1, "current running applications is
greater than config max applications for root.parent")
-
- setParentMaxResourcesErr1 := userTracker.setMaxResources(usage1,
"root.parent")
- assert.Error(t, setParentMaxResourcesErr1, "current resource usage is
greater than config max resource for root.parent")
+ result = userTracker.increaseTrackedResource(queuePath1, TestApp2,
usage1)
+ if !result {
+ t.Fatalf("unable to increase tracked resource: queuepath %s,
app %s, res %v", queuePath1, TestApp2, usage1)
+ }
+ userTracker.setLimits(queuePath1, usage1, 1)
+ userTracker.setLimits("root.parent", usage1, 1)
}
func getUserResource(ut *UserTracker) map[string]*resources.Resource {
diff --git a/pkg/scheduler/utilities_test.go b/pkg/scheduler/utilities_test.go
index 99daf94e..3233c61d 100644
--- a/pkg/scheduler/utilities_test.go
+++ b/pkg/scheduler/utilities_test.go
@@ -80,7 +80,7 @@ func newBasePartition() (*PartitionContext, error) {
"memory": "5",
"vcores": "5",
},
-
MaxApplications: 1,
+
MaxApplications: 8,
},
},
},
@@ -98,7 +98,7 @@ func newBasePartition() (*PartitionContext, error) {
"memory": "10",
"vcores": "10",
},
- MaxApplications: 2,
+ MaxApplications: 10,
},
},
},
@@ -179,7 +179,7 @@ func newConfiguredPartition() (*PartitionContext, error) {
"memory": "5",
"vcores": "5",
},
-
MaxApplications: 2,
+
MaxApplications: 8,
},
},
},
@@ -197,7 +197,7 @@ func newConfiguredPartition() (*PartitionContext, error) {
"memory": "10",
"vcores": "10",
},
- MaxApplications: 2,
+ MaxApplications: 10,
},
},
},
@@ -246,7 +246,7 @@ func newPreemptionConfiguredPartition(parentLimit
map[string]string, leafGuarant
"memory": "5",
"vcores": "5",
},
-
MaxApplications: 1,
+
MaxApplications: 8,
},
},
},
@@ -271,7 +271,7 @@ func newPreemptionConfiguredPartition(parentLimit
map[string]string, leafGuarant
"memory": "5",
"vcores": "5",
},
-
MaxApplications: 1,
+
MaxApplications: 6,
},
},
},
@@ -289,7 +289,7 @@ func newPreemptionConfiguredPartition(parentLimit
map[string]string, leafGuarant
"memory": "5",
"vcores": "5",
},
-
MaxApplications: 2,
+
MaxApplications: 8,
},
},
},
@@ -307,7 +307,7 @@ func newPreemptionConfiguredPartition(parentLimit
map[string]string, leafGuarant
"memory": "10",
"vcores": "10",
},
- MaxApplications: 2,
+ MaxApplications: 10,
},
},
},
@@ -592,8 +592,8 @@ func assertLimits(t *testing.T, userGroup
security.UserGroup, expected *resource
expectedQueuesMaxLimits["root.default"] = make(map[string]interface{})
expectedQueuesMaxLimits["root"][maxresources] =
resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 10,
"vcores": 10})
expectedQueuesMaxLimits["root.default"][maxresources] =
resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 5,
"vcores": 5})
- expectedQueuesMaxLimits["root"][maxapplications] = uint64(2)
- expectedQueuesMaxLimits["root.default"][maxapplications] = uint64(1)
+ expectedQueuesMaxLimits["root"][maxapplications] = uint64(10)
+ expectedQueuesMaxLimits["root.default"][maxapplications] = uint64(8)
assertUserGroupResourceMaxLimits(t, userGroup, expected,
expectedQueuesMaxLimits)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
