feat: Extend the text based configuration to include feature flags and the SaturationDetector's configuration (#1492)

* Added feature gates and saturation config to YAML config

Signed-off-by: Shmuel Kallner <[email protected]>

* Added feature gates and saturation config to EPP config

Signed-off-by: Shmuel Kallner <[email protected]>

* Updated config loader for feature gates and saturation config

Signed-off-by: Shmuel Kallner <[email protected]>

* Moved validation into a separate file

Signed-off-by: Shmuel Kallner <[email protected]>

* Split Datastore initialization into two parts

Signed-off-by: Shmuel Kallner <[email protected]>

* Updated runner to use new config capabilities

Signed-off-by: Shmuel Kallner <[email protected]>

* Updates to tests

Signed-off-by: Shmuel Kallner <[email protected]>

* Updates to documentation

Signed-off-by: Shmuel Kallner <[email protected]>

* Made saturation detector config a pointer in epp config as per review

Signed-off-by: Shmuel Kallner <[email protected]>

* Split config loading into two parts and removed feature gates from config struct

Signed-off-by: Shmuel Kallner <[email protected]>

* Undid changes to Datastore

Signed-off-by: Shmuel Kallner <[email protected]>

* Updates due to config loading split

Signed-off-by: Shmuel Kallner <[email protected]>

* Test updates due to config loading split

Signed-off-by: Shmuel Kallner <[email protected]>

* Undid test changes due to Datastore restore

Signed-off-by: Shmuel Kallner <[email protected]>

---------

Signed-off-by: Shmuel Kallner <[email protected]>

Author: shmuelk

apix/config/v1alpha1/endpointpickerconfig_types.go

@@ -39,13 +39,25 @@ type EndpointPickerConfig struct {
 	// SchedulingProfiles is the list of named SchedulingProfiles
 	// that will be created.
 	SchedulingProfiles []SchedulingProfile `json:"schedulingProfiles"`
+
+	// +optional
+	// FeatureGates is a set of flags that enable various experimental features with the EPP.
+	// If omitted non of these experimental features will be enabled.
+	FeatureGates FeatureGates `json:"featureGates,omitempty"`
+
+	// +optional
+	// SaturationDetector when present specifies the configuration of the
+	// Saturation detector. If not present, default values are used.
+	SaturationDetector *SaturationDetector `json:"saturationDetector,omitempty"`
 }
 
 func (cfg EndpointPickerConfig) String() string {
 	return fmt.Sprintf(
-		"{Plugins: %v, SchedulingProfiles: %v}",
+		"{Plugins: %v, SchedulingProfiles: %v, FeatureGates: %v, SaturationDetector: %v}",
 		cfg.Plugins,
 		cfg.SchedulingProfiles,
+		cfg.FeatureGates,
+		cfg.SaturationDetector,
 	)
 }
 
@@ -118,3 +130,64 @@ func (sp SchedulingPlugin) String() string {
 	}
 	return fmt.Sprintf("{PluginRef: %s%s}", sp.PluginRef, weight)
 }
+
+// FeatureGates is a set of flags that enable various experimental features with the EPP
+type FeatureGates []string
+
+func (fg FeatureGates) String() string {
+	if fg == nil {
+		return "{}"
+	}
+
+	result := ""
+	for _, gate := range fg {
+		result += gate + ","
+	}
+
+	if len(result) > 0 {
+		result = result[:len(result)-1]
+	}
+	return "{" + result + "}"
+}
+
+// SaturationDetector
+type SaturationDetector struct {
+	// +optional
+	// QueueDepthThreshold defines the backend waiting queue size above which a
+	// pod is considered to have insufficient capacity for new requests.
+	QueueDepthThreshold int `json:"queueDepthThreshold,omitempty"`
+
+	// +optional
+	// KVCacheUtilThreshold defines the KV cache utilization (0.0 to 1.0) above
+	// which a pod is considered to have insufficient capacity.
+	KVCacheUtilThreshold float64 `json:"kvCacheUtilThreshold,omitempty"`
+
+	// +optional
+	// MetricsStalenessThreshold defines how old a pod's metrics can be.
+	// If a pod's metrics are older than this, it might be excluded from
+	// "good capacity" considerations or treated as having no capacity for
+	// safety.
+	MetricsStalenessThreshold metav1.Duration `json:"metricsStalenessThreshold,omitempty"`
+}
+
+func (sd *SaturationDetector) String() string {
+	result := ""
+	if sd != nil {
+		if sd.QueueDepthThreshold != 0 {
+			result += fmt.Sprintf("QueueDepthThreshold: %d", sd.QueueDepthThreshold)
+		}
+		if sd.KVCacheUtilThreshold != 0.0 {
+			if len(result) != 0 {
+				result += ", "
+			}
+			result += fmt.Sprintf("KVCacheUtilThreshold: %g", sd.KVCacheUtilThreshold)
+		}
+		if sd.MetricsStalenessThreshold.Duration != 0.0 {
+			if len(result) != 0 {
+				result += ", "
+			}
+			result += fmt.Sprintf("MetricsStalenessThreshold: %s", sd.MetricsStalenessThreshold)
+		}
+	}
+	return "{" + result + "}"
+}

