Pillow, the CouchDB shard manager

If there is one thing that has bothered me about my choice of CouchDB as the main storage system for Sincerial, it’s the lack of an automatic system for shard management. In the early days of a startup, a single server is probably capable of handling all the necessary data. However, a successful service that is built around the harvesting and analysis of data will sooner or later have to shard the dataset across multiple servers. And for a new service, the sooner you have to start sharding, the better. Several good distributed storage systems exist. Google’s original bigtable, HBase (Hadoop’s bigtable equivalent), Cassandra and more solve this particular problem, but there is more to choosing and running a storage system than just data volume scaling. CouchDB has other strengths that made it a good choice for our application, but that is not the topic of this post.

CouchDB-lounge originally written by Kevin Ferguson (macfergus), Vijay Ragunathan (lukatmyshu) and Shaun Lindsay (srlindsay) at Meebo.com will handle distribution of requests to the right servers in a cluster of servers, but you still have to handle resharding or repartitioning manually. CouchDB-lounge consists of two components, dumbproxy handling reading and writing of documents and smartproxy handling views. These require Nginx and Twisted (a Python framework) respectively. If you overshard appropriately, you can scale your data volume a long way before you have to start resharding.

While manually repartitioning a CouchDB database is doable, I’d rather have an automatic way of doing it since I don’t want to make mistakes. In addition, the Sincerial system uses Ruby running with Phusion Passenger in Apache and I didn’t want to add two more frameworks on top of that. This might sound like a not-invented here excuse, but it isn’t or at least I don’t think it is.

When I started developing Pillow, I chose to do so in erlang to match couchdb. The reason was two-fold. First of all I was curious about erlang and I like functional programming. Secondly, CouchDB was written in erlang and there had to be a reason for that. Now I’ve released version 0.3 of Pillow. This version supports automatic resharding, routing of requests to the right shard and views. Reducers need to be written in erlang, but a summing reducer is in place and mappers without reducers are supported out of the box. As such, this version of Pillow has all the functionality I set out to develop, but it does not support the full CouchDB API.

The bulk document API is not supported. And I haven’t tried running standalone CouchApps. The reason being that I am focusing on our needs. I intend to have full support of the CouchDB API eventually, but it might be that I integrate more tightly with CouchDB and use CouchDB as a library to make this happen. This will make supporting javascript reducers easier as well. It would of course be interesting to have Pillow become an integral part of CouchDB as well providing one can still access each CouchDB server directly for maintenance purposes. The latter being one of the reasons I’m in general a bit sceptical to distributed systems that hide the inner workings since often hard to fix problems that may occur.

CouchDB Replication Monitor

CouchDB does replication, but replication needs to be set up after each server restart. This means you need to ensure that replication is restarted whenever the daemon restarts CouchDB. I have never seen replication stop working without a restart, but I prefer being safe to being sorry about replication. To be perfectly honest, I do not trust that my replication initiation after a soft CouchDB restart works properly either so I prefer to monitor the replication and have a safety mechanism in place to restart replication if needed.

There are several ways to monitor replication. You could fetch the status page of all servers and restart replication on servers with an empty page, but that is a kind of brute force approach in my world. A better solution is to use the replication itself to monitor that it works.

Each server updates their timestamp in CouchDB and this is again replicated to the other servers. This gets us a bit of the way, but not all the way. The server you are checking might have received updates from all the other servers, but you don’t know if it’s pushed out anything to the other servers. To solve this, you can add information about the other servers to the local server as well. This will give you a matrix of server replication status.

For each server, you will see the timestamp replicated from the server and a list of timestamps replicated to that server. The latter often being a generation older than the former. Cron can be used to update this data. The cronjob reads all the server timestamps and updates this servers timestamp followed by a list of the other servers timestamp.

A mapper to get a server id to server status out of the db.

map: function(doc) {
  emit(doc._id, doc);
}

Our monitroing database is called server_status. The design containing the mapper is called collections and the view server_list.

A Ruby database checker that can run on cron.

require 'rubygems'
require 'couchrest'
require 'json'
require 'open-uri'

STATUS_DB = 'http://localhost:5984/server_status'
COLLECTIONS = 'collections'
SERVER_LIST = 'server_list'

hostname = ARGV[0]

status_db = CouchRest.database!(STATUS_DB)
status_view = "#{STATUS_DB}/_design/#{COLLECTIONS}/_view/#{SERVER_LIST}"

# Get the current information about this server if available
server_status = begin
  status_db.get(hostname)
rescue RestClient::ResourceNotFound
  {'_id' => hostname}
end

server_status['time'] = Time.new.to_i
# Get the current times of the other servers and update this server's
# view of them
JSON(open(status_view).read)['rows'].map do |row|
  {'server' => row['id'], 'status' => row['value']}
end.each do |status|
  unless status['server'] == hostname
    server_status['servers'][status['server']] = status['status']['time']
  end
end
status_db.save_doc(server_status)

Now you need to determine when to trigger replication restart. This can be handled in the watchdog cronjob. If the highest timestamp seen for this server at other servers is above a threshold, restart replication.

The final loop triggering when the age is above a threshold. The init_replication method just posts a continuous replication trigger to the db:

JSON(open(status_view).read)['rows'].map do |row|
  {'server' => row['id'], 'status' => row['value']}
end.each do |status|
  if server_status['time'] - status['status']['time'] > THRESHOLD
    init_replication(status['server'])
  end
  unless status['server'] == hostname
    server_status['servers'][status['server']] = status['status']['time']
  end
end

Rudimentary init_replication method.

def init_replication(server)
  target = "http://#{server}:5984"
  databases = ['server_status']
  databases.each do |db|
    config = {
            'source' => "#{db}",
            'target' => "#{target}/#{db}",
            'continuous' => true
    }
    payload = JSON.generate(config)
    result = Net::HTTP.new('127.0.0.1', '5984').post(
      '/_replicate', payload, {'content-type' => 'text/x-json'})
    unless result.code == 200
      p "replication to #{target}/#{db} failed with #{result.code}"
    end
  end
end

We have a monitoring view of replication ages in our system. It shows the matrix of timestamps as age in seconds rather than the actual timestamp since the age is the important metric.

A bonus of this replication monitoring system is that we can access the status page from a mobil phone and get an accurate picture of the replication status. This doesn’t worry me now, but it did when we first set it up. Now it’s just a part of our general monitoring view.