wilfred-s commented on code in PR #563:
URL: https://github.com/apache/yunikorn-core/pull/563#discussion_r1236305672
##########
pkg/scheduler/ugm/manager.go:
##########
@@ -35,23 +35,24 @@ var m *Manager
const maxresources = "maxresources"
const maxapplications = "maxapplications"
+const wildcard = "*"
// 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]map[string]interface{} // Hold
limits settings of user '*'
+ groupWildCardLimitsConfig map[string]map[string]interface{} // Hold
limits settings of group '*'
Review Comment:
Simplify this: introduce a struct type that has two entries. Resource and
uint64 to store the wildcard settings in. Store that struct in the map. That
safes one indirection and the continuous conversion of the limit object's
`map[string]string` to a resource.
##########
pkg/scheduler/ugm/queue_tracker.go:
##########
@@ -31,38 +31,141 @@ 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.Logger().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.Logger().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.Logger().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())
+ return false
+ }
+
+ finalResourceUsage := qt.resourceUsage.Clone()
+ finalResourceUsage.AddTo(usage)
+ wildCardQuotaExceeded := false
+
+ // Are there any settings for specific user/group? If not, try wild
card settings
+ if int(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 map[string]interface{}
+ if trackType == user {
+ config = m.getUserWildCardLimitsConfig(qt.queuePath)
+ } else if trackType == group {
+ config = m.getGroupWildCardLimitsConfig(qt.queuePath)
+ }
+ if config != nil {
+ var maxResources map[string]string
+ var maxApplications uint64
+ var ok bool
+ if maxApplications, ok =
config[maxapplications].(uint64); !ok {
+ log.Logger().Debug("Problem in using the wild
card limit max application settings.",
+ zap.String("queue path", queuePath),
+ zap.Int("tracking type",
int(trackType)),
+ zap.Any("max applications ",
config[maxapplications]))
+ }
+ if maxResources, ok =
config[maxresources].(map[string]string); !ok {
+ log.Logger().Debug("Problem in using the wild
card limit max resources settings.",
+ zap.String("queue path", queuePath),
+ zap.Int("tracking type",
int(trackType)),
+ zap.Any("max applications ",
config[maxresources]))
+ }
+ var maxResource *resources.Resource
+ var err error
+ if maxResource, err =
resources.NewResourceFromConf(maxResources); err != nil {
+ log.Logger().Debug("Problem in using the wild
card limit max resources settings.",
+ zap.String("queue path", queuePath),
+ zap.String("max resources after
casting", maxResource.String()),
+ zap.Error(err))
+ return false
+ }
+ wildCardQuotaExceeded = (maxApplications != 0 &&
len(qt.runningApplications)+1 > int(maxApplications)) ||
(!resources.Equals(resources.NewResource(), maxResource) &&
resources.StrictlyGreaterThan(finalResourceUsage, maxResource))
+ log.Logger().Debug("using wild card limit settings as
user/group specific limit settings not set",
+ zap.Int("tracking type", int(trackType)),
+ zap.String("queue path", queuePath),
+ zap.Uint64("wild card max running apps",
maxApplications),
+ zap.String("wild card max resources",
maxResource.String()),
+ zap.Bool("wild card quota exceeded",
wildCardQuotaExceeded))
+ }
Review Comment:
return here if not passed: wildcard has been checked and no specific quota
was set.
##########
pkg/scheduler/ugm/queue_tracker.go:
##########
@@ -48,20 +49,36 @@ func newQueueTracker(queueName string) *QueueTracker {
queueName: queueName,
resourceUsage: resources.NewResource(),
runningApplications: make(map[string]bool),
+ maxResourceUsage: resources.NewResource(),
+ maxRunningApps: 0,
childQueueTrackers: make(map[string]*QueueTracker),
}
return queueTracker
}
-func (qt *QueueTracker) increaseTrackedResource(queuePath string,
applicationID string, usage *resources.Resource) error {
+func (qt *QueueTracker) increaseTrackedResource(queuePath string,
applicationID string, usage *resources.Resource) bool {
log.Logger().Debug("Increasing resource usage",
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())
+ return false
+ }
Review Comment:
agree remove the check
##########
pkg/scheduler/ugm/queue_tracker.go:
##########
@@ -31,38 +31,141 @@ 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.Logger().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.Logger().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.Logger().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())
+ return false
+ }
+
+ finalResourceUsage := qt.resourceUsage.Clone()
+ finalResourceUsage.AddTo(usage)
+ wildCardQuotaExceeded := false
+
Review Comment:
is this a new app to add to the list of running applications?
```
existingApp := runningApplications[appId]
```
Only check maxRunningApps within range if `existingApp == false`
##########
pkg/scheduler/ugm/queue_tracker.go:
##########
@@ -31,38 +31,141 @@ 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.Logger().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.Logger().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.Logger().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())
+ return false
+ }
+
+ finalResourceUsage := qt.resourceUsage.Clone()
+ finalResourceUsage.AddTo(usage)
+ wildCardQuotaExceeded := false
+
+ // Are there any settings for specific user/group? If not, try wild
card settings
+ if int(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 map[string]interface{}
+ if trackType == user {
+ config = m.getUserWildCardLimitsConfig(qt.queuePath)
+ } else if trackType == group {
+ config = m.getGroupWildCardLimitsConfig(qt.queuePath)
+ }
+ if config != nil {
+ var maxResources map[string]string
+ var maxApplications uint64
+ var ok bool
+ if maxApplications, ok =
config[maxapplications].(uint64); !ok {
+ log.Logger().Debug("Problem in using the wild
card limit max application settings.",
+ zap.String("queue path", queuePath),
+ zap.Int("tracking type",
int(trackType)),
+ zap.Any("max applications ",
config[maxapplications]))
+ }
+ if maxResources, ok =
config[maxresources].(map[string]string); !ok {
+ log.Logger().Debug("Problem in using the wild
card limit max resources settings.",
+ zap.String("queue path", queuePath),
+ zap.Int("tracking type",
int(trackType)),
+ zap.Any("max applications ",
config[maxresources]))
+ }
+ var maxResource *resources.Resource
+ var err error
+ if maxResource, err =
resources.NewResourceFromConf(maxResources); err != nil {
+ log.Logger().Debug("Problem in using the wild
card limit max resources settings.",
+ zap.String("queue path", queuePath),
+ zap.String("max resources after
casting", maxResource.String()),
+ zap.Error(err))
+ return false
+ }
+ wildCardQuotaExceeded = (maxApplications != 0 &&
len(qt.runningApplications)+1 > int(maxApplications)) ||
(!resources.Equals(resources.NewResource(), maxResource) &&
resources.StrictlyGreaterThan(finalResourceUsage, maxResource))
+ log.Logger().Debug("using wild card limit settings as
user/group specific limit settings not set",
+ zap.Int("tracking type", int(trackType)),
+ zap.String("queue path", queuePath),
+ zap.Uint64("wild card max running apps",
maxApplications),
+ zap.String("wild card max resources",
maxResource.String()),
+ zap.Bool("wild card quota exceeded",
wildCardQuotaExceeded))
+ }
+ }
+
+ // apply user/group specific limit settings set if configured,
otherwise use wild card limit settings
+ if (int(qt.maxRunningApps) != 0 &&
!resources.Equals(resources.NewResource(), qt.maxResources)) ||
wildCardQuotaExceeded {
Review Comment:
no casting to int just use the uint64
##########
pkg/scheduler/ugm/queue_tracker.go:
##########
@@ -31,38 +31,141 @@ 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.Logger().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.Logger().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.Logger().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())
+ return false
+ }
+
+ finalResourceUsage := qt.resourceUsage.Clone()
+ finalResourceUsage.AddTo(usage)
+ wildCardQuotaExceeded := false
+
+ // Are there any settings for specific user/group? If not, try wild
card settings
+ if int(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 map[string]interface{}
+ if trackType == user {
+ config = m.getUserWildCardLimitsConfig(qt.queuePath)
+ } else if trackType == group {
+ config = m.getGroupWildCardLimitsConfig(qt.queuePath)
+ }
+ if config != nil {
+ var maxResources map[string]string
+ var maxApplications uint64
+ var ok bool
+ if maxApplications, ok =
config[maxapplications].(uint64); !ok {
+ log.Logger().Debug("Problem in using the wild
card limit max application settings.",
+ zap.String("queue path", queuePath),
+ zap.Int("tracking type",
int(trackType)),
+ zap.Any("max applications ",
config[maxapplications]))
+ }
+ if maxResources, ok =
config[maxresources].(map[string]string); !ok {
+ log.Logger().Debug("Problem in using the wild
card limit max resources settings.",
+ zap.String("queue path", queuePath),
+ zap.Int("tracking type",
int(trackType)),
+ zap.Any("max applications ",
config[maxresources]))
+ }
+ var maxResource *resources.Resource
+ var err error
+ if maxResource, err =
resources.NewResourceFromConf(maxResources); err != nil {
+ log.Logger().Debug("Problem in using the wild
card limit max resources settings.",
+ zap.String("queue path", queuePath),
+ zap.String("max resources after
casting", maxResource.String()),
+ zap.Error(err))
+ return false
+ }
+ wildCardQuotaExceeded = (maxApplications != 0 &&
len(qt.runningApplications)+1 > int(maxApplications)) ||
(!resources.Equals(resources.NewResource(), maxResource) &&
resources.StrictlyGreaterThan(finalResourceUsage, maxResource))
+ log.Logger().Debug("using wild card limit settings as
user/group specific limit settings not set",
+ zap.Int("tracking type", int(trackType)),
+ zap.String("queue path", queuePath),
+ zap.Uint64("wild card max running apps",
maxApplications),
+ zap.String("wild card max resources",
maxResource.String()),
+ zap.Bool("wild card quota exceeded",
wildCardQuotaExceeded))
+ }
+ }
+
+ // apply user/group specific limit settings set if configured,
otherwise use wild card limit settings
+ if (int(qt.maxRunningApps) != 0 &&
!resources.Equals(resources.NewResource(), qt.maxResources)) ||
wildCardQuotaExceeded {
+ log.Logger().Debug("applying enforcement checks using limit
settings",
+ zap.Int("tracking type", int(trackType)),
+ zap.String("queue path", queuePath),
+ zap.Uint64("max running apps", qt.maxRunningApps),
+ zap.String("max resources", qt.maxResources.String()),
+ zap.Bool("wild card quota exceeded",
wildCardQuotaExceeded))
+ if (len(qt.runningApplications)+1 > int(qt.maxRunningApps) ||
resources.StrictlyGreaterThan(finalResourceUsage, qt.maxResources)) ||
wildCardQuotaExceeded {
+ log.Logger().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 or wild card user/group",
+ zap.String("queue path", queuePath),
+ zap.Int("tracking type", int(trackType)),
+ zap.Int("current running applications",
len(qt.runningApplications)),
+ zap.Int("max running applications",
int(qt.maxRunningApps)),
+ zap.String("current resource usage",
qt.resourceUsage.String()),
+ zap.String("max resource usage",
qt.maxResources.String()))
+ return false
+ }
+ }
+
+ childQueuePath, immediateChildQueueName := getChildQueuePath(queuePath)
+ if childQueuePath != "" {
+ if qt.childQueueTrackers[immediateChildQueueName] == nil {
+ qt.childQueueTrackers[immediateChildQueueName] =
newQueueTracker(immediateChildQueueName, qt)
+ }
+ result :=
qt.childQueueTrackers[immediateChildQueueName].increaseTrackedResource(childQueuePath,
applicationID, trackType, usage)
+ if !result {
+ return false
+ }
Review Comment:
use
```
if allowed := qt....; !allowed {
return allowed
}
```
--
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]
