Common Messaging Patterns Using Stomp – Part 2

by R.I. Pienaar | Dec 11, 2011 | Code

Yesterday I gave a quick intro to the basics of Message Orientated Middleware, today we’ll build something kewl and useful.

Graphite is a fantastic statistics as a service for your network package. It can store, graph, slice and dice your time series data in ways that was only imaginable in the dark days of just having RRD files. The typical way to get data into it is to just talk to its socket and send some metric. This is great mostly but have some issues:

• You have a huge network and so you might be able to overwhelm its input channel
• You have strict policies about network connections and are not allowed to have all servers open a connection to it directly
• Your network is spread over the globe and sometimes the connections are just not reliable, but you do not wish to loose metrics during this time

Graphite solves this already by having a AMQP input channel but for the sake of seeing how we might solve these problems I’ll show how to build your own Stomp based system to do this.

We will allow all our machines to Produce messages into the Queue and we will have a small pool of Consumers that read the queue and speak to Graphite using the normal TCP protocol. We’d run Graphite and the Consumers on the same machine to give best possible availability to the TCP connections but the Middleware can be anywhere. The TCP connections to Graphite will be persistent and be reused to publish many metrics – a connection pool in other words.

Producer
So first the Producer side of things, this is a simple CLI tool that take a metric and value on the CLI and publish it.

#!/usr/bin/ruby
 
require 'rubygems'
require 'stomp'
 
raise "Please provide a metric and value on the command line" unless ARGV.size == 2
raise "The metric value must be numeric" unless ARGV[1] =~ /^[\d\.]+$/
 
msg = "%s.%s %s %d" % [Socket.gethostname, ARGV[0], ARGV[1], Time.now.utc.to_i]
 
begin
  Timeout::timeout(2) do
    stomp = Stomp::Client.new("", "", "stomp.example.com", 61613)
    stomp.publish("/queue/graphite", msg)
    stomp.close
  end
rescue Timeout::Error
  STDERR.puts "Failed to send metric within the 2 second timeout"
  exit 1
end

This is all there really is to sending a message to the middleware, you’d just run this like

producer.rb load1 `cat /proc/loadavg|cut -f1 -d' '`

Which would result in a message being sent with the body

devco.net.load1 0.1 1323597139

Consumer
The consumer part of this conversation is not a whole lot more complex, you can see it below:

#!/usr/bin/ruby
 
require 'rubygems'
require 'stomp'
 
def graphite
  @graphite ||= TCPSocket.open("localhost", 2003)
end
 
client = Stomp::Connection.new("", "", "stomp.example.com", 61613, true)
 
client.subscribe("/queue/graphite")
 
loop do
  begin
    msg = client.receive
 
    graphite.puts msg
  rescue
    STDERR.puts "Failed to receive from queue: #{$!}"
    sleep 1
    retry
  end
end

This subscribes to the queue, loops forever while reading messages that then get sent to Graphite using a normal TCP socket. This should be a bit more complex to use the transaction properties I mentioned since a crash here will loose a single message.

Results
So that is really all there is to it! You’d totally want to make the receiving end a bit more robust, make it a daemon perhaps using the Daemons or Dante Gems and add some logging. You’d agree though this is extremely simple code that anyone could write and maintain.

This code has a lot of non obvious side effects though simply because we use the Middleware for communication:

• It’s completely decoupled, the Producers don’t know anything about the Consumers other than the message format.
• It’s reliable because the Consumer can die but the Producers would not even be aware or need to care about this
• It’s scalable – by simply starting more Consumers you can consume messages from the queue quicker and in a load balanced way. Contrast this with perhaps writing a single multi threaded server with all that entails.
• It’s trivial to understand how it works and the code is completely readable
• It protects my Graphite from the Thundering Herd Problem by using the middleware as a buffer and only creating a manageable pool of writers to Graphite
• It’s language agnostic, you can produce messages from Perl, Ruby, Java etc
• The network layer can be made resilient without any code changes

You wouldn’t think this 44 lines of code could have all these properties, but they do and this is why I think this style of coding is particularly well suited to Systems Administrators. We are busy people, we do not have time to implement from scratch our own connection pooling, buffers, spools and everything else you would need to try to duplicate these points from scratch. We have 20 minutes and we just want to solve our problem. Languages like Ruby and technologies like Message Orientated Middleware lets you do this.

I’d like to expand on the one aspect a bit – I mentioned that the network topology can change without the code being aware of it and that we might have restricted firewalls preventing everyone from communicating with Graphite. Our 44 lines of code solves these problems with the help of the MOM.

By using the facilities the middleware provides to create complex networks we can distribute our connectivity layer globally as below:

Here we have producers all over the world and our central consumer sitting in the EU somewhere. The queuing and storage characteristics of the middleware is present in every region. The producers in each region only need the ability to communicate with their regional Broker.

