Preparedata timeout config

apix/config/v1alpha1/endpointpickerconfig_types.go

@@ -59,17 +59,22 @@ type EndpointPickerConfig struct {
 	// FlowControl configures the Flow Control layer.
 	// This configuration is only respected if the "flowControl" FeatureGate is enabled.
 	FlowControl *FlowControlConfig `json:"flowControl,omitempty"`
+
+	// +optional
+	// RequestControl configures the request control stage of the EPP pipeline.
+	RequestControl *RequestControlConfig `json:"requestControl,omitempty"`
 }
 
 func (cfg EndpointPickerConfig) String() string {
 	return fmt.Sprintf(
-		"{FeatureGates: %v, Plugins: %v, SchedulingProfiles: %v, Data: %v, SaturationDetector: %v, FlowControl: %v}",
+		"{FeatureGates: %v, Plugins: %v, SchedulingProfiles: %v, Data: %v, SaturationDetector: %v, FlowControl: %v, RequestControl: %v}",
 		cfg.FeatureGates,
 		cfg.Plugins,
 		cfg.SchedulingProfiles,
 		cfg.Data,
 		cfg.SaturationDetector,
 		cfg.FlowControl,
+		cfg.RequestControl,
 	)
 }
 
@@ -183,6 +188,24 @@ type SaturationDetector struct {
 	MetricsStalenessThreshold metav1.Duration `json:"metricsStalenessThreshold,omitempty"`
 }
 
