Explore: classic peer discovery without randomised startup delay

api/v1beta1/rabbitmqcluster_types.go

@@ -280,7 +280,7 @@ type Plugin string
 
 // RabbitMQ-related configuration.
 type RabbitmqClusterConfigurationSpec struct {
-	// List of plugins to enable in addition to essential plugins: rabbitmq_management, rabbitmq_prometheus, and rabbitmq_peer_discovery_k8s.
+	// List of plugins to enable in addition to essential plugins: rabbitmq_management, and rabbitmq_prometheus.
 	// +kubebuilder:validation:MaxItems:=100
 	AdditionalPlugins []Plugin `json:"additionalPlugins,omitempty"`
 	// Modify to add to the rabbitmq.conf file in addition to default configurations set by the operator.

config/crd/bases/rabbitmq.com_rabbitmqclusters.yaml

@@ -3496,7 +3496,7 @@ spec:
                       maxLength: 2000
                       type: string
                     additionalPlugins:
-                      description: 'List of plugins to enable in addition to essential plugins: rabbitmq_management, rabbitmq_prometheus, and rabbitmq_peer_discovery_k8s.'
+                      description: 'List of plugins to enable in addition to essential plugins: rabbitmq_management, and rabbitmq_prometheus.'
                       items:
                         description: A Plugin to enable on the RabbitmqCluster.
                         maxLength: 100

controllers/rabbitmqcluster_controller_test.go

@@ -829,8 +829,8 @@ var _ = Describe("RabbitmqClusterController", func() {
 				}))
 
 			Expect(sts.Spec.Template.Spec.HostNetwork).To(BeFalse())
