Skip to content

Prevent unnecessary RabbitMQ rolling restarts during scaling and cert rotation #643

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
Open

Prevent unnecessary RabbitMQ rolling restarts during scaling and cert rotation #643

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
35 changes: 6 additions & 29 deletions internal/controller/rabbitmq/rabbitmq_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
21 changes: 18 additions & 3 deletions internal/rabbitmq/statefulset.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
},
Expand Down Expand Up @@ -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{
Expand Down
71 changes: 71 additions & 0 deletions internal/rabbitmq/statefulset_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand Down Expand Up @@ -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
Expand Down
14 changes: 8 additions & 6 deletions test/functional/rabbitmq_controller_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand All @@ -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
Expand All @@ -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())
})
})
Expand Down