[YUNIKORN-3121] Replace REST endpoint with orderLog in state dump

Author: mitdesai

pkg/scheduler/objects/queue.go

@@ -1337,49 +1337,6 @@ func (sq *Queue) sortQueues() []*Queue {
 	return sortedQueues
 }
 
-// GetSchedulingOrder returns a sorted shallow copy of the queues for this parent queue along with
-// their eligible applications that would be tried in the current scheduling cycle.
-// This follows the same logic as TryAllocate method.
-// Only queues with a pending resource request are considered. The queues are sorted using the
-// sorting type for the parent queue.
-// Lock free call all locks are taken when needed in called functions
-func (sq *Queue) GetSchedulingOrder() []*dao.SchedulingOrderDAO {
-	if sq.IsLeafQueue() {
-		// For leaf queues, return the queue with its eligible applications
-		appIDs := make([]string, 0)
-
-		// Process the apps (filters out app without pending requests) - same logic as TryAllocate
-		for _, app := range sq.sortApplications(false) {
-			runnableInQueue := sq.canRunApp(app.ApplicationID)
-			runnableByUserLimit := ugm.GetUserManager().CanRunApp(sq.QueuePath, app.ApplicationID, app.user)
-			app.updateRunnableStatus(runnableInQueue, runnableByUserLimit)
-			if app.IsAccepted() && (!runnableInQueue || !runnableByUserLimit) {
-				continue
-			}
-			appIDs = append(appIDs, app.ApplicationID)
-		}
-
-		// Only return this queue if it has eligible applications or pending resources
-		if len(appIDs) > 0 || resources.StrictlyGreaterThanZero(sq.GetPendingResource()) {
-			return []*dao.SchedulingOrderDAO{{
-				QueueName:      sq.QueuePath,
-				ApplicationIDs: appIDs,
-			}}
-		}
-		return nil
-	} else {
-		// For parent queues, process child queues - same logic as TryAllocate
-		result := make([]*dao.SchedulingOrderDAO, 0)
-
-		// Process each sorted child queue using the original sortQueues method
-		for _, child := range sq.sortQueues() {
-			childInfo := child.GetSchedulingOrder()
-			result = append(result, childInfo...)
-		}
-		return result
-	}
-}
-
 // getHeadRoom returns the headroom for the queue. This can never be more than the headroom for the parent.
 // In case there are no nodes in a newly started cluster and no queues have a limit configured this call
 // will return nil.

pkg/scheduler/objects/queue_test.go

@@ -3051,104 +3051,3 @@ func TestQueueBackoffProperties(t *testing.T) {
 	assert.Equal(t, uint64(0), leaf3.GetMaxAppUnschedAskBackoff())
 	assert.Equal(t, 30*time.Second, leaf3.GetBackoffDelay())
 }
