From c93bc7e91dab07b526f2146ec8d4d31c9d4ad2aa Mon Sep 17 00:00:00 2001 From: Loic Marchal Date: Thu, 30 Apr 2026 15:33:27 -0400 Subject: [PATCH 1/5] feature: add slo metrics to flowcontrol Signed-off-by: Loic Marchal --- pkg/epp/flowcontrol/controller/controller.go | 14 ++ .../flowcontrol/controller/internal/item.go | 17 +- pkg/epp/framework/common/request/headers.go | 7 + .../ordering/slodeadline/slo_deadline.go | 11 +- .../ordering/slodeadline/slo_deadline_test.go | 7 +- pkg/epp/metrics/metrics.go | 100 ++++++++-- pkg/epp/metrics/metrics_test.go | 177 +++++++++++++++++- 7 files changed, 301 insertions(+), 32 deletions(-) diff --git a/pkg/epp/flowcontrol/controller/controller.go b/pkg/epp/flowcontrol/controller/controller.go index 2bcb2293be..521f256dad 100644 --- a/pkg/epp/flowcontrol/controller/controller.go +++ b/pkg/epp/flowcontrol/controller/controller.go @@ -31,6 +31,7 @@ import ( "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/contracts" "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/controller/internal" "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/types" + fwkrequest "github.com/llm-d/llm-d-router/pkg/epp/framework/common/request" "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol" "github.com/llm-d/llm-d-router/pkg/epp/metrics" ) @@ -209,21 +210,27 @@ func (fc *FlowController) EnqueueAndWait( flowKey := req.FlowKey() priority := strconv.Itoa(flowKey.Priority) reqBytes := req.ByteSize() + sloClass := metrics.ClassifySLO(extractHeader(req, fwkrequest.TTFTSLOMsHeaderKey)) + metrics.RecordFlowControlSLOIncomingRequest(sloClass, req.InferencePoolName()) metrics.IncFlowControlQueueSize( flowKey.ID, priority, req.InferencePoolName(), + sloClass, req.ModelName(), req.TargetModelName()) defer metrics.DecFlowControlQueueSize( flowKey.ID, priority, req.InferencePoolName(), + sloClass, req.ModelName(), req.TargetModelName()) metrics.AddFlowControlQueueBytes( flowKey.ID, priority, req.InferencePoolName(), + sloClass, req.ModelName(), req.TargetModelName(), reqBytes) defer metrics.SubFlowControlQueueBytes( flowKey.ID, priority, req.InferencePoolName(), + sloClass, req.ModelName(), req.TargetModelName(), reqBytes) // 1. Create the derived context that governs this request's lifecycle (Parent Cancellation + TTL). @@ -328,6 +335,13 @@ func (fc *FlowController) withConnectionWithFallback( return fn(conn, fallback) }) } +func extractHeader(req flowcontrol.FlowControlRequest, name string) string { + infReq := req.InferenceRequest() + if infReq == nil || infReq.Headers == nil { + return "" + } + return fwkrequest.GetHeader(infReq.Headers, name) +} // tryDistribution handles a single attempt to submit a request to the processor. // It uses the provided `conn` to access the registry data plane. diff --git a/pkg/epp/flowcontrol/controller/internal/item.go b/pkg/epp/flowcontrol/controller/internal/item.go index 9fb219a373..f959532fdf 100644 --- a/pkg/epp/flowcontrol/controller/internal/item.go +++ b/pkg/epp/flowcontrol/controller/internal/item.go @@ -26,6 +26,7 @@ import ( "time" "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/types" + "github.com/llm-d/llm-d-router/pkg/epp/framework/common/request" "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol" "github.com/llm-d/llm-d-router/pkg/epp/metrics" ) @@ -159,16 +160,30 @@ func (fi *FlowItem) finalizeInternal(outcome types.QueueOutcome, err error) { duration := time.Since(fi.enqueueTime) flowKey := fi.originalRequest.FlowKey() + outcomeStr := outcome.String() metrics.RecordFlowControlRequestQueueDuration( - flowKey.ID, strconv.Itoa(flowKey.Priority), outcome.String(), + flowKey.ID, strconv.Itoa(flowKey.Priority), outcomeStr, fi.originalRequest.InferencePoolName(), fi.OriginalRequest().ModelName(), fi.OriginalRequest().TargetModelName(), duration) + sloClass := metrics.ClassifySLO(extractHeader(fi.originalRequest, request.TTFTSLOMsHeaderKey)) + metrics.RecordFlowControlSLORequestQueueDuration( + sloClass, outcomeStr, fi.originalRequest.InferencePoolName(), + duration) + fi.done <- finalState close(fi.done) } +func extractHeader(req flowcontrol.FlowControlRequest, name string) string { + infReq := req.InferenceRequest() + if infReq == nil || infReq.Headers == nil { + return "" + } + return request.GetHeader(infReq.Headers, name) +} + // inferOutcome determines the correct QueueOutcome and Error based on the cause of finalization and whether the item // was already admitted to a queue. func inferOutcome(cause error, isQueued bool) (types.QueueOutcome, error) { diff --git a/pkg/epp/framework/common/request/headers.go b/pkg/epp/framework/common/request/headers.go index a0966de27d..2241793046 100644 --- a/pkg/epp/framework/common/request/headers.go +++ b/pkg/epp/framework/common/request/headers.go @@ -18,6 +18,13 @@ package request import "strings" +const ( + // TTFTSLOMsHeaderKey is the request header name for SLO time-to-first-token in milliseconds. + TTFTSLOMsHeaderKey = "x-slo-ttft-ms" + // TPOTSLOMsHeaderKey is the request header name for SLO time-per-output-token in milliseconds. + TPOTSLOMsHeaderKey = "x-slo-tpot-ms" +) + // GetHeader returns the value for key from headers, with case-insensitive lookup. func GetHeader(headers map[string]string, key string) string { if v, ok := headers[key]; ok { diff --git a/pkg/epp/framework/plugins/flowcontrol/ordering/slodeadline/slo_deadline.go b/pkg/epp/framework/plugins/flowcontrol/ordering/slodeadline/slo_deadline.go index 17ccfd5131..c23a35432c 100644 --- a/pkg/epp/framework/plugins/flowcontrol/ordering/slodeadline/slo_deadline.go +++ b/pkg/epp/framework/plugins/flowcontrol/ordering/slodeadline/slo_deadline.go @@ -26,9 +26,9 @@ import ( "strings" "time" + "github.com/llm-d/llm-d-router/pkg/epp/framework/common/request" "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol" "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin" - "github.com/llm-d/llm-d-router/pkg/epp/metadata" ) const ( @@ -37,9 +37,6 @@ const ( // It selects the request with the earliest SLO-based deadline. // For detailed documentation, see README.md. SLODeadlineOrderingPolicyType = "slo-deadline-ordering-policy" - - // sloTtftHeader is the request header name for SLO time-to-first-token in milliseconds. - sloTtftHeader = metadata.TTFTSLOHeaderKey ) func SLODeadlineOrderingPolicyFactory(name string, _ *json.Decoder, _ plugin.Handle) (plugin.Plugin, error) { @@ -90,11 +87,11 @@ func calculateSLODeadline(item flowcontrol.QueueItemAccessor) time.Time { if infReq == nil || infReq.Headers == nil { return sloMaxDeadlineTime } - sloTtft, _ := metadata.GetLowerCaseHeaderValue(infReq.Headers, sloTtftHeader) - if sloTtft == "" { + sloTTFT := request.GetHeader(infReq.Headers, request.TTFTSLOMsHeaderKey) + if sloTTFT == "" { return sloMaxDeadlineTime } - ms, err := strconv.ParseInt(strings.TrimSpace(sloTtft), 10, 64) + ms, err := strconv.ParseInt(strings.TrimSpace(sloTTFT), 10, 64) if err != nil || ms < 0 { return sloMaxDeadlineTime } diff --git a/pkg/epp/framework/plugins/flowcontrol/ordering/slodeadline/slo_deadline_test.go b/pkg/epp/framework/plugins/flowcontrol/ordering/slodeadline/slo_deadline_test.go index 2c9cbc0aa3..68fe7e3fd5 100644 --- a/pkg/epp/framework/plugins/flowcontrol/ordering/slodeadline/slo_deadline_test.go +++ b/pkg/epp/framework/plugins/flowcontrol/ordering/slodeadline/slo_deadline_test.go @@ -22,6 +22,7 @@ import ( "github.com/stretchr/testify/assert" + fwkrequest "github.com/llm-d/llm-d-router/pkg/epp/framework/common/request" "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol" "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol/mocks" "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/scheduling" @@ -46,7 +47,7 @@ func TestSLODeadlinePolicy_WithName(t *testing.T) { func makeSLOItem(id string, received time.Time, sloTTFTMs string) flowcontrol.QueueItemAccessor { req := mocks.NewMockFlowControlRequest(10, id, testFlowKey) req.ReceivedTimestampV = received - req.InferenceRequestV = &scheduling.InferenceRequest{Headers: map[string]string{sloTtftHeader: sloTTFTMs}} + req.InferenceRequestV = &scheduling.InferenceRequest{Headers: map[string]string{fwkrequest.TTFTSLOMsHeaderKey: sloTTFTMs}} return &mocks.MockQueueItemAccessor{ EffectiveTTLV: 0, OriginalRequestV: req, @@ -107,7 +108,7 @@ func TestCalculateSLODeadline(t *testing.T) { // Valid header reqValid := mocks.NewMockFlowControlRequest(1, "valid", testFlowKey) reqValid.ReceivedTimestampV = now - reqValid.InferenceRequestV = &scheduling.InferenceRequest{Headers: map[string]string{sloTtftHeader: "200"}} + reqValid.InferenceRequestV = &scheduling.InferenceRequest{Headers: map[string]string{fwkrequest.TTFTSLOMsHeaderKey: "200"}} accValid := &mocks.MockQueueItemAccessor{OriginalRequestV: reqValid} deadline := calculateSLODeadline(accValid) assert.Equal(t, now.Add(200*time.Millisecond), deadline) @@ -141,7 +142,7 @@ func TestCalculateSLODeadline(t *testing.T) { // Invalid value reqInvalid := mocks.NewMockFlowControlRequest(3, "inv", testFlowKey) - reqInvalid.InferenceRequestV = &scheduling.InferenceRequest{Headers: map[string]string{sloTtftHeader: "x"}} + reqInvalid.InferenceRequestV = &scheduling.InferenceRequest{Headers: map[string]string{fwkrequest.TTFTSLOMsHeaderKey: "x"}} accInvalid := &mocks.MockQueueItemAccessor{OriginalRequestV: reqInvalid} assert.Equal(t, sloMaxDeadlineTime, calculateSLODeadline(accInvalid)) diff --git a/pkg/epp/metrics/metrics.go b/pkg/epp/metrics/metrics.go index c75acee5b0..3e74c9b8d2 100644 --- a/pkg/epp/metrics/metrics.go +++ b/pkg/epp/metrics/metrics.go @@ -312,7 +312,28 @@ var ( append([]string{"fairness_id", "priority", "outcome", "inference_pool"}, modelLabels...), ) - // Deprecated: Use llm_d_epp_flow_control_dispatch_cycle_duration_seconds instead. + flowControlSLORequestQueueDuration = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Subsystem: inferenceExtension, + Name: "flow_control_slo_request_queue_duration_seconds", + Help: metricsutil.HelpMsgWithStability("Distribution of the total time requests spend in the EPP flow control layer, partitioned by SLO class.", compbasemetrics.ALPHA), + Buckets: []float64{ + 0.0001, 0.0005, 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, + }, + }, + []string{"slo_class", "outcome", "inference_pool"}, + ) + + flowControlSLOIncomingRequestsTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Subsystem: inferenceExtension, + Name: "flow_control_slo_incoming_requests_total", + Help: metricsutil.HelpMsgWithStability("Total number of requests that entered the EPP flow control layer via EnqueueAndWait.", compbasemetrics.ALPHA), + }, + []string{"slo_class", "inference_pool"}, + ) + + // Deprecated: Use llm_d_router_epp_flow_control_dispatch_cycle_duration_seconds instead. // Tracked in: https://github.com/llm-d/llm-d-inference-scheduler/issues/1070 flowControlDispatchCycleDuration = prometheus.NewHistogramVec( prometheus.HistogramOpts{ @@ -348,7 +369,7 @@ var ( Name: "flow_control_queue_size", Help: metricsutil.HelpMsgWithStability("[Deprecated: Use llm_d_epp_flow_control_queue_size] Current number of requests actively held in the Flow Control queue.", compbasemetrics.ALPHA), }, - append([]string{"fairness_id", "priority", "inference_pool"}, modelLabels...), + append([]string{"fairness_id", "priority", "inference_pool", "slo_class"}, modelLabels...), ) // Deprecated: Use llm_d_epp_flow_control_queue_bytes instead. @@ -359,7 +380,7 @@ var ( Name: "flow_control_queue_bytes", Help: metricsutil.HelpMsgWithStability("[Deprecated: Use llm_d_epp_flow_control_queue_bytes] Current total size in bytes of requests actively held in the Flow Control queue.", compbasemetrics.ALPHA), }, - append([]string{"fairness_id", "priority", "inference_pool"}, modelLabels...), + append([]string{"fairness_id", "priority", "inference_pool", "slo_class"}, modelLabels...), ) // Deprecated: Use llm_d_epp_flow_control_pool_saturation instead. @@ -462,6 +483,8 @@ func Register(customCollectors ...prometheus.Collector) { metrics.Registry.MustRegister(llmdInferenceExtensionInfo) metrics.Registry.MustRegister(flowControlRequestQueueDuration) metrics.Registry.MustRegister(llmdFlowControlRequestQueueDuration) + metrics.Registry.MustRegister(flowControlSLORequestQueueDuration) + metrics.Registry.MustRegister(flowControlSLOIncomingRequestsTotal) metrics.Registry.MustRegister(flowControlDispatchCycleDuration) metrics.Registry.MustRegister(llmdFlowControlDispatchCycleDuration) metrics.Registry.MustRegister(flowControlQueueSize) @@ -532,6 +555,8 @@ func Reset() { llmdInferenceExtensionInfo.Reset() flowControlRequestQueueDuration.Reset() llmdFlowControlRequestQueueDuration.Reset() + flowControlSLORequestQueueDuration.Reset() + flowControlSLOIncomingRequestsTotal.Reset() flowControlQueueSize.Reset() llmdFlowControlQueueSize.Reset() flowControlQueueBytes.Reset() @@ -801,6 +826,59 @@ func RecordFlowControlRequestQueueDuration( ).Observe(duration.Seconds()) } +// SLO class constants label for flow control SLO metrics (bounded buckets for the TTFT SLO header in ms). +const ( + SLOClassNone = "none" + SLOClassBelowMS200 = "below_ms_200" + SLOClassMS200to399 = "ms_200_399" + SLOClassMS400to599 = "ms_400_599" + SLOClassMS600to799 = "ms_600_799" + SLOClassMS800to1000 = "ms_800_1000" + SLOClassAboveMS1000 = "above_ms_1000" +) + +// ClassifySLO maps a raw SLO header value (in milliseconds) to a bounded SLO class label. +// Returns SLOClassNone when the header is absent or unparseable. +func ClassifySLO(rawHeaderValue string) string { + if rawHeaderValue == "" { + return SLOClassNone + } + ms, err := strconv.ParseInt(rawHeaderValue, 10, 64) + if err != nil || ms < 0 { + return SLOClassNone + } + switch { + case ms < 200: + return SLOClassBelowMS200 + case ms < 400: + return SLOClassMS200to399 + case ms < 600: + return SLOClassMS400to599 + case ms < 800: + return SLOClassMS600to799 + case ms <= 1000: + return SLOClassMS800to1000 + default: + return SLOClassAboveMS1000 + } +} + +// RecordFlowControlSLORequestQueueDuration records the queue duration for a request partitioned by its +// SLO class (derived from the TTFT SLO header "x-slo-ttft-ms"). +func RecordFlowControlSLORequestQueueDuration( + sloClass, outcome, inferencePool string, + duration time.Duration, +) { + flowControlSLORequestQueueDuration.WithLabelValues( + sloClass, outcome, inferencePool, + ).Observe(duration.Seconds()) +} + +// RecordFlowControlSLOIncomingRequest increments the count of requests that entered flow control at EnqueueAndWait. +func RecordFlowControlSLOIncomingRequest(sloClass, inferencePool string) { + flowControlSLOIncomingRequestsTotal.WithLabelValues(sloClass, inferencePool).Inc() +} + // RecordFlowControlDispatchCycleDuration records the duration of a dispatch cycle in the Flow Control layer. func RecordFlowControlDispatchCycleDuration(duration time.Duration) { flowControlDispatchCycleDuration.WithLabelValues().Observe(duration.Seconds()) @@ -822,26 +900,26 @@ func RecordFlowControlRequestEnqueueDuration( } // IncFlowControlQueueSize increments the Flow Control queue size gauge. -func IncFlowControlQueueSize(fairnessID, priority, inferencePool, modelName, targetModelName string) { - flowControlQueueSize.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Inc() +func IncFlowControlQueueSize(fairnessID, priority, inferencePool, sloClass, modelName, targetModelName string) { + flowControlQueueSize.WithLabelValues(fairnessID, priority, inferencePool, sloClass, modelName, targetModelName).Inc() llmdFlowControlQueueSize.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Inc() } // DecFlowControlQueueSize decrements the Flow Control queue size gauge. -func DecFlowControlQueueSize(fairnessID, priority, inferencePool, modelName, targetModelName string) { - flowControlQueueSize.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Dec() +func DecFlowControlQueueSize(fairnessID, priority, inferencePool, sloClass, modelName, targetModelName string) { + flowControlQueueSize.WithLabelValues(fairnessID, priority, inferencePool, sloClass, modelName, targetModelName).Dec() llmdFlowControlQueueSize.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Dec() } // AddFlowControlQueueBytes increments the Flow Control queue bytes gauge. -func AddFlowControlQueueBytes(fairnessID, priority, inferencePool, modelName, targetModelName string, bytes uint64) { - flowControlQueueBytes.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Add(float64(bytes)) +func AddFlowControlQueueBytes(fairnessID, priority, inferencePool, sloClass, modelName, targetModelName string, bytes uint64) { + flowControlQueueBytes.WithLabelValues(fairnessID, priority, inferencePool, sloClass, modelName, targetModelName).Add(float64(bytes)) llmdFlowControlQueueBytes.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Add(float64(bytes)) } // SubFlowControlQueueBytes decrements the Flow Control queue bytes gauge. -func SubFlowControlQueueBytes(fairnessID, priority, inferencePool, modelName, targetModelName string, bytes uint64) { - flowControlQueueBytes.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Sub(float64(bytes)) +func SubFlowControlQueueBytes(fairnessID, priority, inferencePool, sloClass, modelName, targetModelName string, bytes uint64) { + flowControlQueueBytes.WithLabelValues(fairnessID, priority, inferencePool, sloClass, modelName, targetModelName).Sub(float64(bytes)) llmdFlowControlQueueBytes.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Sub(float64(bytes)) } diff --git a/pkg/epp/metrics/metrics_test.go b/pkg/epp/metrics/metrics_test.go index ca0c00a685..5e55cd0f13 100644 --- a/pkg/epp/metrics/metrics_test.go +++ b/pkg/epp/metrics/metrics_test.go @@ -1259,17 +1259,173 @@ func TestFlowControlQueueDurationMetric(t *testing.T) { } } +func TestClassifySLO(t *testing.T) { + testCases := []struct { + raw string + expected string + }{ + {"", SLOClassNone}, + {"not-a-number", SLOClassNone}, + {"-1", SLOClassNone}, + {"0", SLOClassBelowMS200}, + {"50", SLOClassBelowMS200}, + {"199", SLOClassBelowMS200}, + {"200", SLOClassMS200to399}, + {"399", SLOClassMS200to399}, + {"400", SLOClassMS400to599}, + {"500", SLOClassMS400to599}, + {"599", SLOClassMS400to599}, + {"600", SLOClassMS600to799}, + {"799", SLOClassMS600to799}, + {"800", SLOClassMS800to1000}, + {"1000", SLOClassMS800to1000}, + {"1001", SLOClassAboveMS1000}, + {"5000", SLOClassAboveMS1000}, + } + for _, tc := range testCases { + t.Run("raw="+tc.raw, func(t *testing.T) { + require.Equal(t, tc.expected, ClassifySLO(tc.raw)) + }) + } +} + +func TestFlowControlSLOIncomingRequestsTotalMetric(t *testing.T) { + Reset() + + const pool = "pool-1" + + RecordFlowControlSLOIncomingRequest(SLOClassBelowMS200, pool) + RecordFlowControlSLOIncomingRequest(SLOClassBelowMS200, pool) + RecordFlowControlSLOIncomingRequest(SLOClassMS400to599, pool) + RecordFlowControlSLOIncomingRequest(SLOClassNone, "pool-2") + + testCases := []struct { + name string + labels prometheus.Labels + expectCount float64 + }{ + { + name: "below_ms_200, pool-1", + labels: prometheus.Labels{"slo_class": SLOClassBelowMS200, "inference_pool": pool}, + expectCount: 2, + }, + { + name: "ms_400_599, pool-1", + labels: prometheus.Labels{"slo_class": SLOClassMS400to599, "inference_pool": pool}, + expectCount: 1, + }, + { + name: "none, pool-2", + labels: prometheus.Labels{"slo_class": SLOClassNone, "inference_pool": "pool-2"}, + expectCount: 1, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + val, err := testutil.GetCounterMetricValue(flowControlSLOIncomingRequestsTotal.With(tc.labels)) + require.NoError(t, err, "Failed to get counter value for labels %v", tc.labels) + require.Equal(t, tc.expectCount, val, "Counter value mismatch for labels %v", tc.labels) + }) + } +} + +func TestFlowControlSLOQueueDurationMetric(t *testing.T) { + Reset() + + const pool = "pool-1" + + records := []struct { + sloClass string + outcome string + duration time.Duration + }{ + {sloClass: SLOClassBelowMS200, outcome: "Dispatched", duration: 5 * time.Millisecond}, + {sloClass: SLOClassBelowMS200, outcome: "Dispatched", duration: 15 * time.Millisecond}, + {sloClass: SLOClassMS400to599, outcome: "Dispatched", duration: 50 * time.Millisecond}, + {sloClass: SLOClassNone, outcome: "RejectedCapacity", duration: 2 * time.Millisecond}, + {sloClass: SLOClassAboveMS1000, outcome: "Dispatched", duration: 200 * time.Millisecond}, + } + + for _, rec := range records { + RecordFlowControlSLORequestQueueDuration(rec.sloClass, rec.outcome, pool, rec.duration) + } + + testCases := []struct { + name string + labels prometheus.Labels + expectCount uint64 + expectSum float64 + }{ + { + name: "below_ms_200, dispatched", + labels: prometheus.Labels{ + "slo_class": SLOClassBelowMS200, + "outcome": "Dispatched", + "inference_pool": pool, + }, + expectCount: 2, + expectSum: 0.02, // 0.005 + 0.015 + }, + { + name: "ms_400_599, dispatched", + labels: prometheus.Labels{ + "slo_class": SLOClassMS400to599, + "outcome": "Dispatched", + "inference_pool": pool, + }, + expectCount: 1, + expectSum: 0.05, + }, + { + name: "none, rejected", + labels: prometheus.Labels{ + "slo_class": SLOClassNone, + "outcome": "RejectedCapacity", + "inference_pool": pool, + }, + expectCount: 1, + expectSum: 0.002, + }, + { + name: "above_ms_1000, dispatched", + labels: prometheus.Labels{ + "slo_class": SLOClassAboveMS1000, + "outcome": "Dispatched", + "inference_pool": pool, + }, + expectCount: 1, + expectSum: 0.2, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + labels := []string{ + tc.labels["slo_class"], + tc.labels["outcome"], + tc.labels["inference_pool"], + } + hist, err := getHistogramVecLabelValues(t, flowControlSLORequestQueueDuration, labels...) + require.NoError(t, err, "Failed to get histogram for labels %v", tc.labels) + require.Equal(t, tc.expectCount, hist.GetSampleCount(), "Sample count mismatch for labels %v", tc.labels) + require.InDelta(t, tc.expectSum, hist.GetSampleSum(), 0.00001, "Sample sum mismatch for labels %v", tc.labels) + }) + } +} + func TestFlowControlQueueSizeMetric(t *testing.T) { Reset() const ( - pool = "pool-1" - model = "qwen-3" - target = "qwen-3-base" + pool = "pool-1" + model = "qwen-3" + target = "qwen-3-base" + sloClass = SLOClassNone ) // Basic Inc/Dec - IncFlowControlQueueSize("user-a", "100", pool, model, target) + IncFlowControlQueueSize("user-a", "100", pool, sloClass, model, target) val, err := testutil.GetGaugeMetricValue(flowControlQueueSize.WithLabelValues("user-a", "100", pool, model, target)) require.NoError(t, err) require.Equal(t, 1.0, val) @@ -1278,7 +1434,7 @@ func TestFlowControlQueueSizeMetric(t *testing.T) { require.NoError(t, err) require.Equal(t, 1.0, valNew) - DecFlowControlQueueSize("user-a", "100", pool, model, target) + DecFlowControlQueueSize("user-a", "100", pool, sloClass, model, target) val, err = testutil.GetGaugeMetricValue(flowControlQueueSize.WithLabelValues("user-a", "100", pool, model, target)) require.NoError(t, err) require.Equal(t, 0.0, val) @@ -1292,12 +1448,13 @@ func TestFlowControlQueueBytesMetric(t *testing.T) { Reset() const ( - pool = "pool-1" - model = "qwen-3" - target = "qwen-3-base" + pool = "pool-1" + model = "qwen-3" + target = "qwen-3-base" + sloClass = SLOClassNone ) - AddFlowControlQueueBytes("user-a", "100", pool, model, target, 32) + AddFlowControlQueueBytes("user-a", "100", pool, sloClass, model, target, 32) val, err := testutil.GetGaugeMetricValue(flowControlQueueBytes.WithLabelValues("user-a", "100", pool, model, target)) require.NoError(t, err) require.Equal(t, 32.0, val) @@ -1306,7 +1463,7 @@ func TestFlowControlQueueBytesMetric(t *testing.T) { require.NoError(t, err) require.Equal(t, 32.0, valNew) - SubFlowControlQueueBytes("user-a", "100", pool, model, target, 32) + SubFlowControlQueueBytes("user-a", "100", pool, sloClass, model, target, 32) val, err = testutil.GetGaugeMetricValue(flowControlQueueBytes.WithLabelValues("user-a", "100", pool, model, target)) require.NoError(t, err) require.Equal(t, 0.0, val) From 4f69054f7c5abfe0f0bbc6de8a377707f0c1816c Mon Sep 17 00:00:00 2001 From: Loic Marchal Date: Mon, 11 May 2026 23:17:58 -0400 Subject: [PATCH 2/5] move SLO range definition to InferenceObjective Signed-off-by: Loic Marchal --- apix/v1alpha2/inferenceobjective_types.go | 41 ++++++++++++ apix/v1alpha2/zz_generated.deepcopy.go | 55 ++++++++++++++++ .../apix/v1alpha2/inferenceobjectivespec.go | 13 ++++ .../apix/v1alpha2/slorangems.go | 39 +++++++++++ .../apix/v1alpha2/slospec.go | 38 +++++++++++ client-go/applyconfiguration/utils.go | 4 ++ .../bases/llm-d.ai_inferenceobjectives.yaml | 44 +++++++++++++ config/manifests/inferenceobjective.yaml | 10 +++ pkg/epp/flowcontrol/controller/controller.go | 6 +- .../flowcontrol/controller/internal/item.go | 6 +- pkg/epp/metrics/metrics.go | 38 +---------- pkg/epp/metrics/metrics_test.go | 64 +++++-------------- pkg/epp/requestcontrol/director.go | 30 +++++++++ 13 files changed, 303 insertions(+), 85 deletions(-) create mode 100644 client-go/applyconfiguration/apix/v1alpha2/slorangems.go create mode 100644 client-go/applyconfiguration/apix/v1alpha2/slospec.go diff --git a/apix/v1alpha2/inferenceobjective_types.go b/apix/v1alpha2/inferenceobjective_types.go index 9b7b11037c..8ab7777f3e 100644 --- a/apix/v1alpha2/inferenceobjective_types.go +++ b/apix/v1alpha2/inferenceobjective_types.go @@ -76,6 +76,47 @@ type InferenceObjectiveSpec struct { // // +kubebuilder:validation:Required PoolRef PoolObjectReference `json:"poolRef"` + + // SLO defines the expected Service Level Objective ranges for requests targeting this objective. + // When set, the InferenceObjective name is used as the SLO class label in metrics. + // The ranges define the expected bounds for SLO header values (e.g. x-slo-ttft-ms), + // enabling mismatch detection between a request's SLO deadline and its declared objective. + // + // +optional + SLO *SLOSpec `json:"slo,omitempty"` +} + +// SLOSpec defines the expected SLO ranges for this objective. +type SLOSpec struct { + // TTFT defines the expected Time To First Token SLO range in milliseconds. + // The range bounds correspond to values expected in the x-slo-ttft-ms request header. + // + // +optional + TTFT *SLORangeMs `json:"ttft,omitempty"` + + // TPOT defines the expected Time Per Output Token SLO range in milliseconds. + // The range bounds correspond to values expected in the x-slo-tpot-ms request header. + // + // +optional + TPOT *SLORangeMs `json:"tpot,omitempty"` +} + +// SLORangeMs defines a millisecond range for an SLO metric. +// MinMs is inclusive, MaxMs is exclusive: [MinMs, MaxMs). +// Omitting MinMs implies 0 (no lower bound). +// Omitting MaxMs implies no upper bound. +type SLORangeMs struct { + // MinMs is the inclusive lower bound of the SLO range in milliseconds. + // + // +optional + // +kubebuilder:validation:Minimum=0 + MinMs *int64 `json:"minMs,omitempty"` + + // MaxMs is the exclusive upper bound of the SLO range in milliseconds. + // + // +optional + // +kubebuilder:validation:Minimum=0 + MaxMs *int64 `json:"maxMs,omitempty"` } // InferenceObjectiveStatus defines the observed state of InferenceObjective diff --git a/apix/v1alpha2/zz_generated.deepcopy.go b/apix/v1alpha2/zz_generated.deepcopy.go index 494e9d31e7..d4ec1e28b8 100644 --- a/apix/v1alpha2/zz_generated.deepcopy.go +++ b/apix/v1alpha2/zz_generated.deepcopy.go @@ -214,6 +214,11 @@ func (in *InferenceObjectiveSpec) DeepCopyInto(out *InferenceObjectiveSpec) { **out = **in } out.PoolRef = in.PoolRef + if in.SLO != nil { + in, out := &in.SLO, &out.SLO + *out = new(SLOSpec) + (*in).DeepCopyInto(*out) + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InferenceObjectiveSpec. @@ -303,6 +308,56 @@ func (in *PoolObjectReference) DeepCopy() *PoolObjectReference { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *SLORangeMs) DeepCopyInto(out *SLORangeMs) { + *out = *in + if in.MinMs != nil { + in, out := &in.MinMs, &out.MinMs + *out = new(int64) + **out = **in + } + if in.MaxMs != nil { + in, out := &in.MaxMs, &out.MaxMs + *out = new(int64) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SLORangeMs. +func (in *SLORangeMs) DeepCopy() *SLORangeMs { + if in == nil { + return nil + } + out := new(SLORangeMs) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *SLOSpec) DeepCopyInto(out *SLOSpec) { + *out = *in + if in.TTFT != nil { + in, out := &in.TTFT, &out.TTFT + *out = new(SLORangeMs) + (*in).DeepCopyInto(*out) + } + if in.TPOT != nil { + in, out := &in.TPOT, &out.TPOT + *out = new(SLORangeMs) + (*in).DeepCopyInto(*out) + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SLOSpec. +func (in *SLOSpec) DeepCopy() *SLOSpec { + if in == nil { + return nil + } + out := new(SLOSpec) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *TargetModel) DeepCopyInto(out *TargetModel) { *out = *in diff --git a/client-go/applyconfiguration/apix/v1alpha2/inferenceobjectivespec.go b/client-go/applyconfiguration/apix/v1alpha2/inferenceobjectivespec.go index 614c4fb4b7..a815b7e435 100644 --- a/client-go/applyconfiguration/apix/v1alpha2/inferenceobjectivespec.go +++ b/client-go/applyconfiguration/apix/v1alpha2/inferenceobjectivespec.go @@ -30,6 +30,11 @@ type InferenceObjectiveSpecApplyConfiguration struct { Priority *int32 `json:"priority,omitempty"` // PoolRef is a reference to the inference pool, the pool must exist in the same namespace. PoolRef *PoolObjectReferenceApplyConfiguration `json:"poolRef,omitempty"` + // SLO defines the expected Service Level Objective ranges for requests targeting this objective. + // When set, the InferenceObjective name is used as the SLO class label in metrics. + // The ranges define the expected bounds for SLO header values (e.g. x-slo-ttft-ms), + // enabling mismatch detection between a request's SLO deadline and its declared objective. + SLO *SLOSpecApplyConfiguration `json:"slo,omitempty"` } // InferenceObjectiveSpecApplyConfiguration constructs a declarative configuration of the InferenceObjectiveSpec type for use with @@ -53,3 +58,11 @@ func (b *InferenceObjectiveSpecApplyConfiguration) WithPoolRef(value *PoolObject b.PoolRef = value return b } + +// WithSLO sets the SLO field in the declarative configuration to the given value +// and returns the receiver, so that objects can be built by chaining "With" function invocations. +// If called multiple times, the SLO field is set to the value of the last call. +func (b *InferenceObjectiveSpecApplyConfiguration) WithSLO(value *SLOSpecApplyConfiguration) *InferenceObjectiveSpecApplyConfiguration { + b.SLO = value + return b +} diff --git a/client-go/applyconfiguration/apix/v1alpha2/slorangems.go b/client-go/applyconfiguration/apix/v1alpha2/slorangems.go new file mode 100644 index 0000000000..284173e6d7 --- /dev/null +++ b/client-go/applyconfiguration/apix/v1alpha2/slorangems.go @@ -0,0 +1,39 @@ +// Code generated by applyconfiguration-gen. DO NOT EDIT. + +package v1alpha2 + +// SLORangeMsApplyConfiguration represents a declarative configuration of the SLORangeMs type for use +// with apply. +// +// SLORangeMs defines a millisecond range for an SLO metric. +// MinMs is inclusive, MaxMs is exclusive: [MinMs, MaxMs). +// Omitting MinMs implies 0 (no lower bound). +// Omitting MaxMs implies no upper bound. +type SLORangeMsApplyConfiguration struct { + // MinMs is the inclusive lower bound of the SLO range in milliseconds. + MinMs *int64 `json:"minMs,omitempty"` + // MaxMs is the exclusive upper bound of the SLO range in milliseconds. + MaxMs *int64 `json:"maxMs,omitempty"` +} + +// SLORangeMsApplyConfiguration constructs a declarative configuration of the SLORangeMs type for use with +// apply. +func SLORangeMs() *SLORangeMsApplyConfiguration { + return &SLORangeMsApplyConfiguration{} +} + +// WithMinMs sets the MinMs field in the declarative configuration to the given value +// and returns the receiver, so that objects can be built by chaining "With" function invocations. +// If called multiple times, the MinMs field is set to the value of the last call. +func (b *SLORangeMsApplyConfiguration) WithMinMs(value int64) *SLORangeMsApplyConfiguration { + b.MinMs = &value + return b +} + +// WithMaxMs sets the MaxMs field in the declarative configuration to the given value +// and returns the receiver, so that objects can be built by chaining "With" function invocations. +// If called multiple times, the MaxMs field is set to the value of the last call. +func (b *SLORangeMsApplyConfiguration) WithMaxMs(value int64) *SLORangeMsApplyConfiguration { + b.MaxMs = &value + return b +} diff --git a/client-go/applyconfiguration/apix/v1alpha2/slospec.go b/client-go/applyconfiguration/apix/v1alpha2/slospec.go new file mode 100644 index 0000000000..a10059f55f --- /dev/null +++ b/client-go/applyconfiguration/apix/v1alpha2/slospec.go @@ -0,0 +1,38 @@ +// Code generated by applyconfiguration-gen. DO NOT EDIT. + +package v1alpha2 + +// SLOSpecApplyConfiguration represents a declarative configuration of the SLOSpec type for use +// with apply. +// +// SLOSpec defines the expected SLO ranges for this objective. +type SLOSpecApplyConfiguration struct { + // TTFT defines the expected Time To First Token SLO range in milliseconds. + // The range bounds correspond to values expected in the x-slo-ttft-ms request header. + TTFT *SLORangeMsApplyConfiguration `json:"ttft,omitempty"` + // TPOT defines the expected Time Per Output Token SLO range in milliseconds. + // The range bounds correspond to values expected in the x-slo-tpot-ms request header. + TPOT *SLORangeMsApplyConfiguration `json:"tpot,omitempty"` +} + +// SLOSpecApplyConfiguration constructs a declarative configuration of the SLOSpec type for use with +// apply. +func SLOSpec() *SLOSpecApplyConfiguration { + return &SLOSpecApplyConfiguration{} +} + +// WithTTFT sets the TTFT field in the declarative configuration to the given value +// and returns the receiver, so that objects can be built by chaining "With" function invocations. +// If called multiple times, the TTFT field is set to the value of the last call. +func (b *SLOSpecApplyConfiguration) WithTTFT(value *SLORangeMsApplyConfiguration) *SLOSpecApplyConfiguration { + b.TTFT = value + return b +} + +// WithTPOT sets the TPOT field in the declarative configuration to the given value +// and returns the receiver, so that objects can be built by chaining "With" function invocations. +// If called multiple times, the TPOT field is set to the value of the last call. +func (b *SLOSpecApplyConfiguration) WithTPOT(value *SLORangeMsApplyConfiguration) *SLOSpecApplyConfiguration { + b.TPOT = value + return b +} diff --git a/client-go/applyconfiguration/utils.go b/client-go/applyconfiguration/utils.go index 56d725622e..92e43eaf00 100644 --- a/client-go/applyconfiguration/utils.go +++ b/client-go/applyconfiguration/utils.go @@ -36,6 +36,10 @@ func ForKind(kind schema.GroupVersionKind) interface{} { return &apixv1alpha2.ModelMatchApplyConfiguration{} case v1alpha2.SchemeGroupVersion.WithKind("PoolObjectReference"): return &apixv1alpha2.PoolObjectReferenceApplyConfiguration{} + case v1alpha2.SchemeGroupVersion.WithKind("SLORangeMs"): + return &apixv1alpha2.SLORangeMsApplyConfiguration{} + case v1alpha2.SchemeGroupVersion.WithKind("SLOSpec"): + return &apixv1alpha2.SLOSpecApplyConfiguration{} case v1alpha2.SchemeGroupVersion.WithKind("TargetModel"): return &apixv1alpha2.TargetModelApplyConfiguration{} diff --git a/config/crd/bases/llm-d.ai_inferenceobjectives.yaml b/config/crd/bases/llm-d.ai_inferenceobjectives.yaml index e072d4783e..921100a085 100644 --- a/config/crd/bases/llm-d.ai_inferenceobjectives.yaml +++ b/config/crd/bases/llm-d.ai_inferenceobjectives.yaml @@ -99,6 +99,50 @@ spec: Similarly requests with a Priority of -10 will always be served after requests with Priority of 0. format: int32 type: integer + slo: + description: |- + SLO defines the expected Service Level Objective ranges for requests targeting this objective. + When set, the InferenceObjective name is used as the SLO class label in metrics. + The ranges define the expected bounds for SLO header values (e.g. x-slo-ttft-ms), + enabling mismatch detection between a request's SLO deadline and its declared objective. + properties: + tpot: + description: |- + TPOT defines the expected Time Per Output Token SLO range in milliseconds. + The range bounds correspond to values expected in the x-slo-tpot-ms request header. + properties: + maxMs: + description: MaxMs is the exclusive upper bound of the SLO + range in milliseconds. + format: int64 + minimum: 0 + type: integer + minMs: + description: MinMs is the inclusive lower bound of the SLO + range in milliseconds. + format: int64 + minimum: 0 + type: integer + type: object + ttft: + description: |- + TTFT defines the expected Time To First Token SLO range in milliseconds. + The range bounds correspond to values expected in the x-slo-ttft-ms request header. + properties: + maxMs: + description: MaxMs is the exclusive upper bound of the SLO + range in milliseconds. + format: int64 + minimum: 0 + type: integer + minMs: + description: MinMs is the inclusive lower bound of the SLO + range in milliseconds. + format: int64 + minimum: 0 + type: integer + type: object + type: object required: - poolRef type: object diff --git a/config/manifests/inferenceobjective.yaml b/config/manifests/inferenceobjective.yaml index 7a8b6910bb..8e24a99e9d 100644 --- a/config/manifests/inferenceobjective.yaml +++ b/config/manifests/inferenceobjective.yaml @@ -7,6 +7,9 @@ spec: poolRef: group: inference.networking.k8s.io name: vllm-qwen3-32b + slo: + ttft: + minMs: 500 --- apiVersion: llm-d.ai/v1alpha2 kind: InferenceObjective @@ -17,6 +20,9 @@ spec: poolRef: group: inference.networking.k8s.io name: vllm-qwen3-32b + slo: + ttft: + maxMs: 100 --- apiVersion: llm-d.ai/v1alpha2 kind: InferenceObjective @@ -27,3 +33,7 @@ spec: poolRef: group: inference.networking.k8s.io name: vllm-qwen3-32b + slo: + ttft: + minMs: 100 + maxMs: 500 diff --git a/pkg/epp/flowcontrol/controller/controller.go b/pkg/epp/flowcontrol/controller/controller.go index 521f256dad..ae475baff5 100644 --- a/pkg/epp/flowcontrol/controller/controller.go +++ b/pkg/epp/flowcontrol/controller/controller.go @@ -33,6 +33,7 @@ import ( "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/types" fwkrequest "github.com/llm-d/llm-d-router/pkg/epp/framework/common/request" "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol" + "github.com/llm-d/llm-d-router/pkg/epp/metadata" "github.com/llm-d/llm-d-router/pkg/epp/metrics" ) @@ -210,7 +211,10 @@ func (fc *FlowController) EnqueueAndWait( flowKey := req.FlowKey() priority := strconv.Itoa(flowKey.Priority) reqBytes := req.ByteSize() - sloClass := metrics.ClassifySLO(extractHeader(req, fwkrequest.TTFTSLOMsHeaderKey)) + sloClass := extractHeader(req, metadata.ObjectiveKey) + if sloClass == "" { + sloClass = metrics.SLOClassNone + } metrics.RecordFlowControlSLOIncomingRequest(sloClass, req.InferencePoolName()) metrics.IncFlowControlQueueSize( flowKey.ID, priority, diff --git a/pkg/epp/flowcontrol/controller/internal/item.go b/pkg/epp/flowcontrol/controller/internal/item.go index f959532fdf..306bac9661 100644 --- a/pkg/epp/flowcontrol/controller/internal/item.go +++ b/pkg/epp/flowcontrol/controller/internal/item.go @@ -28,6 +28,7 @@ import ( "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/types" "github.com/llm-d/llm-d-router/pkg/epp/framework/common/request" "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol" + "github.com/llm-d/llm-d-router/pkg/epp/metadata" "github.com/llm-d/llm-d-router/pkg/epp/metrics" ) @@ -167,7 +168,10 @@ func (fi *FlowItem) finalizeInternal(outcome types.QueueOutcome, err error) { fi.OriginalRequest().ModelName(), fi.OriginalRequest().TargetModelName(), duration) - sloClass := metrics.ClassifySLO(extractHeader(fi.originalRequest, request.TTFTSLOMsHeaderKey)) + sloClass := extractHeader(fi.originalRequest, metadata.ObjectiveKey) + if sloClass == "" { + sloClass = metrics.SLOClassNone + } metrics.RecordFlowControlSLORequestQueueDuration( sloClass, outcomeStr, fi.originalRequest.InferencePoolName(), duration) diff --git a/pkg/epp/metrics/metrics.go b/pkg/epp/metrics/metrics.go index 3e74c9b8d2..7db6a2ae0a 100644 --- a/pkg/epp/metrics/metrics.go +++ b/pkg/epp/metrics/metrics.go @@ -826,42 +826,8 @@ func RecordFlowControlRequestQueueDuration( ).Observe(duration.Seconds()) } -// SLO class constants label for flow control SLO metrics (bounded buckets for the TTFT SLO header in ms). -const ( - SLOClassNone = "none" - SLOClassBelowMS200 = "below_ms_200" - SLOClassMS200to399 = "ms_200_399" - SLOClassMS400to599 = "ms_400_599" - SLOClassMS600to799 = "ms_600_799" - SLOClassMS800to1000 = "ms_800_1000" - SLOClassAboveMS1000 = "above_ms_1000" -) - -// ClassifySLO maps a raw SLO header value (in milliseconds) to a bounded SLO class label. -// Returns SLOClassNone when the header is absent or unparseable. -func ClassifySLO(rawHeaderValue string) string { - if rawHeaderValue == "" { - return SLOClassNone - } - ms, err := strconv.ParseInt(rawHeaderValue, 10, 64) - if err != nil || ms < 0 { - return SLOClassNone - } - switch { - case ms < 200: - return SLOClassBelowMS200 - case ms < 400: - return SLOClassMS200to399 - case ms < 600: - return SLOClassMS400to599 - case ms < 800: - return SLOClassMS600to799 - case ms <= 1000: - return SLOClassMS800to1000 - default: - return SLOClassAboveMS1000 - } -} +// SLOClassNone is the metric label used when a request has no associated InferenceObjective. +const SLOClassNone = "none" // RecordFlowControlSLORequestQueueDuration records the queue duration for a request partitioned by its // SLO class (derived from the TTFT SLO header "x-slo-ttft-ms"). diff --git a/pkg/epp/metrics/metrics_test.go b/pkg/epp/metrics/metrics_test.go index 5e55cd0f13..c146561f10 100644 --- a/pkg/epp/metrics/metrics_test.go +++ b/pkg/epp/metrics/metrics_test.go @@ -1259,44 +1259,14 @@ func TestFlowControlQueueDurationMetric(t *testing.T) { } } -func TestClassifySLO(t *testing.T) { - testCases := []struct { - raw string - expected string - }{ - {"", SLOClassNone}, - {"not-a-number", SLOClassNone}, - {"-1", SLOClassNone}, - {"0", SLOClassBelowMS200}, - {"50", SLOClassBelowMS200}, - {"199", SLOClassBelowMS200}, - {"200", SLOClassMS200to399}, - {"399", SLOClassMS200to399}, - {"400", SLOClassMS400to599}, - {"500", SLOClassMS400to599}, - {"599", SLOClassMS400to599}, - {"600", SLOClassMS600to799}, - {"799", SLOClassMS600to799}, - {"800", SLOClassMS800to1000}, - {"1000", SLOClassMS800to1000}, - {"1001", SLOClassAboveMS1000}, - {"5000", SLOClassAboveMS1000}, - } - for _, tc := range testCases { - t.Run("raw="+tc.raw, func(t *testing.T) { - require.Equal(t, tc.expected, ClassifySLO(tc.raw)) - }) - } -} - func TestFlowControlSLOIncomingRequestsTotalMetric(t *testing.T) { Reset() const pool = "pool-1" - RecordFlowControlSLOIncomingRequest(SLOClassBelowMS200, pool) - RecordFlowControlSLOIncomingRequest(SLOClassBelowMS200, pool) - RecordFlowControlSLOIncomingRequest(SLOClassMS400to599, pool) + RecordFlowControlSLOIncomingRequest("realtime", pool) + RecordFlowControlSLOIncomingRequest("realtime", pool) + RecordFlowControlSLOIncomingRequest("interactive", pool) RecordFlowControlSLOIncomingRequest(SLOClassNone, "pool-2") testCases := []struct { @@ -1305,13 +1275,13 @@ func TestFlowControlSLOIncomingRequestsTotalMetric(t *testing.T) { expectCount float64 }{ { - name: "below_ms_200, pool-1", - labels: prometheus.Labels{"slo_class": SLOClassBelowMS200, "inference_pool": pool}, + name: "realtime, pool-1", + labels: prometheus.Labels{"slo_class": "realtime", "inference_pool": pool}, expectCount: 2, }, { - name: "ms_400_599, pool-1", - labels: prometheus.Labels{"slo_class": SLOClassMS400to599, "inference_pool": pool}, + name: "interactive, pool-1", + labels: prometheus.Labels{"slo_class": "interactive", "inference_pool": pool}, expectCount: 1, }, { @@ -1340,11 +1310,11 @@ func TestFlowControlSLOQueueDurationMetric(t *testing.T) { outcome string duration time.Duration }{ - {sloClass: SLOClassBelowMS200, outcome: "Dispatched", duration: 5 * time.Millisecond}, - {sloClass: SLOClassBelowMS200, outcome: "Dispatched", duration: 15 * time.Millisecond}, - {sloClass: SLOClassMS400to599, outcome: "Dispatched", duration: 50 * time.Millisecond}, + {sloClass: "realtime", outcome: "Dispatched", duration: 5 * time.Millisecond}, + {sloClass: "realtime", outcome: "Dispatched", duration: 15 * time.Millisecond}, + {sloClass: "interactive", outcome: "Dispatched", duration: 50 * time.Millisecond}, {sloClass: SLOClassNone, outcome: "RejectedCapacity", duration: 2 * time.Millisecond}, - {sloClass: SLOClassAboveMS1000, outcome: "Dispatched", duration: 200 * time.Millisecond}, + {sloClass: "batch", outcome: "Dispatched", duration: 200 * time.Millisecond}, } for _, rec := range records { @@ -1358,9 +1328,9 @@ func TestFlowControlSLOQueueDurationMetric(t *testing.T) { expectSum float64 }{ { - name: "below_ms_200, dispatched", + name: "realtime, dispatched", labels: prometheus.Labels{ - "slo_class": SLOClassBelowMS200, + "slo_class": "realtime", "outcome": "Dispatched", "inference_pool": pool, }, @@ -1368,9 +1338,9 @@ func TestFlowControlSLOQueueDurationMetric(t *testing.T) { expectSum: 0.02, // 0.005 + 0.015 }, { - name: "ms_400_599, dispatched", + name: "interactive, dispatched", labels: prometheus.Labels{ - "slo_class": SLOClassMS400to599, + "slo_class": "interactive", "outcome": "Dispatched", "inference_pool": pool, }, @@ -1388,9 +1358,9 @@ func TestFlowControlSLOQueueDurationMetric(t *testing.T) { expectSum: 0.002, }, { - name: "above_ms_1000, dispatched", + name: "batch, dispatched", labels: prometheus.Labels{ - "slo_class": SLOClassAboveMS1000, + "slo_class": "batch", "outcome": "Dispatched", "inference_pool": pool, }, diff --git a/pkg/epp/requestcontrol/director.go b/pkg/epp/requestcontrol/director.go index 4e33e980f0..383d97c059 100644 --- a/pkg/epp/requestcontrol/director.go +++ b/pkg/epp/requestcontrol/director.go @@ -24,6 +24,7 @@ import ( "fmt" "math/rand" "net" + "strconv" "strings" "sync" "time" @@ -43,6 +44,7 @@ import ( "github.com/llm-d/llm-d-router/pkg/epp/datalayer" "github.com/llm-d/llm-d-router/pkg/epp/datastore" "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/contracts" + fwkrequest "github.com/llm-d/llm-d-router/pkg/epp/framework/common/request" fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer" fwkrc "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requestcontrol" fwkrh "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requesthandling" @@ -292,6 +294,8 @@ func (d *Director) HandleRequest(ctx context.Context, reqCtx *handlers.RequestCo } // Admit may block until flow control admits the request. + warnSLORangeMismatch(logger, infObjective, reqCtx.Request.Headers) + if err := d.admissionController.Admit(ctx, reqCtx, priority); err != nil { return reqCtx, err } @@ -495,6 +499,32 @@ func (d *Director) toSchedulerEndpoints(endpoints []fwkdl.Endpoint) []fwksched.E return result } +func warnSLORangeMismatch(logger logr.Logger, objective *v1alpha2.InferenceObjective, headers map[string]string) { + if objective.Spec.SLO == nil { + return + } + checkRange(logger, objective.Name, "ttft slo", objective.Spec.SLO.TTFT, headers[fwkrequest.TTFTSLOMsHeaderKey]) + checkRange(logger, objective.Name, "tpot slo", objective.Spec.SLO.TPOT, headers[fwkrequest.TPOTSLOMsHeaderKey]) +} + +func checkRange(logger logr.Logger, objectiveName, metric string, sloRange *v1alpha2.SLORangeMs, headerValue string) { + if sloRange == nil || headerValue == "" { + return + } + ms, err := strconv.ParseInt(headerValue, 10, 64) + if err != nil { + return + } + if sloRange.MinMs != nil && ms < *sloRange.MinMs { + logger.V(logutil.VERBOSE).Info("SLO header value below objective range", + "metric", metric, "headerValueMs", ms, "minMs", *sloRange.MinMs, "objective", objectiveName) + } + if sloRange.MaxMs != nil && ms >= *sloRange.MaxMs { + logger.V(logutil.VERBOSE).Info("SLO header value above objective range", + "metric", metric, "headerValueMs", ms, "maxMs", *sloRange.MaxMs, "objective", objectiveName) + } +} + // HandleResponseHeader is called when the response headers are received. func (d *Director) HandleResponseHeader(ctx context.Context, reqCtx *handlers.RequestContext) *handlers.RequestContext { if len(d.requestControlPlugins.responseReceivedPlugins) == 0 { From 95efa42e2d1afe137e6ee4ba2b0c70eab0fb335c Mon Sep 17 00:00:00 2001 From: Loic Marchal Date: Fri, 19 Jun 2026 18:24:46 -0400 Subject: [PATCH 3/5] fix comment Signed-off-by: Loic Marchal --- pkg/epp/metrics/metrics.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/epp/metrics/metrics.go b/pkg/epp/metrics/metrics.go index 7db6a2ae0a..a12c1ca095 100644 --- a/pkg/epp/metrics/metrics.go +++ b/pkg/epp/metrics/metrics.go @@ -333,7 +333,7 @@ var ( []string{"slo_class", "inference_pool"}, ) - // Deprecated: Use llm_d_router_epp_flow_control_dispatch_cycle_duration_seconds instead. + // Deprecated: Use llm_d_epp_flow_control_dispatch_cycle_duration_seconds instead. // Tracked in: https://github.com/llm-d/llm-d-inference-scheduler/issues/1070 flowControlDispatchCycleDuration = prometheus.NewHistogramVec( prometheus.HistogramOpts{ From 36d641d5049de750b6250df70eb6a4fe649164bd Mon Sep 17 00:00:00 2001 From: Loic Marchal Date: Fri, 19 Jun 2026 18:33:00 -0400 Subject: [PATCH 4/5] fix test Signed-off-by: Loic Marchal --- .../flowcontrol/ordering/slodeadline/slo_deadline_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/epp/framework/plugins/flowcontrol/ordering/slodeadline/slo_deadline_test.go b/pkg/epp/framework/plugins/flowcontrol/ordering/slodeadline/slo_deadline_test.go index 68fe7e3fd5..4b56ac9083 100644 --- a/pkg/epp/framework/plugins/flowcontrol/ordering/slodeadline/slo_deadline_test.go +++ b/pkg/epp/framework/plugins/flowcontrol/ordering/slodeadline/slo_deadline_test.go @@ -124,7 +124,7 @@ func TestCalculateSLODeadline(t *testing.T) { reqBoth := mocks.NewMockFlowControlRequest(1, "both", testFlowKey) reqBoth.ReceivedTimestampV = now reqBoth.InferenceRequestV = &scheduling.InferenceRequest{Headers: map[string]string{ - sloTtftHeader: "200", + metadata.TTFTSLOHeaderKey: "200", metadata.OldTTFTSLOHeaderKey: "50", }} accBoth := &mocks.MockQueueItemAccessor{OriginalRequestV: reqBoth} From da1387216de3af2e3643a160b28a7f71a0ac1999 Mon Sep 17 00:00:00 2001 From: Loic Marchal Date: Sat, 20 Jun 2026 00:01:07 -0400 Subject: [PATCH 5/5] clean unnecessary function Signed-off-by: Loic Marchal --- pkg/epp/flowcontrol/controller/controller.go | 15 +++++++-------- pkg/epp/flowcontrol/controller/internal/item.go | 17 ++++++----------- 2 files changed, 13 insertions(+), 19 deletions(-) diff --git a/pkg/epp/flowcontrol/controller/controller.go b/pkg/epp/flowcontrol/controller/controller.go index ae475baff5..a738fd9edc 100644 --- a/pkg/epp/flowcontrol/controller/controller.go +++ b/pkg/epp/flowcontrol/controller/controller.go @@ -211,7 +211,13 @@ func (fc *FlowController) EnqueueAndWait( flowKey := req.FlowKey() priority := strconv.Itoa(flowKey.Priority) reqBytes := req.ByteSize() - sloClass := extractHeader(req, metadata.ObjectiveKey) + sloClass := metrics.SLOClassNone + if r := req.InferenceRequest(); r != nil { + sloClass = fwkrequest.GetHeader(r.Headers, metadata.ObjectiveKey) + if sloClass == "" { + sloClass = metrics.SLOClassNone + } + } if sloClass == "" { sloClass = metrics.SLOClassNone } @@ -339,13 +345,6 @@ func (fc *FlowController) withConnectionWithFallback( return fn(conn, fallback) }) } -func extractHeader(req flowcontrol.FlowControlRequest, name string) string { - infReq := req.InferenceRequest() - if infReq == nil || infReq.Headers == nil { - return "" - } - return fwkrequest.GetHeader(infReq.Headers, name) -} // tryDistribution handles a single attempt to submit a request to the processor. // It uses the provided `conn` to access the registry data plane. diff --git a/pkg/epp/flowcontrol/controller/internal/item.go b/pkg/epp/flowcontrol/controller/internal/item.go index 306bac9661..2f7df49739 100644 --- a/pkg/epp/flowcontrol/controller/internal/item.go +++ b/pkg/epp/flowcontrol/controller/internal/item.go @@ -168,9 +168,12 @@ func (fi *FlowItem) finalizeInternal(outcome types.QueueOutcome, err error) { fi.OriginalRequest().ModelName(), fi.OriginalRequest().TargetModelName(), duration) - sloClass := extractHeader(fi.originalRequest, metadata.ObjectiveKey) - if sloClass == "" { - sloClass = metrics.SLOClassNone + sloClass := metrics.SLOClassNone + if req := fi.originalRequest.InferenceRequest(); req != nil { + sloClass = request.GetHeader(req.Headers, metadata.ObjectiveKey) + if sloClass == "" { + sloClass = metrics.SLOClassNone + } } metrics.RecordFlowControlSLORequestQueueDuration( sloClass, outcomeStr, fi.originalRequest.InferencePoolName(), @@ -180,14 +183,6 @@ func (fi *FlowItem) finalizeInternal(outcome types.QueueOutcome, err error) { close(fi.done) } -func extractHeader(req flowcontrol.FlowControlRequest, name string) string { - infReq := req.InferenceRequest() - if infReq == nil || infReq.Headers == nil { - return "" - } - return request.GetHeader(infReq.Headers, name) -} - // inferOutcome determines the correct QueueOutcome and Error based on the cause of finalization and whether the item // was already admitted to a queue. func inferOutcome(cause error, isQueued bool) (types.QueueOutcome, error) {