Select Page

Replicating NATS Streams between clusters

I’ve mentioned NATS before – the fast and light weight message broker from – but I haven’t yet covered the sister product NATS Streaming before so first some intro.

NATS Streaming is in the same space as Kafka, it’s a stream processing system and like NATS it’s super light weight delivered as a single binary and you do not need anything like Zookeeper. It uses normal NATS for communication and ontop of that builds streaming semantics. Like NATS – and because it uses NATS – it is not well suited to running over long cluster links so you end up with LAN local clusters only.

This presents a challenge since very often you wish to move data out of your LAN. I wrote a Replicator tool for NATS Streaming which I’ll introduce here.


First I guess it’s worth covering what Streaming is, I should preface also that I am quite new in using Stream Processing tools so I am not about to give you some kind of official answer but just what it means to me.

In a traditional queue like ActiveMQ or RabbitMQ, which I covered in my Common Messaging Patterns posts, you do have message storage, persistence etc but those who consume a specific queue are effectively a single group of consumers and messages either go to all or load shared all at the same pace. You can’t really go back and forth over the message store independently as a client. A message gets ack’d once and once it’s been ack’d it’s done being processed.

In a Stream your clients each have their own view over the Stream, they all have their unique progress and point in the Stream they are consuming and they can move backward and forward – and indeed join a cluster of readers if they so wish and then have load balancing with the other group members. A single message can be ack’d many times but once ack’d a specific consumer will not get it again.

This is to me the main difference between a Stream processing system and just a middleware. It’s a huge deal. Without it you will find it hard to build very different business tools centred around the same stream of data since in effect every message can be processed and ack’d many many times vs just once.

Additionally Streams tend to have well defined ordering behaviours and message delivery guarantees and they support clustering etc. much like normal middleware has. There’s a lot of similarity between streams and middleware so it’s a bit hard sometimes to see why you won’t just use your existing queueing infrastructure.

Replicating a NATS Stream

I am busy building a system that will move Choria registration data from regional data centres to a global store. The new Go based Choria daemon has a concept of a Protocol Adapter which can receive messages on the traditional NATS side of Choria and transform them into Stream messages and publish them.

This gets me my data from the high frequency, high concurrency updates from the Choria daemons into a Stream – but the Stream is local to the DC. Indeed in the DC I do want to process these messages to build a metadata store there but I also want to processes these messages for replication upward to my central location(s).

Hence the importance of the properties of Streams that I highlighted above – multiple consumers with multiple views of the Stream.

There are basically 2 options available:

  1. Pick a message from a topic, replicate it, pick the next one, one after the other in a single worker
  2. Have a pool of workers form a queue group and let them share the replication load

At the basic level the first option will retain ordering of the messages – order in the source queue will be the order in the target queue. NATS Streaming will try to redeliver a message that timed out delivery and it won’t move on till that message is handled, thus ordering is safe.

The 2nd option since you have multiple workers you have no way to retain ordering of the messages since workers will go at different rates and retries can happen in any order – it will be much faster though.

I can envision a 3rd option where I have multiple workers replicating data into a temporary store where on the other side I inject them into the queue in order but this seems super prone to failure, so I only support these 2 methods for now.

Limiting the rate of replication

There is one last concern in this scenario, I might have 10s of data centres all with 10s of thousands of nodes. At the DC level I can handle the rate of messages but at the central location where I might have 10s of DCs x 10s of thousands of machines if I had to replicate ALL the data at near real time speed I would overwhelm the central repository pretty quickly.

Now in the case of machine metadata you probably want the first piece of metadata immediately but from then on it’ll be a lot of duplicated data with only small deltas over time. You could be clever and only publish deltas but you have the problem then that should a delta publish go missing you end up with a inconsistent state – this is something that will happen in distributed systems.

So instead I let the replicator inspect your JSON, if your JSON has something like fqdn in it, it can look at that and track it and only publish data for any single matching sender every 1 hour – or whatever you configure.

