metrics: TTFT in streaming mode

Signed-off-by: Jintao Zhang <[email protected]>

Author: tao12345666333

src/semantic-router/pkg/extproc/metrics_integration_test.go

@@ -123,4 +123,42 @@ var _ = Describe("Metrics recording", func() {
 		Expect(afterPrompt).To(BeNumerically(">", beforePrompt))
 		Expect(afterCompletion).To(BeNumerically(">", beforeCompletion))
 	})
+
+	It("records TTFT on first streamed body chunk for SSE responses", func() {
+		ctx := &RequestContext{
+			RequestModel:        "model-stream",
+			ProcessingStartTime: time.Now().Add(-120 * time.Millisecond),
+			Headers:             map[string]string{"accept": "text/event-stream"},
+		}
+
+		// Simulate header phase: SSE content-type indicates streaming
+		respHeaders := &ext_proc.ProcessingRequest_ResponseHeaders{
+			ResponseHeaders: &ext_proc.HttpHeaders{
+				Headers: &core.HeaderMap{Headers: []*core.HeaderValue{{Key: "content-type", Value: "text/event-stream"}}},
+			},
+		}
+
+		before := getHistogramSampleCount("llm_model_ttft_seconds", ctx.RequestModel)
+
+		// Handle response headers (should NOT record TTFT for streaming)
+		response1, err := router.handleResponseHeaders(respHeaders, ctx)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(response1.GetResponseHeaders()).NotTo(BeNil())
+		Expect(ctx.IsStreamingResponse).To(BeTrue())
+		Expect(ctx.TTFTRecorded).To(BeFalse())
+
+		// Now simulate the first streamed body chunk
+		respBody := &ext_proc.ProcessingRequest_ResponseBody{
+			ResponseBody: &ext_proc.HttpBody{Body: []byte("data: chunk-1\n")},
+		}
+
+		response2, err := router.handleResponseBody(respBody, ctx)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(response2.GetResponseBody()).NotTo(BeNil())
+
+		after := getHistogramSampleCount("llm_model_ttft_seconds", ctx.RequestModel)
+		Expect(after).To(BeNumerically(">", before))
+		Expect(ctx.TTFTRecorded).To(BeTrue())
+		Expect(ctx.TTFTSeconds).To(BeNumerically(">", 0))
+	})
 })

src/semantic-router/pkg/extproc/request_handler.go

@@ -108,6 +108,10 @@ type RequestContext struct {
 	StartTime           time.Time
 	ProcessingStartTime time.Time
 
+	// Streaming detection
+	ExpectStreamingResponse bool // set from request Accept header
+	IsStreamingResponse     bool // set from response Content-Type
+
 	// TTFT tracking
 	TTFTRecorded bool
 	TTFTSeconds  float64
@@ -136,7 +140,14 @@ func (r *OpenAIRouter) handleRequestHeaders(v *ext_proc.ProcessingRequest_Reques
 		}
 	}
 
