Skip to content

Commit 7df5d3d

Browse files
nirrozenbaumnirrozenbaum
authored
added cycle state to pick & process results in profile handler (#1040)
* added cycle state to pick + process results of profile handler Signed-off-by: Nir Rozenbaum <[email protected]> * wildcard Signed-off-by: Nir Rozenbaum <[email protected]> --------- Signed-off-by: Nir Rozenbaum <[email protected]>
1 parent de941fd commit 7df5d3d

20 files changed

+44
-43
lines changed

‎conformance/testing-epp/plugins/filter/filter_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -117,7 +117,7 @@ func TestFilter(t *testing.T) {
117117

118118
for _, test := range tests {
119119
t.Run(test.name, func(t *testing.T) {
120-
got := test.filter.Filter(context.Background(), test.req, types.NewCycleState(), test.input)
120+
got := test.filter.Filter(context.Background(), types.NewCycleState(), test.req, test.input)
121121

122122
if diff := cmp.Diff(test.output, got); diff != "" {
123123
t.Errorf("Unexpected output (-want +got): %v", diff)

‎conformance/testing-epp/plugins/filter/request_header_based_filter.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ func (f *HeaderBasedTestingFilter) Type() string {
5050
}
5151

5252
// Filter selects pods that match the IP addresses specified in the request header.
53-
func (f *HeaderBasedTestingFilter) Filter(_ context.Context, request *types.LLMRequest, _ *types.CycleState, pods []types.Pod) []types.Pod {
53+
func (f *HeaderBasedTestingFilter) Filter(_ context.Context, _ *types.CycleState, request *types.LLMRequest, pods []types.Pod) []types.Pod {
5454
headerValue, ok := request.Headers[headerTestEppEndPointSelectionKey]
5555
if !ok || headerValue == "" {
5656
return []types.Pod{}

‎pkg/epp/common/config/configloader_test.go‎

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -468,7 +468,7 @@ func (f *test1) Type() string {
468468
}
469469

470470
// Filter filters out pods that doesn't meet the filter criteria.
471-
func (f *test1) Filter(ctx context.Context, request *types.LLMRequest, cycleState *types.CycleState, pods []types.Pod) []types.Pod {
471+
func (f *test1) Filter(_ context.Context, _ *types.CycleState, _ *types.LLMRequest, pods []types.Pod) []types.Pod {
472472
return pods
473473
}
474474

@@ -482,12 +482,11 @@ func (f *test2) Type() string {
482482
return test2Type
483483
}
484484

485-
func (m *test2) Score(ctx context.Context, request *types.LLMRequest, cycleState *types.CycleState, pods []types.Pod) map[types.Pod]float64 {
485+
func (m *test2) Score(_ context.Context, _ *types.CycleState, _ *types.LLMRequest, _ []types.Pod) map[types.Pod]float64 {
486486
return map[types.Pod]float64{}
487487
}
488488

489-
func (m *test2) PostCycle(ctx context.Context, cycleState *types.CycleState, res *types.ProfileRunResult) {
490-
}
489+
func (m *test2) PostCycle(_ context.Context, _ *types.CycleState, _ *types.ProfileRunResult) {}
491490

492491
// compile-time type validation
493492
var _ framework.Picker = &testPicker{}
@@ -498,7 +497,7 @@ func (p *testPicker) Type() string {
498497
return testPickerType
499498
}
500499

501-
func (p *testPicker) Pick(ctx context.Context, cycleState *types.CycleState, scoredPods []*types.ScoredPod) *types.ProfileRunResult {
500+
func (p *testPicker) Pick(_ context.Context, _ *types.CycleState, _ []*types.ScoredPod) *types.ProfileRunResult {
502501
return nil
503502
}
504503

@@ -511,11 +510,11 @@ func (p *testProfileHandler) Type() string {
511510
return testProfileHandlerType
512511
}
513512

514-
func (p *testProfileHandler) Pick(ctx context.Context, request *types.LLMRequest, profiles map[string]*framework.SchedulerProfile, executionResults map[string]*types.ProfileRunResult) map[string]*framework.SchedulerProfile {
513+
func (p *testProfileHandler) Pick(_ context.Context, _ *types.CycleState, _ *types.LLMRequest, _ map[string]*framework.SchedulerProfile, _ map[string]*types.ProfileRunResult) map[string]*framework.SchedulerProfile {
515514
return nil
516515
}
517516

518-
func (p *testProfileHandler) ProcessResults(ctx context.Context, request *types.LLMRequest, profileResults map[string]*types.ProfileRunResult) (*types.SchedulingResult, error) {
517+
func (p *testProfileHandler) ProcessResults(_ context.Context, _ *types.CycleState, _ *types.LLMRequest, _ map[string]*types.ProfileRunResult) (*types.SchedulingResult, error) {
519518
return nil, nil
520519
}
521520

‎pkg/epp/scheduling/framework/plugins.go‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -38,26 +38,28 @@ type ProfileHandler interface {
3838
plugins.Plugin
3939
// Pick selects the SchedulingProfiles to run from a list of candidate profiles, while taking into consideration the request properties
4040
// and the previously executed SchedluderProfile cycles along with their results.
41-
Pick(ctx context.Context, request *types.LLMRequest, profiles map[string]*SchedulerProfile, profileResults map[string]*types.ProfileRunResult) map[string]*SchedulerProfile
41+
Pick(ctx context.Context, cycleState *types.CycleState, request *types.LLMRequest, profiles map[string]*SchedulerProfile,
42+
profileResults map[string]*types.ProfileRunResult) map[string]*SchedulerProfile
4243

4344
// ProcessResults handles the outcome of the profile runs after all profiles ran.
4445
// It may aggregate results, log test profile outputs, or apply custom logic. It specifies in the SchedulingResult the
4546
// key of the primary profile that should be used to get the request selected destination.
4647
// When a profile run fails, its result in the profileResults map is nil.
47-
ProcessResults(ctx context.Context, request *types.LLMRequest, profileResults map[string]*types.ProfileRunResult) (*types.SchedulingResult, error)
48+
ProcessResults(ctx context.Context, cycleState *types.CycleState, request *types.LLMRequest,
49+
profileResults map[string]*types.ProfileRunResult) (*types.SchedulingResult, error)
4850
}
4951

5052
// Filter defines the interface for filtering a list of pods based on context.
5153
type Filter interface {
5254
plugins.Plugin
53-
Filter(ctx context.Context, request *types.LLMRequest, cycleState *types.CycleState, pods []types.Pod) []types.Pod
55+
Filter(ctx context.Context, cycleState *types.CycleState, request *types.LLMRequest, pods []types.Pod) []types.Pod
5456
}
5557

5658
// Scorer defines the interface for scoring a list of pods based on context.
5759
// Scorers must score pods with a value within the range of [0,1] where 1 is the highest score.
5860
type Scorer interface {
5961
plugins.Plugin
60-
Score(ctx context.Context, request *types.LLMRequest, cycleState *types.CycleState, pods []types.Pod) map[types.Pod]float64
62+
Score(ctx context.Context, cycleState *types.CycleState, request *types.LLMRequest, pods []types.Pod) map[types.Pod]float64
6163
}
6264

6365
// Picker picks the final pod(s) to send the request to.

‎pkg/epp/scheduling/framework/plugins/filter/decision_tree_filter.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -56,9 +56,9 @@ func (f *DecisionTreeFilter) Type() string {
5656
}
5757

5858
// Filter filters out pods that doesn't meet the filter criteria.
59-
func (f *DecisionTreeFilter) Filter(ctx context.Context, request *types.LLMRequest, cycleState *types.CycleState, pods []types.Pod) []types.Pod {
59+
func (f *DecisionTreeFilter) Filter(ctx context.Context, cycleState *types.CycleState, request *types.LLMRequest, pods []types.Pod) []types.Pod {
6060
loggerTrace := log.FromContext(ctx).V(logutil.TRACE)
61-
filteredPod := f.Current.Filter(ctx, request, cycleState, pods)
61+
filteredPod := f.Current.Filter(ctx, cycleState, request, pods)
6262

6363
next := f.NextOnSuccessOrFailure
6464
if len(filteredPod) > 0 {
@@ -71,7 +71,7 @@ func (f *DecisionTreeFilter) Filter(ctx context.Context, request *types.LLMReque
7171
}
7272
loggerTrace.Info("Filter succeeded", "filter", f.Type(), "next", next.Type(), "filteredPodCount", len(filteredPod))
7373
// On success, pass the filtered result to the next filter.
74-
return next.Filter(ctx, request, cycleState, filteredPod)
74+
return next.Filter(ctx, cycleState, request, filteredPod)
7575
} else {
7676
if f.NextOnFailure == nil && f.NextOnSuccessOrFailure == nil {
7777
// No succeeding filters to run, return.
@@ -82,6 +82,6 @@ func (f *DecisionTreeFilter) Filter(ctx context.Context, request *types.LLMReque
8282
}
8383
loggerTrace.Info("Filter failed", "filter", f.Type(), "next", next.Type())
8484
// On failure, pass the initial set of pods to the next filter.
85-
return next.Filter(ctx, request, cycleState, pods)
85+
return next.Filter(ctx, cycleState, request, pods)
8686
}
8787
}

‎pkg/epp/scheduling/framework/plugins/filter/filter_test.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ func (f *filterAll) Type() string {
3939
return "filter-all"
4040
}
4141

42-
func (f *filterAll) Filter(_ context.Context, _ *types.LLMRequest, _ *types.CycleState, pods []types.Pod) []types.Pod {
42+
func (f *filterAll) Filter(_ context.Context, _ *types.CycleState, _ *types.LLMRequest, pods []types.Pod) []types.Pod {
4343
return []types.Pod{}
4444
}
4545

@@ -138,7 +138,7 @@ func TestFilter(t *testing.T) {
138138

139139
for _, test := range tests {
140140
t.Run(test.name, func(t *testing.T) {
141-
got := test.filter.Filter(context.Background(), test.req, types.NewCycleState(), test.input)
141+
got := test.filter.Filter(context.Background(), types.NewCycleState(), test.req, test.input)
142142

143143
if diff := cmp.Diff(test.output, got); diff != "" {
144144
t.Errorf("Unexpected output (-want +got): %v", diff)
@@ -206,7 +206,7 @@ func TestLoRASoftAffinityDistribution(t *testing.T) {
206206
LoraAffinityFilter := NewLoraAffinityFilter(config.Conf.LoraAffinityThreshold)
207207

208208
for range numIterations {
209-
result := LoraAffinityFilter.Filter(context.Background(), req, types.NewCycleState(), pods)
209+
result := LoraAffinityFilter.Filter(context.Background(), types.NewCycleState(), req, pods)
210210

211211
// Check which type of pod was returned
212212
if len(result) != 1 {

‎pkg/epp/scheduling/framework/plugins/filter/least_kvcache_filter.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ func (f *LeastKVCacheFilter) Type() string {
5656
}
5757

5858
// Filter filters out pods that doesn't meet the filter criteria.
59-
func (f *LeastKVCacheFilter) Filter(_ context.Context, _ *types.LLMRequest, _ *types.CycleState, pods []types.Pod) []types.Pod {
59+
func (f *LeastKVCacheFilter) Filter(_ context.Context, _ *types.CycleState, _ *types.LLMRequest, pods []types.Pod) []types.Pod {
6060
filteredPods := []types.Pod{}
6161

6262
min := math.MaxFloat64

‎pkg/epp/scheduling/framework/plugins/filter/least_queue_filter.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ func (f *LeastQueueFilter) Type() string {
5656
}
5757

5858
// Filter filters out pods that doesn't meet the filter criteria.
59-
func (f *LeastQueueFilter) Filter(_ context.Context, _ *types.LLMRequest, _ *types.CycleState, pods []types.Pod) []types.Pod {
59+
func (f *LeastQueueFilter) Filter(_ context.Context, _ *types.CycleState, _ *types.LLMRequest, pods []types.Pod) []types.Pod {
6060
filteredPods := []types.Pod{}
6161

6262
min := math.MaxInt

‎pkg/epp/scheduling/framework/plugins/filter/lora_affinity_filter.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,7 @@ func (f *LoraAffinityFilter) Type() string {
7373
}
7474

7575
// Filter filters out pods that doesn't meet the filter criteria.
76-
func (f *LoraAffinityFilter) Filter(_ context.Context, request *types.LLMRequest, _ *types.CycleState, pods []types.Pod) []types.Pod {
76+
func (f *LoraAffinityFilter) Filter(_ context.Context, _ *types.CycleState, request *types.LLMRequest, pods []types.Pod) []types.Pod {
7777
// Pre-allocate slices with estimated capacity
7878
filtered_affinity := make([]types.Pod, 0, len(pods))
7979
filtered_available := make([]types.Pod, 0, len(pods))

‎pkg/epp/scheduling/framework/plugins/filter/low_queue_filter.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,7 @@ func (f *LowQueueFilter) Type() string {
6767
}
6868

6969
// Filter filters out pods that doesn't meet the filter criteria.
70-
func (f *LowQueueFilter) Filter(_ context.Context, _ *types.LLMRequest, _ *types.CycleState, pods []types.Pod) []types.Pod {
70+
func (f *LowQueueFilter) Filter(_ context.Context, _ *types.CycleState, _ *types.LLMRequest, pods []types.Pod) []types.Pod {
7171
filteredPods := []types.Pod{}
7272

7373
for _, pod := range pods {

0 commit comments

Comments
 (0)