This has the effect that this kind of highly duplicated data is handled continuously in the edge but that it only gets a snapshot replication upwards once a hour for any given node. This solves the problem neatly for me without there being any risks to deltas being lost, it’s also a lot simpler to implement.

Choria Stream Replicator

So finally I present the Choria Stream Replicator. It does all that was described above with a YAML configuration file, something like this:

debug: false                     # default
verbose: false                   # default
logfile: "/path/to/logfile"      # STDOUT default
state_dir: "/path/to/statedir"   # optional
        topic: acme.cmdb
        source_url: nats://source1:4222,nats://source2:4222
        source_cluster_id: dc1
        target_url: nats://target1:4222,nats://target2:4222
        target_cluster_id: dc2
        workers: 10              # optional
        queued: true             # optional
        queue_group: cmdb        # optional
        inspect: host            # optional
        age: 1h                  # optional
        monitor: 10000           # optional
        name: cmdb_replicator    # optional

Please review the README document for full configuration details.

I’ve been running this in a test DC with 1k nodes for a week or so and I am really happy with the results, but be aware this is new software so due care should be given. It’s available as RPMs, has a Puppet module, and I’ll upload some binaries on the next release.

Common Messaging Patterns Using Stomp – Part 5

This is a post in a series of about Middleware for Stomp users, please read the preceding parts starting at 1 before continuing below.

Today changing things around a bit and not so much talking about using Stomp from Ruby but rather how we would monitor ActiveMQ. The ActiveMQ broker has a statistics plugin that you can interact with over Stomp which is particularly nice – being able to interrogate it over the same protocols as you would to use it.

I’ll run through some basic approaches to monitor:

  • The size of queues
  • The memory usage of persisted messages on a queue
  • The rate of messages through a topic or a queue
  • Various memory usage statistics for the broker itself
  • Message counts and rates for the broker as a whole

These are your standard kinds of things you need to know about a running broker in addition to various things like monitoring the length of garbage collections and such which is standard when dealing with Java applications.

Keeping an eye on your queue sizes is very important. I’ve focused a lot on how Queues help you scale by facilitating horizontally adding consumers. Monitoring facilitates the decision making process for how many consumers you need – when to remove some and when to add some.

First you’re going to want to enable the plugin for ActiveMQ, open up your activemq.xml and add the plugin as below and restart when you are done:


A quick word about the output format of the messages you’ll see below. They are a serialized JSON (or XML) representation of a data structure. Unfortunately it isn’t immediately usable without some pre-parsing into a real data structure. The Nagios and Cacti plugins you will see below have a method in them for converting this structure into a normal Ruby hash.

The basic process for requesting stats is a Request Response pattern as per part 3.

stomp.subscribe("/temp-topic/stats", {"transformation" => "jms-map-json"})
# request stats for the random generator queue from part 2
stomp.publish("/queue/ActiveMQ.Statistics.Destination.random_generator", "", {"reply-to" => "/temp-topic/stats"})
puts stomp.receive.body

First we subscribe to a temporary topic that you first saw in Part 2 and we specify that while ActiveMQ will output a JMS Map it should please convert this for us into a JSON document rather than the java structures.

We then request Destination stats for the random_generator queue and finally wait for the response and print it, what you’ll get from it can be seen below:


Queue Statistics
Queue sizes are basically as you saw above, hit the Stats Plugin at /queue/ActiveMQ.Statistics.Destination.<queue name> and you get stats back for the queue in question.

Below table lists the meaning of these values from what I understand – quite conceivable I am wrong about the specifics of ones like enqueueTime for example so happy to be corrected in comments:

