Merge pull request #615 from lmiccini/fix-proxy-sigterm-handling

Fix proxy sigterm handling in amqp-proxy, enable CMR and override terminationGracePeriodSeconds

Author: openshift-merge-bot[bot]

apis/rabbitmq/v1beta1/rabbitmq_webhook.go

@@ -140,6 +140,16 @@ func (spec *RabbitMqSpecCore) Default(isNew bool) {
 		spec.QueueType = &queueType
 	}
 
+	// Migrate terminationGracePeriodSeconds from old default (604800) to new (60).
+	// The old value was inherited from rabbitmq-cluster-operator and causes pods
+	// to get stuck in Terminating state for up to a week.
+	oldGracePeriod := int64(604800)
+	newGracePeriod := int64(60)
+	if spec.TerminationGracePeriodSeconds != nil && *spec.TerminationGracePeriodSeconds == oldGracePeriod {
+		spec.TerminationGracePeriodSeconds = &newGracePeriod
+		rabbitmqlog.Info("Migrating terminationGracePeriodSeconds from old default 604800 to 60")
+	}
+
 	// Force Mirrored → Quorum when upgrading to RabbitMQ 4.x+.
 	// Mirrored queues are not supported in 4.x, so the migration is mandatory.
 	if spec.QueueType != nil && *spec.QueueType == QueueTypeMirrored &&

apis/rabbitmq/v1beta1/rabbitmq_webhook_test.go

@@ -0,0 +1,67 @@
+/*
+Copyright 2025.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1beta1
+
+import "testing"
+
+func TestSpecCoreDefault_MigratesOldGracePeriod(t *testing.T) {
+	oldDefault := int64(604800)
+	spec := RabbitMqSpecCore{
+		TerminationGracePeriodSeconds: &oldDefault,
+	}
+
+	spec.Default(false)
+
+	if spec.TerminationGracePeriodSeconds == nil {
+		t.Fatal("TerminationGracePeriodSeconds should not be nil")
+	}
+	if *spec.TerminationGracePeriodSeconds != 60 {
+		t.Errorf("TerminationGracePeriodSeconds = %d, want 60", *spec.TerminationGracePeriodSeconds)
+	}
+}
+
+func TestSpecCoreDefault_PreservesCustomGracePeriod(t *testing.T) {
+	custom := int64(120)
+	spec := RabbitMqSpecCore{
+		TerminationGracePeriodSeconds: &custom,
+	}
+
+	spec.Default(false)
+
+	if spec.TerminationGracePeriodSeconds == nil {
+		t.Fatal("TerminationGracePeriodSeconds should not be nil")
+	}
+	if *spec.TerminationGracePeriodSeconds != 120 {
+		t.Errorf("TerminationGracePeriodSeconds = %d, want 120", *spec.TerminationGracePeriodSeconds)
+	}
+}
+
+func TestSpecCoreDefault_PreservesNewDefault(t *testing.T) {
+	newDefault := int64(60)
+	spec := RabbitMqSpecCore{
+		TerminationGracePeriodSeconds: &newDefault,
+	}
+
+	spec.Default(false)
+
+	if spec.TerminationGracePeriodSeconds == nil {
+		t.Fatal("TerminationGracePeriodSeconds should not be nil")
+	}
+	if *spec.TerminationGracePeriodSeconds != 60 {
+		t.Errorf("TerminationGracePeriodSeconds = %d, want 60", *spec.TerminationGracePeriodSeconds)
+	}
+}

internal/controller/rabbitmq/data/proxy.py

@@ -18,6 +18,7 @@
 import asyncio
 import argparse
 import logging
+import signal
 import struct
 import sys
 import ssl
@@ -909,14 +910,21 @@ async def main():
     # Start periodic stats
     stats_task = asyncio.create_task(periodic_stats(proxy, args.stats_interval))
 
+    # Handle SIGTERM (sent by kubelet) the same as SIGINT so the proxy
+    # shuts down gracefully instead of hanging until the
+    # terminationGracePeriodSeconds expires.
+    loop = asyncio.get_running_loop()
+    loop.add_signal_handler(signal.SIGTERM, server.close)
+
     try:
         async with server:
             await server.serve_forever()
-    except KeyboardInterrupt:
+    except (KeyboardInterrupt, asyncio.CancelledError):
         logger.info("Shutting down...")
     finally:
         stats_task.cancel()
         proxy.print_stats()
+        logger.info("Proxy stopped")
 
 
 if __name__ == '__main__':

internal/rabbitmq/configmap.go

@@ -137,6 +137,20 @@ func buildOperatorDefaults(r *rabbitmqv1.RabbitMq, IPv6Enabled bool, configVersi
 		)
 	}
 
+	// Enable Continuous Membership Reconciliation (CMR) for multi-node clusters
+	// on RabbitMQ 4.x. CMR automatically grows quorum queues onto nodes that are
+	// missing from the membership, replacing the need for manual
+	// `rabbitmq-queues grow` calls after upgrades or rolling restarts.
+	// trigger_interval (10s) fires when a node joins, covering rolling restarts.
+	// interval (60min default) is a background safety net.
+	if IsVersion4OrLater(configVersion) && getReplicaCount(r) > 1 {
+		config = append(config,
+			"quorum_queue.continuous_membership_reconciliation.enabled            = true",
+			fmt.Sprintf("quorum_queue.continuous_membership_reconciliation.target_group_size = %d", getReplicaCount(r)),
+			"quorum_queue.continuous_membership_reconciliation.auto_remove        = false",
+		)
+	}
+
 	// Prometheus and management bind address
 	config = append(config, "prometheus.tcp.ip                          = ::")
 	config = append(config, "management.tcp.ip                          = ::")