Add APIs for the instantiated plugins to the EPP Handle (kubernetes-sigs#1039)

* Added plugin instance APIs to plugins.Handle

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

* An implementation of the new plugins.Handle APIs

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

* Moved all configuration loading code to new package

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

* Updates due to new and moved APIs

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

* Cleanup of old configuration loading code

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

---------

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

Author: shmuelk

cmd/epp/runner/register.go

@@ -39,6 +39,49 @@ func RegisterAllPlugins() {
 	plugins.Register(scorer.QueueScorerType, scorer.QueueScorerFactory)
 }
 
-// eppHandle is a temporary implementation of the interface plugins.Handle
+// eppHandle is an implementation of the interface plugins.Handle
 type eppHandle struct {
+	plugins plugins.HandlePlugins
+}
+
+// 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() *eppHandle {
+	return &eppHandle{
+		plugins: &eppHandlePlugins{
+			thePlugins: map[string]plugins.Plugin{},
+		},
+	}
 }

cmd/epp/runner/runner.go

@@ -38,7 +38,7 @@ import (
 	conformance_epp "sigs.k8s.io/gateway-api-inference-extension/conformance/testing-epp"
 	"sigs.k8s.io/gateway-api-inference-extension/internal/runnable"
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/common/config"
+	"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
 	latencypredictor "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/latencypredictorasync"
@@ -242,29 +242,28 @@ func (r *Runner) Run(ctx context.Context) error {
 	// ===================================================================
 
 	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)
+		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()...)
 	}
 
 	// --- Initialize Core EPP Components ---

pkg/epp/common/config/configloader.go renamed to pkg/epp/common/config/loader/configloader.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package config
+package loader
 
 import (
 	"errors"
@@ -25,8 +25,11 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/serializer"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 
+	"sigs.k8s.io/gateway-api-inference-extension/api/config/v1alpha1"
 	configapi "sigs.k8s.io/gateway-api-inference-extension/api/config/v1alpha1"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/plugins"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/framework"
 )
 
 var scheme = runtime.NewScheme()
@@ -62,19 +65,61 @@ func LoadConfig(configText []byte, fileName string) (*configapi.EndpointPickerCo
 	return theConfig, nil
 }
 
-func LoadPluginReferences(thePlugins []configapi.PluginSpec, handle plugins.Handle) (map[string]plugins.Plugin, error) {
-	references := map[string]plugins.Plugin{}
+func LoadPluginReferences(thePlugins []configapi.PluginSpec, handle plugins.Handle) error {
 	for _, pluginConfig := range thePlugins {
-		thePlugin, err := InstantiatePlugin(pluginConfig, handle)
+		thePlugin, err := instantiatePlugin(pluginConfig, handle)
 		if err != nil {
-			return nil, err
+			return err
 		}
-		references[pluginConfig.Name] = thePlugin
+		handle.Plugins().AddPlugin(pluginConfig.Name, thePlugin)
 	}
-	return references, nil
+	return nil
+}
+
+func LoadSchedulerConfig(configProfiles []v1alpha1.SchedulingProfile, handle plugins.Handle) (*scheduling.SchedulerConfig, error) {
+
+	var profiles = map[string]*framework.SchedulerProfile{}
+
+	for _, configProfile := range configProfiles {
+		profile := framework.SchedulerProfile{}
+
+		for _, plugin := range configProfile.Plugins {
+			var err error
+			thePlugin := handle.Plugins().Plugin(plugin.PluginRef)
+			if theScorer, ok := thePlugin.(framework.Scorer); ok {
+				if plugin.Weight == nil {
+					return nil, fmt.Errorf("scorer '%s' is missing a weight", plugin.PluginRef)
+				}
+				thePlugin = framework.NewWeightedScorer(theScorer, *plugin.Weight)
+			}
+			err = profile.AddPlugins(thePlugin)
+			if err != nil {
+				return nil, err
+			}
+		}
+		profiles[configProfile.Name] = &profile
+	}
+
+	var profileHandler framework.ProfileHandler
+	var profileHandlerName string
+
+	for pluginName, thePlugin := range handle.Plugins().GetAllPluginsWithNames() {
+		if theProfileHandler, ok := thePlugin.(framework.ProfileHandler); ok {
+			if profileHandler != nil {
+				return nil, fmt.Errorf("only one profile handler is allowed. Both %s and %s are profile handlers", profileHandlerName, pluginName)
+			}
+			profileHandler = theProfileHandler
+			profileHandlerName = pluginName
+		}
+	}
+	if profileHandler == nil {
+		return nil, errors.New("no profile handler was specified")
+	}
+
+	return scheduling.NewSchedulerConfig(profileHandler, profiles), nil
 }
 
-func InstantiatePlugin(pluginSpec configapi.PluginSpec, handle plugins.Handle) (plugins.Plugin, error) {
+func instantiatePlugin(pluginSpec configapi.PluginSpec, handle plugins.Handle) (plugins.Plugin, error) {
 	factory, ok := plugins.Registry[pluginSpec.PluginName]
 	if !ok {
 		return nil, fmt.Errorf("failed to instantiate the plugin. plugin %s not found", pluginSpec.PluginName)