[YUNIKORN-2031] Fix a bug on metric of pending pods/resources

go.mod

@@ -45,6 +45,7 @@ require (
 	github.com/HdrHistogram/hdrhistogram-go v1.0.1 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
 	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect

pkg/scheduler/objects/application.go

@@ -376,7 +376,7 @@ func (sa *Application) timeoutStateTimer(expectedState string, event application
 				sa.notifyRMAllocationReleased(sa.rmID, toRelease, si.TerminationType_TIMEOUT, "releasing placeholders on app complete")
 				sa.clearStateTimer()
 			} else {
-				//nolint: errcheck
+				// nolint: errcheck
 				_ = sa.HandleApplicationEvent(event)
 			}
 		}

pkg/scheduler/objects/queue.go

@@ -651,6 +651,7 @@
 	sq.Lock()
 	defer sq.Unlock()
 	sq.pending = resources.Add(sq.pending, delta)
+	sq.updatePendingResourceMetrics()
 }
 
 // decPendingResource decrements pending resource of this queue and its parents.
@@ -671,6 +672,8 @@
 		log.Log(log.SchedQueue).Warn("Pending resources went negative",
 			zap.String("queueName", sq.QueuePath),
 			zap.Error(err))
+	} else {
+		sq.updatePendingResourceMetrics()
 	}
 }
 
@@ -1017,7 +1020,7 @@
 	}
 	// all OK update this queue
 	sq.allocatedResource = newAllocated
-	sq.updateAllocatedAndPendingResourceMetrics()
+	sq.updateAllocatedResourceMetrics()
 	return nil
 }
 
@@ -1053,7 +1056,7 @@
 	}
 	// all OK update the queue
 	sq.allocatedResource = resources.Sub(sq.allocatedResource, alloc)
-	sq.updateAllocatedAndPendingResourceMetrics()
+	sq.updateAllocatedResourceMetrics()
 	return nil
 }
 
@@ -1066,7 +1069,7 @@
 	defer sq.Unlock()
 	sq.parent.IncPreemptingResource(alloc)
 	sq.preemptingResource = resources.Add(sq.preemptingResource, alloc)
-	sq.updateAllocatedAndPendingResourceMetrics()
+	sq.updatePreemptingResourceMetrics()
 }
 
 // DecPreemptingResource decrements the preempting resources for this queue (recursively).
@@ -1078,7 +1081,7 @@
 	defer sq.Unlock()
 	sq.parent.DecPreemptingResource(alloc)
 	sq.preemptingResource = resources.Sub(sq.preemptingResource, alloc)
-	sq.updateAllocatedAndPendingResourceMetrics()
+	sq.updatePreemptingResourceMetrics()
 }
 
 func (sq *Queue) IsPrioritySortEnabled() bool {
@@ -1553,13 +1556,29 @@
 }
 
 // updateAllocatedAndPendingResourceMetrics updates allocated and pending resource metrics for all queue types.
