Expose MQTT and STOMP endpoints (#292)

* Expose MQTT and STOMP ports when these plugins are enabled

Resolves #271

Author: ansd

Makefile

@@ -14,7 +14,6 @@ list:    ## list Makefile targets
 unit-tests: generate fmt vet manifests ## Run unit tests
 	ginkgo -r api/ internal/
 
-
 integration-tests: generate fmt vet manifests ## Run integration tests
 	ginkgo -r controllers/
 

api/v1beta1/rabbitmqcluster_types.go

@@ -233,8 +233,10 @@ type TLSSpec struct {
 	SecretName string `json:"secretName,omitempty"`
 	// Name of a Secret in the same Namespace as the RabbitmqCluster, containing the Certificate Authority's public certificate for TLS.
 	// This can be the same as SecretName.
+	// Used for mTLS.
 	CaSecretName string `json:"caSecretName,omitempty"`
 	// The Secret defined in CaSecretName must store the Certificate Authority's public certificate under the key specified in CaCertName.
+	// Used for mTLS.
 	CaCertName string `json:"caCertName,omitempty"`
 }
 
@@ -314,6 +316,15 @@ func (cluster *RabbitmqCluster) SingleTLSSecret() bool {
 	return cluster.MutualTLSEnabled() && cluster.Spec.TLS.CaSecretName == cluster.Spec.TLS.SecretName
 }
 
+func (cluster *RabbitmqCluster) AdditionalPluginEnabled(plugin Plugin) bool {
+	for _, p := range cluster.Spec.Rabbitmq.AdditionalPlugins {
+		if p == plugin {
+			return true
+		}
+	}
+	return false
+}
+
 func (rmqStatus *RabbitmqClusterStatus) SetConditions(resources []runtime.Object) {
 	var oldAllPodsReadyCondition *status.RabbitmqClusterCondition
 	var oldClusterAvailableCondition *status.RabbitmqClusterCondition

config/crd/bases/rabbitmq.com_rabbitmqclusters.yaml

@@ -8254,12 +8254,12 @@ spec:
                   caCertName:
                     description: The Secret defined in CaSecretName must store the
                       Certificate Authority's public certificate under the key specified
-                      in CaCertName.
+                      in CaCertName. Used for mTLS.
                     type: string
                   caSecretName:
                     description: Name of a Secret in the same Namespace as the RabbitmqCluster,
                       containing the Certificate Authority's public certificate for
-                      TLS. This can be the same as SecretName.
+                      TLS. This can be the same as SecretName. Used for mTLS.
                     type: string
                   secretName:
                     description: Name of a Secret in the same Namespace as the RabbitmqCluster,

controllers/rabbitmqcluster_controller.go

@@ -355,7 +355,7 @@ func (r *RabbitmqClusterReconciler) allReplicasReady(ctx context.Context, rmq *r
 // enablePlugins - helper function to set the list of enabled plugins in a given RabbitmqCluster pods
 // `rabbitmq-plugins set` disables plugins that are not in the provided list
 func (r *RabbitmqClusterReconciler) enablePlugins(rmq *rabbitmqv1beta1.RabbitmqCluster) error {
-	plugins := resource.NewRabbitMQPlugins(rmq.Spec.Rabbitmq.AdditionalPlugins)
+	plugins := resource.NewRabbitmqPlugins(rmq.Spec.Rabbitmq.AdditionalPlugins)
 	for i := int32(0); i < *rmq.Spec.Replicas; i++ {
 		podName := fmt.Sprintf("%s-%d", rmq.ChildResourceName("server"), i)
 		rabbitCommand := fmt.Sprintf("rabbitmq-plugins set %s", plugins.AsString(" "))

go.mod

@@ -8,13 +8,16 @@ require (
 	github.com/Azure/go-autorest/autorest/adal v0.8.0 // indirect
 	github.com/cespare/xxhash/v2 v2.1.1 // indirect
 	github.com/cloudflare/cfssl v1.4.1
+	github.com/eclipse/paho.mqtt.golang v1.2.0
 	github.com/go-logr/logr v0.1.0
 	github.com/go-logr/zapr v0.1.1 // indirect
+	github.com/go-stomp/stomp v2.0.6+incompatible
 	github.com/golang/groupcache v0.0.0-20191027212112-611e8accdfc9 // indirect
 	github.com/gophercloud/gophercloud v0.5.0 // indirect
 	github.com/onsi/ginkgo v1.14.0
 	github.com/onsi/gomega v1.10.1
 	github.com/prometheus/client_golang v1.2.1 // indirect
+	github.com/smartystreets/goconvey v1.6.4 // indirect
 	github.com/streadway/amqp v0.0.0-20200108173154-1c71cc93ed71
 	go.uber.org/multierr v1.2.0 // indirect
 	golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7
@@ -26,5 +29,4 @@ require (
 	k8s.io/apimachinery v0.18.6
 	k8s.io/client-go v0.18.6
 	sigs.k8s.io/controller-runtime v0.6.2
-	sigs.k8s.io/structured-merge-diff v0.0.0-20190525122527-15d366b2352e // indirect
 )

go.sum

@@ -55,7 +55,7 @@ func (builder *ClientServiceBuilder) Update(object runtime.Object) error {
 	service.Spec.Type = corev1.ServiceType(builder.Instance.Spec.Service.Type)
 	service.Spec.Selector = metadata.LabelSelector(builder.Instance.Name)
 
-	service.Spec.Ports = updatePorts(service.Spec.Ports, builder.Instance.TLSEnabled())
+	service.Spec.Ports = builder.updatePorts(service.Spec.Ports)
 
 	if builder.Instance.Spec.Service.Type == "ClusterIP" || builder.Instance.Spec.Service.Type == "" {
 		for i := range service.Spec.Ports {
@@ -108,20 +108,48 @@ func applySvcOverride(svc *corev1.Service, override *rabbitmqv1beta1.ClientServi
 	return nil
 }
 
-func updatePorts(servicePorts []corev1.ServicePort, tlsEnabled bool) []corev1.ServicePort {
+func (builder *ClientServiceBuilder) updatePorts(servicePorts []corev1.ServicePort) []corev1.ServicePort {
 	servicePortsMap := map[string]corev1.ServicePort{
-		"amqp": corev1.ServicePort{
+		"amqp": {
 			Protocol: corev1.ProtocolTCP,
 			Port:     5672,
 			Name:     "amqp",
 		},
-		"management": corev1.ServicePort{
+		"management": {
 			Protocol: corev1.ProtocolTCP,
 			Port:     15672,
 			Name:     "management",
 		},
 	}
-	if tlsEnabled {
+	if builder.Instance.AdditionalPluginEnabled("rabbitmq_mqtt") {
+		servicePortsMap["mqtt"] = corev1.ServicePort{
+			Protocol: corev1.ProtocolTCP,
+			Port:     1883,
+			Name:     "mqtt",
+		}
+	}
+	if builder.Instance.AdditionalPluginEnabled("rabbitmq_web_mqtt") {
+		servicePortsMap["web-mqtt"] = corev1.ServicePort{
+			Protocol: corev1.ProtocolTCP,
+			Port:     15675,
+			Name:     "web-mqtt",
+		}
+	}
+	if builder.Instance.AdditionalPluginEnabled("rabbitmq_stomp") {
+		servicePortsMap["stomp"] = corev1.ServicePort{
+			Protocol: corev1.ProtocolTCP,
+			Port:     61613,
+			Name:     "stomp",
+		}
+	}
+	if builder.Instance.AdditionalPluginEnabled("rabbitmq_web_stomp") {
+		servicePortsMap["web-stomp"] = corev1.ServicePort{
+			Protocol: corev1.ProtocolTCP,
+			Port:     15674,
+			Name:     "web-stomp",
+		}
+	}
+	if builder.Instance.TLSEnabled() {
 		servicePortsMap["amqps"] = corev1.ServicePort{
 			Protocol: corev1.ProtocolTCP,
 			Port:     5671,

internal/resource/client_service.go

@@ -11,6 +11,7 @@ package resource_test
 
 import (
 	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/ginkgo/extensions/table"
 	. "github.com/onsi/gomega"
 	rabbitmqv1beta1 "github.com/rabbitmq/cluster-operator/api/v1beta1"
 	"github.com/rabbitmq/cluster-operator/internal/resource"
@@ -28,7 +29,7 @@ var _ = Context("ClientServices", func() {
 		scheme   *runtime.Scheme
 	)
 
-	Context("Build", func() {
+	Describe("Build", func() {
 		BeforeEach(func() {
 			scheme = runtime.NewScheme()
 			Expect(rabbitmqv1beta1.AddToScheme(scheme)).To(Succeed())
@@ -57,7 +58,7 @@ var _ = Context("ClientServices", func() {
 		})
 	})
 
-	Context("Update", func() {
+	Describe("Update", func() {
 		BeforeEach(func() {
 			scheme = runtime.NewScheme()
 			Expect(rabbitmqv1beta1.AddToScheme(scheme)).To(Succeed())
@@ -299,9 +300,27 @@ var _ = Context("ClientServices", func() {
 					Port:     15672,
 					Protocol: corev1.ProtocolTCP,
 				}
-				Expect(svc.Spec.Ports).Should(ConsistOf(amqpPort, managementPort))
+				Expect(svc.Spec.Ports).To(ConsistOf(amqpPort, managementPort))
 			})
 
+			DescribeTable("plugins exposing ports",
+				func(plugin, servicePortName string, port int) {
+					instance.Spec.Rabbitmq.AdditionalPlugins = []rabbitmqv1beta1.Plugin{rabbitmqv1beta1.Plugin(plugin)}
+					Expect(serviceBuilder.Update(svc)).To(Succeed())
+
+					expectedPort := corev1.ServicePort{
+						Name:     servicePortName,
+						Port:     int32(port),
+						Protocol: corev1.ProtocolTCP,
+					}
+					Expect(svc.Spec.Ports).To(ContainElement(expectedPort))
+				},
+				Entry("MQTT", "rabbitmq_mqtt", "mqtt", 1883),
+				Entry("MQTT-over-WebSockets", "rabbitmq_web_mqtt", "web-mqtt", 15675),
+				Entry("STOMP", "rabbitmq_stomp", "stomp", 61613),
+				Entry("STOMP-over-WebSockets", "rabbitmq_web_stomp", "web-stomp", 15674),
+			)
+
 			It("updates the service type from ClusterIP to NodePort", func() {
 				svc.Spec.Type = corev1.ServiceTypeClusterIP
 				serviceBuilder.Instance.Spec.Service.Type = "NodePort"

internal/resource/client_service_test.go

@@ -18,7 +18,7 @@ var requiredPlugins = []string{
 
 const pluginsConfig = "plugins-conf"
 
-type RabbitMQPlugins struct {
+type RabbitmqPlugins struct {
 	requiredPlugins   []string
 	additionalPlugins []string
 }
@@ -27,19 +27,19 @@ type RabbitmqPluginsConfigMapBuilder struct {
 	Instance *rabbitmqv1beta1.RabbitmqCluster
 }
 
-func NewRabbitMQPlugins(plugins []rabbitmqv1beta1.Plugin) RabbitMQPlugins {
+func NewRabbitmqPlugins(plugins []rabbitmqv1beta1.Plugin) RabbitmqPlugins {
 	additionalPlugins := make([]string, len(plugins))
 	for i := range additionalPlugins {
 		additionalPlugins[i] = string(plugins[i])
 	}
 
-	return RabbitMQPlugins{
+	return RabbitmqPlugins{
 		requiredPlugins:   requiredPlugins,
 		additionalPlugins: additionalPlugins,
 	}
 }
 
-func (r *RabbitMQPlugins) DesiredPlugins() []string {
+func (r *RabbitmqPlugins) DesiredPlugins() []string {
 	allPlugins := append(r.requiredPlugins, r.additionalPlugins...)
 
 	check := make(map[string]bool)
@@ -53,7 +53,7 @@ func (r *RabbitMQPlugins) DesiredPlugins() []string {
 	return enabledPlugins
 }
 
-func (r *RabbitMQPlugins) AsString(sep string) string {
+func (r *RabbitmqPlugins) AsString(sep string) string {
 	return strings.Join(r.DesiredPlugins(), sep)
 }
 
@@ -92,6 +92,6 @@ func (builder *RabbitmqPluginsConfigMapBuilder) Build() (runtime.Object, error)
 }
 
 func desiredPluginsAsString(additionalPlugins []rabbitmqv1beta1.Plugin) string {
-	plugins := NewRabbitMQPlugins(additionalPlugins)
+	plugins := NewRabbitmqPlugins(additionalPlugins)
 	return "[" + plugins.AsString(",") + "]."
 }

internal/resource/rabbitmq_plugins.go

@@ -25,15 +25,15 @@ var _ = Describe("RabbitMQPlugins", func() {
 	Context("DesiredPlugins", func() {
 		When("AdditionalPlugins is empty", func() {
 			It("returns list of required plugins", func() {
-				plugins := NewRabbitMQPlugins(nil)
+				plugins := NewRabbitmqPlugins(nil)
 				Expect(plugins.DesiredPlugins()).To(ConsistOf([]string{"rabbitmq_peer_discovery_k8s", "rabbitmq_prometheus", "rabbitmq_management"}))
 			})
 		})
 
 		When("AdditionalPlugins are provided", func() {
 			It("returns a concatenated list of plugins", func() {
 				morePlugins := []rabbitmqv1beta1.Plugin{"rabbitmq_shovel", "my_great_plugin"}
-				plugins := NewRabbitMQPlugins(morePlugins)
+				plugins := NewRabbitmqPlugins(morePlugins)
 
 				Expect(plugins.DesiredPlugins()).To(ConsistOf([]string{"rabbitmq_peer_discovery_k8s",
 					"rabbitmq_prometheus",
@@ -47,7 +47,7 @@ var _ = Describe("RabbitMQPlugins", func() {
 		When("AdditionalPlugins are provided with duplicates", func() {
 			It("returns a unique list of plugins", func() {
 				morePlugins := []rabbitmqv1beta1.Plugin{"rabbitmq_management", "rabbitmq_shovel", "my_great_plugin", "rabbitmq_shovel"}
-				plugins := NewRabbitMQPlugins(morePlugins)
+				plugins := NewRabbitmqPlugins(morePlugins)
 
 				Expect(plugins.DesiredPlugins()).To(ConsistOf([]string{"rabbitmq_peer_discovery_k8s",
 					"rabbitmq_prometheus",