This is an automated email from the ASF dual-hosted git repository.
manirajv06 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 902ba02c [YUNIKORN-3313] expose UserGroupResolver type in REST (#1101)
902ba02c is described below
commit 902ba02cb7c4a56fec492ada775f2fa8cd34cf6f
Author: Wilfred Spiegelenburg <[email protected]>
AuthorDate: Thu Jul 9 15:30:30 2026 +0530
[YUNIKORN-3313] expose UserGroupResolver type in REST (#1101)
- Expose the configuration for the UserGroupResolver in the Partition rest
object. Adding the type retrieval from the resolver into the resolver
and partition code.
- Cleanup of the UserGroupCache code, function comments, possible nil
pointer
- Additional unit tests for resolver create and update
- Extend handler tests for resolver check
- Fix context cleanup in handler tests
- Different tests setup new cluster contexts. None of the tests stop the
background processing or clean up the created globals before dropping
the reference to the context.
- The UserGroupCache is protected by a once(). The cache is created when
the partition is created. It cannot be replaced. Depending on the test
run ordering the created partitions might not have the expected resolver
in the cache. Update testing for the cache will also fail.
Update all tests that create a new partition to stop the UserGroupCache.
Stopping the cache will make sure the next partition has the expected
resolver.
Closes: #1101
Signed-off-by: mani <[email protected]>
---
pkg/common/security/ldap_validator_test.go | 2 +-
pkg/common/security/usergroup.go | 52 ++--
pkg/common/security/usergroup_test.go | 32 ++-
pkg/scheduler/context_test.go | 18 ++
pkg/scheduler/partition.go | 25 +-
pkg/scheduler/partition_manager_test.go | 6 +
pkg/scheduler/partition_test.go | 424 ++++++++++++++++++++---------
pkg/scheduler/scheduler_test.go | 3 +
pkg/webservice/dao/partition_info.go | 5 +
pkg/webservice/handlers.go | 3 +
pkg/webservice/handlers_test.go | 68 ++++-
11 files changed, 452 insertions(+), 186 deletions(-)
diff --git a/pkg/common/security/ldap_validator_test.go
b/pkg/common/security/ldap_validator_test.go
index c9f7f7f0..bd42854c 100644
--- a/pkg/common/security/ldap_validator_test.go
+++ b/pkg/common/security/ldap_validator_test.go
@@ -719,7 +719,7 @@ func TestValidateSecretValue(t *testing.T) {
{"Valid bindPassword", common.LdapBindPassword, "password",
false},
{"Valid insecure", common.LdapInsecure, "true", false},
{"Valid SSL", common.LdapSSL, "false", false},
- {"Invalid key", "unknown", "value", true},
+ {"Invalid key", unknown, "value", true},
}
for _, tt := range tests {
diff --git a/pkg/common/security/usergroup.go b/pkg/common/security/usergroup.go
index 956492e3..f05b6766 100644
--- a/pkg/common/security/usergroup.go
+++ b/pkg/common/security/usergroup.go
@@ -45,10 +45,11 @@ var instance *UserGroupCache // The instance of the cache
var once = &sync.Once{} // Make sure we can only create the cache once
var stopped atomic.Bool // whether UserGroupCache is stopped (needed for
multiple partitions)
-// Cache for the user entries.
+// UserGroupCache for the user entries.
type UserGroupCache struct {
lock locking.RWMutex
interval time.Duration
+ myType string
ugs map[string]*UserGroup
// methods that allow mocking of the class or extending to use non OS
solutions
lookup func(userName string) (*user.User, error)
@@ -57,7 +58,7 @@ type UserGroupCache struct {
stop chan struct{}
}
-// The structure of the entry in the cache.
+// UserGroup structure of the entry in the cache.
type UserGroup struct {
User string
Groups []string
@@ -66,34 +67,37 @@ type UserGroup struct {
}
const (
- Default = ""
- Ldap = "ldap"
- Test = "test"
- Os = "os"
+ defType = ""
+ ldapType = "ldap"
+ testType = "test"
+ osType = "os"
)
-// Get the resolver for the user and group info.
+// GetUserGroupCache returns the resolver for the user and group info.
// Current setup allows three resolvers:
// * NO resolver: default, no user or group resolution just return the info
(k8s use case)
// * OS resolver: uses the OS libraries to resolve user and group memberships
// * Test resolver: fake resolution for testing
// * Ldap resolver: uses the LDAP protocol to resolve user and group
memberships
func GetUserGroupCache(ugr configs.UserGroupResolver, ldapConfigReader
ConfigReader, ldapAccess LdapAccess) *UserGroupCache {
- resolver := ugr.Type
once.Do(func() {
- switch resolver {
- case Test:
+ switch ugr.Type {
+ case testType:
log.Log(log.Security).Info("creating test user group
resolver")
instance = GetUserGroupCacheTest()
- case Os:
+ instance.myType = testType
+ case osType:
log.Log(log.Security).Info("creating OS user group
resolver")
instance = GetUserGroupCacheOS()
- case Ldap:
+ instance.myType = osType
+ case ldapType:
log.Log(log.Security).Info("creating LDAP user group
resolver")
instance = GetUserGroupCacheLdap(ldapConfigReader,
ldapAccess)
+ instance.myType = ldapType
default:
log.Log(log.Security).Info("creating UserGroupCache
without resolver")
instance = GetUserGroupNoResolve()
+ instance.myType = defType // do not use the type from
the config as it might not be clean.
}
instance.ugs = make(map[string]*UserGroup)
log.Log(log.Security).Info("starting UserGroupCache cleaner",
@@ -104,7 +108,14 @@ func GetUserGroupCache(ugr configs.UserGroupResolver,
ldapConfigReader ConfigRea
return instance
}
-// Run the cleanup in a separate routine
+// GetResolverType returns the type of resolver configured
+func (c *UserGroupCache) GetResolverType() string {
+ c.lock.RLock()
+ defer c.lock.RUnlock()
+ return c.myType
+}
+
+// run the cleanup in a separate routine
func (c *UserGroupCache) run() {
log.Log(log.Security).Info("Starting user/group cache cleaner")
for {
@@ -120,7 +131,7 @@ func (c *UserGroupCache) run() {
}
}
-// Do the real work for the cache cleanup
+// cleanUpCache clears expired entries from the cache.
func (c *UserGroupCache) cleanUpCache() {
oldest := time.Now().Unix() - poscache
oldestFailed := time.Now().Unix() - negcache
@@ -136,7 +147,7 @@ func (c *UserGroupCache) cleanUpCache() {
}
}
-// reset the cached content, test use only
+// resetCache clears the cached content, test use only
func (c *UserGroupCache) resetCache() {
log.Log(log.Security).Debug("UserGroupCache reset")
instance.lock.Lock()
@@ -149,8 +160,10 @@ func (c *UserGroupCache) ConvertUGI(ugi
*si.UserGroupInformation, force bool) (U
if ugi == nil || ugi.User == "" {
if force {
// app creation is forced, so we need to synthesize a
user / group
- ugi.User = common.AnonymousUser
- ugi.Groups = []string{common.AnonymousGroup}
+ ugi = &si.UserGroupInformation{
+ User: common.AnonymousUser,
+ Groups: []string{common.AnonymousGroup},
+ }
} else {
return UserGroup{}, fmt.Errorf("empty user cannot
resolve")
}
@@ -179,7 +192,7 @@ func (c *UserGroupCache) ConvertUGI(ugi
*si.UserGroupInformation, force bool) (U
return newUG, nil
}
-// Get the user group information. An error will still return a UserGroup.
+// GetUserGroup get the user group information for a singe user. An error will
still return a UserGroup.
// The Failed flag in the object will be set to true for any failures.
// The information is cached, negatively and positively.
func (c *UserGroupCache) GetUserGroup(userName string) (UserGroup, error) {
@@ -236,6 +249,7 @@ func (c *UserGroupCache) GetUserGroup(userName string)
(UserGroup, error) {
return *ug, err
}
+// Stop the currently running cache cleaner and reset the resolver.
func (c *UserGroupCache) Stop() {
// make sure that in case of multiple partitions, we call Stop() only
once (the instance is shared)
// see ClusterContext.Stop()
@@ -254,7 +268,7 @@ func (c *UserGroupCache) Stop() {
log.Log(log.Security).Info("UserGroupCache already stopped")
}
-// Resolve the groups for the user if the user exists
+// resolveGroups resolves the groups for the user if the user exists and
updates the cache.
func (ug *UserGroup) resolveGroups(osUser *user.User, c *UserGroupCache) error
{
// resolve the primary group and add it first
groupName, err := c.lookupGroupID(osUser.Gid)
diff --git a/pkg/common/security/usergroup_test.go
b/pkg/common/security/usergroup_test.go
index 1749ccd9..d2bf478b 100644
--- a/pkg/common/security/usergroup_test.go
+++ b/pkg/common/security/usergroup_test.go
@@ -32,6 +32,8 @@ import (
"github.com/apache/yunikorn-scheduler-interface/lib/go/si"
)
+const unknown = "unknown"
+
func (c *UserGroupCache) getUGsize() int {
c.lock.RLock()
defer c.lock.RUnlock()
@@ -63,7 +65,7 @@ var osResolver = configs.UserGroupResolver{
// UserGroupResolver Config for the unknown resolver
var unknownResolver = configs.UserGroupResolver{
- Type: "unknown",
+ Type: unknown,
}
// UserGroupResolver Config for the LDAP resolver
@@ -71,6 +73,9 @@ var ldapResolver = configs.UserGroupResolver{
Type: "ldap",
}
+// UserGroupResolver Config for the LDAP resolver
+var defResolver = configs.UserGroupResolver{}
+
func TestGetUserGroupCache(t *testing.T) {
testCases := []struct {
name string
@@ -92,6 +97,10 @@ func TestGetUserGroupCache(t *testing.T) {
name: "LdapResolver",
resolver: ldapResolver,
},
+ {
+ name: "DefaultResolver",
+ resolver: defResolver,
+ },
}
for _, tc := range testCases {
@@ -100,6 +109,11 @@ func TestGetUserGroupCache(t *testing.T) {
testCache := GetUserGroupCache(tc.resolver,
&ConfigReaderMock{}, &LdapAccessMock{})
assert.Assert(t, testCache != nil, "Cache create
failed")
assert.Equal(t, 0, testCache.getUGsize(), "Cache is not
empty: %v", testCache.getUGmap())
+ currentType := tc.resolver.Type
+ if tc.resolver.Type == unknown {
+ currentType = defType
+ }
+ assert.Equal(t, testCache.GetResolverType(),
currentType, "Cache type is not correct")
testCache.Stop()
assert.Assert(t, instance == nil, "instance should be
nil")
@@ -230,7 +244,7 @@ func TestBrokenUserGroup(t *testing.T) {
assert.Equal(t, 2, testCache.getUGsize(), "Cache not
updated should have 2 entries %d", len(testCache.ugs))
assert.Equal(t, 4,
testCache.getUGGroupSize("testuser3"), "User 'testuser3' not resolved
correctly: duplicate primary group not filtered %v", ug)
- ug, err = testCache.GetUserGroup("unknown")
+ ug, err = testCache.GetUserGroup(unknown)
assert.ErrorContains(t, err, "lookup failed for user:
unknown")
ug, err = testCache.GetUserGroup("testuser4")
@@ -289,7 +303,7 @@ func TestGetUserGroupFail(t *testing.T) {
// resolve a non existing user
ugi := &si.UserGroupInformation{
- User: "unknown",
+ User: unknown,
Groups: nil,
}
ug, err = testCache.GetUserGroup(ugi.User)
@@ -363,12 +377,12 @@ func TestCacheCleanUp(t *testing.T) {
testCache.lock.Unlock()
// resolve a non existing user
- _, err = testCache.GetUserGroup("unknown")
+ _, err = testCache.GetUserGroup(unknown)
if err == nil {
t.Error("Lookup should have failed: unknown
user")
}
testCache.lock.Lock()
- ug = testCache.ugs["unknown"]
+ ug = testCache.ugs[unknown]
if !ug.failed {
t.Error("User 'unknown' not resolved as a
failure")
}
@@ -427,10 +441,10 @@ func TestIntervalCacheCleanUp(t *testing.T) {
testCache.lock.Unlock()
// resolve a non existing user
- _, err = testCache.GetUserGroup("unknown")
+ _, err = testCache.GetUserGroup(unknown)
assert.Assert(t, err != nil, "Lookup should have
failed: unknown user")
testCache.lock.Lock()
- ug = testCache.ugs["unknown"]
+ ug = testCache.ugs[unknown]
assert.Assert(t, ug.failed, "User 'unknown' not
resolved as a failure")
// expire the failed lookup
@@ -492,7 +506,7 @@ func TestConvertUGI(t *testing.T) {
t.Errorf("User 'testuser1' not resolved
correctly: %v", ug)
}
// try unknown user without groups
- ugi.User = "unknown"
+ ugi.User = unknown
ug, err = testCache.ConvertUGI(ugi, false)
if err == nil {
t.Errorf("unknown user, no groups, convert
should have failed: %v", ug)
@@ -533,7 +547,7 @@ func TestConvertUGI(t *testing.T) {
}
// try unknown user with empty group when forced
- ugi.User = "unknown"
+ ugi.User = unknown
ugi.Groups = []string{}
ug, err = testCache.ConvertUGI(ugi, true)
exceptedGroup := []string{common.AnonymousGroup}
diff --git a/pkg/scheduler/context_test.go b/pkg/scheduler/context_test.go
index f5e4196a..e7ce92c5 100644
--- a/pkg/scheduler/context_test.go
+++ b/pkg/scheduler/context_test.go
@@ -94,6 +94,9 @@ func createTestContext(t *testing.T, partitionName string)
*ClusterContext {
func TestContext_UpdateNode(t *testing.T) {
context := createTestContext(t, pName)
+ // stop the context background processes
+ defer context.Stop()
+
n := &si.NodeInfo{
NodeID: "test-1",
Action: si.NodeInfo_UNKNOWN_ACTION_FROM_RM,
@@ -144,6 +147,8 @@ func TestContext_UpdateNode(t *testing.T) {
func TestContext_AddNode(t *testing.T) {
context := createTestContext(t, pName)
+ // stop the context background processes
+ defer context.Stop()
n := &si.NodeInfo{
NodeID: "test-1",
@@ -172,6 +177,8 @@ func TestContext_AddNode(t *testing.T) {
func TestContext_AddNodeDrained(t *testing.T) {
context := createTestContext(t, pName)
+ // stop the context background processes
+ defer context.Stop()
draining, err := metrics.GetSchedulerMetrics().GetDrainingNodes()
assert.NilError(t, err, "failed to get draining node count")
@@ -211,6 +218,8 @@ func TestContext_AddNodeDrained(t *testing.T) {
func TestContext_AddRMBuildInformation(t *testing.T) {
context := createTestContext(t, pName)
+ // stop the context background processes
+ defer context.Stop()
rmID1 := "myCluster1"
buildInfoMap1 := make(map[string]string)
@@ -254,6 +263,8 @@ func TestContext_ProcessNode(t *testing.T) {
}
}()
context := createTestContext(t, pName)
+ // stop the context background processes
+ defer context.Stop()
request := &si.NodeRequest{
Nodes: []*si.NodeInfo{
@@ -271,6 +282,8 @@ func TestContext_ProcessNode(t *testing.T) {
func TestContextDrainingNodeMetrics(t *testing.T) {
metrics.GetSchedulerMetrics().Reset()
context := createTestContext(t, pName)
+ // stop the context background processes
+ defer context.Stop()
n := getNodeInfoForAddingNode()
err := context.addNode(n, true)
@@ -284,6 +297,8 @@ func TestContextDrainingNodeMetrics(t *testing.T) {
func TestContextDrainingNodeBackToSchedulableMetrics(t *testing.T) {
metrics.GetSchedulerMetrics().Reset()
context := createTestContext(t, pName)
+ // stop the context background processes
+ defer context.Stop()
n := getNodeInfoForAddingNode()
err := context.addNode(n, true)
@@ -299,6 +314,9 @@ func TestContextDrainingNodeBackToSchedulableMetrics(t
*testing.T) {
func TestContext_OnAllocationNotification(t *testing.T) {
context := createTestContext(t, pName)
+ // stop the context background processes
+ defer context.Stop()
+
eventHandler := context.rmEventHandler.(*mockEventHandler)
//nolint:errcheck
var lastAllocEvent *rmevent.RMNewAllocationsEvent
eventHandler.newAllocHandler = func(event
*rmevent.RMNewAllocationsEvent) {
diff --git a/pkg/scheduler/partition.go b/pkg/scheduler/partition.go
index b9489787..28a4859f 100644
--- a/pkg/scheduler/partition.go
+++ b/pkg/scheduler/partition.go
@@ -73,10 +73,10 @@ type PartitionContext struct {
// Scheduling is running continuously as a lock free background task.
Scheduling an application
// acquires a write lock of the application object. While holding the
write lock a list of nodes is
// requested from the partition. This requires a read lock on the
partition.
- // If the partition write lock is held while manipulating an
application a dead lock could occur.
+ // If the partition write lock is held while manipulating an
application a deadlock could occur.
// Since application objects handle their own locks there is no
requirement to hold the partition lock
// while manipulating the application.
- // Similarly adding, updating or removing a node or a queue should only
hold the partition write lock
+ // Similarly, adding, updating or removing a node or a queue should
only hold the partition write lock
// while manipulating the partition information not while manipulating
the underlying objects.
locking.RWMutex
}
@@ -108,7 +108,14 @@ func newPartitionContext(conf configs.PartitionConfig,
rmID string, cc *ClusterC
return pc, nil
}
-// Initialise the partition.
+// GetUserGroupResolverType returns the user group resolver set for the
partition.
+func (pc *PartitionContext) GetUserGroupResolverType() string {
+ pc.RLock()
+ defer pc.RUnlock()
+ return pc.userGroupCache.GetResolverType()
+}
+
+// initialPartitionFromConfig initialises a new partition.
// If the silence flag is set to true, the function will not log queue
creation or node sorting policy, update limit settings, or send a queue event.
func (pc *PartitionContext) initialPartitionFromConfig(conf
configs.PartitionConfig, silence bool) error {
if len(conf.Queues) == 0 || conf.Queues[0].Name != configs.RootQueue {
@@ -568,7 +575,7 @@ func (pc *PartitionContext) createQueue(name string, user
security.UserGroup) (*
return queue, nil
}
-// Get a node from the partition by nodeID.
+// GetNode from the partition by nodeID.
func (pc *PartitionContext) GetNode(nodeID string) *objects.Node {
return pc.nodes.GetNode(nodeID)
}
@@ -1024,13 +1031,15 @@ func (pc *PartitionContext) unReserve(app
*objects.Application, node *objects.No
zap.Int("reservationsRemoved", num))
}
-// Create an ordered node iterator based on the node sort policy set for this
partition.
+// GetNodeIterator returns a node iterator with nodes ordered based on the
node sort policy set for this partition.
+// Reserved nodes are filtered before the iterator is build.
// The iterator is nil if there are no unreserved nodes available.
func (pc *PartitionContext) GetNodeIterator() objects.NodeIterator {
return pc.nodes.GetNodeIterator()
}
-// Create an ordered node iterator based on the node sort policy set for this
partition.
+// GetFullNodeIterator returns a node iterator with nodes ordered based on the
node sort policy set for this partition.
+// Reserved nodes are not filtered before the iterator is build.
// The iterator is nil if there are no nodes available.
func (pc *PartitionContext) GetFullNodeIterator() objects.NodeIterator {
return pc.nodes.GetFullNodeIterator()
@@ -1103,7 +1112,7 @@ func (pc *PartitionContext) GetRejectedApplications()
[]*objects.Application {
func (pc *PartitionContext) getAppsState(appMap
map[string]*objects.Application, state string) []string {
pc.RLock()
defer pc.RUnlock()
- apps := []string{}
+ var apps []string
for appID, app := range appMap {
if app.CurrentState() == state {
apps = append(apps, appID)
@@ -1399,7 +1408,7 @@ func (pc *PartitionContext) getOrStoreForeignAlloc(alloc
*objects.Allocation) bo
}
// calculate overall nodes resource usage and returns a map as the result,
-// where the key is the resource name, e.g memory, and the value is a []int,
+// where the key is the resource name, e.g. memory, and the value is a []int,
// which is a slice with 10 elements,
// each element represents a range of resource usage,
// such as
diff --git a/pkg/scheduler/partition_manager_test.go
b/pkg/scheduler/partition_manager_test.go
index 08cecf40..76fdccbe 100644
--- a/pkg/scheduler/partition_manager_test.go
+++ b/pkg/scheduler/partition_manager_test.go
@@ -48,6 +48,8 @@ func createPartitionContext(t *testing.T) *PartitionContext {
func TestStopPartitionManager(t *testing.T) {
p := createPartitionContext(t)
+ // stop the resolver
+ defer p.userGroupCache.Stop()
p.partitionManager.Stop()
@@ -60,6 +62,8 @@ func TestStopPartitionManager(t *testing.T) {
func TestCleanQueues(t *testing.T) {
p := createPartitionContext(t)
+ // stop the resolver
+ defer p.userGroupCache.Stop()
root := p.GetQueue("root")
assert.Assert(t, root != nil)
@@ -77,6 +81,8 @@ func TestCleanQueues(t *testing.T) {
func TestRemoveAll(t *testing.T) {
p := createPartitionContext(t)
+ // stop the resolver
+ defer p.userGroupCache.Stop()
_, err := p.createQueue("root.test", security.UserGroup{})
assert.NilError(t, err)
diff --git a/pkg/scheduler/partition_test.go b/pkg/scheduler/partition_test.go
index 49949786..f832f2b4 100644
--- a/pkg/scheduler/partition_test.go
+++ b/pkg/scheduler/partition_test.go
@@ -62,21 +62,7 @@ func setupNode(t *testing.T, nodeID string, partition
*PartitionContext, nodeRes
}
func TestNewPartition(t *testing.T) {
- partition, err := newPartitionContext(configs.PartitionConfig{}, "",
nil, false)
- if err == nil || partition != nil {
- t.Fatal("nil inputs should not have returned partition")
- }
- conf := configs.PartitionConfig{Name: "test"}
- partition, err = newPartitionContext(conf, "", nil, false)
- if err == nil || partition != nil {
- t.Fatal("named partition without RM should not have returned
partition")
- }
- partition, err = newPartitionContext(conf, "test", &ClusterContext{},
false)
- if err == nil || partition != nil {
- t.Fatal("partition without root queue should not have returned
partition")
- }
-
- conf = configs.PartitionConfig{
+ conf := configs.PartitionConfig{
Name: "test",
Queues: []configs.QueueConfig{
{
@@ -85,12 +71,8 @@ func TestNewPartition(t *testing.T) {
},
},
}
- partition, err = newPartitionContext(conf, "test", &ClusterContext{},
false)
- if err == nil || partition != nil {
- t.Fatal("partition without root queue should not have returned
partition")
- }
- conf = configs.PartitionConfig{
+ confWithRoot := configs.PartitionConfig{
Name: "test",
Queues: []configs.QueueConfig{
{
@@ -117,10 +99,36 @@ func TestNewPartition(t *testing.T) {
},
},
}
- partition, err = newPartitionContext(conf, "test", &ClusterContext{},
false)
- assert.NilError(t, err, "partition create should not have failed with
error")
- if partition.root.QueuePath != "root" {
- t.Fatal("partition root queue not set as expected")
+ testResolver := confWithRoot
+ testResolver.UserGroupResolver.Type = "test"
+ tests := []struct {
+ name string
+ config configs.PartitionConfig
+ rmID string
+ cc *ClusterContext
+ resolver string
+ wantErr bool
+ }{
+ {"nil", configs.PartitionConfig{}, "", nil, "", true},
+ {"named no RM", configs.PartitionConfig{Name: "test"}, "", nil,
"", true},
+ {"no queue", configs.PartitionConfig{Name: "test"}, "test",
&ClusterContext{}, "", true},
+ {"no root", conf, "test", &ClusterContext{}, "", true},
+ {"OK default resolver", confWithRoot, "test",
&ClusterContext{}, "", false},
+ {"OK test resolver", testResolver, "test", &ClusterContext{},
"test", false},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ partition, err := newPartitionContext(tt.config,
tt.rmID, tt.cc, false)
+ if tt.wantErr {
+ assert.Assert(t, err != nil, "expected error
got nil")
+ assert.Assert(t, partition == nil, "expected
nil partition")
+ return
+ }
+ assert.NilError(t, err, "partition create should not
have failed with error")
+ defer partition.userGroupCache.Stop()
+ assert.Equal(t, partition.root.QueuePath, "root",
"partition root queue not set as expected")
+ assert.Equal(t, partition.GetUserGroupResolverType(),
tt.resolver, "expected resolver type not set based on config")
+ })
}
}
@@ -146,6 +154,7 @@ func TestNewWithPlacement(t *testing.T) {
}
partition, err := newPartitionContext(confWith, rmID, nil, false)
assert.NilError(t, err, "test partition create failed with error")
+ defer partition.userGroupCache.Stop()
// add a rule and check if it is updated
confWith = configs.PartitionConfig{
@@ -195,6 +204,7 @@ func TestNewWithPlacement(t *testing.T) {
func TestAddNode(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "test partition create failed with error")
+ defer partition.userGroupCache.Stop()
err = partition.AddNode(nil)
if err == nil {
t.Fatal("nil node add did not return error")
@@ -230,11 +240,12 @@ func TestRemoveNode(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "test partition create failed with error")
+ defer partition.userGroupCache.Stop()
err = partition.AddNode(newNodeMaxResource("test",
resources.NewResource()))
assert.NilError(t, err, "test node add failed unexpected")
assert.Equal(t, 1, partition.nodes.GetNodeCount(), "node list not
correct")
- // remove non existing node
+ // remove non-existing node
_, _ = partition.removeNode("")
assert.Equal(t, 1, partition.nodes.GetNodeCount(), "nil node should not
remove anything")
_, _ = partition.removeNode("does not exist")
@@ -248,6 +259,7 @@ func TestRemoveNodeWithAllocations(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
defer metrics.GetSchedulerMetrics().Reset()
defer metrics.GetQueueMetrics(defQueue).Reset()
@@ -298,7 +310,7 @@ func TestRemoveNodeWithPlaceholders(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
-
+ defer partition.userGroupCache.Stop()
defer metrics.GetSchedulerMetrics().Reset()
defer metrics.GetQueueMetrics(defQueue).Reset()
@@ -363,6 +375,7 @@ func TestRemoveNodeWithPlaceholders(t *testing.T) {
func TestCalculateNodesResourceUsage(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
oldCapacity :=
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 100})
node := newNodeMaxResource(nodeID1, oldCapacity)
err = partition.AddNode(node)
@@ -399,11 +412,13 @@ func TestCalculateNodesResourceUsage(t *testing.T) {
//
// ensure placeholder has been preempted and released resources has been given
to the request asked for
// ensure preempted placeholder has been accounted under timed out in gang app
placeholder data
+//
+//nolint:funlen
func TestPlaceholderDataWithPlaceholderPreemption(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
-
+ defer partition.userGroupCache.Stop()
defer metrics.GetSchedulerMetrics().Reset()
defer metrics.GetQueueMetrics(defQueue).Reset()
@@ -525,7 +540,7 @@ func TestPlaceholderDataWithPlaceholderPreemption(t
*testing.T) {
// queue quota max size: 16GB / 16cpu
// nodes: 2 * 8GB / 8 cpu
// create an application with allocation: 4 GB / 4 cpu
-// create an gang application requesting: 7 * 2GB / 2cpu
+// create a gang application requesting: 7 * 2GB / 2cpu
// Remove the node where placeholders are running
//
// ensure removed placeholders has been accounted under timed out in gang app
placeholder data
@@ -533,6 +548,7 @@ func TestPlaceholderDataWithNodeRemoval(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
defer metrics.GetSchedulerMetrics().Reset()
defer metrics.GetQueueMetrics(defQueue).Reset()
@@ -611,7 +627,7 @@ func TestPlaceholderDataWithNodeRemoval(t *testing.T) {
// queue quota max size: 16GB / 16cpu
// nodes: 2 * 8GB / 8 cpu
// create an application with allocation: 4 GB / 4 cpu
-// create an gang application requesting: 7 * 2GB / 2cpu
+// create a gang application requesting: 7 * 2GB / 2cpu
// Remove the node where placeholders are running
//
// ensure removed placeholders has been accounted under timed out in gang app
placeholder data
@@ -619,6 +635,7 @@ func TestPlaceholderDataWithRemoval(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
defer metrics.GetSchedulerMetrics().Reset()
defer metrics.GetQueueMetrics(defQueue).Reset()
@@ -716,6 +733,7 @@ func TestRemoveNodeWithReplacement(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
defer metrics.GetSchedulerMetrics().Reset()
defer metrics.GetQueueMetrics(defQueue).Reset()
@@ -790,6 +808,7 @@ func TestRemoveNodeWithReal(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
defer metrics.GetSchedulerMetrics().Reset()
defer metrics.GetQueueMetrics(defQueue).Reset()
@@ -859,6 +878,7 @@ func TestAddApp(t *testing.T) {
defer metrics.GetQueueMetrics(defQueue).Reset()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
// add a new app
app := newApplication(appID1, "default", defQueue)
@@ -913,6 +933,7 @@ func TestAddApp(t *testing.T) {
func TestAddAppForced(t *testing.T) {
partition, err := newBasePartitionNoRootDefault()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
// add a new app to an invalid queue
app := newApplication(appID1, "default", "root.invalid")
@@ -1008,6 +1029,7 @@ func TestAddAppForcedWithPlacement(t *testing.T) {
}
partition, err := newPartitionContext(confWith, rmID, nil, false)
assert.NilError(t, err, "test partition create failed with error")
+ defer partition.userGroupCache.Stop()
// add a new app using tag rule
app := newApplicationTags(appID1, "default", "",
map[string]string{"queue": "root.test"})
@@ -1038,6 +1060,7 @@ func TestAddAppForcedWithPlacement(t *testing.T) {
func TestAddAppTaskGroup(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
// add a new app: TG specified with resource no max set on the queue
task :=
resources.NewResourceFromMap(map[string]resources.Quantity{"vcore": 10000})
@@ -1050,7 +1073,7 @@ func TestAddAppTaskGroup(t *testing.T) {
app = newApplicationTG(appID2, "default", defQueue, task)
assert.Assert(t, resources.Equals(app.GetPlaceholderAsk(), task),
"placeholder ask not set as expected")
- // queue now has fair as sort policy app add should fail
+ // queue now has fair set as the sort policy, app add should fail
queue := partition.GetQueue(defQueue)
_, err = queue.ApplyConf(configs.QueueConfig{
Name: "default",
@@ -1070,6 +1093,7 @@ func TestRemoveApp(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
// add a new app that will just sit around to make sure we remove the
right one
appNotRemoved := "will_not_remove"
@@ -1132,6 +1156,7 @@ func TestRemoveAppAllocs(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
// add a new app that will just sit around to make sure we remove the
right one
appNotRemoved := "will_not_remove"
@@ -1169,7 +1194,7 @@ func TestRemoveAppAllocs(t *testing.T) {
allocs, _ = partition.removeAllocation(release)
assert.Equal(t, 0, len(allocs), "removal request for non existing
application returned allocations: %v", allocs)
assertLimits(t, getTestUserGroup(), resources.Multiply(appRes, 2))
- // create a new release with app, non existing allocation: should just
return
+ // create a new release with app, non-existing allocation: should just
return
release.ApplicationID = appNotRemoved
release.AllocationKey = "does_not_exist"
allocs, _ = partition.removeAllocation(release)
@@ -1195,6 +1220,7 @@ func TestRemoveAllPlaceholderAllocs(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
setupNode(t, nodeID1, partition,
resources.NewResourceFromMap(map[string]resources.Quantity{"vcore": 1000000}))
// add a new app that will just sit around to make sure we remove the
right one
@@ -1225,6 +1251,7 @@ func TestRemoveAllPlaceholderAllocs(t *testing.T) {
func TestCreateQueue(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
// top level should fail
_, err = partition.createQueue("test", security.UserGroup{})
if err == nil {
@@ -1311,6 +1338,7 @@ func TestCreateDeepQueueConfig(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
// There is a queue setup as the config must be valid when we run
root := partition.GetQueue("root")
if root == nil {
@@ -1325,74 +1353,94 @@ func TestCreateDeepQueueConfig(t *testing.T) {
assert.Equal(t, "root.level1.level2.level3.level4.level5",
queue.GetQueuePath(), "root.level1.level2.level3.level4.level5 queue not found
in partition")
}
-func assertUpdateQueues(t *testing.T, resourceType string, resMap
map[string]string) {
- var resExpect *resources.Resource
- var err error
- if len(resMap) > 0 {
- resExpect, err = resources.NewResourceFromConf(resMap)
- assert.NilError(t, err, "resource from conf failed")
- } else {
- resExpect = nil
+func TestUpdateQueues(t *testing.T) {
+ tests := []struct {
+ name string
+ resourceType string
+ resMap map[string]string
+ }{
+ {"single max", "max", map[string]string{"first": "2"}},
+ {"multi max", "max", map[string]string{"first": "4", "second":
"3"}},
+ {"empty max", "max", map[string]string{}},
+ {"single guaranteed", "guaranteed", map[string]string{"third":
"10"}},
+ {"multi guaranteed", "guaranteed", map[string]string{"third":
"2", "fourth": "5"}},
+ {"empty guaranteed", "guaranteed", map[string]string{}},
+ {"single both", "both", map[string]string{"both": "5"}},
+ {"multi both", "both", map[string]string{"bothfirst": "2",
"bothsecond": "5"}},
+ {"empty both", "both", map[string]string{}},
}
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ var resExpect *resources.Resource
+ var err error
+ if len(tt.resMap) > 0 {
+ resExpect, err =
resources.NewResourceFromConf(tt.resMap)
+ assert.NilError(t, err, "resource from conf
failed")
+ } else {
+ resExpect = nil
+ }
- var res configs.Resources
- switch resourceType {
- case "max":
- res = configs.Resources{Max: resMap}
- case "guaranteed":
- res = configs.Resources{Guaranteed: resMap}
- default:
- res = configs.Resources{Max: resMap, Guaranteed: resMap}
- }
+ var res configs.Resources
+ switch tt.resourceType {
+ case "max":
+ res = configs.Resources{Max: tt.resMap}
+ case "guaranteed":
+ res = configs.Resources{Guaranteed: tt.resMap}
+ default:
+ res = configs.Resources{Max: tt.resMap,
Guaranteed: tt.resMap}
+ }
- conf := []configs.QueueConfig{
- {
- Name: "parent",
- Parent: true,
- Resources: res,
- Queues: []configs.QueueConfig{
+ conf := []configs.QueueConfig{
{
- Name: "leaf",
- Parent: false,
- Queues: nil,
+ Name: "parent",
+ Parent: true,
+ Resources: res,
+ Queues: []configs.QueueConfig{
+ {
+ Name: "leaf",
+ Parent: false,
+ Queues: nil,
+ },
+ },
},
- },
- },
- }
+ }
- partition, err := newBasePartition()
- assert.NilError(t, err, "partition create failed")
+ partition, err := newBasePartition()
+ assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
- // There is a queue setup as the config must be valid when we run
- root := partition.GetQueue("root")
- if root == nil {
- t.Error("root queue not found in partition")
- }
+ // There is a queue setup as the config must be valid
when we run
+ root := partition.GetQueue("root")
+ if root == nil {
+ t.Error("root queue not found in partition")
+ }
- err = partition.updateQueues(conf, root)
- assert.NilError(t, err, "queue update from config failed")
- parent := partition.GetQueue("root.parent")
- if parent == nil {
- t.Fatal("parent queue should still exist")
- }
- switch resourceType {
- case "max":
- assert.Assert(t, resources.Equals(parent.GetMaxResource(),
resExpect), "parent queue max resource should have been updated")
- assert.Assert(t,
resources.Equals(parent.GetGuaranteedResource(), nil), "parent queue guaranteed
resource should have been updated")
- case "guaranteed":
- assert.Assert(t, resources.Equals(parent.GetMaxResource(),
nil), "parent queue max resource should have been updated")
- assert.Assert(t,
resources.Equals(parent.GetGuaranteedResource(), resExpect), "parent queue
guaranteed resource should have been updated")
- default:
- assert.Assert(t, resources.Equals(parent.GetMaxResource(),
resExpect), "parent queue max resource should have been updated")
- assert.Assert(t,
resources.Equals(parent.GetGuaranteedResource(), resExpect), "parent queue
guaranteed resource should have been updated")
- }
- leaf := partition.GetQueue("root.parent.leaf")
- if leaf == nil {
- t.Fatal("leaf queue should have been created")
+ err = partition.updateQueues(conf, root)
+ assert.NilError(t, err, "queue update from config
failed")
+ parent := partition.GetQueue("root.parent")
+ if parent == nil {
+ t.Fatal("parent queue should still exist")
+ }
+ switch tt.resourceType {
+ case "max":
+ assert.Assert(t,
resources.Equals(parent.GetMaxResource(), resExpect), "parent queue max
resource should have been updated")
+ assert.Assert(t,
resources.Equals(parent.GetGuaranteedResource(), nil), "parent queue guaranteed
resource should have been updated")
+ case "guaranteed":
+ assert.Assert(t,
resources.Equals(parent.GetMaxResource(), nil), "parent queue max resource
should have been updated")
+ assert.Assert(t,
resources.Equals(parent.GetGuaranteedResource(), resExpect), "parent queue
guaranteed resource should have been updated")
+ default:
+ assert.Assert(t,
resources.Equals(parent.GetMaxResource(), resExpect), "parent queue max
resource should have been updated")
+ assert.Assert(t,
resources.Equals(parent.GetGuaranteedResource(), resExpect), "parent queue
guaranteed resource should have been updated")
+ }
+ leaf := partition.GetQueue("root.parent.leaf")
+ if leaf == nil {
+ t.Fatal("leaf queue should have been created")
+ }
+ })
}
}
-func TestUpdateQueues(t *testing.T) {
+func TestUpdateQueueDefault(t *testing.T) {
conf := []configs.QueueConfig{
{
Name: "parent",
@@ -1403,6 +1451,7 @@ func TestUpdateQueues(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
// There is a queue setup as the config must be valid when we run
root := partition.GetQueue("root")
if root == nil {
@@ -1415,17 +1464,6 @@ func TestUpdateQueues(t *testing.T) {
t.Fatal("default queue should still exist")
}
assert.Assert(t, def.IsDraining(), "'root.default' queue should have
been marked for removal")
-
- assertUpdateQueues(t, "max", map[string]string{"vcore": "2"})
- assertUpdateQueues(t, "max", map[string]string{"vcore": "5"})
- assertUpdateQueues(t, "max", map[string]string{"memory": "5"})
- assertUpdateQueues(t, "guaranteed", map[string]string{"vcore": "2",
"memory": "5"})
- assertUpdateQueues(t, "guaranteed", map[string]string{"vcore": "4",
"memory": "3"})
- assertUpdateQueues(t, "guaranteed", map[string]string{"vcore": "10"})
- assertUpdateQueues(t, "both", map[string]string{"vcore": "2", "memory":
"5"})
- assertUpdateQueues(t, "both", map[string]string{"vcore": "5", "memory":
"2"})
- assertUpdateQueues(t, "both", map[string]string{"vcore": "5"})
- assertUpdateQueues(t, "both", map[string]string{})
}
// TestUpdateQueuesInheritedProperties verifies that a child queue inherits
@@ -1451,6 +1489,7 @@ func TestUpdateQueuesInheritedProperties(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
root := partition.GetQueue("root")
assert.Assert(t, root != nil, "root queue not found")
@@ -1519,6 +1558,7 @@ func TestReAddQueues(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
// There is a queue setup as the config must be valid when we run
+ defer partition.userGroupCache.Stop()
root := partition.GetQueue("root")
if root == nil {
t.Error("root queue not found in partition")
@@ -1549,9 +1589,10 @@ func TestReAddQueues(t *testing.T) {
}
}
-func TestGetApplication(t *testing.T) {
+func TestGetApplicationNoRoot(t *testing.T) {
partition, err := newBasePartitionNoRootDefault()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
app := newApplication(appID1, "default", "root.custom")
err = partition.AddApplication(app)
assert.NilError(t, err, "no error expected while adding the
application")
@@ -1564,18 +1605,22 @@ func TestGetApplication(t *testing.T) {
if partition.GetApplication(appID2) != nil {
t.Fatal("partition added app incorrectly should have failed")
}
-
- partition, err = newBasePartition()
+}
+func TestGetApplication(t *testing.T) {
+ partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
- err = partition.AddApplication(app2)
+ defer partition.userGroupCache.Stop()
+ app := newApplication(appID2, "default", "unknown")
+ err = partition.AddApplication(app)
assert.NilError(t, err, "no error expected while adding the
application")
- assert.Equal(t, partition.GetApplication(appID2), app2, "partition
failed to add app incorrect app returned")
+ assert.Equal(t, partition.GetApplication(appID2), app, "partition
failed to add app incorrect app returned")
}
func TestGetQueue(t *testing.T) {
// get the partition
partition, err := newBasePartition()
assert.NilError(t, err, "test partition create failed with error")
+ defer partition.userGroupCache.Stop()
var nilQueue *objects.Queue
// test partition has a root queue
queue := partition.GetQueue("")
@@ -1604,6 +1649,8 @@ func TestTryAllocate(t *testing.T) {
setupUGM()
partition := createQueuesNodes(t)
assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
if result := partition.tryAllocate(); result != nil {
t.Fatalf("empty cluster allocate returned allocation: %s",
result)
}
@@ -1681,6 +1728,8 @@ func TestRequiredNodeReservation(t *testing.T) {
setupUGM()
partition := createQueuesNodes(t)
assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
node := partition.nodes.GetNode(nodeID1)
if node == nil {
t.Fatal("node-1 should have been created")
@@ -1763,6 +1812,8 @@ func TestRequiredNodeReservation(t *testing.T) {
func TestRequiredNodeCancelOtherReservations(t *testing.T) {
partition := createQueuesNodes(t)
assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
if result := partition.tryAllocate(); result != nil {
t.Fatalf("empty cluster allocate returned allocation: %s",
result)
}
@@ -1841,6 +1892,8 @@ func TestRequiredNodeCancelOtherReservations(t
*testing.T) {
func TestRequiredNodeCancelDSReservations(t *testing.T) {
partition := createQueuesNodes(t)
assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
if result := partition.tryAllocate(); result != nil {
t.Fatalf("empty cluster allocate returned allocation: %s",
result)
}
@@ -1924,6 +1977,8 @@ func TestRequiredNodeNotExist(t *testing.T) {
setupUGM()
partition := createQueuesNodes(t)
assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
if result := partition.tryAllocate(); result != nil {
t.Fatalf("empty cluster allocate returned allocation: %s",
result)
}
@@ -1959,6 +2014,8 @@ func TestRequiredNodeNotExist(t *testing.T) {
func TestRequiredNodeAllocation(t *testing.T) {
partition := createQueuesNodes(t)
assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
if result := partition.tryAllocate(); result != nil {
t.Fatalf("empty cluster allocate returned allocation: %s",
result.Request.String())
}
@@ -2027,6 +2084,7 @@ func assertPreemptedResource(t *testing.T, appSummary
*objects.ApplicationSummar
func TestPreemption(t *testing.T) {
setupUGM()
partition, app1, app2, alloc1, alloc2 := setupPreemption(t)
+ defer partition.userGroupCache.Stop()
res, err := resources.NewResourceFromConf(map[string]string{"vcore":
"5"})
assert.NilError(t, err, "failed to create resource")
@@ -2107,7 +2165,7 @@ func TestPreemptionForRequiredNodeNormalAlloc(t
*testing.T) {
// Preemption followed by a reserved allocation
func TestPreemptionForRequiredNodeReservedAlloc(t *testing.T) {
setupUGM()
- // setup the partition so we can try the real allocation
+ // set up the partition so we can try the real allocation
partition, app := setupPreemptionForRequiredNode(t)
// now try the allocation again: the reserved path
result := partition.tryReservedAllocate()
@@ -2125,6 +2183,7 @@ func TestPreemptionForRequiredNodeReservedAlloc(t
*testing.T) {
func TestPreemptionForRequiredNodeMultipleAttemptsAvoided(t *testing.T) {
partition := createQueuesNodes(t)
assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
app, testHandler := newApplicationWithHandler(appID1, "default",
"root.parent.sub-leaf")
res, err := resources.NewResourceFromConf(map[string]string{"vcore":
"8"})
@@ -2201,7 +2260,7 @@ func
getExpectedQueuesLimitsForPreemptionWithRequiredNode() map[string]map[strin
return expectedQueuesMaxLimits
}
-// setup the partition with existing allocations so we can test preemption
+// set up the partition with existing allocations so we can test preemption
func setupPreemption(t *testing.T) (*PartitionContext, *objects.Application,
*objects.Application, *objects.Allocation, *objects.Allocation) {
partition := createPreemptionQueuesNodes(t)
assert.Assert(t, partition != nil, "partition create failed")
@@ -2261,10 +2320,12 @@ func setupPreemption(t *testing.T) (*PartitionContext,
*objects.Application, *ob
return partition, app1, app2, result1.Request, result2.Request
}
-// setup the partition in a state that we need for multiple tests
+// set up the partition in a state that we need for multiple tests
func setupPreemptionForRequiredNode(t *testing.T) (*PartitionContext,
*objects.Application) {
partition := createQueuesNodes(t)
assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
if result := partition.tryAllocate(); result != nil {
t.Fatalf("empty cluster allocate returned allocation: %s",
result)
}
@@ -2343,6 +2404,8 @@ func TestTryAllocateLarge(t *testing.T) {
setupUGM()
partition := createQueuesNodes(t)
assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
if result := partition.tryAllocate(); result != nil {
t.Fatalf("empty cluster allocate returned allocation: %s",
result)
}
@@ -2374,6 +2437,8 @@ func TestAllocReserveNewNode(t *testing.T) {
setupUGM()
partition := createQueuesNodes(t)
assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
if result := partition.tryAllocate(); result != nil {
t.Fatalf("empty cluster allocate returned result: %s", result)
}
@@ -2443,6 +2508,8 @@ func TestTryAllocateReserve(t *testing.T) {
setupUGM()
partition := createQueuesNodes(t)
assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
if result := partition.tryReservedAllocate(); result != nil {
t.Fatalf("empty cluster reserved allocate returned allocation:
%s", result)
}
@@ -2495,7 +2562,7 @@ func TestTryAllocateReserve(t *testing.T) {
if result != nil {
t.Fatalf("reserved allocation should not return any allocation:
%s", result)
}
- // try non reserved this should allocate
+ // try non-reserved this should allocate
result = partition.tryAllocate()
if result == nil || result.Request == nil {
t.Fatal("allocation did not return any allocation")
@@ -2515,6 +2582,8 @@ func TestTryAllocateWithReserved(t *testing.T) {
setupUGM()
partition := createQueuesNodes(t)
assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
if alloc := partition.tryReservedAllocate(); alloc != nil {
t.Fatalf("empty cluster reserved allocate returned allocation:
%v", alloc)
}
@@ -2566,6 +2635,8 @@ func TestScheduleRemoveReservedAsk(t *testing.T) {
setupUGM()
partition := createQueuesNodes(t)
assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
if result := partition.tryAllocate(); result != nil {
t.Fatalf("empty cluster allocate returned allocation: %s",
result)
}
@@ -2598,7 +2669,7 @@ func TestScheduleRemoveReservedAsk(t *testing.T) {
}
assertLimits(t, getTestUserGroup(),
resources.NewResourceFromMap(map[string]resources.Quantity{"vcore": 16000}))
- // add a asks which should reserve
+ // add an asks which should reserve
ask := newAllocationAsk("alloc-5", appID1, res)
err = app.AddAllocationAsk(ask)
assert.NilError(t, err, "failed to add ask alloc-5 to app")
@@ -2654,6 +2725,8 @@ func TestScheduleRemoveReservedAsk(t *testing.T) {
func TestUpdateRootQueue(t *testing.T) {
partition := createQueuesNodes(t)
assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
res, err := resources.NewResourceFromConf(map[string]string{"vcore":
"20"})
assert.NilError(t, err, "resource creation failed")
assert.Assert(t, resources.Equals(res,
partition.totalPartitionResource), "partition resource not set as expected")
@@ -2753,6 +2826,7 @@ func completeApplicationAndWait(app *objects.Application,
pc *PartitionContext)
func TestCompleteApp(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
app := newApplication("completed", "default", defQueue)
app.SetState(objects.Completing.String())
err = partition.AddApplication(app)
@@ -2769,6 +2843,7 @@ func TestCompleteApp(t *testing.T) {
func TestCleanupFailedApps(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
newApp1 := newApplication("newApp1", "default", defQueue)
newApp2 := newApplication("newApp2", "default", defQueue)
@@ -2789,6 +2864,7 @@ func TestCleanupFailedApps(t *testing.T) {
func TestCleanupCompletedApps(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
completedApp1 := newApplication("completedApp1", "default", defQueue)
completedApp2 := newApplication("completedApp2", "default", defQueue)
@@ -2825,6 +2901,7 @@ func TestCleanupCompletedApps(t *testing.T) {
func TestCleanupRejectedApps(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
rejectedApp := newApplication("new", "default", defQueue)
rejectedMessage := fmt.Sprintf("Failed to place application %s:
application rejected: no placement rule matched", rejectedApp.ApplicationID)
@@ -2850,6 +2927,7 @@ func TestCleanupRejectedApps(t *testing.T) {
func TestUpdateNode(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "test partition create failed with error")
+ defer partition.userGroupCache.Stop()
newRes, err :=
resources.NewResourceFromConf(map[string]string{"memory": "400M", "vcore":
"30"})
assert.NilError(t, err, "failed to create resource")
@@ -2902,21 +2980,25 @@ func TestUpdateNode(t *testing.T) {
assert.Assert(t, partition.GetTotalPartitionResource().IsEmpty())
}
-func TestAddTGApplication(t *testing.T) {
- limit := map[string]string{"vcore": "1"}
- partition, err := newLimitedPartition(limit)
- assert.NilError(t, err, "partition create failed")
- // add a app with TG that does not fit in the queue
- var tgRes *resources.Resource
- tgRes, err = resources.NewResourceFromConf(map[string]string{"vcore":
"10"})
+func tgApp(t *testing.T) *objects.Application {
+ tgRes, err := resources.NewResourceFromConf(map[string]string{"vcore":
"10"})
assert.NilError(t, err, "failed to create resource")
tags := map[string]string{
siCommon.AppTagNamespaceResourceGuaranteed:
"{\"resources\":{\"vcore\":{\"value\":111}}}",
siCommon.AppTagNamespaceResourceQuota:
"{\"resources\":{\"vcore\":{\"value\":2222}}}",
siCommon.AppTagNamespaceResourceMaxApps: "1",
}
- app := newApplicationTGTags(appID1, "default", "root.limited", tgRes,
tags)
- err = partition.AddApplication(app)
+ return newApplicationTGTags(appID1, "default", "root.limited", tgRes,
tags)
+}
+
+func TestAddTGApplication_NotFit(t *testing.T) {
+ limit := map[string]string{"vcore": "1"}
+ partition, err := newLimitedPartition(limit)
+ assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
+ // add an app with TG that does not fit in the queue
+ err = partition.AddApplication(tgApp(t))
if err == nil {
t.Error("app-1 should be rejected due to TG request")
}
@@ -2926,29 +3008,37 @@ func TestAddTGApplication(t *testing.T) {
})), "max resource changed unexpectedly")
assert.Assert(t, queue.GetGuaranteedResource() == nil)
assert.Equal(t, queue.GetMaxApps(), uint64(2), "max running apps should
be 2")
+}
- // add a app with TG that does fit in the queue
- limit = map[string]string{"vcore": "100"}
- partition, err = newLimitedPartition(limit)
+func TestAddTGApplication_Fit(t *testing.T) {
+ // add an app with TG that does fit in the queue
+ limit := map[string]string{"vcore": "100"}
+ partition, err := newLimitedPartition(limit)
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
+ app := tgApp(t)
err = partition.AddApplication(app)
assert.NilError(t, err, "app-1 should have been added to the partition")
assert.Equal(t, partition.getApplication(appID1), app, "partition
failed to add app incorrect app returned")
- queue = partition.GetQueue("root.limited")
+ queue := partition.GetQueue("root.limited")
assert.Assert(t, resources.Equals(queue.GetMaxResource(),
resources.NewResourceFromMap(map[string]resources.Quantity{
"vcore": 100000,
})), "max resource changed unexpectedly")
assert.Assert(t, queue.GetGuaranteedResource() == nil)
assert.Equal(t, queue.GetMaxApps(), uint64(2), "max running apps should
be 2")
+}
- // add a app with TG that does fit in the queue as the resource is not
limited in the queue
- limit = map[string]string{"second": "100"}
- partition, err = newLimitedPartition(limit)
+func TestAddTGApplication_OtherLimit(t *testing.T) {
+ // add an app with TG that does fit in the queue as the resource is not
limited in the queue
+ limit := map[string]string{"second": "100"}
+ partition, err := newLimitedPartition(limit)
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
+ app := tgApp(t)
err = partition.AddApplication(app)
assert.NilError(t, err, "app-1 should have been added to the partition")
assert.Equal(t, partition.getApplication(appID1), app, "partition
failed to add app incorrect app returned")
- queue = partition.GetQueue("root.limited")
+ queue := partition.GetQueue("root.limited")
assert.Assert(t, resources.Equals(queue.GetMaxResource(),
resources.NewResourceFromMap(map[string]resources.Quantity{
"second": 100,
})), "max resource changed unexpectedly")
@@ -2959,7 +3049,9 @@ func TestAddTGApplication(t *testing.T) {
func TestAddTGAppDynamic(t *testing.T) {
partition, err := newPlacementPartition()
assert.NilError(t, err, "partition create failed")
- // add a app with TG that does fit in the dynamic queue (no limit)
+ defer partition.userGroupCache.Stop()
+
+ // add an app with TG that does fit in the dynamic queue (no limit)
var tgRes *resources.Resource
tgRes, err = resources.NewResourceFromConf(map[string]string{"vcore":
"10"})
assert.NilError(t, err, "failed to create resource")
@@ -3027,6 +3119,7 @@ func TestPlaceholderSmallerThanReal(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
tgRes :=
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5,
"second": 5})
phRes :=
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 2,
"second": 2})
@@ -3092,6 +3185,7 @@ func TestPlaceholderSmallerMulti(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
phCount := 5
phRes :=
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 2,
"second": 2})
@@ -3166,6 +3260,7 @@ func TestPlaceholderBiggerThanReal(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
if result := partition.tryPlaceholderAllocate(); result != nil {
t.Fatalf("empty cluster placeholder allocate returned
allocation: %s", result)
}
@@ -3241,6 +3336,7 @@ func TestPlaceholderMatch(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
tgRes :=
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 10,
"second": 10})
phRes :=
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 2,
"second": 2})
@@ -3338,6 +3434,7 @@ func TestPreemptedPlaceholderSkip(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
tgRes :=
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 10,
"second": 10})
phRes :=
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 2,
"second": 2})
@@ -3419,6 +3516,7 @@ func TestTryPlaceholderAllocate(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
if result := partition.tryPlaceholderAllocate(); result != nil {
t.Fatalf("empty cluster placeholder allocate returned
allocation: %s", result)
}
@@ -3546,6 +3644,7 @@ func TestFailReplacePlaceholder(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
if result := partition.tryPlaceholderAllocate(); result != nil {
t.Fatalf("empty cluster placeholder allocate returned
allocation: %s", result)
}
@@ -3642,6 +3741,9 @@ func TestFailReplacePlaceholder(t *testing.T) {
func TestUpdateAllocation(t *testing.T) {
setupUGM()
partition := createQueuesNodes(t)
+ assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
_, allocCreated, err := partition.UpdateAllocation(nil)
assert.NilError(t, err, "nil alloc should not return an error")
assert.Check(t, !allocCreated, "alloc should not have been created")
@@ -3711,6 +3813,8 @@ func TestUpdateAllocation(t *testing.T) {
func TestUpdateAllocationWithQuotaPreemption(t *testing.T) {
setupUGM()
partition := createQuotaPreemptionQueuesNodes(t)
+ defer partition.userGroupCache.Stop()
+
leafQueueConf :=
[]configs.QueueConfig{createLeafQueueConfig(map[string]string{"memory": "5",
"vcore": "5"}, map[string]string{configs.QuotaPreemptionDelay: "1s"})}
leafQueueConf1 :=
[]configs.QueueConfig{createLeafQueueConfig(map[string]string{"memory": "5",
"vcore": "5"}, nil)}
leafQueueConf2 :=
[]configs.QueueConfig{createLeafQueueConfig(map[string]string{"memory": "5",
"vcore": "5"}, map[string]string{configs.QuotaPreemptionDelay: "0s"})}
@@ -3813,9 +3917,9 @@ func TestUpdateAllocationWithQuotaPreemption(t
*testing.T) {
time.Sleep(100 * time.Millisecond)
if tt.allocResult == nil {
- events := testHandler.GetEvents()
+ eventsList := testHandler.GetEvents()
eventsCount := 0
- for _, event := range events {
+ for _, event := range eventsList {
if allocRelease, ok :=
event.(*rmevent.RMReleaseAllocationEvent); ok {
assert.Equal(t,
len(allocRelease.ReleasedAllocations), 3)
assert.Equal(t,
allocRelease.ReleasedAllocations[0].GetTerminationType(),
si.TerminationType_PREEMPTED_BY_SCHEDULER, "")
@@ -3834,6 +3938,9 @@ func TestUpdateAllocationWithQuotaPreemption(t
*testing.T) {
func TestUpdateAllocationWithAsk(t *testing.T) {
setupUGM()
partition := createQueuesNodes(t)
+ assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
askCreated, _, err := partition.UpdateAllocation(nil)
assert.NilError(t, err, "nil ask should not return an error")
assert.Check(t, !askCreated, "ask should not have been created")
@@ -3904,6 +4011,8 @@ func TestUpdateAllocationWithAsk(t *testing.T) {
func TestUpdateAllocationWithAskAndQuotaPreemption(t *testing.T) {
setupUGM()
partition := createQuotaPreemptionQueuesNodes(t)
+ defer partition.userGroupCache.Stop()
+
leafQueueConf :=
[]configs.QueueConfig{createLeafQueueConfig(map[string]string{"memory": "5",
"vcore": "5"}, map[string]string{configs.QuotaPreemptionDelay: "1s"})}
leafQueueConf1 :=
[]configs.QueueConfig{createLeafQueueConfig(map[string]string{"memory": "5",
"vcore": "5"}, nil)}
leafQueueConf2 :=
[]configs.QueueConfig{createLeafQueueConfig(map[string]string{"memory": "5",
"vcore": "5"}, map[string]string{configs.QuotaPreemptionDelay: "0s"})}
@@ -4006,9 +4115,9 @@ func TestUpdateAllocationWithAskAndQuotaPreemption(t
*testing.T) {
time.Sleep(100 * time.Millisecond)
if tt.allocResult == nil {
- events := testHandler.GetEvents()
+ eventsList := testHandler.GetEvents()
eventsCount := 0
- for _, event := range events {
+ for _, event := range eventsList {
if allocRelease, ok :=
event.(*rmevent.RMReleaseAllocationEvent); ok {
assert.Equal(t,
len(allocRelease.ReleasedAllocations), 2)
assert.Equal(t,
allocRelease.ReleasedAllocations[0].GetTerminationType(),
si.TerminationType_PREEMPTED_BY_SCHEDULER, "")
@@ -4028,6 +4137,7 @@ func TestRemoveAllocationAsk(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
// add the app
app := newApplication(appID1, "default", "root.default")
err = partition.AddApplication(app)
@@ -4083,6 +4193,7 @@ func TestUpdatePreemption(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "Partition creation failed")
+ defer partition.userGroupCache.Stop()
assert.Assert(t, partition.IsPreemptionEnabled(), "preemption should be
enabled by default")
assert.Assert(t, !partition.IsQuotaPreemptionEnabled(), "quota
preemption should be disabled by default")
@@ -4110,6 +4221,7 @@ func TestUpdatePreemption(t *testing.T) {
func TestUpdateNodeSortingPolicy(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "Partition creation failed unexpectedly")
+ defer partition.userGroupCache.Stop()
if partition.nodes.GetNodeSortingPolicy().PolicyType().String() !=
policies.FairnessPolicy.String() {
t.Error("Node policy is not set with the default policy which
is fair policy.")
@@ -4193,9 +4305,8 @@ func TestGetNodeSortingPolicyWhenNewPartitionFromConfig(t
*testing.T) {
}
p, err := newPartitionContext(conf, rmID, nil, false)
- if err != nil {
- t.Errorf("Partition creation fail: %s",
err.Error())
- }
+ assert.NilError(t, err, "Partition creation fail should
not have failed")
+ defer p.userGroupCache.Stop()
ans :=
p.nodes.GetNodeSortingPolicy().PolicyType().String()
if ans != tt.want {
@@ -4209,6 +4320,8 @@ func TestTryAllocateMaxRunning(t *testing.T) {
const resType = "vcore"
partition := createQueuesNodes(t)
assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
if result := partition.tryAllocate(); result != nil {
t.Fatalf("empty cluster allocate returned allocation: %s",
result)
}
@@ -4296,6 +4409,7 @@ func TestNewQueueEvents(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err)
+ defer partition.userGroupCache.Stop()
_, err = partition.createQueue("root.test", security.UserGroup{
User: "test",
})
@@ -4326,6 +4440,8 @@ func TestUserHeadroom(t *testing.T) {
setupUGM()
partition, err := newConfiguredPartition()
assert.NilError(t, err, "test partition create failed with error")
+ defer partition.userGroupCache.Stop()
+
var res *resources.Resource
res, err = resources.NewResourceFromConf(map[string]string{"memory":
"10", "vcores": "10"})
assert.NilError(t, err, "failed to create basic resource")
@@ -4456,6 +4572,9 @@ func TestPlaceholderAllocationTracking(t *testing.T) {
const phID3 = "ph-3"
setupUGM()
partition := createQueuesNodes(t)
+ assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
res, err := resources.NewResourceFromConf(map[string]string{"vcore":
"1"})
assert.NilError(t, err, "failed to create resource")
@@ -4529,6 +4648,8 @@ func TestPlaceholderAllocationTracking(t *testing.T) {
func TestReservationTracking(t *testing.T) {
setupUGM()
partition := createQueuesNodes(t)
+ assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
app := newApplication(appID1, "default", "root.parent.sub-leaf")
res, err := resources.NewResourceFromConf(map[string]string{"vcore":
"10"})
@@ -4682,6 +4803,7 @@ func TestLimitMaxApplications(t *testing.T) {
partition, err := newPartitionContext(conf, rmID, nil,
false)
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
// add node1
nodeRes, err :=
resources.NewResourceFromConf(map[string]string{"memory": "10", "vcores": "10"})
@@ -4837,6 +4959,7 @@ func TestLimitMaxApplicationsForReservedAllocation(t
*testing.T) {
partition, err := newPartitionContext(conf, rmID, nil,
false)
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
// add node1
nodeRes, err :=
resources.NewResourceFromConf(map[string]string{"memory": "10", "vcores": "10"})
@@ -4885,6 +5008,7 @@ func TestLimitMaxApplicationsForReservedAllocation(t
*testing.T) {
func TestCalculateOutstandingRequests(t *testing.T) {
partition, err := newBasePartition()
assert.NilError(t, err, "unable to create partition: %v", err)
+ defer partition.userGroupCache.Stop()
// no application&asks
requests := partition.calculateOutstandingRequests()
@@ -4960,6 +5084,7 @@ func
TestPlaceholderAllocationAndReplacementAfterRecovery(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
// add a new app
app := newApplication(appID1, "default", defQueue)
@@ -5013,10 +5138,12 @@ func
TestPlaceholderAllocationAndReplacementAfterRecovery(t *testing.T) {
assert.Equal(t, "tg-1", confirmed.GetTaskGroup())
}
-func TestForeignAllocation(t *testing.T) { //nolint:funlen
+//nolint:funlen
+func TestForeignAllocation(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
nodeRes :=
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 10})
node := newNodeMaxResource(nodeID1, nodeRes)
err = partition.AddNode(node)
@@ -5121,6 +5248,7 @@ func TestAppSchedulingOrderFIFO(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
conf := configs.PartitionConfig{
Name: "test",
Queues: []configs.QueueConfig{
@@ -5221,6 +5349,7 @@ func TestApplicationBackoff(t *testing.T) {
setupUGM()
partition, err := newBasePartition()
assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
conf := configs.PartitionConfig{
Name: "test",
Queues: []configs.QueueConfig{
@@ -5272,3 +5401,28 @@ func TestApplicationBackoff(t *testing.T) {
assert.Assert(t, !deadline.IsZero())
assert.Assert(t, deadline.After(beforeAlloc))
}
+
+func TestUpdateResolver(t *testing.T) {
+ conf := configs.PartitionConfig{
+ Name: "test",
+ Queues: []configs.QueueConfig{
+ {
+ Name: "root",
+ Parent: true,
+ SubmitACL: "*",
+ Queues: nil,
+ },
+ },
+ UserGroupResolver: configs.UserGroupResolver{
+ Type: "test",
+ },
+ }
+ partition, err := newPartitionContext(conf, "resolver",
&ClusterContext{}, false)
+ assert.NilError(t, err, "partition create failed")
+ defer partition.userGroupCache.Stop()
+ assert.Equal(t, partition.GetUserGroupResolverType(), "test", "expected
test resolver to be set based on config")
+ conf.UserGroupResolver.Type = "ldap"
+ err = partition.updatePartitionDetails(conf)
+ assert.NilError(t, err, "unable to update partition config")
+ assert.Equal(t, partition.GetUserGroupResolverType(), "test", "resolver
cannot be updated on running partition")
+}
diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go
index 46a17825..7bc45f44 100644
--- a/pkg/scheduler/scheduler_test.go
+++ b/pkg/scheduler/scheduler_test.go
@@ -35,6 +35,7 @@ func TestInspectOutstandingRequests(t *testing.T) {
scheduler := NewScheduler()
partition, err := newBasePartition()
assert.NilError(t, err, "unable to create partition: %v", err)
+ defer partition.userGroupCache.Stop()
scheduler.clusterContext.partitions["test"] = partition
// two applications with no asks
@@ -114,6 +115,8 @@ func TestTriggerQuotaPreemption(t *testing.T) {
t.Run(tc.name, func(t *testing.T) {
scheduler := NewScheduler()
partition := createQuotaPreemptionQueuesNodes(t)
+ defer partition.userGroupCache.Stop()
+
// override the partition-level flag to match the test
case
partition.quotaPreemptionEnabled =
tc.quotaPreemptionEnabled
scheduler.clusterContext.partitions["test"] = partition
diff --git a/pkg/webservice/dao/partition_info.go
b/pkg/webservice/dao/partition_info.go
index 07e27293..62c8dc62 100644
--- a/pkg/webservice/dao/partition_info.go
+++ b/pkg/webservice/dao/partition_info.go
@@ -23,6 +23,7 @@ type PartitionInfo struct {
Name string `json:"name"`
// no omitempty, name should not be empty
Capacity PartitionCapacity `json:"capacity"`
// no omitempty, omitempty doesn't work on a structure value
NodeSortingPolicy NodeSortingPolicy `json:"nodeSortingPolicy"`
// no omitempty, omitempty doesn't work on a structure value
+ UserGroupResolver UserGroupResolver `json:"userGroupResolver"`
// no omitempty, omitempty doesn't work on a structure value
PreemptionEnabled bool `json:"preemptionEnabled"`
// no omitempty, false shows preemption status better
QuotaPreemptionEnabled bool
`json:"quotaPreemptionEnabled"` // no omitempty, false shows quota preemption
status better
TotalNodes int `json:"totalNodes,omitempty"`
@@ -42,3 +43,7 @@ type NodeSortingPolicy struct {
Type string `json:"type,omitempty"`
ResourceWeights map[string]float64 `json:"resourceWeights,omitempty"`
}
+
+type UserGroupResolver struct {
+ Type string `json:"type,omitempty"`
+}
diff --git a/pkg/webservice/handlers.go b/pkg/webservice/handlers.go
index 2698b525..b3b1e465 100644
--- a/pkg/webservice/handlers.go
+++ b/pkg/webservice/handlers.go
@@ -963,6 +963,9 @@ func getPartitionInfoDAO(lists
map[string]*scheduler.PartitionContext) []*dao.Pa
Type:
partitionContext.GetNodeSortingPolicyType().String(),
ResourceWeights:
partitionContext.GetNodeSortingResourceWeights(),
}
+ partitionInfo.UserGroupResolver = dao.UserGroupResolver{
+ Type: partitionContext.GetUserGroupResolverType(),
+ }
partitionInfo.TotalNodes = partitionContext.GetTotalNodeCount()
appList := partitionContext.GetApplications()
diff --git a/pkg/webservice/handlers_test.go b/pkg/webservice/handlers_test.go
index 2ff1f76a..1ebf7eff 100644
--- a/pkg/webservice/handlers_test.go
+++ b/pkg/webservice/handlers_test.go
@@ -117,13 +117,17 @@ partitions:
const configMultiPartitions = `
partitions:
- name: gpu
+ usergroupresolver:
+ type: test
preemption:
enabled: false
queues:
- name: root
- name: default
+ usergroupresolver:
+ type: ""
nodesortpolicy:
- type: fair
+ type: fair
queues:
- name: root
queues:
@@ -285,7 +289,6 @@ func setup(t *testing.T, config string, partitionCount int)
*scheduler.Partition
ctx, err := scheduler.NewClusterContext(rmID, policyGroup,
[]byte(config))
assert.NilError(t, err, "Error when load clusterInfo from config")
schedulerContext.Store(ctx)
-
assert.Equal(t, partitionCount,
len(schedulerContext.Load().GetPartitionMapClone()))
// Check default partition
@@ -533,6 +536,7 @@ func TestGetConfigYAML(t *testing.T) {
ctx, err := scheduler.NewClusterContext(rmID, policyGroup,
[]byte(startConf))
assert.NilError(t, err, "Error when load clusterInfo from config")
schedulerContext.Store(ctx)
+ defer schedulerContext.Load().Stop()
// No err check: new request always returns correctly
//nolint: errcheck
req, _ := http.NewRequest("GET", "", nil)
@@ -568,6 +572,7 @@ func TestGetConfigYAML(t *testing.T) {
func TestGetConfigJSON(t *testing.T) {
setup(t, startConf, 1)
+ defer schedulerContext.Load().Stop()
// No err check: new request always returns correctly
//nolint: errcheck
req, _ := http.NewRequest("GET", "", nil)
@@ -600,6 +605,7 @@ func TestGetConfigJSON(t *testing.T) {
func TestGetClusterUtilJSON(t *testing.T) {
setup(t, configDefault, 1)
+ defer schedulerContext.Load().Stop()
// check build information of RM
buildInfoMap := make(map[string]string)
@@ -722,6 +728,7 @@ func addAndConfirmApplicationExists(t *testing.T,
partitionName string, partitio
func TestGetPartitionNodesUtilJSON(t *testing.T) {
// setup
partition := setup(t, configDefault, 1)
+ defer schedulerContext.Load().Stop()
appID := "app1"
// create test nodes
@@ -776,12 +783,13 @@ func TestGetNodeUtilisations(t *testing.T) {
// setup partitions
ctx, err := scheduler.NewClusterContext(rmID, policyGroup,
[]byte(configMultiPartitions))
- assert.NilError(t, err, "Error when load clusterInfo from config")
+ assert.NilError(t, err, "Error when creating clusterInfo from config")
schedulerContext.Store(ctx)
- assert.NilError(t, err, "Error when load clusterInfo from config")
- schedulerContext.Load().GetPartition("default")
+ defer schedulerContext.Load().Stop()
defaultPartition :=
schedulerContext.Load().GetPartition(common.GetNormalizedPartitionName("default",
rmID))
+ assert.Assert(t, defaultPartition != nil, "Default partition should not
be nil")
gpuPartition :=
schedulerContext.Load().GetPartition(common.GetNormalizedPartitionName("gpu",
rmID))
+ assert.Assert(t, gpuPartition != nil, "Gpu partition should not be nil")
// add nodes to partitions
node1 := addNode(t, defaultPartition, "node-1",
resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 10}))
@@ -842,10 +850,13 @@ func getNodesUtilByType(t *testing.T, nodesUtilList
[]*dao.NodesUtilDAOInfo, res
func TestPartitions(t *testing.T) { //nolint:funlen
schedulerContext.Store(&scheduler.ClusterContext{})
+ // The resolver is protected by a once(). Need to stop and clear the
resolver before we create a new partition
+ // to ensure the partition will have the expected setup. Clean up after
the test to not impact other tests.
+ defer schedulerContext.Load().Stop()
var req *http.Request
req, err := http.NewRequest("GET", "/ws/v1/partitions",
strings.NewReader(""))
- assert.NilError(t, err, "App Handler request failed")
+ assert.NilError(t, err, "partition handler request failed")
resp := &MockResponseWriter{}
var partitionInfo []*dao.PartitionInfo
@@ -855,6 +866,8 @@ func TestPartitions(t *testing.T) { //nolint:funlen
assert.Check(t, partitionInfo != nil, "partitionInfo should not be nil")
assert.Equal(t, len(partitionInfo), 0)
+ // before we overwrite the context clean up
+ schedulerContext.Load().Stop()
defaultPartition := setup(t, configMultiPartitions, 2)
partitionName := defaultPartition.Name
@@ -913,7 +926,7 @@ func TestPartitions(t *testing.T) { //nolint:funlen
assert.Check(t, allocCreated)
req, err = http.NewRequest("GET", "/ws/v1/partitions",
strings.NewReader(""))
- assert.NilError(t, err, "App Handler request failed")
+ assert.NilError(t, err, "partition handler request failed")
resp = &MockResponseWriter{}
getPartitions(resp, req)
err = json.Unmarshal(resp.outputBytes, &partitionInfo)
@@ -943,15 +956,19 @@ func TestPartitions(t *testing.T) { //nolint:funlen
assert.DeepEqual(t, cs["default"].Capacity.Utilization,
map[string]int64{"memory": 30, "vcore": 70})
assert.Equal(t, cs["default"].State, "Active")
assert.Assert(t, cs["default"].PreemptionEnabled, "preemption should be
enabled on default")
+ assert.Equal(t, cs["default"].NodeSortingPolicy.Type, "fair")
+ assert.Equal(t,
cs["default"].NodeSortingPolicy.ResourceWeights["vcore"], 1.0)
+ assert.Equal(t,
cs["default"].NodeSortingPolicy.ResourceWeights["memory"], 1.0)
assert.Assert(t, cs["gpu"] != nil)
assert.Equal(t, cs["gpu"].ClusterID, "rm-123")
assert.Equal(t, cs["gpu"].Name, "gpu")
- assert.Equal(t, cs["default"].NodeSortingPolicy.Type, "fair")
- assert.Equal(t,
cs["default"].NodeSortingPolicy.ResourceWeights["vcore"], 1.0)
- assert.Equal(t,
cs["default"].NodeSortingPolicy.ResourceWeights["memory"], 1.0)
assert.Equal(t, cs["gpu"].Applications["total"], 0)
assert.Assert(t, !cs["gpu"].PreemptionEnabled, "preemption should be
disabled on gpu")
+
+ // only one resolver for the system, gpu is defined first so we get the
one defined there: test
+ assert.Equal(t, cs["gpu"].UserGroupResolver.Type, "test", "resolver
should be set to test")
+ assert.Equal(t, cs["default"].UserGroupResolver.Type, "test", "resolver
should be set to test")
}
func TestMetricsNotEmpty(t *testing.T) {
@@ -966,7 +983,7 @@ func TestMetricsNotEmpty(t *testing.T) {
//nolint:funlen
func TestGetPartitionQueuesHandler(t *testing.T) {
setup(t, configTwoLevelQueues, 2)
-
+ defer schedulerContext.Load().Stop()
NewWebApp(schedulerContext.Load(), nil)
tMaxResource, err :=
resources.NewResourceFromConf(map[string]string{"memory": "600000"})
@@ -1051,6 +1068,7 @@ func TestGetPartitionQueueHandler(t *testing.T) {
partitionQueuesHandler := "/ws/v1/partition/default/queue/"
queueA := "root.a"
setup(t, configTwoLevelQueues, 2)
+ defer schedulerContext.Load().Stop()
NewWebApp(schedulerContext.Load(), nil)
@@ -1136,6 +1154,8 @@ func TestGetPartitionQueueHandler(t *testing.T) {
func TestGetClusterInfo(t *testing.T) {
schedulerContext.Store(&scheduler.ClusterContext{})
+ // clean up started background routines when done
+ defer schedulerContext.Load().Stop()
resp := &MockResponseWriter{}
req, err := http.NewRequest("GET", "/ws/v1/clusters",
strings.NewReader(""))
assert.NilError(t, err, "error while creating http request")
@@ -1145,6 +1165,8 @@ func TestGetClusterInfo(t *testing.T) {
assert.NilError(t, err)
assert.Equal(t, 0, len(data))
+ // before we create a new context clean up
+ schedulerContext.Load().Stop()
setup(t, configTwoLevelQueues, 2)
resp = &MockResponseWriter{}
@@ -1164,6 +1186,7 @@ func TestGetClusterInfo(t *testing.T) {
func TestGetPartitionNodes(t *testing.T) {
partition := setup(t, configDefault, 1)
+ defer schedulerContext.Load().Stop()
// create test application
appID := "app1"
@@ -1241,6 +1264,7 @@ func TestGetPartitionNodes(t *testing.T) {
func TestGetPartitionNode(t *testing.T) {
partition := setup(t, configDefault, 1)
+ defer schedulerContext.Load().Stop()
// create test application
appID := "app-1"
@@ -1384,6 +1408,7 @@ func TestGetQueueApplicationsHandler(t *testing.T) {
handlerSuffix := "/applications"
defaultQueue := "root.default"
part := setup(t, configDefault, 1)
+ defer schedulerContext.Load().Stop()
// add an application
app := addApp(t, "app-1", part, defaultQueue, false)
@@ -1527,6 +1552,7 @@ func checkIllegalGetAppsRequest(t *testing.T, url string,
params httprouter.Para
func TestGetPartitionApplicationsByStateHandler(t *testing.T) {
defaultPartition := setup(t, configDefault, 1)
+ defer schedulerContext.Load().Stop()
NewWebApp(schedulerContext.Load(), nil)
// add a new application
@@ -1658,6 +1684,7 @@ func checkGetQueueAppByIllegalStateOrStatus(t *testing.T,
partition, queue, stat
func TestGetQueueApplicationsByStateHandler(t *testing.T) {
defaultPartition := setup(t, configDefault, 1)
+ defer schedulerContext.Load().Stop()
NewWebApp(schedulerContext.Load(), nil)
// Accept status
@@ -1697,6 +1724,7 @@ func TestGetQueueApplicationsByStateHandler(t *testing.T)
{
//nolint:funlen
func TestGetApplicationHandler(t *testing.T) {
part := setup(t, configDefault, 1)
+ defer schedulerContext.Load().Stop()
// add 1 application
app := addApp(t, "app-1", part, "root.default", false)
@@ -1973,6 +2001,8 @@ func TestFullStateDumpPath(t *testing.T) {
configs.SetConfigMap(configMap)
prepareSchedulerContext(t)
+ // prepareSchedulerContext hides creating a new context so make sure we
clean up after use
+ defer schedulerContext.Load().Stop()
partitionContext := schedulerContext.Load().GetPartitionMapClone()
ctx := partitionContext[normalizedPartitionName]
@@ -1997,6 +2027,8 @@ func TestFullStateDumpPath(t *testing.T) {
func TestSpecificUserResourceUsage(t *testing.T) {
prepareUserAndGroupContext(t, groupsLimitsConfig)
+ // prepareUserAndGroupContext hides creating a new context so make sure
we clean up after use
+ defer schedulerContext.Load().Stop()
// Test existed user query
req, err := createRequest(t, "/ws/v1/partition/default/usage/user/",
map[string]string{"user": "testuser", "group": "testgroup"})
@@ -2069,6 +2101,8 @@ func TestSpecificUserResourceUsage(t *testing.T) {
func TestSpecificGroupResourceUsage(t *testing.T) {
prepareUserAndGroupContext(t, groupsLimitsConfig)
+ // prepareUserAndGroupContext hides creating a new context so make sure
we clean up after use
+ defer schedulerContext.Load().Stop()
// Test existed group query
req, err := createRequest(t, "/ws/v1/partition/default/usage/group",
map[string]string{"user": "testuser", "group": "testgroup"})
assert.NilError(t, err)
@@ -2135,7 +2169,8 @@ func TestSpecificGroupResourceUsage(t *testing.T) {
func TestUsersAndGroupsResourceUsage(t *testing.T) {
prepareUserAndGroupContext(t, groupsLimitsConfig)
- var req *http.Request
+ // prepareUserAndGroupContext hides creating a new context so make sure
we clean up after use
+ defer schedulerContext.Load().Stop()
req, err := http.NewRequest("GET",
"/ws/v1/partition/default/usage/users", strings.NewReader(""))
assert.NilError(t, err, "Get Users Resource Usage Handler request
failed")
resp := &MockResponseWriter{}
@@ -2189,6 +2224,8 @@ func TestUsersAndGroupsResourceUsage(t *testing.T) {
func TestGetEvents(t *testing.T) {
prepareSchedulerContext(t)
+ // prepareSchedulerContext hides creating a new context so make sure we
clean up after use
+ defer schedulerContext.Load().Stop()
appEvent, nodeEvent, queueEvent := addEvents(t)
checkAllEvents(t, []*si.EventRecord{appEvent, nodeEvent, queueEvent})
@@ -2231,6 +2268,7 @@ func TestGetEventsWhenTrackingDisabled(t *testing.T) {
func TestGetStream(t *testing.T) {
setup(t, configDefault, 1)
+ defer schedulerContext.Load().Stop()
ev, req := initEventsAndCreateRequest(t)
defer ev.Stop()
cancelCtx, cancel := context.WithCancel(context.Background())
@@ -2310,6 +2348,7 @@ func TestGetStream_NotFlusherImpl(t *testing.T) {
func TestGetStream_Count(t *testing.T) {
setup(t, configDefault, 1)
+ defer schedulerContext.Load().Stop()
ev, req := initEventsAndCreateRequest(t)
defer ev.Stop()
cancelCtx, cancel := context.WithCancel(context.Background())
@@ -2397,6 +2436,7 @@ func TestGetStream_NoWriteDeadline(t *testing.T) {
func TestGetStream_SetWriteDeadlineFails(t *testing.T) {
setup(t, configDefault, 1)
+ defer schedulerContext.Load().Stop()
ev, req := initEventsAndCreateRequest(t)
defer ev.Stop()
resp := NewResponseRecorderWithDeadline()
@@ -2650,8 +2690,8 @@ func prepareSchedulerContext(t *testing.T) {
config := []byte(configDefault)
var err error
ctx, err := scheduler.NewClusterContext(rmID, policyGroup, config)
- schedulerContext.Store(ctx)
assert.NilError(t, err, "Error when load clusterInfo from config")
+ schedulerContext.Store(ctx)
assert.Equal(t, 1, len(schedulerContext.Load().GetPartitionMapClone()))
}
@@ -2773,7 +2813,7 @@ func runHealthCheckTest(t *testing.T, expected
*dao.SchedulerHealthDAOInfo) {
func TestGetPartitionRuleHandler(t *testing.T) {
setup(t, configDefault, 1)
-
+ defer schedulerContext.Load().Stop()
NewWebApp(schedulerContext.Load(), nil)
// test partition not exists
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
