Clean up filters (#802)

Author: liu-cong

pkg/epp/scheduling/config.go

@@ -16,7 +16,11 @@ limitations under the License.
 
 package scheduling
 
-import "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
+import (
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins/filter"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins/picker"
+)
 
 // NewSchedulerConfig creates a new SchedulerConfig object with the given plugins.
 func NewSchedulerConfig(preSchedulePlugins []plugins.PreSchedule, filters []plugins.Filter, scorers map[plugins.Scorer]int,
@@ -39,16 +43,14 @@ type SchedulerConfig struct {
 	postSchedulePlugins []plugins.PostSchedule
 }
 
-var defPlugin = &defaultPlugin{}
-
 // When the scheduler is initialized with NewScheduler function, this config will be used as default.
 // it's possible to call NewSchedulerWithConfig to pass a different argument.
 
 // For build time plugins changes, it's recommended to change the defaultConfig variable in this file.
 var defaultConfig = &SchedulerConfig{
 	preSchedulePlugins:  []plugins.PreSchedule{},
-	filters:             []plugins.Filter{defPlugin},
+	filters:             []plugins.Filter{&filter.SheddableRequestFilter{}, filter.LowLatencyFilter},
 	scorers:             map[plugins.Scorer]int{},
-	picker:              defPlugin,
+	picker:              &picker.RandomPicker{},
 	postSchedulePlugins: []plugins.PostSchedule{},
 }

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

@@ -276,3 +276,41 @@ func (pp podPredicate) and(another podPredicate) podPredicate {
 		return pp(req, pod) && another(req, pod)
 	}
 }
+
+var LowLatencyFilter = &DecisionTreeFilter{
+	Current: LowQueueFilter,
+	NextOnSuccess: &DecisionTreeFilter{
+		Current: LoRAAffinityFilter,
+		NextOnSuccessOrFailure: &DecisionTreeFilter{
+			Current: LeastQueueFilter,
+			NextOnSuccessOrFailure: &DecisionTreeFilter{
+				Current: LeastKVCacheFilter,
+			},
+		},
+	},
+	NextOnFailure: &DecisionTreeFilter{
+		Current: LeastQueueFilter,
+		NextOnSuccessOrFailure: &DecisionTreeFilter{
+			Current: LoRAAffinityFilter,
+			NextOnSuccessOrFailure: &DecisionTreeFilter{
+				Current: LeastKVCacheFilter,
+			},
+		},
+	},
+}
+
+type SheddableRequestFilter struct{}
+
+func (p *SheddableRequestFilter) Name() string {
+	return "SheddableRequestFilter"
+}
+
+func (p *SheddableRequestFilter) Filter(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
+	if ctx.Req.Critical {
+		// Allow all pods to pass through if the request is critical, even if all pods reach their capacity.
+		return pods
+	}
+
+	// Only allow pods that have enough capacity to handle the request.
+	return HasCapacityFilter.Filter(ctx, pods)
+}

pkg/epp/scheduling/scheduler.go

@@ -26,47 +26,11 @@ import (
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins/filter"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins/picker"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
 	errutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/error"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
 
-var (
-	lowLatencyFilter = &filter.DecisionTreeFilter{
-		Current: filter.LowQueueFilter,
-		NextOnSuccess: &filter.DecisionTreeFilter{
-			Current: filter.LoRAAffinityFilter,
-			NextOnSuccessOrFailure: &filter.DecisionTreeFilter{
-				Current: filter.LeastQueueFilter,
-				NextOnSuccessOrFailure: &filter.DecisionTreeFilter{
-					Current: filter.LeastKVCacheFilter,
-				},
-			},
-		},
-		NextOnFailure: &filter.DecisionTreeFilter{
-			Current: filter.LeastQueueFilter,
-			NextOnSuccessOrFailure: &filter.DecisionTreeFilter{
-				Current: filter.LoRAAffinityFilter,
-				NextOnSuccessOrFailure: &filter.DecisionTreeFilter{
-					Current: filter.LeastKVCacheFilter,
-				},
-			},
-		},
-	}
-
-	sheddableRequestFilter = &filter.DecisionTreeFilter{
-		// When there is at least one model server that's not queuing requests, and still has KV
-		// cache below a certain threshold, we consider this model server has capacity to handle
-		// a sheddable request without impacting critical requests.
-		Current:       filter.HasCapacityFilter,
-		NextOnSuccess: lowLatencyFilter,
-		// If all pods are queuing or running above the KVCache threshold, we drop the sheddable
-		// request to make room for critical requests. for this, we don't define nextOnFailure.
-	}
-)
-
 func NewScheduler(datastore Datastore) *Scheduler {
 	return NewSchedulerWithConfig(datastore, defaultConfig)
 }
@@ -206,19 +170,3 @@ func (s *Scheduler) runPostSchedulePlugins(ctx *types.SchedulingContext, res *ty
 		metrics.RecordSchedulerPluginProcessingLatency(plugins.PostSchedulePluginType, plugin.Name(), time.Since(before))
 	}
 }
-
-type defaultPlugin struct {
-	picker.RandomPicker
-}
-
-func (p *defaultPlugin) Name() string {
-	return "DefaultPlugin"
-}
-
-func (p *defaultPlugin) Filter(ctx *types.SchedulingContext, pods []types.Pod) []types.Pod {
-	if ctx.Req.Critical {
-		return lowLatencyFilter.Filter(ctx, pods)
-	}
-
-	return sheddableRequestFilter.Filter(ctx, pods)
-}