Add default queue type config to vhost crd

api/v1beta1/vhost_types.go

@@ -21,6 +21,10 @@ type VhostSpec struct {
 	Name    string   `json:"name"`
 	Tracing bool     `json:"tracing,omitempty"`
 	Tags    []string `json:"tags,omitempty"`
+	// Default queue type for this vhost; can be set to quorum, classic or stream.
+	// Supported in RabbitMQ 3.11.12 or above.
+	// +kubebuilder:validation:Enum=quorum;classic;stream
+	DefaultQueueType string `json:"defaultQueueType,omitempty"`
 	// Reference to the RabbitmqCluster that the vhost will be created in.
 	// Required property.
 	// +kubebuilder:validation:Required

api/v1beta1/vhost_types_test.go

@@ -99,4 +99,39 @@ var _ = Describe("Vhost", func() {
 			Name: "random-cluster",
 		}))
 	})
+
+	Context("Default queue types", func() {
+		var qTypeVhost = &Vhost{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "some-vhost",
+				Namespace: namespace,
+			},
+			Spec: VhostSpec{
+				Name: "some-vhost",
+				RabbitmqClusterReference: RabbitmqClusterReference{
+					Name: "random-cluster",
+				},
+			},
+		}
+		It("creates a vhost with default queue type configured", func() {
+			qTypeVhost.Spec.DefaultQueueType = "stream"
+			Expect(k8sClient.Create(ctx, qTypeVhost)).To(Succeed())
+
+			fetched := &Vhost{}
+			Expect(k8sClient.Get(ctx, types.NamespacedName{
+				Name:      qTypeVhost.Name,
+				Namespace: qTypeVhost.Namespace,
+			}, fetched)).To(Succeed())
+			Expect(fetched.Spec.DefaultQueueType).To(Equal("stream"))
+			Expect(fetched.Spec.RabbitmqClusterReference).To(Equal(RabbitmqClusterReference{
+				Name: "random-cluster",
+			}))
+		})
+
+		It("fails when default queue type is invalid", func() {
+			qTypeVhost.Spec.DefaultQueueType = "aqueuetype"
+			Expect(k8sClient.Create(ctx, qTypeVhost)).To(HaveOccurred())
+			Expect(k8sClient.Create(ctx, qTypeVhost)).To(MatchError(`Vhost.rabbitmq.com "some-vhost" is invalid: spec.defaultQueueType: Unsupported value: "aqueuetype": supported values: "quorum", "classic", "stream"`))
+		})
+	})
 })

api/v1beta1/vhost_webhook_test.go

@@ -15,8 +15,9 @@ var _ = Describe("vhost webhook", func() {
 			Name: "test-vhost",
 		},
 		Spec: VhostSpec{
-			Name:    "test",
-			Tracing: false,
+			Name:             "test",
+			Tracing:          false,
+			DefaultQueueType: "classic",
 			RabbitmqClusterReference: RabbitmqClusterReference{
 				Name: "a-cluster",
 			},
@@ -84,5 +85,11 @@ var _ = Describe("vhost webhook", func() {
 			newVhost.Spec.Tags = []string{"new-tag"}
 			Expect(newVhost.ValidateUpdate(&vhost)).To(Succeed())
 		})
+
+		It("allows updates on vhost.spec.defaultQueueType", func() {
+			newVhost := vhost.DeepCopy()
+			newVhost.Spec.DefaultQueueType = "quorum"
+			Expect(newVhost.ValidateUpdate(&vhost)).To(Succeed())
+		})
 	})
 })

config/crd/bases/rabbitmq.com_vhosts.yaml

@@ -37,6 +37,14 @@ spec:
           spec:
             description: VhostSpec defines the desired state of Vhost
             properties:
+              defaultQueueType:
+                description: Default queue type for this vhost; can be set to quorum,
+                  classic or stream. Supported in RabbitMQ 3.11.12 or above.
+                enum:
+                - quorum
+                - classic
+                - stream
+                type: string
               name:
                 description: Name of the vhost; see https://www.rabbitmq.com/vhosts.html.
                 type: string

docs/api/rabbitmq.com.ref.asciidoc

@@ -1226,6 +1226,7 @@ VhostSpec defines the desired state of Vhost
 | *`name`* __string__ | Name of the vhost; see https://www.rabbitmq.com/vhosts.html.
 | *`tracing`* __boolean__ | 
 | *`tags`* __string array__ | 
+| *`defaultQueueType`* __string__ | Default queue type for this vhost; can be set to quorum, classic or stream. Supported in RabbitMQ 3.11.12 or above.
 | *`rabbitmqClusterReference`* __xref:{anchor_prefix}-github-com-rabbitmq-messaging-topology-operator-api-v1beta1-rabbitmqclusterreference[$$RabbitmqClusterReference$$]__ | Reference to the RabbitmqCluster that the vhost will be created in. Required property.
 |===
 

