Open stream ports when osr plugin is enabled

- the osr plugin rabbitmq_multi_dc_replication enables
stream plugin as a dependendy and connect to stream port

Author: ChunyiLyu

api/v1beta1/rabbitmqcluster_types.go

@@ -351,6 +351,11 @@ func (cluster *RabbitmqCluster) AdditionalPluginEnabled(plugin Plugin) bool {
 	return false
 }
 
+// the OSR plugin `rabbitmq_multi_dc_replication` enables `rabbitmq_stream` as a dependency
+func (cluster *RabbitmqCluster) StreamNeeded() bool {
+	return cluster.AdditionalPluginEnabled("rabbitmq_stream") || cluster.AdditionalPluginEnabled("rabbitmq_multi_dc_replication")
+}
+
 // +kubebuilder:object:root=true
 
 // RabbitmqClusterList contains a list of RabbitmqClusters.

internal/resource/service.go

@@ -149,7 +149,7 @@ func (builder *ServiceBuilder) generateServicePortsMapOnlyTLSListeners() map[str
 		}
 	}
 
-	if builder.Instance.AdditionalPluginEnabled("rabbitmq_stream") {
+	if builder.Instance.StreamNeeded() {
 		servicePortsMap["streams"] = corev1.ServicePort{
 			Protocol:   corev1.ProtocolTCP,
 			Port:       5551,
@@ -232,7 +232,7 @@ func (builder *ServiceBuilder) generateServicePortsMap() map[string]corev1.Servi
 		}
 	}
 
-	if builder.Instance.AdditionalPluginEnabled("rabbitmq_stream") {
+	if builder.Instance.StreamNeeded() {
 		servicePortsMap["stream"] = corev1.ServicePort{
 			Protocol:   corev1.ProtocolTCP,
 			Port:       5552,
@@ -270,7 +270,7 @@ func (builder *ServiceBuilder) generateServicePortsMap() map[string]corev1.Servi
 				TargetPort: intstr.FromInt(8883),
 			}
 		}
-		if builder.Instance.AdditionalPluginEnabled("rabbitmq_stream") {
+		if builder.Instance.StreamNeeded() {
 			servicePortsMap["streams"] = corev1.ServicePort{
 				Protocol:   corev1.ProtocolTCP,
 				Port:       5551,

internal/resource/service_test.go

@@ -169,6 +169,21 @@ var _ = Context("Services", func() {
 				})
 			})
 
+			When("rabbitmq_multi_dc_replication is enabled", func() {
+				It("opens port for streams", func() {
+					instance.Spec.Rabbitmq.AdditionalPlugins = []rabbitmqv1beta1.Plugin{"rabbitmq_multi_dc_replication"}
+					Expect(serviceBuilder.Update(svc)).To(Succeed())
+					Expect(svc.Spec.Ports).To(ContainElements([]corev1.ServicePort{
+						{
+							Name:       "streams",
+							Protocol:   corev1.ProtocolTCP,
+							Port:       5551,
+							TargetPort: intstr.FromInt(5551),
+						},
+					}))
+				})
+			})
+
 			When("DisableNonTLSListeners is set to true", func() {
 				It("only exposes tls ports in the service", func() {
 					instance.Spec.TLS.DisableNonTLSListeners = true
@@ -234,6 +249,7 @@ var _ = Context("Services", func() {
 					Entry("STOMP", "rabbitmq_stomp", "stomps", 61614),
 					Entry("STOMP-over-WebSockets", "rabbitmq_web_stomp", "web-stomp-tls", 15673),
 					Entry("Stream", "rabbitmq_stream", "streams", 5551),
+					Entry("OSR", "rabbitmq_multi_dc_replication", "streams", 5551),
 				)
 			})
 		})
@@ -466,6 +482,7 @@ var _ = Context("Services", func() {
 				Entry("STOMP", "rabbitmq_stomp", "stomp", 61613),
 				Entry("STOMP-over-WebSockets", "rabbitmq_web_stomp", "web-stomp", 15674),
 				Entry("Stream", "rabbitmq_stream", "stream", 5552),
+				Entry("OSR", "rabbitmq_multi_dc_replication", "stream", 5552),
 			)
 
 			It("updates the service type from ClusterIP to NodePort", func() {

internal/resource/statefulset.go

@@ -762,7 +762,7 @@ func (builder *StatefulSetBuilder) updateContainerPorts() []corev1.ContainerPort
 		})
 	}
 
-	if builder.Instance.AdditionalPluginEnabled("rabbitmq_stream") {
+	if builder.Instance.StreamNeeded() {
 		ports = append(ports, corev1.ContainerPort{
 			Name:          "stream",
 			ContainerPort: 5552,
@@ -799,7 +799,7 @@ func (builder *StatefulSetBuilder) updateContainerPorts() []corev1.ContainerPort
 			})
 		}
 
-		if builder.Instance.AdditionalPluginEnabled("rabbitmq_stream") {
+		if builder.Instance.StreamNeeded() {
 			ports = append(ports, corev1.ContainerPort{
 				Name:          "streams",
 				ContainerPort: 5551,
@@ -860,7 +860,7 @@ func (builder *StatefulSetBuilder) updateContainerPortsOnlyTLSListeners() []core
 		})
 	}
 
-	if builder.Instance.AdditionalPluginEnabled("rabbitmq_stream") {
+	if builder.Instance.StreamNeeded() {
 		ports = append(ports, corev1.ContainerPort{
 			Name:          "streams",
 			ContainerPort: 5551,

internal/resource/statefulset_test.go

@@ -599,6 +599,21 @@ var _ = Describe("StatefulSet", func() {
 				}))
 			})
 
+			It("opens tls port for stream when rabbitmq_multi_dc_replication is enabled", func() {
+				instance.Spec.TLS.SecretName = "tls-secret"
+				instance.Spec.Rabbitmq.AdditionalPlugins = []rabbitmqv1beta1.Plugin{"rabbitmq_multi_dc_replication"}
+				Expect(stsBuilder.Update(statefulSet)).To(Succeed())
+
+				rabbitmqContainerSpec := extractContainer(statefulSet.Spec.Template.Spec.Containers, "rabbitmq")
+
+				Expect(rabbitmqContainerSpec.Ports).To(ContainElements([]corev1.ContainerPort{
+					{
+						Name:          "streams",
+						ContainerPort: 5551,
+					},
+				}))
+			})
+
 			When("Mutual TLS (same secret) is enabled", func() {
 				It("opens tls ports when rabbitmq_web_mqtt and rabbitmq_web_stomp are configured", func() {
 					instance.Spec.TLS.SecretName = "tls-secret"
@@ -685,7 +700,7 @@ var _ = Describe("StatefulSet", func() {
 				})
 
 				It("disables non tls ports for mqtt, stomp and stream if enabled", func() {
-					instance.Spec.Rabbitmq.AdditionalPlugins = []rabbitmqv1beta1.Plugin{"rabbitmq_mqtt", "rabbitmq_stomp", "rabbitmq_stream"}
+					instance.Spec.Rabbitmq.AdditionalPlugins = []rabbitmqv1beta1.Plugin{"rabbitmq_mqtt", "rabbitmq_stomp", "rabbitmq_stream", "rabbitmq_multi_dc_replication"}
 					Expect(stsBuilder.Update(statefulSet)).To(Succeed())
 
 					rabbitmqContainerSpec := extractContainer(statefulSet.Spec.Template.Spec.Containers, "rabbitmq")
@@ -808,6 +823,7 @@ var _ = Describe("StatefulSet", func() {
 			Entry("STOMP", "rabbitmq_stomp", "stomp", 61613),
 			Entry("STOMP-over-WebSockets", "rabbitmq_web_stomp", "web-stomp", 15674),
 			Entry("Stream", "rabbitmq_stream", "stream", 5552),
+			Entry("OSR", "rabbitmq_multi_dc_replication", "stream", 5552),
 		)
 
 		It("uses required Environment Variables", func() {