Don't restart the cluster after scale-out (#1612)

* Don't restart the cluster after scale-out

Co-authored-by: David Ansari <[email protected]>

---------

Co-authored-by: David Ansari <[email protected]>

Author: mkuratczyk

controllers/reconcile_rabbitmq_configurations.go

@@ -34,7 +34,7 @@ func (r *RabbitmqClusterReconciler) annotateIfNeeded(ctx context.Context, logger
 		annotationKey string
 	)
 
-	switch builder.(type) {
+	switch b := builder.(type) {
 
 	case *resource.RabbitmqPluginsConfigMapBuilder:
 		if operationResult != controllerutil.OperationResultUpdated {
@@ -45,7 +45,7 @@ func (r *RabbitmqClusterReconciler) annotateIfNeeded(ctx context.Context, logger
 		annotationKey = pluginsUpdateAnnotation
 
 	case *resource.ServerConfigMapBuilder:
-		if operationResult != controllerutil.OperationResultUpdated {
+		if operationResult != controllerutil.OperationResultUpdated || !b.UpdateRequiresStsRestart {
 			return nil
 		}
 		obj = &corev1.ConfigMap{}

controllers/reconcile_rabbitmq_configurations_test.go

@@ -1,14 +1,16 @@
 package controllers_test
 
 import (
-	rabbitmqv1beta1 "github.com/rabbitmq/cluster-operator/v2/api/v1beta1"
 	"strings"
 	"time"
 
+	rabbitmqv1beta1 "github.com/rabbitmq/cluster-operator/v2/api/v1beta1"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
 )
 
 var _ = Describe("Reconcile rabbitmq Configurations", func() {
@@ -78,4 +80,35 @@ var _ = Describe("Reconcile rabbitmq Configurations", func() {
 		Entry(nil, "advancedConfig"),
 		Entry(nil, "envConfig"),
 	)
+
+	Context("scale out", func() {
+		It("does not restart StatefulSet", func() {
+			cluster = &rabbitmqv1beta1.RabbitmqCluster{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: defaultNamespace,
+					Name:      "rabbitmq-scale-out",
+				},
+			}
+			Expect(client.Create(ctx, cluster)).To(Succeed())
+			waitForClusterCreation(ctx, cluster, client)
+
+			cfm := configMap(ctx, cluster, "server-conf")
+			Expect(cfm.Annotations).ShouldNot(HaveKey("rabbitmq.com/serverConfUpdatedAt"))
+			sts := statefulSet(ctx, cluster)
+			Expect(sts.Annotations).ShouldNot(HaveKey("rabbitmq.com/lastRestartAt"))
+
+			Expect(updateWithRetry(cluster, func(r *rabbitmqv1beta1.RabbitmqCluster) {
+				r.Spec.Replicas = ptr.To(int32(5))
+			})).To(Succeed())
+
+			Consistently(func() map[string]string {
+				return configMap(ctx, cluster, "server-conf").Annotations
+			}, 3, 0.3).ShouldNot(HaveKey("rabbitmq.com/serverConfUpdatedAt"))
+
+			Consistently(func() map[string]string {
+				sts := statefulSet(ctx, cluster)
+				return sts.Spec.Template.Annotations
+			}, 3, 0.3).ShouldNot(HaveKey("rabbitmq.com/lastRestartAt"))
+		})
+	})
 })

internal/resource/configmap.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/rabbitmq/cluster-operator/v2/internal/metadata"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -59,10 +60,11 @@ prometheus.ssl.port      = 15691
 
 type ServerConfigMapBuilder struct {
 	*RabbitmqResourceBuilder
+	UpdateRequiresStsRestart bool
 }
 
 func (builder *RabbitmqResourceBuilder) ServerConfigMap() *ServerConfigMapBuilder {
-	return &ServerConfigMapBuilder{builder}
+	return &ServerConfigMapBuilder{builder, true}
 }
 
 func (builder *ServerConfigMapBuilder) Build() (client.Object, error) {
@@ -82,6 +84,7 @@ func (builder *ServerConfigMapBuilder) UpdateMayRequireStsRecreate() bool {
 
 func (builder *ServerConfigMapBuilder) Update(object client.Object) error {
 	configMap := object.(*corev1.ConfigMap)
+	previousConfigMap := configMap.DeepCopy()
 
 	ini.PrettySection = false // Remove trailing new line because rabbitmq.conf has only a default section.
 	operatorConfiguration, err := ini.Load([]byte(defaultRabbitmqConf))
@@ -247,6 +250,43 @@ func (builder *ServerConfigMapBuilder) Update(object client.Object) error {
 		return fmt.Errorf("failed setting controller reference: %w", err)
 	}
 
+	updatedConfigMap := configMap.DeepCopy()
+	if err := removeConfigNotRequiringNodeRestart(previousConfigMap); err != nil {
+		return err
+	}
+	if err := removeConfigNotRequiringNodeRestart(updatedConfigMap); err != nil {
+		return err
+	}
+	if equality.Semantic.DeepEqual(previousConfigMap, updatedConfigMap) {
+		builder.UpdateRequiresStsRestart = false
+	}
+
+	return nil
+}
+
+// removeConfigNotRequiringNodeRestart removes configuration data that does not require a restart of RabbitMQ nodes.
+// For example, the target cluster size hint changes after adding nodes to a cluster, but there's no reason
+// to restart already running nodes.
+func removeConfigNotRequiringNodeRestart(configMap *corev1.ConfigMap) error {
+	operatorConf := configMap.Data["operatorDefaults.conf"]
+	if operatorConf == "" {
+		return nil
+	}
+	conf, err := ini.Load([]byte(operatorConf))
+	if err != nil {
+		return fmt.Errorf("failed to load operatorDefaults.conf when deciding whether to restart STS: %w", err)
+	}
+	defaultSection := conf.Section("")
+	for _, key := range defaultSection.KeyStrings() {
+		if strings.HasPrefix(key, "cluster_formation.target_cluster_size_hint") {
+			defaultSection.DeleteKey(key)
+		}
+	}
+	var b strings.Builder
+	if _, err := conf.WriteTo(&b); err != nil {
+		return fmt.Errorf("failed to write operatorDefaults.conf when deciding whether to restart STS: %w", err)
+	}
+	configMap.Data["operatorDefaults.conf"] = b.String()
 	return nil
 }
 

internal/resource/configmap_test.go

@@ -12,6 +12,7 @@ package resource_test
 import (
 	"bytes"
 	"fmt"
+
 	"k8s.io/utils/ptr"
 
 	. "github.com/onsi/ginkgo/v2"
@@ -579,6 +580,34 @@ CONSOLE_LOG=new`
 				})
 			})
 		})
+
+		Describe("UpdateRequiresStsRestart", func() {
+			BeforeEach(func() {
+				Expect(configMapBuilder.Update(configMap)).To(Succeed())
+				Expect(configMapBuilder.UpdateRequiresStsRestart).To(BeTrue())
+			})
+			When("the config does not change", func() {
+				It("does not restart StatefulSet", func() {
+					Expect(configMapBuilder.Update(configMap)).To(Succeed())
+					Expect(configMapBuilder.UpdateRequiresStsRestart).To(BeFalse())
+				})
+			})
+			When("the only config change is cluster formation nodes", func() {
+				It("does not require the StatefulSet to be restarted", func() {
+					instance.Spec.Replicas = ptr.To(int32(3))
+					Expect(configMapBuilder.Update(configMap)).To(Succeed())
+					Expect(configMapBuilder.UpdateRequiresStsRestart).To(BeFalse())
+				})
+			})
+			When("config change includes more than cluster formation nodes", func() {
+				It("requires the StatefulSet to be restarted", func() {
+					instance.Spec.Replicas = ptr.To(int32(3))
+					instance.Spec.Rabbitmq.AdditionalConfig = "foo = bar"
+					Expect(configMapBuilder.Update(configMap)).To(Succeed())
+					Expect(configMapBuilder.UpdateRequiresStsRestart).To(BeTrue())
+				})
+			})
+		})
 	})
 
 	Context("UpdateMayRequireStsRecreate", func() {