Measure EPP processing latency at the handler boundary

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

Author: satyamg1620

docs/metrics.md

@@ -112,12 +112,12 @@ Label `{name}` (the pool name).
 
 ### EPP processing overhead
 
-Unlabelled. Together these cover the EPP's own cost on the request and response paths.
+Unlabelled.
 
 | Name | Type | Notes |
 |---|---|---|
-| `request_processing_duration_seconds` | Histogram | Request orchestration latency, from receipt through endpoint selection and request preparation; excludes admission control. |
-| `response_processing_duration_seconds` | Histogram | Response handling latency, accumulated across the response handlers; excludes model-server generation time. |
+| `request_processing_duration_seconds` | Histogram | Time from request receipt until the request body has been handled. Includes admission control, so under the flow control feature gate this covers queue wait; `flow_control_request_queue_duration_seconds` separates it out. |
+| `response_processing_duration_seconds` | Histogram | Sum of the per-chunk handler slices for a streamed response, so model-server generation time between chunks is excluded. For a non-streaming response, the interval from response headers to completion. |
 
 ### Plugin, info, and model rewrite
 

pkg/epp/handlers/server.go

@@ -136,9 +136,13 @@ 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 is the EPP cost of handling the response. For a
+	// streamed response it is the sum of the per-chunk handler slices, since the
+	// gaps between chunks are model server generation time. For a non-streaming
+	// response it is the single interval from responseHeadersReceivedAt onward,
+	// during which the response is entirely in EPP's hands.
 	responseProcessingDuration time.Duration
+	responseHeadersReceivedAt  time.Time
 
 	Response *Response
 
@@ -269,6 +273,14 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 		},
 	}
 
+	// Request-phase failures (parser resolution, body parsing, admission
+	// rejection) leave the switch before the success path, so both call this.
+	// Flow-control rejections carry the queue wait and are the slowest samples;
+	// dropping them would bias the histogram low.
+	recordRequestProcessing := sync.OnceFunc(func() {
+		metrics.RecordRequestProcessingLatency(time.Since(reqCtx.RequestReceivedTimestamp))
+	})
+
 	// 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
@@ -476,11 +488,13 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 				if parseResult.SkipResponseProcessing {
 					reqCtx.RequestState = RequestResponseProcessingSkipped
 				}
+
+				recordRequestProcessing()
 			}
 		case *extProcPb.ProcessingRequest_RequestTrailers:
 			// This is currently unused.
 		case *extProcPb.ProcessingRequest_ResponseHeaders:
-			respHeaderStart := time.Now()
+			respHeadersReceivedAt := time.Now()
 			for _, header := range v.ResponseHeaders.Headers.GetHeaders() {
 				value := string(header.RawValue)
 				loggerTrace.Info("header", "key", header.Key, "value", value)
@@ -494,7 +508,8 @@ 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)
+			reqCtx.responseHeadersReceivedAt = respHeadersReceivedAt
+			reqCtx.responseProcessingDuration += time.Since(respHeadersReceivedAt)
 
 		case *extProcPb.ProcessingRequest_ResponseBody:
 			endOfStream := v.ResponseBody.EndOfStream
@@ -532,6 +547,7 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 
 		// Handle the err and fire an immediate response.
 		if err != nil {
+			recordRequestProcessing()
 			if logger.V(logutil.DEBUG).Enabled() {
 				logger.V(logutil.DEBUG).Error(err, "Failed to process request", "request", req)
 			} else {
@@ -582,7 +598,13 @@ 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)
+	if modelStreaming || reqCtx.responseHeadersReceivedAt.IsZero() {
+		reqCtx.responseProcessingDuration += time.Since(start)
+	} else {
+		// Supersedes the header slice already accumulated: the interval since the
+		// response headers arrived covers it and the body wait in between.
+		reqCtx.responseProcessingDuration = time.Since(reqCtx.responseHeadersReceivedAt)
+	}
 }
 
 // rewriteModelName replaces occurrences of the target (internal) model name with the

