From 675e3c9ffe2a02e05725f2927c031ab3c68dc082 Mon Sep 17 00:00:00 2001 From: Luca Miccini Date: Tue, 28 Jul 2026 17:02:07 +0200 Subject: [PATCH] =?UTF-8?q?Only=20enable=20AMQP=20proxy=20for=20Mirrored?= =?UTF-8?q?=E2=86=92Quorum=20queue=20migration?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The proxy sidecar was incorrectly enabled for all 3.x→4.x RabbitMQ upgrades, even when clusters were already using Quorum queues. The proxy is only needed when migrating from Mirrored to Quorum queues to handle the durability semantics difference during the transition. Add Status.QueueType == Mirrored check to the isVersionUpgradeWithMigration condition so version-only upgrades on existing Quorum clusters skip the unnecessary proxy sidecar. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../rabbitmq/rabbitmq_controller.go | 3 +- test/functional/rabbitmq_controller_test.go | 67 +++++++++++++++++-- .../rabbitmq-upgrade-no-tls/02-assert.yaml | 28 ++------ .../rabbitmq-version-upgrade/02-assert.yaml | 16 ++--- 4 files changed, 74 insertions(+), 40 deletions(-) diff --git a/internal/controller/rabbitmq/rabbitmq_controller.go b/internal/controller/rabbitmq/rabbitmq_controller.go index c862fcdb..5e1ee09c 100644 --- a/internal/controller/rabbitmq/rabbitmq_controller.go +++ b/internal/controller/rabbitmq/rabbitmq_controller.go @@ -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 { diff --git a/test/functional/rabbitmq_controller_test.go b/test/functional/rabbitmq_controller_test.go index d75c0b25..f024e75d 100644 --- a/test/functional/rabbitmq_controller_test.go +++ b/test/functional/rabbitmq_controller_test.go @@ -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{} @@ -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() { @@ -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()) }) @@ -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()) diff --git a/test/kuttl/tests/rabbitmq-upgrade-no-tls/02-assert.yaml b/test/kuttl/tests/rabbitmq-upgrade-no-tls/02-assert.yaml index fe251d8e..4a1cb8cd 100644 --- a/test/kuttl/tests/rabbitmq-upgrade-no-tls/02-assert.yaml +++ b/test/kuttl/tests/rabbitmq-upgrade-no-tls/02-assert.yaml @@ -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" diff --git a/test/kuttl/tests/rabbitmq-version-upgrade/02-assert.yaml b/test/kuttl/tests/rabbitmq-version-upgrade/02-assert.yaml index ec396f48..e58d7f67 100644 --- a/test/kuttl/tests/rabbitmq-version-upgrade/02-assert.yaml +++ b/test/kuttl/tests/rabbitmq-version-upgrade/02-assert.yaml @@ -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"