Rename methods, cleanup

Author: rahulgurnani

cmd/epp/runner/runner.go

@@ -450,9 +450,10 @@ func (r *Runner) parseConfigurationPhaseTwo(ctx context.Context, rawConfig *conf
 	// Add requestControl plugins
 	r.requestControlConfig.AddPlugins(handle.GetAllPlugins()...)
 
-	// Sort prepare data plugins in DAG order (topological sort). Also check prepare data plugins for cycles.
-	if r.requestControlConfig.PrepareDataPluginGraph() != nil {
-		return nil, errors.New("failed to load the configuration - prepare data plugins have cyclic dependencies")
+	// Sort data plugins in DAG order (topological sort). Also check DAG for cycles.
+	// Also Order PrepareData plugins based on data dependencies.
+	if err := r.requestControlConfig.ValidateAndOrderDataDependencies(); err != nil {
+		return nil, fmt.Errorf("failed to load the configuration - %w", err)
 	}
 	// TODO(#1970): Remove feature gate check once prepare data plugins are stable.
 	if !r.featureGates[datalayer.PrepareDataPluginsFeatureGate] {

pkg/epp/requestcontrol/dag.go

@@ -64,7 +64,7 @@ func buildDAG(producers map[string]plugin.ProducerPlugin, consumers map[string]p
 
 // sortPlugins builds the dependency graph and returns the plugins ordered in topological order.
 // If there is a cycle, it returns an error.
-func sortPlugins(dag map[string][]string, plugins []fwk.PrepareDataPlugin) ([]fwk.PrepareDataPlugin, error) {
+func sortPlugins(dag map[string][]string, plugins []fwk.PrepareDataPlugin) ([]string, error) {
 	nameToPlugin := map[string]fwk.PrepareDataPlugin{}
 	for _, plugin := range plugins {
 		nameToPlugin[plugin.TypedName().String()] = plugin
@@ -73,12 +73,7 @@ func sortPlugins(dag map[string][]string, plugins []fwk.PrepareDataPlugin) ([]fw
 	if err != nil {
 		return nil, err
 	}
-	orderedPlugins := []fwk.PrepareDataPlugin{}
-	for _, pluginName := range sortedPlugins {
-		orderedPlugins = append(orderedPlugins, nameToPlugin[pluginName])
-	}
-
-	return orderedPlugins, err
+	return sortedPlugins, nil
 }
 
 // TopologicalSort performs Kahn's Algorithm on a DAG.

pkg/epp/requestcontrol/dag_test.go

@@ -162,20 +162,16 @@ func TestPrepareDataGraph(t *testing.T) {
 				t.Errorf("prepareDataGraph() mismatch (-want +got):\n%s", diff)
 			}
 
-			orderedPluginNames := make([]string, len(orderedPlugins))
-			for i, p := range orderedPlugins {
-				orderedPluginNames[i] = p.TypedName().String()
-			}
 			assertTopologicalOrder(t, dag, orderedPlugins)
 		})
 	}
 }
 
-func assertTopologicalOrder(t *testing.T, dag map[string][]string, ordered []fwk.PrepareDataPlugin) {
+func assertTopologicalOrder(t *testing.T, dag map[string][]string, ordered []string) {
 	t.Helper()
 	positions := make(map[string]int)
 	for i, p := range ordered {
-		positions[p.TypedName().String()] = i
+		positions[p] = i
 	}
 
 	for node, dependencies := range dag {

pkg/epp/requestcontrol/request_control_config.go

@@ -144,14 +144,10 @@ func (c *Config) ProducerConsumerPlugins() (map[string]plugin.ProducerPlugin, ma
 	return producers, consumers
 }
 
-// PrepareDataPluginGraph creates data dependency graph and sorts the plugins in topological order.
+// ValidateDataDependencies creates a data dependency graph and sorts the plugins in topological order.
 // If a cycle is detected, it returns an error.
-func (c *Config) PrepareDataPluginGraph() error {
+func (c *Config) ValidateAndOrderDataDependencies() error {
 	producers, consumers := c.ProducerConsumerPlugins()
-	// TODO(#1988): Add all producer and consumer plugins to the graph.
-	if len(c.prepareDataPlugins) == 0 {
-		return nil
-	}
 	dag, err := buildDAG(producers, consumers)
 	if err != nil {
 		return err
@@ -160,7 +156,22 @@ func (c *Config) PrepareDataPluginGraph() error {
 	if err != nil {
 		return err
 	}
-	c.prepareDataPlugins = plugins
+	c.orderPrepareDataPlugins(plugins)
 
 	return nil
 }
+
+// orderPrepareDataPlugins reorders the prepareDataPlugins in the Config based on the given sorted plugin names.
+func (c *Config) orderPrepareDataPlugins(sortedPluginNames []string) {
+	sortedPlugins := make([]fwk.PrepareDataPlugin, 0, len(sortedPluginNames))
+	nameToPlugin := make(map[string]fwk.PrepareDataPlugin)
+	for _, plugin := range NewConfig().prepareDataPlugins {
+		nameToPlugin[plugin.TypedName().String()] = plugin
+	}
+	for _, name := range sortedPluginNames {
+		if plugin, ok := nameToPlugin[name]; ok {
+			sortedPlugins = append(sortedPlugins, plugin)
+		}
+	}
+	c.prepareDataPlugins = sortedPlugins
+}