-
-func TestGetSchedulingOrder(t *testing.T) {
-	// create the root
-	root, err := createRootQueue(nil)
-	root.sortType = policies.FifoSortPolicy
-	assert.NilError(t, err, "queue create failed")
-
-	// setup queue hierarchy
-	parent1, err := createManagedQueue(root, "parent1", true, nil)
-	parent1.sortType = policies.FifoSortPolicy
-	assert.NilError(t, err, "failed to create parent1 queue")
-	parent2, err := createManagedQueue(root, "parent2", true, nil)
-	assert.NilError(t, err, "failed to create parent2 queue")
-
-	leaf1, err := createManagedQueue(parent1, "leaf1", false, nil)
-	assert.NilError(t, err, "failed to create leaf1 queue")
-	leaf2, err := createManagedQueue(parent1, "leaf2", false, nil)
-	assert.NilError(t, err, "failed to create leaf2 queue")
-	leaf3, err := createManagedQueue(parent2, "leaf3", false, nil)
-	assert.NilError(t, err, "failed to create leaf3 queue")
-
-	// resource for asks
-	res, err := resources.NewResourceFromConf(map[string]string{"first": "1"})
-	assert.NilError(t, err, "failed to create basic resource")
-
-	// App1 in leaf1 with pending resources
-	app1 := newApplication("app1", "default", leaf1.QueuePath)
-	app1.queue = leaf1
-	leaf1.AddApplication(app1)
-	err = app1.AddAllocationAsk(newAllocationAsk("alloc-1", "app1", res))
-	assert.NilError(t, err, "failed to add allocation ask to app1")
-	leaf1.incPendingResource(res)
-
-	// App2 in leaf1 with no pending resources
-	app2 := newApplication("app2", "default", leaf1.QueuePath)
-	app2.queue = leaf1
-	leaf1.AddApplication(app2)
-
-	// App3 in leaf2 with pending resources
-	app3 := newApplication("app3", "default", leaf2.QueuePath)
-	app3.queue = leaf2
-	leaf2.AddApplication(app3)
-	err = app3.AddAllocationAsk(newAllocationAsk("alloc-3", "app3", res))
-	assert.NilError(t, err, "failed to add allocation ask to app3")
-	leaf2.incPendingResource(res)
-
-	// App4 in leaf3 with pending resources, but cannot run
-	app4 := newApplication("app4", "default", leaf3.QueuePath)
-	app4.queue = leaf3
-	leaf3.AddApplication(app4)
-	err = app4.AddAllocationAsk(newAllocationAsk("alloc-4", "app4", res))
-	assert.NilError(t, err, "failed to add allocation ask to app4")
-	leaf3.incPendingResource(res)
-	// Make app4 not runnable by setting queue to full
-	leaf3.maxRunningApps = 1
-	leaf3.runningApps = 1
-
-	// Get scheduling order from root
-	order := root.GetSchedulingOrder()
-
-	// Expected order based on FIFO:
-	// 1. parent1 -> leaf1 (app1)
-	// 2. parent1 -> leaf2 (app3)
-	// 3. parent2 -> leaf3 (no apps, because app4 is not runnable but queue has pending)
-	assert.Equal(t, len(order), 3, "incorrect number of scheduling order entries")
-
-	// Entry 1: leaf1
-	assert.Equal(t, order[0].QueueName, "root.parent1.leaf1")
-	assert.Equal(t, len(order[0].ApplicationIDs), 1, "leaf1 should have one app")
-	assert.Equal(t, order[0].ApplicationIDs[0], "app1")
-
-	// Entry 2: leaf2
-	assert.Equal(t, order[1].QueueName, "root.parent1.leaf2")
-	assert.Equal(t, len(order[1].ApplicationIDs), 1, "leaf2 should have one app")
-	assert.Equal(t, order[1].ApplicationIDs[0], "app3")
-
-	// Entry 3: leaf3
-	assert.Equal(t, order[2].QueueName, "root.parent2.leaf3")
-	assert.Equal(t, len(order[2].ApplicationIDs), 0, "leaf3 should have no runnable apps")
-
-	// Test a leaf queue directly
-	order = leaf1.GetSchedulingOrder()
-	assert.Equal(t, len(order), 1, "direct call on leaf1 failed")
-	assert.Equal(t, order[0].QueueName, "root.parent1.leaf1")
-	assert.Equal(t, len(order[0].ApplicationIDs), 1)
-	assert.Equal(t, order[0].ApplicationIDs[0], "app1")
-
-	// Test a leaf queue with no pending resources
-	leaf4, err := createManagedQueue(parent2, "leaf4", false, nil)
-	assert.NilError(t, err, "failed to create leaf4")
-	order = leaf4.GetSchedulingOrder()
-	assert.Equal(t, len(order), 0, "leaf queue with no pending resources should return nil/empty")
-
-	// Test a parent with no pending children
-	parent3, err := createManagedQueue(root, "parent3", true, nil)
-	assert.NilError(t, err, "failed to create parent3")
-	_, err = createManagedQueue(parent3, "leaf5", false, nil)
-	assert.NilError(t, err, "failed to create leaf5")
-	order = parent3.GetSchedulingOrder()
-	assert.Equal(t, len(order), 0, "parent with no pending children should return empty")
-}

