Merge "Make RabbitMQ stream retention policy configurable"

3 files changed, 52 insertions, 1 deletions

diff --git a/ansible/roles/rabbitmq/defaults/main.yml b/ansible/roles/rabbitmq/defaults/main.yml
index 72cb1e657..0a1c7b1ce 100644
--- a/ansible/roles/rabbitmq/defaults/main.yml
+++ b/ansible/roles/rabbitmq/defaults/main.yml
@@ -89,6 +89,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 %}
 }