Added usage attributes (#418)

Author: itsomri

cmd/fairshare-simulator/main.go

@@ -56,7 +56,7 @@ func (s *server) simulateHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	queues := SimulateSetResourcesShare(req.TotalResource, req.Queues)
+	queues := SimulateSetResourcesShare(req.TotalResource, 0, req.Queues)
 
 	resp := make(map[string]QueueFairShare)
 	for id, qa := range queues {
@@ -92,12 +92,12 @@ func main() {
 	}
 }
 
-func SimulateSetResourcesShare(totalResource rs.ResourceQuantities, queueOverrides []rs.QueueOverrides) map[common_info.QueueID]*rs.QueueAttributes {
+func SimulateSetResourcesShare(totalResource rs.ResourceQuantities, kValue float64, queueOverrides []rs.QueueOverrides) map[common_info.QueueID]*rs.QueueAttributes {
 	queues := make(map[common_info.QueueID]*rs.QueueAttributes)
 	for _, qo := range queueOverrides {
 		qa := qo.ToQueueAttributes()
 		queues[qa.UID] = qa
 	}
-	resource_division.SetResourcesShare(totalResource, queues)
+	resource_division.SetResourcesShare(totalResource, kValue, queues)
 	return queues
 }

pkg/scheduler/api/queue_info/queue_info.go

@@ -35,6 +35,7 @@ type QueueInfo struct {
 	ParentQueue       common_info.QueueID
 	ChildQueues       []common_info.QueueID
 	Resources         QueueQuota
+	ResourceUsage     QueueUsage
 	Priority          int
 	CreationTimestamp metav1.Time
 	PreemptMinRuntime *metav1.Duration

pkg/scheduler/api/queue_info/quota_info.go

@@ -27,13 +27,11 @@ type QueueUsage struct {
 }
 
 type ClusterUsage struct {
-	ClusterCapacity QueueUsage                          `json:"cluster"`
-	Queues          map[common_info.QueueID]*QueueUsage `json:"queues"`
+	Queues map[common_info.QueueID]*QueueUsage `json:"queues"`
 }
 
 func NewClusterUsage() *ClusterUsage {
 	return &ClusterUsage{
-		ClusterCapacity: QueueUsage{},
-		Queues:          make(map[common_info.QueueID]*QueueUsage),
+		Queues: make(map[common_info.QueueID]*QueueUsage),
 	}
 }

pkg/scheduler/cache/cluster_info/cluster_info_test.go

@@ -173,11 +173,6 @@ func TestSnapshotUsage(t *testing.T) {
 		{
 			name: "BasicUsage",
 			usage: &queue_info.ClusterUsage{
-				ClusterCapacity: queue_info.QueueUsage{
-					CPU:    10,
-					Memory: 10,
-					GPU:    10,
-				},
 				Queues: map[common_info.QueueID]*queue_info.QueueUsage{
 					"queue-1": {
 						CPU:    10,
@@ -188,11 +183,6 @@ func TestSnapshotUsage(t *testing.T) {
 			},
 			err: nil,
 			expectedUsage: &queue_info.ClusterUsage{
-				ClusterCapacity: queue_info.QueueUsage{
-					CPU:    10,
-					Memory: 10,
-					GPU:    10,
-				},
 				Queues: map[common_info.QueueID]*queue_info.QueueUsage{
 					"queue-1": {
 						CPU:    10,
@@ -211,11 +201,6 @@ func TestSnapshotUsage(t *testing.T) {
 		{
 			name: "Error and usage",
 			usage: &queue_info.ClusterUsage{
-				ClusterCapacity: queue_info.QueueUsage{
-					CPU:    11,
-					Memory: 11,
-					GPU:    11,
-				},
 				Queues: map[common_info.QueueID]*queue_info.QueueUsage{
 					"queue-1": {
 						CPU:    11,
@@ -235,7 +220,6 @@ func TestSnapshotUsage(t *testing.T) {
 			return
 		}
 		assert.NotNil(t, actual)
-		assert.Equal(t, expected.ClusterCapacity, actual.ClusterCapacity)
 		assert.Equal(t, len(expected.Queues), len(actual.Queues))
 		for queueID, expectedUsage := range expected.Queues {
 			actualUsage, ok := actual.Queues[queueID]

pkg/scheduler/cache/usagedb/usagedb_test.go

@@ -87,15 +87,13 @@ func TestGetResourceUsage(t *testing.T) {
 			name: "fresh data available",
 			setupLister: func(l *UsageLister) {
 				usage := queue_info.NewClusterUsage()
-				usage.ClusterCapacity.GPU = 10
 				usage.Queues["queue1"] = &queue_info.QueueUsage{GPU: 5}
 				now := time.Now()
 				l.lastUsageData = usage
 				l.lastUsageDataTime = &now
 			},
 			wantUsage: func() *queue_info.ClusterUsage {
 				usage := queue_info.NewClusterUsage()
-				usage.ClusterCapacity.GPU = 10
 				usage.Queues["queue1"] = &queue_info.QueueUsage{GPU: 5}
 				return usage
 			}(),
@@ -104,14 +102,12 @@ func TestGetResourceUsage(t *testing.T) {
 			name: "stale data",
 			setupLister: func(l *UsageLister) {
 				usage := queue_info.NewClusterUsage()
-				usage.ClusterCapacity.GPU = 10
 				staleTime := time.Now().Add(-10 * time.Minute)
 				l.lastUsageData = usage
 				l.lastUsageDataTime = &staleTime
 			},
 			wantUsage: func() *queue_info.ClusterUsage {
 				usage := queue_info.NewClusterUsage()
-				usage.ClusterCapacity.GPU = 10
 				return usage
 			}(),
 			wantErr: true,

pkg/scheduler/metrics/metrics.go

@@ -51,6 +51,9 @@ var (
 	queueFairShareCPU           *prometheus.GaugeVec
 	queueFairShareMemory        *prometheus.GaugeVec
 	queueFairShareGPU           *prometheus.GaugeVec
+	queueCPUUsage               *prometheus.GaugeVec
+	queueMemoryUsage            *prometheus.GaugeVec
+	queueGPUUsage               *prometheus.GaugeVec
 )
 
 func init() {
@@ -168,6 +171,25 @@ func InitMetrics(namespace string) {
 			Name:      "queue_fair_share_gpu",
 			Help:      "GPU Fair share of queue, as a gauge. Values in GPU devices",
 		}, []string{"queue_name"})
+
+	queueCPUUsage = promauto.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: namespace,
+			Name:      "queue_cpu_usage_cores",
+			Help:      "CPU usage of queue, as a gauge. Value is proportional to cpu*hours usage with time decay applied",
+		}, []string{"queue_name"})
+	queueMemoryUsage = promauto.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: namespace,
+			Name:      "queue_memory_usage_gb",
+			Help:      "Memory usage of queue, as a gauge. Value is proportional to memory*hours usage with time decay applied",
+		}, []string{"queue_name"})
+	queueGPUUsage = promauto.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: namespace,
+			Name:      "queue_gpu_usage_devices",
+			Help:      "GPU usage of queue, as a gauge. Value is proportional to gpu*hours usage with time decay applied",
+		}, []string{"queue_name"})
 }
 
 // UpdateOpenSessionDuration updates latency for open session, including all plugins
@@ -242,6 +264,19 @@ func ResetQueueFairShare() {
 	queueFairShareGPU.Reset()
 }
 
+// UpdateQueueUsage updates usage of queue for a resource
+func UpdateQueueUsage(queueName string, cpu, memory, gpu float64) {
+	queueCPUUsage.WithLabelValues(queueName).Set(cpu)
+	queueMemoryUsage.WithLabelValues(queueName).Set(memory)
+	queueGPUUsage.WithLabelValues(queueName).Set(gpu)
+}
+
+func ResetQueueUsage() {
+	queueCPUUsage.Reset()
+	queueMemoryUsage.Reset()
+	queueGPUUsage.Reset()
+}
+
 // RegisterPreemptionAttempts records number of attempts for preemption
 func RegisterPreemptionAttempts() {
 	preemptionAttempts.Inc()

pkg/scheduler/plugins/proportion/proportion.go

@@ -220,6 +220,7 @@ func (pp *proportionPlugin) calculateResourcesProportion(ssn *framework.Session)
 	log.InfraLogger.V(6).Infof("Calculating resource proportion")
 
 	pp.setTotalResources(ssn)
+
 	pp.createQueueAttributes(ssn)
 	log.InfraLogger.V(3).Infof("Total allocatable resources are <%s>, number of nodes: <%d>, number of "+
 		"queues: <%d>", pp.totalResource, len(ssn.Nodes), len(pp.queues))
@@ -322,6 +323,11 @@ func (pp *proportionPlugin) createQueueResourceAttrs(ssn *framework.Session) {
 		overQuotaWeight = queue.Resources.GPU.OverQuotaWeight
 		queueAttributes.SetQuotaResources(rs.GpuResource, deserved, limit, overQuotaWeight)
 
+		usage, found := ssn.ResourceUsage.Queues[queue.UID]
+		if found {
+			queueAttributes.SetResourceUsage(*usage)
+		}
+
 		pp.queues[queue.UID] = queueAttributes
 		log.InfraLogger.V(7).Infof("Added queue attributes for queue <%s>", queue.Name)
 	}
@@ -380,22 +386,23 @@ func (pp *proportionPlugin) updateQueuesResourceUsageForPendingJob(queueId commo
 
 func (pp *proportionPlugin) setFairShare() {
 	topQueues := pp.getTopQueues()
+	metrics.ResetQueueUsage()
 	metrics.ResetQueueFairShare()
-	pp.setFairShareForQueues(pp.totalResource, topQueues)
+	pp.setFairShareForQueues(pp.totalResource, 1, topQueues)
 }
 
-func (pp *proportionPlugin) setFairShareForQueues(totalResources rs.ResourceQuantities,
+func (pp *proportionPlugin) setFairShareForQueues(totalResources rs.ResourceQuantities, kValue float64,
 	queues map[common_info.QueueID]*rs.QueueAttributes) {
 
 	if len(queues) == 0 {
 		return
 	}
 
-	resource_division.SetResourcesShare(totalResources, queues)
+	resource_division.SetResourcesShare(totalResources, kValue, queues)
 	for _, queue := range queues {
 		childQueues := pp.getChildQueues(queue)
 		resources := queue.GetFairShare()
-		pp.setFairShareForQueues(resources, childQueues)
+		pp.setFairShareForQueues(resources, kValue, childQueues)
 	}
 }
 

pkg/scheduler/plugins/proportion/resource_division/resource_division.go

@@ -23,18 +23,18 @@ type remainingRequestedResource struct {
 	remainingAmount float64
 }
 
-func SetResourcesShare(totalResource rs.ResourceQuantities, queues map[common_info.QueueID]*rs.QueueAttributes) {
+func SetResourcesShare(totalResource rs.ResourceQuantities, kValue float64, queues map[common_info.QueueID]*rs.QueueAttributes) {
 	for _, resource := range rs.AllResources {
-		setResourceShare(totalResource[resource], resource, queues)
+		setResourceShare(totalResource[resource], kValue, resource, queues)
 	}
 	reportDivisionResult(queues)
 }
 
-func setResourceShare(totalAmount float64, resourceName rs.ResourceName, queues map[common_info.QueueID]*rs.QueueAttributes) float64 {
+func setResourceShare(totalAmount, kValue float64, resourceName rs.ResourceName, queues map[common_info.QueueID]*rs.QueueAttributes) float64 {
 	log.InfraLogger.V(6).Infof("About to start calculating %v fairShare, totalAmount: <%v>", resourceName, totalAmount)
 	remainingAmount := setDeservedResource(totalAmount, queues, resourceName)
 	if remainingAmount > 0 {
-		remainingAmount = divideOverQuotaResource(remainingAmount, queues, resourceName)
+		remainingAmount = divideOverQuotaResource(remainingAmount, kValue, queues, resourceName)
 	} else {
 		remainingAmount = 0
 	}
@@ -54,24 +54,34 @@ func reportDivisionResult(queues map[common_info.QueueID]*rs.QueueAttributes) {
 			gpuResourceShare.FairShare,
 		)
 
+		metrics.UpdateQueueUsage(
+			queue.Name,
+			cpuResourceShare.GetUsage(),
+			memoryResourceShare.GetUsage(),
+			gpuResourceShare.GetUsage(),
+		)
+
 		log.InfraLogger.V(3).Infof("Resource division result for queue <%v>: "+
 			"Queue Priority: <%d>, "+
-			"GPU: deserved: <%v>, requested: <%v>, maxAllowed: <%v>, fairShare: <%v> "+
-			"CPU (cores): deserved: <%v>, requested: <%v>, maxAllowed: <%v>, fairShare: <%v> "+
-			"memory (GB): deserved: <%v>, requested: <%v>, maxAllowed: <%v>, fairShare: <%v> ",
+			"GPU: deserved: <%v>, requested: <%v>, maxAllowed: <%v>, usage: <%v>, fairShare: <%v> "+
+			"CPU (cores): deserved: <%v>, requested: <%v>, maxAllowed: <%v>, usage: <%v>, fairShare: <%v> "+
+			"memory (GB): deserved: <%v>, requested: <%v>, maxAllowed: <%v>, usage: <%v>, fairShare: <%v> ",
 			queue.Name,
 			queue.Priority,
 			resource_info.HumanizeResource(gpuResourceShare.Deserved, 1),
 			resource_info.HumanizeResource(gpuResourceShare.GetRequestableShare(), 1),
 			resource_info.HumanizeResource(gpuResourceShare.MaxAllowed, 1),
+			resource_info.HumanizeResource(gpuResourceShare.GetUsage(), 1),
 			resource_info.HumanizeResource(gpuResourceShare.FairShare, 1),
 			resource_info.HumanizeResource(cpuResourceShare.Deserved, resource_info.MilliCPUToCores),
 			resource_info.HumanizeResource(cpuResourceShare.GetRequestableShare(), resource_info.MilliCPUToCores),
 			resource_info.HumanizeResource(cpuResourceShare.MaxAllowed, resource_info.MilliCPUToCores),
+			resource_info.HumanizeResource(cpuResourceShare.GetUsage(), resource_info.MilliCPUToCores),
 			resource_info.HumanizeResource(cpuResourceShare.FairShare, resource_info.MilliCPUToCores),
 			resource_info.HumanizeResource(memoryResourceShare.Deserved, resource_info.MemoryToGB),
 			resource_info.HumanizeResource(memoryResourceShare.GetRequestableShare(), resource_info.MemoryToGB),
 			resource_info.HumanizeResource(memoryResourceShare.MaxAllowed, resource_info.MemoryToGB),
+			resource_info.HumanizeResource(memoryResourceShare.GetUsage(), resource_info.MemoryToGB),
 			resource_info.HumanizeResource(memoryResourceShare.FairShare, resource_info.MemoryToGB))
 	}
 }
@@ -95,15 +105,15 @@ func setDeservedResource(
 	return remainingAmount
 }
 
-func divideOverQuotaResource(totalResourceAmount float64, queues map[common_info.QueueID]*rs.QueueAttributes,
+func divideOverQuotaResource(totalResourceAmount, kValue float64, queues map[common_info.QueueID]*rs.QueueAttributes,
 	resourceName rs.ResourceName) (remainingAmount float64) {
 	queuesByPriority, priorities := getQueuesByPriority(queues)
 	remainingRequested := make(map[int]map[common_info.QueueID]*remainingRequestedResource)
 	remainingAmount = totalResourceAmount
 
 	for _, priority := range priorities {
 		var newRemainingRequested map[common_info.QueueID]*remainingRequestedResource
-		remainingAmount, newRemainingRequested = divideUpToFairShare(remainingAmount, queuesByPriority[priority], resourceName)
+		remainingAmount, newRemainingRequested = divideUpToFairShare(remainingAmount, kValue, queuesByPriority[priority], resourceName)
 		if remainingRequested[priority] == nil {
 			remainingRequested[priority] = make(map[common_info.QueueID]*remainingRequestedResource)
 		}
@@ -148,9 +158,10 @@ func getQueuesByPriority(queues map[common_info.QueueID]*rs.QueueAttributes) (ma
 	return queuesByPriority, priorities
 }
 
-func divideUpToFairShare(totalResourceAmount float64, queues map[common_info.QueueID]*rs.QueueAttributes,
+func divideUpToFairShare(totalResourceAmount, kValue float64, queues map[common_info.QueueID]*rs.QueueAttributes,
 	resourceName rs.ResourceName) (remainingAmount float64, remainingRequested map[common_info.QueueID]*remainingRequestedResource) {
 	remainingRequested = map[common_info.QueueID]*remainingRequestedResource{}
+
 	for {
 		shouldRunAnotherRound := false
 		amountToGiveInCurrentRound := totalResourceAmount