Merge pull request #4302 from mkanoor/bugzilla_1258648

Fixed Automate Method handling

lib/miq_automation_engine/engine/miq_ae_method.rb

@@ -86,30 +86,12 @@ def self.invoke_external(cmd, workspace, serialize_workspace = false)
 
       ENV['MIQ_TOKEN'] = ws.guid unless ws.nil?
 
-      rc = nil
-      final_stderr = nil
-      begin
-        require 'open4'
-        status = Open4.popen4(*cmd) do |pid, stdin, stdout, stderr|
-          stdin.close
-          final_stderr = stderr.each_line.map(&:strip)
-          stdout.each_line { |msg| $miq_ae_logger.info  "Method STDOUT: #{msg.strip}" }
-          final_stderr.each { |msg| $miq_ae_logger.error "Method STDERR: #{msg}" }
-        end
-
-        unless ws.nil?
-          ws.reload unless ws.nil?
-          ws.setters.each { |uri, value| workspace.varset(uri, value) } unless ws.setters.nil?
-          ws.delete
-        end
-        rc  = status.exitstatus
-        msg = "Method exited with rc=#{verbose_rc(rc)}"
-      rescue => err
-        $miq_ae_logger.error("Method exec failed because (#{err.class}:#{err.message})")
-        rc = MIQ_ABORT
-        msg = "Method execution failed"
+      rc, msg, final_stderr = run_method(*cmd)
+      if ws
+        ws.reload
+        ws.setters.each { |uri, value| workspace.varset(uri, value) } unless ws.setters.nil?
+        ws.delete
       end
-
       process_ruby_method_results(rc, msg, final_stderr)
     end
 
@@ -198,34 +180,14 @@ def self.verbose_rc(rc)
     end
 
     def self.run_ruby_method(body, preamble = nil)
-      begin
-        require 'open4'
-        ActiveRecord::Base.connection_pool.release_connection
-
-        rc = nil
-        final_stderr = nil
-        Bundler.with_clean_env do
-          status = Open4.popen4(Gem.ruby) do |pid, stdin, stdout, stderr|
-            stdin.puts(preamble.to_s)
-            stdin.puts(body)
-            stdin.puts(RUBY_METHOD_POSTSCRIPT) unless preamble.blank?
-            stdin.close
-
-            final_stderr = stderr.each_line.map(&:strip)
-            stdout.each_line { |msg| $miq_ae_logger.info  "Method STDOUT: #{msg.strip}" }
-            final_stderr.each { |msg| $miq_ae_logger.error "Method STDERR: #{msg}" }
-          end
-
-          rc = status.exitstatus
+      ActiveRecord::Base.connection_pool.release_connection
+      Bundler.with_clean_env do
+        run_method(Gem.ruby) do |stdin|
+          stdin.puts(preamble.to_s)
+          stdin.puts(body)
+          stdin.puts(RUBY_METHOD_POSTSCRIPT) unless preamble.blank?
         end
-
-        msg = "Method exited with rc=#{verbose_rc(rc)}"
-      rescue => err
-        $miq_ae_logger.error("Method exec failed because (#{err.class}:#{err.message})")
-        rc = MIQ_ABORT
-        msg = "Method execution failed"
       end
-      return rc, msg, final_stderr
     end
 
     def self.process_ruby_method_results(rc, msg, stderr)
@@ -289,5 +251,55 @@ def self.invoke_inline_ruby(aem, obj, inputs)
         end
       end
     end
+
+    def self.run_method(cmd)
+      require 'open4'
+      rc = nil
+      final_stderr = []
+      threads = []
+      method_pid = nil
+      begin
+        status = Open4.popen4(*cmd) do |pid, stdin, stdout, stderr|
+          method_pid = pid
+          yield stdin if block_given?
+          stdin.close
+          threads << Thread.new do
+            stdout.each_line { |msg| $miq_ae_logger.info "Method STDOUT: #{msg.strip}" }
+          end
+          threads << Thread.new do
+            stderr.each_line do |msg|
+              msg = msg.chomp
+              final_stderr << msg
+              $miq_ae_logger.error "Method STDERR: #{msg}"
+            end
+          end
+          threads.each(&:join)
+        end
+        rc  = status.exitstatus
+        msg = "Method exited with rc=#{verbose_rc(rc)}"
+        method_pid = nil
+        threads = []
+      rescue => err
+        $miq_ae_logger.error("Method exec failed because (#{err.class}:#{err.message})")
+        rc = MIQ_ABORT
+        msg = "Method execution failed"
+      ensure
+        cleanup(method_pid, threads)
+      end
+      return rc, msg, final_stderr.presence
+    end
+
+    def self.cleanup(method_pid, threads)
+      if method_pid
+        begin
+          $miq_ae_logger.error("Terminating non responsive method with pid #{method_pid.inspect}")
+          Process.kill("TERM", method_pid)
+          Process.wait(method_pid)
+        rescue Errno::ESRCH, RangeError => err
+          $miq_ae_logger.error("Error terminating #{method_pid.inspect} exception #{err}")
+        end
+      end
+      threads.each(&:exit)
+    end
   end
 end

spec/lib/miq_automation_engine/miq_ae_method_dispatch_spec.rb

