Minimum Air Induction

This article introduces a plugin component for the RabbitMQ broker that relays AMQP messages to a remote broker. The component is called Shovel and it is a deployed as an embedded message consumer on an instance of RabbitMQ. It receives messages from the broker running in the same address space and in turn relays them to a remote broker. The use case for Shovel is a scenario where a message producer sends messages over a high-speed LAN to a broker which then asynchronously relays them over a low-speed WAN to a destination broker.

Introduction

This diagram shows the high level deployment layout:

In this scenario, the producer sends messages to the relay broker via a quick LAN link. Shovel, which is installed as an embedded consumer, relays the messages it receives to the destination broker via the slow WAN link, which in turn, delivers the message to the end consumer. This frees the producer from having to send a message via a slow remote link.

Using Shovel as an embedded consumer avoids the overhead of TCP communication between itself and the local broker. Also, because Shovel uses the Erlang AMQP client in the embedded modus, it eliminates the necessity to marshal and unmarshal AMQP frames at both the client and server ends.

Currently, Shovel performs a simple forwarding mechanism with no aggregation or compression. These features could be added to future version of Shovel, but it would good to get community feedback in order to find out what kinds of aggregation and compression would suit various applications.

Getting Set Up

To get set up, you will require these three components:

1. The RabbitMQ server;
2. The Erlang AMQP client;
3. The Shovel application.

Firstly, you will need RabbitMQ installed on the machine (or machines) where the relay and destination brokers are going to be run. Strictly speaking, you can just run RabbitMQ straight out of the shell, but to do so, you will need the compiled beam files in the Rabbit ebin directory to be on the load path of the Erlang VM you are starting. This can be acheived by supplying the path to the ebin directory using the -pa argument to the erl command. Alternatively, you can put a soft link to the Rabbit directory in the lib/erlang/lib directory inside your OTP installation.

Secondly, you will need to download and build the Erlang AMQP client. This is achieved by unpacking the source archive and running make at the command line. You can clone the Mercurial repository as follows:

$ hg clone http://hg.opensource.lshift.net/rabbitmq-erlang-client

Although this is not required in order to run Shovel, there is an introduction to the Erlang AMQP client for those who are interested how the client is embedded into the broker.

Thirdly, you can get the source code for shovel from Github either as a tarball or by checking out using git:

$ git clone git://github.com/0x6e6562/shovel.git

After this, change into the shovel directory and run the default make target.

Running The Scenario

In this scenario, I'm running both brokers on one physical machine, but I'm binding the RabbitMQ TCP listener to different network interfaces. The destination broker is going to listen on 192.168.1.66 and the relay broker is going to listen on 127.0.0.1, but you can change these to whatever you want. In fact, if you do run the brokers on separate machines, then you do not even have to bother to configure the listener interface, it will use 0.0.0.0 by default.

First of all, start the destination broker. To override the default listener interface for RabbitMQ, you need to supply the tcp_listeners property via the command line:

$ erl -mnesia dir DIR -boot start_sasl \
-rabbit tcp_listeners '[{"192.168.1.66", 5672}]' \
-s rabbit

The DIR variable is just some directory where you would like mnesia to store its transaction logs in.

When that has successfully started, start the relay broker with the embedded shovel application:

$ erl -pa PATH_TO_SHOVEL_EBIN PATH_TO_ERLANG_CLIENT_EBIN \
-mnesia dir DIR \
-rabbit tcp_listeners '[{"127.0.0.1", 5672}]' \
-boot start_sasl -s rabbit -s shovel

The PATH_TO_SHOVEL_EBIN variable is the path to directory where the compiled beam files are located for the Shovel application. Similarly, the PATH_TO_ERLANG_CLIENT_EBIN variable is the ebin directory of the Erlang AMQP client. This procedure is just setting the load path for the Erlang VM to be able to load in the required modules at run time.

If Shovel starts successfully, it should show up in the list of managed OTP applications when you use the which_applications/0 function in the Erlang shell:

1> application:which_applications().
[{shovel,"Shovel AMQP Relay","0.1"},
 {rabbit,"RabbitMQ","%%VERSION%%"},
 {mnesia,"MNESIA  CXC 138 12","4.4.3"},
 {os_mon,"CPO  CXC 138 46","2.1.6"},
 {sasl,"SASL  CXC 138 11","2.1.5.3"},
 {stdlib,"ERTS  CXC 138 10","1.15.3"},
 {kernel,"ERTS  CXC 138 10","2.12.3"}]
2>

From this point on, you can start sending to the relay broker and receiving messages from the destination broker using any AMQP client that you like. To keep this discussion simple though, I am just going to demonstrate the relay mechanism using the Erlang AMQP client as a producer and a consumer.

To see this, start the consumer that will process messages from the destination broker. To do this, start a new Erlang shell and type in the following commands:

$ erl -pa PATH_TO_SHOVEL_EBIN PATH_TO_ERLANG_CLIENT_EBIN

Eshell V5.6.3  (abort with ^G)
1> Pid = consumer:start("192.168.1.66",<<"X">>,<<"q">>,<<"">>).
Subscribed: <<"amq.ctag-MEA7t5BeGQQRnipzslrsjg==">>
<0.38.0>

This shows that a consumer process was started (with the Pid <0.38.0>) and that it is subscribed to a remote queue with a broker-generated subscription tag (amq.ctag-MEA7t5BeGQQRnipzslrsjg==).

From either the same or a different shell, start a producer to send messages to the relay broker:

2> producer:start("127.0.0.1",<<"">>,<<"q">>,1).
ok

In the shell that you started the consumer in, you should see some indication that a message has been consumed:

Rec'd msg: [<<0,0,4,190,0,2,108,186,0,11,10,109>>]

And that's it basically.

Discussion

This simple demonstration shows how to leverage the asynchronous nature of AMQP to provide a fast local endpoint for clients that produce messages at a fast rate, but require the messages to be delivered to a slow remote endpoint.

The Shovel application is quite simple and utilizes the locality abstraction provided by the Erlang client to run as an embedded consumer inside the same Erlang VM as the RabbitMQ broker. Currently it just does a straight relay to the remote broker, but it could be adapted to buffer up messages depending on time and/or volume before relaying it to the destination broker. Also, the option of message level compression has not been explored. It would be useful to find out how Shovel works in an actual WAN environment to see the effects of the various parameters such as buffer size and compression factors on a real application. Hence, any input from the community is welcomed to shape the Shovel application.

A further extension point is the possibility to make this functionality a part of the RabbitMQ broker code base. By doing so, it would make the packaging and deployment of the relay component a lot simpler as well as being able to exploit further optimization potential within the server. However, the first goal is to provide a simple working solution for the RabbitMQ community, who in turn can provide feedback that may flow into an eventual integrated server side solution.