destinationName The name of the queue in JMS URL format
enqueueCount Amount of messages that was sent to the queue and committed to it
inflightCount Messages sent to the consumers but not consumed – they might be sat in the prefetch buffers
dequeueCount The opposite of enqueueCount – messages sent from the queue to consumers
dispatchCount Like dequeueCount but includes messages that might been rolled back
expiredCount Messages can have a maximum life, these are ones thats expired
maxEnqueueTime The maximum amount of time a message sat on the queue before being consumed
minEnqueueTime The minimum amount of time a message sat on the queue before being consumed
averageEnqueueTime The average amount of time a message sat on the queue before being consumed
memoryUsage Memory used by messages stored in the queue
memoryPercentUsage Percentage of available queue memory used
memoryLimit Total amount of memory this queue can use
size How many messages are currently in the queue
consumerCount Consumers currently subscribed to this queue
producerCount Producers currently producing messages

I have written a nagios plugin that can check the queue sizes:

$ check_activemq_queue.rb --host localhost --user nagios --password passw0rd --queue random_generator --queue-warn 10 --queue-crit 20
OK: ActiveMQ random_generator has 1 messages

You can see there’s enough information about the specific queue to be able to draw rate of messages, consumer counts and all sorts of useful information. I also have a quick script that will return all this data in a format suitable for use by Cacti:

$ activemq-cacti-plugin.rb --host localhost --user nagios --password passw0rd --report exim.stats
size:0 dispatchCount:168951 memoryUsage:0 averageEnqueueTime:1629.42897052992 enqueueCount:168951 minEnqueueTime:0.0 consumerCount:1 producerCount:0 memoryPercentUsage:0 destinationName:queue://exim.stats messagesCached:0 memoryLimit:20971520 inflightCount:0 dequeueCount:168951 expiredCount:0 maxEnqueueTime:328585.0

Broker Statistics
Getting stats for the broker is more of the same, just send a message to /queue/ActiveMQ.Statistics.Broker and tell it where to reply to, you’ll get a message back with these properties, I am only listing ones not seen above, the meanings is the same except in the broker stats its totals for all queues and topics.

storePercentUsage Total percentage of storage used for all queues
storeLimit Total storage space available
storeUsage Storage space currently used
tempLimit Total temporary space available
brokerId Unique ID for this broker that you will see in Advisory messages

dataDirectory Where the broker is configured to store its data for queue persistence etc
brokerName The name this broker was given in its configuration file

Additionally there would be a value for each of your connectors listing the URL to it including protocol and port

Again I have a Cacti plugin to get these values out in a format usable in Cacti data sources:

$ activemq-cacti-plugin.rb --host localhost --user nagios --password passw0rd --report broker
stomp+ssl:stomp+ssl storePercentUsage:81 size:5597 ssl:ssl vm:vm://web3 dataDirectory:/var/log/activemq/activemq-data dispatchCount:169533 brokerName:web3 openwire:tcp://web3:6166 storeUsage:869933776 memoryUsage:1564 tempUsage:0 averageEnqueueTime:1623.90502285799 enqueueCount:174080 minEnqueueTime:0.0 producerCount:0 memoryPercentUsage:0 tempLimit:104857600 messagesCached:0 consumerCount:2 memoryLimit:20971520 storeLimit:1073741824 inflightCount:9 dequeueCount:169525 brokerId:ID:web3-44651-1280002111036-0:0 tempPercentUsage:0 stomp:stomp://web3:6163 maxEnqueueTime:328585.0 expiredCount:0

You can find the plugins mentioned above in my GitHub account.

In the same location is a generic checker that publishes a message and wait for its return within a specified number of seconds – good turn around test for your broker.

I don’t really have good templates to share but you can see a Cacti graph I built below with the above plugins.

Common Messaging Patterns Using Stomp – Part 4

This is an ongoing post in a series of posts about Middlware for Stomp users, please read parts 1, 2 and 3 of this series first before continuing below.

Back in Part 2 we wrote a little system to ship metrics from nodes into Graphite via Stomp. This solved the goals of the problem then but now lets see what to do when our needs change.

Graphite is like RRD where it would summarize data over time and eventually discard old data. Contrast that with OpenTSDB that never summarizes or delete data and can store billions of data points. Imagine we want to use Graphite for a short term reporting service for our data but we also need to store the data long term without losing any data. So we really want to store the data in 2 locations.

