ivanvanderbyl / cloudist Goto Github PK

View Code? Open in Web Editor NEW

174.0 4.0 4.0 199 KB

Cloudist is a simple, highly scalable job queue for Ruby applications

Home Page: testpilot.me

License: MIT License

Ruby 100.00%

cloudist's Introduction

This was an experiment, it's no longer maintained. Go learn Elixir/Phoenix :)

Cloudist

Cloudist is a simple, highly scalable job queue for Ruby applications, it can run within Rails, DaemonKit or your own custom application. Cloudist uses AMQP (RabbitMQ mainly) for transport and provides a simple DSL for queuing jobs and receiving responses including logs, exceptions and job progress.

Cloudist can be used to distribute long running tasks such as encoding a video, generating PDFs, scraping site data or even just sending emails. Unlike other job queues (DelayedJob etc) Cloudist does not load your entire Rails stack into memory for every worker, and it is not designed to, instead it expects all the data your worker requires to complete a job to be sent in the initial job request. This means your workers stay slim and can scale very quickly and even run on EC2 micros outside your applications environment without any further configuration.

Installation

gem install cloudist

Or if your app has a Gemfile:

gem 'cloudist', '~> 0.4.4'

Usage

Refer to examples.

Configuration

The only configuration required to get going are the AMQP settings, these can be set in two ways:

Using the AMQP_URL environment variable with value of amqp://username:password@localhost:5672/vhost
Updating the settings hash manually:

Cloudist.settings = {:user => 'guest', :pass => 'password', :vhost => '/', :host => 'localhost', :port => 5672}

Now and what's coming

Cloudist was developed to provide the messaging layer used within TestPilot Continuous Integration service.

TestPilot still uses Cloudist heavily and a number of features will be merged in soon.

Acknowledgements

Portions of this gem are based on code from the following projects:

Heroku's Droid gem
Lizzy
Minion

Contributing to Cloudist

Check out the latest master to make sure the feature hasn't been implemented or the bug hasn't been fixed yet
Check out the issue tracker to make sure someone already hasn't requested it and/or contributed it
Fork the project
Start a feature/bugfix branch e.g. git checkout -b feature-my-awesome-idea or bugfix-this-does-not-work
Commit and push until you are happy with your contribution
Make sure to add tests for it. This is important so I don't break it in a future version unintentionally.
Please try not to mess with the Rakefile, version, or history. If you want to have your own version, or is otherwise necessary, that is fine, but please isolate to its own commit so I can cherry-pick around it.

Authors

IvanVanderbyl - Blog

Copyright

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

cloudist's People

Contributors

Stargazers

Watchers

Forkers

akhil epictetus joevandyk jik0

cloudist's Issues

Re-architecting Cloudist to avoid message loss and improve scalability

TL;DR

Cloudist is broken, and as it's a crucial component in TestPilot, we are offering a free (forever) account on TestPilot to anyone who fixes it.

Note
References to "Ruby AMQP" mean https://github.com/ruby-amqp/amqp

The Problem

After using Cloudist in production we noticed it would lock up and messages wouldn't be sent or received. This is due partly to the way Ruby AMQP 0.6.7 works and was incorrectly implemented by me in the first instance. Essentially two things (that we are currently aware of) play a factor in this:

We open a new Channel every time we reply and never close them. This is an obvious problem, however solving this using the Async design of Ruby AMQP is tricky. One option would be to use Bunny and send all reply messages synchronously, another might be caching channels/queues instead of creating new ones - this is tricky because it needs to be thread safe and work within EventMachine.
We were not setting QoS correctly because Ruby AMQP ignores the prefetch option. We now know that we have to call #prefetch(1) on the channel after it connects, something like this:

    @connection = AMQP.connect(:host => "localhost") do
      puts "Connected to AMQP broker. Running #{AMQP::VERSION} version of the gem..."
    end

    @channel  = AMQP::Channel.new(connection) do |channel, open_ok|
      puts "Channel ##{channel.id} is now open!"
      @channel.prefetch(1, false) do
        puts "basic.qos callback has fired"
      end
    end

    @queue    = channel.queue(queue_name, :auto_delete => true)
    @exchange = channel.direct("")

The Solution

Re-architecting Cloudist to be less of a job queue or more of a high-performance two way messaging DSL which we can build a job queue on top of, optionally moving all messaging code to an adapter pattern so we can swap out AMQP for anything else which fits the bill, such as Beanstalkd or even Redis.

Solving the above problems: (1.) We will need to cache the reply queues instead of opening a new one every time we have a message to send, or design it in such a way that we can use Bunny for publishing and AMQP for consuming. (2.) Use the above code and update Cloudist to support Ruby AMQP 0.8+

