wilfred-s commented on code in PR #563:
URL: https://github.com/apache/yunikorn-core/pull/563#discussion_r1237969497
##########
pkg/scheduler/objects/application.go:
##########
@@ -1631,24 +1631,22 @@ 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",
+ if allowed :=
ugm.GetUserManager().IncreaseTrackedResource(sa.queuePath, sa.ApplicationID,
resource, sa.user); !allowed {
+ log.Log(log.SchedUGM).Error("Not allowed to increase the user
resource usage",
Review Comment:
Not an error, debug at best.
Also need to be careful that we do not log the same information in multiple
places. This same thing is logged in ugm as a WARN. Preferably log in ugm and
not the application.
##########
pkg/scheduler/ugm/manager.go:
##########
@@ -33,25 +33,24 @@ import (
var once sync.Once
var m *Manager
-const maxresources = "maxresources"
-const maxapplications = "maxapplications"
+const wildcard = "*"
Review Comment:
This should already be defined somewhere I would assume, if not it has to be
moved to common constants.
##########
pkg/scheduler/ugm/queue_tracker.go:
##########
@@ -31,102 +29,197 @@ import (
type QueueTracker struct {
queueName string
+ parent *QueueTracker
+ 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, nil)
+ return qt
}
-func newQueueTracker(queueName string) *QueueTracker {
- log.Log(log.SchedUGM).Debug("Creating queue tracker object for queue",
- zap.String("queue", queueName))
+func newQueueTracker(queueName string, parent *QueueTracker) *QueueTracker {
queueTracker := &QueueTracker{
queueName: queueName,
+ parent: parent,
resourceUsage: resources.NewResource(),
runningApplications: make(map[string]bool),
+ maxResources: resources.NewResource(),
+ maxRunningApps: 0,
childQueueTrackers: make(map[string]*QueueTracker),
}
+ parentQueueName := ""
+ if parent != nil {
+ queueTracker.queuePath = queueTracker.parent.queuePath +
configs.DOT + queueName
+ parentQueueName = parent.queueName
+ } else if queueName == configs.RootQueue {
+ queueTracker.queuePath = configs.RootQueue
+ }
+ log.Log(log.SchedUGM).Debug("Created queue tracker object for queue",
+ zap.String("queue", queueName),
+ zap.String("parent", parentQueueName),
+ zap.String("queue path ", queueTracker.queuePath))
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(immediateChildQueueName, qt)
+ }
+ 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.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())
- }
- qt.resourceUsage.SubFrom(usage)
- if removeApp {
- delete(qt.runningApplications, applicationID)
- }
- log.Log(log.SchedUGM).Debug("Successfully decreased resource usage",
- zap.String("queue path", queuePath),
- zap.String("application", applicationID),
- zap.Stringer("resource", usage),
- 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
+ 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)
}
} 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)
+ return false, false
}
}
+ 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",
+ zap.String("queue path", queuePath),
+ zap.String("application", applicationID),
+ zap.Stringer("resource", usage),
+ zap.Stringer("total resource after decreasing",
qt.resourceUsage),
+ zap.Int("total applications after decreasing",
len(qt.runningApplications)))
+
// 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("Can queue tracker be removed?",
+ zap.String("queue path ", queuePath),
+ zap.Int("no. of child queue trackers",
len(qt.childQueueTrackers)),
+ zap.Int("no. of running applications",
len(qt.runningApplications)),
+ zap.Bool("current resource usage",
resources.IsZero(qt.resourceUsage)),
+ zap.Uint64("max running applications", qt.maxRunningApps),
+ zap.String("max resources", qt.maxResources.String()),
+ zap.Bool("remove QT", removeQT))
Review Comment:
Lot of overhead: we know exactly when we need to remove,
Create a single DEBUG log line:
```
log.Log(log.SchedUGM).Debug("Remove queue tracker",
zap.String("queue path ", queuePath),
zap.Bool("remove ", removeQT))
```
##########
pkg/scheduler/ugm/manager.go:
##########
@@ -195,60 +159,79 @@ func (m *Manager) DecreaseTrackedResource(queuePath
string, applicationID string
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",
+ log.Log(log.SchedUGM).Debug("Decreasing resource usage for
user",
Review Comment:
turn if around
##########
pkg/scheduler/partition_test.go:
##########
@@ -346,7 +346,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.Multiply(appRes, 0))
Review Comment:
Why is this not a `resource.Zero` ?
##########
pkg/scheduler/ugm/manager.go:
##########
@@ -419,30 +348,40 @@ 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).Debug("Problem in using the wild
card limit max resources settings.",
Review Comment:
Assume all resources can be parsed: the config must have passed the
validation before we get here
##########
pkg/scheduler/ugm/manager.go:
##########
@@ -475,113 +416,119 @@ 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, limit configs.Limit,
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 {
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
- }
+ } else if user != "" {
if err := m.setUserLimits(user, limit, 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, limit configs.Limit,
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 {
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
- }
+ } else if group != "" {
if err := m.setGroupLimits(group, limit, 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.setMaxResources(resources.NewResource(), queuePath)
+ ut.setMaxApplications(0, queuePath)
Review Comment:
Use one function to set both, propagate that into all objects
##########
pkg/scheduler/ugm/queue_tracker.go:
##########
@@ -31,102 +29,197 @@ import (
type QueueTracker struct {
queueName string
+ parent *QueueTracker
+ 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, nil)
+ return qt
}
-func newQueueTracker(queueName string) *QueueTracker {
- log.Log(log.SchedUGM).Debug("Creating queue tracker object for queue",
- zap.String("queue", queueName))
+func newQueueTracker(queueName string, parent *QueueTracker) *QueueTracker {
queueTracker := &QueueTracker{
queueName: queueName,
+ parent: parent,
resourceUsage: resources.NewResource(),
runningApplications: make(map[string]bool),
+ maxResources: resources.NewResource(),
+ maxRunningApps: 0,
childQueueTrackers: make(map[string]*QueueTracker),
}
+ parentQueueName := ""
+ if parent != nil {
+ queueTracker.queuePath = queueTracker.parent.queuePath +
configs.DOT + queueName
+ parentQueueName = parent.queueName
+ } else if queueName == configs.RootQueue {
+ queueTracker.queuePath = configs.RootQueue
+ }
+ log.Log(log.SchedUGM).Debug("Created queue tracker object for queue",
+ zap.String("queue", queueName),
+ zap.String("parent", parentQueueName),
+ zap.String("queue path ", queueTracker.queuePath))
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(immediateChildQueueName, qt)
+ }
+ 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.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())
- }
- qt.resourceUsage.SubFrom(usage)
- if removeApp {
- delete(qt.runningApplications, applicationID)
- }
- log.Log(log.SchedUGM).Debug("Successfully decreased resource usage",
- zap.String("queue path", queuePath),
- zap.String("application", applicationID),
- zap.Stringer("resource", usage),
- 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
+ removeQT, decreased :=
qt.childQueueTrackers[immediateChildQueueName].decreaseTrackedResource(childQueuePath,
applicationID, usage, removeApp)
Review Comment:
Change to == nil and directly return logging the message
##########
pkg/scheduler/ugm/queue_tracker.go:
##########
@@ -201,11 +274,75 @@ 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)
usage.Children = append(usage.Children, childUsage)
}
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 {
Review Comment:
Parent not needed if we use recursive checks.
root.parent.**child**.leaf
-> slice of [root, parent, child, leaf]
```
manager -> user.rootQT.unlinkChild(path []slice)
root.unlinkChild([parent, child, leaf])
root -> find parent
parent.unlinkChild([child.leaf])
parent -> find child
child.unlinkQT([leaf])
if slice.len ==1 leaf can be removed?
delete(qt.childQueueTrackers, leaf.queueName)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
