integration tests

bin/redstorm

@@ -5,11 +5,11 @@ require 'rake'
 
 begin
   # will work from gem, since lib dir is in gem require_paths
-  require 'red_storm'
+  require 'red_storm/application'
 rescue LoadError
   # will work within RedStorm dev project
   $:.unshift './lib'
-  require 'red_storm'
+  require 'red_storm/application'
 end
 
-RedStorm::Application.new.run(ARGV.dup)
+RedStorm::Application.run(ARGV.dup)

lib/red_storm.rb

@@ -3,7 +3,6 @@ require 'rubygems'
 require 'red_storm/environment'
 require 'red_storm/version'
 require 'red_storm/configuration'
-require 'red_storm/application'
 require 'red_storm/simple_bolt'
 require 'red_storm/simple_spout'
 require 'red_storm/simple_topology'

lib/red_storm/application.rb

@@ -1,10 +1,40 @@
+require 'red_storm/version'
+require 'red_storm/environment'
+
+CWD = Dir.pwd
+TARGET_DIR = "#{CWD}/target"
+TARGET_LIB_DIR = "#{TARGET_DIR}/lib"
+TARGET_SRC_DIR = "#{TARGET_DIR}/src"
+TARGET_GEM_DIR = "#{TARGET_DIR}/gems/gems"
+TARGET_SPECS_DIR = "#{TARGET_DIR}/gems/specifications"
+TARGET_CLASSES_DIR = "#{TARGET_DIR}/classes"  
+TARGET_DEPENDENCY_DIR = "#{TARGET_DIR}/dependency"
+TARGET_DEPENDENCY_UNPACKED_DIR = "#{TARGET_DIR}/dependency-unpacked"
+TARGET_CLUSTER_JAR = "#{TARGET_DIR}/cluster-topology.jar"
+
+REDSTORM_JAVA_SRC_DIR = "#{RedStorm::REDSTORM_HOME}/src/main"
+REDSTORM_LIB_DIR = "#{RedStorm::REDSTORM_HOME}/lib"
+
+SRC_EXAMPLES = "#{RedStorm::REDSTORM_HOME}/examples"
+DST_EXAMPLES = "#{CWD}/examples"
+
+
 module RedStorm
   
   class Application 
     TASKS_FILE = "#{RedStorm::REDSTORM_HOME}/lib/tasks/red_storm.rake" 
 
-    def usage
-      puts("usage: redstorm install")
+    def self.local_storm_command(class_file, ruby_mode = nil)
+      "java -Djruby.compat.version=#{RedStorm.jruby_mode_token(ruby_mode)} -cp \"#{TARGET_CLASSES_DIR}:#{TARGET_DEPENDENCY_DIR}/*\" redstorm.TopologyLauncher local #{class_file}"
+    end
+
+    def self.cluster_storm_command(class_file, ruby_mode = nil)
+      "storm jar #{TARGET_CLUSTER_JAR} -Djruby.compat.version=#{RedStorm.jruby_mode_token(ruby_mode)} redstorm.TopologyLauncher cluster #{class_file}"
+    end
+    
+    def self.usage
+      puts("usage: redstorm version")
+      puts("       redstorm install")
       puts("       redstorm deps")
       puts("       redstorm build")
       puts("       redstorm examples")
@@ -15,9 +45,12 @@ module RedStorm
       exit(1)
     end
 
-    def run(args)
+    def self.run(args)
       if args.size > 0
-        if ["install", "examples", "jar", "bundle", "deps", "build"].include?(args[0])
+        if args[0] == "version"
+          puts("RedStorm v#{VERSION}")
+          exit
+        elsif ["install", "examples", "jar", "bundle", "deps", "build"].include?(args[0])
           load(TASKS_FILE)
           Rake::Task[args.shift].invoke(args.join(":"))
           exit

lib/tasks/red_storm.rake

@@ -2,27 +2,11 @@ require 'ant'
 require 'jruby/jrubyc'
 require 'pompompom'
 require 'red_storm'
- 
+require 'red_storm/application'
+
 INSTALL_STORM_VERSION = "0.7.4"
 INSTALL_JRUBY_VERSION = "1.6.7.2"
 
