From 0d008d5d5f7498d567f117e5019c5af44033eb48 Mon Sep 17 00:00:00 2001 From: Luca Miccini Date: Tue, 28 Jul 2026 11:10:27 +0200 Subject: [PATCH] Prevent unnecessary RabbitMQ rolling restarts during scaling and cert rotation Three changes to align with the upstream rabbitmq-cluster-operator and prevent quorum queue corruption caused by unnecessary rolling restarts: 1. Remove TLS secret data and replica count from config hash Stop hashing TLS secret content (tlsSecretData, caSecretData) and the replica count into the pod-template annotation. The secret names are still hashed so that switching to a different cert triggers a restart, but content changes (cert-manager re-issuance, CA chain updates, rotation) no longer cause rolling restarts. TLS certificates are mounted via projected volumes that kubelet updates in-place, and RabbitMQ 4.x reloads them from disk automatically. During initial deployment the playbook scales RabbitMQ from 1 to 3 replicas, which adds new DNS SANs to the Certificate CR and triggers cert-manager to re-issue. When secret data was hashed, this caused a rolling restart before quorum queue replicas had synced, leading to Raft log divergence and a permanent crash loop (ra_server assertion "PLIdx < LastApplied"). 2. Make PreStop hook replica-count-independent The PreStop command previously varied based on spec.replicas (skipping quorum checks for single-replica clusters). This meant scaling from 1 to 3 changed the pod template, triggering a restart of existing pods. Now the quorum check is conditional at runtime: it queries rabbit_nodes:list_running() and only runs await_online_quorum_plus_one when more than one node is active. Per-command timeouts are derived from TerminationGracePeriodSeconds using a proportional split (T/6 for quorum and mirror checks, T/2 for drain) so that drain always gets a chance to run within the grace period. With the default 60s grace period this yields 10s/10s/30s. 3. Set explicit UpdateStrategy on the StatefulSet Always set RollingUpdate with Partition=0, matching the upstream cluster-operator. This prevents stale partition values (from migration or manual debugging) from silently blocking pod updates. 4. Add unit tests for PreStop hook and UpdateStrategy Cover default and custom TerminationGracePeriodSeconds values, and validate RollingUpdate with Partition=0. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../rabbitmq/rabbitmq_controller.go | 35 ++------- internal/rabbitmq/statefulset.go | 21 +++++- internal/rabbitmq/statefulset_test.go | 71 +++++++++++++++++++ test/functional/rabbitmq_controller_test.go | 14 ++-- 4 files changed, 103 insertions(+), 38 deletions(-) diff --git a/internal/controller/rabbitmq/rabbitmq_controller.go b/internal/controller/rabbitmq/rabbitmq_controller.go index c862fcdb..48ab4221 100644 --- a/internal/controller/rabbitmq/rabbitmq_controller.go +++ b/internal/controller/rabbitmq/rabbitmq_controller.go @@ -388,45 +388,22 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (result ct } // Calculate hash for config tracking. - // Include TLS secret content so that certificate rotation (e.g. by - // cert-manager) triggers a rolling restart of the StatefulSet pods. + // TLS cert content is NOT included: certs are mounted via projected + // volumes that kubelet updates in-place, and RabbitMQ 4.x reloads + // them automatically. Hashing cert data here would trigger a + // rolling restart on every cert-manager update, which can corrupt + // quorum queues when replicas haven't fully synced yet. hashInput := map[string]interface{}{ "ipv6Enabled": IPv6Enabled, "fipsEnabled": fipsEnabled, "tlsSecret": instance.Spec.TLS.SecretName, + "caSecret": instance.Spec.TLS.CaSecretName, "additionalConfig": instance.Spec.Rabbitmq.AdditionalConfig, "advancedConfig": instance.Spec.Rabbitmq.AdvancedConfig, "envConfig": instance.Spec.Rabbitmq.EnvConfig, "erlangInetConfig": instance.Spec.Rabbitmq.ErlangInetConfig, "additionalPlugins": instance.Spec.Rabbitmq.AdditionalPlugins, "containerImage": instance.Spec.ContainerImage, - "replicas": instance.Spec.Replicas, - } - if instance.Spec.TLS.SecretName != "" { - tlsSecret := &corev1.Secret{} - if err := r.Get(ctx, types.NamespacedName{ - Name: instance.Spec.TLS.SecretName, - Namespace: instance.Namespace, - }, tlsSecret); err != nil { - if !k8s_errors.IsNotFound(err) { - return ctrl.Result{}, fmt.Errorf("failed to get TLS secret %s: %w", instance.Spec.TLS.SecretName, err) - } - } else { - hashInput["tlsSecretData"] = tlsSecret.Data - } - if instance.Spec.TLS.CaSecretName != "" && instance.Spec.TLS.CaSecretName != instance.Spec.TLS.SecretName { - caSecret := &corev1.Secret{} - if err := r.Get(ctx, types.NamespacedName{ - Name: instance.Spec.TLS.CaSecretName, - Namespace: instance.Namespace, - }, caSecret); err != nil { - if !k8s_errors.IsNotFound(err) { - return ctrl.Result{}, fmt.Errorf("failed to get CA secret %s: %w", instance.Spec.TLS.CaSecretName, err) - } - } else { - hashInput["caSecretData"] = caSecret.Data - } - } } configMapHash, err := util.ObjectHash(hashInput) if err != nil { diff --git a/internal/rabbitmq/statefulset.go b/internal/rabbitmq/statefulset.go index a8d73848..8bf6d1be 100644 --- a/internal/rabbitmq/statefulset.go +++ b/internal/rabbitmq/statefulset.go @@ -143,6 +143,12 @@ func StatefulSet( ServiceName: fmt.Sprintf("%s-nodes", r.Name), Replicas: replicas, PodManagementPolicy: appsv1.ParallelPodManagement, + UpdateStrategy: appsv1.StatefulSetUpdateStrategy{ + RollingUpdate: &appsv1.RollingUpdateStatefulSetStrategy{ + Partition: ptr.To(int32(0)), + }, + Type: appsv1.RollingUpdateStatefulSetStrategyType, + }, Selector: &metav1.LabelSelector{ MatchLabels: matchls, }, @@ -392,10 +398,19 @@ func buildReadinessProbe(r *rabbitmqv1.RabbitMq) *corev1.Probe { // before draining. For single-replica clusters, quorum checks are // skipped since there are no other nodes to maintain quorum. func buildLifecycle(r *rabbitmqv1.RabbitMq) *corev1.Lifecycle { - preStopCmd := `if [ ! -z "$(cat /etc/pod-info/skipPreStopChecks)" ]; then exit 0; fi; rabbitmq-upgrade await_online_quorum_plus_one -t 600 && rabbitmq-upgrade await_online_synchronized_mirror -t 600 || true && rabbitmq-upgrade drain -t 600` - if r.Spec.Replicas != nil && *r.Spec.Replicas <= 1 { - preStopCmd = `if [ ! -z "$(cat /etc/pod-info/skipPreStopChecks)" ]; then exit 0; fi; rabbitmq-upgrade drain -t 600` + // Derive per-command timeouts from TerminationGracePeriodSeconds using a + // proportional split (T/6 for quorum checks, T/2 for drain) so that drain + // always gets a chance to run within the grace period. + timeout := int64(60) + if r.Spec.TerminationGracePeriodSeconds != nil { + timeout = *r.Spec.TerminationGracePeriodSeconds } + preStopCmd := fmt.Sprintf( + `if [ ! -z "$(cat /etc/pod-info/skipPreStopChecks)" ]; then exit 0; fi; `+ + `if rabbitmqctl eval 'length(rabbit_nodes:list_running()) > 1.' 2>/dev/null | grep -q true; then `+ + `rabbitmq-upgrade await_online_quorum_plus_one -t %d && rabbitmq-upgrade await_online_synchronized_mirror -t %d || true; fi; `+ + `rabbitmq-upgrade drain -t %d`, + timeout/6, timeout/6, timeout/2) return &corev1.Lifecycle{ PreStop: &corev1.LifecycleHandler{ Exec: &corev1.ExecAction{ diff --git a/internal/rabbitmq/statefulset_test.go b/internal/rabbitmq/statefulset_test.go index b760bec1..f95caf57 100644 --- a/internal/rabbitmq/statefulset_test.go +++ b/internal/rabbitmq/statefulset_test.go @@ -15,10 +15,12 @@ limitations under the License. package rabbitmq import ( + "fmt" "strings" "testing" rabbitmqv1 "github.com/openstack-k8s-operators/infra-operator/apis/rabbitmq/v1beta1" + appsv1 "k8s.io/api/apps/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/utils/ptr" ) @@ -144,6 +146,75 @@ func TestBuildWipeDataInitContainer_DifferentVersions(t *testing.T) { } } +func TestStatefulSet_PreStopHook(t *testing.T) { + tests := []struct { + name string + gracePeriod *int64 + wantQuorum int64 + wantMirror int64 + wantDrain int64 + }{ + {"default (nil) uses 60s", nil, 10, 10, 30}, + {"explicit 60s", ptr.To(int64(60)), 10, 10, 30}, + {"explicit 120s", ptr.To(int64(120)), 20, 20, 60}, + {"explicit 600s", ptr.To(int64(600)), 100, 100, 300}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + r := newTestRabbitMq("test-mq") + r.Spec.TerminationGracePeriodSeconds = tt.gracePeriod + sts := StatefulSet(r, "hash", nil, nil, "4.2", false, ProxyConfig{}) + + lifecycle := sts.Spec.Template.Spec.Containers[0].Lifecycle + if lifecycle == nil || lifecycle.PreStop == nil || lifecycle.PreStop.Exec == nil { + t.Fatal("PreStop exec hook must be set") + } + cmd := lifecycle.PreStop.Exec.Command + if len(cmd) != 3 || cmd[0] != "/bin/bash" || cmd[1] != "-c" { + t.Fatalf("unexpected command structure: %v", cmd) + } + script := cmd[2] + + wantQuorum := fmt.Sprintf("await_online_quorum_plus_one -t %d", tt.wantQuorum) + if !strings.Contains(script, wantQuorum) { + t.Errorf("script missing %q:\n%s", wantQuorum, script) + } + wantMirror := fmt.Sprintf("await_online_synchronized_mirror -t %d", tt.wantMirror) + if !strings.Contains(script, wantMirror) { + t.Errorf("script missing %q:\n%s", wantMirror, script) + } + wantDrain := fmt.Sprintf("drain -t %d", tt.wantDrain) + if !strings.Contains(script, wantDrain) { + t.Errorf("script missing %q:\n%s", wantDrain, script) + } + if !strings.Contains(script, "skipPreStopChecks") { + t.Error("script missing skipPreStopChecks guard") + } + if !strings.Contains(script, "rabbit_nodes:list_running()") { + t.Error("script missing runtime node count check") + } + }) + } +} + +func TestStatefulSet_UpdateStrategy(t *testing.T) { + r := newTestRabbitMq("test-mq") + sts := StatefulSet(r, "hash", nil, nil, "4.2", false, ProxyConfig{}) + + if sts.Spec.UpdateStrategy.Type != appsv1.RollingUpdateStatefulSetStrategyType { + t.Errorf("UpdateStrategy.Type = %q, want %q", + sts.Spec.UpdateStrategy.Type, appsv1.RollingUpdateStatefulSetStrategyType) + } + if sts.Spec.UpdateStrategy.RollingUpdate == nil { + t.Fatal("UpdateStrategy.RollingUpdate must be set") + } + if *sts.Spec.UpdateStrategy.RollingUpdate.Partition != 0 { + t.Errorf("Partition = %d, want 0", + *sts.Spec.UpdateStrategy.RollingUpdate.Partition) + } +} + func TestStatefulSet_ReplicaCount(t *testing.T) { tests := []struct { name string diff --git a/test/functional/rabbitmq_controller_test.go b/test/functional/rabbitmq_controller_test.go index d75c0b25..12413679 100644 --- a/test/functional/rabbitmq_controller_test.go +++ b/test/functional/rabbitmq_controller_test.go @@ -2240,7 +2240,7 @@ var _ = Describe("RabbitMQ Controller", func() { }) }) - When("TLS secret is rotated, config hash changes to trigger rolling restart", func() { + When("TLS secret data is rotated, config hash remains stable", func() { var certSecret *corev1.Secret BeforeEach(func() { @@ -2254,7 +2254,7 @@ var _ = Describe("RabbitMQ Controller", func() { DeferCleanup(th.DeleteInstance, rabbitmq) }) - It("should update the config-hash annotation on the StatefulSet when TLS secret changes", func() { + It("should NOT update the config-hash annotation when TLS secret data changes", func() { SimulateRabbitMQClusterReady(rabbitmqName) // Get initial config-hash from the StatefulSet @@ -2274,12 +2274,14 @@ var _ = Describe("RabbitMQ Controller", func() { g.Expect(th.K8sClient.Update(th.Ctx, secret)).Should(Succeed()) }, timeout, interval).Should(Succeed()) - // Verify config-hash changed on the StatefulSet (triggers rolling restart) - Eventually(func(g Gomega) { + // Verify config-hash did NOT change — cert rotation is handled + // by projected volumes and RabbitMQ 4.x automatic reload, not + // by rolling restart. Hashing cert data caused quorum queue + // corruption during initial deployment. + Consistently(func(g Gomega) { sts := GetRabbitMQStatefulSet(rabbitmqName) newHash := sts.Spec.Template.Annotations["config-hash"] - g.Expect(newHash).NotTo(BeEmpty()) - g.Expect(newHash).NotTo(Equal(initialHash), "config-hash should change when TLS secret is rotated") + g.Expect(newHash).To(Equal(initialHash), "config-hash should not change on TLS secret data rotation") }, timeout, interval).Should(Succeed()) }) })