feature: add slo-based deadline ordering policy (kubernetes-sigs#2531)

loicmarchal · web-flow · commit e7a78e4f838f · 2026-03-14T05:05:35.000+05:30
* feature: add slo-based deadline ordering policy

* Move GetHeader function to utils package

* Update plugin guide
diff --git a/cmd/epp/runner/runner.go b/cmd/epp/runner/runner.go
@@ -460,6 +460,7 @@ func (r *Runner) registerInTreePlugins() {
 	fwkplugin.Register(fairness.RoundRobinFairnessPolicyType, fairness.RoundRobinFairnessPolicyFactory)
 	fwkplugin.Register(ordering.FCFSOrderingPolicyType, ordering.FCFSOrderingPolicyFactory)
 	fwkplugin.Register(ordering.EDFOrderingPolicyType, ordering.EDFOrderingPolicyFactory)
+	fwkplugin.Register(ordering.SLODeadlineOrderingPolicyType, ordering.SLODeadlineOrderingPolicyFactory)
 	// Latency predictor plugins
 	fwkplugin.Register(predictedlatency.PredictedLatencyPluginType, predictedlatency.PredictedLatencyFactory)
 	// register filter for test purpose only (used in conformance tests)
diff --git a/pkg/common/request/headers.go b/pkg/common/request/headers.go
@@ -16,6 +16,22 @@ limitations under the License.
 
 package request
 
+import "strings"
+
 const (
 	RequestIdHeaderKey = "x-request-id"
 )
+
+// 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 {
+		return v
+	}
+	lower := strings.ToLower(key)
+	for k, v := range headers {
+		if strings.ToLower(k) == lower {
+			return v
+		}
+	}
+	return ""
+}
diff --git a/pkg/common/request/headers_test.go b/pkg/common/request/headers_test.go
@@ -0,0 +1,31 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package request
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetHeader(t *testing.T) {
+	headers := map[string]string{"X-SLO-TTFT-MS": "42", "Other": "x"}
+	assert.Equal(t, "42", GetHeader(headers, "X-SLO-TTFT-MS"))
+	assert.Equal(t, "42", GetHeader(headers, "x-slo-ttft-ms"))
+	assert.Equal(t, "", GetHeader(headers, "missing"))
+	assert.Equal(t, "", GetHeader(nil, "k"))
+}
diff --git a/pkg/epp/flowcontrol/framework/plugins/ordering/doc.go b/pkg/epp/flowcontrol/framework/plugins/ordering/doc.go
@@ -51,4 +51,9 @@ limitations under the License.
 //   - EDF ("Earliest Deadline First") ("edf-ordering-policy"): Orders requests by their absolute deadline
 //     (EnqueueTime + TTL).
 //     This maximizes the number of requests served before their deadlines expire.
+//
+//   - SLO Deadline ("slo-deadline-ordering-policy"): Orders requests by an SLO-based (service level objective) deadline
+//     computed as ReceivedTimestamp + x-slo-ttft-ms header (interpreted as milliseconds).
+//     Requests without a valid header are scheduled after SLO-bound requests.
+//     This maximizes the number of requests served before the deadlines computed on the defined SLO expire.
 package ordering
