Making RabbitMQ Stew

I just wanted to write this up real quick for anyone who may run into the same problem I did...

I'm working on a project where I need to load some work from a queue for indexing into a Xapian Database (coming from CouchDB). So, RabbitMQ seemed like a good fit long-term for such a task.

Anyway, After getting RabbitMQ installed, I went to start'er up and noticed that it kept failing. So, off to the logs we go...and that is where I found the following:

Mnesia('rabbit@[myhostname]'): ** WARNING ** Mnesia is overloaded: {dump_log, ...

Hmm, well as much as I'd love to be pegged for resources I knew this wasn't the case. So after some googling I had found some suggestions that would squelch the warnings. What really seemed to do the trick was starting up rabbitmq-server with the following parameters: -mnesia dump_log_write_threshold 50000 -mnesia dc_dump_limit 40 (this was suggested from the Million User Comet Application

Now, since I know people don't really enjoy reading the documentation on things, I'll make it easier on you. From the mnesia documentation here is a quick description for the above-mentioned parameters.

-mnesia dc_dump_limit Number. Controls how often disc_copies tables are dumped from memory. Tables are dumped when filesize(Log) > (filesize(Tab)/Dc_dump_limit). Lower values reduces cpu overhead but increases disk space and startup times. The default is 4.

and

-mnesia dump_log_write_threshold Max, where Max is an integer which specifies the maximum number of writes allowed to the transaction log before a new dump of the log is performed. It defaults to 100 log writes.

I'd like to note that I can omit dump_log_write_threshold and still get rabbit up and running, so it may or may not help you out.

Now that I had begun passing these parameters to rabbitmq-server I was now getting a new error in the logs:

=INFO REPORT==== 7-Jan-2009::11:52:45 ===
application: rabbit
exited: {{timeout_waiting_for_tables,
[user,user_vhost,vhost,rabbit_config,listener,durable_routes,
route,reverse_route,durable_exchanges,exchange,
durable_queues,amqqueue]},
{rabbit,start,[normal,[]]}}
type: temporary

Oy, alright - so the tables don't seem to either be there or they just refuse to be created. Let's see if I can drop the mnedia tables that we created when starting the server and start over again.

So, I...
cd /var/lib/rabbitmq/mnesia/
an "ls" revealed two rabbit directories: rabbit/ and rabbit_20090107031256/
rm -rf rabbit/
rm -rf rabbit_20090107031256/

This path...is now clear! Let's start rabbitmq-server up again, but this time, without all the voodoo.
rabbitmq-server -detached

Starts up fine and now if we check our logs, we should see something like:
=INFO REPORT==== 7-Jan-2009::12:40:48 ===

Rolling persister log to "/var/lib/rabbitmq/mnesia/rabbit/rabbit_persister.LOG.previous"

=INFO REPORT==== 7-Jan-2009::12:40:48 ===
started TCP listener on 0.0.0.0:5672


The short of it - whipe your mnesia directories and start over.

Is your DataMapper CouchDB adapter PUT'ing where it shouldn't?

It sounds gross, sure...but this was actually a simple problem I foresee someone running into and potentially running around mad about for an hour if they're not too keen on how DataMapper/CouchDB work together. Even if you DO know how they work together, this was still a fun problem to track down.

So let's take an example model that's extremely simple:

class Person
include DataMapper::Resource
# CouchDB Specific
property :id, String, :key => true, :serial => true, :field => :_id
property :rev, String, :field => :_rev
#App Specific
property :name, String, :nullable => false, :length => 75, :key => true
end


Looks harmless, right? That's what I thought too! Well, if you begin your application and get to the point of actually creating a Person you'll be greeted with an error that kicks off with "EOFError: end of file reached ".

So what's happening here?

Basically in the adapter, there is a LOC in the create method where it checks to see if there is a key or not (Yea, the one we've specified above as :name).

This creates an assumption of using PUT instead of POST and that is where the issue stems from. So, to fix this problem, all you need to do is DROP the :key => true and things will run as they should.

Now you're probably asking "But, what if I NEED to have that key there, I NEED to relate my models with a foreign key". Not to sound harsh, but just to be clear, you may want to revisit why you're using CouchDB in the first place.

So what happened (the multi-key CouchDB question)?

So, if any of you have been following along with my fun ride upon the cozy cushions of CouchDB, you'll be happy to hear there is hope (well, some - I had to pick up a book on Erlang).

Anyhow, if you follow the link here you will see that there was a list of proposed fixes for my desire to pass in multiple keys to CouchDB instead of just one. Granted, yes it could get out of hand, you could pass in 1,000,000 keys. But so what! Either way you're going to have to query for them, whether it's a query after a query a billion times or if you can somehow lump them together.

So as you can see there were a few options:

1. We could execute our query with each key each time and lump them together. The downside of this is that I really don't want Ruby to have to lump all of those results together, I can only imagine the paging for it would get incredibly *blah*. Plus the memory usage for Ruby to merge all of those DM::Collection objects together would likely be the equivalent of a Chihuahua passing a peach seed. Not pretty.

2. We could use a full text search using Lucene, which CouchDB already has implemented. I suppose the only downside of that would be...??? I'll need to think of something, it's really not a bad idea, it just may come to me later. The extent of my full text search capabilities have been with MySQL and...oh, hold on I keep laughing, it sucked. bad. I really hope CouchDB's is better - I'm sure it is, I haven't seen anything so far that is crap!

3. We (read: Me {Bradford}) hack at the CouchDB internals and allow for views to accept multiple keys for comparison rather than one. I'd like to see this happen via something to the effect of:

A single key query URL
http://.../_view/foo/bars?key="mykey"

A Multi-key query URL
http://.../_view/foo/bars?key[]="mykey1"&key[]="mykey2"
OR
http://.../_view/foo/bars?keys=["mykey1","mykey2","mykey3"]

I'd like to keep it along the lines of form control arrays key[] this way you could just analyze the parameter itself to see what it was (an array or a string) and do the appropriate things with it. Whereas adding a parameter keys may need to be repeated throughout the code (Not likely, but the former makes sense to me most).

I'd like to extend a warm welcome to anyone willing to walk the lines of CouchDB's internals with me in order to implement such functionality, I promise to keep you in line as long as you return the favor. As I said, I'm JUST NOW learning Erlang and am horribly under-qualified, but hey, I really feel strongly about this functionality, so, what the hell, right?

Bring on the opinions!

It's all about document design with CouchDB

Hello everyone, been stalking the mailing list for a while and thought this might be worthy of a post as I was asked to solve it, yet, I couldn't!

Let's take the classic blog example document with the following fields/values:

"_id": "1f2fc3955b91aed5e7369f0b0ba8214e",
"_rev": "1226709986",
"Author": "Bradford",
"Type": "Post",
"Body": "Just mentioning this for a sample blog post.",
"PostedDate": "2008-07-02T23:22:12-04:00",
"Subject": "My Fine Blog Post",
"Tags": ["octopus","hockey","squidward","bradford","recreation"]

Next, I'd like to find each blog post that contains ANY of the following tags ["octopus","hockey"]. Now, generally speaking this isn't so bad. We could write a simple view:

function (doc) {
  if (doc.Type == 'Post') {
    for (var i = 0;i < doc.tags.length; i++) {
      emit(doc.tags[i],doc);
    }
  }
}

We would get back each one of our tags as a key, yea? Only if we supplied one at a time. So how does one go about supplying a range, array (not sure what we'd call it here) of keys to be searched on? http://...?key=["octopus","hockey"] maybe? I'm unsure of the plan of attack for such a thing. Maybe I'm just going about it in the wrong direction. Any thoughts?

So I finally bit the bullet and did it...a merb, datamapper and couchdb lovechild.

I got up this morning, wonderful day, got in to the office early, ground up some rocket fuel coffee beans and began my development day. Started off pretty sane, but slowly it turned into a full-blown hackfest!

Suprise, suprise...look whose coding!

Yes, my Cuban bee's after finally hitting my limitations with the CouchDB 0.7.2 adapter I took one for the team and got it working for 0.8.0.

This is by no means complete as I am going to build paging into the adapter as well before I submit it for usage in DataMapper. So without any further delay...Here are the goodies. I'll again just drop in what I've overloaded so you can have a pretty seamless transition if you need it. Also, this is currently running off of the CouchDB 0.8.0 trunk (see this page for subversion linkage)

Also, note the TODO, it should be done later this evening all things considered. Last but not least, I hope this helps someone other than myself, enjoy!

module DataMapper
  module Types
    class Object < DataMapper::Type
      primitive String
      size 65535
      lazy true
      track :hash

      def self.dump(value, property)
        value.to_json unless value.nil?
      end

      def self.load(value, property)
        value.nil? ? nil : value
      end
    end
  end
end

module DataMapper
  class Collection < LazyArray
    attr_reader :_total_rows
    private
      
    def initialize(query, options = {}, &block)
      assert_kind_of 'query', query, Query

      unless block_given?
        raise ArgumentError, 'a block must be supplied for lazy loading results', caller
      end

      @query          = query
      @key_properties = model.key(repository.name)
      @_total_rows = options[:total_rows] || 0
      super()
      load_with(&block)
    end

  end
end

module DataMapper
  module Adapters
    class CouchDBAdapter < AbstractAdapter

      def read_one(query)
        doc = request do |http|
          http.request(build_request(query))
        end
        unless doc["total_rows"] == 0
          data = doc['rows'].first
          query.model.load(
            query.fields.map do |property|
              data["value"][property.field.to_s]
            end,
            query)
        end
      end

      def read_many(query)
        doc = request do |http|
          http.request(build_request(query))
        end
        Collection.new(query, {:total_rows => doc['total_rows']}) do |collection|
          doc['rows'].each do |doc|
            collection.load(
              query.fields.map do |property|
                property.typecast(doc["value"][property.field.to_s])
              end
            )
          end
        end
      end
      def ad_hoc_request(query)
        if query.order.empty?
          key = "null"
        else
          key = (query.order.map do |order|
            "doc.#{order.property.field}"
          end).join(", ")
          key = "[#{key}]"
        end

        options = []
        options << "count=#{query.limit}" if query.limit
        options << "skip=#{query.offset}" if query.offset
        options = options.empty? ? nil : "?#{options.join('&')}"

        request = Net::HTTP::Post.new("/#{self.escaped_db_name}/_temp_view#{options}")
        request["Content-Type"] = "application/json"

        if query.conditions.empty?
          request.body = '{"language":"javascript","map":"function(doc) { if (doc.type == \'' + query.model.name.downcase + '\') { emit(' + key + ', doc);} }"}'
        else
          conditions = query.conditions.map do |operator, property, value|
            condition = "doc.#{property.field}"
            condition << case operator
            when :eql   then " == '#{value}'"
            when :not   then " != '#{value}'"
            when :gt    then " > #{value}"
            when :gte   then " >= #{value}"
            when :lt    then " < #{value}"
            when :lte   then " <= #{value}"
            when :like  then like_operator(value)
            end
          end
          request.body = '{"language":"javascript","map":"function(doc) {if (doc.type == \'' + query.model.name.downcase + '\') { if (' + conditions.join(' && ') + ') { emit(' + key + ', doc);}}}"}'
        end

        request
      end
      
      def request(parse_result = true, &block)
        res = nil
        Net::HTTP.start(@uri.host, @uri.port) do |http|
          res = yield(http)
        end
        # debugger
        JSON.parse(res.body) if parse_result
      end
      
      module Migration
        def create_model_storage(repository, model)
          assert_kind_of 'repository', repository, Repository
          assert_kind_of 'model',      model,      Resource::ClassMethods

          uri = "/#{self.escaped_db_name}/_design/#{model.storage_name(self.name)}"          
          view = Net::HTTP::Put.new(uri)
          view['content_type'] = "javascript"
          views = model.views.reject {|key, value| value.nil?}
          # TODO: This absolutely should be handled up a level.
          # You shouuld pass view a hash 
          #{:map => "function(doc){...}", :reduce => "function(doc){...}"}
          # We'll get there...
          view.body = { :views => views.each {|k,v| views[k] = {:map => v}} }.to_json

          request do |http|
            http.request(view)
          end
        end

        def destroy_model_storage(repository, model)
          assert_kind_of 'repository', repository, Repository
          assert_kind_of 'model',      model,      Resource::ClassMethods

          uri = "/#{self.escaped_db_name}/_design/#{model.storage_name(self.name)}"
          response = http_get(uri)
          unless response['error']
            uri += "?rev=#{response["_rev"]}"
            http_delete(uri)
          end
        end
      end
      
    end
  end
end