feat: Added a factory function for the DecisionTree filter (kubernetes-sigs#1053)

* Added a factory function for the DecisionTreeFilter

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

* Added tests of the factory function of the DecisionTreeFilter

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

* Registered the factory function of the DecisionTreeFilter

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

* Refactored the configuration loading

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

---------

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

Author: shmuelk

cmd/epp/runner/register.go

@@ -27,6 +27,7 @@ import (
 
 // 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)

cmd/epp/runner/runner.go

@@ -242,29 +242,10 @@ func (r *Runner) Run(ctx context.Context) error {
 
 	// ===================================================================
 
-	if len(*configText) != 0 || len(*configFile) != 0 {
-		theConfig, err := loader.LoadConfig([]byte(*configText), *configFile)
-		if err != nil {
-			setupLog.Error(err, "Failed to load the configuration")
-			return err
-		}
-
-		epp := newEppHandle()
-
-		err = loader.LoadPluginReferences(theConfig.Plugins, epp)
-		if err != nil {
-			setupLog.Error(err, "Failed to instantiate the plugins")
-			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
-		r.requestControlConfig.AddPlugins(epp.Plugins().GetAllPlugins()...)
+	err = r.parseConfiguration()
+	if err != nil {
+		setupLog.Error(err, "Failed to parse the configuration")
+		return err
 	}
 
 	// --- Initialize Core EPP Components ---
@@ -357,6 +338,31 @@ func (r *Runner) initializeScheduler(datastore datastore.Datastore) (*scheduling
 	return scheduler, nil
 }
 
+func (r *Runner) parseConfiguration() 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()
+
+		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

pkg/epp/scheduling/framework/plugins/filter/decision_tree_filter.go

@@ -18,13 +18,21 @@ package filter
 
 import (
 	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
 
 	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/plugins"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/framework"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling/types"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
 
+const (
+	DecisionTreeFilterType = "decision-tree"
+)
+
 // compile-time type assertion
 var _ framework.Filter = &DecisionTreeFilter{}
 
@@ -47,6 +55,82 @@ type DecisionTreeFilter struct {
 	NextOnSuccessOrFailure framework.Filter
 }
 
+type decisionTreeFilterParameters struct {
+	Current                *decisionTreeFilterEntry `json:"current"`
+	NextOnSuccess          *decisionTreeFilterEntry `json:"nextOnSuccess"`
+	NextOnFailure          *decisionTreeFilterEntry `json:"nextOnFailure"`
+	NextOnSuccessOrFailure *decisionTreeFilterEntry `json:"nextOnSuccessOrFailure"`
+}
+
+type decisionTreeFilterEntry struct {
+	PluginRef    *string                       `json:"pluginRef"`
+	DecisionTree *decisionTreeFilterParameters `json:"decisionTree"`
+}
+
+func DecisionTreeFilterFactory(name string, rawParameters json.RawMessage, handle plugins.Handle) (plugins.Plugin, error) {
+	parameters := decisionTreeFilterParameters{}
+	if err := json.Unmarshal(rawParameters, &parameters); err != nil {
+		return nil, fmt.Errorf("failed to parse the parameters of the '%s' filter - %w", name, err)
+	}
+	return loadDecisionTree(&parameters, handle)
+}
+
+func loadDecisionTree(parameters *decisionTreeFilterParameters, handle plugins.Handle) (*DecisionTreeFilter, error) {
+	result := &DecisionTreeFilter{}
+	var err error
+
+	if parameters.Current == nil {
+		return nil, errors.New("a current filter must be specified")
+	}
+	result.Current, err = loadDecisionTreeEntry(parameters.Current, handle)
+	if err != nil {
+		return nil, err
+	}
+
+	if parameters.NextOnSuccess != nil {
+		result.NextOnSuccess, err = loadDecisionTreeEntry(parameters.NextOnSuccess, handle)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if parameters.NextOnFailure != nil {
+		result.NextOnFailure, err = loadDecisionTreeEntry(parameters.NextOnFailure, handle)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if parameters.NextOnSuccessOrFailure != nil {
+		result.NextOnSuccessOrFailure, err = loadDecisionTreeEntry(parameters.NextOnSuccessOrFailure, handle)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return result, nil
+}
+
+func loadDecisionTreeEntry(entry *decisionTreeFilterEntry, handle plugins.Handle) (framework.Filter, error) {
+	if entry.PluginRef != nil && entry.DecisionTree != nil {
+		return nil, errors.New("both pluginRef and decisionTree may not be specified")
+	}
+
+	if entry.PluginRef != nil {
+		instance := handle.Plugins().Plugin(*entry.PluginRef)
+		if instance == nil {
+			return nil, errors.New(*entry.PluginRef + " is a reference to an undefined Plugin")
+		}
+		if theFilter, ok := instance.(framework.Filter); ok {
+			return theFilter, nil
+		}
+		return nil, errors.New(*entry.PluginRef + " is not a filter")
+	} else if entry.DecisionTree != nil {
+		return loadDecisionTree(entry.DecisionTree, handle)
+	}
+	return nil, errors.New("either pluginRef or decisionTree must be specified")
+}
+
 // Type returns the type of the filter.
 func (f *DecisionTreeFilter) Type() string {
 	if f == nil {