From 3b835bbcca4cefb387c576c580fee31a74a913c9 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 10 Jul 2026 17:58:00 +0000 Subject: [PATCH 1/2] Initial plan From 865c7964c43f64b8bf1f1214df43a1476d5c2699 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 10 Jul 2026 18:08:53 +0000 Subject: [PATCH 2/2] handlers: guard latency metrics when timestamps are absent When HandleResponseBody reaches end-of-stream it previously called all latency metric helpers unconditionally, including RecordNormalizedTimePerOutputToken, RecordRequestLatencies, RecordRequestTTFT, and RecordRequestTPOT. Those helpers reject zero/unordered timestamps and log errors, causing deterministic failures in tests that exercise response parsing or response size accumulation without initialising timing fields. Guard each metric call behind the timestamp checks it requires: - RecordNormalizedTimePerOutputToken and RecordRequestLatencies: only when RequestReceivedTimestamp and ResponseCompleteTimestamp are non-zero. - RecordRequestTTFT: only when RequestReceivedTimestamp and FirstTokenTimestamp are non-zero. - RecordRequestTPOT: only when all three timestamps are non-zero. - RecordResponseSizes is always recorded at end-of-stream. Add TestHandleResponseBodyZeroTimestamps to confirm that end-of-stream handling with zero timestamps records no latency metrics and still records the response size. Signed-off-by: GitHub Copilot --- pkg/epp/handlers/response.go | 14 +++++--- pkg/epp/handlers/response_test.go | 57 +++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 4 deletions(-) diff --git a/pkg/epp/handlers/response.go b/pkg/epp/handlers/response.go index 8cb3a057da..ac2fd23cdb 100644 --- a/pkg/epp/handlers/response.go +++ b/pkg/epp/handlers/response.go @@ -86,11 +86,17 @@ func (s *StreamingServer) HandleResponseBody(ctx context.Context, reqCtx *Reques } } if endOfStream { - metrics.RecordNormalizedTimePerOutputToken(ctx, reqCtx.IncomingModelName, reqCtx.TargetModelName, fairnessID, priority, reqCtx.RequestReceivedTimestamp, reqCtx.ResponseCompleteTimestamp, reqCtx.Usage.CompletionTokens) - metrics.RecordRequestLatencies(ctx, reqCtx.IncomingModelName, reqCtx.TargetModelName, fairnessID, priority, reqCtx.RequestReceivedTimestamp, reqCtx.ResponseCompleteTimestamp) + if !reqCtx.RequestReceivedTimestamp.IsZero() && !reqCtx.ResponseCompleteTimestamp.IsZero() { + metrics.RecordNormalizedTimePerOutputToken(ctx, reqCtx.IncomingModelName, reqCtx.TargetModelName, fairnessID, priority, reqCtx.RequestReceivedTimestamp, reqCtx.ResponseCompleteTimestamp, reqCtx.Usage.CompletionTokens) + metrics.RecordRequestLatencies(ctx, reqCtx.IncomingModelName, reqCtx.TargetModelName, fairnessID, priority, reqCtx.RequestReceivedTimestamp, reqCtx.ResponseCompleteTimestamp) + } metrics.RecordResponseSizes(reqCtx.IncomingModelName, reqCtx.TargetModelName, fairnessID, priority, reqCtx.ResponseSize) - metrics.RecordRequestTTFT(ctx, reqCtx.IncomingModelName, reqCtx.TargetModelName, fairnessID, priority, reqCtx.modelServerStreaming, reqCtx.RequestReceivedTimestamp, reqCtx.FirstTokenTimestamp) - metrics.RecordRequestTPOT(ctx, reqCtx.IncomingModelName, reqCtx.TargetModelName, fairnessID, priority, reqCtx.RequestReceivedTimestamp, reqCtx.FirstTokenTimestamp, reqCtx.ResponseCompleteTimestamp, reqCtx.Usage.CompletionTokens) + if !reqCtx.RequestReceivedTimestamp.IsZero() && !reqCtx.FirstTokenTimestamp.IsZero() { + metrics.RecordRequestTTFT(ctx, reqCtx.IncomingModelName, reqCtx.TargetModelName, fairnessID, priority, reqCtx.modelServerStreaming, reqCtx.RequestReceivedTimestamp, reqCtx.FirstTokenTimestamp) + } + if !reqCtx.RequestReceivedTimestamp.IsZero() && !reqCtx.FirstTokenTimestamp.IsZero() && !reqCtx.ResponseCompleteTimestamp.IsZero() { + metrics.RecordRequestTPOT(ctx, reqCtx.IncomingModelName, reqCtx.TargetModelName, fairnessID, priority, reqCtx.RequestReceivedTimestamp, reqCtx.FirstTokenTimestamp, reqCtx.ResponseCompleteTimestamp, reqCtx.Usage.CompletionTokens) + } } return s.director.HandleResponseBody(ctx, reqCtx, endOfStream) } diff --git a/pkg/epp/handlers/response_test.go b/pkg/epp/handlers/response_test.go index 87f8fc06da..ac652dcdb8 100644 --- a/pkg/epp/handlers/response_test.go +++ b/pkg/epp/handlers/response_test.go @@ -309,6 +309,63 @@ func TestHandleResponseBodyWithoutSchedulingRequest(t *testing.T) { require.Equal(t, uint64(1), histogram.GetSampleCount()) } +// TestHandleResponseBodyZeroTimestamps verifies that end-of-stream handling with zero +// timestamps does not emit latency metrics or produce error log entries. +func TestHandleResponseBodyZeroTimestamps(t *testing.T) { + ctx := logutil.NewTestLoggerIntoContext(context.Background()) + eppmetrics.Register() + eppmetrics.Reset() + t.Cleanup(eppmetrics.Reset) + + server := &StreamingServer{ + parserRegistry: NewParserRegistry([]fwkrh.Parser{openai.NewOpenAIParser()}, logr.Discard()), + } + server.director = &mockDirector{} + // RequestReceivedTimestamp, FirstTokenTimestamp, and ResponseCompleteTimestamp are all zero. + reqCtx := &RequestContext{ + IncomingModelName: "incoming-model", + TargetModelName: "target-model", + Response: &Response{ + Headers: map[string]string{}, + }, + SchedulingRequest: &fwksched.InferenceRequest{FairnessID: metadata.DefaultFairnessID}, + } + + require.NotPanics(t, func() { + server.HandleResponseBody(ctx, reqCtx, []byte("data"), true) + }) + + // Latency metrics must not be recorded when timestamps are absent. + families, err := ctrlmetrics.Registry.Gather() + require.NoError(t, err) + latencyMetrics := []string{ + "llm_d_epp_request_ntpot_seconds", + "llm_d_epp_request_duration_seconds", + "llm_d_epp_request_ttft_seconds", + "llm_d_epp_request_streaming_tpot_seconds", + } + for _, family := range families { + for _, name := range latencyMetrics { + if family.GetName() == name { + for _, metric := range family.GetMetric() { + if metricHasLabels(metric, map[string]string{"model_name": "incoming-model"}) { + t.Errorf("latency metric %q must not be recorded with zero timestamps, got sample count %d", + name, metric.GetHistogram().GetSampleCount()) + } + } + } + } + } + + // ResponseSize should still be recorded. + sizeHistogram := findHistogramMetric(t, "llm_d_epp_response_size_bytes", map[string]string{ + "model_name": "incoming-model", + "target_model_name": "target-model", + "fairness_id": metadata.DefaultFairnessID, + }) + require.Equal(t, uint64(1), sizeHistogram.GetSampleCount()) +} + func findHistogramMetric(t *testing.T, name string, labels map[string]string) *dto.Histogram { t.Helper()