Merge pull request #45 from arturopie/runners-event

Runners event

.gitignore

@@ -18,5 +18,7 @@ coverage
 rdoc
 pkg
 tags
+.rvmrc
+hydra-runner.log
 
 ## PROJECT::SPECIFIC

lib/hydra.rb

@@ -13,4 +13,4 @@ require 'hydra/listener/minimal_output'
 require 'hydra/listener/report_generator'
 require 'hydra/listener/notifier'
 require 'hydra/listener/progress_bar'
-
+require 'hydra/runner_listener/abstract'

lib/hydra/master.rb

@@ -64,6 +64,10 @@ module Hydra #:nodoc:
         listener = eval(l)
         @event_listeners << listener if listener.is_a?(Hydra::Listener::Abstract)
       end
+
+      @string_runner_event_listeners = Array( opts.fetch( 'runner_listeners' ) { nil } )
+
+      @runner_log_file = opts.fetch('runner_log_file') { nil }
       @verbose = opts.fetch('verbose') { false }
       @autosort = opts.fetch('autosort') { true }
       @sync = opts.fetch('sync') { nil }
@@ -160,7 +164,7 @@ module Hydra #:nodoc:
       pipe = Hydra::Pipe.new
       child = SafeFork.fork do
         pipe.identify_as_child
-        Hydra::Worker.new(:io => pipe, :runners => runners, :verbose => @verbose)
+        Hydra::Worker.new(:io => pipe, :runners => runners, :verbose => @verbose, :runner_listeners => @string_runner_event_listeners, :runner_log_file => @runner_log_file )
       end
 
       pipe.identify_as_parent
@@ -172,7 +176,7 @@ module Hydra #:nodoc:
 
       runners = worker.fetch('runners') { raise "You must specify the number of runners"  }
       command = worker.fetch('command') {
-        "RAILS_ENV=#{@environment} ruby -e \"require 'rubygems'; require 'hydra'; Hydra::Worker.new(:io => Hydra::Stdio.new, :runners => #{runners}, :verbose => #{@verbose});\""
+        "RAILS_ENV=#{@environment} ruby -e \"require 'rubygems'; require 'hydra'; Hydra::Worker.new(:io => Hydra::Stdio.new, :runners => #{runners}, :verbose => #{@verbose}, :runner_listeners => \'#{@string_runner_event_listeners}\', :runner_log_file => \'#{@runner_log_file}\' );\""
       }
 
       trace "Booting SSH worker"

lib/hydra/runner.rb

@@ -13,14 +13,23 @@ module Hydra #:nodoc:
   class Runner
     include Hydra::Messages::Runner
     traceable('RUNNER')
+
+    DEFAULT_LOG_FILE = 'hydra-runner.log'
+
     # Boot up a runner. It takes an IO object (generally a pipe from its
     # parent) to send it messages on which files to execute.
     def initialize(opts = {})
-      @io = opts.fetch(:io) { raise "No IO Object" } 
+      redirect_output( opts.fetch( :runner_log_file ) { DEFAULT_LOG_FILE } )
-      @verbose = opts.fetch(:verbose) { false }      
+      reg_trap_sighup
-      $stdout.sync = true
-      trace 'Booted. Sending Request for file'
 
+      @io = opts.fetch(:io) { raise "No IO Object" }
+      @verbose = opts.fetch(:verbose) { false }
+      @event_listeners = Array( opts.fetch( :runner_listeners ) { nil } )
+
+      $stdout.sync = true
+      runner_begin
+
+      trace 'Booted. Sending Request for file'
       @io.write RequestFile.new
       begin
         process_messages
@@ -30,6 +39,20 @@ module Hydra #:nodoc:
       end
     end
 
+    def reg_trap_sighup
+      for sign in [:SIGHUP, :INT]
+        trap sign do
+          stop
+        end
+      end
+      @runner_began = true
+    end
+
+    def runner_begin
+      trace "Firing runner_begin event"
+      @event_listeners.each {|l| l.runner_begin( self ) }
+    end
+
     # Run a test file and report the results
     def run_file(file)
       trace "Running file: #{file}"
