Add EPP request/response processing latency metrics (#1658)

* add EPP request/response processing latency metrics

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

* address review feedback on processing latency metrics

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

* fix subsystem prefix in processing latency metric tests

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

* Measure EPP processing latency at the handler boundary

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

* Assert EPP processing latency metrics in e2e preset

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

---------

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

Author: satyamg1620

docs/metrics.md

@@ -110,6 +110,15 @@ Label `{name}` (the pool name).
 | `scheduler_e2e_duration_seconds` | Histogram | End-to-end scheduling latency. |
 | `scheduler_attempts_total` | Counter | Scheduling attempts; labels `{status, target_model_name, endpoint_name, namespace, port}`. |
 
+### EPP processing overhead
+
+Unlabelled.
+
+| Name | Type | Notes |
+|---|---|---|
+| `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
 
 | Name | Type | Notes |

pkg/epp/handlers/server.go

@@ -136,6 +136,14 @@ type RequestContext struct {
 	RequestDroppedReason errcommon.RequestDroppedReason
 	modelServerStreaming bool
 
+	// 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
 
 	reqHeaderResp  *extProcPb.ProcessingResponse
@@ -265,6 +273,25 @@ 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
+	// 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() {
@@ -461,10 +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:
+			respHeadersReceivedAt := time.Now()
 			for _, header := range v.ResponseHeaders.Headers.GetHeaders() {
 				value := string(header.RawValue)
 				loggerTrace.Info("header", "key", header.Key, "value", value)
@@ -478,12 +508,15 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 			reqCtx.RequestState = ResponseReceived
 			reqCtx = s.HandleResponseHeaders(ctx, reqCtx, v)
 			reqCtx.respHeaderResp = s.generateResponseHeaderResponse(reqCtx)
+			reqCtx.responseHeadersReceivedAt = respHeadersReceivedAt
+			reqCtx.responseProcessingDuration += time.Since(respHeadersReceivedAt)
 
 		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 +526,7 @@ 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)
 			} else {
 				respBody = append(respBody, chunk...)
 				if endOfStream {
@@ -513,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 {
@@ -553,6 +588,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 +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)
 	}
+	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

@@ -279,6 +279,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 the request body has been handled, including admission control.", compbasemetrics.ALPHA),
+			Buckets: []float64{
+				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{},
+	)
+
+	llmdResponseProcessingLatency = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Subsystem: LLMDRouterEndpointPickerSubsystem,
+			Name:      "response_processing_duration_seconds",
+			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.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{},
+	)
 )
 
 // --- llm-d Info Metrics ---

pkg/epp/metrics/metrics.go

@@ -458,6 +458,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)
@@ -528,6 +530,8 @@ func Reset() {
 	llmdSchedulerAttemptsTotal.Reset()
 	pluginProcessingLatencies.Reset()
 	llmdPluginProcessingLatencies.Reset()
+	llmdRequestProcessingLatency.Reset()
+	llmdResponseProcessingLatency.Reset()
 	inferenceExtensionInfo.Reset()
 	llmdInferenceExtensionInfo.Reset()
 	flowControlRequestQueueDuration.Reset()
@@ -765,6 +769,18 @@ func RecordSchedulerE2ELatency(duration time.Duration) {
 	llmdSchedulerE2ELatency.WithLabelValues().Observe(duration.Seconds())
 }
 
+// RecordRequestProcessingLatency records the EPP request processing latency,
+// 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
+// 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

@@ -881,6 +881,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_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_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,21 @@
+# 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.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.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

@@ -0,0 +1,20 @@
+# 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.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.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.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

test/e2e/requests_test.go

@@ -344,6 +344,8 @@ func verifyMetrics(infPoolName string, numTargetPorts int) {
 		"llm_d_epp_request_running",
 		"llm_d_epp_ready_endpoints",
 		"llm_d_epp_info",
+		"llm_d_epp_request_processing_duration_seconds",
+		"llm_d_epp_response_processing_duration_seconds",
 	}
 	expectedMetrics := make([]string, 0, len(preset)+len(decodePods)*numTargetPorts*2)
 	expectedMetrics = append(expectedMetrics, preset...)