diff --git a/docs/metrics.md b/docs/metrics.md index f3091a5a69..bee8be0657 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. + +| 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 | diff --git a/pkg/epp/handlers/server.go b/pkg/epp/handlers/server.go index c7d82054e3..89849c6ea8 100644 --- a/pkg/epp/handlers/server.go +++ b/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 diff --git a/pkg/epp/metrics/llm_d_router_metrics.go b/pkg/epp/metrics/llm_d_router_metrics.go index 26e9d06f32..e65ba27b30 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 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 --- diff --git a/pkg/epp/metrics/metrics.go b/pkg/epp/metrics/metrics.go index 8011acd5ff..ca5ae09a05 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 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 { diff --git a/pkg/epp/metrics/metrics_test.go b/pkg/epp/metrics/metrics_test.go index ca0c00a685..3589fc9b09 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_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 { 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..1053aa0d57 --- /dev/null +++ b/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 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..e5bfbdeecf --- /dev/null +++ b/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 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...)