Extend FederationSpec with QueueType and ResourceCleanupMode (#974)

* Extend FederationSpec with QueueType and ResourceCleanupMode

Author: jaroslaw-bochniak

api/v1beta1/federation_types.go

@@ -40,6 +40,19 @@ type FederationSpec struct {
 	// +kubebuilder:validation:Enum=delete;retain
 	// +kubebuilder:default:=delete
 	DeletionPolicy string `json:"deletionPolicy,omitempty"`
+	// The queue type of the internal upstream queue used by exchange federation.
+	// Defaults to classic (a single replica queue type). Set to quorum to use a replicated queue type.
+	// Changing the queue type will delete and recreate the upstream queue by default.
+	// This may lead to messages getting lost or not routed anywhere during the re-declaration.
+	// To avoid that, set resource-cleanup-mode key to never.
+	// This requires manually deleting the old upstream queue so that it can be recreated with the new type.
+	// +kubebuilder:validation:Enum=classic;quorum
+	QueueType string `json:"queueType,omitempty"`
+	// Whether to delete the internal upstream queue when federation links stop.
+	// By default, the internal upstream queue is deleted immediately when a federation link stops.
+	// Set to never to keep the upstream queue around and collect messages even when changing federation configuration.
+	// +kubebuilder:validation:Enum=default;never
+	ResourceCleanupMode string `json:"resourceCleanupMode,omitempty"`
 }
 
 // FederationStatus defines the observed state of Federation

api/v1beta1/federation_types_test.go

@@ -64,14 +64,16 @@ var _ = Describe("Federation spec", func() {
 				UriSecret: &corev1.LocalObjectReference{
 					Name: "a-secret",
 				},
-				Expires:        1000,
-				MessageTTL:     1000,
-				MaxHops:        100,
-				PrefetchCount:  50,
-				ReconnectDelay: 10,
-				TrustUserId:    true,
-				Exchange:       "an-exchange",
-				AckMode:        "no-ack",
+				Expires:             1000,
+				MessageTTL:          1000,
+				MaxHops:             100,
+				PrefetchCount:       50,
+				ReconnectDelay:      10,
+				TrustUserId:         true,
+				Exchange:            "an-exchange",
+				AckMode:             "no-ack",
+				QueueType:           "quorum",
+				ResourceCleanupMode: "never",
 				RabbitmqClusterReference: RabbitmqClusterReference{
 					Name: "some-cluster",
 				},
@@ -99,6 +101,8 @@ var _ = Describe("Federation spec", func() {
 		Expect(fetched.Spec.MaxHops).To(Equal(100))
 		Expect(fetched.Spec.PrefetchCount).To(Equal(50))
 		Expect(fetched.Spec.ReconnectDelay).To(Equal(10))
+		Expect(fetched.Spec.QueueType).To(Equal("quorum"))
+		Expect(fetched.Spec.ResourceCleanupMode).To(Equal("never"))
 	})
 
 	When("creating a federation with an invalid 'AckMode' value", func() {

config/crd/bases/rabbitmq.com_federations.yaml

@@ -73,6 +73,18 @@ spec:
                 type: integer
               queue:
                 type: string
+              queueType:
+                description: |-
+                  The queue type of the internal upstream queue used by exchange federation.
+                  Defaults to classic (a single replica queue type). Set to quorum to use a replicated queue type.
+                  Changing the queue type will delete and recreate the upstream queue by default.
+                  This may lead to messages getting lost or not routed anywhere during the re-declaration.
+                  To avoid that, set resource-cleanup-mode key to never.
+                  This requires manually deleting the old upstream queue so that it can be recreated with the new type.
+                enum:
+                - classic
+                - quorum
+                type: string
               rabbitmqClusterReference:
                 description: |-
                   Reference to the RabbitmqCluster that this federation upstream will be created in.
@@ -108,6 +120,15 @@ spec:
                 type: object
               reconnectDelay:
                 type: integer
+              resourceCleanupMode:
+                description: |-
+                  Whether to delete the internal upstream queue when federation links stop.
+                  By default, the internal upstream queue is deleted immediately when a federation link stops.
+                  Set to never to keep the upstream queue around and collect messages even when changing federation configuration.
+                enum:
+                - default
+                - never
+                type: string
               trustUserId:
                 type: boolean
               uriSecret:

controllers/binding_controller.go

@@ -17,7 +17,7 @@ import (
 	"reflect"
 
 	"github.com/go-logr/logr"
-	rabbithole "github.com/michaelklishin/rabbit-hole/v2"
+	rabbithole "github.com/michaelklishin/rabbit-hole/v3"
 	topology "github.com/rabbitmq/messaging-topology-operator/api/v1beta1"
 	"github.com/rabbitmq/messaging-topology-operator/internal"
 	"github.com/rabbitmq/messaging-topology-operator/rabbitmqclient"

docs/api/rabbitmq.com.ref.asciidoc

@@ -496,6 +496,15 @@ Required property.
 | *`queue`* __string__ | 
 | *`deletionPolicy`* __string__ | DeletionPolicy defines the behavior of federation in the RabbitMQ cluster when the corresponding custom resource is deleted.
 Can be set to 'delete' or 'retain'. Default is 'delete'.
+| *`queueType`* __string__ | The queue type of the internal upstream queue used by exchange federation.
+Defaults to classic (a single replica queue type). Set to quorum to use a replicated queue type.
+Changing the queue type will delete and recreate the upstream queue by default.
+This may lead to messages getting lost or not routed anywhere during the re-declaration.
+To avoid that, set resource-cleanup-mode key to never.
+This requires manually deleting the old upstream queue so that it can be recreated with the new type.
+| *`resourceCleanupMode`* __string__ | Whether to delete the internal upstream queue when federation links stop.
+By default, the internal upstream queue is deleted immediately when a federation link stops.
+Set to never to keep the upstream queue around and collect messages even when changing federation configuration.
 |===
 
 

go.mod

@@ -9,9 +9,9 @@ require (
 	github.com/go-logr/logr v1.4.2
 	github.com/google/uuid v1.6.0
 	github.com/hashicorp/vault/api v1.16.0
-	github.com/michaelklishin/rabbit-hole/v2 v2.16.0
+	github.com/michaelklishin/rabbit-hole/v3 v3.2.0
 	github.com/onsi/ginkgo/v2 v2.23.3
-	github.com/onsi/gomega v1.36.3
+	github.com/onsi/gomega v1.37.0
 	github.com/rabbitmq/cluster-operator/v2 v2.12.1
 	gopkg.in/ini.v1 v1.67.0
 	k8s.io/api v0.32.3