-			Expect(sts.Spec.Template.Spec.Volumes).To(ConsistOf(
-				corev1.Volume{
+			Expect(sts.Spec.Template.Spec.Volumes).To(ConsistOf([]corev1.Volume{
+				{
 					Name: "additional-config",
 					VolumeSource: corev1.VolumeSource{
 						ConfigMap: &corev1.ConfigMapVolumeSource{
@@ -841,7 +841,7 @@ var _ = Describe("RabbitmqClusterController", func() {
 						},
 					},
 				},
-				corev1.Volume{
+				{
 					Name: "rabbitmq-confd",
 					VolumeSource: corev1.VolumeSource{
 						Projected: &corev1.ProjectedVolumeSource{
@@ -881,7 +881,7 @@ var _ = Describe("RabbitmqClusterController", func() {
 						},
 					},
 				},
-				corev1.Volume{
+				{
 					Name: "plugins-conf",
 					VolumeSource: corev1.VolumeSource{
 						ConfigMap: &corev1.ConfigMapVolumeSource{
@@ -893,19 +893,19 @@ var _ = Describe("RabbitmqClusterController", func() {
 					},
 				},
 
-				corev1.Volume{
+				{
 					Name: "rabbitmq-plugins",
 					VolumeSource: corev1.VolumeSource{
 						EmptyDir: &corev1.EmptyDirVolumeSource{},
 					},
 				},
-				corev1.Volume{
+				{
 					Name: "rabbitmq-erlang-cookie",
 					VolumeSource: corev1.VolumeSource{
 						EmptyDir: &corev1.EmptyDirVolumeSource{},
 					},
 				},
-				corev1.Volume{
+				{
 					Name: "erlang-cookie-secret",
 					VolumeSource: corev1.VolumeSource{
 						Secret: &corev1.SecretVolumeSource{
@@ -914,7 +914,7 @@ var _ = Describe("RabbitmqClusterController", func() {
 						},
 					},
 				},
-				corev1.Volume{
+				{
 					Name: "pod-info",
 					VolumeSource: corev1.VolumeSource{
 						DownwardAPI: &corev1.DownwardAPIVolumeSource{
@@ -930,7 +930,14 @@ var _ = Describe("RabbitmqClusterController", func() {
 							},
 						},
 					},
-				}))
+				},
+				{
+					Name: "rabbitmq-configs",
+					VolumeSource: corev1.VolumeSource{
+						EmptyDir: &corev1.EmptyDirVolumeSource{},
+					},
+				},
+			}))
 
 			Expect(extractContainer(sts.Spec.Template.Spec.Containers, "additional-container").Image).To(Equal("my-great-image"))
 		})

docs/api/rabbitmq.com.ref.asciidoc

@@ -148,7 +148,7 @@ RabbitMQ-related configuration.
 [cols="25a,75a", options="header"]
 |===
 | Field | Description
-| *`additionalPlugins`* __xref:{anchor_prefix}-github-com-rabbitmq-cluster-operator-api-v1beta1-plugin[$$Plugin$$] array__ | List of plugins to enable in addition to essential plugins: rabbitmq_management, rabbitmq_prometheus, and rabbitmq_peer_discovery_k8s.
+| *`additionalPlugins`* __xref:{anchor_prefix}-github-com-rabbitmq-cluster-operator-api-v1beta1-plugin[$$Plugin$$] array__ | List of plugins to enable in addition to essential plugins: rabbitmq_management, and rabbitmq_prometheus.
 | *`additionalConfig`* __string__ | Modify to add to the rabbitmq.conf file in addition to default configurations set by the operator. Modifying this property on an existing RabbitmqCluster will trigger a StatefulSet rolling restart and will cause rabbitmq downtime. For more information on this config, see https://www.rabbitmq.com/configure.html#config-file
 | *`advancedConfig`* __string__ | Specify any rabbitmq advanced.config configurations to apply to the cluster. For more information on advanced config, see https://www.rabbitmq.com/configure.html#advanced-config-file
 | *`envConfig`* __string__ | Modify to add to the rabbitmq-env.conf file. Modifying this property on an existing RabbitmqCluster will trigger a StatefulSet rolling restart and will cause rabbitmq downtime. For more information on env config, see https://www.rabbitmq.com/man/rabbitmq-env.conf.5.html
@@ -298,7 +298,7 @@ Spec is the desired state of the RabbitmqCluster Custom Resource.
 | Field | Description
 | *`replicas`* __integer__ | Replicas is the number of nodes in the RabbitMQ cluster. Each node is deployed as a Replica in a StatefulSet. Only 1, 3, 5 replicas clusters are tested. This value should be an odd number to ensure the resultant cluster can establish exactly one quorum of nodes in the event of a fragmenting network partition.
 | *`image`* __string__ | Image is the name of the RabbitMQ docker image to use for RabbitMQ nodes in the RabbitmqCluster. Must be provided together with ImagePullSecrets in order to use an image in a private registry.
-| *`imagePullSecrets`* __link:https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.20/#localobjectreference-v1-core[$$LocalObjectReference$$] array__ | List of Secret resource containing access credentials to the registry for the RabbitMQ image. Required if the docker registry is private.
+| *`imagePullSecrets`* __link:https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.20/#localobjectreference-v1-core[$$LocalObjectReference$$]__ | List of Secret resource containing access credentials to the registry for the RabbitMQ image. Required if the docker registry is private.
 | *`service`* __xref:{anchor_prefix}-github-com-rabbitmq-cluster-operator-api-v1beta1-rabbitmqclusterservicespec[$$RabbitmqClusterServiceSpec$$]__ | The desired state of the Kubernetes Service to create for the cluster.
 | *`persistence`* __xref:{anchor_prefix}-github-com-rabbitmq-cluster-operator-api-v1beta1-rabbitmqclusterpersistencespec[$$RabbitmqClusterPersistenceSpec$$]__ | The desired persistent storage configuration for each Pod in the cluster.
 | *`resources`* __link:https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.20/#resourcerequirements-v1-core[$$ResourceRequirements$$]__ | The desired compute resource requirements of Pods in the cluster.

internal/resource/configmap.go

@@ -27,14 +27,13 @@ import (
 const (
 	ServerConfigMapName = "server-conf"
 	defaultRabbitmqConf = `
-cluster_formation.peer_discovery_backend = rabbit_peer_discovery_k8s
-cluster_formation.k8s.host = kubernetes.default
-cluster_formation.k8s.address_type = hostname
 cluster_partition_handling = pause_minority
 queue_master_locator = min-masters
 disk_free_limit.absolute = 2GB
+cluster_formation.peer_discovery_backend = classic_config
+cluster_formation.discovery_retry_interval = 1000
 cluster_formation.randomized_startup_delay_range.min = 0
-cluster_formation.randomized_startup_delay_range.max = 60`
+cluster_formation.randomized_startup_delay_range.max = 0`
 
 	defaultTLSConf = `
 ssl_options.certfile = /etc/rabbitmq-tls/tls.crt
@@ -87,6 +86,13 @@ func (builder *ServerConfigMapBuilder) Update(object client.Object) error {
 	}
 	defaultSection := operatorConfiguration.Section("")
 
+	// Enforce pod 0 to form the cluster.
+	if _, err := defaultSection.NewKey(
+		"cluster_formation.classic_config.nodes.1",
+		fmt.Sprintf("rabbit@%s-0.%s.%s", builder.Instance.ChildResourceName(stsSuffix), builder.Instance.ChildResourceName(headlessServiceSuffix), builder.Instance.Namespace)); err != nil {
+		return err
+	}
+
 	if _, err := defaultSection.NewKey("cluster_name", builder.Instance.Name); err != nil {
 		return err
 	}

internal/resource/configmap_test.go

@@ -25,19 +25,6 @@ import (
 	defaultscheme "k8s.io/client-go/kubernetes/scheme"
 )
 
-func defaultRabbitmqConf(instanceName string) string {
-	return iniString(`
-cluster_formation.peer_discovery_backend = rabbit_peer_discovery_k8s
-cluster_formation.k8s.host               = kubernetes.default
-cluster_formation.k8s.address_type       = hostname
-cluster_partition_handling               = pause_minority
-queue_master_locator                     = min-masters
-disk_free_limit.absolute                 = 2GB
-cluster_formation.randomized_startup_delay_range.min = 0
-cluster_formation.randomized_startup_delay_range.max = 60
-cluster_name                             = ` + instanceName)
-}
-
 var _ = Describe("GenerateServerConfigMap", func() {
 	var (
 		instance         rabbitmqv1beta1.RabbitmqCluster
@@ -139,7 +126,18 @@ var _ = Describe("GenerateServerConfigMap", func() {
 		It("returns the default rabbitmq configuration", func() {
 			builder.Instance.Spec.Rabbitmq.AdditionalConfig = ""
 
-			expectedConfiguration := defaultRabbitmqConf(builder.Instance.Name)
+			name := builder.Instance.Name
+			expectedConfiguration := fmt.Sprintf(
+				`cluster_partition_handling                           = pause_minority
+queue_master_locator                                 = min-masters
+disk_free_limit.absolute                             = 2GB
+cluster_formation.peer_discovery_backend             = classic_config
+cluster_formation.discovery_retry_interval           = 1000
+cluster_formation.randomized_startup_delay_range.min = 0
+cluster_formation.randomized_startup_delay_range.max = 0
+cluster_formation.classic_config.nodes.1             = rabbit@%s-server-0.%s-nodes.%s
+cluster_name                                         = %s
+`, name, name, builder.Instance.Namespace, name)
 
 			Expect(configMapBuilder.Update(configMap)).To(Succeed())
 			Expect(configMap.Data).To(HaveKeyWithValue("operatorDefaults.conf", expectedConfiguration))

internal/resource/headless_service.go

@@ -21,9 +21,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 )
 
-const (
-	headlessServiceSuffix = "nodes"
-)
+const headlessServiceSuffix = "nodes"
 
 type HeadlessServiceBuilder struct {
 	*RabbitmqResourceBuilder

internal/resource/rabbitmq_plugins.go

@@ -15,8 +15,7 @@ import (
 )
 
 var requiredPlugins = []string{
-	"rabbitmq_peer_discovery_k8s", // required for clustering
-	"rabbitmq_prometheus",         // enforce prometheus metrics
+	"rabbitmq_prometheus", // enforce prometheus metrics
 	"rabbitmq_management",
 }
 

internal/resource/rabbitmq_plugins_test.go

@@ -28,7 +28,7 @@ var _ = Describe("RabbitMQPlugins", func() {
 		When("AdditionalPlugins is empty", func() {
 			It("returns list of required plugins", func() {
 				plugins := NewRabbitmqPlugins(nil)
-				Expect(plugins.DesiredPlugins()).To(ConsistOf([]string{"rabbitmq_peer_discovery_k8s", "rabbitmq_prometheus", "rabbitmq_management"}))
+				Expect(plugins.DesiredPlugins()).To(ConsistOf([]string{"rabbitmq_prometheus", "rabbitmq_management"}))
 			})
 		})
 
@@ -37,7 +37,7 @@ var _ = Describe("RabbitMQPlugins", func() {
 				morePlugins := []rabbitmqv1beta1.Plugin{"rabbitmq_shovel", "my_great_plugin"}
 				plugins := NewRabbitmqPlugins(morePlugins)
 
-				Expect(plugins.DesiredPlugins()).To(ConsistOf([]string{"rabbitmq_peer_discovery_k8s",
+				Expect(plugins.DesiredPlugins()).To(ConsistOf([]string{
 					"rabbitmq_prometheus",
 					"rabbitmq_management",
 					"my_great_plugin",
@@ -51,7 +51,7 @@ var _ = Describe("RabbitMQPlugins", func() {
 				morePlugins := []rabbitmqv1beta1.Plugin{"rabbitmq_management", "rabbitmq_shovel", "my_great_plugin", "rabbitmq_shovel"}
 				plugins := NewRabbitmqPlugins(morePlugins)
 
-				Expect(plugins.DesiredPlugins()).To(ConsistOf([]string{"rabbitmq_peer_discovery_k8s",
+				Expect(plugins.DesiredPlugins()).To(ConsistOf([]string{
 					"rabbitmq_prometheus",
 					"rabbitmq_management",
 					"my_great_plugin",
@@ -117,10 +117,7 @@ var _ = Describe("RabbitMQPlugins", func() {
 			})
 
 			It("adds list of default plugins", func() {
-				expectedEnabledPlugins := "[" +
-					"rabbitmq_peer_discovery_k8s," +
-					"rabbitmq_prometheus," +
-					"rabbitmq_management]."
+				expectedEnabledPlugins := "[rabbitmq_prometheus,rabbitmq_management]."
 
 				obj, err := configMapBuilder.Build()
 				Expect(err).NotTo(HaveOccurred())
@@ -184,7 +181,6 @@ var _ = Describe("RabbitMQPlugins", func() {
 						builder.Instance.Spec.Rabbitmq.AdditionalPlugins = []rabbitmqv1beta1.Plugin{"rabbitmq_management", "rabbitmq_management", "rabbitmq_shovel", "my_great_plugin"}
 
 						expectedEnabledPlugins := "[" +
-							"rabbitmq_peer_discovery_k8s," +
 							"rabbitmq_prometheus," +
 							"rabbitmq_management," +
 							"rabbitmq_shovel," +
@@ -199,15 +195,14 @@ var _ = Describe("RabbitMQPlugins", func() {
 				When("previous data is present", func() {
 					BeforeEach(func() {
 						configMap.Data = map[string]string{
-							"enabled_plugins": "[rabbitmq_peer_discovery_k8s,rabbitmq_shovel]",
+							"enabled_plugins": "[rabbitmq_prometheus,rabbitmq_shovel]",
 						}
 					})
 
 					It("updates enabled_plugins with unique list of default and additionalPlugins", func() {
 						builder.Instance.Spec.Rabbitmq.AdditionalPlugins = []rabbitmqv1beta1.Plugin{"rabbitmq_management", "rabbitmq_management", "rabbitmq_shovel", "my_great_plugin"}
 
 						expectedEnabledPlugins := "[" +
-							"rabbitmq_peer_discovery_k8s," +
 							"rabbitmq_prometheus," +
 							"rabbitmq_management," +
 							"rabbitmq_shovel," +

internal/resource/statefulset.go

@@ -31,9 +31,11 @@ import (
 )
 
 const (
+	stsSuffix           string = "server"
 	initContainerCPU    string = "100m"
 	initContainerMemory string = "500Mi"
 	defaultPVCName      string = "persistence"
+	configsVolume       string = "rabbitmq-configs"
 	DeletionMarker      string = "skipPreStopChecks"
 )
 
@@ -54,7 +56,7 @@ func (builder *StatefulSetBuilder) Build() (client.Object, error) {
 
 	sts := &appsv1.StatefulSet{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      builder.Instance.ChildResourceName("server"),
+			Name:      builder.Instance.ChildResourceName(stsSuffix),
 			Namespace: builder.Instance.Namespace,
 		},
 		Spec: appsv1.StatefulSetSpec{
@@ -280,17 +282,19 @@ func sortEnvVar(envVar []corev1.EnvVar) {
 }
 
 // sortVolumeMounts always returns '/var/lib/rabbitmq/' and '/var/lib/rabbitmq/mnesia/' first in the list.
-// this is to ensure '/var/lib/rabbitmq/' always mounts before '/var/lib/rabbitmq/mnesia/' to aviod shadowing
+// This is to ensure '/var/lib/rabbitmq/' always mounts before '/var/lib/rabbitmq/mnesia/' to avoid shadowing
 // popular open-sourced container runtimes like docker and containerD will sort mounts in alphabetical order to
-// avoid this issue, but there's no guarantee that all container runtime would do so
+// avoid this issue, but there's no guarantee that all container runtime would do so.
+// Likewise '/etc/rabbitmq/conf.d/' always mounts before '/etc/rabbitmq/conf.d/<config-file>'
 func sortVolumeMounts(mounts []corev1.VolumeMount) {
 	for i, m := range mounts {
-		if m.Name == "rabbitmq-erlang-cookie" {
+		switch m.Name {
+		case "rabbitmq-erlang-cookie":
 			mounts[0], mounts[i] = mounts[i], mounts[0]
-			continue
-		}
-		if m.Name == defaultPVCName {
+		case defaultPVCName:
 			mounts[1], mounts[i] = mounts[i], mounts[1]
+		case configsVolume:
+			mounts[2], mounts[i] = mounts[i], mounts[2]
 		}
 	}
 }
@@ -405,6 +409,12 @@ func (builder *StatefulSetBuilder) podTemplateSpec(previousPodAnnotations map[st
 				},
 			},
 		},
+		{
+			Name: configsVolume,
+			VolumeSource: corev1.VolumeSource{
+				EmptyDir: &corev1.EmptyDirVolumeSource{},
+			},
+		},
 	}
 
 	if builder.Instance.Spec.Rabbitmq.AdvancedConfig != "" || builder.Instance.Spec.Rabbitmq.EnvConfig != "" {
@@ -426,6 +436,10 @@ func (builder *StatefulSetBuilder) podTemplateSpec(previousPodAnnotations map[st
 			Name:      "persistence",
 			MountPath: "/var/lib/rabbitmq/mnesia/",
 		},
+		{
+			Name:      configsVolume,
+			MountPath: "/etc/rabbitmq/conf.d/",
+		},
 		{
 			Name:      "rabbitmq-plugins",
 			MountPath: "/operator",
@@ -505,6 +519,8 @@ func (builder *StatefulSetBuilder) podTemplateSpec(previousPodAnnotations map[st
 		volumes = append(volumes, tlsProjectedVolume)
 	}
 
+	writeRetryLimit := "echo 'cluster_formation.discovery_retry_limit = %d' > /etc/rabbitmq/conf.d/12-cluster_formation.conf"
+
 	return corev1.PodTemplateSpec{
 		ObjectMeta: metav1.ObjectMeta{
 			Annotations: metadata.ReconcileAnnotations(previousPodAnnotations, defaultPodAnnotations),
@@ -559,7 +575,7 @@ func (builder *StatefulSetBuilder) podTemplateSpec(previousPodAnnotations map[st
 						},
 					},
 					Command: []string{
-						"sh", "-c", "cp /tmp/erlang-cookie-secret/.erlang.cookie /var/lib/rabbitmq/.erlang.cookie " +
+						"/bin/bash", "-c", "cp /tmp/erlang-cookie-secret/.erlang.cookie /var/lib/rabbitmq/.erlang.cookie " +
 							"&& chown 999:999 /var/lib/rabbitmq/.erlang.cookie " +
 							"&& chmod 600 /var/lib/rabbitmq/.erlang.cookie ; " +
 							"cp /tmp/rabbitmq-plugins/enabled_plugins /operator/enabled_plugins " +
@@ -568,7 +584,22 @@ func (builder *StatefulSetBuilder) podTemplateSpec(previousPodAnnotations map[st
 							"echo '[default]' > /var/lib/rabbitmq/.rabbitmqadmin.conf " +
 							"&& sed -e 's/default_user/username/' -e 's/default_pass/password/' /tmp/default_user.conf >> /var/lib/rabbitmq/.rabbitmqadmin.conf " +
 							"&& chown 999:999 /var/lib/rabbitmq/.rabbitmqadmin.conf " +
-							"&& chmod 600 /var/lib/rabbitmq/.rabbitmqadmin.conf",
+							"&& chmod 600 /var/lib/rabbitmq/.rabbitmqadmin.conf ; " +
+							// All nodes wait (for max 1 day) until pod 0 created the cluster.
+							"if [[ \"$MY_POD_NAME\" == *-server-0 ]]; then " +
+							fmt.Sprintf(writeRetryLimit, 1) + "; else " +
+							fmt.Sprintf(writeRetryLimit, 60*60*24) + "; fi",
+					},
+					Env: []corev1.EnvVar{
+						{
+							Name: "MY_POD_NAME",
+							ValueFrom: &corev1.EnvVarSource{
+								FieldRef: &corev1.ObjectFieldSelector{
+									FieldPath:  "metadata.name",
+									APIVersion: "v1",
+								},
+							},
+						},
 					},
 					Resources: corev1.ResourceRequirements{
 						Limits: map[corev1.ResourceName]k8sresource.Quantity{
@@ -606,6 +637,10 @@ func (builder *StatefulSetBuilder) podTemplateSpec(previousPodAnnotations map[st
 							MountPath: "/tmp/default_user.conf",
 							SubPath:   "default_user.conf",
 						},
+						{
+							Name:      configsVolume,
+							MountPath: "/etc/rabbitmq/conf.d/",
+						},
 					},
 				},
 			},