address review feedback on processing latency metrics

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

Author: satyamg1620

pkg/epp/handlers/server.go

@@ -269,6 +269,17 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 		},
 	}
 
+	// Record EPP response processing latency once when the stream ends. Using a
+	// defer (rather than emitting on end-of-stream) ensures aborted streams
+	// (client cancel or upstream error before EOS) are also recorded, so the
+	// metric is not biased toward fully streamed responses. The guard skips
+	// requests that never reached the response phase.
+	defer func() {
+		if reqCtx.responseProcessingDuration > 0 {
+			metrics.RecordResponseProcessingLatency(reqCtx.responseProcessingDuration)
+		}
+	}()
+
 	buf := s.bufferPool.Get().(*bytes.Buffer)
 	buf.Reset()
 	defer func() {
@@ -501,9 +512,6 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 				// 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 {
@@ -575,7 +583,6 @@ func (s *StreamingServer) finishResponse(ctx context.Context, reqCtx *RequestCon
 		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

@@ -257,7 +257,7 @@ var (
 		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),
+			Help:      metricsutil.HelpMsgWithStability("EPP request orchestration latency distribution in seconds, from request receipt through endpoint selection and request preparation, excluding admission-control 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,
 			},

pkg/epp/metrics/testdata/llm_d_request_processing_duration_seconds_metric

@@ -1,4 +1,4 @@
-# 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.
+# HELP llm_d_router_epp_request_processing_duration_seconds [ALPHA] EPP request orchestration latency distribution in seconds, from request receipt through endpoint selection and request preparation, excluding admission-control time.
 # 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

pkg/epp/requestcontrol/director.go

@@ -234,9 +234,11 @@ func (d *Director) getInferenceObjective(ctx context.Context, reqCtx *handlers.R
 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.
+	// request_processing_duration measures EPP's own orchestration cost for a
+	// request (receipt through endpoint selection and request preparation). Time
+	// spent in admission control is subtracted: under flow control it is
+	// dominated by the queue wait, which is load- rather than EPP-driven and is
+	// tracked separately by flow_control_request_queue_duration_seconds.
 	defer func() { metrics.RecordRequestProcessingLatency(time.Since(start) - admissionWait) }()
 
 	tracer := tracing.Tracer("llm-d-router/pkg/epp/requestcontrol")