@@ -53,7 +76,17 @@ module Hydra #:nodoc:
 
     # Stop running
     def stop
-      @running = false
+      runner_end if @runner_began
+      @runner_began = @running = false
+    end
+
+    def runner_end
+      trace "Ending runner #{self.inspect}"
+      @event_listeners.each {|l| l.runner_end( self ) }
+    end
+
+    def format_exception(ex)
+      "#{ex.class.name}: #{ex.message}\n    #{ex.backtrace.join("\n    ")}"
     end
 
     private
@@ -74,7 +107,7 @@ module Hydra #:nodoc:
           end
         rescue IOError => ex
           trace "Runner lost Worker"
-          @running = false
+          stop
         end
       end
     end
@@ -83,10 +116,6 @@ module Hydra #:nodoc:
       "Error in #{file}:\n  #{format_exception(ex)}"
     end
 
-    def format_exception(ex)
-      "#{ex.class.name}: #{ex.message}\n    #{ex.backtrace.join("\n    ")}"
-    end
-
     # Run all the Test::Unit Suites in a ruby file
     def run_test_unit_file(file)
       begin
@@ -263,5 +292,15 @@ module Hydra #:nodoc:
         end
       end.compact
     end
+
+    def redirect_output file_name
+      begin
+        $stderr = $stdout =  File.open(file_name, 'a')
+      rescue
+        # it should always redirect output in order to handle unexpected interruption
+        # successfully
+        $stderr = $stdout =  File.open(DEFAULT_LOG_FILE, 'a')
+      end
+    end
   end
 end

lib/hydra/runner_listener/abstract.rb

@@ -0,0 +1,23 @@
+module Hydra #:nodoc:
+  module RunnerListener #:nodoc:
+    # Abstract listener that implements all the events
+    # but does nothing.
+    class Abstract
+      # Create a new listener.
+      #
+      # Output: The IO object for outputting any information.
+      # Defaults to STDOUT, but you could pass a file in, or STDERR
+      def initialize(output = $stdout)
+        @output = output
+      end
+
+      # Fired by the runner just before requesting the first file
+      def runner_begin( runner )
+      end
+
+      # Fired by the runner just after stoping
+      def runner_end( runner )
+      end
+    end
+  end
+end

lib/hydra/tasks.rb

@@ -41,6 +41,10 @@ module Hydra #:nodoc:
     # Set to false if you don't want to show the total running time
     attr_accessor :show_time
 
+    # Set to a valid file path if you want to save the output of the runners
+    # in a log file
+    attr_accessor :runner_log_file
+
     #
     # Search for the hydra config file
     def find_config_file
@@ -98,7 +102,8 @@ module Hydra #:nodoc:
         :autosort => @autosort,
         :files => @files,
         :listeners => @listeners,
-        :environment => @environment
+        :environment => @environment,
+        :runner_log_file => @runner_log_file
       }
       if @config
         @opts.merge!(:config => @config)

lib/hydra/worker.rb

@@ -21,6 +21,15 @@ module Hydra #:nodoc:
       @listeners = []
 
       load_worker_initializer
+
+      @runner_event_listeners = Array(opts.fetch(:runner_listeners) { nil })
+      @runner_event_listeners.select{|l| l.is_a? String}.each do |l|
+        @runner_event_listeners.delete_at(@runner_event_listeners.index(l))
+        listener = eval(l)
+        @runner_event_listeners << listener if listener.is_a?(Hydra::RunnerListener::Abstract)
+      end
+      @runner_log_file = opts.fetch(:runner_log_file) { nil }
+
       boot_runners(opts.fetch(:runners) { 1 })
       @io.write(Hydra::Messages::Worker::WorkerBegin.new)
 
@@ -83,7 +92,7 @@ module Hydra #:nodoc:
         pipe = Hydra::Pipe.new
         child = SafeFork.fork do
           pipe.identify_as_child
