Merge pull request #119 from perihelios/outputs-left-open

Fix for #118 (agent hangs if STDOUT/STDERR left open by hook)

Author: rohkat-aws

lib/instance_agent.rb

@@ -14,6 +14,7 @@
   require 'instance_agent/log'
   require 'instance_agent/platform'
   require 'instance_agent/platform/linux_util'
+  require 'instance_agent/platform/thread_joiner'
   require 'instance_agent/agent/plugin'
   require 'instance_agent/agent/base'
   require 'instance_agent/runner/master'

lib/instance_agent/platform/thread_joiner.rb

@@ -0,0 +1,13 @@
+module InstanceAgent
+  class ThreadJoiner
+    def initialize(timeout_sec)
+      @timeout_epoch = Time.now.to_i + timeout_sec
+    end
+
+    def joinOrFail(thread, &block)
+      if !thread.join([@timeout_epoch - Time.now.to_i, 0].max)
+        yield(thread) if block_given?
+      end
+    end
+  end
+end

lib/instance_agent/plugins/codedeploy/hook_executor.rb

@@ -47,6 +47,7 @@ class ScriptError < StandardError
         SCRIPT_TIMED_OUT_CODE = 3
         SCRIPT_FAILED_CODE = 4
         UNKNOWN_ERROR_CODE = 5
+        OUTPUTS_LEFT_OPEN_CODE = 6
         def initialize(error_code, script_name, log)
           @error_code = error_code
           @script_name = script_name
@@ -160,12 +161,21 @@ def execute_script(script, script_log_file)
             stdin.close
             stdout_thread = Thread.new{stdout.each_line { |line| log_script("[stdout]" + line.to_s, script_log_file)}}
             stderr_thread = Thread.new{stderr.each_line { |line| log_script("[stderr]" + line.to_s, script_log_file)}}
-            if !wait_thr.join(script.timeout)
+            thread_joiner = InstanceAgent::ThreadJoiner.new(script.timeout)
+            thread_joiner.joinOrFail(wait_thr) do
               Process.kill(signal, wait_thr.pid)
               raise Timeout::Error
             end
-            stdout_thread.join
-            stderr_thread.join
+            thread_joiner.joinOrFail(stdout_thread) do
+              script_error = "Script at specified location: #{script.location} failed to close STDOUT"
+              log :error, script_error
+              raise ScriptError.new(ScriptError::OUTPUTS_LEFT_OPEN_CODE, script.location, @script_log), script_error
+            end
+            thread_joiner.joinOrFail(stderr_thread) do
+              script_error = "Script at specified location: #{script.location} failed to close STDERR"
+              log :error, script_error
+              raise ScriptError.new(ScriptError::OUTPUTS_LEFT_OPEN_CODE, script.location, @script_log), script_error
+            end
             exit_status = wait_thr.value.exitstatus
           end
           if(exit_status != 0)

test/instance_agent/platform/thread_joiner_test.rb

@@ -0,0 +1,90 @@
+require 'test_helper'
+
+class ThreadJoinerTest < InstanceAgentTestCase
+  context 'ThreadJoiner' do
+    setup do
+      @timeout_sec = 30
+      @thread1 = mock('thread1')
+      @thread2 = mock('thread2')
+      @start_time = Time.now
+      Time.stubs(:now).returns(start_time)
+      @joiner = InstanceAgent::ThreadJoiner.new(@timeout_sec)
+    end
+
+    context 'with time left' do
+      setup do
+	@thread1.expects(:join).with(@timeout_sec).returns(1)
+	@thread2.expects(:join).with(@timeout_sec - 13).returns(1)
+      end
+
+      should 'join threads with proper timeout' do
+        @joiner.joinOrFail(@thread1)
+        Time.stubs(:now).returns(@start_time + 13)
+        @joiner.joinOrFail(@thread2)
+      end
+    end
+
+    context 'with no time left' do
+      setup do
+        @thread1.expects(:join).with(0)
+        @thread2.expects(:join).with(0)
+      end
+
+      should 'join threads for zero seconds' do
+        Time.stubs(:now).returns(@start_time + @timeout_sec)
+        @joiner.joinOrFail(@thread1)
+        Time.stubs(:now).returns(@start_time + @timeout_sec + 1)
+        @joiner.joinOrFail(@thread2)
+      end
+    end
+
+    context 'when a block is provided' do
+      context 'and thread does not complete' do
+        setup do
+          @thread1.expects(:join).returns(nil)
+        end
+
+        should 'call block' do
+          called = false
+          @joiner.joinOrFail(@thread1) do
+            called = true
+          end
+
+          assert_true called
+        end
+
+        should 'pass thread to block' do
+          thread = nil
+          @joiner.joinOrFail(@thread1) do | th |
+            thread = th
+          end
+
+          assert_equal @thread1, thread
+        end
+
+        should 'propagate exception back from block' do
+          assert_raise RuntimeError do
+            @joiner.joinOrFail(@thread1) do
+              raise 'thread did not complete'
+            end
+          end
+        end
+      end
+
+      context 'and thread completes' do
+        setup do
+          @thread1.expects(:join).returns(1)
+        end
+
+        should 'not call block' do
+          called = false
+          @joiner.joinOrFail(@thread1) do
+            called = true
+          end
+
+          assert_false called
+        end
+      end
+    end
+  end
+end

