Merge branch 'main' into fix/fc-saturation-staleness

Author: LukeAVanDrie

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/flowcontrol/controller/controller.go

@@ -378,8 +378,15 @@ func (fc *FlowController) tryDistribution(
 	return item, finalErr
 }
 
+func finalizeOnControllerShutdown(item *internal.FlowItem) (types.QueueOutcome, error) {
+	item.Finalize(types.ErrFlowControllerNotRunning)
+
+	finalState := item.FinalState()
+	return finalState.Outcome, finalState.Err
+}
+
 // awaitFinalization blocks until an item is finalized, either by the processor (synchronously) or by the controller
-// itself due to context expiry (asynchronously).
+// itself due to context expiry or shutdown (asynchronously).
 func (fc *FlowController) awaitFinalization(
 	reqCtx context.Context,
 	item *internal.FlowItem,
@@ -388,13 +395,20 @@ func (fc *FlowController) awaitFinalization(
 	case <-reqCtx.Done():
 		// Asynchronous Finalization (Controller-initiated):
 		// The request Context expired (Cancellation/TTL) while the item was being processed.
+		if fc.parentCtx.Err() != nil {
+			return finalizeOnControllerShutdown(item)
+		}
+
 		cause := context.Cause(reqCtx)
 		item.Finalize(cause)
 
 		// The processor will eventually discard this "zombie" item during its cleanup sweep.
 		finalState := item.FinalState()
 		return finalState.Outcome, finalState.Err
 
+	case <-fc.parentCtx.Done():
+		return finalizeOnControllerShutdown(item)
+
 	case finalState := <-item.Done():
 		// Synchronous Finalization (Processor-initiated):
 		// The processor finalized the item (Dispatch, Reject, Shutdown).

pkg/epp/flowcontrol/controller/controller_test.go

@@ -367,6 +367,71 @@ func TestFlowController_EnqueueAndWait(t *testing.T) {
 			assert.Equal(t, types.QueueOutcomeRejectedOther, outcome,
 				"outcome should be QueueOutcomeRejectedOther on shutdown")
 		})
+		t.Run("OnControllerShutdownDuringFinalization", func(t *testing.T) {
+			t.Parallel()
+			ctx, cancel := context.WithCancel(t.Context())
+			h := newUnitHarness(ctx, t, &Config{}, nil, nil)
+			item := internal.NewItem(newTestRequest(defaultFlowKey), 0, time.Now())
+
+			result := make(chan struct {
+				outcome types.QueueOutcome
+				err     error
+			}, 1)
+			go func() {
+				outcome, err := h.fc.awaitFinalization(context.Background(), item)
+				result <- struct {
+					outcome types.QueueOutcome
+					err     error
+				}{outcome: outcome, err: err}
+			}()
+
+			cancel()
+			select {
+			case r := <-result:
+				require.Error(t, r.err, "awaitFinalization must fail when controller shuts down")
+				assert.ErrorIs(t, r.err, types.ErrFlowControllerNotRunning,
+					"error should wrap ErrFlowControllerNotRunning")
+				assert.Equal(t, types.QueueOutcomeRejectedOther, r.outcome,
+					"outcome should be QueueOutcomeRejectedOther on shutdown")
+			case <-time.After(time.Second):
+				t.Fatal("awaitFinalization did not return after controller shutdown")
+			}
+		})
+		t.Run("OnControllerShutdownTakesPrecedenceOverRequestCancellation", func(t *testing.T) {
+			t.Parallel()
+			ctx, cancel := context.WithCancel(t.Context())
+			h := newUnitHarness(ctx, t, &Config{}, nil, nil)
+			reqCtx, reqCancel := context.WithCancel(context.Background())
+			item := internal.NewItem(newTestRequest(defaultFlowKey), 0, time.Now())
+
+			reqCancel()
+			cancel()
+
+			outcome, err := h.fc.awaitFinalization(reqCtx, item)
+			require.Error(t, err, "awaitFinalization must fail when controller shuts down")
+			assert.ErrorIs(t, err, types.ErrFlowControllerNotRunning,
+				"controller shutdown should take precedence over request cancellation")
+			assert.Equal(t, types.QueueOutcomeRejectedOther, outcome,
+				"shutdown should return the rejected outcome")
+		})
+		t.Run("OnControllerShutdownPreservesQueuedOutcome", func(t *testing.T) {
+			t.Parallel()
+			ctx, cancel := context.WithCancel(t.Context())
+			h := newUnitHarness(ctx, t, &Config{}, nil, nil)
+			item := internal.NewItem(newTestRequest(defaultFlowKey), 0, time.Now())
+			item.SetHandle(&fwkfcmocks.MockQueueItemHandle{})
+
+			cancel()
+
+			outcome, err := h.fc.awaitFinalization(context.Background(), item)
+			require.Error(t, err, "awaitFinalization must fail when controller shuts down")
+			assert.ErrorIs(t, err, types.ErrEvicted,
+				"a queued item should be evicted, not rejected, during shutdown")
+			assert.ErrorIs(t, err, types.ErrFlowControllerNotRunning,
+				"queued shutdown should preserve the shutdown cause")
+			assert.Equal(t, types.QueueOutcomeEvictedOther, outcome,
+				"a queued item should return the evicted outcome")
+		})
 
 		t.Run("OnRegistryConnectionError", func(t *testing.T) {
 			t.Parallel()

pkg/epp/flowcontrol/integration_test.go

@@ -782,6 +782,11 @@ func TestConcurrentEnqueueDuringShutdown(t *testing.T) {
 
 	h := newHarness(t, harnessOpts{
 		detector: detector,
+		controllerCfg: &controller.Config{
+			DefaultRequestTTL:        0,
+			ExpiryCleanupInterval:    10 * time.Millisecond,
+			EnqueueChannelBufferSize: 100,
+		},
 	})
 
 	key := flowcontrol.FlowKey{ID: "flow-a", Priority: 0}
@@ -791,8 +796,8 @@ func TestConcurrentEnqueueDuringShutdown(t *testing.T) {
 	for i := 0; i < numRequests; i++ {
 		id := fmt.Sprintf("req-%d", i)
 		go func() {
-			req := &testRequest{id: id, key: key, byteSize: 100, ttl: 5 * time.Minute}
-			outcome, err := h.fc.EnqueueAndWait(h.ctx, req)
+			req := &testRequest{id: id, key: key, byteSize: 100}
+			outcome, err := h.fc.EnqueueAndWait(context.Background(), req)
 			results <- dispatchResult{id: id, outcome: outcome, err: err}
 		}()
 	}
@@ -809,6 +814,8 @@ func TestConcurrentEnqueueDuringShutdown(t *testing.T) {
 				"no request should dispatch (detector is blocked and controller is shutting down)")
 			require.Error(t, r.err,
 				"every request should receive an error during shutdown")
+			require.ErrorIs(t, r.err, fcTypes.ErrFlowControllerNotRunning,
+				"every request should report the shutdown cause")
 		case <-time.After(5 * time.Second):
 			t.Fatalf("request %d hung during concurrent shutdown", i)
 		}

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

@@ -464,6 +464,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)
@@ -537,6 +539,8 @@ func Reset() {
 	llmdSchedulerAttemptsTotal.Reset()
 	pluginProcessingLatencies.Reset()
 	llmdPluginProcessingLatencies.Reset()
+	llmdRequestProcessingLatency.Reset()
+	llmdResponseProcessingLatency.Reset()
 	inferenceExtensionInfo.Reset()
 	llmdInferenceExtensionInfo.Reset()
 	flowControlRequestQueueDuration.Reset()
@@ -775,6 +779,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 {