Dynamic shovel publisher not blocked on low disk alarm

[omonnier-swlabs]

This issue is related to following rabbitmq discussion https://groups.google.com/g/rabbitmq-users/c/iVdixNm4-Jk

Versions:

• RabbitMQ 3.9.13
• Erlang 24.2.1

Configure a dynamic shovel like this:

{
"ack-mode": "on-confirm",
"dest-add-forward-headers": true,
"dest-add-timestamp-header": true,
"dest-exchange": "outputexchange",
"dest-exchange-key": "fromshovel",
"dest-protocol": "amqp091",
"dest-uri": "amqp://",
"src-delete-after": "never",
"src-protocol": "amqp091",
"src-queue": "inputqueue",
"src-uri": "amqp://172.17.0.1:5673/%2F"
}

When RabbitMQ low disk alarms fires, publishers are expected to be blocked as we can see in the logs:

2023-01-31 13:54:18.494092+00:00 [info] <0.297.0> Free disk space is insufficient. Free bytes: 566611968. Limit: 629145600
2023-01-31 13:54:18.494174+00:00 [warn] <0.293.0> *** Publishers will be blocked until this alarm clears ***

But the shovel continues to publish on the destination exchange.
Only other kind of publishers connections are in "blocked" state.
The shovel connection is still in "Idle" state.

[michaelklishin]

It must be the "direct connection" part of Shovel as all network clients would be blocked the same way.

[ikavgo]

there is whole bunch of code in amqp091 shovel that handles blocked connections -

rabbitmq-server/deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl

Line 295 in c88ec8a

update_blocked_by(connection_blocked, true, State);

, for example.

That shovel also seems to be using normal amqp_connection.

[ikavgo]

another potentially interesting thing is that when destination connection is blocked, messages added to pending list which looks boundless.