-          Hydra::Runner.new(:io => pipe, :verbose => @verbose)
+          Hydra::Runner.new(:io => pipe, :verbose => @verbose, :runner_listeners => @runner_event_listeners, :runner_log_file => @runner_log_file )
         end
         pipe.identify_as_parent
         @runners << { :pid => child, :io => pipe, :idle => false }
@@ -121,7 +130,7 @@ module Hydra #:nodoc:
             end
           rescue IOError => ex
             trace "Worker lost Master"
-            Thread.exit
+            shutdown
           end
         end
       end

test/fixtures/hydra_worker_init.rb

@@ -0,0 +1,2 @@
+require '../test/fixtures/runner_listeners.rb'
+require '../test/fixtures/master_listeners.rb'

test/fixtures/master_listeners.rb

@@ -0,0 +1,10 @@
+module HydraExtension
+  module Listener
+    class WorkerBeganFlag < Hydra::Listener::Abstract
+      # Fired after runner processes have been started
+      def worker_begin(worker)
+        FileUtils.touch File.expand_path(File.join(Dir.consistent_tmpdir, 'worker_began_flag'))
+      end
+    end
+  end
+end

test/fixtures/runner_listeners.rb

@@ -0,0 +1,23 @@
+module HydraExtension
+  module RunnerListener
+    class RunnerBeginTest < Hydra::RunnerListener::Abstract
+      # Fired by the runner just before requesting the first file
+      def runner_begin( runner )
+        FileUtils.touch File.expand_path(File.join(Dir.consistent_tmpdir, 'alternate_hydra_test.txt'))
+      end
+    end
+
+    class RunnerEndTest < Hydra::RunnerListener::Abstract
+      # Fired by the runner just before requesting the first file
+      def runner_begin( runner )
+        FileUtils.touch File.expand_path(File.join(Dir.consistent_tmpdir, 'runner_began_flag')) #used to know when the runner is ready
+      end
+      # Fired by the runner just after stoping
+      def runner_end( runner )
+        # NOTE: do not use trace here
+        #runner.trace "Ending runner"
+        FileUtils.touch File.expand_path(File.join(Dir.consistent_tmpdir, 'alternate_hydra_test.txt'))
+      end
+    end
+  end
+end

test/master_test.rb

@@ -1,4 +1,6 @@
 require File.join(File.dirname(__FILE__), 'test_helper')
+require File.join(File.dirname(__FILE__), 'fixtures', 'runner_listeners')
+require File.join(File.dirname(__FILE__), 'fixtures', 'master_listeners')
 
 class MasterTest < Test::Unit::TestCase
   context "with a file to test and a destination to verify" do
@@ -115,8 +117,8 @@ class MasterTest < Test::Unit::TestCase
         :workers => [{
           :type => :ssh,
           :connect => 'localhost',
-          :directory => File.expand_path(File.join(File.dirname(__FILE__), '..', 'lib')),
+          :directory => remote_dir_path,
-          :runners => 1 
+          :runners => 1
         }]
       )
       assert File.exists?(target_file)
@@ -178,4 +180,204 @@ class MasterTest < Test::Unit::TestCase
       assert !File.exists?(File.join(remote, 'test_b.rb')), "B was not deleted"
     end
   end
