[Metrics] Add running requests gauge metric

Author: JeffLuoo

Diff for: pkg/epp/handlers/streamingserver.go

@@ -137,6 +137,7 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 		case *extProcPb.ProcessingRequest_RequestTrailers:
 			// This is currently unused.
 		case *extProcPb.ProcessingRequest_ResponseHeaders:
+			metrics.DecRunningRequests(reqCtx.Model)
 			for _, header := range v.ResponseHeaders.Headers.GetHeaders() {
 				value := string(header.RawValue)
 
@@ -322,6 +323,7 @@ func (s *StreamingServer) HandleRequestBody(
 	if !ok {
 		return reqCtx, errutil.Error{Code: errutil.BadRequest, Msg: "model not found in request"}
 	}
+	metrics.IncRunningRequests(model)
 
 	modelName := model
 

Diff for: pkg/epp/metrics/metrics.go

@@ -121,6 +121,16 @@ var (
 		[]string{"model_name", "target_model_name"},
 	)
 
+	runningRequests = compbasemetrics.NewGaugeVec(
+		&compbasemetrics.GaugeOpts{
+			Subsystem:      InferenceModelComponent,
+			Name:           "running_requests",
+			Help:           "Inference model number of running requests in each model.",
+			StabilityLevel: compbasemetrics.ALPHA,
+		},
+		[]string{"model_name"},
+	)
+
 	// Inference Pool Metrics
 	inferencePoolAvgKVCache = compbasemetrics.NewGaugeVec(
 		&compbasemetrics.GaugeOpts{
@@ -155,6 +165,7 @@ func Register() {
 		legacyregistry.MustRegister(responseSizes)
 		legacyregistry.MustRegister(inputTokens)
 		legacyregistry.MustRegister(outputTokens)
+		legacyregistry.MustRegister(runningRequests)
 
 		legacyregistry.MustRegister(inferencePoolAvgKVCache)
 		legacyregistry.MustRegister(inferencePoolAvgQueueSize)
@@ -209,6 +220,20 @@ func RecordOutputTokens(modelName, targetModelName string, size int) {
 	}
 }
 
+// IncRunningRequests increases the current running requests.
+func IncRunningRequests(modelName string) {
+	if modelName != "" {
+		runningRequests.WithLabelValues(modelName).Inc()
+	}
+}
+
+// DecRunningRequests decreases the current running requests.
+func DecRunningRequests(modelName string) {
+	if modelName != "" {
+		runningRequests.WithLabelValues(modelName).Dec()
+	}
+}
+
 func RecordInferencePoolAvgKVCache(name string, utilization float64) {
 	inferencePoolAvgKVCache.WithLabelValues(name).Set(utilization)
 }

Diff for: pkg/epp/metrics/metrics_test.go

@@ -36,6 +36,7 @@ const (
 	ResponseSizesMetric     = InferenceModelComponent + "_response_sizes"
 	InputTokensMetric       = InferenceModelComponent + "_input_tokens"
 	OutputTokensMetric      = InferenceModelComponent + "_output_tokens"
+	RunningRequestsMetric   = InferenceModelComponent + "_running_requests"
 	KVCacheAvgUsageMetric   = InferencePoolComponent + "_average_kv_cache_utilization"
 	QueueAvgSizeMetric      = InferencePoolComponent + "_average_queue_size"
 )
@@ -345,6 +346,66 @@ func TestRecordResponseMetrics(t *testing.T) {
 	}
 }
 
+func TestRunningRequestsMetrics(t *testing.T) {
+	type request struct {
+		modelName string
+		complete  bool // true -> request is completed, false -> runing request
+	}
+
+	scenarios := []struct {
+		name     string
+		requests []request
+	}{
+		{
+			name: "basic test",
+			requests: []request{
+				{
+					modelName: "m1",
+					complete:  false,
+				},
+				{
+					modelName: "m1",
+					complete:  false,
+				},
+				{
+					modelName: "m1",
+					complete:  true,
+				},
+				{
+					modelName: "m2",
+					complete:  false,
+				},
+			},
+		},
+	}
+
+	Register()
+	for _, scenario := range scenarios {
+		t.Run(scenario.name, func(t *testing.T) {
+			for _, req := range scenario.requests {
+				if req.complete {
+					DecRunningRequests(req.modelName)
+				} else {
+					IncRunningRequests(req.modelName)
+				}
+			}
+
+			wantRunningRequests, err := os.Open("testdata/running_requests_metrics")
+			defer func() {
+				if err := wantRunningRequests.Close(); err != nil {
+					t.Error(err)
+				}
+			}()
+			if err != nil {
+				t.Fatal(err)
+			}
+			if err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, wantRunningRequests, RunningRequestsMetric); err != nil {
+				t.Error(err)
+			}
+		})
+	}
+}
+
 func TestInferencePoolMetrics(t *testing.T) {
 	scenarios := []struct {
 		name         string

Diff for: pkg/epp/metrics/testdata/running_requests_metrics

@@ -0,0 +1,4 @@
+# HELP inference_model_running_requests [ALPHA] Inference model number of running requests in each model.
+# TYPE inference_model_running_requests gauge
+inference_model_running_requests{model_name="m1"} 1
+inference_model_running_requests{model_name="m2"} 1

Diff for: site-src/guides/metrics.md

@@ -49,6 +49,7 @@ curl -i ${IP}:${PORT}/v1/completions -H 'Content-Type: application/json' -d '{
 | inference_model_response_sizes               | Distribution     | Distribution of response size in bytes.                           | `model_name`=&lt;model-name&gt; <br> `target_model_name`=&lt;target-model-name&gt; | ALPHA       |
 | inference_model_input_tokens                 | Distribution     | Distribution of input token count.                                | `model_name`=&lt;model-name&gt; <br> `target_model_name`=&lt;target-model-name&gt; | ALPHA       |
 | inference_model_output_tokens                | Distribution     | Distribution of output token count.                               | `model_name`=&lt;model-name&gt; <br> `target_model_name`=&lt;target-model-name&gt; | ALPHA       |
+| inference_model_running_requests                | Gauge     | Number of running requests for each model.             | `model_name`=&lt;model-name&gt;  | ALPHA       |
 | inference_pool_average_kv_cache_utilization  | Gauge            | The average kv cache utilization for an inference server pool.    | `name`=&lt;inference-pool-name&gt;                                                 | ALPHA       |
 | inference_pool_average_queue_size            | Gauge            | The average number of requests pending in the model server queue. | `name`=&lt;inference-pool-name&gt;                                                 | ALPHA       |