BenjaminBraunDev
diff --git a/‎cmd/epp/runner/register.go‎
Lines changed: 97 additions & 0 deletions b/‎cmd/epp/runner/register.go‎
Lines changed: 97 additions & 0 deletions
diff --git a/‎cmd/epp/runner/runner.go‎
Lines changed: 92 additions & 36 deletions b/‎cmd/epp/runner/runner.go‎
Lines changed: 92 additions & 36 deletions
@@ -0,0 +1,97 @@
+/*
+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"
+
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/plugins"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/framework/plugins/filter"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/framework/plugins/multi/prefix"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/framework/plugins/picker"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/framework/plugins/profile"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/framework/plugins/scorer"
+)
+
+// RegisterAllPlugins registers the factory functions of all known plugins
+func RegisterAllPlugins() {
+	plugins.Register(filter.DecisionTreeFilterType, filter.DecisionTreeFilterFactory)
+	plugins.Register(filter.LeastKVCacheFilterType, filter.LeastKVCacheFilterFactory)
+	plugins.Register(filter.LeastQueueFilterType, filter.LeastQueueFilterFactory)
+	plugins.Register(filter.LoraAffinityFilterType, filter.LoraAffinityFilterFactory)
+	plugins.Register(filter.LowQueueFilterType, filter.LowQueueFilterFactory)
+	plugins.Register(prefix.PrefixCachePluginType, prefix.PrefixCachePluginFactory)
+	plugins.Register(picker.MaxScorePickerType, picker.MaxScorePickerFactory)
+	plugins.Register(picker.RandomPickerType, picker.RandomPickerFactory)
+	plugins.Register(profile.SingleProfileHandlerType, profile.SingleProfileHandlerFactory)
+	plugins.Register(scorer.KvCacheScorerType, scorer.KvCacheScorerFactory)
+	plugins.Register(scorer.QueueScorerType, scorer.QueueScorerFactory)
+}
+
+// eppHandle is an implementation of the interface plugins.Handle
+type eppHandle struct {
+	ctx     context.Context
+	plugins plugins.HandlePlugins
+}
+
+// Context returns a context the plugins can use, if they need one
+func (h *eppHandle) Context() context.Context {
+	return h.ctx
+}
+
+// Plugins returns the sub-handle for working with instantiated plugins
+func (h *eppHandle) Plugins() plugins.HandlePlugins {
+	return h.plugins
+}
+
+// eppHandlePlugins implements the set of APIs to work with instantiated plugins
+type eppHandlePlugins struct {
+	thePlugins map[string]plugins.Plugin
+}
+
+// Plugin returns the named plugin instance
+func (h *eppHandlePlugins) Plugin(name string) plugins.Plugin {
+	return h.thePlugins[name]
+}
+
+// AddPlugin adds a plugin to the set of known plugin instances
+func (h *eppHandlePlugins) AddPlugin(name string, plugin plugins.Plugin) {
+	h.thePlugins[name] = plugin
+}
+
+// GetAllPlugins returns all of the known plugins
+func (h *eppHandlePlugins) GetAllPlugins() []plugins.Plugin {
+	result := make([]plugins.Plugin, 0)
+	for _, plugin := range h.thePlugins {
+		result = append(result, plugin)
+	}
+	return result
+}
+
+// GetAllPluginsWithNames returns al of the known plugins with their names
+func (h *eppHandlePlugins) GetAllPluginsWithNames() map[string]plugins.Plugin {
+	return h.thePlugins
+}
+
+func newEppHandle(ctx context.Context) *eppHandle {
+	return &eppHandle{
+		ctx: ctx,
+		plugins: &eppHandlePlugins{
+			thePlugins: map[string]plugins.Plugin{},
+		},
+	}
+}
@@ -45,7 +45,7 @@ import (
 	"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/loader"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/common/config/loader"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
 
 	// Import the latency predictor package
@@ -325,6 +325,11 @@ func (r *Runner) Run(ctx context.Context) error {
 		setupLog.Error(err, "Failed to create controller manager")
 		return err
 	}
+	err = setupPprofHandlers(mgr)
+	if err != nil {
+		setupLog.Error(err, "Failed to setup pprof handlers")
+		return err
+	}
 
 	// ===================================================================
 	// == Latency Predictor Integration
@@ -368,37 +373,40 @@ func (r *Runner) Run(ctx context.Context) error {
 		}
 	}
 
+	// START DIFF
+	// below is what was incomming
+	err = r.parseConfiguration(ctx)
+	if err != nil {
+		setupLog.Error(err, "Failed to parse the configuration")
+		return err
+	}
+
+	// below is what was current
 	if len(*configText) != 0 || len(*configFile) != 0 {
-		theConfig, err := config.LoadConfig([]byte(*configText), *configFile)
+		theConfig, err := loader.LoadConfig([]byte(*configText), *configFile)
 		if err != nil {
 			setupLog.Error(err, "Failed to load the configuration")
 			return err
 		}
 
-		epp := eppHandle{}
-		instantiatedPlugins, err := config.LoadPluginReferences(theConfig.Plugins, epp)
+		epp := newEppHandle()
+
+		err = loader.LoadPluginReferences(theConfig.Plugins, epp)
 		if err != nil {
 			setupLog.Error(err, "Failed to instantiate the plugins")
 			return err
 		}
-	}
 
-	r.schedulerConfig, err = scheduling.LoadSchedulerConfig(theConfig.SchedulingProfiles, instantiatedPlugins)
-	if err != nil {
-		setupLog.Error(err, "Failed to create Scheduler configuration")
-		return err
-	}
-
-	err = r.parsePluginsConfiguration(ctx)
-	if err != nil {
-		setupLog.Error(err, "Failed to parse plugins configuration")
-		return err
-	}
+		r.schedulerConfig, err = loader.LoadSchedulerConfig(theConfig.SchedulingProfiles, epp)
+		if err != nil {
+			setupLog.Error(err, "Failed to create Scheduler configuration")
+			return err
+		}
 
-	// Add requestcontrol plugins
-	if instantiatedPlugins != nil {
-		r.requestControlConfig = requestcontrol.LoadRequestControlConfig(instantiatedPlugins)
+		// Add requestControl plugins
+		r.requestControlConfig.AddPlugins(epp.Plugins().GetAllPlugins()...)
 	}
+	// END DIFF
 
 	// --- Initialize Core EPP Components ---
 	if r.schedulerConfig == nil {
@@ -418,18 +426,20 @@ func (r *Runner) Run(ctx context.Context) error {
 
 	// --- Setup ExtProc Server Runner ---
 	serverRunner := &runserver.ExtProcServerRunner{
-		GrpcPort:                         *grpcPort,
-		PoolNamespacedName:               poolNamespacedName,
-		PoolGKNN:                         poolGKNN,
-		Datastore:                        datastore,
-		SecureServing:                    *secureServing,
-		HealthChecking:                   *healthChecking,
-		CertPath:                         *certPath,
-		RefreshPrometheusMetricsInterval: *refreshPrometheusMetricsInterval,
-		MetricsStalenessThreshold:        *metricsStalenessThreshold,
-		Director:                         director,
-		SaturationDetector:               saturationDetector,
-		LatencyPredictor:                 predictor,
+		GrpcPort:                                 *grpcPort,
+		PoolNamespacedName:                       poolNamespacedName,
+		PoolGKNN:                                 poolGKNN,
+		Datastore:                                datastore,
+		SecureServing:                            *secureServing,
+		HealthChecking:                           *healthChecking,
+		CertPath:                                 *certPath,
+		RefreshPrometheusMetricsInterval:         *refreshPrometheusMetricsInterval,
+		MetricsStalenessThreshold:                *metricsStalenessThreshold,
+		Director:                                 director,
+		SaturationDetector:                       saturationDetector,
+		LatencyPredictor:                         predictor,
+		DestinationEndpointHintMetadataNamespace: *destinationEndpointHintMetadataNamespace,
+		DestinationEndpointHintKey:               *destinationEndpointHintKey,
 	}
 	if err := serverRunner.SetupWithManager(ctx, mgr); err != nil {
 		setupLog.Error(err, "Failed to setup EPP controllers")
@@ -507,19 +517,19 @@ func (r *Runner) parsePluginsConfiguration(ctx context.Context) error {
 
 func (r *Runner) initializeScheduler(datastore datastore.Datastore, predictor *latencypredictor.Predictor) (*scheduling.Scheduler, error) {
 	if r.schedulerConfig != nil {
-		return scheduling.NewSchedulerWithConfig(datastore, r.schedulerConfig), nil
+		return scheduling.NewSchedulerWithConfig(r.schedulerConfig), nil
 	}
 
 	// otherwise, no one configured from outside scheduler config. use existing configuration
-	scheduler := scheduling.NewScheduler(datastore)
+	scheduler := scheduling.NewScheduler()
 	if schedulerV2 {
 		queueScorerWeight := envutil.GetEnvInt("QUEUE_SCORE_WEIGHT", scorer.DefaultQueueScorerWeight, setupLog)
 		kvCacheScorerWeight := envutil.GetEnvInt("KV_CACHE_SCORE_WEIGHT", scorer.DefaultKVCacheScorerWeight, setupLog)
 
 		schedulerProfile := framework.NewSchedulerProfile().
 			WithScorers(framework.NewWeightedScorer(scorer.NewQueueScorer(), queueScorerWeight),
 				framework.NewWeightedScorer(scorer.NewKVCacheScorer(), kvCacheScorerWeight)).
-			WithPicker(picker.NewMaxScorePicker())
+			WithPicker(picker.NewMaxScorePicker(picker.DefaultMaxNumOfEndpoints))
 
 		if prefixCacheScheduling {
 			prefixScorerWeight := envutil.GetEnvInt("PREFIX_CACHE_SCORE_WEIGHT", prefix.DefaultScorerWeight, setupLog)
@@ -529,16 +539,41 @@ func (r *Runner) initializeScheduler(datastore datastore.Datastore, predictor *l
 		}
 
 		schedulerConfig := scheduling.NewSchedulerConfig(profile.NewSingleProfileHandler(), map[string]*framework.SchedulerProfile{"schedulerv2": schedulerProfile})
-		scheduler = scheduling.NewSchedulerWithConfig(datastore, schedulerConfig)
+		scheduler = scheduling.NewSchedulerWithConfig(schedulerConfig)
 	}
 
 	if reqHeaderBasedSchedulerForTesting {
-		scheduler = conformance_epp.NewReqHeaderBasedScheduler(datastore)
+		scheduler = conformance_epp.NewReqHeaderBasedScheduler()
 	}
 
 	return scheduler, nil
 }
 
+func (r *Runner) parseConfiguration(ctx context.Context) error {
+	if len(*configText) != 0 || len(*configFile) != 0 {
+		theConfig, err := loader.LoadConfig([]byte(*configText), *configFile)
+		if err != nil {
+			return fmt.Errorf("failed to load the configuration - %w", err)
+		}
+
+		epp := newEppHandle(ctx)
+
+		err = loader.LoadPluginReferences(theConfig.Plugins, epp)
+		if err != nil {
+			return fmt.Errorf("failed to instantiate the plugins - %w", err)
+		}
+
+		r.schedulerConfig, err = loader.LoadSchedulerConfig(theConfig.SchedulingProfiles, epp)
+		if err != nil {
+			return fmt.Errorf("failed to create Scheduler configuration - %w", err)
+		}
+
+		// Add requestControl plugins
+		r.requestControlConfig.AddPlugins(epp.Plugins().GetAllPlugins()...)
+	}
+	return nil
+}
+
 func initLogging(opts *zap.Options) {
 	// Unless -zap-log-level is explicitly set, use -v
 	useV := true
@@ -652,3 +687,24 @@ func (p *predictorRunnable) Start(ctx context.Context) error {
 	p.predictor.Stop()
 	return nil
 }
+
+// setupPprofHandlers only implements the pre-defined profiles:
+// https://cs.opensource.google/go/go/+/refs/tags/go1.24.4:src/runtime/pprof/pprof.go;l=108
+func setupPprofHandlers(mgr ctrl.Manager) error {
+	var err error
+	profiles := []string{
+		"heap",
+		"goroutine",
+		"allocs",
+		"threadcreate",
+		"block",
+		"mutex",
+	}
+	for _, p := range profiles {
+		err = mgr.AddMetricsServerExtraHandler("/debug/pprof/"+p, pprof.Handler(p))
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}