Validate all producer consumer plugins

Author: rahulgurnani

pkg/epp/requestcontrol/dag.go

@@ -20,35 +20,39 @@ import (
 	"errors"
 	"slices"
 
+	"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"
 )
 
 // buildDAG builds a dependency graph among data preparation plugins based on their
 // produced and consumed data keys.
-func buildDAG(plugins []fwk.PrepareDataPlugin) (map[string][]string, error) {
+func buildDAG(producers map[string]plugin.ProducerPlugin, consumers map[string]plugin.ConsumerPlugin) (map[string][]string, error) {
 	dag := make(map[string][]string)
-	for _, plugin := range plugins {
-		dag[plugin.TypedName().String()] = []string{}
-	}
 	// Create dependency graph as a DAG.
-	for i := range plugins {
-		for j := range plugins {
-			if i == j {
+	for _, producer := range producers {
+		dag[producer.TypedName().String()] = []string{}
+	}
+	for _, consumer := range consumers {
+		dag[consumer.TypedName().String()] = []string{}
+	}
+	for pName, producer := range producers {
+		for cName, consumer := range consumers {
+			if pName == cName {
 				continue
 			}
 			// Check whether plugin[i] produces something consumed by plugin[j]. In that case, j depends on i.
-			if plugins[i].Produces() != nil && plugins[j].Consumes() != nil {
-				for producedKey, producedData := range plugins[i].Produces() {
+			if producer.Produces() != nil && consumer.Consumes() != nil {
+				for producedKey, producedData := range producer.Produces() {
 					// If plugin j consumes the produced key, then j depends on i. We can break after the first match.
-					if consumedData, ok := plugins[j].Consumes()[producedKey]; ok {
+					if consumedData, ok := consumer.Consumes()[producedKey]; ok {
 						// Check types are same. Reflection is avoided here for simplicity.
 						// TODO(#1985): Document this detail in IGW docs.
 						if producedData != consumedData {
 							return nil, errors.New("data type mismatch between produced and consumed data for key: " + producedKey)
 						}
-						iPluginName := plugins[i].TypedName().String()
-						jPluginName := plugins[j].TypedName().String()
-						dag[jPluginName] = append(dag[jPluginName], iPluginName)
+						// Consumer depends on producer, so add an edge from consumer to producer.
+						dag[cName] = append(dag[cName], pName)
 						break
 					}
 				}

pkg/epp/requestcontrol/dag_test.go

@@ -124,7 +124,17 @@ func TestPrepareDataGraph(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			dag, err := buildDAG(tc.plugins)
+			producers := make(map[string]fwkplugin.ProducerPlugin)
+			consumers := make(map[string]fwkplugin.ConsumerPlugin)
+			for _, p := range tc.plugins {
+				if pp, ok := p.(fwkplugin.ProducerPlugin); ok {
+					producers[p.TypedName().String()] = pp
+				}
+				if cp, ok := p.(fwkplugin.ConsumerPlugin); ok {
+					consumers[p.TypedName().String()] = cp
+				}
+			}
+			dag, err := buildDAG(producers, consumers)
 			if err != nil {
 				if tc.expectError {
 					assert.Error(t, err)

pkg/epp/requestcontrol/request_control_config.go

@@ -106,14 +106,54 @@ func (c *Config) AddPlugins(pluginObjects ...plugin.Plugin) {
 	}
 }
 
+// ProducerConsumerPlugins iterates through all registered plugins and returns two slices:
+// one containing the names of plugins that implement the ProducerPlugin interface,
+// and another for plugins that implement the ConsumerPlugin interface.
+func (c *Config) ProducerConsumerPlugins() (map[string]plugin.ProducerPlugin, map[string]plugin.ConsumerPlugin) {
+	var producers map[string]plugin.ProducerPlugin
+	var consumers map[string]plugin.ConsumerPlugin
+
+	// Collect all unique plugins from the config.
+	allPlugins := make(map[string]plugin.Plugin)
+	for _, p := range c.admissionPlugins {
+		allPlugins[p.TypedName().String()] = p
+	}
+	for _, p := range c.prepareDataPlugins {
+		allPlugins[p.TypedName().String()] = p
+	}
+	for _, p := range c.preRequestPlugins {
+		allPlugins[p.TypedName().String()] = p
+	}
+	for _, p := range c.responseReceivedPlugins {
+		allPlugins[p.TypedName().String()] = p
+	}
+	for _, p := range c.responseStreamingPlugins {
+		allPlugins[p.TypedName().String()] = p
+	}
+	for _, p := range c.responseCompletePlugins {
+		allPlugins[p.TypedName().String()] = p
+	}
+
+	for name, p := range allPlugins {
+		if producer, ok := p.(plugin.ProducerPlugin); ok {
+			producers[name] = producer
+		}
+		if consumer, ok := p.(plugin.ConsumerPlugin); ok {
+			consumers[name] = consumer
+		}
+	}
+	return producers, consumers
+}
+
 // PrepareDataPluginGraph creates data dependency graph and sorts the plugins in topological order.
 // If a cycle is detected, it returns an error.
 func (c *Config) PrepareDataPluginGraph() 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(c.prepareDataPlugins)
+	dag, err := buildDAG(producers, consumers)
 	if err != nil {
 		return err
 	}