+
+  context "with a runner_end event" do
+    setup do
+      # avoid having other tests interfering with us
+      sleep(0.2)
+      FileUtils.rm_f(target_file)
+      FileUtils.rm_f(alternate_target_file)
+
+      @runner_began_flag = File.expand_path(File.join(Dir.consistent_tmpdir, 'runner_began_flag')) #used to know when the worker is ready
+      FileUtils.rm_f(@runner_began_flag)
+
+      @runner_listener = 'HydraExtension::RunnerListener::RunnerEndTest.new' # runner_end method that creates alternate_target_file
+      @master_listener = HydraExtension::Listener::WorkerBeganFlag.new  #used to know when the runner is up
+    end
+
+    teardown do
+      FileUtils.rm_f(target_file)
+      FileUtils.rm_f(alternate_target_file)
+    end
+
+    context "running a local worker" do
+      should "run runner_end on successful termination" do
+        @pid = Process.fork do
+            Hydra::Master.new(
+              :files => [test_file] * 6,
+              :autosort => false,
+              :listeners => [@master_listener],
+              :runner_listeners => [@runner_listener],
+              :verbose => false
+            )
+          end
+        Process.waitpid @pid
+
+        assert_file_exists alternate_target_file
+      end
+
+      should "run runner_end after interruption signal" do
+        add_infinite_worker_begin_to @master_listener
+
+        capture_stderr do # redirect stderr
+          @pid = Process.fork do
+            Hydra::Master.new(
+              :files => [test_file],
+              :autosort => false,
+              :listeners => [@master_listener],
+              :runner_listeners => [@runner_listener],
+              :verbose => false
+            )
+          end
+        end
+        wait_for_runner_to_begin
+
+        Process.kill 'SIGINT', @pid
+        Process.waitpid @pid
+
+        assert_file_exists alternate_target_file
+      end
+    end
+
+    context "running a remote worker" do
+      setup do
+        copy_worker_init_file # this method has a protection to avoid erasing an existing worker_init_file
+      end
+
+      teardown do
+        FileUtils.rm_f(@remote_init_file) unless @protect_init_file
+      end
+
+      should "run runner_end on successful termination" do
+        capture_stderr do # redirect stderr
+          @pid = Process.fork do
+            Hydra::Master.new(
+              :files => [test_file],
+              :autosort => false,
+              :listeners => [@master_listener],
+              :runner_listeners => [@runner_listener],
+              :workers => [{
+                :type => :ssh,
+                :connect => 'localhost',
+                :directory => remote_dir_path,
+                :runners => 1
+              }],
+              :verbose => false
+            )
+          end
+        end
+        Process.waitpid @pid
+
+        assert_file_exists alternate_target_file
+      end
+    end
+  end
+
+  context "redirecting runner's output and errors" do
+    setup do
+      # avoid having other tests interfering with us
+      sleep(0.2)
+      FileUtils.rm_f(target_file)
+      FileUtils.rm_f(runner_log_file)
+      FileUtils.rm_f("#{remote_dir_path}/#{runner_log_file}")
+    end
+
+    teardown do
+      FileUtils.rm_f(target_file)
+      FileUtils.rm_f(runner_log_file)
+      FileUtils.rm_f("#{remote_dir_path}/#{runner_log_file}")
+    end
+
+    should "create a runner log file when usign local worker and passing a log file name" do
+      @pid = Process.fork do
+        Hydra::Master.new(
+              :files => [test_file],
+              :runner_log_file => runner_log_file,
+              :verbose => false
+            )
+      end
+      Process.waitpid @pid
+
+      assert_file_exists target_file # ensure the test was successfully ran
+      assert_file_exists runner_log_file
+    end
+
+    should "create a runner log file when usign remote worker and passing a log file name" do
+      @pid = Process.fork do
+        Hydra::Master.new(
+              :files => [test_file],
+              :workers => [{
+                :type => :ssh,
+                :connect => 'localhost',
+                :directory => remote_dir_path,
+                :runners => 1
+              }],
+              :verbose => false,
+              :runner_log_file => runner_log_file
+            )
+      end
+      Process.waitpid @pid
+
+      assert_file_exists target_file # ensure the test was successfully ran
+      assert_file_exists "#{remote_dir_path}/#{runner_log_file}"
+    end
+
+    should "create the default runner log file when passing an incorrect log file path" do
+      default_log_file = "#{remote_dir_path}/#{Hydra::Runner::DEFAULT_LOG_FILE}" # hydra-runner.log"
+      FileUtils.rm_f(default_log_file)
+
+      @pid = Process.fork do
+        Hydra::Master.new(
+              :files => [test_file],
+              :workers => [{
+                :type => :ssh,
+                :connect => 'localhost',
+                :directory => remote_dir_path,
+                :runners => 1
+              }],
+              :verbose => false,
+              :runner_log_file => 'invalid-dir/#{runner_log_file}'
+            )
+      end
+      Process.waitpid @pid
+
+      assert_file_exists target_file # ensure the test was successfully ran
+      assert_file_exists default_log_file #default log file
+      assert !File.exists?( "#{remote_dir_path}/#{runner_log_file}" )
+
+      FileUtils.rm_f(default_log_file)
+    end
+  end
+
+  private
+
+  def runner_log_file
+    "my-hydra-runner.log"
+  end
+
+  def add_infinite_worker_begin_to master_listener
+    class << master_listener
+      def worker_begin( worker )
+        super
+        sleep 1 while true #ensure the process doesn't finish before killing it
+      end
+    end
+  end
+
+  #  this requires that a worker_begin listener creates a file named worker_began_flag in tmp directory
+  def wait_for_runner_to_begin
+    assert_file_exists @runner_began_flag
+  end
+
+  # with a protection to avoid erasing something important in lib
+  def copy_worker_init_file
+    @remote_init_file = "#{remote_dir_path}/#{File.basename( hydra_worker_init_file )}"
+    if File.exists?( @remote_init_file )
+      $stderr.puts "\nWARNING!!!: #{@remote_init_file} exits and this test needs to create a new file here. Make sure there is nothing inportant in that file and remove it before running this test\n\n"
+      @protect_init_file = true
+      exit
+    end
+    # copy the hydra_worker_init to the correct location
+    FileUtils.cp(hydra_worker_init_file, remote_dir_path)
+  end
 end

