Add rabbit memory limit

api/v1beta1/rabbitmqcluster_types.go

@@ -314,6 +314,10 @@ func (cluster *RabbitmqCluster) MutualTLSEnabled() bool {
 	return cluster.TLSEnabled() && cluster.Spec.TLS.CaSecretName != ""
 }
 
+func (cluster *RabbitmqCluster) MemoryLimited() bool {
+	return cluster.Spec.Resources != nil && cluster.Spec.Resources.Limits != nil && !cluster.Spec.Resources.Limits.Memory().IsZero()
+}
+
 func (cluster *RabbitmqCluster) SingleTLSSecret() bool {
 	return cluster.MutualTLSEnabled() && cluster.Spec.TLS.CaSecretName == cluster.Spec.TLS.SecretName
 }

api/v1beta1/rabbitmqcluster_types_test.go

@@ -113,6 +113,16 @@ var _ = Describe("RabbitmqCluster", func() {
 			Expect(created.MutualTLSEnabled()).To(BeTrue())
 		})
 
+		It("can be queried if memory limits are provided", func() {
+			created := generateRabbitmqClusterObject("rabbit-mem-limit")
+			Expect(created.MemoryLimited()).To(BeTrue())
+
+			created.Spec.Resources = &corev1.ResourceRequirements{
+				Limits: map[corev1.ResourceName]resource.Quantity{},
+			}
+			Expect(created.MemoryLimited()).To(BeFalse())
+		})
+
 		It("is validated", func() {
 			By("checking the replica count", func() {
 				nOne := int32(-1)

internal/resource/configmap.go

@@ -12,6 +12,7 @@ package resource
 import (
 	"bytes"
 	"fmt"
+
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
 	"gopkg.in/ini.v1"
@@ -87,6 +88,12 @@ func (builder *ServerConfigMapBuilder) Update(object runtime.Object) error {
 		}
 	}
 
+	if builder.Instance.MemoryLimited() {
+		if _, err := defaultSection.NewKey("total_memory_available_override_value", fmt.Sprintf("%d", removeHeadroom(builder.Instance.Spec.Resources.Limits.Memory().Value()))); err != nil {
+			return err
+		}
+	}
+
 	rmqProperties := builder.Instance.Spec.Rabbitmq
 	if err := cfg.Append([]byte(rmqProperties.AdditionalConfig)); err != nil {
 		return fmt.Errorf("failed to append spec.rabbitmq.additionalConfig: %w", err)
@@ -129,3 +136,13 @@ func updateProperty(configMapData map[string]string, key string, value string) {
 		configMapData[key] = value
 	}
 }
+
+// The Erlang VM needs headroom above Rabbit to avoid being OOM killed
+// We set the headroom to be the smaller amount of 20% memory or 2GiB
+func removeHeadroom(memLimit int64) int64 {
+	const GiB int64 = 1073741824
+	if memLimit/5 > 2*GiB {
+		return memLimit - 2*GiB
+	}
+	return memLimit - memLimit/5
+}

internal/resource/configmap_test.go

@@ -10,11 +10,14 @@
 package resource_test
 
 import (
+	"fmt"
+
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	rabbitmqv1beta1 "github.com/rabbitmq/cluster-operator/api/v1beta1"
 	"github.com/rabbitmq/cluster-operator/internal/resource"
 	corev1 "k8s.io/api/core/v1"
+	k8sresource "k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -249,6 +252,27 @@ ssl_options.verify                              = verify_peer`)))
 			})
 		})
 
+		Context("Memory Limits", func() {
+			It("sets a RabbitMQ memory limit with headroom when memory limits are specified", func() {
+				const GiB int64 = 1073741824
+				instance = rabbitmqv1beta1.RabbitmqCluster{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "rabbit-mem-limit",
+					},
+					Spec: rabbitmqv1beta1.RabbitmqClusterSpec{
+						Resources: &corev1.ResourceRequirements{
+							Limits: map[corev1.ResourceName]k8sresource.Quantity{
+								corev1.ResourceMemory: k8sresource.MustParse("10Gi"),
+							},
+						},
+					},
+				}
+
+				Expect(configMapBuilder.Update(configMap)).To(Succeed())
+				Expect(configMap.Data).To(HaveKeyWithValue("rabbitmq.conf", ContainSubstring(fmt.Sprintf("total_memory_available_override_value           = %d", 8*GiB))))
+			})
+		})
+
 		Context("labels", func() {
 			BeforeEach(func() {
 				instance = rabbitmqv1beta1.RabbitmqCluster{