docs/examples/vhosts/README.md

@@ -0,0 +1,4 @@
+# Vhost examples
+
+This section contains 1 example for creating a RabbitMQ vhost.
+Note that setting default queue type `spec.defaultQueueType` is only supported by RabbitMQ server version `3.11.12` or above.

docs/examples/vhosts/vhost.yaml

@@ -3,7 +3,8 @@ kind: Vhost
 metadata:
   name: test-vhost
 spec:
-  name: test-vhost # vhost name
+  name: test-vhost # vhost name; required and cannot be updated
+  defaultQueueType: quorum # default queue type for this vhost; require RabbitMQ version 3.11.12 or above
   rabbitmqClusterReference:
     name: test # rabbitmqCluster must exist in the same namespace as this resource
 # status:

internal/vhost_settings.go

@@ -16,7 +16,8 @@ import (
 
 func GenerateVhostSettings(v *topology.Vhost) *rabbithole.VhostSettings {
 	return &rabbithole.VhostSettings{
-		Tracing: v.Spec.Tracing,
-		Tags:    v.Spec.Tags,
+		Tracing:          v.Spec.Tracing,
+		Tags:             v.Spec.Tags,
+		DefaultQueueType: v.Spec.DefaultQueueType,
 	}
 }

internal/vhost_settings_test.go

@@ -32,4 +32,10 @@ var _ = Describe("GenerateVhostSettings", func() {
 		settings := internal.GenerateVhostSettings(v)
 		Expect(settings.Tags).To(ConsistOf("tag1", "tag2", "multi_dc_replication"))
 	})
+
+	It("sets default queue type according to vhost.spec.defaultQueueType", func() {
+		v.Spec.DefaultQueueType = "stream"
+		settings := internal.GenerateVhostSettings(v)
+		Expect(settings.DefaultQueueType).To(Equal("stream"))
+	})
 })

system_tests/utils.go

@@ -184,7 +184,8 @@ func basicTestRabbitmqCluster(name, namespace string) *rabbitmqv1beta1.RabbitmqC
 			Namespace: namespace,
 		},
 		Spec: rabbitmqv1beta1.RabbitmqClusterSpec{
-			Replicas: pointer.Int32Ptr(1),
+			Replicas: pointer.Int32(1),
+			Image:    "rabbitmq:3-management",
 			Resources: &corev1.ResourceRequirements{
 				Requests: corev1.ResourceList{
 					corev1.ResourceMemory: resource.MustParse("100Mi"),

system_tests/vhost_system_test.go

@@ -28,8 +28,9 @@ var _ = Describe("vhost", func() {
 				Namespace: namespace,
 			},
 			Spec: topology.VhostSpec{
-				Name: "test",
-				Tags: []string{"multi_dc_replication"},
+				Name:             "test",
+				Tags:             []string{"multi_dc_replication"},
+				DefaultQueueType: "stream",
 				RabbitmqClusterReference: topology.RabbitmqClusterReference{
 					Name: rmq.Name,
 				},
@@ -49,6 +50,7 @@ var _ = Describe("vhost", func() {
 		Expect(fetched.Tracing).To(BeFalse())
 		Expect(fetched.Tags).To(HaveLen(1))
 		Expect(fetched.Tags[0]).To(Equal("multi_dc_replication"))
+		Expect(fetched.DefaultQueueType).To(Equal("stream"))
 
 		By("updating status condition 'Ready'")
 		updatedVhost := topology.Vhost{}
@@ -76,6 +78,17 @@ var _ = Describe("vhost", func() {
 		updateTest.Spec.Name = "new-name"
 		Expect(k8sClient.Update(ctx, &updateTest).Error()).To(ContainSubstring("spec.name: Forbidden: updates on name and rabbitmqClusterReference are all forbidden"))
 
+		By("updating vhosts configuration")
+		Expect(k8sClient.Get(ctx, types.NamespacedName{Name: vhost.Name, Namespace: vhost.Namespace}, vhost)).To(Succeed())
+		vhost.Spec.Tracing = false
+		Expect(k8sClient.Update(ctx, vhost, &client.UpdateOptions{})).To(Succeed())
+		Eventually(func() bool {
+			var err error
+			fetched, err = rabbitClient.GetVhost(vhost.Spec.Name)
+			Expect(err).NotTo(HaveOccurred())
+			return fetched.Tracing
+		}, 30, 2).Should(BeFalse())
+
 		By("deleting a vhost")
 		Expect(k8sClient.Delete(ctx, vhost)).To(Succeed())
 		var err error