pkg/epp/metrics/llm_d_router_metrics.go

@@ -284,9 +284,9 @@ var (
 		prometheus.HistogramOpts{
 			Subsystem: LLMDRouterEndpointPickerSubsystem,
 			Name:      "request_processing_duration_seconds",
-			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),
+			Help:      metricsutil.HelpMsgWithStability("EPP request processing latency distribution in seconds, from request receipt until the request body has been handled, including admission control.", compbasemetrics.ALPHA),
 			Buckets: []float64{
-				0.0001, 0.0002, 0.0005, 0.001, 0.002, 0.005, 0.01, 0.02, 0.05, 0.1,
+				0.0005, 0.001, 0.002, 0.005, 0.01, 0.015, 0.025, 0.04, 0.06, 0.1, 0.25, 0.5, 1, 2.5, 5, 10,
 			},
 		},
 		[]string{},
@@ -296,9 +296,9 @@ var (
 		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),
+			Help:      metricsutil.HelpMsgWithStability("EPP response processing latency distribution in seconds: the sum of per-chunk handler time for a streamed response, or the interval from response headers to completion for a non-streaming response.", compbasemetrics.ALPHA),
 			Buckets: []float64{
-				0.0001, 0.0002, 0.0005, 0.001, 0.002, 0.005, 0.01, 0.02, 0.05, 0.1,
+				0.0001, 0.00025, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5,
 			},
 		},
 		[]string{},

pkg/epp/metrics/metrics.go

@@ -770,14 +770,13 @@ func RecordSchedulerE2ELatency(duration time.Duration) {
 }
 
 // RecordRequestProcessingLatency records the EPP request processing latency,
-// measured from request receipt through endpoint selection and request
-// preparation, excluding admission-control time.
+// measured from request receipt until the request body has been handled.
 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.
+// RecordResponseProcessingLatency records the EPP response processing latency
+// for a single request.
 func RecordResponseProcessingLatency(duration time.Duration) {
 	llmdResponseProcessingLatency.WithLabelValues().Observe(duration.Seconds())
 }

pkg/epp/metrics/testdata/llm_d_request_processing_duration_seconds_metric

@@ -1,15 +1,21 @@
-# HELP llm_d_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.
+# HELP llm_d_epp_request_processing_duration_seconds [ALPHA] EPP request processing latency distribution in seconds, from request receipt until the request body has been handled, including admission control.
 # TYPE llm_d_epp_request_processing_duration_seconds histogram
-llm_d_epp_request_processing_duration_seconds_bucket{le="0.0001"} 0
-llm_d_epp_request_processing_duration_seconds_bucket{le="0.0002"} 1
 llm_d_epp_request_processing_duration_seconds_bucket{le="0.0005"} 1
 llm_d_epp_request_processing_duration_seconds_bucket{le="0.001"} 2
 llm_d_epp_request_processing_duration_seconds_bucket{le="0.002"} 3
 llm_d_epp_request_processing_duration_seconds_bucket{le="0.005"} 4
 llm_d_epp_request_processing_duration_seconds_bucket{le="0.01"} 5
-llm_d_epp_request_processing_duration_seconds_bucket{le="0.02"} 6
-llm_d_epp_request_processing_duration_seconds_bucket{le="0.05"} 7
+llm_d_epp_request_processing_duration_seconds_bucket{le="0.015"} 6
+llm_d_epp_request_processing_duration_seconds_bucket{le="0.025"} 6
+llm_d_epp_request_processing_duration_seconds_bucket{le="0.04"} 7
+llm_d_epp_request_processing_duration_seconds_bucket{le="0.06"} 7
 llm_d_epp_request_processing_duration_seconds_bucket{le="0.1"} 8
