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>

5 files changed, 93 insertions, 1 deletions

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 %}
 }
diff --git a/doc/source/reference/message-queues/rabbitmq.rst b/doc/source/reference/message-queues/rabbitmq.rst
index 1eed7ef14..139697633 100644
--- a/doc/source/reference/message-queues/rabbitmq.rst
+++ b/doc/source/reference/message-queues/rabbitmq.rst
@@ -209,3 +209,34 @@ Currently, membership changes for streams `is not entirely safe
 <https://github.com/rabbitmq/rabbitmq-server/discussions/14246>`__, so this
 script should only be used when the RabbitMQ cluster is in a known healthy
 state.
+
+Streams Retention Period
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+When using RabbitMQ streams for fanout queues by setting
+``om_enable_rabbitmq_stream_fanout: true``, users can set retention policy for
+them with the use of two variables ``rabbitmq_stream_max_segment_size_bytes``
+and ``rabbitmq_stream_segment_max_age`` to avoid running out of disk space
+eventually.
+
+Default configuration set segments of a stream queues to have maximum size of
+500 MB (`RabbitMQ default <https://www.rabbitmq.com/docs/streams#declaring>`__)
+and the retention time of 1800 seconds once a segment reaches the maximum size
+(`oslo.messaging default
+<https://docs.openstack.org/oslo.messaging/latest/configuration/opts.html#oslo_messaging_rabbit.rabbit_transient_queues_ttl>`__).
+These default values will leave large number of ready messages in stream
+queues even though old ones are removed by the retention policy.
+So it is recommended to tune them based on how busy the cloud is.
+
+``rabbitmq_stream_max_segment_size_bytes`` sets the maximum size of stream
+segments. This variable needs to be positive integer.
+
+``rabbitmq_stream_segment_max_age`` sets the 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 (e.g. 24h for 24 hours).
+
+.. code-block:: yaml
+
+   # Example custom retention policy configuration
+   rabbitmq_stream_max_segment_size_bytes: 5000 # 5 KB
+   rabbitmq_stream_segment_max_age: "60s" # 60 seconds
diff --git a/releasenotes/notes/make-rabbitmq-stream-retention-policy-configurable-99388b8113d5fcac.yaml b/releasenotes/notes/make-rabbitmq-stream-retention-policy-configurable-99388b8113d5fcac.yaml
new file mode 100644
index 000000000..7f14e29b7
--- /dev/null
+++ b/releasenotes/notes/make-rabbitmq-stream-retention-policy-configurable-99388b8113d5fcac.yaml
@@ -0,0 +1,10 @@
+---
+features:
+  - |
+    New variables ``rabbitmq_stream_max_segment_size_bytes`` and
+    ``rabbitmq_stream_segment_max_age`` are added to make RabbitMQ stream
+    retention policy configurable.
+    These two new variables are only applied when
+    ``om_enable_rabbitmq_stream_fanout`` is set to true.
+    It is recommended to tune the values based on user's environment to
+    prevent leaving too many ready messages in stream queues.