@@ -0,0 +1,110 @@
+require "spec_helper"
+include AutomationSpecHelper
+
+describe "MiqAeMethodDispatch" do
+  before do
+    @method_name     = 'MY_METHOD'
+    @method_instance = 'MY_METHOD_INSTANCE'
+    @method_class    = 'MY_METHOD_CLASS'
+    @domain          = 'SPEC_DOMAIN'
+    @namespace       = 'NS1'
+    @root_class      = "TOP_OF_THE_WORLD"
+    @root_instance   = "EVEREST"
+    @automate_args   = {:namespace        => @namespace,
+                        :class_name       => @root_class,
+                        :instance_name    => @root_instance,
+                        :automate_message => 'create'}
+    MiqServer.stub(:my_zone).and_return('default')
+    @pidfile = File.join(Dir.mktmpdir, "rip_van_winkle.pid")
+    clear_domain
+  end
+
+  def clear_domain
+    MiqAeDomain.find_by_name(@domain).try(:destroy)
+  end
+
+  def rip_van_winkle_script
+    <<-'RUBY'
+      pidfile    = $evm.inputs['pidfile']
+      File.open(pidfile, 'w') { |file| file.write(Process.pid.to_s) }
+      10.times do
+          STDERR.puts "Hello from stderr channel"
+          STDOUT.puts "Hello from stdout channel"
+      end
+      # Intentional Sleep, this process gets terminated by the
+      # Automate Engine, since it thinks its unresponsive
+      # look at long running method spec in this file
+      sleep(60)
+    RUBY
+  end
+
+  def std_script
+    <<-'RUBY'
+      pidfile    = $evm.inputs['pidfile']
+      File.open(pidfile, 'w') { |file| file.write(Process.pid.to_s) }
+      $evm.root['method_pid'] = Process.pid
+      10.times do
+          STDERR.puts "Hello from stderr channel"
+          STDOUT.puts "Hello from stdout channel"
+      end
+      exit(MIQ_OK)
+    RUBY
+  end
+
+  def setup_model(method_script)
+    dom = FactoryGirl.create(:miq_ae_domain, :enabled => true, :name => @domain)
+    ns  = FactoryGirl.create(:miq_ae_namespace, :parent_id => dom.id, :name => @namespace)
+    @ns_fqname = ns.fqname
+    create_method_class(:namespace => @ns_fqname, :name => @method_class,
+                        :method_script => method_script)
+    create_root_class(:namespace => @ns_fqname, :name => @root_class)
+  end
+
+  def create_method_class(attrs = {})
+    params = {'pidfile' => {'aetype'        => 'attribute',
+                            'datatype'      => 'string',
+                            'default_value' => @pidfile}
+             }
+    method_script = attrs.delete(:method_script)
+    ae_fields = {'execute' => {:aetype => 'method', :datatype => 'string'}}
+    ae_instances = {@method_instance => {'execute' => {:value => @method_name}}}
+    ae_methods = {@method_name => {:scope => 'instance', :location => 'inline',
+                                   :data => method_script,
+                                   :language => 'ruby', 'params' => params}}
+
+    FactoryGirl.create(:miq_ae_class, :with_instances_and_methods,
+                       attrs.merge('ae_fields'    => ae_fields,
+                                   'ae_instances' => ae_instances,
+                                   'ae_methods'   => ae_methods))
+  end
+
+  def create_root_class(attrs = {})
+    ae_fields = {'rel1' => {:aetype => 'relationship', :datatype => 'string'}}
+    fqname = "/#{@domain}/#{@namespace}/#{@method_class}/#{@method_instance}"
+    ae_instances = {@root_instance => {'rel1' => {:value => fqname}}}
+    FactoryGirl.create(:miq_ae_class, :with_instances_and_methods,
+                       attrs.merge('ae_fields'    => ae_fields,
+                                   'ae_methods'   => {},
+                                   'ae_instances' => ae_instances))
+  end
+
+  it "long running method" do
+    File.delete(@pidfile) if File.exist?(@pidfile)
+    setup_model(rip_van_winkle_script)
+    # Set the timeout to 2 seconds so we can terminate
+    # unresponsive method
+    send_ae_request_via_queue(@automate_args, 2)
+    status, _msg, _ws = deliver_ae_request_from_queue
+    expect(status).to eql 'timeout'
+    pid = File.read(@pidfile).to_i
+    expect { Process.getpgid(pid) }.to raise_error(Errno::ESRCH)
+  end
+
+  it "run method that writes to stderr and stdout" do
+    setup_model(std_script)
+    send_ae_request_via_queue(@automate_args)
+    _, _, ws = deliver_ae_request_from_queue
+    pid = File.read(@pidfile).to_i
+    expect(ws.root['method_pid']).to eql pid
+  end
+end

spec/support/automation_spec_helper.rb

@@ -57,11 +57,13 @@ def default_ae_model_attributes(attrs = {})
       :instance_name => 'instance1')
   end
 
-  def send_ae_request_via_queue(args)
-    MiqQueue.put(:role        => 'automate',
-                 :class_name  => 'MiqAeEngine',
-                 :method_name => 'deliver',
-                 :args        => [args])
+  def send_ae_request_via_queue(args, timeout = nil)
+    queue_args = {:role        => 'automate',
+                  :class_name  => 'MiqAeEngine',
+                  :method_name => 'deliver',
+                  :args        => [args]}
+    queue_args.merge!(:msg_timeout => timeout) if timeout
+    MiqQueue.put(queue_args)
   end
 
   def deliver_ae_request_from_queue