Externalize Scheduler's saturation logic

This commit refactors the request processing pipeline, externalizing
saturation detection and criticality-based service differentiation
from the Scheduler. These responsibilities are now primarily managed by
the RequestControl.Director.

This change is a preparatory step for the introduction of a new
Flow Controller component, which will eventually absorb these admission
control duties.

Key changes include:

- Introduced `PreDispatch` method to `RequestControl.Director` It
  utilizes the `SaturationDetector` for admission control of
  non-critical requests and handles request criticality to determine if
  saturation checks are bypassed.
- The saturation detection logic for dropping non-critical requests
  is intentionally preserved within the `Director` at this stage.
  This allows the option to bypass the future Flow Controller
  component during its maturation, ensuring the existing saturation
  and sheddable request behavior can be maintained as a fallback.
- Simplified the `Scheduler` to focus solely on preference-based
  filtering and pod selection for requests that have already been
  admitted by the `Director`.
- Removed the `SheddableRequestFilter` and the distinct
  critical/sheddable filter paths from the `Scheduler`'s internal logic.
  The `Scheduler` now applies a single, unified preference filter chain
  to all incoming requests.
- Updated `main.go` to instantiate the `SaturationDetector`, wiring it
  into the request handling flow.
- Updated tests across `scheduler_test.go`, `director_test.go`, and
  `filter_test.go` to align with the new component responsibilities,
  adding additional coverage where necessary.

This refactoring leads to a cleaner architecture, making the `Scheduler`
a more focused component and centralizing initial admission control logic,
while paving the way for the future Flow Controller.

This is aligned with the direction in `0683-epp-architecture-proposal`
and should be nearly no-op in terms of EPP behavior.

Author: LukeAVanDrie

cmd/epp/main.go

@@ -43,9 +43,9 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics/collectors"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/saturationdetector"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling"
 	"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/plugins/prefix"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/plugins/scorer"
@@ -157,13 +157,19 @@ func run() error {
 	})
 	setupLog.Info("Flags processed", "flags", flags)
 
-	// Init runtime.
+	// --- Load Configurations from Environment Variables ---
+	// Note: Scheduler config is loaded via its package init currently. We may
+	// want to load it here explicitly:
+	sdConfig := saturationdetector.LoadConfigFromEnv()
+
+	// --- Get Kubernetes Config ---
 	cfg, err := ctrl.GetConfig()
 	if err != nil {
-		setupLog.Error(err, "Failed to get rest config")
+		setupLog.Error(err, "Failed to get Kubernetes rest config")
 		return err
 	}
 