We have a few options open to use:

  • Send the metric twice from every node, once to Graphite and once to OpenTSDB.
  • Write a software router that receives metrics on one queue and then route the metric to 2 other queues in the middleware.
  • Use facilities internal to the middleware to do the routing for us

The first option is an obvious bad idea and should just be avoided – this would be the worst case scenario for data collection at scale. The 3rd seems like the natural choice here but first we need to understand the facilities the middleware provides. Todays article will explore what ActiveMQ can do for you in this regard.

The 2nd seems an odd fit but as you’ll see below the capabilities for internal routing at the middleware layer isn’t all that exciting, useful in some cases but I think most projects will reach for some kind of message router in code sooner or later.

Virtual Destinations
If you think back to part 2 you’ll remember we have a publisher that publishes data into a queue and any number of consumers that consumes the queue. The queue will load balance the messages for us thus helping us scale.

In order to also create OpenTSDB data we essentially need to double up the consumer side into 2 groups. Ideally each set of consumers will be scalable horizontally and the sets of consumers should both get a copy of all the data – in other words we need 2 queues with all the data in it, one for Graphite and one for OpenTSDB.

You will also remember that Topics have the behavior of duplicating data they receive to all consumers of the topics. So really what we want is to attach 2 queues to a single topic. This way the topic will duplicate the data and the queues will be used for the scalable consumption of the data.

ActiveMQ provides a feature called Virtual Topics that solves this exact problem by convention. You publish messages to a predictably named topic and then you can create any number of queues that will all get a copy of that message.

The image above shows the convention:

  • Publish to /topic/VirtualTopic.metrics
  • Create consumers for /queue/Consumer.Graphite.VirtualTopic.metrics

Create as many of the consumer queues as you want, changing Graphite for some unique name and each of the resulting queues will behave like a normal queue with all the load balancing, storage and other queue like behaviors but all the queues will get a copy of all the data.

You can customize the name pattern of these queues by changing the ActiveMQ configuration files. I really like this approach to solving the problem vs approaches found in other brokers since this is all done by convention and you do not need to change your code to set up a bunch of internal structures that describes the routing topology. I consider routing topology that is living in code of the consumers to be a form of hard coding. Using this approach all I need to do is make sure the names of the destinations to publish to and consume from is configurable strings.

Our Graphite consumer would not need to change other than the name of the queue it should read and ditto for the producer.

If we find that we simply could not change the code for the consumers/producer or if it just was not a configurable setting you can still achieve this behavior by using something called a Composite Destinations in ActiveMQ that could describe this behavior purely in the config file with any arbitrarily named queues and topics.

Selective Consumers
Imagine we wish to give each one of our thousands of servers a unique destination on the middleware so that we can send machines a command directly. You could simple create queues like /queue/ and keep creating queues per server.

The problem with this approach is that internally to ActiveMQ each queue is a thread. So you’d be creating thousands of threads – not ideal.

As we saw before in Part 3 messages can have headers – there we used the reply-to header. Below you’ll find some code that sets an arbitrary header:

stomp.publish("/queue/nodes", "service restart httpd", {"fqdn" => ""})

We are publishing a message with the text service restart httpd to a queue and we are setting a fqdn header.

Now if every server in our estate subscribed to this one queue with the knowledge you have at this point of Queues this would have the effect of sending this restart request to some random one of our servers, not ideal!

The JMS specification allow for something called selectors to be used while subscribing to a destination:

stomp.subscribe("/queue/nodes", {"selector" => "fqdn = ''"})

The selector header sets the logic to apply to every message which will help decide if you get the message on your subscription or not. The selector language is defined using SQL 92 language and you can generally apply logic to any header in the message.

This way we set up a queue for all our servers without the overhead of 1000s of threads.

The choice for when to use a queue like this and when to use a traditional queue comes down to weighing up the overhead of validating all the SQL statements vs creating all the threads. There are also some side effects if you have a cluster of brokers – the queue traffic gets duplicated to all cluster brokers where with traditional queues the traffic only gets send to a broker if that broker actually has any subscribers interested in this data.