+llm_d_epp_request_processing_duration_seconds_bucket{le="0.25"} 9
+llm_d_epp_request_processing_duration_seconds_bucket{le="0.5"} 9
+llm_d_epp_request_processing_duration_seconds_bucket{le="1"} 9
+llm_d_epp_request_processing_duration_seconds_bucket{le="2.5"} 9
+llm_d_epp_request_processing_duration_seconds_bucket{le="5"} 9
+llm_d_epp_request_processing_duration_seconds_bucket{le="10"} 9
 llm_d_epp_request_processing_duration_seconds_bucket{le="+Inf"} 9
 llm_d_epp_request_processing_duration_seconds_sum 0.2835
 llm_d_epp_request_processing_duration_seconds_count 9

pkg/epp/metrics/testdata/llm_d_response_processing_duration_seconds_metric

@@ -1,15 +1,20 @@
-# HELP llm_d_epp_response_processing_duration_seconds [ALPHA] EPP response processing latency distribution in seconds, accumulated across response handlers and excluding model server generation time.
+# HELP llm_d_epp_response_processing_duration_seconds [ALPHA] EPP response processing latency distribution in seconds: the sum of per-chunk handler time for a streamed response, or the interval from response headers to completion for a non-streaming response.
 # TYPE llm_d_epp_response_processing_duration_seconds histogram
 llm_d_epp_response_processing_duration_seconds_bucket{le="0.0001"} 0
-llm_d_epp_response_processing_duration_seconds_bucket{le="0.0002"} 1
+llm_d_epp_response_processing_duration_seconds_bucket{le="0.00025"} 1
 llm_d_epp_response_processing_duration_seconds_bucket{le="0.0005"} 1
 llm_d_epp_response_processing_duration_seconds_bucket{le="0.001"} 2
-llm_d_epp_response_processing_duration_seconds_bucket{le="0.002"} 3
+llm_d_epp_response_processing_duration_seconds_bucket{le="0.0025"} 3
 llm_d_epp_response_processing_duration_seconds_bucket{le="0.005"} 4
 llm_d_epp_response_processing_duration_seconds_bucket{le="0.01"} 5
-llm_d_epp_response_processing_duration_seconds_bucket{le="0.02"} 6
+llm_d_epp_response_processing_duration_seconds_bucket{le="0.025"} 6
 llm_d_epp_response_processing_duration_seconds_bucket{le="0.05"} 7
 llm_d_epp_response_processing_duration_seconds_bucket{le="0.1"} 8
+llm_d_epp_response_processing_duration_seconds_bucket{le="0.25"} 9
+llm_d_epp_response_processing_duration_seconds_bucket{le="0.5"} 9
+llm_d_epp_response_processing_duration_seconds_bucket{le="1"} 9
+llm_d_epp_response_processing_duration_seconds_bucket{le="2.5"} 9
+llm_d_epp_response_processing_duration_seconds_bucket{le="5"} 9
 llm_d_epp_response_processing_duration_seconds_bucket{le="+Inf"} 9
 llm_d_epp_response_processing_duration_seconds_sum 0.2835
 llm_d_epp_response_processing_duration_seconds_count 9

pkg/epp/requestcontrol/director.go

@@ -232,15 +232,6 @@ 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 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")
 	ctx, span := tracer.Start(ctx, "gateway.request_orchestration", trace.WithSpanKind(trace.SpanKindServer))
 	defer func() {
@@ -301,11 +292,8 @@ func (d *Director) HandleRequest(ctx context.Context, reqCtx *handlers.RequestCo
 	}
 
 	// Admit may block until flow control admits the request.
-	admitStart := time.Now()
-	admitErr := d.admissionController.Admit(ctx, reqCtx, priority)
-	admissionWait = time.Since(admitStart)
-	if admitErr != nil {
-		return reqCtx, admitErr
+	if err := d.admissionController.Admit(ctx, reqCtx, priority); err != nil {
+		return reqCtx, err
 	}
 
 	endpointCandidates := d.endpointCandidates.Locate(ctx, reqCtx.Request.Metadata)