emr-trigger.rb

#!/usr/bin/ruby
#
# This script triggers builds for completed EMR tasks based on data in the metadata domain.  We assume that AWS
# credentials are specified in the 'AWS_ACCESS_KEY_ID' and 'AWS_SECRET_ACCESS_KEY' environment variables.
#
# Note that each invocation will only make one query (and possibly many puts) to SimpleDB.  If there 
# are a large number of post-EMR tasks to trigger, some may need to wait until the next invocation. 
#
# Hudson builds may be parameterized by selecting the "This build is parameterized" option in the job 
# configuration.  This script will pass JSON in the 'next_on_cron_server_job_params' to the build URL.  The JSON 
# should be in the following format:
#
# { "parameter" : [
#    {"name": "param1", "value": "val1"},
#    {"name": "param2", "value": "val2"},
#       ...
# ]}
#
# Note that this JSON is passed directly via a command line parameter, so all quotes must be escaped.
#
# Author: darren (darren@bizo.com)
# Date: 2009/12/17
#####

require 'rubygems'
require 'optparse'
require 'right_aws'  # gem install right_aws

domain = ""

opts = OptionParser.new
opts.banner = "Usage: #{$0} [-d DOMAIN]"
opts.on("-d", "--domain DOMAIN", "specify the domain to use") { |val|
  domain = val
}
opts.on("-h", "-?", "--help", "Display this screen" ) do
  puts opts
  exit
end

args = opts.parse(ARGV)

if (domain == "") then
  puts "Error: No domain specified."
  puts opts
  exit 1
end

sdb = RightAws::SdbInterface.new(ENV['AWS_ACCESS_KEY_ID'], ENV['AWS_SECRET_ACCESS_KEY'])

query_response = sdb.query(domain, "['next_on_cron_server_triggered' = 'false']")
unless query_response[:items].nil?
  query_response[:items].each { |item|
    status=`/opt/elastic-mapreduce-ruby/elastic-mapreduce --list -j #{item} | grep #{item} | awk '{print $2}' -`
    if "COMPLETED" == status.chomp
      puts "#{item} is completed"
      next_job_response = sdb.get_attributes(domain,item,nil)
      unless next_job_response[:attributes].nil? || next_job_response[:attributes]['next_on_cron_server_job_name'].nil?
        next_job_response[:attributes]['next_on_cron_server_job_name'].each { |next_job|
          job_params = next_job_response[:attributes]['next_on_cron_server_job_params']
          if job_params.nil?
            puts "Triggering #{next_job}"
            `curl localhost:8080/hudson/job/#{next_job}/build`
          else
            puts "Triggering #{next_job} with params #{job_params}"
            `curl -d 'json=#{job_params}' localhost:8080/hudson/job/#{next_job}/build`
          end
          if $?.exitstatus == 0
              sdb.put_attributes(domain,item,{ 'next_on_cron_server_triggered' => [ 'true' ] },true)             
          else
            puts "Could not trigger next build via HTTP call to localhost." 
          end
        }
      end
    else
      puts "#{item} is not completed"
    end 
  }
end