test/instance_agent/plugins/codedeploy/hook_executor_test.rb

@@ -173,7 +173,7 @@ def create_full_hook_executor
             File.stubs(:exist?).with(@script_location).returns(false)
           end
 
-          should "raise and exception" do
+          should "raise an exception" do
             assert_raised_with_message("Script does not exist at specified location: #{File.expand_path(@deployment_root_dir)}/deployment-archive/test", ScriptError)do
               @hook_executor.execute
             end
@@ -224,6 +224,9 @@ def create_full_hook_executor
               @value = mock
               @wait_thr.stubs(:value).returns(@value)
               @wait_thr.stubs(:join).returns(1000)
+              @thread_joiner = mock('thread_joiner')
+              @thread_joiner.stubs(:joinOrFail)
+              InstanceAgent::ThreadJoiner.stubs(:new).returns(@thread_joiner)
             end
 
             context 'scripts fail for unknown reason' do
@@ -248,9 +251,9 @@ def create_full_hook_executor
                 @app_spec =  { "version" => 0.0, "os" => "linux", "hooks" => {'ValidateService'=> [{"location"=>"test", "timeout"=>"30"}]}}
                 YAML.stubs(:load).returns(@app_spec)
                 @hook_executor = create_full_hook_executor
-                @wait_thr.stubs(:join).with(30).returns(nil)
+                @thread_joiner.expects(:joinOrFail).with(@wait_thr).yields
+                InstanceAgent::ThreadJoiner.expects(:new).with(30).returns(@thread_joiner)
                 @wait_thr.stubs(:pid).returns(1234)
-                mock_pipe = mock
               end
 
               context "with process group support" do
@@ -281,7 +284,47 @@ def create_full_hook_executor
                 end
               end
             end
-            
+
+            context "scripts fail to close outputs" do
+              setup do
+                timeout = 144
+                InstanceAgent::ThreadJoiner.expects(:new).with(timeout).returns(@thread_joiner)
+                @stdout_thread = mock('stdout_thread')
+                @stderr_thread = mock('stderr_thread')
+                Thread.stubs(:new).returns(@stdout_thread, @stderr_thread)
+                Open3.stubs(:popen3).with(@child_env, @script_location, :pgroup => true).yields([@mock_pipe,@mock_pipe,@mock_pipe,@wait_thr])
+                @app_spec = {"version" => 0.0, "os" => "linux", "hooks" => {'ValidateService'=>[{'location'=>'test', 'timeout'=>"#{timeout}"}]}}
+                YAML.stubs(:load).returns(@app_spec)
+                @hook_executor = create_full_hook_executor
+              end
+
+              context "STDOUT left open" do
+                setup do
+                  @thread_joiner.expects(:joinOrFail).with(@stdout_thread).yields
+                  InstanceAgent::Log.expects(:send).with(:error, "InstanceAgent::Plugins::CodeDeployPlugin::HookExecutor: Script at specified location: test failed to close STDOUT")
+                end
+
+                should "raise an exception" do
+                  assert_raised_with_message("Script at specified location: test failed to close STDOUT", ScriptError) do
+                    @hook_executor.execute
+                  end
+                end
+              end
+
+              context "STDERR left open" do
+                setup do
+                  @thread_joiner.expects(:joinOrFail).with(@stderr_thread).yields
+                  InstanceAgent::Log.expects(:send).with(:error, "InstanceAgent::Plugins::CodeDeployPlugin::HookExecutor: Script at specified location: test failed to close STDERR")
+                end
+
+                should "raise an exception" do
+                  assert_raised_with_message("Script at specified location: test failed to close STDERR", ScriptError) do
+                    @hook_executor.execute
+                  end
+                end
+              end
+            end
+
             context "Scripts run with a runas" do
               setup do
                 @app_spec =  { "version" => 0.0, "os" => "linux", "hooks" => {'ValidateService'=> [{"location"=>"test", "runas"=>"user"}]}}