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,7 +9,7 @@ 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/rabbitmq/cluster-operator/v2 v2.12.1

go.sum

@@ -148,8 +148,8 @@ github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovk
 github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
 github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
 github.com/mattn/go-sqlite3 v1.14.6/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
-github.com/michaelklishin/rabbit-hole/v2 v2.16.0 h1:RvTPW3DnmyR7R1XTbET0ILRneU1Ou3vzIVj4YHBIE/g=
-github.com/michaelklishin/rabbit-hole/v2 v2.16.0/go.mod h1:wnHAhXYVncuXKdF/3mdywRE4vzBgn4k07Z+HjdNGMpM=
+github.com/michaelklishin/rabbit-hole/v3 v3.2.0 h1:N4YdHFj36MP5059Csze9B4TTZPS6j6HPJm9bBeZgvJk=
+github.com/michaelklishin/rabbit-hole/v3 v3.2.0/go.mod h1:LTyucfaAV/Y++Y6aVfAmsc6lvKw3y0WEyQa+yPAXcXc=
 github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY=
@@ -293,6 +293,8 @@ golang.org/x/crypto v0.12.0/go.mod h1:NF0Gs7EO5K4qLn+Ylc+fih8BSTeIjAP05siRnAh98y
 golang.org/x/crypto v0.13.0/go.mod h1:y6Z2r+Rw4iayiXXAIxJIDAJ1zMW4yaTpebo8fPOliYc=
 golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
 golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
+golang.org/x/crypto v0.35.0 h1:b15kiHdrGCHrP6LvwaQ3c03kgNhhiMgvlhxHQhmg2Xs=
+golang.org/x/crypto v0.35.0/go.mod h1:dy7dXNW32cAb/6/PRuTNsix8T+vJAqvuIy5Bli/x0YQ=
 golang.org/x/crypto v0.36.0 h1:AnAEvhDddvBdpY+uR+MyHmuZzzNqXSe/GvuDeob5L34=
 golang.org/x/crypto v0.36.0/go.mod h1:Y4J0ReaxCR1IMaabaSMugxJES1EpwhBHhv2bDHklZvc=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
@@ -334,6 +336,8 @@ golang.org/x/net v0.14.0/go.mod h1:PpSgVXXLK0OxS0F31C1/tv6XNguvCrnXIDrFMspZIUI=
 golang.org/x/net v0.15.0/go.mod h1:idbUs1IY1+zTqbi8yxTbhexhEEk5ur9LInksu6HrEpk=
 golang.org/x/net v0.16.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
 golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
+golang.org/x/net v0.36.0 h1:vWF2fRbw4qslQsQzgFqZff+BItCvGFQqKzKIzx1rmoA=
+golang.org/x/net v0.36.0/go.mod h1:bFmbeoIPfrw4sMHNhb4J9f6+tPziuGjq7Jk/38fxi1I=
 golang.org/x/net v0.37.0 h1:1zLorHbz+LYj7MQlSf1+2tPIIgibq2eL5xkrGk6f+2c=
 golang.org/x/net v0.37.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8=
 golang.org/x/oauth2 v0.6.0/go.mod h1:ycmewcwgD4Rpr3eZJLSB4Kyyljb3qDh40vJ8STE5HKw=
@@ -389,6 +393,8 @@ golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.14.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
+golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik=
 golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
@@ -406,6 +412,8 @@ golang.org/x/term v0.11.0/go.mod h1:zC9APTIj3jG3FdV/Ons+XE1riIZXG4aZ4GTHiPZJPIU=
 golang.org/x/term v0.12.0/go.mod h1:owVbMEjm3cBLCHdkQu9b1opXd4ETQWc3BhuQGKgXgvU=
 golang.org/x/term v0.13.0/go.mod h1:LTmsnFJwVN6bCy1rVCoS+qHT1HhALEFxKncY3WNNh4U=
 golang.org/x/term v0.15.0/go.mod h1:BDl952bC7+uMoWR75FIrCDx79TPU9oHkTZ9yRbYOrX0=
