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()) }) })