From dd88771c5656664c24db64fed2e95b0a6a18d7b6 Mon Sep 17 00:00:00 2001 From: satyamg1620 Date: Mon, 15 Jun 2026 17:49:31 +0530 Subject: [PATCH 1/5] add EPP request/response processing latency metrics Signed-off-by: satyamg1620 --- pkg/epp/handlers/server.go | 14 +++++ pkg/epp/metrics/llm_d_router_metrics.go | 24 +++++++++ pkg/epp/metrics/metrics.go | 16 ++++++ pkg/epp/metrics/metrics_test.go | 54 +++++++++++++++++++ ...request_processing_duration_seconds_metric | 15 ++++++ ...esponse_processing_duration_seconds_metric | 15 ++++++ pkg/epp/requestcontrol/director.go | 14 ++++- 7 files changed, 150 insertions(+), 2 deletions(-) create mode 100644 pkg/epp/metrics/testdata/llm_d_request_processing_duration_seconds_metric create mode 100644 pkg/epp/metrics/testdata/llm_d_response_processing_duration_seconds_metric diff --git a/pkg/epp/handlers/server.go b/pkg/epp/handlers/server.go index c7d82054e3..1660e13333 100644 --- a/pkg/epp/handlers/server.go +++ b/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 diff --git a/pkg/epp/metrics/llm_d_router_metrics.go b/pkg/epp/metrics/llm_d_router_metrics.go index 26e9d06f32..a2a01e6129 100644 --- a/pkg/epp/metrics/llm_d_router_metrics.go +++ b/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 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 --- diff --git a/pkg/epp/metrics/metrics.go b/pkg/epp/metrics/metrics.go index 8011acd5ff..05e37e0d71 100644 --- a/pkg/epp/metrics/metrics.go +++ b/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 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 { diff --git a/pkg/epp/metrics/metrics_test.go b/pkg/epp/metrics/metrics_test.go index ca0c00a685..136953e0f0 100644 --- a/pkg/epp/metrics/metrics_test.go +++ b/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_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 { diff --git a/pkg/epp/metrics/testdata/llm_d_request_processing_duration_seconds_metric b/pkg/epp/metrics/testdata/llm_d_request_processing_duration_seconds_metric new file mode 100644 index 0000000000..5c658076ab --- /dev/null +++ b/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 diff --git a/pkg/epp/metrics/testdata/llm_d_response_processing_duration_seconds_metric b/pkg/epp/metrics/testdata/llm_d_response_processing_duration_seconds_metric new file mode 100644 index 0000000000..937789868d --- /dev/null +++ b/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 diff --git a/pkg/epp/requestcontrol/director.go b/pkg/epp/requestcontrol/director.go index 4e33e980f0..e677493190 100644 --- a/pkg/epp/requestcontrol/director.go +++ b/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() { @@ -292,8 +299,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) From 6359ce1c824a4ce919c0abd8134c7491236ba1fa Mon Sep 17 00:00:00 2001 From: satyamg1620 Date: Mon, 15 Jun 2026 20:31:11 +0530 Subject: [PATCH 2/5] address review feedback on processing latency metrics Signed-off-by: satyamg1620 --- pkg/epp/handlers/server.go | 15 +++++++++++---- pkg/epp/metrics/llm_d_router_metrics.go | 2 +- ...m_d_request_processing_duration_seconds_metric | 2 +- pkg/epp/requestcontrol/director.go | 8 +++++--- 4 files changed, 18 insertions(+), 9 deletions(-) diff --git a/pkg/epp/handlers/server.go b/pkg/epp/handlers/server.go index 1660e13333..369c629f3b 100644 --- a/pkg/epp/handlers/server.go +++ b/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 diff --git a/pkg/epp/metrics/llm_d_router_metrics.go b/pkg/epp/metrics/llm_d_router_metrics.go index a2a01e6129..aeb8d0b277 100644 --- a/pkg/epp/metrics/llm_d_router_metrics.go +++ b/pkg/epp/metrics/llm_d_router_metrics.go @@ -284,7 +284,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, }, diff --git a/pkg/epp/metrics/testdata/llm_d_request_processing_duration_seconds_metric b/pkg/epp/metrics/testdata/llm_d_request_processing_duration_seconds_metric index 5c658076ab..f7057fa5aa 100644 --- a/pkg/epp/metrics/testdata/llm_d_request_processing_duration_seconds_metric +++ b/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 diff --git a/pkg/epp/requestcontrol/director.go b/pkg/epp/requestcontrol/director.go index e677493190..ae6f3266d2 100644 --- a/pkg/epp/requestcontrol/director.go +++ b/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") From e5cbb5c27c3b43e1a2ac66bd7e5606d176934b1f Mon Sep 17 00:00:00 2001 From: satyamg1620 Date: Mon, 20 Jul 2026 18:14:17 +0530 Subject: [PATCH 3/5] fix subsystem prefix in processing latency metric tests Signed-off-by: satyamg1620 --- docs/metrics.md | 9 ++++++ pkg/epp/metrics/metrics.go | 3 +- pkg/epp/metrics/metrics_test.go | 4 +-- ...request_processing_duration_seconds_metric | 30 +++++++++---------- ...esponse_processing_duration_seconds_metric | 30 +++++++++---------- 5 files changed, 43 insertions(+), 33 deletions(-) diff --git a/docs/metrics.md b/docs/metrics.md index f3091a5a69..fd61798555 100644 --- a/docs/metrics.md +++ b/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. Together these cover the EPP's own cost on the request and response paths. + +| 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. | + ### Plugin, info, and model rewrite | Name | Type | Notes | diff --git a/pkg/epp/metrics/metrics.go b/pkg/epp/metrics/metrics.go index 05e37e0d71..d899ff5dc7 100644 --- a/pkg/epp/metrics/metrics.go +++ b/pkg/epp/metrics/metrics.go @@ -770,7 +770,8 @@ func RecordSchedulerE2ELatency(duration time.Duration) { } // RecordRequestProcessingLatency records the EPP request processing latency, -// measured from request receipt until an endpoint is selected. +// measured from request receipt through endpoint selection and request +// preparation, excluding admission-control time. func RecordRequestProcessingLatency(duration time.Duration) { llmdRequestProcessingLatency.WithLabelValues().Observe(duration.Seconds()) } diff --git a/pkg/epp/metrics/metrics_test.go b/pkg/epp/metrics/metrics_test.go index 136953e0f0..3589fc9b09 100644 --- a/pkg/epp/metrics/metrics_test.go +++ b/pkg/epp/metrics/metrics_test.go @@ -903,7 +903,7 @@ func TestRequestProcessingLatency(t *testing.T) { t.Fatal(err) } defer want.Close() - if err := promtestutil.GatherAndCompare(metrics.Registry, want, "llm_d_router_epp_request_processing_duration_seconds"); err != nil { + if err := promtestutil.GatherAndCompare(metrics.Registry, want, "llm_d_epp_request_processing_duration_seconds"); err != nil { t.Error(err) } } @@ -930,7 +930,7 @@ func TestResponseProcessingLatency(t *testing.T) { t.Fatal(err) } defer want.Close() - if err := promtestutil.GatherAndCompare(metrics.Registry, want, "llm_d_router_epp_response_processing_duration_seconds"); err != nil { + if err := promtestutil.GatherAndCompare(metrics.Registry, want, "llm_d_epp_response_processing_duration_seconds"); err != nil { t.Error(err) } } diff --git a/pkg/epp/metrics/testdata/llm_d_request_processing_duration_seconds_metric b/pkg/epp/metrics/testdata/llm_d_request_processing_duration_seconds_metric index f7057fa5aa..30408d03ee 100644 --- a/pkg/epp/metrics/testdata/llm_d_request_processing_duration_seconds_metric +++ b/pkg/epp/metrics/testdata/llm_d_request_processing_duration_seconds_metric @@ -1,15 +1,15 @@ -# 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 -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 +# 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. +# 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.1"} 8 +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 diff --git a/pkg/epp/metrics/testdata/llm_d_response_processing_duration_seconds_metric b/pkg/epp/metrics/testdata/llm_d_response_processing_duration_seconds_metric index 937789868d..c2f9ca8281 100644 --- a/pkg/epp/metrics/testdata/llm_d_response_processing_duration_seconds_metric +++ b/pkg/epp/metrics/testdata/llm_d_response_processing_duration_seconds_metric @@ -1,15 +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 +# 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. +# 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.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.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.05"} 7 +llm_d_epp_response_processing_duration_seconds_bucket{le="0.1"} 8 +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 From 59a5cbf4d1b277bdd2fd05e86395d27fead5301b Mon Sep 17 00:00:00 2001 From: satyamg1620 Date: Wed, 22 Jul 2026 04:15:15 +0530 Subject: [PATCH 4/5] Measure EPP processing latency at the handler boundary Signed-off-by: satyamg1620 --- docs/metrics.md | 6 ++-- pkg/epp/handlers/server.go | 32 ++++++++++++++++--- pkg/epp/metrics/llm_d_router_metrics.go | 8 ++--- pkg/epp/metrics/metrics.go | 7 ++-- ...request_processing_duration_seconds_metric | 16 +++++++--- ...esponse_processing_duration_seconds_metric | 13 +++++--- pkg/epp/requestcontrol/director.go | 16 ++-------- 7 files changed, 59 insertions(+), 39 deletions(-) diff --git a/docs/metrics.md b/docs/metrics.md index fd61798555..bee8be0657 100644 --- a/docs/metrics.md +++ b/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 diff --git a/pkg/epp/handlers/server.go b/pkg/epp/handlers/server.go index 369c629f3b..89849c6ea8 100644 --- a/pkg/epp/handlers/server.go +++ b/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 diff --git a/pkg/epp/metrics/llm_d_router_metrics.go b/pkg/epp/metrics/llm_d_router_metrics.go index aeb8d0b277..e65ba27b30 100644 --- a/pkg/epp/metrics/llm_d_router_metrics.go +++ b/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{}, diff --git a/pkg/epp/metrics/metrics.go b/pkg/epp/metrics/metrics.go index d899ff5dc7..ca5ae09a05 100644 --- a/pkg/epp/metrics/metrics.go +++ b/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()) } diff --git a/pkg/epp/metrics/testdata/llm_d_request_processing_duration_seconds_metric b/pkg/epp/metrics/testdata/llm_d_request_processing_duration_seconds_metric index 30408d03ee..1053aa0d57 100644 --- a/pkg/epp/metrics/testdata/llm_d_request_processing_duration_seconds_metric +++ b/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 diff --git a/pkg/epp/metrics/testdata/llm_d_response_processing_duration_seconds_metric b/pkg/epp/metrics/testdata/llm_d_response_processing_duration_seconds_metric index c2f9ca8281..e5bfbdeecf 100644 --- a/pkg/epp/metrics/testdata/llm_d_response_processing_duration_seconds_metric +++ b/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 diff --git a/pkg/epp/requestcontrol/director.go b/pkg/epp/requestcontrol/director.go index ae6f3266d2..4e33e980f0 100644 --- a/pkg/epp/requestcontrol/director.go +++ b/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) From 93a9bc1767130bdcb071f3b32c9690e1592fdf86 Mon Sep 17 00:00:00 2001 From: satyamg1620 Date: Wed, 22 Jul 2026 04:49:03 +0530 Subject: [PATCH 5/5] Assert EPP processing latency metrics in e2e preset Signed-off-by: satyamg1620 --- test/e2e/requests_test.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/test/e2e/requests_test.go b/test/e2e/requests_test.go index e7af53b926..ce1566796d 100644 --- a/test/e2e/requests_test.go +++ b/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...)