The middleware layer is reliably connected in a Mesh topology but in the event that transatlantic communications are interrupted the US broker will store the metrics till the connection problem is resolved. At that point it will forward the messages on to the EU broker and finally to the Consumer.

We can deploy brokers in a HA configuration regionally to protect against failure there. This is very well suited for multi DC deployments, deployments in the cloud where you have machines in different Regions and Availability Zones etc.

This is also an approach you could use to also allow your DMZ machines to publish metrics without needing the ability to connect directly to the Graphite service. The middleware layer is very flexible in how it’s clustered, who makes the connections etc so it’s ideal for that.

Conclusion
So in the end with just a bit of work once we’ve invested in the underlying MOM technology and deployed that we have solved a bunch of very complex problems using very simple techniques.

While this was done with reliability and scalability in mind for me possibly the bigger win is that we now have a simple network wide service for creating metrics. You can write to the queue from almost any language and you can easily allow your developers to just emit metrics from their Java code and you can emit metrics from the system side perhaps by reusing Munin.

Using code that is not a lot more complex than this I have been able to gather 10s of thousands of Munin metrics in a very short period of time into Graphite. Was able to up my collection frequency to once every minute instead of the traditional 5 minutes and was able to do that with a load average below 1 vs below 30 for Munin. This is probably more to do with Graphite being superior than anything else but the other properties outlined above makes this very appealing. Nodes push their statistics soon as they are built and I never need to edit a Munin config file anymore to tell it where my servers are.

This enabling of all parties in the organization to quickly and easily create metrics without having an operations bottleneck is a huge win and at the heart of what it means to be a DevOps Practitioner.

Part 3 has been written, please read that next.

Common Messaging Patterns Using Stomp – Part 1

by R.I. Pienaar | Dec 11, 2011 | Code

As most people who follow this blog know I’m quite a fan of messaging based architectures. I am also a big fan of Ruby and I like the simplicity of the Stomp Gem to create messaging applications rather than some of the more complex options like those based on Event Machine (which I am hard pressed to even consider Ruby) or the various AMQP ones.

So I wanted to do a few blog posts on basic messaging principals and patterns and how to use those with the Stomp gem in Ruby. I think Ruby is a great choice for systems development – it’s not perfect by any stretch – but it’s a great replacement for all the things systems people tend to reach to Perl for.

Message Orientated Middleware represents a new way of inter process communications, different from previous approaches that were in-process, reliant on file system sockets or even TCP or UDP sockets. While consumers and producers connect to the middleware using TCP you simply cannot really explain how messaging works in relation to TCP. It’s a new transport that brings with it its own concepts, addressing and reliability.

There are some parallels to TCP/IP wrt to reliability as per TCP and unreliability as per UDP but that’s really where it ends – Messaging based IPC is very different and best to learn the semantics. TCP is to middleware as Ethernet frames are to TCP, it’s just one of the possible ways middleware brokers can communicate and is at a much lower level and works very differently.

Why use MOM
There are many reasons but it comes down to promoting a style of applications that scales well. Mostly it does this by a few means:

• Promotes application design that breaks complex applications into simple single function building blocks that’s easy to develop, test and scale.
• Application building blocks aren’t tightly coupled, doesn’t maintain state and can scale independently of other building blocks
• The middleware layer implementation is transparent to the application – network topologies, routing, ACLs etc can change without application code change
• The brokers provide a lot of the patterns you need for scaling – load balancing, queuing, persistence, eventual consistency, etc
• Mature brokers are designed to be scalable and highly available – very complex problems that you really do not want to attempt to solve on your own

There are many other reasons but for me these are the big ticket items – especially the 2nd one.

Note of warning though while mature brokers are fast, scalable and reliably they are not some magical silver bullet. You might be able to handle 100s of thousands of messages a second on commodity hardware but it has limits and trade offs. Enable persistence or reliable messaging and that number drops drastically.

Even without enabling reliability or persistence you can easily do dumb things that overwhelm your broker – they do not scale infinitely and each broker has design trade offs. Some dedicate single threads to topics/queues that can become a bottleneck, others do not replicate queues across a cluster and so you end up with SPOFs that you might not have expected.

Message passing might appear to be instantaneous but they do not defeat the speed of light, it’s only fast relative to the network distance, network latencies and latencies in your hardware or OS kernel.

If you wish to design a complex application that relies heavily on your middleware for HA and scaling you should expect to spend as much time learning, tuning, monitoring, trending and recovering from crashes as you might with your DBMS, Web Server or any other big complex component of your system.

Types of User
There are basically 2 terms that are used to describe actors in a message orientated system. You have software that produce messages called Producers and ones consuming them called Consumers. Your application might be both a producer and a consumer but these terms are what I’ll use to describe the roles of various actors in a system.

