Skip to content

Only enable AMQP proxy for Mirrored to Quorum queue migration #648

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lmiccini wants to merge 1 commit into
base: main
Choose a base branch
from
+74 −40
Open

Only enable AMQP proxy for Mirrored to Quorum queue migration #648

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion internal/controller/rabbitmq/rabbitmq_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -739,7 +739,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (result ct
// never connect to quorum queues without the durability proxy.
if instance.Status.ProxyRequired != "True" && instance.Spec.QueueType != nil && *instance.Spec.QueueType == rabbitmqv1beta1.QueueTypeQuorum {
isVersionUpgradeWithMigration := instance.Spec.TargetVersion != nil && *instance.Spec.TargetVersion != "" &&
rabbitmq.Is3xTo4xUpgrade(instance.Status.CurrentVersion, *instance.Spec.TargetVersion)
rabbitmq.Is3xTo4xUpgrade(instance.Status.CurrentVersion, *instance.Spec.TargetVersion) &&
instance.Status.QueueType == rabbitmqv1beta1.QueueTypeMirrored
isQueueTypeMigration := instance.Status.WipeReason == rabbitmqv1beta1.WipeReasonQueueTypeMigration

if isVersionUpgradeWithMigration || isQueueTypeMigration {
Expand Down
67 changes: 60 additions & 7 deletions test/functional/rabbitmq_controller_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1596,21 +1596,23 @@ var _ = Describe("RabbitMQ Controller", func() {
When("a 3.x to 4.x upgrade with Quorum queue type", func() {
BeforeEach(func() {
spec := GetDefaultRabbitMQSpec()
spec["queueType"] = "Mirrored"
rabbitmq := CreateRabbitMQ(rabbitmqName, spec)
DeferCleanup(th.DeleteInstance, rabbitmq)

SimulateRabbitMQClusterReady(rabbitmqName)

// Set CurrentVersion to 3.9 to simulate existing 3.x cluster
// Set CurrentVersion to 3.9 and QueueType to Mirrored to simulate existing 3.x cluster
Eventually(func(g Gomega) {
instance := &rabbitmqv1.RabbitMq{}
g.Expect(k8sClient.Get(ctx, rabbitmqName, instance)).Should(Succeed())
instance.Status.CurrentVersion = "3.9"
instance.Status.QueueType = rabbitmqv1.QueueTypeMirrored
g.Expect(th.K8sClient.Status().Update(ctx, instance)).Should(Succeed())
}, timeout, interval).Should(Succeed())
})

It("should set ProxyRequired for 3.x to 4.x upgrade with Quorum", func() {
It("should set ProxyRequired for 3.x to 4.x upgrade when migrating from Mirrored to Quorum", func() {
// Set TargetVersion and QueueType
Eventually(func(g Gomega) {
instance := &rabbitmqv1.RabbitMq{}
Expand All @@ -1635,6 +1637,52 @@ var _ = Describe("RabbitMQ Controller", func() {
g.Expect(instance.Status.ProxyRequired).To(Equal("True"))
}, timeout, interval).Should(Succeed())
})

})

When("a 3.x to 4.x upgrade on a cluster already using Quorum queues", func() {
BeforeEach(func() {
spec := GetDefaultRabbitMQSpec()
rabbitmq := CreateRabbitMQ(rabbitmqName, spec)
DeferCleanup(th.DeleteInstance, rabbitmq)

SimulateRabbitMQClusterReady(rabbitmqName)

// Set CurrentVersion to 3.9 but keep QueueType as Quorum
Eventually(func(g Gomega) {
instance := &rabbitmqv1.RabbitMq{}
g.Expect(k8sClient.Get(ctx, rabbitmqName, instance)).Should(Succeed())
instance.Status.CurrentVersion = "3.9"
instance.Status.QueueType = rabbitmqv1.QueueTypeQuorum
g.Expect(th.K8sClient.Status().Update(ctx, instance)).Should(Succeed())
}, timeout, interval).Should(Succeed())
})

It("should NOT set ProxyRequired for 3.x to 4.x upgrade when already on Quorum", func() {
// Set TargetVersion to trigger the 3.x→4.x upgrade
Eventually(func(g Gomega) {
instance := &rabbitmqv1.RabbitMq{}
g.Expect(k8sClient.Get(ctx, rabbitmqName, instance)).Should(Succeed())
instance.Spec.TargetVersion = ptr.To("4.2")
instance.Spec.QueueType = ptr.To(rabbitmqv1.QueueTypeQuorum)
g.Expect(th.K8sClient.Update(ctx, instance)).Should(Succeed())
}, timeout, interval).Should(Succeed())

// Wait for upgrade to reach WaitingForCluster
Eventually(func(g Gomega) {
instance := GetRabbitMQ(rabbitmqName)
g.Expect(string(instance.Status.UpgradePhase)).To(Equal(string(rabbitmqv1.UpgradePhaseWaitingForCluster)))
}, timeout, interval).Should(Succeed())

SimulateRabbitMQClusterReady(rabbitmqName)

// ProxyRequired should NOT be set — no Mirrored→Quorum migration needed
Eventually(func(g Gomega) {
instance := GetRabbitMQ(rabbitmqName)
g.Expect(instance.Status.CurrentVersion).To(Equal("4.2"))
g.Expect(instance.Status.ProxyRequired).ToNot(Equal("True"))
}, timeout, interval).Should(Succeed())
})
})

When("AnnotationClientsReconfigured is set", func() {
Expand Down Expand Up @@ -1880,17 +1928,19 @@ var _ = Describe("RabbitMQ Controller", func() {
When("a 3.x to 4.x upgrade includes proxy sidecar in StatefulSet", func() {
BeforeEach(func() {
spec := GetDefaultRabbitMQSpec()
spec["queueType"] = "Mirrored"
rabbitmq := CreateRabbitMQ(rabbitmqName, spec)
DeferCleanup(th.DeleteInstance, rabbitmq)

SimulateRabbitMQClusterReady(rabbitmqName)

// Set CurrentVersion to 3.9 and QueueType to Quorum
// Set CurrentVersion to 3.9 and QueueType to Mirrored to simulate a
// pre-upgrade cluster that needs Mirrored→Quorum migration
Eventually(func(g Gomega) {
instance := &rabbitmqv1.RabbitMq{}
g.Expect(k8sClient.Get(ctx, rabbitmqName, instance)).Should(Succeed())
instance.Status.CurrentVersion = "3.9"
instance.Status.QueueType = rabbitmqv1.QueueTypeQuorum
instance.Status.QueueType = rabbitmqv1.QueueTypeMirrored
g.Expect(th.K8sClient.Status().Update(ctx, instance)).Should(Succeed())
}, timeout, interval).Should(Succeed())
})
Expand Down Expand Up @@ -2081,27 +2131,30 @@ var _ = Describe("RabbitMQ Controller", func() {
When("a version upgrade completes the full state machine lifecycle", func() {
BeforeEach(func() {
spec := GetDefaultRabbitMQSpec()
spec["queueType"] = "Mirrored"
rabbitmq := CreateRabbitMQ(rabbitmqName, spec)
DeferCleanup(th.DeleteInstance, rabbitmq)

SimulateRabbitMQClusterReady(rabbitmqName)

// Set CurrentVersion to 3.9 and QueueType to Quorum
// Set CurrentVersion to 3.9 and QueueType to Mirrored to simulate pre-upgrade state
Eventually(func(g Gomega) {
instance := &rabbitmqv1.RabbitMq{}
g.Expect(k8sClient.Get(ctx, rabbitmqName, instance)).Should(Succeed())
instance.Status.CurrentVersion = "3.9"
instance.Status.QueueType = rabbitmqv1.QueueTypeQuorum
instance.Status.QueueType = rabbitmqv1.QueueTypeMirrored
g.Expect(th.K8sClient.Status().Update(ctx, instance)).Should(Succeed())
}, timeout, interval).Should(Succeed())
})

It("should transition through all upgrade phases and reach clean state", func() {
// Set TargetVersion to trigger upgrade
// Set TargetVersion and QueueType to Quorum to trigger upgrade
// (simulates what the webhook does: Mirrored + 4.x → Quorum)
Eventually(func(g Gomega) {
instance := &rabbitmqv1.RabbitMq{}
g.Expect(k8sClient.Get(ctx, rabbitmqName, instance)).Should(Succeed())
instance.Spec.TargetVersion = ptr.To("4.2")
instance.Spec.QueueType = ptr.To(rabbitmqv1.QueueTypeQuorum)
g.Expect(th.K8sClient.Update(ctx, instance)).Should(Succeed())
}, timeout, interval).Should(Succeed())

Expand Down
28 changes: 4 additions & 24 deletions test/kuttl/tests/rabbitmq-upgrade-no-tls/02-assert.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,30 +39,10 @@ commands:
echo "PASS: wipeReason is cleared"
- script: |
set -e
# Verify proxy sidecar container is present in StatefulSet
# Verify proxy sidecar container is NOT present (cluster was already on Quorum, no migration needed)
CONTAINERS=$(oc get statefulset -n $NAMESPACE rabbitmq-upgrade-notls-server -o jsonpath='{.spec.template.spec.containers[*].name}')
if ! echo "$CONTAINERS" | grep -qw "amqp-proxy"; then
echo "FAIL: Proxy sidecar container not found. Containers: $CONTAINERS"
if echo "$CONTAINERS" | grep -qw "amqp-proxy"; then
echo "FAIL: Proxy sidecar should not be present for version-only upgrade. Containers: $CONTAINERS"
exit 1
fi
echo "PASS: Proxy sidecar container present: $CONTAINERS"
- script: |
set -e
# Verify proxy listens on plain AMQP port 5672 (no TLS)
PROXY_PORT=$(oc get statefulset -n $NAMESPACE rabbitmq-upgrade-notls-server -o \
jsonpath='{.spec.template.spec.containers[?(@.name=="amqp-proxy")].ports[0].containerPort}')
if [ "$PROXY_PORT" != "5672" ]; then
echo "FAIL: Expected proxy to listen on port 5672 (plain AMQP), got: $PROXY_PORT"
exit 1
fi
echo "PASS: Proxy listens on plain AMQP port $PROXY_PORT"
- script: |
set -e
# Verify proxy command does NOT contain TLS args
ARGS=$(oc get statefulset -n $NAMESPACE rabbitmq-upgrade-notls-server -o \
jsonpath='{.spec.template.spec.containers[?(@.name=="amqp-proxy")].args}')
if echo "$ARGS" | grep -q "tls-cert"; then
echo "FAIL: Proxy should not have TLS args in non-TLS mode. Args: $ARGS"
exit 1
fi
echo "PASS: Proxy has no TLS args"
echo "PASS: No proxy sidecar (version-only upgrade on Quorum cluster): $CONTAINERS"
16 changes: 8 additions & 8 deletions test/kuttl/tests/rabbitmq-version-upgrade/02-assert.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,19 +42,19 @@ commands:
echo "PASS: wipeReason is cleared"
- script: |
set -e
# Verify proxyRequired is set (3.x to 4.x upgrade with Quorum queues)
# Verify proxyRequired is NOT set (cluster was already on Quorum, no migration needed)
PROXY=$(oc get rabbitmq.rabbitmq.openstack.org -n $NAMESPACE rabbitmq-upgrade -o jsonpath='{.status.proxyRequired}')
if [ "$PROXY" != "True" ]; then
echo "FAIL: Expected proxyRequired=True after 3.x→4.x upgrade, got: $PROXY"
if [ "$PROXY" = "True" ]; then
echo "FAIL: proxyRequired should not be True for version-only upgrade on Quorum cluster, got: $PROXY"
exit 1
fi
echo "PASS: proxyRequired is $PROXY"
echo "PASS: proxyRequired is not set (no Mirrored→Quorum migration)"
- script: |
set -e
# Verify proxy sidecar container is present in StatefulSet
# Verify proxy sidecar container is NOT present (no queue type migration)
CONTAINERS=$(oc get statefulset -n $NAMESPACE rabbitmq-upgrade-server -o jsonpath='{.spec.template.spec.containers[*].name}')
if ! echo "$CONTAINERS" | grep -qw "amqp-proxy"; then
echo "FAIL: Proxy sidecar container not found. Containers: $CONTAINERS"
if echo "$CONTAINERS" | grep -qw "amqp-proxy"; then
echo "FAIL: Proxy sidecar should not be present for version-only upgrade. Containers: $CONTAINERS"
exit 1
fi
echo "PASS: Proxy sidecar container present: $CONTAINERS"
echo "PASS: No proxy sidecar (version-only upgrade): $CONTAINERS"