You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@yunikorn.apache.org by wi...@apache.org on 2023/08/21 07:25:25 UTC
[yunikorn-core] branch master updated: [YUNIKORN-1858] Handle group limit config changes (#600)
This is an automated email from the ASF dual-hosted git repository.
wilfreds pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-core.git
The following commit(s) were added to refs/heads/master by this push:
new fa43406d [YUNIKORN-1858] Handle group limit config changes (#600)
fa43406d is described below
commit fa43406dd9e7664749b6c5283e1460b5ae1ca9e3
Author: Manikandan R <ma...@gmail.com>
AuthorDate: Mon Aug 21 17:20:58 2023 +1000
[YUNIKORN-1858] Handle group limit config changes (#600)
handle config changes for groups and clean up linked usage tracking
objects when the config changes.
Closes: #600
Signed-off-by: Wilfred Spiegelenburg <wi...@apache.org>
---
pkg/scheduler/objects/application.go | 1 +
pkg/scheduler/partition_test.go | 20 ++++++++++++
pkg/scheduler/ugm/group_tracker.go | 36 +++++++++++++--------
pkg/scheduler/ugm/group_tracker_test.go | 18 +++++------
pkg/scheduler/ugm/manager.go | 51 +++++++++---------------------
pkg/scheduler/ugm/queue_tracker.go | 56 +++++++++++++++++++++++++++------
pkg/scheduler/ugm/user_tracker.go | 2 +-
pkg/scheduler/utilities_test.go | 4 +++
8 files changed, 118 insertions(+), 70 deletions(-)
diff --git a/pkg/scheduler/objects/application.go b/pkg/scheduler/objects/application.go
index bdbf72fd..1f47c7e1 100644
--- a/pkg/scheduler/objects/application.go
+++ b/pkg/scheduler/objects/application.go
@@ -681,6 +681,7 @@ func (sa *Application) AddAllocationAsk(ask *AllocationAsk) error {
log.Log(log.SchedApplication).Info("ask added successfully to application",
zap.String("appID", sa.ApplicationID),
+ zap.String("appID", sa.user.User),
zap.String("ask", ask.GetAllocationKey()),
zap.Bool("placeholder", ask.IsPlaceholder()),
zap.Stringer("pendingDelta", delta))
diff --git a/pkg/scheduler/partition_test.go b/pkg/scheduler/partition_test.go
index d1f55669..9f4c8125 100644
--- a/pkg/scheduler/partition_test.go
+++ b/pkg/scheduler/partition_test.go
@@ -3527,6 +3527,26 @@ func TestUserHeadroom(t *testing.T) {
if alloc != nil {
t.Fatal("allocation should not happen on other nodes as well")
}
+ partition.removeApplication("app-5")
+
+ app6 := newApplicationWithUser("app-6", "default", "root.parent.sub-leaf", security.UserGroup{
+ User: "testuser1",
+ Groups: []string{"testgroup1"},
+ })
+ res, err = resources.NewResourceFromConf(map[string]string{"memory": "3", "vcores": "3"})
+ assert.NilError(t, err, "failed to create resource")
+
+ err = partition.AddApplication(app6)
+ assert.NilError(t, err, "failed to add app-6 to partition")
+ err = app6.AddAllocationAsk(newAllocationAsk(allocID, "app-6", res))
+ assert.NilError(t, err, "failed to add ask alloc-1 to app-6")
+
+ // app 6 would be allocated as headroom is nil because no limits configured for 'testuser1' user an
+ alloc = partition.tryAllocate()
+ if alloc == nil {
+ t.Fatal("allocation did not return any allocation")
+ }
+ assert.Equal(t, alloc.GetResult(), objects.Allocated, "result is not the expected allocated")
}
func TestPlaceholderAllocationTracking(t *testing.T) {
diff --git a/pkg/scheduler/ugm/group_tracker.go b/pkg/scheduler/ugm/group_tracker.go
index ff410dc2..8fde8618 100644
--- a/pkg/scheduler/ugm/group_tracker.go
+++ b/pkg/scheduler/ugm/group_tracker.go
@@ -27,9 +27,9 @@ import (
)
type GroupTracker struct {
- groupName string // Name of the group for which usage is being tracked upon
- applications map[string]bool // Hold applications currently run by all users belong to this group
- queueTracker *QueueTracker // Holds the actual resource usage of queue path where application run
+ groupName string // Name of the group for which usage is being tracked upon
+ applications map[string]string // Hold applications currently run by all users belong to this group
+ queueTracker *QueueTracker // Holds the actual resource usage of queue path where application run
sync.RWMutex
}
@@ -38,19 +38,19 @@ func newGroupTracker(group string) *GroupTracker {
queueTracker := newRootQueueTracker()
groupTracker := &GroupTracker{
groupName: group,
- applications: make(map[string]bool),
+ applications: make(map[string]string),
queueTracker: queueTracker,
}
return groupTracker
}
-func (gt *GroupTracker) increaseTrackedResource(queuePath, applicationID string, usage *resources.Resource) bool {
+func (gt *GroupTracker) increaseTrackedResource(queuePath, applicationID string, usage *resources.Resource, user string) bool {
if gt == nil {
return true
}
gt.Lock()
defer gt.Unlock()
- gt.applications[applicationID] = true
+ gt.applications[applicationID] = user
return gt.queueTracker.increaseTrackedResource(queuePath, applicationID, group, usage)
}
@@ -66,7 +66,7 @@ func (gt *GroupTracker) decreaseTrackedResource(queuePath, applicationID string,
return gt.queueTracker.decreaseTrackedResource(queuePath, applicationID, usage, removeApp)
}
-func (gt *GroupTracker) getTrackedApplications() map[string]bool {
+func (gt *GroupTracker) getTrackedApplications() map[string]string {
gt.RLock()
defer gt.RUnlock()
return gt.applications
@@ -123,15 +123,25 @@ func (gt *GroupTracker) canBeRemoved() bool {
return len(gt.queueTracker.childQueueTrackers) == 0 && len(gt.queueTracker.runningApplications) == 0
}
-func (gt *GroupTracker) removeApp(applicationID string) {
- gt.Lock()
- defer gt.Unlock()
- delete(gt.applications, applicationID)
-}
-
func (gt *GroupTracker) getName() string {
if gt == nil {
return common.Empty
}
return gt.groupName
}
+
+func (gt *GroupTracker) decreaseAllTrackedResourceUsage(queuePath string) map[string]string {
+ if gt == nil {
+ return nil
+ }
+ gt.Lock()
+ defer gt.Unlock()
+ applications := gt.queueTracker.decreaseTrackedResourceUsageDownwards(queuePath)
+ removedApplications := make(map[string]string)
+ for app := range applications {
+ if u, ok := gt.applications[app]; ok {
+ removedApplications[app] = u
+ }
+ }
+ return removedApplications
+}
diff --git a/pkg/scheduler/ugm/group_tracker_test.go b/pkg/scheduler/ugm/group_tracker_test.go
index 7d0ce280..5dbbe544 100644
--- a/pkg/scheduler/ugm/group_tracker_test.go
+++ b/pkg/scheduler/ugm/group_tracker_test.go
@@ -40,7 +40,7 @@ func TestGTIncreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource: error %t, res %v", err, usage1)
}
- result := groupTracker.increaseTrackedResource(queuePath1, TestApp1, usage1)
+ result := groupTracker.increaseTrackedResource(queuePath1, TestApp1, usage1, user.User)
if !result {
t.Fatalf("unable to increase tracked resource: queuepath %s, app %s, res %v", queuePath1, TestApp1, usage1)
}
@@ -49,7 +49,7 @@ func TestGTIncreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource: error %t, res %v", err, usage2)
}
- result = groupTracker.increaseTrackedResource(queuePath2, TestApp2, usage2)
+ result = groupTracker.increaseTrackedResource(queuePath2, TestApp2, usage2, user.User)
if !result {
t.Fatalf("unable to increase tracked resource: queuepath %s, app %s, res %v", queuePath2, TestApp2, usage2)
}
@@ -58,7 +58,7 @@ func TestGTIncreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource: error %t, res %v", err, usage3)
}
- result = groupTracker.increaseTrackedResource(queuePath3, TestApp3, usage3)
+ result = groupTracker.increaseTrackedResource(queuePath3, TestApp3, usage3, user.User)
if !result {
t.Fatalf("unable to increase tracked resource: queuepath %s, app %s, res %v", queuePath3, TestApp3, usage3)
}
@@ -67,7 +67,7 @@ func TestGTIncreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource: error %t, res %v", err, usage3)
}
- result = groupTracker.increaseTrackedResource(queuePath4, TestApp4, usage4)
+ result = groupTracker.increaseTrackedResource(queuePath4, TestApp4, usage4, user.User)
if !result {
t.Fatalf("unable to increase tracked resource: queuepath %s, app %s, res %v", queuePath4, TestApp4, usage4)
}
@@ -92,7 +92,7 @@ func TestGTDecreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource: error %t, res %v", err, usage1)
}
- result := groupTracker.increaseTrackedResource(queuePath1, TestApp1, usage1)
+ result := groupTracker.increaseTrackedResource(queuePath1, TestApp1, usage1, user.User)
if !result {
t.Fatalf("unable to increase tracked resource: queuepath %s, app %s, res %v", queuePath1, TestApp1, usage1)
}
@@ -102,7 +102,7 @@ func TestGTDecreaseTrackedResource(t *testing.T) {
if err != nil {
t.Errorf("new resource create returned error or wrong resource: error %t, res %v", err, usage2)
}
- result = groupTracker.increaseTrackedResource(queuePath2, TestApp2, usage2)
+ result = groupTracker.increaseTrackedResource(queuePath2, TestApp2, usage2, user.User)
if !result {
t.Fatalf("unable to increase tracked resource: queuepath %s, app %s, res %v", queuePath2, TestApp2, usage2)
}
@@ -172,7 +172,7 @@ func TestGTSetMaxLimits(t *testing.T) {
t.Errorf("new resource create returned error or wrong resource: error %t, res %v", err, usage1)
}
- result := groupTracker.increaseTrackedResource(queuePath1, TestApp1, usage1)
+ result := groupTracker.increaseTrackedResource(queuePath1, TestApp1, usage1, user.User)
if !result {
t.Fatalf("unable to increase tracked resource: queuepath %s, app %s, res %v", queuePath1, TestApp1, usage1)
}
@@ -180,11 +180,11 @@ func TestGTSetMaxLimits(t *testing.T) {
groupTracker.setLimits(queuePath1, resources.Multiply(usage1, 5), 5)
groupTracker.setLimits("root.parent", resources.Multiply(usage1, 10), 10)
- result = groupTracker.increaseTrackedResource(queuePath1, TestApp2, usage1)
+ result = groupTracker.increaseTrackedResource(queuePath1, TestApp2, usage1, user.User)
if !result {
t.Fatalf("unable to increase tracked resource: queuepath %s, app %s, res %v", queuePath1, TestApp2, usage1)
}
- result = groupTracker.increaseTrackedResource(queuePath1, TestApp3, usage1)
+ result = groupTracker.increaseTrackedResource(queuePath1, TestApp3, usage1, user.User)
if !result {
t.Fatalf("unable to increase tracked resource: queuepath %s, app %s, res %v", queuePath1, TestApp2, usage1)
}
diff --git a/pkg/scheduler/ugm/manager.go b/pkg/scheduler/ugm/manager.go
index d71ad032..76141573 100644
--- a/pkg/scheduler/ugm/manager.go
+++ b/pkg/scheduler/ugm/manager.go
@@ -408,46 +408,20 @@ func (m *Manager) processGroupConfig(group string, limitConfig *LimitConfig, que
// 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 ut.IsQueuePathTrackedCompletely(queuePath) {
- // Traverse all the group trackers linked to user through different applications and remove the linkage
- for appID, gt := range ut.appGroupTrackers {
- if gt != nil {
- g := gt.groupName
- // Is there any limit config set for group in the current configuration? If not, then remove the linkage by setting it to nil
- if ok := groupLimits[g]; !ok {
- log.Log(log.SchedUGM).Debug("Removed the linkage between user and group through applications",
- zap.String("user", u),
- zap.String("group", gt.groupName),
- zap.String("application id", appID),
- zap.String("queue path", queuePath))
- // removing the linkage only happens here by setting it to nil and deleting app id
- // but group resource usage so far remains as it is because we don't have app id wise resource usage with in group as of now.
- // YUNIKORN-1858 handles the group resource usage properly
- // In case of only one (last) application, group tracker would be removed from the manager.
- ut.setGroupForApp(appID, nil)
- gt.removeApp(appID)
- if len(gt.getTrackedApplications()) == 0 {
- log.Log(log.SchedUGM).Debug("Is this app the only running application in group?",
- zap.String("user", u),
- zap.String("group", gt.groupName),
- zap.Int("no. of applications", len(gt.getTrackedApplications())),
- zap.String("application id", appID),
- zap.String("queue path", queuePath))
- delete(m.groupTrackers, g)
- }
- }
- }
+ // Clear already configured limits of group for which limits have been configured before but not now
+ for _, gt := range m.groupTrackers {
+ appUsersMap := m.clearEarlierSetGroupLimits(gt, queuePath, groupLimits)
+ if len(appUsersMap) > 0 {
+ for app, user := range appUsersMap {
+ ut := m.userTrackers[user]
+ ut.setGroupForApp(app, nil)
}
}
- m.clearEarlierSetUserLimits(ut, queuePath, userLimits)
}
- // Clear already configured limits of group for which limits have been configured before but not now
- for _, gt := range m.groupTrackers {
- m.clearEarlierSetGroupLimits(gt, queuePath, groupLimits)
+ // Clear already configured limits of user for which limits have been configured before but not now
+ for _, ut := range m.userTrackers {
+ m.clearEarlierSetUserLimits(ut, queuePath, userLimits)
}
return nil
}
@@ -480,7 +454,8 @@ func (m *Manager) clearEarlierSetUserLimits(ut *UserTracker, queuePath string, u
}
}
-func (m *Manager) clearEarlierSetGroupLimits(gt *GroupTracker, queuePath string, groupLimits map[string]bool) {
+func (m *Manager) clearEarlierSetGroupLimits(gt *GroupTracker, queuePath string, groupLimits map[string]bool) map[string]string {
+ appUsersMap := make(map[string]string)
// Is this group already tracked for the queue path?
if gt.IsQueuePathTrackedCompletely(queuePath) {
g := gt.groupName
@@ -489,6 +464,7 @@ func (m *Manager) clearEarlierSetGroupLimits(gt *GroupTracker, queuePath string,
log.Log(log.SchedUGM).Debug("Need to clear earlier set configs for group",
zap.String("group", g),
zap.String("queue path", queuePath))
+ appUsersMap = gt.decreaseAllTrackedResourceUsage(queuePath)
// Is there any running applications in end queue of this queue path? If not, then remove the linkage between end queue and its immediate parent
if gt.IsUnlinkRequired(queuePath) {
gt.UnlinkQT(queuePath)
@@ -506,6 +482,7 @@ func (m *Manager) clearEarlierSetGroupLimits(gt *GroupTracker, queuePath string,
}
}
}
+ return appUsersMap
}
func (m *Manager) setUserLimits(user string, limitConfig *LimitConfig, queuePath string) error {
diff --git a/pkg/scheduler/ugm/queue_tracker.go b/pkg/scheduler/ugm/queue_tracker.go
index c793ffc0..edcb0c07 100644
--- a/pkg/scheduler/ugm/queue_tracker.go
+++ b/pkg/scheduler/ugm/queue_tracker.go
@@ -88,7 +88,7 @@ func (qt *QueueTracker) increaseTrackedResource(queuePath string, applicationID
zap.String("queue path", queuePath),
zap.Bool("existing app", existingApp),
zap.Uint64("max running apps", qt.maxRunningApps),
- zap.String("max resources", qt.maxResources.String()))
+ zap.Stringer("max resources", qt.maxResources))
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",
@@ -97,8 +97,8 @@ func (qt *QueueTracker) increaseTrackedResource(queuePath string, applicationID
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()))
+ zap.Stringer("current resource usage", qt.resourceUsage),
+ zap.Stringer("max resource usage", qt.maxResources))
return false
}
}
@@ -118,7 +118,7 @@ func (qt *QueueTracker) increaseTrackedResource(queuePath string, applicationID
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.Stringer("wild card max resources", config.maxResources),
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))
@@ -129,8 +129,8 @@ func (qt *QueueTracker) increaseTrackedResource(queuePath string, applicationID
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()))
+ zap.Stringer("current resource usage", qt.resourceUsage),
+ zap.Stringer("max resource usage", config.maxResources))
return false
}
}
@@ -158,7 +158,7 @@ func (qt *QueueTracker) increaseTrackedResource(queuePath string, 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("max resource usage", qt.maxResources),
zap.Stringer("total resource after increasing", qt.resourceUsage),
zap.Int("total applications after increasing", len(qt.runningApplications)))
return true
@@ -239,7 +239,7 @@ func (qt *QueueTracker) setLimit(queuePath string, maxResource *resources.Resour
log.Log(log.SchedUGM).Debug("Setting limits",
zap.String("queue path", queuePath),
zap.Uint64("max applications", maxApps),
- zap.String("max resources", maxResource.String()))
+ zap.Stringer("max resources", maxResource))
childQueueTracker := qt.getChildQueueTracker(queuePath)
childQueueTracker.maxRunningApps = maxApps
childQueueTracker.maxResources = maxResource
@@ -268,8 +268,8 @@ func (qt *QueueTracker) headroom(queuePath string) *resources.Resource {
log.Log(log.SchedUGM).Debug("Calculated headroom",
zap.String("queue path", queuePath),
zap.String("queue", qt.queueName),
- zap.String("max resource", qt.maxResources.String()),
- zap.String("headroom", headroom.String()))
+ zap.Stringer("max resource", qt.maxResources),
+ zap.Stringer("headroom", headroom))
return headroom
}
return nil
@@ -364,3 +364,39 @@ func (qt *QueueTracker) UnlinkQT(queuePath string) bool {
}
return false
}
+
+// decreaseTrackedResourceUsageDownwards queuePath either could be parent or leaf queue path. If it is parent queue path, then traverse upto the end leaf
+// recursively for all child queues, reset resourceUsage and runningApplications to the default value.
+// Once downward traversal has been completed, traverse downwards using decreaseTrackedResourceUsageUpwards
+func (qt *QueueTracker) decreaseTrackedResourceUsageDownwards(queuePath string) map[string]bool {
+ childQueueTracker := qt.getChildQueueTracker(queuePath)
+ childQueueTrackers := childQueueTracker.childQueueTrackers
+ removedApplications := make(map[string]bool)
+ for _, childQT := range childQueueTrackers {
+ if len(childQT.runningApplications) > 0 && childQT.resourceUsage != resources.NewResource() {
+ removedApplications = childQT.runningApplications
+ childQT.resourceUsage = resources.NewResource()
+ childQT.runningApplications = make(map[string]bool)
+ childQT.decreaseTrackedResourceUsageDownwards(childQT.queuePath)
+ }
+ }
+ qt.decreaseTrackedResourceUsageUpwards(queuePath)
+ return removedApplications
+}
+
+// decreaseTrackedResourceUsageUpwards Traverse upwards all the way upto the root starting from last queue in queuePath,
+// reset resourceUsage and runningApplications to the default value.
+func (qt *QueueTracker) decreaseTrackedResourceUsageUpwards(queuePath string) {
+ childQueuePath, immediateChildQueueName := getChildQueuePath(queuePath)
+ if childQueuePath != common.Empty {
+ 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))
+ }
+ qt.childQueueTrackers[immediateChildQueueName].decreaseTrackedResourceUsageUpwards(childQueuePath)
+ }
+ if len(qt.runningApplications) > 0 && qt.resourceUsage != resources.NewResource() {
+ qt.resourceUsage = resources.NewResource()
+ qt.runningApplications = make(map[string]bool)
+ }
+}
diff --git a/pkg/scheduler/ugm/user_tracker.go b/pkg/scheduler/ugm/user_tracker.go
index 0dcb4b95..f0c86486 100644
--- a/pkg/scheduler/ugm/user_tracker.go
+++ b/pkg/scheduler/ugm/user_tracker.go
@@ -63,7 +63,7 @@ func (ut *UserTracker) increaseTrackedResource(queuePath, applicationID string,
zap.String("queue path", queuePath),
zap.String("application", applicationID),
zap.Stringer("resource", usage))
- increasedGroupUsage := gt.increaseTrackedResource(queuePath, applicationID, usage)
+ increasedGroupUsage := gt.increaseTrackedResource(queuePath, applicationID, usage, ut.userName)
if !increasedGroupUsage {
_, decreased := ut.queueTracker.decreaseTrackedResource(queuePath, applicationID, usage, false)
if !decreased {
diff --git a/pkg/scheduler/utilities_test.go b/pkg/scheduler/utilities_test.go
index d3bf9683..e4eb9e97 100644
--- a/pkg/scheduler/utilities_test.go
+++ b/pkg/scheduler/utilities_test.go
@@ -426,6 +426,10 @@ func newApplication(appID, partition, queueName string) *objects.Application {
User: "testuser",
Groups: []string{"testgroup"},
}
+ return newApplicationWithUser(appID, partition, queueName, user)
+}
+
+func newApplicationWithUser(appID, partition, queueName string, user security.UserGroup) *objects.Application {
siApp := &si.AddApplicationRequest{
ApplicationID: appID,
QueueName: queueName,
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@yunikorn.apache.org
For additional commands, e-mail: issues-help@yunikorn.apache.org