add_pending(Elem, State = #{dest := Dest}) ->
    Pending = maps:get(pending, Dest, queue:new()),
    State#{dest => Dest#{pending => queue:in(Elem, Pending)}}.

[ikavgo]

Hi @omonnier-swlabs ,

There are many code lines concerning flow/blocking state management in both 091 shovel and respective tests. So at this point I would like to ask you for reproduction steps. Please describe what exactly you are doing, what you are seeing as well as provide more logs.

[gomoripeti]

another potentially interesting thing is that when destination connection is blocked, messages added to pending list which looks boundless.

iirc it is bound by the prefetch count on the source side. (The pending list is only used in case of ack-mode: on-publish or on-confirm)

[gomoripeti]

As far as I know it is a known characteristics/feature/limitation that direct connections are not blocked by resource alarms, that includes also shovels with a local destination. If you would like to shovel between two clusters you can consider defining the shovel on the source cluster as a workaround and have proper flow control on the destination connection (which goes via the network) as was mentioned by Michael and Iliia.

[michaelklishin]

Indeed using full URIs (not amqp://) should result in network connections being used everywhere, and using the same resource alarm notifications and flow control behavior that all other clients provide/experience.

[omonnier-swlabs]

thanks @michaelklishin for your feedback.

Indeed, I tested a full destination URI in the shovel config and the shovel connection now becomes "blocked" as expected, when the low disk alarm fires. Here are the details of the reproduction steps:

Here is the setup, composed of 2 rabbimq servers

| rabbitmq-src | rabbitmq-dest |
| src_queue ---|--> shovel -> dest_ex -> dest_queue |

Start rabbitmq-src and create src_queue

$ docker run -d --hostname localhost --name rabbitmq-src -p 5672:5672 -p 8080:15672 rabbitmq:3-management
$ docker exec rabbitmq-src rabbitmqadmin declare queue name=src_queue queue_type=quorum

start rabbitmq-dest and create dest_ex, dest_queue and binding

$ docker run -d --hostname localhost --name rabbitmq-dest -p 5673:5672 -p 8081:15672 rabbitmq:3-management
$ docker exec rabbitmq-dest rabbitmqadmin declare queue name=dest_queue queue_type=quorum
$ docker exec rabbitmq-dest rabbitmqadmin declare exchange name=dest_ex type=topic
$ docker exec rabbitmq-dest rabbitmqadmin declare binding source=dest_ex destination=dest_queue routing_key=#

connect both rabbitmq through a network bridge

$ docker network create rabbitmq-net
$ docker network connect rabbitmq-net rabbitmq-src
$ docker network connect rabbitmq-net rabbitmq-dest

on rabbitmq-dest, enable rabbitmq_shovel_management plugin and create the shovel

$ docker exec rabbitmq-dest rabbitmq-plugins enable rabbitmq_shovel_management

CASE 1: dest-uri="amqp://"

$ docker exec rabbitmq-dest rabbitmqctl set_parameter shovel my-shovel '{"ack-mode": "on-confirm","dest-add-forward-headers": true,"dest-add-timestamp-header": true,"dest-exchange": "dest_ex","dest-exchange-key": "fromshovel","dest-protocol": "amqp091","dest-uri": "amqp://","src-delete-after": "never","src-protocol": "amqp091","src-queue": "src_queue","src-uri": "amqp://rabbitmq-src:5672/%2F"}'

Check shovel status is OK:

$ docker exec rabbitmq-dest rabbitmqctl shovel_status --formatter=pretty_table
Shovel status on node rabbit@localhost
┌─────────────┬──────────────────────┬──────────────────────────┬──────────────────────┬─────────────────────┬───────────┬──────────────────────────────┬─────────────────┬──────────────┬─────────┬────────────────────┬─────────┬───────┐
│ destination │ destination_exchange │ destination_exchange_key │ destination_protocol │ last_changed │ name │ source │ source_protocol │ source_queue │ state │ termination_reason │ type │ vhost │
├─────────────┼──────────────────────┼──────────────────────────┼──────────────────────┼─────────────────────┼───────────┼──────────────────────────────┼─────────────────┼──────────────┼─────────┼────────────────────┼─────────┼───────┤
│ amqp:// │ dest_ex │ fromshovel │ amqp091 │ 2023-03-29 09:15:12 │ my-shovel │ amqp://rabbitmq-src:5672/%2F │ amqp091 │ src_queue │ running │ │ dynamic │ / │
└─────────────┴──────────────────────┴──────────────────────────┴──────────────────────┴─────────────────────┴───────────┴──────────────────────────────┴─────────────────┴──────────────┴─────────┴────────────────────┴─────────┴───────┘

Simulate a disk full on rabbitmq-dest

See the current free disk space and limit
$ docker exec rabbitmq-dest rabbitmqctl status
...
Low free disk space watermark: 0.05 gb
Free disk space: 904.6608 gb

Set disk free limit to a value higher than Free disk space

$ docker exec rabbitmq-dest rabbitmqctl set_disk_free_limit 1000GB

And ensure in the logs that the alarm fires:

$ docker logs rabbitmq-dest
2023-03-29 09:23:46.553156+00:00 [info] <0.302.0> Disk free limit set to 1000000MB
2023-03-29 09:23:46.555002+00:00 [info] <0.302.0> Free disk space is insufficient. Free bytes: 904662163456. Limit: 1000000000000
2023-03-29 09:23:46.555100+00:00 [warning] <0.298.0> disk resource limit alarm set on node rabbit@localhost.
2023-03-29 09:23:46.555100+00:00 [warning] <0.298.0>
2023-03-29 09:23:46.555100+00:00 [warning] <0.298.0> **********************************************************
2023-03-29 09:23:46.555100+00:00 [warning] <0.298.0> *** Publishers will be blocked until this alarm clears ***
2023-03-29 09:23:46.555100+00:00 [warning] <0.298.0> **********************************************************
2023-03-29 09:23:46.555100+00:00 [warning] <0.298.0>

Now publish a message to rabbitmq-src:src_queue

-> The shovel will get and republish the message into rabbitmq-dest:dest_ex, despite the low disk alarm

CASE 2: Now testing a full shovel destination uri: dest-uri="amqp://rabbitmq-dest:5672/%2F"

Clear the previously created shovel

$ docker exec rabbitmq-dest rabbitmqctl clear_parameter shovel my-shovel

$ docker exec rabbitmq-dest rabbitmqctl set_parameter shovel my-shovel '{"ack-mode": "on-confirm","dest-add-forward-headers": true,"dest-add-timestamp-header": true,"dest-exchange": "dest_ex","dest-exchange-key": "fromshovel","dest-protocol": "amqp091","dest-uri": "amqp://rabbitmq-dest:5672/%2F","src-delete-after": "never","src-protocol": "amqp091","src-queue": "src_queue","src-uri": "amqp://rabbitmq-src:5672/%2F"}'

Check shovel status is OK:

$ docker exec rabbitmq-dest rabbitmqctl shovel_status --formatter=pretty_table
Shovel status on node rabbit@localhost
┌───────────────────────────────┬──────────────────────┬──────────────────────────┬──────────────────────┬─────────────────────┬───────────┬──────────────────────────────┬─────────────────┬──────────────┬─────────┬────────────────────┬─────────┬───────┐
│ destination │ destination_exchange │ destination_exchange_key │ destination_protocol │ last_changed │ name │ source │ source_protocol │ source_queue │ state │ termination_reason │ type │ vhost │
├───────────────────────────────┼──────────────────────┼──────────────────────────┼──────────────────────┼─────────────────────┼───────────┼──────────────────────────────┼─────────────────┼──────────────┼─────────┼────────────────────┼─────────┼───────┤
│ amqp://rabbitmq-dest:5672/%2F │ dest_ex │ fromshovel │ amqp091 │ 2023-03-29 09:43:51 │ my-shovel │ amqp://rabbitmq-src:5672/%2F │ amqp091 │ src_queue │ running │ │ dynamic │ / │
└───────────────────────────────┴──────────────────────┴──────────────────────────┴──────────────────────┴─────────────────────┴───────────┴──────────────────────────────┴─────────────────┴──────────────┴─────────┴────────────────────┴─────────┴───────┘

Publish a message to rabbitmq-src:src_queue
The shovel connection is now marked as blocked in RabbitMQ Admin UI
The shovel does not republish the message into rabbitmq-dest:dest_ex

CASE 3: dest-uri="amqp://localhost:5672"

-> Same expected behavior as CASE 2