add EPP request/response processing latency metrics

Signed-off-by: satyamg1620 <Satyam.Gupta.3@ibm.com>

Author: satyamg1620

pkg/epp/handlers/server.go

@@ -136,6 +136,10 @@ type RequestContext struct {
 	RequestDroppedReason errcommon.RequestDroppedReason
 	modelServerStreaming bool
 
+	// responseProcessingDuration accumulates the time EPP spends in its response
+	// handlers across all response events, excluding model server generation time.
+	responseProcessingDuration time.Duration
+
 	Response *Response
 
 	reqHeaderResp  *extProcPb.ProcessingResponse
@@ -465,6 +469,7 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 		case *extProcPb.ProcessingRequest_RequestTrailers:
 			// This is currently unused.
 		case *extProcPb.ProcessingRequest_ResponseHeaders:
+			respHeaderStart := time.Now()
 			for _, header := range v.ResponseHeaders.Headers.GetHeaders() {
 				value := string(header.RawValue)
 				loggerTrace.Info("header", "key", header.Key, "value", value)
@@ -478,12 +483,14 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 			reqCtx.RequestState = ResponseReceived
 			reqCtx = s.HandleResponseHeaders(ctx, reqCtx, v)
 			reqCtx.respHeaderResp = s.generateResponseHeaderResponse(reqCtx)
+			reqCtx.responseProcessingDuration += time.Since(respHeaderStart)
 
 		case *extProcPb.ProcessingRequest_ResponseBody:
 			endOfStream := v.ResponseBody.EndOfStream
 			chunk := v.ResponseBody.Body
 
 			if reqCtx.modelServerStreaming {
+				respBodyStart := time.Now()
 				if endOfStream {
 					reqCtx.ResponseComplete = true
 					reqCtx.ResponseCompleteTimestamp = time.Now()
@@ -493,6 +500,10 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 				chunk = rewriteModelName(chunk, reqCtx.TargetModelName, reqCtx.IncomingModelName)
 				// For streaming response, we send response chunk back to envoy every time we received it.
 				reqCtx.respBodyResp = generateResponseBodyResponses(chunk, endOfStream, reqCtx.Response.DynamicMetadata)
+				reqCtx.responseProcessingDuration += time.Since(respBodyStart)
+				if endOfStream {
+					metrics.RecordResponseProcessingLatency(reqCtx.responseProcessingDuration)
+				}
 			} else {
 				respBody = append(respBody, chunk...)
 				if endOfStream {
@@ -553,6 +564,7 @@ func (s *StreamingServer) finishResponse(ctx context.Context, reqCtx *RequestCon
 		return
 	}
 
+	start := time.Now()
 	reqCtx.ResponseComplete = true
 	reqCtx.ResponseCompleteTimestamp = time.Now()
 	reqCtx = s.HandleResponseBody(ctx, reqCtx, body, true)
@@ -562,6 +574,8 @@ func (s *StreamingServer) finishResponse(ctx context.Context, reqCtx *RequestCon
 		// For non-streaming response, we send response back to envoy after receiving all the response body.
 		reqCtx.respBodyResp = generateResponseBodyResponses(body, setEos, reqCtx.Response.DynamicMetadata)
 	}
+	reqCtx.responseProcessingDuration += time.Since(start)
+	metrics.RecordResponseProcessingLatency(reqCtx.responseProcessingDuration)
 }
 
 // rewriteModelName replaces occurrences of the target (internal) model name with the

pkg/epp/metrics/llm_d_router_metrics.go

@@ -252,6 +252,30 @@ var (
 		},
 		[]string{"extension_point", "plugin_type", "plugin_name"},
 	)
+
+	llmdRequestProcessingLatency = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Subsystem: LLMDRouterEndpointPickerSubsystem,
+			Name:      "request_processing_duration_seconds",
+			Help:      metricsutil.HelpMsgWithStability("EPP request processing latency distribution in seconds, from request receipt until an endpoint is selected, excluding flow-control admission wait.", compbasemetrics.ALPHA),
+			Buckets: []float64{
+				0.0001, 0.0002, 0.0005, 0.001, 0.002, 0.005, 0.01, 0.02, 0.05, 0.1,
+			},
+		},
+		[]string{},
+	)
+
+	llmdResponseProcessingLatency = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Subsystem: LLMDRouterEndpointPickerSubsystem,
+			Name:      "response_processing_duration_seconds",
+			Help:      metricsutil.HelpMsgWithStability("EPP response processing latency distribution in seconds, accumulated across response handlers and excluding model server generation time.", compbasemetrics.ALPHA),
+			Buckets: []float64{
+				0.0001, 0.0002, 0.0005, 0.001, 0.002, 0.005, 0.01, 0.02, 0.05, 0.1,
+			},
+		},
+		[]string{},
+	)
 )
 
 // --- llm-d Info Metrics ---