So you need to carefully consider the implications and do some tests with your work load, message sizes, message frequencies, amount of consumers etc.

There is a 3rd option that combines these 2 techniques. You’d create queues sourcing from the topic based on JMS Selectors deciding what data hits what queue. You would set up this arrangement in the ActiveMQ config file.

This, as far as I am aware, covers all the major areas internal to ActiveMQ that you can use to apply some routing and duplication of messages.

These methods are useful and solves some problems but as I pointed out it’s not really that flexible. In a later part of this series I will look into software routers from software like Apache Camel and how to write your own.

From a technology choices point of view future self is now thanking past self for building the initial metrics system using MOM since rather than go back to the drawing board when our needs changed we were able to solve our problems by virtue of the fact that we built it on a flexible foundation using well known patterns and without changing much if any actual code.

This series continue in part 5.

Common Messaging Patterns Using Stomp – Part 3

Yesterday I showed a detailed example of a Asynchronous system using MOM. Please read part 1 and part 2 of this series first before continuing below.

The system shown yesterday was Asynchronous since there is no coupling, no conversation or time constraints. The Producer does not know or care what happens to the messages once Published or when that happens. This is a kind of nirvana for distributed systems but sadly it’s just not possible to solve every problem using this pattern.

Today I’ll show how to use MOM technologies to solve a different kind of problem. Specifically I will show how large retailers scale their web properties using these technologies to create web sites that is more resilient to failure, easier to scale and easier to manage.

Imagine you are just hitting buy on some online retailers web page, perhaps they have implemented a 1-click based buying system where the very next page would be a Thank You page showing some details of your order and also recommendations of what other things you might like. It would have some personalized menus and in some cases even a personalized look and feel.

By the time you see this page your purchase is not complete, it is still going on in the background but you have a fast acknowledge back and immediately you are being enticed to spend more money with relevant products.

To achieve this in a PHP or Rails world you would typically have a page that runs top down and do things like generate your CSS page, generate your personalized menu then write some record into a database perhaps for a system like delayed job to process the purchase later on and finally it would do a bunch of SQL queries to find the related items.

This approach is very difficult to scale, all the hard work happens in your front controller, it has to be able to communicate with all the technology you choose in the backend and you end up with a huge monolithic chunk of code that can rapidly become a nightmare. If you need more capacity to render the recommendations you have no choice but to scale up the entire stack.

The better approach is to decouple all of the bits needed to generate a web page, if you take the narrative above you would have small single purpose services that does the following:

  • Take a 1-click order request, save it and provide an order number back. Start an Asynchronous process to fulfill the order.
  • Generate CSS for the custom look and feel for user X
  • Generate Menu for logged in user X
  • Generate recommendation for product Y based on browsing history for user X

Here we have 4 possible services that could exist on the network and that do not really relate to each other in any way. They are decoupled, do not share state with each other and can do their work in parallel independently from each other.

Your front controller now would become something that simply Published to the MOM requests for each of the 4 services providing just the information each service needs and then wait for the responses. Once all 4 responses were received the page would be assembled and rendered. If some response does not arrive in time a graceful failure can be done – like render a generic menu, or do not show the recommendations only show the Thank You text.

There are many benefits to this approach, I’ll highlight some I find compelling below:

  • You can scale each Service independently based on performance patterns – more order processors as this requires slower ACID writes into databases etc.
  • You can use different technology where appropriate. Your payment systems might be .Net while your CSS generation is in Node.JS and recommendations are in Java
  • Each system can be thought of as a simple standalone island with its own tech stack, monitoring etc, thinking about and scaling small components is much easier than a monolithic system
  • You can separate your payment processing from the rest of your network for PCI compliance by only allowing the middleware to connect into a private subnet where all actual credit information lives
  • Individual services can be upgraded or replaced with new ones much easier than in a monolithic system thus making the lives of Operations much better and lowering risk in ongoing maintenance of the system.
  • Individual services can be reused – the recommendation engine isn’t just a engine that gets called at the end of a sale but also while browsing through the store, the same service can serve both types of request