diff --git a/pkg/epp/flowcontrol/framework/plugins/ordering/slo_deadline.go b/pkg/epp/flowcontrol/framework/plugins/ordering/slo_deadline.go
@@ -0,0 +1,135 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package ordering
+
+import (
+	"encoding/json"
+	"strconv"
+	"strings"
+	"time"
+
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/common/request"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/flowcontrol"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/plugin"
+)
+
+const (
+	// SLODeadlineOrderingPolicyType orders requests by an SLO-based deadline
+	//
+	// It selects the request with the earliest SLO-based deadline, computed as `ReceivedTimestamp() + x-slo-ttft-ms header (interpreted as milliseconds)`.
+	// Requests without a valid x-slo-ttft-ms header are treated as having no deadline and are scheduled after SLO-bound requests,
+	// with FCFS as a tie-breaker.
+	SLODeadlineOrderingPolicyType = "slo-deadline-ordering-policy"
+
+	// sloTtftHeader is the request header name for SLO time-to-first-token in milliseconds.
+	sloTtftHeader = "x-slo-ttft-ms"
+)
+
+func SLODeadlineOrderingPolicyFactory(name string, _ json.RawMessage, _ plugin.Handle) (plugin.Plugin, error) {
+	return newSLODeadlinePolicy().withName(name), nil
+}
+
+type sloDeadlinePolicy struct {
+	name string
+}
+
+var _ flowcontrol.OrderingPolicy = &sloDeadlinePolicy{}
+
+func newSLODeadlinePolicy() *sloDeadlinePolicy {
+	return &sloDeadlinePolicy{
+		name: SLODeadlineOrderingPolicyType,
+	}
+}
+
+func (p *sloDeadlinePolicy) withName(name string) *sloDeadlinePolicy {
+	if name != "" {
+		p.name = name
+	}
+	return p
+}
+
+func (p *sloDeadlinePolicy) Name() string {
+	return p.name
+}
+
+// RequiredQueueCapabilities returns the queue capabilities required by this policy.
+func (p *sloDeadlinePolicy) RequiredQueueCapabilities() []flowcontrol.QueueCapability {
+	return []flowcontrol.QueueCapability{flowcontrol.CapabilityPriorityConfigurable}
+}
+
+func (p *sloDeadlinePolicy) TypedName() plugin.TypedName {
+	return plugin.TypedName{
+		Type: SLODeadlineOrderingPolicyType,
+		Name: p.name,
+	}
+}
+
+var sloMaxDeadlineTime = time.Unix(0, 1<<63-1)
+
+// calculateSLODeadline computes the SLO-based deadline for a request: ReceivedTimestamp + x-slo-ttft-ms (ms).
+// The header is read from the InferenceRequest()'s headers. If the header is missing, empty, or invalid,
+// the request is assigned a far-future deadline so it sorts after SLO-bound requests.
+func calculateSLODeadline(item flowcontrol.QueueItemAccessor) time.Time {
+	req := item.OriginalRequest()
+	if req == nil {
+		return sloMaxDeadlineTime
+	}
+	infReq := req.InferenceRequest()
+	if infReq == nil || infReq.Headers == nil {
+		return sloMaxDeadlineTime
+	}
+	sloTtft := request.GetHeader(infReq.Headers, sloTtftHeader)
+	if sloTtft == "" {
+		return sloMaxDeadlineTime
+	}
+	ms, err := strconv.ParseInt(strings.TrimSpace(sloTtft), 10, 64)
+	if err != nil || ms < 0 {
+		return sloMaxDeadlineTime
+	}
+	return req.ReceivedTimestamp().Add(time.Duration(ms) * time.Millisecond)
+}
+
+// Less returns true if item 'a' should be dispatched before item 'b'.
+// It orders by SLO deadline (earliest first), using FCFS as a tie-breaker.
+func (p *sloDeadlinePolicy) Less(a, b flowcontrol.QueueItemAccessor) bool {
+	if a == nil && b == nil {
+		return false
+	}
+	if a == nil {
+		return false
+	}
+	if b == nil {
+		return true
+	}
+	deadlineA := calculateSLODeadline(a)
+	deadlineB := calculateSLODeadline(b)
+	if !deadlineA.Equal(deadlineB) {
+		return deadlineA.Before(deadlineB)
+	}
+	reqA := a.OriginalRequest()
+	reqB := b.OriginalRequest()
+	if reqA == nil && reqB == nil {
+		return false
+	}
+	if reqA == nil {
+		return false
+	}
+	if reqB == nil {
+		return true
+	}
+	return reqA.ReceivedTimestamp().Before(reqB.ReceivedTimestamp())
+}
diff --git a/pkg/epp/flowcontrol/framework/plugins/ordering/slo_deadline_test.go b/pkg/epp/flowcontrol/framework/plugins/ordering/slo_deadline_test.go
@@ -0,0 +1,145 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package ordering
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/flowcontrol"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/flowcontrol/mocks"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/scheduling"
+)
+
+func TestSLODeadlinePolicy_Name(t *testing.T) {
+	t.Parallel()
+	policy := newSLODeadlinePolicy()
+	assert.Equal(t, SLODeadlineOrderingPolicyType, policy.Name())
+}
+
+func TestSLODeadlinePolicy_WithName(t *testing.T) {
+	t.Parallel()
+	policy := newSLODeadlinePolicy().withName("test-name")
+	assert.Equal(t, "test-name", policy.Name())
+}
+
+func TestSLODeadlinePolicy_RequiredQueueCapabilities(t *testing.T) {
+	t.Parallel()
+	policy := newSLODeadlinePolicy()
+	caps := policy.RequiredQueueCapabilities()
+	require.Len(t, caps, 1)
+	assert.Equal(t, flowcontrol.CapabilityPriorityConfigurable, caps[0])
+}
+
+// makeSLOItem builds a QueueItemAccessor with the given SLO header and received time.
+func makeSLOItem(id string, received time.Time, sloTTFTMs string) flowcontrol.QueueItemAccessor {
+	req := mocks.NewMockFlowControlRequest(10, id, testFlowKey)
+	req.ReceivedTimestampV = received
+	req.InferenceRequestV = &scheduling.LLMRequest{Headers: map[string]string{sloTtftHeader: sloTTFTMs}}
+	return &mocks.MockQueueItemAccessor{
+		EffectiveTTLV:    0,
+		OriginalRequestV: req,
+	}
+}
+
+func TestSLODeadline_Less(t *testing.T) {
+	t.Parallel()
+	policy := newSLODeadlinePolicy()
+
+	now := time.Now()
+
+	// A: received now, 100ms SLO → deadline now+100ms
+	itemA := makeSLOItem("a", now, "100")
+	// B: received now, 50ms SLO → deadline now+50ms (earlier)
+	itemB := makeSLOItem("b", now, "50")
+	// C: received now+20ms, 50ms SLO → deadline now+20ms+50ms = now+70ms (after B but earlier than A)
+	itemC := makeSLOItem("c", now.Add(20*time.Millisecond), "50")
+	// D: no header → far-future deadline
+	reqD := mocks.NewMockFlowControlRequest(10, "d", testFlowKey)
+	reqD.ReceivedTimestampV = now
+	reqD.InferenceRequestV = &scheduling.LLMRequest{Headers: map[string]string{}}
+	itemD := &mocks.MockQueueItemAccessor{EffectiveTTLV: 0, OriginalRequestV: reqD}
+	// E: same deadline as B (received 1s earlier + 1050ms SLO = now+50ms), earlier ReceivedTimestamp → wins tie-breaker
+	itemE := makeSLOItem("e", now.Add(-time.Second), "1050")
+
+	testCases := []struct {
+		name     string
+		a        flowcontrol.QueueItemAccessor
+		b        flowcontrol.QueueItemAccessor
+		expected bool
+	}{
+		{"earlier SLO deadline first (B before A)", itemB, itemA, true},
+		{"later SLO deadline after (A after B)", itemA, itemB, false},
+		{"received later but earlier deadline (C before A)", itemC, itemA, true},
+		{"SLO-bound before no-header (A before D)", itemA, itemD, true},
+		{"no-header after SLO-bound (D after A)", itemD, itemA, false},
+		{"same deadline: earlier ReceivedTimestamp first (E before B)", itemE, itemB, true},
+		{"same deadline: later ReceivedTimestamp after (B after E)", itemB, itemE, false},
+		{"a is nil → b wins", nil, itemA, false},
+		{"b is nil → a wins", itemA, nil, true},
+		{"both nil → false", nil, nil, false},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			assert.Equal(t, tc.expected, policy.Less(tc.a, tc.b))
+		})
+	}
+}
+
+func TestCalculateSLODeadline(t *testing.T) {
+	t.Parallel()
+
+	now := time.Now()
+
+	// Valid header
+	reqValid := mocks.NewMockFlowControlRequest(1, "valid", testFlowKey)
+	reqValid.ReceivedTimestampV = now
+	reqValid.InferenceRequestV = &scheduling.LLMRequest{Headers: map[string]string{sloTtftHeader: "200"}}
+	accValid := &mocks.MockQueueItemAccessor{OriginalRequestV: reqValid}
+	deadline := calculateSLODeadline(accValid)
+	assert.Equal(t, now.Add(200*time.Millisecond), deadline)
+
+	// Missing header
+	reqNoHeader := mocks.NewMockFlowControlRequest(2, "no", testFlowKey)
+	reqNoHeader.InferenceRequestV = &scheduling.LLMRequest{Headers: map[string]string{}}
+	accNoHeader := &mocks.MockQueueItemAccessor{OriginalRequestV: reqNoHeader}
+	assert.Equal(t, sloMaxDeadlineTime, calculateSLODeadline(accNoHeader))
+
+	reqNoHeader.InferenceRequestV = &scheduling.LLMRequest{Headers: map[string]string{"x-some-header": "200"}}
+	accNoHeader = &mocks.MockQueueItemAccessor{OriginalRequestV: reqNoHeader}
+	assert.Equal(t, sloMaxDeadlineTime, calculateSLODeadline(accNoHeader))
+
+	// Invalid value
+	reqInvalid := mocks.NewMockFlowControlRequest(3, "inv", testFlowKey)
+	reqInvalid.InferenceRequestV = &scheduling.LLMRequest{Headers: map[string]string{sloTtftHeader: "x"}}
+	accInvalid := &mocks.MockQueueItemAccessor{OriginalRequestV: reqInvalid}
+	assert.Equal(t, sloMaxDeadlineTime, calculateSLODeadline(accInvalid))
+
+	// Nil OriginalRequest
+	accNilReq := &mocks.MockQueueItemAccessor{OriginalRequestV: nil}
+	assert.Equal(t, sloMaxDeadlineTime, calculateSLODeadline(accNilReq))
+
+	// Nil InferenceRequest
+	reqNilInfReq := mocks.NewMockFlowControlRequest(4, "no-inf-req", testFlowKey)
+	accNoInfReq := &mocks.MockQueueItemAccessor{OriginalRequestV: reqNilInfReq}
+	assert.Equal(t, sloMaxDeadlineTime, calculateSLODeadline(accNoInfReq))
+}
diff --git a/site-src/guides/epp-configuration/config-text.md b/site-src/guides/epp-configuration/config-text.md
@@ -312,6 +312,13 @@ An Ordering Policy that implements Earliest Deadline First. It prioritizes reque
 - *Type*: edf-ordering-policy
 - *Parameters*: none
 
+#### SLODeadlineOrderingPolicy
+
+An Ordering Policy that orders requests by an SLO-based deadline, computed from the time the request is received by the server. It prioritizes requests with the earliest such deadline.
+
+- *Type*: slo-deadline-ordering-policy
+- *Parameters*: none
+
 ## Scheduling Profiles
 
 The `schedulingProfiles` section defines the set of scheduling profiles that can be used in scheduling
