wilfred-s commented on code in PR #563:
URL: https://github.com/apache/yunikorn-core/pull/563#discussion_r1239308847
##########
pkg/log/logger.go:
##########
@@ -258,7 +258,7 @@ func UpdateLoggingConfig(config map[string]string) {
// with one that checks the enabled log level first.
func initLoggingConfig(config map[string]string) {
levelMap := make(map[string]zapcore.Level)
- levelMap[nullLogger] = zapcore.InfoLevel
+ levelMap[nullLogger] = zapcore.DebugLevel
Review Comment:
unintended change?
##########
pkg/scheduler/objects/application.go:
##########
@@ -1637,24 +1637,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).Debug("User quota exceeded unexpectedly.",
Review Comment:
This is logged in the ugm and queueTracker already, choice made to log in
queueTracker only
##########
pkg/scheduler/ugm/manager.go:
##########
@@ -475,127 +413,128 @@ 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),
+ zap.Int("New max apps", 0),
+ zap.String("New max resource",
resources.NewResource().String()))
}
- 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),
+ zap.Int("New max apps", 0),
+ zap.String("New max resource",
resources.NewResource().String()))
Review Comment:
they do not tell us anything new they will always be zeros, remove
##########
pkg/scheduler/ugm/manager.go:
##########
@@ -77,112 +84,69 @@ func (m *Manager) IncreaseTrackedResource(queuePath
string, applicationID string
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)
+ return false
}
m.Lock()
defer m.Unlock()
var userTracker *UserTracker
if m.userTrackers[user.User] == nil {
+ log.Log(log.SchedUGM).Debug("User tracker doesn't exists.
Creating user tracker.",
+ zap.String("user", user.User),
+ zap.String("queue path", queuePath),
+ zap.String("application", applicationID),
+ zap.Stringer("resource", usage))
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]
}
- err := userTracker.increaseTrackedResource(queuePath, applicationID,
usage)
- if err != nil {
- log.Log(log.SchedUGM).Error("Problem in increasing the user
resource usage",
+ 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 {
+ log.Log(log.SchedUGM).Warn("Not allowed to increase 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
+ zap.Stringer("resource", usage))
+ return increased
}
Review Comment:
This is logged in the app and queueTracker already, choice made to log in
queueTracker only
##########
pkg/scheduler/ugm/manager.go:
##########
@@ -77,112 +84,69 @@ func (m *Manager) IncreaseTrackedResource(queuePath
string, applicationID string
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)
+ return false
Review Comment:
Change the level of this message to Debug as it shows a bug in our code, not
something that should ever happen.
##########
pkg/scheduler/ugm/manager.go:
##########
@@ -77,112 +84,69 @@ func (m *Manager) IncreaseTrackedResource(queuePath
string, applicationID string
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)
+ return false
}
m.Lock()
defer m.Unlock()
var userTracker *UserTracker
if m.userTrackers[user.User] == nil {
+ log.Log(log.SchedUGM).Debug("User tracker doesn't exists.
Creating user tracker.",
+ zap.String("user", user.User),
+ zap.String("queue path", queuePath),
+ zap.String("application", applicationID),
+ zap.Stringer("resource", usage))
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]
}
- err := userTracker.increaseTrackedResource(queuePath, applicationID,
usage)
- if err != nil {
- log.Log(log.SchedUGM).Error("Problem in increasing the user
resource usage",
+ 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 {
+ log.Log(log.SchedUGM).Warn("Not allowed to increase 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
+ zap.Stringer("resource", usage))
+ 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",
+ 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 {
+ log.Log(log.SchedUGM).Warn("Not allowed to increase the
group resource usage",
Review Comment:
This is logged in the app and queueTracker already, choice made to log in
queueTracker only
##########
pkg/scheduler/ugm/queue_tracker.go:
##########
@@ -33,71 +31,163 @@ type QueueTracker struct {
queueName 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 {
Review Comment:
new signature: newQueueTracker(queuePath, queueName string) *QueueTracker
add the queuePath filed with the full path to the object:
```
qt.queuePath := queuePath + "." + queueName
qt.queueName := queueName
```
the call for the root queue gets an empty path:
```
rootQueue := newQueueTracker("", configs.RootQueue)
```
when we create a new child we use the current path and the childname:
```
qt.childQueueTracker[child] = newQueueTracker(qt.path, childName)
```
Looking up the queue in the wildcard list:
```
wildcardQuota := getUserWildCardLimitsConfig(qt.path)
```
##########
pkg/scheduler/ugm/manager.go:
##########
@@ -195,60 +159,77 @@ 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",
- 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 {
+ 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 {
+ log.Log(log.SchedUGM).Warn("Not allowed to decrease 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))
+ 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 {
+ 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 {
+ log.Log(log.SchedUGM).Warn("Not allowed to decrease the user
resource usage",
Review Comment:
This is logged in the app and queueTracker already, choice made to log in
queueTracker only
##########
pkg/scheduler/ugm/manager.go:
##########
@@ -475,127 +413,128 @@ 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),
+ zap.Int("New max apps", 0),
+ zap.String("New max resource",
resources.NewResource().String()))
Review Comment:
they do not tell us anything new they will always be zeros, remove
--
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]