Why can't TestPilot use Xyz job queue
TestPilot was built from the ground up to provide realtime build feedback, so much so that it will feel like running your tests locally – on the 30 machines you don't have sitting in your office. We don't believe any job queues that currently exist for Ruby allow us to do this.

The new architecture

I would like to separate Cloudist into 3 components which can be used together or independently.

Messaging, this would handle publishing and consuming of messages in a common format, including setting of headers and encoding decoding of payload.
Application/Worker, this would be our current Rails integration for class based listeners and workers, essentially building a layer which sits on top of 1. to use Cloudist as a message queue with all the advantages of being able to send realtime replies of progress, logs, errors and updates as the job is processed. This involves a listener in the Rails app and a worker instance deployed somewhere else completely standalone from the Rails app but using Cloudist it should be supplied with all the information it needs to process the job, and optionally do RPC calls back to the Rails app for more info if required.
Adapter pattern. It would be great if we could switch out the use of AMQP for Redis or Beanstalkd if desired.

But isn't re-architecting Cloudist overkill to solving this problem? On face value yes, but I think it will help simplify the design of Cloudist so that it will be of more benefit to the Ruby community and more than just another job queue.

The reward

I will give everyone involved in solving this problem free unlimited access to TestPilot once we launch, because this is the one blocking issue stopping us launching.

How to get involved.

Fork Cloudist
Start a feature/bugfix branch e.g. git checkout -b feature/my-awesome-idea or support/this-does-not-work
Commit and push until you are happy with your contribution
Make sure to add tests for it. This is important so I don't break it in a future version unintentionally.
Issue a Pull request.

Add comments below.

Examples not working

Using ruby 1.9.3p194 and cloudist 0.4.4, the examples fail with segmentation fault. Am I doing something wrong?

No `data` available to workers under ruby1.8

I encountered the following exception when running the "make sandwich" examples with ruby1.8.