pkg/epp/metrics/metrics.go

@@ -455,6 +455,8 @@ func Register(customCollectors ...prometheus.Collector) {
 		metrics.Registry.MustRegister(llmdSchedulerAttemptsTotal)
 		metrics.Registry.MustRegister(pluginProcessingLatencies)
 		metrics.Registry.MustRegister(llmdPluginProcessingLatencies)
+		metrics.Registry.MustRegister(llmdRequestProcessingLatency)
+		metrics.Registry.MustRegister(llmdResponseProcessingLatency)
 		metrics.Registry.MustRegister(inferenceExtensionInfo)
 		metrics.Registry.MustRegister(llmdInferenceExtensionInfo)
 		metrics.Registry.MustRegister(flowControlRequestQueueDuration)
@@ -521,6 +523,8 @@ func Reset() {
 	llmdSchedulerAttemptsTotal.Reset()
 	pluginProcessingLatencies.Reset()
 	llmdPluginProcessingLatencies.Reset()
+	llmdRequestProcessingLatency.Reset()
+	llmdResponseProcessingLatency.Reset()
 	inferenceExtensionInfo.Reset()
 	llmdInferenceExtensionInfo.Reset()
 	flowControlRequestQueueDuration.Reset()
@@ -715,6 +719,18 @@ func RecordSchedulerE2ELatency(duration time.Duration) {
 	llmdSchedulerE2ELatency.WithLabelValues().Observe(duration.Seconds())
 }
 
+// RecordRequestProcessingLatency records the EPP request processing latency,
+// measured from request receipt until an endpoint is selected.
+func RecordRequestProcessingLatency(duration time.Duration) {
+	llmdRequestProcessingLatency.WithLabelValues().Observe(duration.Seconds())
+}
+
+// RecordResponseProcessingLatency records the EPP response processing latency,
+// accumulated across response handlers for a single request.
+func RecordResponseProcessingLatency(duration time.Duration) {
+	llmdResponseProcessingLatency.WithLabelValues().Observe(duration.Seconds())
+}
+
 // RecordSchedulerAttempt records a scheduling attempt with status and endpoint information.
 func RecordSchedulerAttempt(err error, targetModelName string, result *fwksched.SchedulingResult) {
 	if err != nil {

pkg/epp/metrics/metrics_test.go

@@ -845,6 +845,60 @@ func TestSchedulerE2ELatency(t *testing.T) {
 	}
 }
 
+func TestRequestProcessingLatency(t *testing.T) {
+	Reset()
+	durations := []time.Duration{
+		200 * time.Microsecond,
+		800 * time.Microsecond,
+		1500 * time.Microsecond,
+		3 * time.Millisecond,
+		8 * time.Millisecond,
+		15 * time.Millisecond,
+		30 * time.Millisecond,
+		75 * time.Millisecond,
+		150 * time.Millisecond,
+	}
+	for _, duration := range durations {
+		RecordRequestProcessingLatency(duration)
+	}
+
+	want, err := os.Open("testdata/llm_d_request_processing_duration_seconds_metric")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer want.Close()
+	if err := promtestutil.GatherAndCompare(metrics.Registry, want, "llm_d_router_epp_request_processing_duration_seconds"); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestResponseProcessingLatency(t *testing.T) {
+	Reset()
+	durations := []time.Duration{
+		200 * time.Microsecond,
+		800 * time.Microsecond,
+		1500 * time.Microsecond,
+		3 * time.Millisecond,
+		8 * time.Millisecond,
+		15 * time.Millisecond,
+		30 * time.Millisecond,
+		75 * time.Millisecond,
+		150 * time.Millisecond,
+	}
+	for _, duration := range durations {
+		RecordResponseProcessingLatency(duration)
+	}
+
+	want, err := os.Open("testdata/llm_d_response_processing_duration_seconds_metric")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer want.Close()
+	if err := promtestutil.GatherAndCompare(metrics.Registry, want, "llm_d_router_epp_response_processing_duration_seconds"); err != nil {
+		t.Error(err)
+	}
+}
+
 func TestFlowControlDispatchCycleLengthMetric(t *testing.T) {
 	Reset()
 	scenarios := []struct {

pkg/epp/metrics/testdata/llm_d_request_processing_duration_seconds_metric

@@ -0,0 +1,15 @@
+# HELP llm_d_router_epp_request_processing_duration_seconds [ALPHA] EPP request processing latency distribution in seconds, from request receipt until an endpoint is selected, excluding flow-control admission wait.
+# TYPE llm_d_router_epp_request_processing_duration_seconds histogram
+llm_d_router_epp_request_processing_duration_seconds_bucket{le="0.0001"} 0
+llm_d_router_epp_request_processing_duration_seconds_bucket{le="0.0002"} 1
+llm_d_router_epp_request_processing_duration_seconds_bucket{le="0.0005"} 1
+llm_d_router_epp_request_processing_duration_seconds_bucket{le="0.001"} 2
+llm_d_router_epp_request_processing_duration_seconds_bucket{le="0.002"} 3
+llm_d_router_epp_request_processing_duration_seconds_bucket{le="0.005"} 4
+llm_d_router_epp_request_processing_duration_seconds_bucket{le="0.01"} 5
+llm_d_router_epp_request_processing_duration_seconds_bucket{le="0.02"} 6
+llm_d_router_epp_request_processing_duration_seconds_bucket{le="0.05"} 7
+llm_d_router_epp_request_processing_duration_seconds_bucket{le="0.1"} 8
+llm_d_router_epp_request_processing_duration_seconds_bucket{le="+Inf"} 9
+llm_d_router_epp_request_processing_duration_seconds_sum 0.2835
+llm_d_router_epp_request_processing_duration_seconds_count 9

pkg/epp/metrics/testdata/llm_d_response_processing_duration_seconds_metric

@@ -0,0 +1,15 @@
+# HELP llm_d_router_epp_response_processing_duration_seconds [ALPHA] EPP response processing latency distribution in seconds, accumulated across response handlers and excluding model server generation time.
+# TYPE llm_d_router_epp_response_processing_duration_seconds histogram
+llm_d_router_epp_response_processing_duration_seconds_bucket{le="0.0001"} 0
+llm_d_router_epp_response_processing_duration_seconds_bucket{le="0.0002"} 1
+llm_d_router_epp_response_processing_duration_seconds_bucket{le="0.0005"} 1
+llm_d_router_epp_response_processing_duration_seconds_bucket{le="0.001"} 2
+llm_d_router_epp_response_processing_duration_seconds_bucket{le="0.002"} 3
+llm_d_router_epp_response_processing_duration_seconds_bucket{le="0.005"} 4
+llm_d_router_epp_response_processing_duration_seconds_bucket{le="0.01"} 5
+llm_d_router_epp_response_processing_duration_seconds_bucket{le="0.02"} 6
+llm_d_router_epp_response_processing_duration_seconds_bucket{le="0.05"} 7
+llm_d_router_epp_response_processing_duration_seconds_bucket{le="0.1"} 8
+llm_d_router_epp_response_processing_duration_seconds_bucket{le="+Inf"} 9
+llm_d_router_epp_response_processing_duration_seconds_sum 0.2835
+llm_d_router_epp_response_processing_duration_seconds_count 9

pkg/epp/requestcontrol/director.go

@@ -232,6 +232,13 @@ func (d *Director) getInferenceObjective(ctx context.Context, reqCtx *handlers.R
 // HandleRequest orchestrates the request lifecycle.
 // It always returns the requestContext even in the error case, as the request context is used in error handling.
 func (d *Director) HandleRequest(ctx context.Context, reqCtx *handlers.RequestContext, inferenceRequestBody *fwkrh.InferenceRequestBody) (_ *handlers.RequestContext, err error) {
+	start := time.Now()
+	var admissionWait time.Duration
+	// request_processing_duration captures EPP's own processing cost, so the
+	// flow-control admission wait (tracked separately by
+	// flow_control_request_queue_duration_seconds) is subtracted out.
+	defer func() { metrics.RecordRequestProcessingLatency(time.Since(start) - admissionWait) }()
+
 	tracer := tracing.Tracer("llm-d-router/pkg/epp/requestcontrol")
 	ctx, span := tracer.Start(ctx, "gateway.request_orchestration", trace.WithSpanKind(trace.SpanKindServer))
 	defer func() {
@@ -284,8 +291,11 @@ func (d *Director) HandleRequest(ctx context.Context, reqCtx *handlers.RequestCo
 	}
 
 	// Admit may block until flow control admits the request.
-	if err := d.admissionController.Admit(ctx, reqCtx, priority); err != nil {
-		return reqCtx, err
+	admitStart := time.Now()
+	admitErr := d.admissionController.Admit(ctx, reqCtx, priority)
+	admissionWait = time.Since(admitStart)
+	if admitErr != nil {
+		return reqCtx, admitErr
 	}
 
 	endpointCandidates := d.endpointCandidates.Locate(ctx, reqCtx.Request.Metadata)