Merge branch 'kubernetes-sigs:main' into parallel-plugins2

Author: rahulgurnani

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,19 +434,28 @@ 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()...)
@@ -445,8 +464,49 @@ func (r *Runner) parsePluginsConfiguration(ctx context.Context, ds datastore.Dat
 		return errors.New("failed to load the configuration - prepare data plugins have cyclic dependencies")
 	}
 
+	// 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

@@ -26,12 +26,12 @@ require (
 	golang.org/x/sync v0.18.0
 	google.golang.org/grpc v1.76.0
 	google.golang.org/protobuf v1.36.10
-	k8s.io/api v0.34.1
-	k8s.io/apiextensions-apiserver v0.34.1
-	k8s.io/apimachinery v0.34.1
-	k8s.io/client-go v0.34.1
-	k8s.io/code-generator v0.34.1
-	k8s.io/component-base v0.34.1
+	k8s.io/api v0.34.2
+	k8s.io/apiextensions-apiserver v0.34.2
+	k8s.io/apimachinery v0.34.2
+	k8s.io/client-go v0.34.2
+	k8s.io/code-generator v0.34.2
+	k8s.io/component-base v0.34.2
 	k8s.io/utils v0.0.0-20250820121507-0af2bda4dd1d
 	sigs.k8s.io/controller-runtime v0.22.4
 	// Update the CONTROLLER_TOOLS_VERSION in Makefile when bumping controller-tools.
@@ -124,7 +124,7 @@ require (
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	k8s.io/apiserver v0.34.1 // indirect
+	k8s.io/apiserver v0.34.2 // indirect
 	k8s.io/gengo/v2 v2.0.0-20250820003526-c297c0c1eb9d // indirect
 	k8s.io/klog/v2 v2.130.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20250814151709-d7b6acb124c3 // indirect