From 1b42101d82d716387af9c970e4573f766579f7f6 Mon Sep 17 00:00:00 2001
From: Seunghun Lee <seunghun@stackhpc.com>
Date: Wed, 25 Jun 2025 11:09:40 +0100
Subject: Make RabbitMQ stream retention policy configurable

When fanout queues are set to be RabbitMQ stream, number of ready
messages in stream queues grow large.
This is because retention policy of stream is only applied when stream
segments hit the maximum size.

This change make the maximum size of RabbitMQ stream segment and the
retention time configurable by new variables
``rabbitmq_stream_max_segment_size_bytes`` and
``rabbitmq_stream_segment_max_age`` respectively.

NOTE: These are only applied when ``om_enable_rabbitmq_stream_fanout``
is set to ``true``.

Change-Id: Iade5e525f0de8981a0e03e3340ad7552c2f40121
Signed-off-by: Seunghun Lee <seunghun@stackhpc.com>
---
 ansible/roles/rabbitmq/defaults/main.yml            | 13 +++++++++++++
 ansible/roles/rabbitmq/tasks/precheck.yml           | 21 +++++++++++++++++++++
 .../roles/rabbitmq/templates/definitions.json.j2    | 19 ++++++++++++++++++-
 3 files changed, 52 insertions(+), 1 deletion(-)

(limited to 'ansible')

diff --git a/ansible/roles/rabbitmq/defaults/main.yml b/ansible/roles/rabbitmq/defaults/main.yml
index f1ed69f38..9ccbb6b0a 100644
--- a/ansible/roles/rabbitmq/defaults/main.yml
+++ b/ansible/roles/rabbitmq/defaults/main.yml
@@ -88,6 +88,19 @@ rabbitmq_message_ttl_ms: 600000
 rabbitmq_queue_expiry_ms: 3600000
 rabbitmq_extra_config: {}
 
+### RabbitMQ stream queue retention policy ###
+# These policies will only be applied when om_enable_rabbitmq_stream_fanout is true
+
+# Maximum size of stream segments. This variable needs to be positive integer.
+# Default value follows RabbitMQ default size of a segment in bytes: 500000000 (500MB)
+# https://www.rabbitmq.com/docs/streams#declaring
+rabbitmq_stream_max_segment_size_bytes: 500000000
+# Retention time of segments that reached the maximum size.
+# This variable needs to be string with valid options of Y, M, D, h, m, s.
+# Default value follows oslo.messaging default: 1800s
+# https://docs.openstack.org/oslo.messaging/latest/configuration/opts.html#oslo_messaging_rabbit.rabbit_transient_queues_ttl
+rabbitmq_stream_segment_max_age: "1800s"
+
 ####################
 # Plugins
 ####################
diff --git a/ansible/roles/rabbitmq/tasks/precheck.yml b/ansible/roles/rabbitmq/tasks/precheck.yml
index 19f192dfc..a5c8f6340 100644
--- a/ansible/roles/rabbitmq/tasks/precheck.yml
+++ b/ansible/roles/rabbitmq/tasks/precheck.yml
@@ -161,3 +161,24 @@
       when:
         - item.name is search('_fanout')
         - om_enable_rabbitmq_stream_fanout | bool
+
+- name: Check if RabbitMQ stream queue policy variables are valid
+  when: om_enable_rabbitmq_stream_fanout | bool
+  block:
+    - name: Check if the value of rabbitmq_stream_max_segment_size_bytes is valid
+      ansible.builtin.assert:
+        that:
+          - rabbitmq_stream_max_segment_size_bytes is integer
+          - rabbitmq_stream_max_segment_size_bytes > 0
+        fail_msg: >
+          rabbitmq_stream_max_segment_size_bytes needs to be a positive integer.
+          Please check its value again.
+
+    - name: Check if the value of rabbitmq_stream_segment_max_age is valid
+      ansible.builtin.assert:
+        that:
+          - rabbitmq_stream_segment_max_age is string
+          - 'rabbitmq_stream_segment_max_age is regex("^[0-9]+[YMDhms]$")'
+        fail_msg: >
+          rabbitmq_stream_segment_max_age needs to have numeric string that ends with one of the units in [Y, M, D, h, m, s].
+          Please check its value again.
diff --git a/ansible/roles/rabbitmq/templates/definitions.json.j2 b/ansible/roles/rabbitmq/templates/definitions.json.j2
index 298706633..319e95f13 100644
--- a/ansible/roles/rabbitmq/templates/definitions.json.j2
+++ b/ansible/roles/rabbitmq/templates/definitions.json.j2
@@ -1,3 +1,4 @@
+#jinja2: trim_blocks: False
 {
   "vhosts": [
     {"name": "/"}
@@ -10,5 +11,21 @@
     {"user": "{{ role_rabbitmq_user }}", "vhost": "/", "configure": ".*", "write": ".*", "read": ".*"}{% if role_rabbitmq_monitoring_user is defined and role_rabbitmq_monitoring_user %},
     {"user": "{{ role_rabbitmq_monitoring_user }}", "vhost": "/", "configure": "^$", "write": "^$", "read": ".*"}{% endif %}
   ],
-  "policies":[]
+{% if om_enable_rabbitmq_stream_fanout | bool %}
+  "policies": [
+    {
+      "vhost": "/",
+      "name": "stream-retention-policy",
+      "apply-to": "streams",
+      "pattern": ".+",
+      "definition": {
+        "stream-max-segment-size-bytes": {{ rabbitmq_stream_max_segment_size_bytes }},
+        "max-age": "{{ rabbitmq_stream_segment_max_age }}"
+      },
+      "priority": 1
+    }
+  ]
+{% else %}
+  "policies": []
+{% endif %}
 }
-- 
cgit v1.2.3