+// RequestControlConfig configures the request control stage.
+type RequestControlConfig struct {
+	// +optional
+	// PrepareDataTimeout defines the timeout for PrepareData plugins.
+	// If omitted, defaults to 400ms.
+	PrepareDataTimeout *metav1.Duration `json:"prepareDataTimeout,omitempty"`
+}
+
+func (rc *RequestControlConfig) String() string {
+	if rc == nil {
+		return "{}"
+	}
+	if rc.PrepareDataTimeout == nil {
+		return "{PrepareDataTimeout: default(400ms)}"
+	}
+	return "{PrepareDataTimeout: " + rc.PrepareDataTimeout.String() + "}"
+}
+
 func (sd *SaturationDetector) String() string {
 	result := ""
 	if sd != nil {

cmd/epp/runner/runner.go

@@ -460,6 +460,10 @@ func (r *Runner) parseConfigurationPhaseTwo(ctx context.Context, rawConfig *conf
 
 	r.schedulerConfig = cfg.SchedulerConfig
 
+	if cfg.RequestControlConfig != nil && cfg.RequestControlConfig.PrepareDataTimeout > 0 {
+		r.requestControlConfig.WithPrepareDataTimeout(cfg.RequestControlConfig.PrepareDataTimeout)
+	}
+
 	// Add requestControl plugins
 	r.requestControlConfig.AddPlugins(handle.GetAllPlugins()...)
 

cmd/epp/runner/runner_test.go

@@ -0,0 +1,67 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package runner
+
+import (
+	"context"
+	"reflect"
+	"testing"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	configapi "sigs.k8s.io/gateway-api-inference-extension/apix/config/v1alpha1"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
+)
+
+func TestParseConfigurationPhaseTwoAppliesPrepareDataTimeout(t *testing.T) {
+	t.Parallel()
+
+	r := NewRunner()
+	r.registerInTreePlugins()
+
+	rawConfig := &configapi.EndpointPickerConfig{
+		RequestControl: &configapi.RequestControlConfig{
+			PrepareDataTimeout: &metav1.Duration{Duration: 125 * time.Millisecond},
+		},
+	}
+
+	ctx := context.Background()
+	epFactory := datalayer.NewEndpointFactory(nil, time.Millisecond)
+	ds := datastore.NewDatastore(ctx, epFactory, 0)
+
+	if _, err := r.parseConfigurationPhaseTwo(ctx, rawConfig, ds); err != nil {
+		t.Fatalf("parseConfigurationPhaseTwo failed: %v", err)
+	}
+
+	got := readPrepareDataTimeout(t, r.requestControlConfig)
+	want := 125 * time.Millisecond
+	if got != want {
+		t.Fatalf("prepareDataTimeout = %v, want %v", got, want)
+	}
+}
+
+func readPrepareDataTimeout(t *testing.T, cfg any) time.Duration {
+	t.Helper()
+
+	v := reflect.ValueOf(cfg).Elem().FieldByName("prepareDataTimeout")
+	if v.Kind() != reflect.Int64 {
+		t.Fatalf("unexpected kind for prepareDataTimeout: %v", v.Kind())
+	}
+	return time.Duration(v.Int())
+}

pkg/epp/config/config.go

@@ -17,6 +17,8 @@ limitations under the License.
 package config
 
 import (
+	"time"
+
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/flowcontrol"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/saturationdetector/framework/plugins/utilizationdetector"
@@ -29,4 +31,10 @@ type Config struct {
 	SaturationDetectorConfig *utilizationdetector.Config
 	DataConfig               *datalayer.Config
 	FlowControlConfig        *flowcontrol.Config
+	RequestControlConfig     *RequestControlConfig
+}
+
+// RequestControlConfig holds configuration for request control behaviors.
+type RequestControlConfig struct {
+	PrepareDataTimeout time.Duration
 }

pkg/epp/config/loader/configloader.go

@@ -118,6 +118,7 @@ func InstantiateAndConfigure(
 		SaturationDetectorConfig: buildSaturationConfig(rawConfig.SaturationDetector),
 		DataConfig:               dataConfig,
 		FlowControlConfig:        flowControlConfig,
+		RequestControlConfig:     buildRequestControlConfig(rawConfig.RequestControl),
 	}, nil
 }
 
@@ -245,6 +246,17 @@ func buildSaturationConfig(apiConfig *configapi.SaturationDetector) *utilization
 	return cfg
 }
 
+func buildRequestControlConfig(apiConfig *configapi.RequestControlConfig) *config.RequestControlConfig {
+	if apiConfig == nil {
+		return nil
+	}
+	cfg := &config.RequestControlConfig{}
+	if apiConfig.PrepareDataTimeout != nil && apiConfig.PrepareDataTimeout.Duration > 0 {
+		cfg.PrepareDataTimeout = apiConfig.PrepareDataTimeout.Duration
+	}
+	return cfg
+}
+
 func buildDataLayerConfig(rawDataConfig *configapi.DataLayerConfig, dataLayerEnabled bool, handle fwkplugin.Handle) (*datalayer.Config, error) {
 	if dataLayerEnabled && (rawDataConfig == nil || rawDataConfig.Sources == nil) { // enabled but no configuration
 		return nil, errors.New("the Datalayer has been enabled. You must specify the Data section in the configuration")

pkg/epp/config/loader/validation.go

@@ -17,6 +17,7 @@ limitations under the License.
 package loader
 
 import (
+	"errors"
 	"fmt"
 
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -32,6 +33,9 @@ func validateConfig(cfg *configapi.EndpointPickerConfig) error {
 	if err := validateSchedulingProfiles(cfg); err != nil {
 		return fmt.Errorf("scheduling profile validation failed: %w", err)
 	}
+	if err := validateRequestControl(cfg.RequestControl); err != nil {
+		return fmt.Errorf("request control validation failed: %w", err)
+	}
 	return nil
 }
 
@@ -78,3 +82,13 @@ func validateFeatureGates(gates configapi.FeatureGates) error {
 
 	return nil
 }
+
+func validateRequestControl(cfg *configapi.RequestControlConfig) error {
+	if cfg == nil || cfg.PrepareDataTimeout == nil {
+		return nil
+	}
+	if cfg.PrepareDataTimeout.Duration <= 0 {
+		return errors.New("prepareDataTimeout must be greater than 0")
+	}
+	return nil
+}

pkg/epp/requestcontrol/director.go

@@ -43,13 +43,6 @@ import (
 	requtil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/request"
 )
 
-const (
-	// TODO(https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/2081):
-	// Make this timeout configurable per-plugin or globally via the Director configuration to support plugins with
-	// varying latency profiles.
-	prepareDataTimeout = 400 * time.Millisecond
-)
-
 // Datastore defines the interface required by the Director.
 type Datastore interface {
 	PoolGet() (*datalayer.EndpointPool, error)
@@ -352,7 +345,13 @@ func (d *Director) runPrepareDataPlugins(ctx context.Context,
 	if len(d.requestControlPlugins.prepareDataPlugins) == 0 {
 		return nil
 	}
-	return prepareDataPluginsWithTimeout(prepareDataTimeout, d.requestControlPlugins.prepareDataPlugins, ctx, request, endpoints)
+	return prepareDataPluginsWithTimeout(
+		d.requestControlPlugins.prepareDataTimeout,
+		d.requestControlPlugins.prepareDataPlugins,
+		ctx,
+		request,
+		endpoints,
+	)
 }
 
 func (d *Director) runAdmissionPlugins(ctx context.Context,

pkg/epp/requestcontrol/request_control_config.go

@@ -17,10 +17,14 @@ limitations under the License.
 package requestcontrol
 
 import (
+	"time"
+
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/plugin"
 	fwk "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/requestcontrol"
 )
 
+const defaultPrepareDataTimeout = 400 * time.Millisecond
+
 // NewConfig creates a new Config object and returns its pointer.
 func NewConfig() *Config {
 	return &Config{
@@ -30,6 +34,7 @@ func NewConfig() *Config {
 		responseReceivedPlugins:  []fwk.ResponseReceived{},
 		responseStreamingPlugins: []fwk.ResponseStreaming{},
 		responseCompletePlugins:  []fwk.ResponseComplete{},
+		prepareDataTimeout:       defaultPrepareDataTimeout,
 	}
 }
 
@@ -41,6 +46,7 @@ type Config struct {
 	responseReceivedPlugins  []fwk.ResponseReceived
 	responseStreamingPlugins []fwk.ResponseStreaming
 	responseCompletePlugins  []fwk.ResponseComplete
+	prepareDataTimeout       time.Duration
 }
 
 // WithPreRequestPlugins sets the given plugins as the PreRequest plugins.
@@ -77,6 +83,12 @@ func (c *Config) WithPrepareDataPlugins(plugins ...fwk.PrepareDataPlugin) *Confi
 	return c
 }
 
+// WithPrepareDataTimeout sets the timeout for PrepareData plugins.
+func (c *Config) WithPrepareDataTimeout(timeout time.Duration) *Config {
+	c.prepareDataTimeout = timeout
+	return c
+}
+
 // WithAdmissionPlugins sets the given plugins as the AdmitRequest plugins.
 func (c *Config) WithAdmissionPlugins(plugins ...fwk.AdmissionPlugin) *Config {
 	c.admissionPlugins = plugins