-	// Allow the request to continue
+	// Detect if the client expects a streaming response (SSE)
+	if accept, ok := ctx.Headers["accept"]; ok {
+		if strings.Contains(strings.ToLower(accept), "text/event-stream") {
+			ctx.ExpectStreamingResponse = true
+		}
+	}
+
+	// Prepare base response
 	response := &ext_proc.ProcessingResponse{
 		Response: &ext_proc.ProcessingResponse_RequestHeaders{
 			RequestHeaders: &ext_proc.HeadersResponse{
@@ -148,6 +159,10 @@ func (r *OpenAIRouter) handleRequestHeaders(v *ext_proc.ProcessingRequest_Reques
 		},
 	}
 
+	// If streaming is expected, we rely on Envoy config to set response_body_mode: STREAMED for SSE.
+	// Some Envoy/control-plane versions may not support per-message ModeOverride; avoid compile-time coupling here.
+	// The Accept header is still recorded on context for downstream logic.
+
 	return response, nil
 }
 

src/semantic-router/pkg/extproc/response_handler.go

@@ -3,6 +3,7 @@ package extproc
 import (
 	"encoding/json"
 	"strconv"
+	"strings"
 	"time"
 
 	core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
@@ -17,6 +18,9 @@ import (
 func (r *OpenAIRouter) handleResponseHeaders(v *ext_proc.ProcessingRequest_ResponseHeaders, ctx *RequestContext) (*ext_proc.ProcessingResponse, error) {
 	// Detect upstream HTTP status and record non-2xx as errors
 	if v != nil && v.ResponseHeaders != nil && v.ResponseHeaders.Headers != nil {
+		// Determine if the response is streaming based on Content-Type
+		ctx.IsStreamingResponse = isStreamingContentType(v.ResponseHeaders.Headers)
+
 		if statusCode := getStatusFromHeaders(v.ResponseHeaders.Headers); statusCode != 0 {
 			if statusCode >= 500 {
 				metrics.RecordRequestError(getModelFromCtx(ctx), "upstream_5xx")
@@ -26,8 +30,10 @@ func (r *OpenAIRouter) handleResponseHeaders(v *ext_proc.ProcessingRequest_Respo
 		}
 	}
 
-	// Best-effort TTFT measurement: record on first response headers if we have a start time and model
-	if ctx != nil && !ctx.TTFTRecorded && !ctx.ProcessingStartTime.IsZero() && ctx.RequestModel != "" {
+	// Best-effort TTFT measurement:
+	// - For non-streaming responses, record on first response headers (approx TTFB ~= TTFT)
+	// - For streaming responses (SSE), defer TTFT until the first response body chunk arrives
+	if ctx != nil && !ctx.IsStreamingResponse && !ctx.TTFTRecorded && !ctx.ProcessingStartTime.IsZero() && ctx.RequestModel != "" {
 		ttft := time.Since(ctx.ProcessingStartTime).Seconds()
 		if ttft > 0 {
 			metrics.RecordModelTTFT(ctx.RequestModel, ttft)
@@ -79,13 +85,58 @@ func getModelFromCtx(ctx *RequestContext) string {
 	return ctx.RequestModel
 }
 
+// isStreamingContentType checks if the response content-type indicates streaming (SSE)
+func isStreamingContentType(headerMap *core.HeaderMap) bool {
+	if headerMap == nil {
+		return false
+	}
+	for _, hv := range headerMap.Headers {
+		if strings.ToLower(hv.Key) == "content-type" {
+			val := hv.Value
+			if val == "" && len(hv.RawValue) > 0 {
+				val = string(hv.RawValue)
+			}
+			if strings.Contains(strings.ToLower(val), "text/event-stream") {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // handleResponseBody processes the response body
 func (r *OpenAIRouter) handleResponseBody(v *ext_proc.ProcessingRequest_ResponseBody, ctx *RequestContext) (*ext_proc.ProcessingResponse, error) {
 	completionLatency := time.Since(ctx.StartTime)
 
 	// Process the response for caching
 	responseBody := v.ResponseBody.Body
 
+	// If this is a streaming response (e.g., SSE), record TTFT on the first body chunk
+	// and skip JSON parsing/caching which are not applicable for SSE chunks.
+	if ctx.IsStreamingResponse {
+		if ctx != nil && !ctx.TTFTRecorded && !ctx.ProcessingStartTime.IsZero() && ctx.RequestModel != "" {
+			ttft := time.Since(ctx.ProcessingStartTime).Seconds()
+			if ttft > 0 {
+				metrics.RecordModelTTFT(ctx.RequestModel, ttft)
+				ctx.TTFTSeconds = ttft
+				ctx.TTFTRecorded = true
+				observability.Infof("Recorded TTFT on first streamed body chunk: %.3fs", ttft)
+			}
+		}
+
+		// For streaming chunks, just continue (no token parsing or cache update)
+		response := &ext_proc.ProcessingResponse{
+			Response: &ext_proc.ProcessingResponse_ResponseBody{
+				ResponseBody: &ext_proc.BodyResponse{
+					Response: &ext_proc.CommonResponse{
+						Status: ext_proc.CommonResponse_CONTINUE,
+					},
+				},
+			},
+		}
+		return response, nil
+	}
+
 	// Parse tokens from the response JSON using OpenAI SDK types
 	var parsed openai.ChatCompletion
 	if err := json.Unmarshal(responseBody, &parsed); err != nil {