-CWD = Dir.pwd
-TARGET_DIR = "#{CWD}/target"
-TARGET_LIB_DIR = "#{TARGET_DIR}/lib"
-TARGET_SRC_DIR = "#{TARGET_DIR}/src"
-TARGET_GEM_DIR = "#{TARGET_DIR}/gems/gems"
-TARGET_SPECS_DIR = "#{TARGET_DIR}/gems/specifications"
-TARGET_CLASSES_DIR = "#{TARGET_DIR}/classes"  
-TARGET_DEPENDENCY_DIR = "#{TARGET_DIR}/dependency"
-TARGET_DEPENDENCY_UNPACKED_DIR = "#{TARGET_DIR}/dependency-unpacked"
-TARGET_CLUSTER_JAR = "#{TARGET_DIR}/cluster-topology.jar"
-
-REDSTORM_JAVA_SRC_DIR = "#{RedStorm::REDSTORM_HOME}/src/main"
-REDSTORM_LIB_DIR = "#{RedStorm::REDSTORM_HOME}/lib"
-
-SRC_EXAMPLES = "#{RedStorm::REDSTORM_HOME}/examples"
-DST_EXAMPLES = "#{CWD}/examples"
-
 module JavaZip
   import 'java.util.zip.ZipFile'
 end
@@ -33,13 +17,13 @@ task :launch, :env, :ruby_mode, :class_file do |t, args|
   
   command = case args[:env]
   when "local"
-    "java -Djruby.compat.version=#{version_token} -cp \"#{TARGET_CLASSES_DIR}:#{TARGET_DEPENDENCY_DIR}/*\" redstorm.TopologyLauncher local #{args[:class_file]}"
+    RedStorm::Application.local_storm_command(args[:class_file], args[:ruby_mode])
   when "cluster"
     unless File.exist?(TARGET_CLUSTER_JAR)
       puts("cluster jar file #{TARGET_CLUSTER_JAR} not found. Generate it using $redstorm jar DIR1 [DIR2, ...]")
       exit(1)
     end
-    "storm jar #{TARGET_CLUSTER_JAR} -Djruby.compat.version=#{version_token} redstorm.TopologyLauncher cluster #{args[:class_file]}"
+    RedStorm::Application.cluster_storm_command(args[:class_file], args[:ruby_mode])
   end
 
   puts("launching #{command}")

test/README.md

@@ -0,0 +1,3 @@
+# Integration Tests
+
+These are integration tests to automate the testing of topology submission in a local and remote cluster. More documentation on usage to come.

test/integration/run_all.sh