+// Deprecated: use specific metric update function for efficiency.
 func (sq *Queue) updateAllocatedAndPendingResourceMetrics() {
+	sq.updateAllocatedResourceMetrics()
+	sq.updatePendingResourceMetrics()
+	sq.updatePreemptingResourceMetrics()
+}
+
+// updateAllocatedResourceMetrics updates allocated resource metrics for all queue types.
+func (sq *Queue) updateAllocatedResourceMetrics() {
 	for k, v := range sq.allocatedResource.Resources {
 		metrics.GetQueueMetrics(sq.QueuePath).SetQueueAllocatedResourceMetrics(k, float64(v))
 	}
+}
+
+// updatePendingResourceMetrics updates pending resource metrics for all queue types.
+func (sq *Queue) updatePendingResourceMetrics() {
 	for k, v := range sq.pending.Resources {
 		metrics.GetQueueMetrics(sq.QueuePath).SetQueuePendingResourceMetrics(k, float64(v))
 	}
+}
+
+// updatePendingResourceMetrics updates preempting resource metrics for all queue types.
+func (sq *Queue) updatePreemptingResourceMetrics() {
 	for k, v := range sq.preemptingResource.Resources {
 		metrics.GetQueueMetrics(sq.QueuePath).SetQueuePreemptingResourceMetrics(k, float64(v))
 	}

pkg/scheduler/objects/queue_test.go

@@ -23,15 +23,19 @@ import (
 	"reflect"
 	"sort"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+	promtu "github.com/prometheus/client_golang/prometheus/testutil"
 	"gotest.tools/v3/assert"
 
 	"github.com/apache/yunikorn-core/pkg/common"
 	"github.com/apache/yunikorn-core/pkg/common/configs"
 	"github.com/apache/yunikorn-core/pkg/common/resources"
 	"github.com/apache/yunikorn-core/pkg/events"
+	"github.com/apache/yunikorn-core/pkg/metrics"
 	"github.com/apache/yunikorn-core/pkg/scheduler/objects/template"
 	"github.com/apache/yunikorn-core/pkg/scheduler/policies"
 	siCommon "github.com/apache/yunikorn-scheduler-interface/lib/go/common"
@@ -266,6 +270,11 @@ func TestPriorityCalc(t *testing.T) {
 }
 
 func TestPendingCalc(t *testing.T) {
+	// Reset existing metric storage; otherwise this unit test would get metrics populated by other UTs.
+	// In long run, to make the metrics code more testable, we should pass instantiable Metrics obj to Queue
+	// instead of using a global Metrics obj at pkg/metrics/init.go.
+	metrics.Reset()
+
 	// create the root
 	root, err := createRootQueue(nil)
 	assert.NilError(t, err, "queue create failed")
@@ -284,13 +293,32 @@ func TestPendingCalc(t *testing.T) {
 	if !resources.Equals(leaf.pending, allocRes) {
 		t.Errorf("leaf queue pending allocation failed to increment expected %v, got %v", allocRes, leaf.pending)
 	}
+	metrics := []string{"yunikorn_root_queue_resource", "yunikorn_root_leaf_queue_resource"}
+	want := concatQueueResourceMetric(metrics, []string{`
+yunikorn_root_queue_resource{resource="memory",state="pending"} 100
+yunikorn_root_queue_resource{resource="vcores",state="pending"} 10
+`, `
+yunikorn_root_leaf_queue_resource{resource="memory",state="pending"} 100
+yunikorn_root_leaf_queue_resource{resource="vcores",state="pending"} 10
+`},
+	)
+	assert.NilError(t, promtu.GatherAndCompare(prometheus.DefaultGatherer, strings.NewReader(want), metrics...), "unexpected metrics")
 	leaf.decPendingResource(allocRes)
 	if !resources.IsZero(root.pending) {
 		t.Errorf("root queue pending allocation failed to decrement expected 0, got %v", root.pending)
 	}
 	if !resources.IsZero(leaf.pending) {
 		t.Errorf("leaf queue pending allocation failed to decrement expected 0, got %v", leaf.pending)
 	}
+	want = concatQueueResourceMetric(metrics, []string{`
+yunikorn_root_queue_resource{resource="memory",state="pending"} 0
+yunikorn_root_queue_resource{resource="vcores",state="pending"} 0
+`, `
+yunikorn_root_leaf_queue_resource{resource="memory",state="pending"} 0
+yunikorn_root_leaf_queue_resource{resource="vcores",state="pending"} 0
+`},
+	)
+	assert.NilError(t, promtu.GatherAndCompare(prometheus.DefaultGatherer, strings.NewReader(want), metrics...), "unexpected metrics")
 	// Not allowed to go negative: both will be zero after this
 	newRes := resources.Multiply(allocRes, 2)
 	root.pending = newRes
@@ -302,6 +330,30 @@ func TestPendingCalc(t *testing.T) {
 	if !resources.IsZero(leaf.GetPendingResource()) {
 		t.Errorf("leaf queue pending allocation should have failed to decrement expected zero, got %v", leaf.pending)
 	}
+	want = concatQueueResourceMetric(metrics, []string{`
+yunikorn_root_queue_resource{resource="memory",state="pending"} 0
+yunikorn_root_queue_resource{resource="vcores",state="pending"} 0
+`, `
+yunikorn_root_leaf_queue_resource{resource="memory",state="pending"} 0
+yunikorn_root_leaf_queue_resource{resource="vcores",state="pending"} 0
+`},
+	)
+	assert.NilError(t, promtu.GatherAndCompare(prometheus.DefaultGatherer, strings.NewReader(want), metrics...), "unexpected metrics")
+}
+
+const (
+	QueueResourceMetricHelp = "# HELP %v Queue resource metrics. State of the resource includes `guaranteed`, `max`, `allocated`, `pending`, `preempting`."
+	QueueResourceMetricType = "# TYPE %v gauge"
+)
+
+func concatQueueResourceMetric(metricNames, metricVals []string) string {
+	var out string
+	for i, metricName := range metricNames {
+		out = out + fmt.Sprintf(QueueResourceMetricHelp, metricName) + "\n"
+		out = out + fmt.Sprintf(QueueResourceMetricType, metricName) + "\n"
+		out += strings.TrimLeft(metricVals[i], "\n")
+	}
+	return out
 }
 
 func TestGetChildQueueInfo(t *testing.T) {