Do not restart StatefulSet on scale out

If the RabbitMQ cluster is under heavy load and is being scaled out
(i.e. more RabbitMQ nodes added to the RabbitMQ cluster), existing nodes
shouldn't be restarted.

Before this commit, exising nodes were restarted because the ConfigMap
gets updated since new peers get included for peer discovery.
However, existing nodes do not need this new peer discovery
configuration since the cluster is already formed.

Author: ansd

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,81 +1,113 @@
 package controllers_test
 
 import (
-	rabbitmqv1beta1 "github.com/rabbitmq/cluster-operator/api/v1beta1"
 	"time"
 
+	rabbitmqv1beta1 "github.com/rabbitmq/cluster-operator/api/v1beta1"
+
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/ginkgo/extensions/table"
 	. "github.com/onsi/gomega"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/pointer"
 )
 
 var _ = Describe("Reconcile rabbitmq Configurations", func() {
 	var (
 		cluster          *rabbitmqv1beta1.RabbitmqCluster
 		defaultNamespace = "default"
 	)
-	DescribeTable("Server configurations updates",
-		func(testCase string) {
-			// create rabbitmqcluster
+
+	AfterEach(func() {
+		Expect(client.Delete(ctx, cluster)).To(Succeed())
+		waitForClusterDeletion(ctx, cluster, client)
+	})
+
+	DescribeTable("Server configurations updates", func(testCase string) {
+		cluster = &rabbitmqv1beta1.RabbitmqCluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: defaultNamespace,
+				Name:      "rabbitmq-" + testCase,
+			},
+		}
+		Expect(client.Create(ctx, cluster)).To(Succeed())
+		waitForClusterCreation(ctx, cluster, client)
+
+		// ensure that cfm and statefulSet does not have annotations set when cluster just got created
+		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"))
+
+		// update rabbitmq server configurations
+		Expect(updateWithRetry(cluster, func(r *rabbitmqv1beta1.RabbitmqCluster) {
+			if testCase == "additional-config" {
+				r.Spec.Rabbitmq.AdditionalConfig = "test_config=0"
+			}
+			if testCase == "advanced-config" {
+				r.Spec.Rabbitmq.AdvancedConfig = "sample-advanced-config."
+			}
+			if testCase == "env-config" {
+				r.Spec.Rabbitmq.EnvConfig = "some-env-variable"
+			}
+		})).To(Succeed())
+
+		By("annotating the server-conf ConfigMap")
+		// ensure annotations from the server-conf ConfigMap
+		var annotations map[string]string
+		Eventually(func() map[string]string {
+			cfm := configMap(ctx, cluster, "server-conf")
+			annotations = cfm.Annotations
+			return annotations
+		}, 5).Should(HaveKey("rabbitmq.com/serverConfUpdatedAt"))
+		_, err := time.Parse(time.RFC3339, annotations["rabbitmq.com/serverConfUpdatedAt"])
+		Expect(err).NotTo(HaveOccurred(), "Annotation rabbitmq.com/serverConfUpdatedAt was not a valid RFC3339 timestamp")
+
+		By("annotating the sts podTemplate")
+		// ensure statefulSet annotations
+		Eventually(func() map[string]string {
+			sts := statefulSet(ctx, cluster)
+			annotations = sts.Spec.Template.Annotations
+			return annotations
+		}, 5).Should(HaveKey("rabbitmq.com/lastRestartAt"))
+		_, err = time.Parse(time.RFC3339, annotations["rabbitmq.com/lastRestartAt"])
+		Expect(err).NotTo(HaveOccurred(), "Annotation rabbitmq.com/lastRestartAt was not a valid RFC3339 timestamp")
+	},
+
+		Entry("spec.rabbitmq.additionalConfig is updated", "additional-config"),
+		Entry("spec.rabbitmq.advancedConfig is updated", "advanced-config"),
+		Entry("spec.rabbitmq.envConfig is updated", "env-config"),
+	)
+
+	Context("scale out", func() {
+		It("does not restart StatefulSet", func() {
 			cluster = &rabbitmqv1beta1.RabbitmqCluster{
 				ObjectMeta: metav1.ObjectMeta{
-					Name:      "rabbitmq-" + testCase,
 					Namespace: defaultNamespace,
+					Name:      "rabbitmq-scale-out",
 				},
 			}
 			Expect(client.Create(ctx, cluster)).To(Succeed())
 			waitForClusterCreation(ctx, cluster, client)
 
-			// ensure that cfm and statefulSet does not have annotations set when configurations haven't changed
 			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"))
 
-			// update rabbitmq server configurations
 			Expect(updateWithRetry(cluster, func(r *rabbitmqv1beta1.RabbitmqCluster) {
-				if testCase == "additional-config" {
-					r.Spec.Rabbitmq.AdditionalConfig = "test_config=0"
-				}
-				if testCase == "advanced-config" {
-					r.Spec.Rabbitmq.AdvancedConfig = "sample-advanced-config."
-				}
-				if testCase == "env-config" {
-					r.Spec.Rabbitmq.EnvConfig = "some-env-variable"
-				}
+				r.Spec.Replicas = pointer.Int32Ptr(5)
 			})).To(Succeed())
 
-			By("annotating the server-conf ConfigMap")
-			// ensure annotations from the server-conf ConfigMap
-			var annotations map[string]string
-			Eventually(func() map[string]string {
-				cfm := configMap(ctx, cluster, "server-conf")
-				annotations = cfm.Annotations
-				return annotations
-			}, 5).Should(HaveKey("rabbitmq.com/serverConfUpdatedAt"))
-			_, err := time.Parse(time.RFC3339, annotations["rabbitmq.com/serverConfUpdatedAt"])
-			Expect(err).NotTo(HaveOccurred(), "Annotation rabbitmq.com/serverConfUpdatedAt was not a valid RFC3339 timestamp")
-
-			By("annotating the sts podTemplate")
-			// ensure statefulSet annotations
-			Eventually(func() map[string]string {
-				sts := statefulSet(ctx, cluster)
-				annotations = sts.Spec.Template.Annotations
-				return annotations
-			}, 5).Should(HaveKey("rabbitmq.com/lastRestartAt"))
-			_, err = time.Parse(time.RFC3339, annotations["rabbitmq.com/lastRestartAt"])
-			Expect(err).NotTo(HaveOccurred(), "Annotation rabbitmq.com/lastRestartAt was not a valid RFC3339 timestamp")
-
-			// delete rmq cluster
-			Expect(client.Delete(ctx, cluster)).To(Succeed())
-			waitForClusterDeletion(ctx, cluster, client)
-		},
+			Consistently(func() map[string]string {
+				return configMap(ctx, cluster, "server-conf").Annotations
+			}, 3, 0.3).ShouldNot(HaveKey("rabbitmq.com/serverConfUpdatedAt"))
 
-		Entry("spec.rabbitmq.additionalConfig is updated", "additional-config"),
-		Entry("spec.rabbitmq.advancedConfig is updated", "advanced-config"),
-		Entry("spec.rabbitmq.envConfig is updated", "env-config"),
-	)
+			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

@@ -14,19 +14,20 @@ import (
 	"strings"
 
 	"sigs.k8s.io/controller-runtime/pkg/client"
-
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
 	"gopkg.in/ini.v1"
 
 	"github.com/rabbitmq/cluster-operator/internal/metadata"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
 )
 
 const (
 	ServerConfigMapName = "server-conf"
+	operatorDefaults    = "operatorDefaults.conf"
 	defaultRabbitmqConf = `
 cluster_partition_handling = pause_minority
 queue_master_locator = min-masters
@@ -55,10 +56,12 @@ prometheus.ssl.port      = 15691
 
 type ServerConfigMapBuilder struct {
 	*RabbitmqResourceBuilder
+	// Set to true if the config change requires RabbitMQ nodes to be restarted.
+	UpdateRequiresStsRestart bool
 }
 
 func (builder *RabbitmqResourceBuilder) ServerConfigMap() *ServerConfigMapBuilder {
-	return &ServerConfigMapBuilder{builder}
+	return &ServerConfigMapBuilder{builder, true}
 }
 
 func (builder *ServerConfigMapBuilder) Build() (client.Object, error) {
@@ -78,6 +81,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))
@@ -215,7 +219,7 @@ func (builder *ServerConfigMapBuilder) Update(object client.Object) error {
 		configMap.Data = make(map[string]string)
 	}
 
-	configMap.Data["operatorDefaults.conf"] = rmqConfBuffer.String()
+	configMap.Data[operatorDefaults] = rmqConfBuffer.String()
 
 	rmqConfBuffer.Reset()
 
@@ -237,6 +241,45 @@ func (builder *ServerConfigMapBuilder) Update(object client.Object) error {
 		return fmt.Errorf("failed setting controller reference: %v", 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.
+// As of now, this data consists of peer discovery nodes:
+// In the case of scale out (i.e. adding more RabbitMQ nodes to the RabbitMQ cluster), new RabbitMQ nodes will
+// be added to the peer discovery configuration. New nodes will receive the configuration containing all peers.
+// However, exisiting nodes do not need to receive the new peer discovery configuration since the cluster is already formed.
+func removeConfigNotRequiringNodeRestart(configMap *corev1.ConfigMap) error {
+	operatorConf := configMap.Data[operatorDefaults]
+	if operatorConf == "" {
+		return nil
+	}
+	conf, err := ini.Load([]byte(operatorConf))
+	if err != nil {
+		return err
+	}
+	defaultSection := conf.Section("")
+	for _, key := range defaultSection.KeyStrings() {
+		if strings.HasPrefix(key, "cluster_formation.classic_config.nodes.") {
+			defaultSection.DeleteKey(key)
+		}
+	}
+	var b strings.Builder
+	if _, err := conf.WriteTo(&b); err != nil {
+		return err
+	}
+	configMap.Data[operatorDefaults] = b.String()
 	return nil
 }
 

internal/resource/configmap_test.go

@@ -542,7 +542,7 @@ CONSOLE_LOG=new`
 			Expect(configMap.Annotations).To(BeEmpty())
 		})
 
-		When("multiple replicas", func() {
+		Context("multiple replicas", func() {
 			It("adds nodes for peer discovery", func() {
 				instance.Spec.Replicas = pointer.Int32Ptr(3)
 
@@ -557,7 +557,7 @@ cluster_formation.classic_config.nodes.3             = rabbit@%s-server-2.%s-nod
 				)))
 			})
 		})
-		When("no replicas", func() {
+		Context("no replicas", func() {
 			It("does not add nodes for peer discovery", func() {
 				instance.Spec.Replicas = pointer.Int32Ptr(0)
 
@@ -567,6 +567,34 @@ cluster_formation.classic_config.nodes.3             = rabbit@%s-server-2.%s-nod
 				))
 			})
 		})
+
+		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 = pointer.Int32Ptr(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 = pointer.Int32Ptr(3)
+					instance.Spec.Rabbitmq.AdditionalConfig = "foo = bar"
+					Expect(configMapBuilder.Update(configMap)).To(Succeed())
+					Expect(configMapBuilder.UpdateRequiresStsRestart).To(BeTrue())
+				})
+			})
+		})
 	})
 
 	Context("UpdateMayRequireStsRecreate", func() {