@@ -0,0 +1,39 @@
+TOPOLOGIES=test/topology/*.rb
+COMMANDS=("redstorm" "bin/redstorm" "bundle exec redstorm")
+REDSTORM=""
+
+# figure correct command
+for c in "${COMMANDS[@]}"; do
+  $c version &> /dev/null
+  if [ $? -eq 0 ]; then
+    REDSTORM=$c
+    break
+  fi
+done
+if [ "$REDSTORM" == "" ]; then
+  echo "redstorm command not found"
+  exit 1
+fi
+
+if [[ $@ != "noinstall" ]]; then
+  # install target
+  rm -rf target
+  bundle install
+  $REDSTORM install 
+  $REDSTORM bundle topology
+  $REDSTORM jar test
+fi
+
+echo "runnig integration tests..."
+
+# run local mode tests
+for t in $TOPOLOGIES; do
+  echo -n "local $t " 
+  ruby test/integration/run_local.rb $t
+done
+
+# run cluster mode tests
+for t in $TOPOLOGIES; do
+  echo -n "cluster $t "
+  ruby test/integration/run_remote.rb $t
+done

test/integration/run_local.rb

@@ -0,0 +1,28 @@
+begin
+  # will work from gem, since lib dir is in gem require_paths
+  require 'red_storm/application'
+rescue LoadError
+  # will work within RedStorm dev project
+  $:.unshift './lib'
+  require 'red_storm/application'
+end
+
+require 'redis'
+
+topology_class_path = ARGV[0]
+topology_class = topology_class_path.split("/").last
+
+@redis = Redis.new(:host => "localhost", :port => 6379)
+@redis.del(topology_class)
+
+command = RedStorm::Application.local_storm_command(topology_class_path)
+pid = spawn("#{command} > /dev/null", :out=>"/dev/null")
+
+result = @redis.blpop(topology_class, :timeout => 30)
+sleep(5) if result
+
+if result.nil? || result[1] != "SUCCESS"
+  puts("FAILED, bad result=#{result.inspect}")
+  exit(1)
+end
+puts("SUCCESS")

test/integration/run_remote.rb

@@ -0,0 +1,36 @@
+begin
+  # will work from gem, since lib dir is in gem require_paths
+  require 'red_storm/application'
+rescue LoadError
+  # will work within RedStorm dev project
+  $:.unshift './lib'
+  require 'red_storm/application'
+end
+
+require 'redis'
+
+topology_class_path = ARGV[0]
+topology_class = topology_class_path.split("/").last
+
+@redis = Redis.new(:host => "localdevhost", :port => 6379)
+@redis.del(topology_class)
+
+command = RedStorm::Application.cluster_storm_command(topology_class_path)
+unless system("#{command} > /dev/null")
+  puts("FAILED, #{$!}")
+  exit(1)
+end
+
+result = @redis.blpop(topology_class, :timeout => 60)
+
+command = "storm kill #{topology_class.split(".").first} > /dev/null"
+unless system(command)
+  puts("FAILED, #{$!}")
+  exit(1)
+end
+
+if result.nil? || result[1] != "SUCCESS"
+  puts("FAILED, bad result=#{result.inspect}")
+  exit(1)
+end
+puts("SUCCESS")

test/topology/basic_topology.rb

@@ -0,0 +1,48 @@
+require 'red_storm'
+require 'thread'
+require 'redis'
+
+class SingleTupleSpout < RedStorm::SimpleSpout
+  output_fields :string
+
+  on_init do
+    @q = Queue.new
+    @q << "SUCCESS"
+  end
+
+  on_send do
+    # avoid putting the thread to sleep endlessly on @q.pop which will prevent local cluster.shutdown
+    sleep(1)
+    @q.pop unless @q.empty?
+  end
+end
+
+class RedisPushBolt < RedStorm::SimpleBolt
+  on_init {@redis = Redis.new(:host => "localhost", :port => 6379)}
+
+  on_receive :emit => false do |tuple|
+    data = tuple.getValue(0).to_s
+    @redis.lpush(File.basename(__FILE__), data)
+  end
+end
+
+class BasicTopology < RedStorm::SimpleTopology
+  spout SingleTupleSpout, :parallelism => 1
+
+  bolt RedisPushBolt, :parallelism => 1 do
+    source SingleTupleSpout, :global
+  end
+
+  configure do |environment|
+    debug false
+  end
+
+  on_submit do |environment|
+    case environment
+    when :local
+      sleep(10)
+      cluster.shutdown
+    end
+  end
+
+end

test/topology/reliable_topology.rb

@@ -0,0 +1,91 @@
+require 'red_storm'
+require 'thread'
+require 'redis'
+
+class ReliableSpout < RedStorm::SimpleSpout
+  output_fields :string
+
+  on_init do
+    @redis = Redis.new(:host => "localhost", :port => 6379)
+    @ids = [1, 2]
+
+    @q = Queue.new
+    @ids.each{|id| @q << id}
+  end
+
+  on_fail do |id|
+    log.info("***** FAIL #{id}")
+  end
+
+  on_ack do |id|
+    @ids.delete(id)
+    log.info("***** ACK #{id}")
+    if @ids.empty?
+      log.info("*** SUCCESS")
+      @redis.lpush(File.basename(__FILE__), "SUCCESS")
+    end
+  end
+
+  on_send :reliable => true do
+    # avoid putting the thread to sleep endlessly on @q.pop which will prevent local cluster.shutdown
+    sleep(1)
+    unless @q.empty?
+      id = @q.pop
+      [id, "DATA#{id}"] # reliable & autoemit, first element must be message_id
+    end
+  end
+end
+
+class AckBolt < RedStorm::SimpleBolt
+  on_receive :emit => false do |tuple|
+    ack(tuple)
+  end
+end
+
+class ImplicitPassthruBolt < RedStorm::SimpleBolt
+  output_fields :string
+
+  on_receive :emit => true, :ack => true, :anchor => true do |tuple|
+    tuple.getString(0)
+  end
+end
+
+class ExplicitPassthruBolt < RedStorm::SimpleBolt
+  output_fields :string
+
+  on_receive :emit => false do |tuple|
+    anchored_emit(tuple, tuple.getString(0))
+    ack(tuple)
+  end
+end
+
+class ReliableTopology < RedStorm::SimpleTopology
+  spout ReliableSpout, :parallelism => 1
+
+  bolt ImplicitPassthruBolt, :parallelism => 1 do
+    source ReliableSpout, :shuffle
+  end
+
+  bolt ExplicitPassthruBolt, :parallelism => 1 do
+    source ImplicitPassthruBolt, :shuffle
+  end
+
+  bolt AckBolt, :parallelism => 1 do
+    source ExplicitPassthruBolt, :shuffle
+  end
+
+  configure do |environment|
+    debug true
+    message_timeout_secs 10
+    num_ackers 2
+  end
+
+  on_submit do |environment|
+    case environment
+    when :local
+      sleep(10)
+      cluster.shutdown
+    end
+  end
+
+end