apix/config/v1alpha1/zz_generated.deepcopy.go

@@ -44,9 +44,11 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 
+	configapi "sigs.k8s.io/gateway-api-inference-extension/apix/config/v1alpha1"
 	"sigs.k8s.io/gateway-api-inference-extension/internal/runnable"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/common"
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/config"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/config/loader"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer"
 	dlmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer/metrics"
@@ -77,6 +79,11 @@ const (
 	// enableExperimentalFlowControlLayer defines the environment variable used as a feature flag for the pluggable flow
 	// control layer.
 	enableExperimentalFlowControlLayer = "ENABLE_EXPERIMENTAL_FLOW_CONTROL_LAYER"
+
+	// Saturation Detector deprecated configuration environment variables
+	EnvSdQueueDepthThreshold       = "SD_QUEUE_DEPTH_THRESHOLD"
+	EnvSdKVCacheUtilThreshold      = "SD_KV_CACHE_UTIL_THRESHOLD"
+	EnvSdMetricsStalenessThreshold = "SD_METRICS_STALENESS_THRESHOLD"
 )
 
 // TODO: this is hardcoded for POC only. This needs to be hooked up to our text-based config story.
@@ -143,6 +150,7 @@ func NewRunner() *Runner {
 // Runner is used to run epp with its plugins
 type Runner struct {
 	eppExecutableName    string // the EPP executable name
+	featureGates         config.FeatureConfig
 	requestControlConfig *requestcontrol.Config
 	schedulerConfig      *scheduling.SchedulerConfig
 	customCollectors     []prometheus.Collector
@@ -200,24 +208,33 @@ func (r *Runner) Run(ctx context.Context) error {
 	})
 	setupLog.Info("Flags processed", "flags", flags)
 
-	// --- Load Configurations from Environment Variables ---
-	sdConfig := saturationdetector.LoadConfigFromEnv()
-
 	// --- Get Kubernetes Config ---
 	cfg, err := ctrl.GetConfig()
 	if err != nil {
 		setupLog.Error(err, "Failed to get Kubernetes rest config")
 		return err
 	}
 
+	rawConfig, featureGates, err := r.parseConfigurationPhaseOne(ctx)
+	if err != nil {
+		setupLog.Error(err, "Failed to parse configuration")
+		return err
+	}
+	r.featureGates = featureGates
+
 	// --- Setup Datastore ---
-	useDatalayerV2 := env.GetEnvBool(enableExperimentalDatalayerV2, false, setupLog)
-	epf, err := r.setupMetricsCollection(setupLog, useDatalayerV2)
+	epf, err := r.setupMetricsCollection(setupLog, r.featureGates[datalayer.FeatureGate])
 	if err != nil {
 		return err
 	}
 	datastore := datastore.NewDatastore(ctx, epf, int32(*modelServerMetricsPort))
 
+	eppConfig, err := r.parseConfigurationPhaseTwo(ctx, rawConfig, datastore)
+	if err != nil {
+		setupLog.Error(err, "Failed to parse configuration")
+		return err
+	}
+
 	// --- Setup Metrics Server ---
 	customCollectors := []prometheus.Collector{collectors.NewInferencePoolMetricsCollector(datastore)}
 	if r.customCollectors != nil {
@@ -297,12 +314,6 @@ func (r *Runner) Run(ctx context.Context) error {
 		runtime.SetBlockProfileRate(1)
 	}
 
-	err = r.parsePluginsConfiguration(ctx, datastore)
-	if err != nil {
-		setupLog.Error(err, "Failed to parse plugins configuration")
-		return err
-	}
-
 	// --- Initialize Core EPP Components ---
 	if r.schedulerConfig == nil {
 		err := errors.New("scheduler config must be set either by config api or through code")
@@ -314,12 +325,11 @@ func (r *Runner) Run(ctx context.Context) error {
 
 	scheduler := scheduling.NewSchedulerWithConfig(r.schedulerConfig)
 
-	saturationDetector := saturationdetector.NewDetector(sdConfig, setupLog)
+	saturationDetector := saturationdetector.NewDetector(eppConfig.SaturationDetectorConfig, setupLog)
 
 	// --- Admission Control Initialization ---
-	enableFlowControl := env.GetEnvBool(enableExperimentalFlowControlLayer, false, setupLog)
 	var admissionController requestcontrol.AdmissionController
-	if enableFlowControl {
+	if r.featureGates[flowcontrol.FeatureGate] {
 		setupLog.Info("Initializing experimental Flow Control layer")
 		fcCfg, err := flowControlConfig.ValidateAndApplyDefaults()
 		if err != nil {
@@ -367,7 +377,7 @@ func (r *Runner) Run(ctx context.Context) error {
 		MetricsStalenessThreshold:        *metricsStalenessThreshold,
 		Director:                         director,
 		SaturationDetector:               saturationDetector,
-		UseExperimentalDatalayerV2:       useDatalayerV2, // pluggable data layer feature flag
+		UseExperimentalDatalayerV2:       r.featureGates[datalayer.FeatureGate], // pluggable data layer feature flag
 	}
 	if err := serverRunner.SetupWithManager(ctx, mgr); err != nil {
 		setupLog.Error(err, "Failed to setup EPP controllers")
@@ -410,9 +420,9 @@ func (r *Runner) registerInTreePlugins() {
 	plugins.Register(testfilter.HeaderBasedTestingFilterType, testfilter.HeaderBasedTestingFilterFactory)
 }
 
-func (r *Runner) parsePluginsConfiguration(ctx context.Context, ds datastore.Datastore) error {
+func (r *Runner) parseConfigurationPhaseOne(ctx context.Context) (*configapi.EndpointPickerConfig, config.FeatureConfig, error) {
 	if *configText == "" && *configFile == "" {
-		return nil // configuring through code, not through file
+		return nil, nil, nil // configuring through code, not through file
 	}
 
 	logger := log.FromContext(ctx)
@@ -424,25 +434,75 @@ func (r *Runner) parsePluginsConfiguration(ctx context.Context, ds datastore.Dat
 		var err error
 		configBytes, err = os.ReadFile(*configFile)
 		if err != nil {
-			return fmt.Errorf("failed to load config from a file '%s' - %w", *configFile, err)
+			return nil, nil, fmt.Errorf("failed to load config from a file '%s' - %w", *configFile, err)
 		}
 	}
 
+	loader.RegisterFeatureGate(datalayer.FeatureGate)
+	loader.RegisterFeatureGate(flowcontrol.FeatureGate)
+
 	r.registerInTreePlugins()
+
+	return loader.LoadConfigPhaseOne(configBytes, logger)
+}
+
+func (r *Runner) parseConfigurationPhaseTwo(ctx context.Context, rawConfig *configapi.EndpointPickerConfig, ds datastore.Datastore) (*config.Config, error) {
+	logger := log.FromContext(ctx)
 	handle := plugins.NewEppHandle(ctx, ds.PodList)
-	config, err := loader.LoadConfig(configBytes, handle, logger)
+	cfg, err := loader.LoadConfigPhaseTwo(rawConfig, handle, logger)
 
 	if err != nil {
-		return fmt.Errorf("failed to load the configuration - %w", err)
+		return nil, fmt.Errorf("failed to load the configuration - %w", err)
 	}
 
-	r.schedulerConfig = config.SchedulerConfig
+	r.schedulerConfig = cfg.SchedulerConfig
 
 	// Add requestControl plugins
 	r.requestControlConfig.AddPlugins(handle.GetAllPlugins()...)
 
+	// Handler deprecated configuration options
+	r.deprecatedConfigurationHelper(cfg, logger)
+
 	logger.Info("loaded configuration from file/text successfully")
-	return nil
+	return cfg, nil
+}
+
+func (r *Runner) deprecatedConfigurationHelper(cfg *config.Config, logger logr.Logger) {
+	// Handle deprecated environment variable based feature flags
+
+	if _, ok := os.LookupEnv(enableExperimentalDatalayerV2); ok {
+		logger.Info("Enabling the experimental Data Layer V2 using environment variables is deprecated")
+		r.featureGates[datalayer.FeatureGate] = env.GetEnvBool(enableExperimentalDatalayerV2, false, logger)
+	}
+	if _, ok := os.LookupEnv(enableExperimentalFlowControlLayer); ok {
+		logger.Info("Enabling the experimental Flow Control layer using environment variables is deprecated")
+		r.featureGates[flowcontrol.FeatureGate] = env.GetEnvBool(enableExperimentalFlowControlLayer, false, setupLog)
+	}
+
+	// Handle deprecated environment variable base Saturation Detector configuration
+
+	if _, ok := os.LookupEnv(EnvSdQueueDepthThreshold); ok {
+		logger.Info("Configuring Saturation Detector using environment variables is deprecated")
+		cfg.SaturationDetectorConfig.QueueDepthThreshold =
+			env.GetEnvInt(EnvSdQueueDepthThreshold, saturationdetector.DefaultQueueDepthThreshold, logger)
+		if cfg.SaturationDetectorConfig.QueueDepthThreshold <= 0 {
+			cfg.SaturationDetectorConfig.QueueDepthThreshold = saturationdetector.DefaultQueueDepthThreshold
+		}
+	}
+	if _, ok := os.LookupEnv(EnvSdKVCacheUtilThreshold); ok {
+		logger.Info("Configuring Saturation Detector using environment variables is deprecated")
+		cfg.SaturationDetectorConfig.KVCacheUtilThreshold = env.GetEnvFloat(EnvSdKVCacheUtilThreshold, saturationdetector.DefaultKVCacheUtilThreshold, logger)
+		if cfg.SaturationDetectorConfig.KVCacheUtilThreshold <= 0 || cfg.SaturationDetectorConfig.KVCacheUtilThreshold >= 1 {
+			cfg.SaturationDetectorConfig.KVCacheUtilThreshold = saturationdetector.DefaultKVCacheUtilThreshold
+		}
+	}
+	if _, ok := os.LookupEnv(EnvSdMetricsStalenessThreshold); ok {
+		logger.Info("Configuring Saturation Detector using environment variables is deprecated")
+		cfg.SaturationDetectorConfig.MetricsStalenessThreshold = env.GetEnvDuration(EnvSdMetricsStalenessThreshold, saturationdetector.DefaultMetricsStalenessThreshold, logger)
+		if cfg.SaturationDetectorConfig.MetricsStalenessThreshold <= 0 {
+			cfg.SaturationDetectorConfig.MetricsStalenessThreshold = saturationdetector.DefaultMetricsStalenessThreshold
+		}
+	}
 }
 
 func (r *Runner) setupMetricsCollection(setupLog logr.Logger, useExperimentalDatalayer bool) (datalayer.EndpointFactory, error) {

cmd/epp/runner/runner.go

@@ -73,7 +73,7 @@ nav:
         - InferencePool Rollout: guides/inferencepool-rollout.md
       - Metrics and Observability: guides/metrics-and-observability.md
       - Configuration Guide:
-          - Configuring the plugins via configuration YAML file: guides/epp-configuration/config-text.md
+          - Configuring the EndPoint Picker via configuration YAML file: guides/epp-configuration/config-text.md
           - Prefix Cache Aware Plugin: guides/epp-configuration/prefix-aware.md
       - Migration Guide: guides/ga-migration.md
       - Troubleshooting Guide: guides/troubleshooting.md

mkdocs.yml

@@ -16,9 +16,15 @@ limitations under the License.
 
 package config
 
-import "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling"
+import (
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/saturationdetector"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling"
+)
 
 // Config is the configuration loaded from the text based configuration
 type Config struct {
-	SchedulerConfig *scheduling.SchedulerConfig
+	SchedulerConfig          *scheduling.SchedulerConfig
+	SaturationDetectorConfig *saturationdetector.Config
 }
+
+type FeatureConfig map[string]bool