This pattern is often known as Request Response or similar terms. You should only use it when absolutely needed as it increases coupling and effectively turn your service into a Synchronous system but it does have it’s uses and advantages as seen above.

Sample Code
I’ll show 2 quick samples of how this conversation flow works in code and expand a bit into the details wrt to the ActiveMQ JMS Broker. The examples will just have the main working part of the code not the bits that would set up connections to the brokers etc, look in part 2 for some of that.

My example will create a service that generates random data using OpenSSL, maybe you have some reason to create a very large number of these and you need to distribute it across many machines so you do not run out of entropy.

As this is basically a Client / Server relationship I will use these terms, first the client part – the part that requests a random number from the server:

stomp.publish("/queue/random_generator", "random", {"reply-to" => "/temp-queue/random_replies"})
Timeout::timeout(2) do
   msg = stomp.receive
   puts "Got random number: #{msg.body}"    

This is pretty simple, the only new thing here is that we are subscribing first to a Temporary Queue that we will receive the responses on and we send the request including this queue name. Below will have some more detail on temp queues and temp topics. The timeout part is important you need this to be able to handle the case where all of the number generators died or if the service is just too overloaded to service the request.

Here is the server part, it gets a request then generates the number and replies to the reply-to destination.

require 'openssl'
loop do
      msg = client.receive
      number = OpenSSL::Random.random_bytes(8).unpack("Q").first
      stomp.publish(msg.headers["reply-to"], number)
      puts "Failed to generate random number: #{$!}"

You could start instances of this code on 10 servers and the MOM will load share the requests across the workers thus spreading out the available entropy across 10 machines.

When run this will give you nice big random numbers like 11519368947872272894. The web page in our example would follow a very similar process only it would post the requests to each of the services mentioned and then just wait for all the responses to come in and render them.

Temporary Destination
The big thing here is that we’re using a Temporary Queue for the replies. The behavior of temporary destinations differ from broker to broker and how the Stomp library needs to be used also changes. For ActiveMQ the behavior and special headers etc can be seen in their docs.

When you subscribe to a temporary destination like the client code above internally to ActiveMQ it sets up a queue that has a your connection as an exclusive subscriber. Internally the name would be something else entirely from what you gave it, it would be unique and exclusive to you. Here is an example for a Temporary Queue setup on a remote broker:


If you were to puts the contents of msg.headers[“reply-to”] in the server code you would see the translated queue name as above. The broker does this transparently for you.

Other processes can write to this unique destination but your connection would be the only one able to consume message from it. Soon as your connection closes or you unsubscribe from it the broker will free the queue, delete any messages on it and anyone else trying to write to it will get an exception.

Temporary queues and this magical translation happens even across a cluster of brokers so you can spread this out geographically and it would work.

Setting up a temporary queue and informing a network of brokers about it is a costly process so you should always try to set up a temporary queue early on in the process life time and reuse it for all the work you have to do.

If you need to correlate responses to their requests then you should use the correlation-id header for that – set it on the request and when constructing the reply read it from the request and set it again on the new reply.

This series continue in part 4.

Common Messaging Patterns Using Stomp – Part 2

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.

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.

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],]
  Timeout::timeout(2) do
    stomp ="", "", "", 61613)
    stomp.publish("/queue/graphite", msg)
rescue Timeout::Error
  STDERR.puts "Failed to send metric within the 2 second timeout"
  exit 1

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 0.1 1323597139

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

require 'rubygems'
require 'stomp'
def graphite
  @graphite ||="localhost", 2003)
client ="", "", "", 61613, true)
loop do
    msg = client.receive
    graphite.puts msg
    STDERR.puts "Failed to receive from queue: #{$!}"
    sleep 1

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.

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.

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.