+	// --- Setup Manager ---
 	poolNamespacedName := types.NamespacedName{
 		Name:      *poolName,
 		Namespace: *poolNamespace,
@@ -174,7 +180,7 @@ func run() error {
 		return err
 	}
 
-	// Set up mapper for metric scraping.
+	// --- Setup Datastore ---
 	mapping, err := backendmetrics.NewMetricMapping(
 		*totalQueuedRequestsMetric,
 		*kvCacheUsagePercentageMetric,
@@ -185,14 +191,12 @@ func run() error {
 		return err
 	}
 	verifyMetricMapping(*mapping, setupLog)
-
 	pmf := backendmetrics.NewPodMetricsFactory(&backendmetrics.PodMetricsClientImpl{MetricMapping: mapping}, *refreshMetricsInterval)
-	// Setup runner.
 	ctx := ctrl.SetupSignalHandler()
+	appDatastore := datastore.NewDatastore(ctx, pmf)
 
-	datastore := datastore.NewDatastore(ctx, pmf)
-
-	scheduler := scheduling.NewScheduler(datastore)
+	// --- Initialize EPP Components ---
+	appScheduler := scheduling.NewScheduler(appDatastore)
 	if schedulerV2 == "true" {
 		queueScorerWeight := envutil.GetEnvInt("QUEUE_SCORE_WEIGHT", scorer.DefaultQueueScorerWeight, setupLog)
 		kvCacheScorerWeight := envutil.GetEnvInt("KV_CACHE_SCORE_WEIGHT", scorer.DefaultKVCacheScorerWeight, setupLog)
@@ -207,47 +211,62 @@ func run() error {
 		}
 		schedulerConfig := scheduling.NewSchedulerConfig(
 			[]plugins.PreSchedule{},
-			[]plugins.Filter{filter.NewSheddableCapacityFilter()},
+			[]plugins.Filter{},
 			scorers,
 			picker.NewMaxScorePicker(),
 			[]plugins.PostSchedule{},
 			[]plugins.PostResponse{},
 			schedConfigOpts...)
-		scheduler = scheduling.NewSchedulerWithConfig(datastore, schedulerConfig)
+		appScheduler = scheduling.NewSchedulerWithConfig(appDatastore, schedulerConfig)
+	}
+
+	appSaturationDetector, err := saturationdetector.NewDetector(
+		*sdConfig,
+		appDatastore,
+		ctrl.Log.WithName("saturation-detector"),
+	)
+	if err != nil {
+		setupLog.Error(err, "Failed to create SaturationDetector")
+		return err
 	}
+
+	// --- Setup ExtProc Server Runner ---
 	serverRunner := &runserver.ExtProcServerRunner{
 		GrpcPort:                                 *grpcPort,
 		DestinationEndpointHintMetadataNamespace: *destinationEndpointHintMetadataNamespace,
 		DestinationEndpointHintKey:               *destinationEndpointHintKey,
 		PoolNamespacedName:                       poolNamespacedName,
-		Datastore:                                datastore,
+		Datastore:                                appDatastore,
 		SecureServing:                            *secureServing,
 		CertPath:                                 *certPath,
 		RefreshPrometheusMetricsInterval:         *refreshPrometheusMetricsInterval,
-		Scheduler:                                scheduler,
+		Scheduler:                                appScheduler,
+		SaturationDetector:                       appSaturationDetector,
 	}
 	if err := serverRunner.SetupWithManager(ctx, mgr); err != nil {
-		setupLog.Error(err, "Failed to setup ext-proc controllers")
+		setupLog.Error(err, "Failed to setup EPP controllers")
 		return err
 	}
 
+	// --- Add Runnables to Manager ---
+
 	// Register health server.
-	if err := registerHealthServer(mgr, ctrl.Log.WithName("health"), datastore, *grpcHealthPort); err != nil {
+	if err := registerHealthServer(mgr, ctrl.Log.WithName("health"), appDatastore, *grpcHealthPort); err != nil {
 		return err
 	}
 
 	// Register ext-proc server.
-	if err := mgr.Add(serverRunner.AsRunnable(ctrl.Log.WithName("ext-proc"))); err != nil {
-		setupLog.Error(err, "Failed to register ext-proc gRPC server")
+	if err := registerExtProcServer(mgr, serverRunner, ctrl.Log.WithName("ext-proc")); err != nil {
 		return err
 	}
 
 	// Register metrics handler.
-	if err := registerMetricsHandler(mgr, *metricsPort, cfg, datastore); err != nil {
+	if err := registerMetricsHandler(mgr, *metricsPort, cfg, appDatastore); err != nil {
 		return err
 	}
 
-	// Start the manager. This blocks until a signal is received.
+	// --- Start Manager ---
+	// This blocks until a signal is received.
 	setupLog.Info("Controller manager starting")
 	if err := mgr.Start(ctx); err != nil {
 		setupLog.Error(err, "Error starting controller manager")
@@ -275,6 +294,17 @@ func initLogging(opts *zap.Options) {
 	ctrl.SetLogger(logger)
 }
 
+// registerExtProcServer adds the ExtProcServerRunner as a Runnable to the
+// manager.
+func registerExtProcServer(mgr manager.Manager, runner *runserver.ExtProcServerRunner, logger logr.Logger) error {
+	if err := mgr.Add(runner.AsRunnable(logger)); err != nil {
+		setupLog.Error(err, "Failed to register ext-proc gRPC server runnable")
+		return err
+	}
+	setupLog.Info("ExtProc server runner added to manager.")
+	return nil
+}
+
 // registerHealthServer adds the Health gRPC server as a Runnable to the given manager.
 func registerHealthServer(mgr manager.Manager, logger logr.Logger, ds datastore.Datastore, port int) error {
 	srv := grpc.NewServer()
@@ -364,5 +394,4 @@ func verifyMetricMapping(mapping backendmetrics.MetricMapping, logger logr.Logge
 	if mapping.LoraRequestInfo == nil {
 		logger.Info("Not scraping metric: LoraRequestInfo")
 	}
-
 }