test/runner_test.rb

@@ -1,4 +1,5 @@
 require File.join(File.dirname(__FILE__), 'test_helper')
+require File.join(File.dirname(__FILE__), 'fixtures', 'runner_listeners')
 
 class RunnerTest < Test::Unit::TestCase
   context "with a file to test and a destination to verify" do
@@ -79,46 +80,91 @@ class RunnerTest < Test::Unit::TestCase
       # because of all the crap cucumber pulls in
       # we run this in a fork to not contaminate
       # the main test environment
-      pid = Process.fork do
+      capture_stderr do # redirect stderr
-        runner = Hydra::Runner.new(:io => File.new('/dev/null', 'w'))
+        pid = Process.fork do
-        runner.run_file(cucumber_feature_file)
+          runner = Hydra::Runner.new(:io => File.new('/dev/null', 'w'))
-        assert File.exists?(target_file)
+          runner.run_file(cucumber_feature_file)
-        assert_equal "HYDRA", File.read(target_file)
+          assert File.exists?(target_file)
-        
+          assert_equal "HYDRA", File.read(target_file)
-        FileUtils.rm_f(target_file)
+
-        
+          FileUtils.rm_f(target_file)
-        runner.run_file(alternate_cucumber_feature_file)
+
-        assert File.exists?(alternate_target_file)
+          runner.run_file(alternate_cucumber_feature_file)
-        assert_equal "HYDRA", File.read(alternate_target_file)
+          assert File.exists?(alternate_target_file)
-        assert !File.exists?(target_file)
+          assert_equal "HYDRA", File.read(alternate_target_file)
+          assert !File.exists?(target_file)
+        end
+        Process.wait pid
       end
-      Process.wait pid
     end
 
     should "be able to run a runner over ssh" do
       ssh = Hydra::SSH.new(
-        'localhost', 
+        'localhost',
         File.expand_path(File.join(File.dirname(__FILE__), '..', 'lib')),
         "ruby -e \"require 'rubygems'; require 'hydra'; Hydra::Runner.new(:io => Hydra::Stdio.new, :verbose => true);\""
       )
       assert ssh.gets.is_a?(Hydra::Messages::Runner::RequestFile)
       ssh.write(Hydra::Messages::Worker::RunFile.new(:file => test_file))