+golang.org/x/term v0.29.0 h1:L6pJp37ocefwRRtYPKSWOWzOtWSxVajvz2ldH/xi3iU=
+golang.org/x/term v0.29.0/go.mod h1:6bl4lRlvVuDgSf3179VpIxBF0o10JUpXWOnI7nErv7s=
 golang.org/x/term v0.30.0 h1:PQ39fJZ+mfadBm0y5WlL4vlM7Sx1Hgf13sMIY2+QS9Y=
 golang.org/x/term v0.30.0/go.mod h1:NYYFdzHoI5wRh/h5tDMdMqCqPJZEuNqVR5xJLd/n67g=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -424,6 +432,8 @@ golang.org/x/text v0.11.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/text v0.12.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM=
+golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY=
 golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY=
 golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4=
 golang.org/x/time v0.9.0 h1:EsRrnYcQiGH+5FfbgvV4AP7qEZstoyrHB0DzarOQ4ZY=
@@ -464,6 +474,8 @@ google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQ
 google.golang.org/protobuf v1.28.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
+google.golang.org/protobuf v1.36.3 h1:82DV7MYdb8anAVi3qge1wSnMDrnKK7ebr+I0hHRN1BU=
+google.golang.org/protobuf v1.36.3/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
 google.golang.org/protobuf v1.36.5 h1:tPhr+woSbjfYvY6/GPufUoYizxw1cF/yFoxJ2fmpwlM=
 google.golang.org/protobuf v1.36.5/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

internal/binding.go

@@ -12,7 +12,7 @@ package internal
 import (
 	"encoding/json"
 	"fmt"
-	rabbithole "github.com/michaelklishin/rabbit-hole/v2"
+	rabbithole "github.com/michaelklishin/rabbit-hole/v3"
 	topology "github.com/rabbitmq/messaging-topology-operator/api/v1beta1"
 	"strings"
 )

internal/exchange_settings.go

@@ -12,7 +12,7 @@ package internal
 import (
 	"encoding/json"
 	"fmt"
-	rabbithole "github.com/michaelklishin/rabbit-hole/v2"
+	rabbithole "github.com/michaelklishin/rabbit-hole/v3"
 	topology "github.com/rabbitmq/messaging-topology-operator/api/v1beta1"
 )
 

internal/federation_definition.go

@@ -10,22 +10,24 @@ This product may include a number of subcomponents with separate copyright notic
 package internal
 
 import (
-	rabbithole "github.com/michaelklishin/rabbit-hole/v2"
+	rabbithole "github.com/michaelklishin/rabbit-hole/v3"
 	topology "github.com/rabbitmq/messaging-topology-operator/api/v1beta1"
 	"strings"
 )
 
 func GenerateFederationDefinition(f *topology.Federation, uri string) rabbithole.FederationDefinition {
 	return rabbithole.FederationDefinition{
-		Uri:            strings.Split(uri, ","),
-		Expires:        f.Spec.Expires,
-		MessageTTL:     int32(f.Spec.MessageTTL),
-		MaxHops:        f.Spec.MaxHops,
-		PrefetchCount:  f.Spec.PrefetchCount,
-		ReconnectDelay: f.Spec.ReconnectDelay,
-		AckMode:        f.Spec.AckMode,
-		TrustUserId:    f.Spec.TrustUserId,
-		Exchange:       f.Spec.Exchange,
-		Queue:          f.Spec.Queue,
+		Uri:                 strings.Split(uri, ","),
+		Expires:             f.Spec.Expires,
+		MessageTTL:          int32(f.Spec.MessageTTL),
+		MaxHops:             f.Spec.MaxHops,
+		PrefetchCount:       f.Spec.PrefetchCount,
+		ReconnectDelay:      f.Spec.ReconnectDelay,
+		AckMode:             f.Spec.AckMode,
+		TrustUserId:         f.Spec.TrustUserId,
+		Exchange:            f.Spec.Exchange,
+		Queue:               f.Spec.Queue,
+		QueueType:           f.Spec.QueueType,
+		ResourceCleanupMode: f.Spec.ResourceCleanupMode,
 	}
 }