proxy_action.rb in foreman-tasks-0.15.1

- old
+ new

@@ -24,12 +24,19 @@
 
     class ProxyActionStopped; end
 
     def plan(proxy, klass, options)
       options[:connection_options] ||= {}
-      default_connection_options.each { |key, value| options[:connection_options][key] ||= value }
-      plan_self(options.merge(:proxy_url => proxy.url, :proxy_action_name => klass.to_s))
+      default_connection_options.each do |key, value|
+        options[:connection_options][key] = value unless options[:connection_options].key?(key)
+      end
+      plan_self(options.merge(:proxy_url => proxy.url, :proxy_action_name => klass.to_s, :proxy_version => proxy_version(proxy)))
+      # Just saving the RemoteTask is enough when using batch triggering
+      # It will be picked up by the ProxyBatchTriggering middleware
+      if input[:use_batch_triggering] && with_batch_triggering?(input[:proxy_version])
+        prepare_remote_task.save!
+      end
     end
 
     def run(event = nil)
       with_connection_error_handling(event) do |event|
         case event
@@ -62,24 +69,26 @@
       @remote_task ||= ForemanTasks::RemoteTask.find_by(:execution_plan_id => execution_plan_id, :step_id => run_step_id)
     end
 
     def trigger_proxy_task
       suspend do |_suspended_action|
-        response = proxy.trigger_task(proxy_action_name,
-                                      input.merge(:callback => { :task_id => task.id,
-                                                                 :step_id => run_step_id }))
-        ::ForemanTasks::RemoteTask.new(:remote_task_id => response['task_id'], :execution_plan_id => execution_plan_id,
-                                       :state => 'triggered', :proxy_url => input[:proxy_url], :step_id => run_step_id).save!
-        output[:proxy_task_id] = response['task_id']
+        remote_task = prepare_remote_task
+        remote_task.trigger(proxy_action_name, proxy_input)
+        output[:proxy_task_id] = remote_task.remote_task_id
       end
     end
 
+    def proxy_input(task_id = task.id)
+      input.merge(:callback => { :task_id => task_id,
+                                 :step_id => run_step_id })
+    end
+
     def check_task_status
-      response = proxy.status_of_task(output[:proxy_task_id])
+      response = proxy.status_of_task(proxy_task_id)
       if %w[stopped paused].include? response['state']
         if response['result'] == 'error'
-          raise ::Foreman::Exception, _('The smart proxy task %s failed.') % output[:proxy_task_id]
+          raise ::Foreman::Exception, _('The smart proxy task %s failed.') % proxy_task_id
         else
           on_data(response['actions'].find { |block_action| block_action['class'] == proxy_action_name }['output'])
         end
       else
         suspend
@@ -88,18 +97,18 @@
 
     def cancel_proxy_task
       if output[:cancel_sent]
         error! ForemanTasks::Task::TaskCancelledException.new(_('Cancel enforced: the task might be still running on the proxy'))
       else
-        proxy.cancel_task(output[:proxy_task_id])
+        proxy.cancel_task(proxy_task_id)
         output[:cancel_sent] = true
         suspend
       end
     end
 
     def abort_proxy_task
-      proxy.cancel_task(output[:proxy_task_id])
+      proxy.cancel_task(proxy_task_id)
       error! ForemanTasks::Task::TaskCancelledException.new(_('Task aborted: the task might be still running on the proxy'))
     end
 
     def on_resume
       # TODO: add logic to load the data from the external action
@@ -130,19 +139,24 @@
     # @override String name of an action to be triggered on server
     def proxy_action_name
       input[:proxy_action_name]
     end
 
+    # @override String name of a operation to be triggered on server
+    def proxy_operation_name
+      input[:proxy_operation_name]
+    end
+
     def proxy
       ProxyAPI::ForemanDynflow::DynflowProxy.new(:url => input[:proxy_url])
     end
 
     def proxy_output(live = false)
       if output.key?(:proxy_output) || state == :error
         output.fetch(:proxy_output, {})
-      elsif live && output[:proxy_task_id]
-        proxy_data = proxy.status_of_task(output[:proxy_task_id])['actions'].detect { |action| action['class'] == proxy_action_name }
+      elsif live && proxy_task_id
+        proxy_data = proxy.status_of_task(proxy_task_id)['actions'].detect { |action| action['class'] == proxy_action_name }
         proxy_data.fetch('output', {})
       else
         {}
       end
     end
@@ -172,19 +186,30 @@
 
     def default_connection_options
       # Fails if the plan is not finished within 60 seconds from the first task trigger attempt on the smart proxy
       # If the triggering fails, it retries 3 more times with 15 second delays
       { :retry_interval => Setting['foreman_tasks_proxy_action_retry_interval'] || 15,
-        :retry_count    => Setting['foreman_tasks_proxy_action_retry_count'] || 4 }
+        :retry_count    => Setting['foreman_tasks_proxy_action_retry_count'] || 4,
+        :proxy_batch_triggering => Setting['foreman_tasks_proxy_batch_trigger'] || false }
     end
 
+    def with_batch_triggering?(proxy_version)
+      (proxy_version[:major] == 1 && proxy_version[:minor] > 20) || proxy_version[:major] > 1 &&
+        input.fetch(:connection_options, {}).fetch(:proxy_batch_triggering, false)
+    end
+
     def clean_remote_task(*_args)
       remote_task.destroy! if remote_task
     end
 
     private
 
+    def proxy_version(proxy)
+      match = proxy.statuses[:version].version['version'].match(/(\d+)\.(\d+)\.(\d+)/)
+      { :major => match[1].to_i, :minor => match[2].to_i, :patch => match[3].to_i }
+    end
+
     def failed_proxy_tasks
       metadata[:failed_proxy_tasks] ||= []
     end
 
     def with_connection_error_handling(event = nil)
@@ -196,11 +221,11 @@
         handle_connection_exception(e, event)
       end
     end
 
     def format_exception(exception)
-      { :proxy_task_id => output[:proxy_task_id],
+      { :proxy_task_id => proxy_task_id,
         :exception_class => exception.class.name,
         :exception_message => exception.message,
         :timestamp => Time.now.to_f }
     end
 
@@ -215,8 +240,19 @@
                             event
         end
       else
         raise exception
       end
+    end
+
+    def prepare_remote_task
+      ::ForemanTasks::RemoteTask.new(:execution_plan_id => execution_plan_id,
+                                     :proxy_url => input[:proxy_url],
+                                     :step_id => run_step_id,
+                                     :operation => proxy_operation_name)
+    end
+
+    def proxy_task_id
+      output[:proxy_task_id] ||= remote_task.try(:remote_task_id)
     end
   end
 end