If I run ruby worker_client.rb 1 (to request only one sandwich because I'm not very hungry) worker code will appear to behave as intended the first time. But when I ask for another sandwich (either by running my command line again, or by asking for 2 sandwiches after restarting the worker) the second time, just after the "Finished" displays I see the following exception in the console where I have been running the worker.

julien@dev:/data/sandbox7$ ruby worker.rb
I, [2011-07-21T07:35:26.577251 #23390]  INFO -- : Started Worker
Prefetch: 1
I, [2011-07-21T07:35:26.578369 #23390]  INFO -- : AMQP Subscribed: queue=make.sandwich exchange=
I, [2011-07-21T07:35:31.311294 #23390]  INFO -- : JOB (2c285500-9589-012e-61b3-08002795e76b) Make sandwich with white bread
D, [2011-07-21T07:35:36.320343 #23390] DEBUG -- : Finished Job in 5.01775193214417 seconds
worker.rb:23: warning: Object#id will be deprecated; use Object#object_id
worker.rb:23: undefined local variable or method `data' for Cloudist:Module (NameError)
        from /var/lib/gems/1.8/gems/amq-client-0.7.0.alpha9/lib/amq/client/exchange.rb:101:in `call'
        from /var/lib/gems/1.8/gems/amq-client-0.7.0.alpha9/lib/amq/client/exchange.rb:101:in `publish'
        from /var/lib/gems/1.8/gems/amqp-0.8.0.rc9/lib/amqp/exchange.rb:457:in `publish'
        from /var/lib/gems/1.8/gems/eventmachine-0.12.10/lib/em/deferrable.rb:47:in `call'
        from /var/lib/gems/1.8/gems/eventmachine-0.12.10/lib/em/deferrable.rb:47:in `callback'
        from /var/lib/gems/1.8/gems/amqp-0.8.0.rc9/lib/amqp/channel.rb:253:in `once_open'
        from /var/lib/gems/1.8/gems/amqp-0.8.0.rc9/lib/amqp/exchange.rb:456:in `publish'
        from /var/lib/gems/1.8/gems/eventmachine-0.12.10/lib/eventmachine.rb:996:in `call'
        from /var/lib/gems/1.8/gems/eventmachine-0.12.10/lib/eventmachine.rb:996:in `run_deferred_callbacks'
        from /var/lib/gems/1.8/gems/eventmachine-0.12.10/lib/eventmachine.rb:996:in `each'
        from /var/lib/gems/1.8/gems/eventmachine-0.12.10/lib/eventmachine.rb:996:in `run_deferred_callbacks'
        from /var/lib/gems/1.8/gems/eventmachine-0.12.10/lib/eventmachine.rb:256:in `run_machine'
        from /var/lib/gems/1.8/gems/eventmachine-0.12.10/lib/eventmachine.rb:256:in `run'
        from /var/lib/gems/1.8/gems/amqp-0.8.0.rc9/lib/amqp/connection.rb:39:in `start'
        from /var/lib/gems/1.8/gems/cloudist-0.4.3/lib/cloudist.rb:61:in `start'
        from worker.rb:19
julien@dev:/data/sandbox7$

If I run the same worker with ruby 1.9.2 (and the same version of installed gems) everything seems to run fine.

I have the following versions installed:

julien@dev:/data/sandbox7$ ruby -v
ruby 1.8.7 (2010-08-16 patchlevel 302) [x86_64-linux]
julien@dev:/data/sandbox7$ gem -v
1.3.7
julien@dev:/data/sandbox7$ gem list

*** LOCAL GEMS ***

activesupport (3.0.9)
amq-client (0.7.0.alpha9)
amq-protocol (0.7.0)
amqp (0.8.0.rc9)
bundler (1.0.12)
cloudist (0.4.3)
eventmachine (0.12.10)
hashie (1.0.0)
i18n (0.6.0)
json (1.4.6)
macaddr (1.0.0)
uuid (2.3.2)
julien@dev:/data/sandbox7$

I can work around the problem by running the worker code with ruby 1.9.2 but I would prefer to work with 1.8, as I am trying to build an application with QtRuby that will also make good use of cloudist.

Sincerely,
Julien

Bump amqp dependency to 0.8.1?

It's set to a pre-release version right now.

Problems running tests

  [[email protected]:~/projects/cloudist(master)]  $ rake
WARNING: 'require 'rake/rdoctask'' is deprecated.  Please use 'require 'rdoc/task' (in RDoc 2.4.2+)' instead.
    at /Users/joe/.rvm/gems/ree-1.8.7-2011.03@n2/gems/rake-0.9.2.2/lib/rake/rdoctask.rb
/Users/joe/.rvm/rubies/ree-1.8.7-2011.03/bin/ruby -S bundle exec rspec spec/cloudist/basic_queue_spec.rb spec/cloudist/job_spec.rb spec/cloudist/message_spec.rb spec/cloudist/messaging_spec.rb spec/cloudist/payload_spec.rb spec/cloudist/payload_spec_2_spec.rb spec/cloudist/queue_spec.rb spec/cloudist/request_spec.rb spec/cloudist/utils_spec.rb spec/cloudist_spec.rb spec/core_ext/string_spec.rb
F.....FFFFFF.FFFFF.F.F........F...Prefetch: 1
FPrefetch: 1
FPrefetch: 1
FPrefetch: 1
FPrefetch: 1
FPrefetch: 1
F......

Failures:

  1) Cloudist Cloudist::Queues::BasicQueue should create a queue and exchange
     Failure/Error: bq = Cloudist::Queues::BasicQueue.new("make.sandwich")
     RuntimeError:
       AMQP can only be used from within EM.run {}
     # ./spec/../lib/cloudist/queues/basic_queue.rb:35:in `new'
     # ./spec/../lib/cloudist/queues/basic_queue.rb:35:in `setup'
     # ./spec/../lib/cloudist/queues/basic_queue.rb:25:in `initialize'
     # ./spec/cloudist/basic_queue_spec.rb:19:in `new'
     # ./spec/cloudist/basic_queue_spec.rb:19

  2) Cloudist::Payload should extract published_on from data
     Failure/Error: payload.headers['published_on'].should == "12345678"
       expected: "12345678"
            got: 1321297561.04786 (using ==)
     # ./spec/cloudist/payload_spec.rb:29

  3) Cloudist::Payload should extract custom event hash from data
     Failure/Error: payload.body.should == {:bread=>"white"}
       expected: {:bread=>"white"}
            got: <#Hashie::Mash bread="white" event_hash="foo"> (using ==)
       Diff:
       @@ -1,2 +1,2 @@
       -{:bread=>"white"}
       +{"event_hash"=>"foo", "bread"=>"white"}
     # ./spec/cloudist/payload_spec.rb:34

  4) Cloudist::Payload should parse JSON message
     Failure/Error: payload = Cloudist::Payload.new(Marshal.dump({:bread => 'white', :event_hash => 'foo'}))
     JSON::ParserError:
       705: unexpected token at {:event_hashfoo:
       bread"
       white'
     # ./spec/../lib/cloudist/encoding.rb:13:in `decode'
     # ./spec/../lib/cloudist/payload.rb:12:in `initialize'
     # ./spec/cloudist/payload_spec.rb:39:in `new'
     # ./spec/cloudist/payload_spec.rb:39

  5) Cloudist::Payload should parse custom headers
     Failure/Error: payload = Cloudist::Payload.new(Marshal.dump({:bread => 'white', :event_hash => 'foo'}), {:published_on => 12345})
     JSON::ParserError:
       705: unexpected token at {:event_hashfoo:
       bread"
       white'
     # ./spec/../lib/cloudist/encoding.rb:13:in `decode'
     # ./spec/../lib/cloudist/payload.rb:12:in `initialize'
     # ./spec/cloudist/payload_spec.rb:44:in `new'
     # ./spec/cloudist/payload_spec.rb:44

  6) Cloudist::Payload should create a unique event hash
     Failure/Error: payload.create_event_hash.size.should == 32
     NoMethodError:
       undefined method `create_event_hash' for #<Cloudist::Payload:0x107893048>
     # ./spec/../lib/cloudist/payload.rb:86:in `method_missing'
     # ./spec/cloudist/payload_spec.rb:50

  7) Cloudist::Payload should not create a new event hash unless it doesn't have one
     Failure/Error: payload.event_hash.size.should == 32
     NoMethodError:
       undefined method `event_hash' for #<Cloudist::Payload:0x10788f3a8>
     # ./spec/../lib/cloudist/payload.rb:86:in `method_missing'
     # ./spec/cloudist/payload_spec.rb:55

  8) Cloudist::Payload should format payload for sending
     Failure/Error: body.should == Marshal.dump({:bread => 'white'})
       expected: "\004\b{\006:\nbread\"\nwhite"
            got: "{\"bread\":\"white\"}" (using ==)
       Diff:
       @@ -1,4 +1,2 @@
       {:
       -bread"
       -white
       +{"bread":"white"}
     # ./spec/cloudist/payload_spec.rb:70

  9) Cloudist::Payload should generate a unique payload ID
     Failure/Error: payload.id.size.should == 32
       expected: 32
            got: 36 (using ==)
     # ./spec/cloudist/payload_spec.rb:77

  10) Cloudist::Payload should allow setting of payload ID
     Failure/Error: payload.id = "2345"
     NoMethodError:
       undefined method `id=' for #<Cloudist::Payload:0x10787d2e8>
     # ./spec/../lib/cloudist/payload.rb:86:in `method_missing'
     # ./spec/cloudist/payload_spec.rb:82

  11) Cloudist::Payload should allow changing of payload after being published
     Failure/Error: payload.publish
     NoMethodError:
       undefined method `publish' for #<Cloudist::Payload:0x10787a610>
     # ./spec/../lib/cloudist/payload.rb:86:in `method_missing'
     # ./spec/cloudist/payload_spec.rb:88

  12) Cloudist::Payload should freeze
     Failure/Error: payload.publish
     NoMethodError:
       undefined method `publish' for #<Cloudist::Payload:0x107875db8>
     # ./spec/../lib/cloudist/payload.rb:86:in `method_missing'
     # ./spec/cloudist/payload_spec.rb:96

  13) Cloudist::Payload should not overwrite passed in headers
     Failure/Error: payload.headers[:ttl].should == "25"
       expected: "25"
            got: 25 (using ==)
     # ./spec/cloudist/payload_spec.rb:114

  14) Cloudist::Payload should be able to transport an error
     Failure/Error: payload = Cloudist::Payload.new(e, {:message_type => 'error'})
     NoMethodError:
       undefined method `each_pair' for #<ArgumentError: FAILED>
     # ./spec/../lib/cloudist/payload.rb:13:in `new'
     # ./spec/../lib/cloudist/payload.rb:13:in `initialize'
     # ./spec/cloudist/payload_spec.rb:127:in `new'
     # ./spec/cloudist/payload_spec.rb:127

  15) Cloudist::Payload should parse custom headers
     Failure/Error: payload = Cloudist::Payload.new(Marshal.dump({:bread => 'white'}), {:published_on => 12345, :message_id => "foo"})
     JSON::ParserError:
       705: unexpected token at {:
       bread"
       white'
     # ./spec/../lib/cloudist/encoding.rb:13:in `decode'
     # ./spec/../lib/cloudist/payload.rb:12:in `initialize'
     # ./spec/cloudist/payload_spec_2_spec.rb:50:in `new'
     # ./spec/cloudist/payload_spec_2_spec.rb:50

  16) Cloudist::Request should return ttl
     Failure/Error: q = Cloudist::JobQueue.new('test.queue')
     RuntimeError:
       AMQP can only be used from within EM.run {}
     # ./spec/../lib/cloudist/queues/basic_queue.rb:35:in `new'
     # ./spec/../lib/cloudist/queues/basic_queue.rb:35:in `setup'
     # ./spec/../lib/cloudist/queues/basic_queue.rb:25:in `initialize'
     # ./spec/../lib/cloudist/queues/job_queue.rb:10:in `initialize'
     # ./spec/cloudist/request_spec.rb:8:in `new'
     # ./spec/cloudist/request_spec.rb:8

  17) Cloudist::Request should have a payload
     Failure/Error: q = Cloudist::JobQueue.new('test.queue')
     RuntimeError:
       AMQP can only be used from within EM.run {}
     # ./spec/../lib/cloudist/queues/basic_queue.rb:35:in `new'
     # ./spec/../lib/cloudist/queues/basic_queue.rb:35:in `setup'
     # ./spec/../lib/cloudist/queues/basic_queue.rb:25:in `initialize'
     # ./spec/../lib/cloudist/queues/job_queue.rb:10:in `initialize'
     # ./spec/cloudist/request_spec.rb:8:in `new'
     # ./spec/cloudist/request_spec.rb:8

  18) Cloudist::Request should be 1 minute old
     Failure/Error: q = Cloudist::JobQueue.new('test.queue')
     RuntimeError:
       AMQP can only be used from within EM.run {}
     # ./spec/../lib/cloudist/queues/basic_queue.rb:35:in `new'
     # ./spec/../lib/cloudist/queues/basic_queue.rb:35:in `setup'
     # ./spec/../lib/cloudist/queues/basic_queue.rb:25:in `initialize'
     # ./spec/../lib/cloudist/queues/job_queue.rb:10:in `initialize'
     # ./spec/cloudist/request_spec.rb:8:in `new'
     # ./spec/cloudist/request_spec.rb:8

  19) Cloudist::Request should not be expired
     Failure/Error: q = Cloudist::JobQueue.new('test.queue')
     RuntimeError:
       AMQP can only be used from within EM.run {}
     # ./spec/../lib/cloudist/queues/basic_queue.rb:35:in `new'
     # ./spec/../lib/cloudist/queues/basic_queue.rb:35:in `setup'
     # ./spec/../lib/cloudist/queues/basic_queue.rb:25:in `initialize'
     # ./spec/../lib/cloudist/queues/job_queue.rb:10:in `initialize'
     # ./spec/cloudist/request_spec.rb:8:in `new'
     # ./spec/cloudist/request_spec.rb:8

  20) Cloudist::Request should not be acked yet
     Failure/Error: q = Cloudist::JobQueue.new('test.queue')
     RuntimeError:
       AMQP can only be used from within EM.run {}
     # ./spec/../lib/cloudist/queues/basic_queue.rb:35:in `new'
     # ./spec/../lib/cloudist/queues/basic_queue.rb:35:in `setup'
     # ./spec/../lib/cloudist/queues/basic_queue.rb:25:in `initialize'
     # ./spec/../lib/cloudist/queues/job_queue.rb:10:in `initialize'
     # ./spec/cloudist/request_spec.rb:8:in `new'
     # ./spec/cloudist/request_spec.rb:8

  21) Cloudist::Request should be ackable
     Failure/Error: q = Cloudist::JobQueue.new('test.queue')
     RuntimeError:
       AMQP can only be used from within EM.run {}
     # ./spec/../lib/cloudist/queues/basic_queue.rb:35:in `new'
     # ./spec/../lib/cloudist/queues/basic_queue.rb:35:in `setup'
     # ./spec/../lib/cloudist/queues/basic_queue.rb:25:in `initialize'
     # ./spec/../lib/cloudist/queues/job_queue.rb:10:in `initialize'
     # ./spec/cloudist/request_spec.rb:8:in `new'
     # ./spec/cloudist/request_spec.rb:8

Finished in 0.02376 seconds
46 examples, 21 failures
rake aborted!
ruby -S bundle exec rspec spec/cloudist/basic_queue_spec.rb spec/cloudist/job_spec.rb spec/cloudist/message_spec.rb spec/cloudist/messaging_spec.rb spec/cloudist/payload_spec.rb spec/cloudist/payload_spec_2_spec.rb spec/cloudist/queue_spec.rb spec/cloudist/request_spec.rb spec/cloudist/utils_spec.rb spec/cloudist_spec.rb spec/core_ext/string_spec.rb failed

Tasks: TOP => default => spec
(See full trace by running task with --trace)

Could be my environment? Ruby 1.9.2 on OSX, using rabbitmq 2.5.1.

Also, have you looked at using travis-ci? I think they support rabbitmq.