Support for Erlang INET configuration

api/v1beta1/rabbitmqcluster_types.go

@@ -385,6 +385,10 @@ type RabbitmqClusterConfigurationSpec struct {
 	// For more information on env config, see https://www.rabbitmq.com/man/rabbitmq-env.conf.5.html
 	// +kubebuilder:validation:MaxLength:=100000
 	EnvConfig string `json:"envConfig,omitempty"`
+	// Erlang Inet configuration to apply to the Erlang VM running rabbit.
+	// See also: https://www.erlang.org/doc/apps/erts/inet_cfg.html
+	// +kubebuilder:validation:MaxLength:=2000
+	ErlangInetConfig string `json:"erlangInetConfig,omitempty"`
 }
 
 // The settings for the persistent storage desired for each Pod in the RabbitmqCluster.
@@ -407,6 +411,10 @@ type RabbitmqClusterServiceSpec struct {
 	Type corev1.ServiceType `json:"type,omitempty"`
 	// Annotations to add to the Service.
 	Annotations map[string]string `json:"annotations,omitempty"`
+	// IPFamilyPolicy represents the dual-stack-ness requested or required by a Service
+	// See also: https://pkg.go.dev/k8s.io/api/core/v1#IPFamilyPolicy
+	// +kubebuilder:validation:Enum=SingleStack;PreferDualStack;RequireDualStack
+	IPFamilyPolicy *corev1.IPFamilyPolicy `json:"ipFamilyPolicy,omitempty"`
 }
 
 func (cluster *RabbitmqCluster) TLSEnabled() bool {

api/v1beta1/rabbitmqcluster_types_test.go

@@ -135,6 +135,24 @@ var _ = Describe("RabbitmqCluster", func() {
 				Expect(apierrors.IsInvalid(k8sClient.Create(context.Background(), invalidService))).To(BeTrue())
 				Expect(k8sClient.Create(context.Background(), invalidService)).To(MatchError(ContainSubstring("supported values: \"ClusterIP\", \"LoadBalancer\", \"NodePort\"")))
 			})
+
+			By("checking the IP family policy", func() {
+				invalidSvc := generateRabbitmqClusterObject("madeup-family-policy")
+				policy := corev1.IPFamilyPolicy("my-awesome-policy")
+				invalidSvc.Spec.Service.IPFamilyPolicy = &policy
+				Expect(apierrors.IsInvalid(k8sClient.Create(context.Background(), invalidSvc))).To(BeTrue())
+			})
+		})
+
+		It("can be created with Erlang configuration", func() {
+			created := generateRabbitmqClusterObject("erlang-configuration")
+			erlangConfig := "{some_config, 123}."
+			created.Spec.Rabbitmq.ErlangInetConfig = erlangConfig
+			Expect(k8sClient.Create(context.Background(), created)).To(Succeed())
+
+			got := &RabbitmqCluster{}
+			Expect(k8sClient.Get(context.Background(), getKey(created), got)).To(Succeed())
+			Expect(got.Spec.Rabbitmq.ErlangInetConfig).To(Equal(erlangConfig))
 		})
 
 		Describe("ChildResourceName", func() {

config/crd/bases/rabbitmq.com_rabbitmqclusters.yaml

@@ -10,7 +10,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   annotations:
-    controller-gen.kubebuilder.io/version: v0.12.1
+    controller-gen.kubebuilder.io/version: v0.13.0
   name: rabbitmqclusters.rabbitmq.com
 spec:
   group: rabbitmq.com
@@ -4127,6 +4127,10 @@ spec:
                       description: 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
                       maxLength: 100000
                       type: string
+                    erlangInetConfig:
+                      description: 'Erlang Inet configuration to apply to the Erlang VM running rabbit. See also: https://www.erlang.org/doc/apps/erts/inet_cfg.html'
+                      maxLength: 2000
+                      type: string
                   type: object
                 replicas:
                   default: 1
@@ -4233,6 +4237,13 @@ spec:
                         type: string
                       description: Annotations to add to the Service.
                       type: object
+                    ipFamilyPolicy:
+                      description: 'IPFamilyPolicy represents the dual-stack-ness requested or required by a Service See also: https://pkg.go.dev/k8s.io/api/core/v1#IPFamilyPolicy'
+                      enum:
+                        - SingleStack
+                        - PreferDualStack
+                        - RequireDualStack
+                      type: string
                     type:
                       default: ClusterIP
                       description: 'Type of Service to create for the cluster. Must be one of: ClusterIP, LoadBalancer, NodePort. For more info see https://pkg.go.dev/k8s.io/api/core/v1#ServiceType'

controllers/rabbitmqcluster_controller.go

@@ -151,6 +151,8 @@ func (r *RabbitmqClusterReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 
 	logger.Info("Start reconciling")
 
+	// FIXME: marshalling is expensive. We are marshalling only for the sake of logging.
+	// 	Implement Stringer interface instead
 	instanceSpec, err := json.Marshal(rabbitmqCluster.Spec)
 	if err != nil {
 		logger.Error(err, "Failed to marshal cluster spec")
@@ -290,6 +292,7 @@ func (r *RabbitmqClusterReconciler) updateStatusConditions(ctx context.Context,
 
 	if !reflect.DeepEqual(rmq.Status.Conditions, oldConditions) {
 		if err = r.Status().Update(ctx, rmq); err != nil {
+			// FIXME: must fetch again to avoid the conflict
 			if k8serrors.IsConflict(err) {
 				logger.Info("failed to update status because of conflict; requeueing...",
 					"namespace", rmq.Namespace,

controllers/rabbitmqcluster_controller_test.go

@@ -13,6 +13,7 @@ package controllers_test
 import (
 	"context"
 	"fmt"
+	"k8s.io/utils/ptr"
 	"time"
 
 	"k8s.io/utils/pointer"
@@ -339,6 +340,23 @@ var _ = Describe("RabbitmqClusterController", func() {
 			Expect(clientSvc.Spec.Type).Should(Equal(corev1.ServiceTypeLoadBalancer))
 			Expect(clientSvc.Annotations).Should(HaveKeyWithValue("annotations", "cr-annotation"))
 		})
+
+		It("creates the service with the expected IP family policy", func() {
+			cluster = &rabbitmqv1beta1.RabbitmqCluster{
+				ObjectMeta: metav1.ObjectMeta{Name: "rabbit-with-ip-family", Namespace: defaultNamespace},
+			}
+			cluster.Spec.Service.IPFamilyPolicy = ptr.To(corev1.IPFamilyPolicyPreferDualStack)
+
+			Expect(client.Create(ctx, cluster)).To(Succeed())
+
+			clientSvc := service(ctx, cluster, "")
+			Expect(clientSvc.Spec.IPFamilyPolicy).ToNot(BeNil())
+			Expect(clientSvc.Spec.IPFamilyPolicy).To(BeEquivalentTo(ptr.To("PreferDualStack")))
+
+			headlessSvc := service(ctx, cluster, "nodes")
+			Expect(headlessSvc.Spec.IPFamilyPolicy).ToNot(BeNil())
+			Expect(headlessSvc.Spec.IPFamilyPolicy).To(BeEquivalentTo(ptr.To("PreferDualStack")))
+		})
 	})
 
 	Context("Resource requirements configurations", func() {

docs/api/rabbitmq.com.ref.asciidoc

@@ -154,6 +154,7 @@ RabbitMQ-related configuration.
 | *`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
+| *`erlangInetConfig`* __string__ | Erlang Inet configuration to apply to the Erlang VM running rabbit. See also: https://www.erlang.org/doc/apps/erts/inet_cfg.html
 |===
 
 
@@ -283,6 +284,7 @@ Settable attributes for the Service resource.
 | Field | Description
 | *`type`* __link:https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.23/#servicetype-v1-core[$$ServiceType$$]__ | Type of Service to create for the cluster. Must be one of: ClusterIP, LoadBalancer, NodePort. For more info see https://pkg.go.dev/k8s.io/api/core/v1#ServiceType
 | *`annotations`* __object (keys:string, values:string)__ | Annotations to add to the Service.
+| *`ipFamilyPolicy`* __link:https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.23/#ipfamilypolicy-v1-core[$$IPFamilyPolicy$$]__ | IPFamilyPolicy represents the dual-stack-ness requested or required by a Service See also: https://pkg.go.dev/k8s.io/api/core/v1#IPFamilyPolicy
 |===
 
 

docs/examples/ipv6/README.md

@@ -0,0 +1,10 @@
+# IPv6 example
+
+This example shows the required configuration to force the Erlang VM to use
+IPv6. RabbitMQ relies on Erlang's INET module for network interaction.
+
+You can deploy this example using:
+
+```shell
+kubectl apply -f rabbitmq.yaml
+```

docs/examples/ipv6/rabbitmq.yaml

@@ -0,0 +1,30 @@
+---
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: rabbits
+  labels:
+    app: rabbitmq
+---
+apiVersion: rabbitmq.com/v1beta1
+kind: RabbitmqCluster
+metadata:
+  name: rabbit-ipv6
+  namespace: rabbits
+  labels:
+    app: rabbitmq
+spec:
+  resources:
+    requests: {}
+    limits: {}
+  rabbitmq:
+    erlangInetConfig: |
+      {inet6, true}.
+    envConfig: |
+        SERVER_ADDITIONAL_ERL_ARGS="-kernel inetrc '/etc/rabbitmq/erl_inetrc'  -proto_dist inet6_tcp"
+        RABBITMQ_CTL_ERL_ARGS="-proto_dist inet6_tcp"
+    additionalConfig: |
+      cluster_formation.k8s.host = kubernetes.default.svc.cluster.local
+  replicas: 3
+  service:
+    ipFamilyPolicy: "PreferDualStack"

go.mod

@@ -23,7 +23,7 @@ require (
 	k8s.io/apimachinery v0.28.2
 	k8s.io/client-go v0.28.1
 	k8s.io/klog/v2 v2.100.1
-	k8s.io/utils v0.0.0-20230505201702-9f6742963106
+	k8s.io/utils v0.0.0-20230726121419-3b25d923346b
 	sigs.k8s.io/controller-runtime v0.15.1
 	sigs.k8s.io/controller-runtime/tools/setup-envtest v0.0.0-20220217024943-cfd92767d28e
 	sigs.k8s.io/controller-tools v0.13.0

go.sum

@@ -566,8 +566,8 @@ k8s.io/klog/v2 v2.100.1 h1:7WCHKK6K8fNhTqfBhISHQ97KrnJNFZMcQvKp7gP/tmg=
 k8s.io/klog/v2 v2.100.1/go.mod h1:y1WjHnz7Dj687irZUWR/WLkLc5N1YHtjLdmgWjndZn0=
 k8s.io/kube-openapi v0.0.0-20230816210353-14e408962443 h1:CAIciCnJnSOQxPd0xvpV6JU3D4AJvnYbImPpFpO9Hnw=
 k8s.io/kube-openapi v0.0.0-20230816210353-14e408962443/go.mod h1:wZK2AVp1uHCp4VamDVgBP2COHZjqD1T68Rf0CM3YjSM=
-k8s.io/utils v0.0.0-20230505201702-9f6742963106 h1:EObNQ3TW2D+WptiYXlApGNLVy0zm/JIBVY9i+M4wpAU=
-k8s.io/utils v0.0.0-20230505201702-9f6742963106/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
+k8s.io/utils v0.0.0-20230726121419-3b25d923346b h1:sgn3ZU783SCgtaSJjpcVVlRqd6GSnlTLKgpAAttJvpI=
+k8s.io/utils v0.0.0-20230726121419-3b25d923346b/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
 sigs.k8s.io/controller-runtime v0.15.1 h1:9UvgKD4ZJGcj24vefUFgZFP3xej/3igL9BsOUTb/+4c=
 sigs.k8s.io/controller-runtime v0.15.1/go.mod h1:7ngYvp1MLT+9GeZ+6lH3LOlcHkp/+tzA/fmHa4iq9kk=
 sigs.k8s.io/controller-runtime/tools/setup-envtest v0.0.0-20220217024943-cfd92767d28e h1:Z4+OH6QT2Xy2aqyN5BjyBEGnINrFzkHlMqLCyrE0A+g=

internal/resource/configmap.go

@@ -98,7 +98,7 @@ func (builder *ServerConfigMapBuilder) Update(object client.Object) error {
 		return err
 	}
 
-	userConfiguration := ini.Empty(ini.LoadOptions{})
+	userConfiguration := ini.Empty()
 	userConfigurationSection := userConfiguration.Section("")
 
 	if builder.Instance.TLSEnabled() {
@@ -241,6 +241,7 @@ func (builder *ServerConfigMapBuilder) Update(object client.Object) error {
 
 	updateProperty(configMap.Data, "advanced.config", rmqProperties.AdvancedConfig)
 	updateProperty(configMap.Data, "rabbitmq-env.conf", rmqProperties.EnvConfig)
+	updateProperty(configMap.Data, "erl_inetrc", rmqProperties.ErlangInetConfig)
 
 	if err := controllerutil.SetControllerReference(builder.Instance, configMap, builder.Scheme); err != nil {
 		return fmt.Errorf("failed setting controller reference: %w", err)

internal/resource/configmap_test.go

@@ -183,7 +183,7 @@ var _ = Describe("GenerateServerConfigMap", func() {
 				Expect(configMap.Data).To(HaveKeyWithValue("advanced.config", "[my-awesome-config]."))
 			})
 
-			It("does set data.advancedConfig when empty", func() {
+			It("does not set data.advancedConfig when empty", func() {
 				instance.Spec.Rabbitmq.AdvancedConfig = ""
 				Expect(configMapBuilder.Update(configMap)).To(Succeed())
 				Expect(configMap.Data).ToNot(HaveKey("advanced.config"))
@@ -559,6 +559,26 @@ CONSOLE_LOG=new`
 			Expect(configMapBuilder.Update(configMap)).To(Succeed())
 			Expect(configMap.Annotations).To(BeEmpty())
 		})
+
+		Context("Erlang INET configuration", func() {
+			It("sets erlangInetRc key", func() {
+				instance.Spec.Rabbitmq.ErlangInetConfig = "{any-config, is-set}."
+				Expect(configMapBuilder.Update(configMap)).To(Succeed())
+				Expect(configMap.Data).To(HaveKeyWithValue("erl_inetrc", "{any-config, is-set}."))
+			})
+
+			When("erlangInetRc is removed", func() {
+				It("deletes the key", func() {
+					instance.Spec.Rabbitmq.ErlangInetConfig = "any string is set, rabbit will do validation"
+					Expect(configMapBuilder.Update(configMap)).To(Succeed())
+					Expect(configMap.Data).To(HaveKey("erl_inetrc"))
+
+					instance.Spec.Rabbitmq.ErlangInetConfig = ""
+					Expect(configMapBuilder.Update(configMap)).To(Succeed())
+					Expect(configMap.Data).ToNot(HaveKey("erl_inetrc"))
+				})
+			})
+		})
 	})
 
 	Context("UpdateMayRequireStsRecreate", func() {

internal/resource/headless_service.go

@@ -59,17 +59,18 @@ func (builder *HeadlessServiceBuilder) Update(object client.Object) error {
 			{
 				Protocol:   corev1.ProtocolTCP,
 				Port:       4369,
-				TargetPort: intstr.FromInt(4369),
+				TargetPort: intstr.FromInt32(4369),
 				Name:       "epmd",
 			},
 			{
 				Protocol:   corev1.ProtocolTCP,
 				Port:       25672,
-				TargetPort: intstr.FromInt(25672),
+				TargetPort: intstr.FromInt32(25672),
 				Name:       "cluster-rpc", // aka distribution port
 			},
 		},
 		PublishNotReadyAddresses: true,
+		IPFamilyPolicy:           builder.Instance.Spec.Service.IPFamilyPolicy,
 	}
 
 	if err := controllerutil.SetControllerReference(builder.Instance, service, builder.Scheme); err != nil {

internal/resource/headless_service_test.go

@@ -19,6 +19,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	defaultscheme "k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/utils/ptr"
 )
 
 var _ = Describe("HeadlessService", func() {
@@ -170,11 +171,11 @@ var _ = Describe("HeadlessService", func() {
 					},
 				},
 			}
-			err := serviceBuilder.Update(service)
-			Expect(err).NotTo(HaveOccurred())
 		})
 
 		It("sets the required Spec", func() {
+			Expect(serviceBuilder.Update(service)).To(Succeed())
+
 			expectedSpec := corev1.ServiceSpec{
 				Type:      corev1.ServiceTypeClusterIP,
 				ClusterIP: "None",
@@ -198,7 +199,38 @@ var _ = Describe("HeadlessService", func() {
 				},
 				PublishNotReadyAddresses: true,
 			}
+			Expect(service.Spec).To(Equal(expectedSpec))
+		})
+
+		It("sets the IP family", func() {
+			dualStack := ptr.To(corev1.IPFamilyPolicyPreferDualStack)
+			instance.Spec.Service.IPFamilyPolicy = dualStack
+			Expect(serviceBuilder.Update(service)).To(Succeed())
 
+			expectedSpec := corev1.ServiceSpec{
+				Type:      corev1.ServiceTypeClusterIP,
+				ClusterIP: "None",
+				Selector: map[string]string{
+					"app.kubernetes.io/name": "rabbit-spec",
+				},
+				SessionAffinity: corev1.ServiceAffinityNone,
+				Ports: []corev1.ServicePort{
+					{
+						Protocol:   corev1.ProtocolTCP,
+						Port:       4369,
+						TargetPort: intstr.FromInt32(4369),
+						Name:       "epmd",
+					},
+					{
+						Protocol:   corev1.ProtocolTCP,
+						Port:       25672,
+						TargetPort: intstr.FromInt32(25672),
+						Name:       "cluster-rpc",
+					},
+				},
+				PublishNotReadyAddresses: true,
+				IPFamilyPolicy:           dualStack,
+			}
 			Expect(service.Spec).To(Equal(expectedSpec))
 		})
 	})

internal/resource/service.go

@@ -58,6 +58,7 @@ func (builder *ServiceBuilder) Update(object client.Object) error {
 	service.Labels = metadata.GetLabels(builder.Instance.Name, builder.Instance.Labels)
 	service.Spec.Type = builder.Instance.Spec.Service.Type
 	service.Spec.Selector = metadata.LabelSelector(builder.Instance.Name)
+	service.Spec.IPFamilyPolicy = builder.Instance.Spec.Service.IPFamilyPolicy
 
 	service.Spec.Ports = builder.updatePorts(service.Spec.Ports)