Types of Message
In the Stomp world there really are only two types of message destinations. Message destinations have names like /topic/my.input or /queue/my.input. Here we have 2 message sources – the one is a Topic and the other is a Queue. The format of these names might even change between broker implementations.

There are some riffs on these 2 types of message source – you get short lived private destinations, queues that vanish soon as all subscribers are gone, you get topics that behave like queues and so forth. The basic principals you need to know are just Topics and Queues and detail on these can be seen below, the rest builds on these.

Topics
A topic is basically a named broadcast zone. If I produce a single message into a topic called /topic/my.input and there are 10 consumers of that topic then all 10 will get a copy of the message. Messages are not stored when you aren’t around to read them – it’s a stream of messages that you can just tap into as needed.

There might be some buffers involved which means if you’re a bit slow to consume messages you will have a few 100 or 1000 there waiting depending on your settings, but this is just a buffer it’s not really a store and shouldn’t be relied on. If your process crash the buffer is lost. If the buffer overflow messages are lost.

The use of topic is often described as having Publish and Subscribe semantics since consumers Subscribe and every subscriber will get messages that are published.

Topics are often used in cases where you do not need reliable handling of your data. A stock symbol or high frequency status message from your monitoring system might go over topics. If you miss the current stock price soon enough the next update will come that would supersede the previous one so why would you queue them, perfect use for a broadcast based system.

Queues
Instead of broadcasting messages queues will store messages if no-one is around to consume them and a queue will load balance the work load across consumers.

This style of message is often used to create async workers that does some kind of long running task.

Imagine you need to convert many documents from MS Word to PDF – maybe after someone uploaded it to your site. You would create each job request in a queue and your converter process consumes the queue. If the converter is too slow or you need more capacity you simply need to add more consumers – perhaps even on different servers – and the middleware will ensure the traffic is load shared across the consumers.

You can therefore focus on a single function process – convert document to PDF – and the horizontal scalability comes at a very small cost on the part of the code since the middleware handles most of that for you. Messages are stored in the broker reliably and if you choose can even survive broker crashes and server reboots.

Additionally queues generally have a transaction metaphor, you start a transaction when you begin to process the document and if you crash mid processing the message will be requeued for later processing. To avoid a infinite loop of a bad message that crash all Consumers the brokers will also have a Dead Letter Queue where messages that have been retried too many times will go to sit in Limbo for an administrator to investigate.

These few basic features enable you to create software that resilient to failure, scalable and not susceptible to thundering herd problems. You can easily monitor the size of queues and know if your workers are not keeping up so you can provision more worker capacity – or retire unneeded capacity.

Demonstration
Playing with these concepts is very easy, you need a middleware broker and the Stomp library for Ruby, follow the steps below to install both in a simple sandbox that you can delete when you’re done. I’ll assume you installed Ruby, Rubygems and Python with your OS package management.

Note I am using CoilMQ here instead of the Ruby Stompserver since Stompserver has some bug with queues – they just don’t work right at all.

$ export GEM_HOME=/tmp/gems
$ export PYTHONPATH=/tmp/python
$ gem install stomp
$ mkdir /tmp/python; easy_install -d /tmp/python CoilMQ
$ /tmp/python/coilmq

At this point you have a working Stomp server that is listening on port 61613, you can just ^C it when you are done. If you want to do the stuff below using more than one machine then add to the command line for stompserver -b 0.0.0.0 and make sure port 61613 is open on your machine. The exercises below will work fine on one machine or twenty.

To test topics we first create multiple consumers, I suggest you do this in Screen and open multiple terms, for each terminal set the GEM_HOME as above.

Start 2 or 3 of these, these are consumers on the topic:

$ STOMP_HOST=localhost STOMP_PORT=61613 /tmp/gems/bin/stompcat /topic/my.input
Connecting to stomp://localhost:61613 as 
Getting output from /topic/my.input

Now we’ll create 1 producer and send a few messages, just type into the console I typed 1, 2, 3 (ignore the warnings about deprecation):

$ STOMP_HOST=localhost STOMP_PORT=61613 /tmp/gems/bin/catstomp /topic/my.input
Connecting to stomp://localhost:61613 as
Sending input to /topic/my.input
1
2
3

You should see these messages showing up on each of your consumers at roughly the same time and all your consumers should have received each message.

Now try the same with /queue/my.input instead of the topic and you should see that the messages are distributed evenly across your consumers.

You should also try to create messages with no consumers present and then subscribe consumers to the queue or topic, you’ll notice the difference in persistence behavior between topics and queues right away

When you’re done you can ^C everything and just rm /tmp/python and /tmp/gems.

That’s it for today, I’ll post several follow up posts soon.

UPDATE: part 2 has been published.