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 2bcb2293be..a738fd9edc 100644 --- a/pkg/epp/flowcontrol/controller/controller.go +++ b/pkg/epp/flowcontrol/controller/controller.go @@ -31,7 +31,9 @@ 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/metadata" "github.com/llm-d/llm-d-router/pkg/epp/metrics" ) @@ -209,21 +211,36 @@ func (fc *FlowController) EnqueueAndWait( flowKey := req.FlowKey() priority := strconv.Itoa(flowKey.Priority) reqBytes := req.ByteSize() + 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 + } + 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). diff --git a/pkg/epp/flowcontrol/controller/internal/item.go b/pkg/epp/flowcontrol/controller/internal/item.go index 9fb219a373..2f7df49739 100644 --- a/pkg/epp/flowcontrol/controller/internal/item.go +++ b/pkg/epp/flowcontrol/controller/internal/item.go @@ -26,7 +26,9 @@ 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/metadata" "github.com/llm-d/llm-d-router/pkg/epp/metrics" ) @@ -159,12 +161,24 @@ 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.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(), + duration) + fi.done <- finalState close(fi.done) } 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..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 @@ -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) @@ -123,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} @@ -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..a12c1ca095 100644 --- a/pkg/epp/metrics/metrics.go +++ b/pkg/epp/metrics/metrics.go @@ -312,6 +312,27 @@ var ( append([]string{"fairness_id", "priority", "outcome", "inference_pool"}, modelLabels...), ) + 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_epp_flow_control_dispatch_cycle_duration_seconds instead. // Tracked in: https://github.com/llm-d/llm-d-inference-scheduler/issues/1070 flowControlDispatchCycleDuration = prometheus.NewHistogramVec( @@ -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,25 @@ func RecordFlowControlRequestQueueDuration( ).Observe(duration.Seconds()) } +// 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"). +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 +866,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..c146561f10 100644 --- a/pkg/epp/metrics/metrics_test.go +++ b/pkg/epp/metrics/metrics_test.go @@ -1259,17 +1259,143 @@ func TestFlowControlQueueDurationMetric(t *testing.T) { } } +func TestFlowControlSLOIncomingRequestsTotalMetric(t *testing.T) { + Reset() + + const pool = "pool-1" + + RecordFlowControlSLOIncomingRequest("realtime", pool) + RecordFlowControlSLOIncomingRequest("realtime", pool) + RecordFlowControlSLOIncomingRequest("interactive", pool) + RecordFlowControlSLOIncomingRequest(SLOClassNone, "pool-2") + + testCases := []struct { + name string + labels prometheus.Labels + expectCount float64 + }{ + { + name: "realtime, pool-1", + labels: prometheus.Labels{"slo_class": "realtime", "inference_pool": pool}, + expectCount: 2, + }, + { + name: "interactive, pool-1", + labels: prometheus.Labels{"slo_class": "interactive", "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: "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: "batch", 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: "realtime, dispatched", + labels: prometheus.Labels{ + "slo_class": "realtime", + "outcome": "Dispatched", + "inference_pool": pool, + }, + expectCount: 2, + expectSum: 0.02, // 0.005 + 0.015 + }, + { + name: "interactive, dispatched", + labels: prometheus.Labels{ + "slo_class": "interactive", + "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: "batch, dispatched", + labels: prometheus.Labels{ + "slo_class": "batch", + "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 +1404,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 +1418,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 +1433,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) 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 {