Feat/global plugins

internal/apis/admin/plugin.go

@@ -18,6 +18,7 @@ type PluginInterface interface {
 	Get(string) (*adminv1.Plugin, *APIResponse)
 	CreateInRoute(string, *adminv1.Plugin) (*adminv1.Plugin, *APIResponse)
 	CreateInService(string, *adminv1.Plugin) (*adminv1.Plugin, *APIResponse)
+	CreateGlobal(*adminv1.Plugin) (*adminv1.Plugin, *APIResponse)
 	Patch(string, *adminv1.Plugin) (*adminv1.Plugin, *APIResponse)
 	Delete(string) error
 
@@ -69,6 +70,12 @@ func (a *pluginAPI) CreateInService(id string, plugin *adminv1.Plugin) (*adminv1
 	return a.create("services", id, plugin)
 }
 
+func (a *pluginAPI) CreateGlobal(plugin *adminv1.Plugin) (*adminv1.Plugin, *APIResponse) {
+	out := &adminv1.Plugin{}
+	err := a.client.Create(plugin, out)
+	return out, err
+}
+
 func (a *pluginAPI) Patch(id string, route *adminv1.Plugin) (*adminv1.Plugin, *APIResponse) {
 	out := &adminv1.Plugin{}
 	err := a.client.Patch(id, route, out)

internal/apis/plugin/v1/types.go

@@ -27,6 +27,9 @@ type KongPlugin struct {
 
 	// Config contains the plugin configuration.
 	Config Configuration `json:"config,omitempty"`
+
+	// Name is the name of the plugin to which to apply the config
+	Name string `json:"name,omitempty"`
 }
 
 // KongPluginList is a top-level list type. The client methods for lists are automatically created.

internal/ingress/controller/kong.go

@@ -30,16 +30,16 @@ import (
 
 	"github.com/fatih/structs"
 	"github.com/golang/glog"
-	configurationv1 "github.com/kong/kubernetes-ingress-controller/internal/apis/configuration/v1"
-	"github.com/pkg/errors"
-	extensions "k8s.io/api/extensions/v1beta1"
-	"k8s.io/apimachinery/pkg/util/sets"
-
 	kong "github.com/kong/kubernetes-ingress-controller/internal/apis/admin"
 	kongadminv1 "github.com/kong/kubernetes-ingress-controller/internal/apis/admin/v1"
+	configurationv1 "github.com/kong/kubernetes-ingress-controller/internal/apis/configuration/v1"
+	pluginv1 "github.com/kong/kubernetes-ingress-controller/internal/apis/plugin/v1"
 	"github.com/kong/kubernetes-ingress-controller/internal/ingress"
 	"github.com/kong/kubernetes-ingress-controller/internal/ingress/annotations"
 	"github.com/kong/kubernetes-ingress-controller/internal/net/ssl"
+	"github.com/pkg/errors"
+	extensions "k8s.io/api/extensions/v1beta1"
+	"k8s.io/apimachinery/pkg/util/sets"
 )
 
 // OnUpdate is called periodically by syncQueue to keep the configuration in sync.
@@ -92,6 +92,11 @@ func (n *NGINXController) OnUpdate(ingressCfg *ingress.Configuration) error {
 		return err
 	}
 
+	err = n.syncGlobalPlugins()
+	if err != nil {
+		return err
+	}
+
 	checkServices, err := n.syncServices(ingressCfg)
 	if err != nil {
 		return err
@@ -114,6 +119,91 @@ func (n *NGINXController) OnUpdate(ingressCfg *ingress.Configuration) error {
 	return nil
 }
 
+func (n *NGINXController) syncGlobalPlugins() error {
+	glog.Infof("syncing global plugins")
+
+	targetGlobalPlugins, err := n.store.ListGlobalKongPlugins()
+	if err != nil {
+		return err
+	}
+	targetPluginMap := make(map[string]*pluginv1.KongPlugin)
+	var duplicates []string // keep track of duplicate
+
+	for i := 0; i < len(targetGlobalPlugins); i++ {
+		name := targetGlobalPlugins[i].Name
+		// empty name skip it
+		if name == "" {
+			continue
+		}
+		if _, ok := targetPluginMap[name]; ok {
+			glog.Error("Multiple KongPlugin definitions found with 'global' annotation for :", name,
+				", the plugin will not be applied")
+			duplicates = append(duplicates, name)
+			continue
+		}
+		targetPluginMap[name] = targetGlobalPlugins[i]
+	}
+
+	// remove duplicates
+	for _, plugin := range duplicates {
+		delete(targetPluginMap, plugin)
+	}
+
+	client := n.cfg.Kong.Client
+	plugins, err := client.Plugins().List(nil)
+	if err != nil {
+		return err
+	}
+
+	// plugins in Kong
+	currentGlobalPlugins := make(map[string]kongadminv1.Plugin)
+	for _, plugin := range plugins.Items {
+		if plugin.Route == "" && plugin.Service == "" && plugin.Consumer == "" {
+			currentGlobalPlugins[plugin.Name] = plugin
+		}
+	}
+
+	// sync plugins to Kong
+	for pluginName, kongPlugin := range targetPluginMap {
+		// plugin exists?
+		if pluginInKong, ok := currentGlobalPlugins[pluginName]; !ok {
+			// no, create it
+			p := &kongadminv1.Plugin{
+				Name:   pluginName,
+				Config: kongadminv1.Configuration(kongPlugin.Config),
+			}
+			_, res := client.Plugins().CreateGlobal(p)
+			if res.StatusCode != http.StatusCreated {
+				return errors.Wrap(res.Error(), fmt.Sprintf("creating a global Kong plugin %v", p))
+			}
+		} else {
+			// plugin exists, is the configuration up to date
+			if !pluginDeepEqual(kongPlugin.Config, &pluginInKong) {
+				// no, update it
+				p := &kongadminv1.Plugin{
+					Name:   pluginName,
+					Config: kongadminv1.Configuration(kongPlugin.Config),
+				}
+				_, res := client.Plugins().Patch(pluginInKong.ID, p)
+				if res.StatusCode != http.StatusOK {
+					return errors.Wrap(res.Error(), fmt.Sprintf("updating a global Kong plugin %v", p))
+				}
+			}
+		}
+		// remove from the current list, all that remain in the current list will be deleted
+		delete(currentGlobalPlugins, pluginName)
+	}
+
+	// delete the ones not configured in k8s
+	for _, plugin := range currentGlobalPlugins {
+		err := client.Plugins().Delete(plugin.ID)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // syncTargets reconciles the state between the ingress controller and
 // kong comparing the endpoints in Kubernetes and the targets in a
 // particular kong upstream. To avoid downtimes we create the new targets

internal/ingress/controller/store/store.go

@@ -22,6 +22,9 @@ import (
 	"sync"
 	"time"
 
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/selection"
+
 	"github.com/eapache/channels"
 	"github.com/golang/glog"
 
@@ -92,6 +95,8 @@ type Storer interface {
 	ListKongConsumers() []*consumerv1.KongConsumer
 
 	ListKongCredentials() []*credentialv1.KongCredential
+
+	ListGlobalKongPlugins() ([]*pluginv1.KongPlugin, error)
 }
 
 // EventType type of event associated with an informer
@@ -657,3 +662,24 @@ func (s k8sStore) ListKongCredentials() []*credentialv1.KongCredential {
 
 	return credentials
 }
+
+func (s k8sStore) ListGlobalKongPlugins() ([]*pluginv1.KongPlugin, error) {
+
+	var plugins []*pluginv1.KongPlugin
+	// var globalPlugins []*pluginv1.KongPlugin
+	req, err := labels.NewRequirement("global", selection.Equals, []string{"true"})
+	if err != nil {
+		return nil, err
+	}
+	err = cache.ListAll(s.listers.Kong.Plugin,
+		labels.NewSelector().Add(*req),
+		func(ob interface{}) {
+			if p, ok := ob.(*pluginv1.KongPlugin); ok {
+				plugins = append(plugins, p)
+			}
+		})
+	if err != nil {
+		return nil, err
+	}
+	return plugins, nil
+}