How to set Stream retention policy

[ankitgr8]

Community Support Policy

• I have read RabbitMQ's Community Support Policy
• I run RabbitMQ 4.x, the only series currently covered by community support
• I promise to provide all relevant information (versions, logs from all nodes, rabbitmq-diagnostics output, detailed reproduction steps)

RabbitMQ version used

other (please specify)

Erlang version used

27.2.x

Operating system (distribution) used

Linux

How is RabbitMQ deployed?

Other

rabbitmq-diagnostics status output

Status of node [email protected] ...
Runtime

OS PID: 62
OS: Linux
Uptime (seconds): 2309883
Is under maintenance?: false
RabbitMQ version: 3.13.7
RabbitMQ release series support status: see https://www.rabbitmq.com/release-information
Node name: [email protected]
Erlang configuration: Erlang/OTP 26 [erts-14.2.3] [source] [64-bit] [smp:16:16] [ds:16:16:10] [async-threads:1]
Crypto library: OpenSSL 1.0.2zi-fips 1 Aug 2023
Erlang processes: 34643 used, 1048576 limit
Scheduler run queue: 1
Cluster heartbeat timeout (net_ticktime): 60

Plugins

Enabled plugin file: /etc/rabbitmq/enabled_plugins
Enabled plugins:

• rabbitmq_prometheus
• rabbitmq_auth_backend_http
• rabbitmq_federation
• rabbitmq_peer_discovery_k8s
• rabbitmq_auth_mechanism_ssl
• rabbitmq_stream
• accept
• rabbitmq_stream_common
• rabbitmq_peer_discovery_common
• prometheus
• rabbitmq_management_agent
• rabbitmq_web_dispatch
  "diagnostic.txt" 113L, 3185C
• prometheus
• rabbitmq_management_agent
• rabbitmq_web_dispatch
• amqp_client
• cowboy
• cowlib

Data directory

Node data directory: /var/rabbitmq/data/[email protected]
Raft data directory: /var/rabbitmq/data/[email protected]/quorum/[email protected]
rvice-rabbitmq-nodes.qaperf-rmq

Config files

• /etc/rabbitmq/advanced.config
• /etc/rabbitmq/rabbitmq.conf

Log file(s)

• /var/rabbitmq/data/rabbit_connection.log
• /var/rabbitmq/data/rabbitmq.log

Alarms

(none)

Memory

Total memory used: 7.0954 gb
Calculation strategy: rss
Memory high watermark setting: 0.4 of available memory, computed to: 24.0518 gb

File Descriptors

Total: 32105, limit: 1048479
Sockets: 72, limit: 943629

Free Disk Space

Low free disk space watermark: 0.05 gb
Free disk space: 524.0077 gb

Totals

Connection count: 72
Queue count: 8128
Virtual host count: 1

Listeners

Interface: [::], port: 5551, protocol: stream/ssl, purpose: stream/ssl
Interface: [::], port: 15692, protocol: http/prometheus, purpose: Prometheus exporter API over HTTP
Interface: [::], port: 25672, protocol: clustering, purpose: inter-node and CLI tool communication
Interface: [::], port: 5671, protocol: amqp/ssl, purpose: AMQP 0-9-1 and AMQP 1.0 over TLS

Logs from node 1 (with sensitive values edited out)

See https://www.rabbitmq.com/docs/logging to learn how to collect logs

# PASTE LOG HERE, BETWEEN BACKTICKS

Logs from node 2 (if applicable, with sensitive values edited out)

See https://www.rabbitmq.com/docs/logging to learn how to collect logs

# PASTE LOG HERE, BETWEEN BACKTICKS

Logs from node 3 (if applicable, with sensitive values edited out)

See https://www.rabbitmq.com/docs/logging to learn how to collect logs

# PASTE LOG HERE, BETWEEN BACKTICKS

rabbitmq.conf

See https://www.rabbitmq.com/docs/configure#config-location to learn how to find rabbitmq.conf file location

# PASTE rabbitmq.conf HERE, BETWEEN BACKTICKS

Steps to deploy RabbitMQ cluster

rabbitmq is deployed as 3 node cluster.

Steps to reproduce the behavior in question

I created the rabbitmq policy

[root@rabbitmq-cluster-1 rabbitmq]# rabbitmqctl list_policies
Listing policies for vhost "/" ...
vhostnamepatternapply-todefinitionpriority
/gms.dbi.dlq.policygms.events.global.dbi.qqueues{"dead-letter-exchange":"gms.dbi.x.deadletter"}0
/stream_retentiongms2itastreams{"max-age":"1m"}0

Now when I am trying to use the same , it throwing below error

[root@rabbitmq-cluster-1 rabbitmq]# rabbitmq-streams set_stream_retention_policy gms2ita stream_retention
Setting retention policy of stream queue gms2ita to stream_retention ...
Error:
{:badarg, [{:erlang, :list_to_integer, [[]], [error_info: %{module: :erl_erts_errors}]}, {:rabbit_amqqueue, :check_max_age, 1, [file: ~
c"rabbit_amqqueue.erl", line: 1036]}, {:rabbit_stream_queue, :set_retention_policy, 3, [file: ~c"rabbit_stream_queue.erl", line: 941]}]
}
[root@rabbitmq-cluster-1 rabbitmq]#

Can any one help what I am doing wrong

advanced.config

See https://www.rabbitmq.com/docs/configure#config-location to learn how to find advanced.config file location

# PASTE advanced.config HERE, BETWEEN BACKTICKS

Application code

# PASTE CODE HERE, BETWEEN BACKTICKS

Kubernetes deployment file

# Relevant parts of K8S deployment that demonstrate how RabbitMQ is deployed
# PASTE YAML HERE, BETWEEN BACKTICKS

What problem are you trying to solve?

trying to set the stream retention policy but failed to do so

[acogoluegnes]

You just need to set up a policy. Do not use the set_stream_retention_policy command (we will probably remove it in the future).

By setting an appropriate regex in a policy you can target one or several streams and apply the retention with the max-age field.

On another note, you happily checked "I run RabbitMQ 4.x, the only series currently covered by community support" whereas the status output says "RabbitMQ version: 3.13.7". Let's consider it a honest mistake but please be more careful in the future.

[ankitgr8]

Thanks for the reply, I knew I was using older version hence I specify .. "RabbitMQ version used" as other.. But there was no way to mention the same.. hence added the output of diagnostics

[michaelklishin]

All "other" versions are out of community support. Please upgrade before starting new discussions.

[ankitgr8]

And Once we set the policy, will it be honoured on existing streams .. Example I created a stream with 24 hour retention.. then I applied the retention policy for 1 hour... i would expect the messages other than 1 hour to be removed from the stream... Is the expectation correct.

Is there any way/cli option to verify if the streams retention is being updated.. because even after the policy is applied.. the original retention shown in stream arguments is still 24 hours..

[acogoluegnes]

Retention is enforced when the current segment is closed and a new one is created, so you won't see old messages going away on a idle stream, the stream must see new messages coming. See the note there https://www.rabbitmq.com/docs/streams#retention.

[ankitgr8]

Thanks that helps