merge has capacity filter with sheddable filter. (#809)

* merge has capacity filter with sheddable filter.

has capacity only use was for sheddable requests (passthrough for critical ones).

Signed-off-by: Nir Rozenbaum <[email protected]>

* Update pkg/epp/scheduling/plugins/filter/filter_test.go

Co-authored-by: Cong Liu <[email protected]>

---------

Signed-off-by: Nir Rozenbaum <[email protected]>
Co-authored-by: Cong Liu <[email protected]>

Author: nirrozenbaum

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

@@ -131,8 +131,9 @@ func TestFilter(t *testing.T) {
 			},
 		},
 		{
-			name:   "lowQueueAndLessThanKVCacheThresholdPredicate",
-			filter: &HasCapacityFilter{queueThreshold: 0, kvCacheThreshold: 0.8},
+			name:   "SheddableCapacityFilter, sheddable request",
+			req:    &types.LLMRequest{Critical: false},
+			filter: &SheddableCapacityFilter{queueThreshold: 0, kvCacheThreshold: 0.8},
 			input: []types.Pod{
 				&types.PodMetrics{
 					// This pod should be returned.

pkg/epp/scheduling/plugins/filter/has_capacity_filter.go renamed to pkg/epp/scheduling/plugins/filter/sheddable_capacity_filter.go

@@ -23,29 +23,33 @@ import (
 )
 
 // compile-time type validation
-var _ plugins.Filter = &HasCapacityFilter{}
+var _ plugins.Filter = &SheddableCapacityFilter{}
 
-// NewHasCapacityFilter returns a new HasCapacityFilter.
-func NewHasCapacityFilter() *HasCapacityFilter {
-	return &HasCapacityFilter{
+// NewSheddableCapacityFilter returns a new SheddableCapacityFilter.
+func NewSheddableCapacityFilter() *SheddableCapacityFilter {
+	return &SheddableCapacityFilter{
 		queueThreshold:   config.Conf.QueueThresholdCritical,
 		kvCacheThreshold: config.Conf.KVCacheThreshold,
 	}
 }
 
-// HasCapacityFilter filters only pods that has capacity for sheddable requests.
-type HasCapacityFilter struct {
+// SheddableCapacityFilter filters only pods that has capacity for sheddable requests.
+type SheddableCapacityFilter struct {
 	queueThreshold   int
 	kvCacheThreshold float64
 }
 
 // Name returns the name of the filter.
-func (f *HasCapacityFilter) Name() string {
-	return "has-capacity"
+func (f *SheddableCapacityFilter) Name() string {
+	return "sheddable-capacity"
 }
 
 // Filter filters out pods that doesn't meet the filter criteria.
-func (f *HasCapacityFilter) Filter(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
+func (f *SheddableCapacityFilter) Filter(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
+	if ctx.Req.Critical {
+		return pods // // Allow all pods to passthrough if the request is critical, even if all pods reach their capacity.
+	}
+
 	filteredPods := []types.Pod{}
 
 	for _, pod := range pods {

pkg/epp/scheduling/plugins/filter/sheddable_request_filter.go

@@ -66,7 +66,7 @@ func NewScheduler(datastore Datastore) *Scheduler {
 
 	defaultConfig := &SchedulerConfig{
 		preSchedulePlugins:  []plugins.PreSchedule{},
-		filters:             []plugins.Filter{filter.NewSheddableRequestFilter(), lowLatencyFilter},
+		filters:             []plugins.Filter{filter.NewSheddableCapacityFilter(), lowLatencyFilter},
 		scorers:             map[plugins.Scorer]int{},
 		picker:              &picker.RandomPicker{},
 		postSchedulePlugins: []plugins.PostSchedule{},