pkg/scheduler/partition.go

@@ -507,11 +507,6 @@ func (pc *PartitionContext) GetPartitionQueues() dao.PartitionQueueDAOInfo {
 	return partitionQueueDAOInfo
 }
 
-// GetPartitionSchedulingOrder builds the sorted queue info for the whole queue structure to pass to the webservice
-func (pc *PartitionContext) GetPartitionSchedulingOrder() []*dao.SchedulingOrderDAO {
-	return pc.root.GetSchedulingOrder()
-}
-
 // GetPlacementRules returns the current active rule set as dao to expose to the webservice
 func (pc *PartitionContext) GetPlacementRules() []*dao.RuleDAO {
 	return pc.getPlacementManager().GetRulesDAO()
@@ -1720,3 +1715,31 @@ func (pc *PartitionContext) getReservationCount() int {
 	defer pc.RUnlock()
 	return pc.reservations
 }
+
+// GetOrderLog returns a snapshot of applications with pending requests grouped by queue
+func (pc *PartitionContext) GetOrderLog() []*dao.OrderLogEntry {
+	pc.RLock()
+	defer pc.RUnlock()
+
+	// Build a map of queue -> applications with pending requests
+	queueAppMap := make(map[string][]string)
+
+	for _, app := range pc.applications {
+		// Only include apps with pending resources
+		if resources.StrictlyGreaterThanZero(app.GetPendingResource()) {
+			queuePath := app.GetQueuePath()
+			queueAppMap[queuePath] = append(queueAppMap[queuePath], app.ApplicationID)
+		}
+	}
+
+	// Convert map to slice
+	result := make([]*dao.OrderLogEntry, 0, len(queueAppMap))
+	for queueName, appIDs := range queueAppMap {
+		result = append(result, &dao.OrderLogEntry{
+			QueueName:      queueName,
+			ApplicationIDs: appIDs,
+		})
+	}
+
+	return result
+}

pkg/scheduler/partition_test.go

@@ -4998,3 +4998,116 @@ func TestGetPartitionQueueDAOInfo(t *testing.T) {
 	assert.Equal(t, "root.parent.leaf", leafDAO.QueueName)
 	assert.Equal(t, 0, len(leafDAO.Children))
 }
+
+func TestOrderLog(t *testing.T) {
+	// Create partition with multiple queues
+	conf := configs.PartitionConfig{
+		Name: "test",
+		Queues: []configs.QueueConfig{
+			{
+				Name:      "root",
+				Parent:    true,
+				SubmitACL: "*",
+				Queues: []configs.QueueConfig{
+					{Name: "default", Parent: false},
+					{Name: "production", Parent: false},
+				},
+			},
+		},
+	}
+	partition, err := newPartitionContext(conf, rmID, nil, false)
+	assert.NilError(t, err, "partition create failed")
+
+	// Empty partition should have empty order log
+	orderLog := partition.GetOrderLog()
+	assert.Equal(t, 0, len(orderLog), "initial order log should be empty")
+
+	// Add app1 to default queue with pending request
+	res := resources.NewResourceFromMap(map[string]resources.Quantity{"memory": 100, "vcore": 10})
+	app1 := newApplication(appID1, "default", "root.default")
+	err = partition.AddApplication(app1)
+	assert.NilError(t, err, "failed to add app1")
+	err = app1.AddAllocationAsk(newAllocationAsk("alloc-1", appID1, res))
+	assert.NilError(t, err, "failed to add ask to app1")
+
+	// Order log should show app1 in default queue
+	orderLog = partition.GetOrderLog()
+	assert.Equal(t, 1, len(orderLog), "should have 1 queue entry")
+	assert.Equal(t, "root.default", orderLog[0].QueueName)
+	assert.Equal(t, 1, len(orderLog[0].ApplicationIDs), "should have 1 app")
+	assert.Equal(t, appID1, orderLog[0].ApplicationIDs[0])
+
+	// Add app2 to same queue
+	app2 := newApplication(appID2, "default", "root.default")
+	err = partition.AddApplication(app2)
+	assert.NilError(t, err, "failed to add app2")
+	err = app2.AddAllocationAsk(newAllocationAsk("alloc-2", appID2, res))
+	assert.NilError(t, err, "failed to add ask to app2")
+
+	// Order log should show both apps in default queue
+	orderLog = partition.GetOrderLog()
+	assert.Equal(t, 1, len(orderLog), "should have 1 queue entry")
+	assert.Equal(t, 2, len(orderLog[0].ApplicationIDs), "should have 2 apps")
+	assert.Assert(t, contains(orderLog[0].ApplicationIDs, appID1), "should contain app1")
+	assert.Assert(t, contains(orderLog[0].ApplicationIDs, appID2), "should contain app2")
+
+	// Add app3 to production queue
+	app3 := newApplication(appID3, "production", "root.production")
+	err = partition.AddApplication(app3)
+	assert.NilError(t, err, "failed to add app3")
+	err = app3.AddAllocationAsk(newAllocationAsk("alloc-3", appID3, res))
+	assert.NilError(t, err, "failed to add ask to app3")
+
+	// Order log should show both queues
+	orderLog = partition.GetOrderLog()
+	assert.Equal(t, 2, len(orderLog), "should have 2 queue entries")
+	queueNames := make([]string, len(orderLog))
+	for i, entry := range orderLog {
+		queueNames[i] = entry.QueueName
+	}
+	assert.Assert(t, contains(queueNames, "root.default"), "should contain default queue")
+	assert.Assert(t, contains(queueNames, "root.production"), "should contain production queue")
+
+	// Remove allocation from app1 - it should disappear from order log
+	app1.RemoveAllocationAsk("alloc-1")
+	orderLog = partition.GetOrderLog()
+	assert.Equal(t, 2, len(orderLog), "should still have 2 queue entries")
+	for _, entry := range orderLog {
+		if entry.QueueName == defQueue {
+			assert.Equal(t, 1, len(entry.ApplicationIDs), "default queue should have 1 app")
+			assert.Equal(t, appID2, entry.ApplicationIDs[0], "should only have app2")
+		}
+	}
+
+	// Test snapshot immutability
+	snapshot1 := partition.GetOrderLog()
+	app4 := newApplication("app-4", "default", "root.default")
+	err = partition.AddApplication(app4)
+	assert.NilError(t, err, "failed to add app4")
+	err = app4.AddAllocationAsk(newAllocationAsk("alloc-4", "app-4", res))
+	assert.NilError(t, err, "failed to add ask to app4")
+	snapshot2 := partition.GetOrderLog()
+
+	// First snapshot should not be affected by new app
+	for _, entry := range snapshot1 {
+		if entry.QueueName == defQueue {
+			assert.Equal(t, 1, len(entry.ApplicationIDs), "snapshot1 should still have 1 app in default")
+		}
+	}
+	// Second snapshot should have the new app
+	for _, entry := range snapshot2 {
+		if entry.QueueName == defQueue {
+			assert.Equal(t, 2, len(entry.ApplicationIDs), "snapshot2 should have 2 apps in default")
+		}
+	}
+}
+
+// Helper function to check if a slice contains a string
+func contains(slice []string, str string) bool {
+	for _, s := range slice {
+		if s == str {
+			return true
+		}
+	}
+	return false
+}

pkg/webservice/dao/scheduling_order.go pkg/webservice/dao/order_log_info.gopkg/webservice/dao/scheduling_order.go renamed to pkg/webservice/dao/order_log_info.go

@@ -6,7 +6,9 @@
  to you under the Apache License, Version 2.0 (the
  "License"); you may not use this file except in compliance
  with the License.  You may obtain a copy of the License at
+
      http://www.apache.org/licenses/LICENSE-2.0
+
  Unless required by applicable law or agreed to in writing, software
  distributed under the License is distributed on an "AS IS" BASIS,
  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -16,8 +18,8 @@
 
 package dao
 
-// SchedulingOrderDAO contains queue and application information for scheduling order
-type SchedulingOrderDAO struct {
+// OrderLogEntry represents applications with pending requests in a queue
+type OrderLogEntry struct {
 	QueueName      string   `json:"queueName"`
-	ApplicationIDs []string `json:"applicationIDs"` // Application IDs that would be tried in current scheduling cycle
+	ApplicationIDs []string `json:"applicationIDs"` // Applications with pending allocation requests
 }

pkg/webservice/handlers.go

@@ -707,11 +707,11 @@ func getPartitionQueues(w http.ResponseWriter, r *http.Request) {
 		buildJSONErrorResponse(w, MissingParamsName, http.StatusBadRequest)
 		return
 	}
-	partitionName := vars.ByName("partition")
+	partition := vars.ByName("partition")
 	var partitionQueuesDAOInfo dao.PartitionQueueDAOInfo
-	var partition = schedulerContext.Load().GetPartitionWithoutClusterID(partitionName)
-	if partition != nil {
-		partitionQueuesDAOInfo = partition.GetPartitionQueues()
+	var partitionContext = schedulerContext.Load().GetPartitionWithoutClusterID(partition)
+	if partitionContext != nil {
+		partitionQueuesDAOInfo = partitionContext.GetPartitionQueues()
 	} else {
 		buildJSONErrorResponse(w, PartitionDoesNotExists, http.StatusNotFound)
 		return
@@ -721,27 +721,6 @@ func getPartitionQueues(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func getPartitionSchedulingOrder(w http.ResponseWriter, r *http.Request) {
-	writeHeaders(w, r.Method)
-	vars := httprouter.ParamsFromContext(r.Context())
-	if vars == nil {
-		buildJSONErrorResponse(w, MissingParamsName, http.StatusBadRequest)
-		return
-	}
-	partitionName := vars.ByName("partition")
-	var partitionSchedulingOrderDAOInfo []*dao.SchedulingOrderDAO
-	var partition = schedulerContext.Load().GetPartitionWithoutClusterID(partitionName)
-	if partition != nil {
-		partitionSchedulingOrderDAOInfo = partition.GetPartitionSchedulingOrder()
-	} else {
-		buildJSONErrorResponse(w, PartitionDoesNotExists, http.StatusNotFound)
-		return
-	}
-	if err := json.NewEncoder(w).Encode(partitionSchedulingOrderDAOInfo); err != nil {
-		buildJSONErrorResponse(w, err.Error(), http.StatusInternalServerError)
-	}
-}
-
 func getPartitionQueue(w http.ResponseWriter, r *http.Request) {
 	writeHeaders(w, r.Method)
 	vars := httprouter.ParamsFromContext(r.Context())
@@ -1159,6 +1138,17 @@ func getPartitionQueuesDAO(lists map[string]*scheduler.PartitionContext) []dao.P
 	return result
 }
 
+func getOrderLogDAO(lists map[string]*scheduler.PartitionContext) map[string][]*dao.OrderLogEntry {
+	result := make(map[string][]*dao.OrderLogEntry)
+
+	for _, partition := range lists {
+		partitionName := common.GetPartitionNameWithoutClusterID(partition.Name)
+		result[partitionName] = partition.GetOrderLog()
+	}
+
+	return result
+}
+
 func getClusterDAO(lists map[string]*scheduler.PartitionContext) []*dao.ClusterDAOInfo {
 	result := make([]*dao.ClusterDAOInfo, 0, len(lists))