Skip to content

Commit 1b3f0b8

Browse files
committed
[YUNIKORN-3121] Add REST Endpoint for Scheduling Order Visibility
1 parent e41a426 commit 1b3f0b8

6 files changed

Lines changed: 211 additions & 0 deletions

File tree

pkg/scheduler/objects/queue.go

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1337,6 +1337,49 @@ func (sq *Queue) sortQueues() []*Queue {
13371337
return sortedQueues
13381338
}
13391339

1340+
// GetSchedulingOrder returns a sorted shallow copy of the queues for this parent queue along with
1341+
// their eligible applications that would be tried in the current scheduling cycle.
1342+
// This follows the same logic as TryAllocate method.
1343+
// Only queues with a pending resource request are considered. The queues are sorted using the
1344+
// sorting type for the parent queue.
1345+
// Lock free call all locks are taken when needed in called functions
1346+
func (sq *Queue) GetSchedulingOrder() []*dao.SchedulingOrderDAO {
1347+
if sq.IsLeafQueue() {
1348+
// For leaf queues, return the queue with its eligible applications
1349+
appIDs := make([]string, 0)
1350+
1351+
// Process the apps (filters out app without pending requests) - same logic as TryAllocate
1352+
for _, app := range sq.sortApplications(false) {
1353+
runnableInQueue := sq.canRunApp(app.ApplicationID)
1354+
runnableByUserLimit := ugm.GetUserManager().CanRunApp(sq.QueuePath, app.ApplicationID, app.user)
1355+
app.updateRunnableStatus(runnableInQueue, runnableByUserLimit)
1356+
if app.IsAccepted() && (!runnableInQueue || !runnableByUserLimit) {
1357+
continue
1358+
}
1359+
appIDs = append(appIDs, app.ApplicationID)
1360+
}
1361+
1362+
// Only return this queue if it has eligible applications or pending resources
1363+
if len(appIDs) > 0 || resources.StrictlyGreaterThanZero(sq.GetPendingResource()) {
1364+
return []*dao.SchedulingOrderDAO{{
1365+
QueueName: sq.QueuePath,
1366+
ApplicationIDs: appIDs,
1367+
}}
1368+
}
1369+
return nil
1370+
} else {
1371+
// For parent queues, process child queues - same logic as TryAllocate
1372+
result := make([]*dao.SchedulingOrderDAO, 0)
1373+
1374+
// Process each sorted child queue using the original sortQueues method
1375+
for _, child := range sq.sortQueues() {
1376+
childInfo := child.GetSchedulingOrder()
1377+
result = append(result, childInfo...)
1378+
}
1379+
return result
1380+
}
1381+
}
1382+
13401383
// getHeadRoom returns the headroom for the queue. This can never be more than the headroom for the parent.
13411384
// In case there are no nodes in a newly started cluster and no queues have a limit configured this call
13421385
// will return nil.

pkg/scheduler/partition.go

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -507,6 +507,11 @@ func (pc *PartitionContext) GetPartitionQueues() dao.PartitionQueueDAOInfo {
507507
return partitionQueueDAOInfo
508508
}
509509

