feat(parser) add support for KongClusterPlugins

KongClusterPlugins allows user to have Plugin configuration at a
cluster-level and share the configuration across multiple workspaces.

This also helps users who want to have different personas for defining a
plugin configuration and using a plugin on a service or ingress as
access to KongClusterPlugin can be setup independent of other resources.

KongPlugin in the namespace takes precedence over KongClusterPlugin.

Global plugins for KongClusterPlugins are also supported.

Based on feedback, in future, we will deperecate global-level KongPlugin
support and instead only KongClusterPlugins will support the global
label. This will solve the problem of a user with access to a
single namespace manipulating proxy behavior for services in all namespaces.

cli/ingress-controller/main.go

@@ -285,6 +285,11 @@ func main() {
 	cacheStores.Plugin = kongPluginInformer.GetStore()
 	informers = append(informers, kongPluginInformer)
 
+	kongClusterPluginInformer := kongInformerFactory.Configuration().V1().KongClusterPlugins().Informer()
+	kongClusterPluginInformer.AddEventHandler(reh)
+	cacheStores.ClusterPlugin = kongClusterPluginInformer.GetStore()
+	informers = append(informers, kongClusterPluginInformer)
+
 	kongConsumerInformer := kongInformerFactory.Configuration().V1().KongConsumers().Informer()
 	kongConsumerInformer.AddEventHandler(reh)
 	cacheStores.Consumer = kongConsumerInformer.GetStore()

internal/ingress/controller/parser/parser.go

@@ -964,7 +964,7 @@ func (p *Parser) fillPlugins(state KongState) []Plugin {
 		namespace, kongPluginName := identifier[0], identifier[1]
 		plugin, err := p.getPlugin(namespace, kongPluginName)
 		if err != nil {
-			glog.Errorf("reading KongPlugin '%v/%v': %v", namespace,
+			glog.Errorf("fetching KongPlugin '%v/%v': %v", namespace,
 				kongPluginName, err)
 			continue
 		}
@@ -998,7 +998,7 @@ func (p *Parser) fillPlugins(state KongState) []Plugin {
 func (p *Parser) globalPlugins() ([]Plugin, error) {
 	globalPlugins, err := p.store.ListGlobalKongPlugins()
 	if err != nil {
-		return nil, errors.Wrap(err, "error listing global plugins:")
+		return nil, errors.Wrap(err, "error listing global KongPlugins:")
 	}
 	res := make(map[string]Plugin)
 	var duplicates []string // keep track of duplicate
@@ -1028,6 +1028,31 @@ func (p *Parser) globalPlugins() ([]Plugin, error) {
 			Plugin: kongPluginFromK8SPlugin(k8sPlugin),
 		}
 	}
+
+	globalClusterPlugins, err := p.store.ListGlobalKongClusterPlugins()
+	if err != nil {
+		return nil, errors.Wrap(err, "error listing global KongClusterPlugins")
+	}
+	for i := 0; i < len(globalClusterPlugins); i++ {
+		k8sPlugin := *globalClusterPlugins[i]
+		pluginName := k8sPlugin.PluginName
+		// empty pluginName skip it
+		if pluginName == "" {
+			glog.Errorf("KongPlugin '%v' does not specify a plugin name",
+				k8sPlugin.Name)
+			continue
+		}
+		if _, ok := res[pluginName]; ok {
+			glog.Error("Multiple KongPlugin definitions found with"+
+				" 'global' annotation for '", pluginName,
+				"', the plugin will not be applied")
+			duplicates = append(duplicates, pluginName)
+			continue
+		}
+		res[pluginName] = Plugin{
+			Plugin: kongPluginFromK8SClusterPlugin(k8sPlugin),
+		}
+	}
 	for _, plugin := range duplicates {
 		delete(res, plugin)
 	}
@@ -1123,7 +1148,28 @@ func (p *Parser) getPlugin(namespace, name string) (kong.Plugin, error) {
 	var plugin kong.Plugin
 	k8sPlugin, err := p.store.GetKongPlugin(namespace, name)
 	if err != nil {
-		return plugin, errors.Wrapf(err, "fetching KongPlugin")
+		// if no namespaced plugin definition, then
+		// search for cluster level-plugin definition
+		if errors.As(err, &store.ErrNotFound{}) {
+			clusterPlugin, err := p.store.GetKongClusterPlugin(name)
+			// not found
+			if errors.As(err, &store.ErrNotFound{}) {
+				return plugin, errors.New(
+					"no KongPlugin or KongClusterPlugin was found")
+			}
+			if err != nil {
+				return plugin, err
+			}
+			if clusterPlugin.PluginName == "" {
+				return plugin, errors.Errorf("invalid empty 'plugin' property")
+			}
+			plugin = kongPluginFromK8SClusterPlugin(*clusterPlugin)
+			return plugin, err
+		}
+		// handle other errors
+		if err != nil {
+			return plugin, err
+		}
 	}
 	// ignore plugins with no name
 	if k8sPlugin.PluginName == "" {
@@ -1133,21 +1179,53 @@ func (p *Parser) getPlugin(namespace, name string) (kong.Plugin, error) {
 	return plugin, nil
 }
 
-func kongPluginFromK8SPlugin(k8sPlugin configurationv1.KongPlugin) kong.Plugin {
-	plugin := kong.Plugin{
-		Name:   kong.String(k8sPlugin.PluginName),
-		Config: kong.Configuration(k8sPlugin.Config).DeepCopy(),
+// plugin is a intermediate type to hold plugin related configuration
+type plugin struct {
+	Name   string
+	Config configurationv1.Configuration
+
+	RunOn     string
+	Disabled  bool
+	Protocols []string
+}
+
+func toKongPlugin(plugin plugin) kong.Plugin {
+	result := kong.Plugin{
+		Name:   kong.String(plugin.Name),
+		Config: kong.Configuration(*plugin.Config.DeepCopy()),
 	}
-	if k8sPlugin.RunOn != "" {
-		plugin.RunOn = kong.String(k8sPlugin.RunOn)
+	if plugin.RunOn != "" {
+		result.RunOn = kong.String(plugin.RunOn)
 	}
-	if k8sPlugin.Disabled {
-		plugin.Enabled = kong.Bool(false)
+	if plugin.Disabled {
+		result.Enabled = kong.Bool(false)
 	}
-	if len(k8sPlugin.Protocols) > 0 {
-		plugin.Protocols = kong.StringSlice(k8sPlugin.Protocols...)
+	if len(plugin.Protocols) > 0 {
+		result.Protocols = kong.StringSlice(plugin.Protocols...)
 	}
-	return plugin
+	return result
+}
+
+func kongPluginFromK8SClusterPlugin(k8sPlugin configurationv1.KongClusterPlugin) kong.Plugin {
+	return toKongPlugin(plugin{
+		Name:   k8sPlugin.PluginName,
+		Config: k8sPlugin.Config,
+
+		RunOn:     k8sPlugin.RunOn,
+		Disabled:  k8sPlugin.Disabled,
+		Protocols: k8sPlugin.Protocols,
+	})
+}
+
+func kongPluginFromK8SPlugin(k8sPlugin configurationv1.KongPlugin) kong.Plugin {
+	return toKongPlugin(plugin{
+		Name:   k8sPlugin.PluginName,
+		Config: k8sPlugin.Config,
+
+		RunOn:     k8sPlugin.RunOn,
+		Disabled:  k8sPlugin.Disabled,
+		Protocols: k8sPlugin.Protocols,
+	})
 }
 
 // getEndpoints returns a list of <endpoint ip>:<port> for a given service/target port combination.