feat: add mqtt-proxy plugin in ApisixRoute (#1056)

.github/workflows/spell-checker.yml

@@ -35,5 +35,5 @@ jobs:
           wget -O - -q https://git.io/misspell | sh -s -- -b .
       - name: Misspell
         run: |
-          find . -name "*.go" -type f | xargs ./misspell -error
+          find . -name "*.go" -type f | xargs ./misspell -i mosquitto -error
           find docs -type f | xargs ./misspell -error

docs/en/latest/references/apisix_route_v2.md

@@ -73,6 +73,10 @@ Meaning of each field in the spec of ApisixRoute are followed, the top level fie
 | stream[].backend.servicePort         | integer or string  | The backend service port, can be the port number or the name defined in the service object.                                                                                                                                       |
 | stream[].backend.resolveGranularity  | string             | See [Service Resolve Granularity](#service-resolve-granularity) for the details.                                                                                                                                                  |
 | stream[].backend.subset              | string             | Subset specifies a subset for the target Service. The subset should be pre-definedin ApisixUpstream about this service.                                                                                                           |
+| stream[].plugins                     | array              | A series of APISIX plugins that will be executed once this route rule is matched                                                                                                                                                  |
+| stream[].plugins[].name              | string             | The plugin name, see [docs](http://apisix.apache.org/docs/apisix/getting-started) for learning the available plugins.                                                                                                             |
+| stream[].plugins[].enable            | boolean            | Whether the plugin would be used                                                                                                                                                                                                  |
+| stream[].plugins[].config            | object             | The configuration of the plugin that must have the same fields as in APISIX.                                                                                                                                                      |
 
 ## Expression Operators
 

pkg/api/validation/apisix_route_test.go

@@ -124,7 +124,7 @@ func Test_validatePlugin(t *testing.T) {
 	fakeClient := newFakeSchemaClient()
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			gotValid, _ := validatePlugin(fakeClient, tt.pluginName, v2.ApisixRouteHTTPPluginConfig(tt.pluginConfig))
+			gotValid, _ := validatePlugin(fakeClient, tt.pluginName, v2.ApisixRoutePluginConfig(tt.pluginConfig))
 			if gotValid != tt.wantValid {
 				t.Errorf("validatePlugin() gotValid = %v, want %v", gotValid, tt.wantValid)
 			}

pkg/kube/apisix/apis/config/v2/types.go

@@ -70,7 +70,7 @@ type ApisixRouteHTTP struct {
 	Backends         []ApisixRouteHTTPBackend  `json:"backends,omitempty" yaml:"backends,omitempty"`
 	Websocket        bool                      `json:"websocket" yaml:"websocket"`
 	PluginConfigName string                    `json:"plugin_config_name,omitempty" yaml:"plugin_config_name,omitempty"`
-	Plugins          []ApisixRouteHTTPPlugin   `json:"plugins,omitempty" yaml:"plugins,omitempty"`
+	Plugins          []ApisixRoutePlugin       `json:"plugins,omitempty" yaml:"plugins,omitempty"`
 	Authentication   ApisixRouteAuthentication `json:"authentication,omitempty" yaml:"authentication,omitempty"`
 }
 
@@ -152,19 +152,19 @@ type ApisixRouteHTTPMatchExprSubject struct {
 	Name string `json:"name" yaml:"name"`
 }
 
-// ApisixRouteHTTPPlugin represents an APISIX plugin.
-type ApisixRouteHTTPPlugin struct {
+// ApisixRoutePlugin represents an APISIX plugin.
+type ApisixRoutePlugin struct {
 	// The plugin name.
 	Name string `json:"name" yaml:"name"`
 	// Whether this plugin is in use, default is true.
 	Enable bool `json:"enable" yaml:"enable"`
 	// Plugin configuration.
-	Config ApisixRouteHTTPPluginConfig `json:"config" yaml:"config"`
+	Config ApisixRoutePluginConfig `json:"config" yaml:"config"`
 }
 
-// ApisixRouteHTTPPluginConfig is the configuration for
+// ApisixRoutePluginConfig is the configuration for
 // any plugins.
-type ApisixRouteHTTPPluginConfig map[string]interface{}
+type ApisixRoutePluginConfig map[string]interface{}
 
 // ApisixRouteAuthentication is the authentication-related
 // configuration in ApisixRoute.
@@ -189,16 +189,16 @@ type ApisixRouteAuthenticationJwtAuth struct {
 	Cookie string `json:"cookie,omitempty" yaml:"cookie,omitempty"`
 }
 
-func (p ApisixRouteHTTPPluginConfig) DeepCopyInto(out *ApisixRouteHTTPPluginConfig) {
+func (p ApisixRoutePluginConfig) DeepCopyInto(out *ApisixRoutePluginConfig) {
 	b, _ := json.Marshal(&p)
 	_ = json.Unmarshal(b, out)
 }
 
-func (p *ApisixRouteHTTPPluginConfig) DeepCopy() *ApisixRouteHTTPPluginConfig {
+func (p *ApisixRoutePluginConfig) DeepCopy() *ApisixRoutePluginConfig {
 	if p == nil {
 		return nil
 	}
-	out := new(ApisixRouteHTTPPluginConfig)
+	out := new(ApisixRoutePluginConfig)
 	p.DeepCopyInto(out)
 	return out
 }
@@ -210,6 +210,7 @@ type ApisixRouteStream struct {
 	Protocol string                   `json:"protocol" yaml:"protocol"`
 	Match    ApisixRouteStreamMatch   `json:"match" yaml:"match"`
 	Backend  ApisixRouteStreamBackend `json:"backend" yaml:"backend"`
+	Plugins  []ApisixRoutePlugin      `json:"plugins,omitempty" yaml:"plugins,omitempty"`
 }
 
 // ApisixRouteStreamMatch represents the match conditions of stream route.
@@ -691,9 +692,9 @@ type ApisixPluginConfig struct {
 
 // ApisixPluginConfigSpec defines the desired state of ApisixPluginConfigSpec.
 type ApisixPluginConfigSpec struct {
-	// Plugins contains a list of ApisixRouteHTTPPlugin
+	// Plugins contains a list of ApisixRoutePlugin
 	// +required
-	Plugins []ApisixRouteHTTPPlugin `json:"plugins" yaml:"plugins"`
+	Plugins []ApisixRoutePlugin `json:"plugins" yaml:"plugins"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/providers/apisix/translation/apisix_route.go

@@ -818,6 +818,20 @@ func (t *translator) translateStreamRouteV2(ctx *translation.TranslateContext, a
 			)
 			return err
 		}
+
+		// add stream route plugins
+		pluginMap := make(apisixv1.Plugins)
+		for _, plugin := range part.Plugins {
+			if !plugin.Enable {
+				continue
+			}
+			if plugin.Config != nil {
+				pluginMap[plugin.Name] = plugin.Config
+			} else {
+				pluginMap[plugin.Name] = make(map[string]interface{})
+			}
+		}
+
 		sr := apisixv1.NewDefaultStreamRoute()
 		name := apisixv1.ComposeStreamRouteName(ar.Namespace, ar.Name, part.Name)
 		sr.ID = id.GenID(name)
@@ -827,6 +841,7 @@ func (t *translator) translateStreamRouteV2(ctx *translation.TranslateContext, a
 			return err
 		}
 		sr.UpstreamId = ups.ID
+		sr.Plugins = pluginMap
 		ctx.AddStreamRoute(sr)
 		if !ctx.CheckUpstreamExist(ups.Name) {
 			ctx.AddUpstream(ups)

pkg/types/apisix/v1/types.go

@@ -372,6 +372,7 @@ type StreamRoute struct {
 	SNI        string            `json:"sni,omitempty" yaml:"sni,omitempty"`
 	UpstreamId string            `json:"upstream_id,omitempty" yaml:"upstream_id,omitempty"`
 	Upstream   *Upstream         `json:"upstream,omitempty" yaml:"upstream,omitempty"`
+	Plugins    Plugins           `json:"plugins,omitempty" yaml:"plugins,omitempty"`
 }
 
 // GlobalRule represents the global_rule object in APISIX.

samples/deploy/crd/v1/ApisixRoute.yaml

@@ -821,6 +821,22 @@ spec:
                         required:
                           - serviceName
                           - servicePort
+                      plugins:
+                        type: array
+                        items:
+                          type: object
+                          properties:
+                            name:
+                              type: string
+                              minLength: 1
+                            enable:
+                              type: boolean
+                            config:
+                              type: object
+                              x-kubernetes-preserve-unknown-fields: true # we have to enable it since plugin config
+                        required:
+                          - name
+                          - enable
             status:
               type: object
               properties:

test/e2e/go.mod

@@ -5,6 +5,7 @@ go 1.18
 require (
 	github.com/apache/apisix-ingress-controller v0.0.0-20210105024109-72e53386de5a
 	github.com/apache/apisix-ingress-controller/test/e2e/testbackend v0.0.0
+	github.com/eclipse/paho.mqtt.golang v1.3.5
 	github.com/gavv/httpexpect/v2 v2.3.1
 	github.com/gorilla/websocket v1.5.0
 	github.com/gruntwork-io/terratest v0.40.19

test/e2e/go.sum

@@ -101,6 +101,11 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
+github.com/dustin/go-humanize v0.0.0-20171111073723-bb3d318650d4/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
+github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
+github.com/eclipse/paho.mqtt.golang v1.3.5 h1:sWtmgNxYM9P2sP+xEItMozsR3w0cqZFlqnNN1bdl41Y=
+github.com/eclipse/paho.mqtt.golang v1.3.5/go.mod h1:eTzb4gxwwyWpqBUHGQZ4ABAV7+Jgm1PklsYT/eo8Hcc=
+github.com/elazarl/goproxy v0.0.0-20170405201442-c4fc26588b6e/go.mod h1:/Zj4wYkgs4iZTTu3o/KG3Itv/qCCa8VVMlb3i9OVuzc=
 github.com/elazarl/goproxy v0.0.0-20180725130230-947c36da3153/go.mod h1:/Zj4wYkgs4iZTTu3o/KG3Itv/qCCa8VVMlb3i9OVuzc=
 github.com/elazarl/goproxy v0.0.0-20190911111923-ecfe977594f1 h1:yY9rWGoXv1U5pl4gxqlULARMQD7x0QG85lqEXTWysik=
 github.com/emicklei/go-restful v0.0.0-20170410110728-ff4f55a20633/go.mod h1:otzb+WCGbkyDHkqmQmT5YD2WR4BBwUdeQoFo8l/7tVs=
@@ -574,6 +579,7 @@ golang.org/x/net v0.0.0-20200222125558-5a598a2470a0/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200301022130-244492dfa37a/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200324143707-d3edc9973b7e/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
+golang.org/x/net v0.0.0-20200425230154-ff2c4b7c35a0/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200501053045-e0ff5e5a1de5/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200506145744-7e3656a0809f/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200513185701-a91f0712d120/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=

test/e2e/scaffold/scaffold.go

@@ -34,6 +34,7 @@ import (
 	"time"
 
 	"github.com/apache/apisix-ingress-controller/pkg/config"
+	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/gavv/httpexpect/v2"
 	"github.com/gruntwork-io/terratest/modules/k8s"
 	"github.com/gruntwork-io/terratest/modules/testing"
@@ -283,6 +284,13 @@ func (s *Scaffold) NewAPISIXClientWithTLSOverTCP(host string) *httpexpect.Expect
 	})
 }
 
+func (s *Scaffold) NewMQTTClient() mqtt.Client {
+	opts := mqtt.NewClientOptions()
+	opts.AddBroker(fmt.Sprintf("tcp://%s", s.apisixTCPTunnel.Endpoint()))
+	client := mqtt.NewClient(opts)
+	return client
+}
+
 func (s *Scaffold) DNSResolver() *net.Resolver {
 	return &net.Resolver{
 		PreferGo: false,