510+
// GetPartitionSchedulingOrder builds the sorted queue info for the whole queue structure to pass to the webservice
511+
func (pc *PartitionContext) GetPartitionSchedulingOrder() []*dao.SchedulingOrderDAO {
512+
return pc.root.GetSchedulingOrder()
513+
}
514+
510515
// GetPlacementRules returns the current active rule set as dao to expose to the webservice
511516
func (pc *PartitionContext) GetPlacementRules() []*dao.RuleDAO {
512517
return pc.getPlacementManager().GetRulesDAO()
Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
/*
2+
Licensed to the Apache Software Foundation (ASF) under one
3+
or more contributor license agreements. See the NOTICE file
4+
distributed with this work for additional information
5+
regarding copyright ownership. The ASF licenses this file
6+
to you under the Apache License, Version 2.0 (the
7+
"License"); you may not use this file except in compliance
8+
with the License. You may obtain a copy of the License at
9+
http://www.apache.org/licenses/LICENSE-2.0
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package dao
18+
19+
// SchedulingOrderDAO contains queue and application information for scheduling order
20+
type SchedulingOrderDAO struct {
21+
QueueName string `json:"queueName"`
22+
ApplicationIDs []string `json:"applicationIDs"` // Application IDs that would be tried in current scheduling cycle
23+
}

pkg/webservice/handlers.go

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -721,6 +721,27 @@ func getPartitionQueues(w http.ResponseWriter, r *http.Request) {
721721
}
722722
}
723723

724+
func getPartitionSchedulingOrder(w http.ResponseWriter, r *http.Request) {
725+
writeHeaders(w, r.Method)
726+
vars := httprouter.ParamsFromContext(r.Context())
727+
if vars == nil {
728+
buildJSONErrorResponse(w, MissingParamsName, http.StatusBadRequest)
729+
return
730+
}
731+
partitionName := vars.ByName("partition")
732+
var partitionSchedulingOrderDAOInfo []*dao.SchedulingOrderDAO
733+
var partition = schedulerContext.Load().GetPartitionWithoutClusterID(partitionName)
734+
if partition != nil {
735+
partitionSchedulingOrderDAOInfo = partition.GetPartitionSchedulingOrder()
736+
} else {
737+
buildJSONErrorResponse(w, PartitionDoesNotExists, http.StatusNotFound)
738+
return
739+
}
740+
if err := json.NewEncoder(w).Encode(partitionSchedulingOrderDAOInfo); err != nil {
741+
buildJSONErrorResponse(w, err.Error(), http.StatusInternalServerError)
742+
}
743+
}
744+
724745
func getPartitionQueue(w http.ResponseWriter, r *http.Request) {
725746
writeHeaders(w, r.Method)
726747
vars := httprouter.ParamsFromContext(r.Context())

pkg/webservice/handlers_test.go

Lines changed: 113 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3016,6 +3016,119 @@ func TestSetMaxRESTResponseSize(t *testing.T) {
30163016
assert.Equal(t, uint64(10000), maxRESTResponseSize.Load())
30173017
}
30183018

3019+
func TestGetPartitionSchedulingOrderHandler(t *testing.T) {
3020+
// Setup partition with default configuration
3021+
part := setup(t, configDefault, 1)
3022+
3023+
// Create applications with pending requests to trigger scheduling order
3024+
app1 := addApp(t, "app-1", part, "root.default", false)
3025+
app2 := addApp(t, "app-2", part, "root.default", false)
3026+
app3 := addApp(t, "app-3", part, "root.default", false)
3027+
3028+
// Add pending allocation requests to applications
3029+
res1 := &si.Resource{
3030+
Resources: map[string]*si.Quantity{"memory": {Value: 100}, "vcore": {Value: 1}},
3031+
}
3032+
ask1 := objects.NewAllocationFromSI(&si.Allocation{
3033+
ApplicationID: "app-1",
3034+
PartitionName: part.Name,
3035+
ResourcePerAlloc: res1})
3036+
err := app1.AddAllocationAsk(ask1)
3037+
assert.NilError(t, err, "ask should have been added to app-1")
3038+
3039+
res2 := &si.Resource{
3040+
Resources: map[string]*si.Quantity{"memory": {Value: 200}, "vcore": {Value: 2}},
3041+
}
3042+
ask2 := objects.NewAllocationFromSI(&si.Allocation{
3043+
ApplicationID: "app-2",
3044+
PartitionName: part.Name,
3045+
ResourcePerAlloc: res2})
3046+
err = app2.AddAllocationAsk(ask2)
3047+
assert.NilError(t, err, "ask should have been added to app-2")
3048+
3049+
res3 := &si.Resource{
3050+
Resources: map[string]*si.Quantity{"memory": {Value: 150}, "vcore": {Value: 1}},
3051+
}
3052+
ask3 := objects.NewAllocationFromSI(&si.Allocation{
3053+
ApplicationID: "app-3",
3054+
PartitionName: part.Name,
3055+
ResourcePerAlloc: res3})
3056+
err = app3.AddAllocationAsk(ask3)
3057+
assert.NilError(t, err, "ask should have been added to app-3")
3058+
3059+
NewWebApp(schedulerContext.Load(), nil)
3060+
3061+
// Test successful scheduling order retrieval
3062+
var req *http.Request
3063+
req, err = createRequest(t, "/ws/v1/partition/default/schedulingorder", map[string]string{"partition": partitionNameWithoutClusterID})
3064+
assert.NilError(t, err, "Get Scheduling Order Handler request failed")
3065+
resp := &MockResponseWriter{}
3066+
var schedulingOrderDao []*dao.SchedulingOrderDAO
3067+
getPartitionSchedulingOrder(resp, req)
3068+
err = json.Unmarshal(resp.outputBytes, &schedulingOrderDao)
3069+
assert.NilError(t, err, unmarshalError)
3070+
3071+
// Validate response structure and content
3072+
assert.Assert(t, len(schedulingOrderDao) > 0, "scheduling order should contain at least one queue")
3073+
3074+
// Verify that queues with pending resources are included
3075+
queueFound := false
3076+
for _, queueOrder := range schedulingOrderDao {
3077+
if queueOrder.QueueName == "root.default" {
3078+
queueFound = true
3079+
// Verify that applications with pending requests are included
3080+
assert.Assert(t, len(queueOrder.ApplicationIDs) >= 3, "queue root.default should have at least 3 applications")
3081+
// Check that our test applications are in the list
3082+
appFound1, appFound2, appFound3 := false, false, false
3083+
for _, appID := range queueOrder.ApplicationIDs {
3084+
if appID == "app-1" {
3085+
appFound1 = true
3086+
}
3087+
if appID == "app-2" {
3088+
appFound2 = true
3089+
}
3090+
if appID == "app-3" {
3091+
appFound3 = true
3092+
}
3093+
}
3094+
assert.Assert(t, appFound1, "app-1 should be in scheduling order")
3095+
assert.Assert(t, appFound2, "app-2 should be in scheduling order")
3096+
assert.Assert(t, appFound3, "app-3 should be in scheduling order")
3097+
}
3098+
}
3099+
assert.Assert(t, queueFound, "queue root.default should be found in scheduling order")
3100+
3101+
// Test nonexistent partition
3102+
req, err = createRequest(t, "/ws/v1/partition/default/schedulingorder", map[string]string{"partition": "notexists"})
3103+
assert.NilError(t, err, "Get Scheduling Order Handler request failed")
3104+
resp = &MockResponseWriter{}
3105+
getPartitionSchedulingOrder(resp, req)
3106+
assertPartitionNotExists(t, resp)
3107+
3108+
// Test missing params name
3109+
req, err = http.NewRequest("GET", "/ws/v1/partition/default/schedulingorder", strings.NewReader(""))
3110+
assert.NilError(t, err, "Get Scheduling Order Handler request failed")
3111+
resp = &MockResponseWriter{}
3112+
getPartitionSchedulingOrder(resp, req)
3113+
assertParamsMissing(t, resp)
3114+
3115+
// Test empty scheduling order (partition with no pending requests)
3116+
// Create a new partition with no applications
3117+
emptyPart := setup(t, configDefault, 1)
3118+
// Add an application but without pending requests
3119+
addApp(t, "app-empty", emptyPart, "root.default", false)
3120+
3121+
req, err = createRequest(t, "/ws/v1/partition/default/schedulingorder", map[string]string{"partition": partitionNameWithoutClusterID})
3122+
assert.NilError(t, err, "Get Scheduling Order Handler request failed")
3123+
resp = &MockResponseWriter{}
3124+
var emptySchedulingOrderDao []*dao.SchedulingOrderDAO
3125+
getPartitionSchedulingOrder(resp, req)
3126+
err = json.Unmarshal(resp.outputBytes, &emptySchedulingOrderDao)
3127+
assert.NilError(t, err, unmarshalError)
3128+
// Should return empty array when no queues have pending resources
3129+
assert.Equal(t, len(emptySchedulingOrderDao), 0, "scheduling order should be empty when no pending requests exist")
3130+
}
3131+
30193132
type ResponseRecorderWithDeadline struct {
30203133
*httptest.ResponseRecorder
30213134
setWriteFails bool

pkg/webservice/routes.go

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,12 @@ var webRoutes = routes{
9090
"/ws/v1/partition/:partition/queues",
9191
getPartitionQueues,
9292
},
93+
route{
94+
"Scheduler",
95+
"GET",
96+
"/ws/v1/partition/:partition/schedulingorder",
97+
getPartitionSchedulingOrder,
98+
},
9399
route{
94100
"Scheduler",
95101
"GET",

0 commit comments

Comments
 (0)