-      
+
       # grab its response. This makes us wait for it to finish
       echo = ssh.gets # get the ssh echo
       response = ssh.gets
 
       assert_equal Hydra::Messages::Runner::Results, response.class
-      
+
       # tell it to shut down
       ssh.write(Hydra::Messages::Worker::Shutdown.new)
 
       ssh.close
-      
+
       # ensure it ran
       assert File.exists?(target_file)
       assert_equal "HYDRA", File.read(target_file)
     end
+
+    context "using runner events" do
+      context "on successful termination" do
+        setup do
+          @pipe = Hydra::Pipe.new
+          @parent = Process.fork do
+            request_a_file_and_verify_completion(@pipe, test_file)
+          end
+        end
+
+        should "fire runner_begin event" do
+          run_the_runner(@pipe,  [HydraExtension::RunnerListener::RunnerBeginTest.new] )
+          Process.wait(@parent)
+
+          # ensure runner_begin was fired
+          assert_file_exists alternate_target_file
+        end
+
+        should "fire runner_end event" do
+          run_the_runner(@pipe,  [HydraExtension::RunnerListener::RunnerEndTest.new] )
+          Process.wait(@parent)
+
+          assert_file_exists alternate_target_file
+        end
+      end
+
+      should "fire runner_end event after losing communication with worker" do
+        pipe = Hydra::Pipe.new
+        parent = Process.fork do
+          pipe.identify_as_parent
+
+          # grab its response.
+          response = pipe.gets
+          pipe.close #this will be detected by the runner and it should call runner_end
+        end
+
+        run_the_runner(pipe,  [HydraExtension::RunnerListener::RunnerEndTest.new] )
+        Process.wait(parent)
+
+        # ensure runner_end was fired
+        assert File.exists?( alternate_target_file )
+      end
+    end
   end
 
   module RunnerTestHelper
@@ -140,9 +186,9 @@ class RunnerTest < Test::Unit::TestCase
       assert_equal "HYDRA", File.read(target_file)
     end
 
-    def run_the_runner(pipe)
+    def run_the_runner(pipe, listeners = [])
       pipe.identify_as_child
-      Hydra::Runner.new(:io => pipe)
+      Hydra::Runner.new( :io => pipe, :runner_listeners => listeners )
     end
   end
   include RunnerTestHelper

test/test_helper.rb

@@ -4,6 +4,7 @@ gem 'shoulda', '2.10.3'
 gem 'rspec', '2.0.0.beta.19'
 require 'shoulda'
 require 'tmpdir'
+require "stringio"
 
 $LOAD_PATH.unshift(File.join(File.dirname(__FILE__), '..', 'lib'))
 $LOAD_PATH.unshift(File.dirname(__FILE__))
@@ -56,6 +57,38 @@ class Test::Unit::TestCase
   def conflicting_test_file
     File.expand_path(File.join(File.dirname(__FILE__), 'fixtures', 'conflicting.rb'))
   end
+
+  def remote_dir_path
+    File.expand_path(File.join(File.dirname(__FILE__), '..', 'lib'))
+  end
+
+  def hydra_worker_init_file
+    File.expand_path(File.join(File.dirname(__FILE__), 'fixtures', 'hydra_worker_init.rb'))
+  end
+
+  def capture_stderr
+    # The output stream must be an IO-like object. In this case we capture it in
+    # an in-memory IO object so we can return the string value. You can assign any
+    # IO object here.
+    previous_stderr, $stderr = $stderr, StringIO.new
+    yield
+    $stderr.string
+  ensure
+    # Restore the previous value of stderr (typically equal to STDERR).
+    $stderr = previous_stderr
+  end
+
+  #this method allow us to wait for a file for a maximum number of time, so the
+  #test can pass in slower machines. This helps to speed up the tests
+  def assert_file_exists file, time_to_wait = 2
+    time_begin = Time.now
+
+    until Time.now - time_begin >= time_to_wait or File.exists?( file ) do
+      sleep 0.01
+    end
+
+    assert File.exists?( file )
+  end
 end
 
 module Hydra #:nodoc: