feat: Add metric that records length of queue for each model server pods (kubernetes-sigs#776)

This metric is helpful to determine if the scheduling algorithm is
working as expected to load the traffic to each pod.

New metric uses the prometheus collector because the pod list will
change each time the underlying model server deployment changes and will
cause the metric cardinality to grow.

Author: JeffLuoo

cmd/epp/main.go

@@ -41,6 +41,7 @@ import (
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics/collectors"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling"
 	runserver "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/server"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
@@ -199,7 +200,7 @@ func run() error {
 	}
 
 	// Register metrics handler.
-	if err := registerMetricsHandler(mgr, *metricsPort, cfg); err != nil {
+	if err := registerMetricsHandler(mgr, *metricsPort, cfg, datastore); err != nil {
 		return err
 	}
 
@@ -247,8 +248,9 @@ func registerHealthServer(mgr manager.Manager, logger logr.Logger, ds datastore.
 }
 
 // registerMetricsHandler adds the metrics HTTP handler as a Runnable to the given manager.
-func registerMetricsHandler(mgr manager.Manager, port int, cfg *rest.Config) error {
+func registerMetricsHandler(mgr manager.Manager, port int, cfg *rest.Config, ds datastore.Datastore) error {
 	metrics.Register()
+	legacyregistry.CustomMustRegister(collectors.NewInferencePoolMetricsCollector(ds))
 
 	metrics.RecordInferenceExtensionInfo()
 

pkg/epp/metrics/collectors/inference_pool.go

@@ -0,0 +1,80 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed 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.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package collectors
+
+import (
+	"k8s.io/component-base/metrics"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
+)
+
+var (
+	descInferencePoolPerPodQueueSize = metrics.NewDesc(
+		"inference_pool_per_pod_queue_size",
+		"The total number of requests pending in the model server queue for each underlying pod.",
+		[]string{
+			"name",
+			"model_server_pod",
+		}, nil,
+		metrics.ALPHA,
+		"",
+	)
+)
+
+type inferencePoolMetricsCollector struct {
+	metrics.BaseStableCollector
+
+	ds datastore.Datastore
+}
+
+// Check if inferencePoolMetricsCollector implements necessary interface
+var _ metrics.StableCollector = &inferencePoolMetricsCollector{}
+
+// NewInferencePoolMetricsCollector implements the metrics.StableCollector interface and
+// exposes metrics about inference pool.
+func NewInferencePoolMetricsCollector(ds datastore.Datastore) metrics.StableCollector {
+	return &inferencePoolMetricsCollector{
+		ds: ds,
+	}
+}
+
+// DescribeWithStability implements the metrics.StableCollector interface.
+func (c *inferencePoolMetricsCollector) DescribeWithStability(ch chan<- *metrics.Desc) {
+	ch <- descInferencePoolPerPodQueueSize
+}
+
+// CollectWithStability implements the metrics.StableCollector interface.
+func (c *inferencePoolMetricsCollector) CollectWithStability(ch chan<- metrics.Metric) {
+	pool, err := c.ds.PoolGet()
+	if err != nil {
+		return
+	}
+
+	podMetrics := c.ds.PodGetAll()
+	if len(podMetrics) == 0 {
+		return
+	}
+
+	for _, pod := range podMetrics {
+		ch <- metrics.NewLazyConstMetric(
+			descInferencePoolPerPodQueueSize,
+			metrics.GaugeValue,
+			float64(pod.GetMetrics().WaitingQueueSize),
+			pool.Name,
+			pod.GetPod().NamespacedName.Name,
+		)
+	}
+}

pkg/epp/metrics/collectors/inference_pool_test.go

@@ -0,0 +1,101 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed 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.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package collectors
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/component-base/metrics/testutil"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/gateway-api-inference-extension/api/v1alpha2"
+	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
+)
+
+var (
+	pod1 = &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "pod1",
+		},
+	}
+	pod1NamespacedName = types.NamespacedName{Name: pod1.Name, Namespace: pod1.Namespace}
+	pod1Metrics        = &backendmetrics.Metrics{
+		WaitingQueueSize:    100,
+		KVCacheUsagePercent: 0.2,
+		MaxActiveModels:     2,
+	}
+)
+
+func TestNoMetricsCollected(t *testing.T) {
+	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.FakePodMetricsClient{}, time.Second)
+	datastore := datastore.NewDatastore(context.Background(), pmf)
+
+	collector := &inferencePoolMetricsCollector{
+		ds: datastore,
+	}
+
+	if err := testutil.CustomCollectAndCompare(collector, strings.NewReader(""), ""); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestMetricsCollected(t *testing.T) {
+	pmc := &backendmetrics.FakePodMetricsClient{
+		Res: map[types.NamespacedName]*backendmetrics.Metrics{
+			pod1NamespacedName: pod1Metrics,
+		},
+	}
+	pmf := backendmetrics.NewPodMetricsFactory(pmc, time.Millisecond)
+	ds := datastore.NewDatastore(context.Background(), pmf)
+
+	scheme := runtime.NewScheme()
+	fakeClient := fake.NewClientBuilder().
+		WithScheme(scheme).
+		Build()
+
+	inferencePool := &v1alpha2.InferencePool{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-pool",
+		},
+		Spec: v1alpha2.InferencePoolSpec{
+			TargetPortNumber: 8000,
+		},
+	}
+	_ = ds.PoolSet(context.Background(), fakeClient, inferencePool)
+	_ = ds.PodUpdateOrAddIfNotExist(pod1)
+
+	time.Sleep(1 * time.Second)
+
+	collector := &inferencePoolMetricsCollector{
+		ds: ds,
+	}
+	err := testutil.CustomCollectAndCompare(collector, strings.NewReader(`
+		# HELP inference_pool_per_pod_queue_size [ALPHA] The total number of requests pending in the model server queue for each underlying pod.
+		# TYPE inference_pool_per_pod_queue_size gauge
+		inference_pool_per_pod_queue_size{model_server_pod="pod1",name="test-pool"} 100
+`), "inference_pool_per_pod_queue_size")
+	if err != nil {
+		t.Fatal(err)
+	}
+}

pkg/epp/metrics/metrics_test.go

@@ -40,6 +40,7 @@ const (
 	RunningRequestsMetric              = InferenceModelComponent + "_running_requests"
 	KVCacheAvgUsageMetric              = InferencePoolComponent + "_average_kv_cache_utilization"
 	QueueAvgSizeMetric                 = InferencePoolComponent + "_average_queue_size"
+	PerPodQueueSizeMetrics             = InferencePoolComponent + "_per_pod_queue_size"
 )
 
 func TestRecordRequestCounterandSizes(t *testing.T) {

site-src/guides/metrics.md

@@ -34,6 +34,7 @@ curl -i ${IP}:${PORT}/v1/completions -H 'Content-Type: application/json' -d '{
 | inference_model_running_requests                | Gauge     | Number of running requests for each model.             | `model_name`=<model-name>  | ALPHA       |
 | inference_pool_average_kv_cache_utilization  | Gauge            | The average kv cache utilization for an inference server pool.    | `name`=<inference-pool-name>                                                 | ALPHA       |
 | inference_pool_average_queue_size            | Gauge            | The average number of requests pending in the model server queue. | `name`=<inference-pool-name>                                                 | ALPHA       |
+| inference_pool_per_pod_queue_size            | Gauge            | The total number of queue for each model server pod under the inference pool         | `model_server_pod`=<model-server-pod-name> `name`=<inference-pool-name>                             | ALPHA       |
 | inference_pool_ready_pods                    | Gauge            | The number of ready pods for an inference server pool.            | `name`=<inference-pool-name>                                                 | ALPHA       |
 | inference_extension_info                     | Gauge            | The general information of the current build.                     | `commit`=<hash-of-the-build>                                                 | ALPHA       |