Merge pull request #104 from iNecas/sub-plans-new

Fixes #9508 - use Dynflow sub-plans support for sub tasks and bulk actions

app/lib/actions/action_with_sub_plans.rb

@@ -0,0 +1,30 @@
+module Actions
+
+  class Actions::ActionWithSubPlans < Actions::EntryAction
+
+    middleware.use Actions::Middleware::KeepCurrentUser
+
+    include Dynflow::Action::WithSubPlans
+
+    def plan(*args)
+      raise NotImplementedError
+    end
+
+    def humanized_output
+      return unless counts_set?
+      _("%{total} tasks, %{success} success, %{failed} fail") %
+          { total:   output[:total_count],
+            success: output[:success_count],
+            failed:  output[:failed_count] }
+    end
+
+    def run_progress
+      if counts_set? && output[:total_count] > 0
+        (output[:success_count] + output[:failed_count]).to_f / output[:total_count]
+      else
+        0.1
+      end
+    end
+
+  end
+end

app/lib/actions/bulk_action.rb

@@ -1,14 +1,6 @@
 module Actions
 
-  class BulkAction < Actions::EntryAction
-
-    middleware.use Actions::Middleware::KeepCurrentUser
-
-    SubPlanFinished = Algebrick.type do
-      fields! :execution_plan_id => String,
-              :success           => type { variants TrueClass, FalseClass }
-    end
-
+  class BulkAction < Actions::ActionWithSubPlans
     # == Parameters:
     # actions_class::
     #   Class of action to trigger on targets
@@ -32,6 +24,10 @@ def humanized_name
       end
     end
 
+    def rescue_strategy
+      Dynflow::Action::Rescue::Skip
+    end
+
     def humanized_input
       a_sub_task = task.sub_tasks.first
       if a_sub_task
@@ -40,32 +36,6 @@ def humanized_input
       end
     end
 
-    def humanized_output
-      return unless counts_set?
-      _("%{total} tasks, %{success} success, %{failed} fail") %
-          { total:   output[:total_count],
-            success: output[:success_count],
-            failed:  output[:failed_count] }
-    end
-
-    def run(event = nil)
-      case(event)
-      when nil
-        if output[:total_count]
-          resume
-        else
-          initiate_sub_plans
-        end
-      when SubPlanFinished
-        mark_as_done(event.execution_plan_id, event.success)
-        if done?
-          check_for_errors!
-        else
-          suspend
-        end
-      end
-    end
-
     # @api override when the logic for the initiation of the subtasks
     #      is different from the default one
     def create_sub_plans
@@ -74,85 +44,10 @@ def create_sub_plans
       targets = target_class.where(:id => input[:target_ids])
 
       targets.map do |target|
-        ForemanTasks.trigger(action_class, target, *input[:args])
+        trigger(action_class, target, *input[:args])
       end
     end
 
-    def initiate_sub_plans
-      output.update(total_count: 0,
-                    failed_count: 0,
-                    success_count: 0)
-
-      planned, failed = create_sub_plans.partition(&:planned?)
-
-      sub_plan_ids = ((planned + failed).map(&:execution_plan_id))
-      set_parent_task_id(sub_plan_ids)
-
-      output[:total_count] = sub_plan_ids.size
-      output[:failed_count] = failed.size
-
-      if planned.any?
-        wait_for_sub_plans(planned)
-      else
-        check_for_errors!
-      end
-    end
-
-    def resume
-      if task.sub_tasks.active.any?
-        fail _("Some sub tasks are still not finished")
-      end
-    end
-
-    def rescue_strategy
-      Dynflow::Action::Rescue::Skip
-    end
-
-    def wait_for_sub_plans(plans)
-      suspend do |suspended_action|
-        plans.each do |plan|
-          plan.finished.do_then do |value|
-            suspended_action << SubPlanFinished[plan.execution_plan_id,
-                                                value.result == :success]
-          end
-        end
-      end
-    end
-
-    def mark_as_done(plan_id, success)
-      if success
-        output[:success_count] += 1
-      else
-        output[:failed_count] += 1
-      end
-    end
-
-    def done?
-      if counts_set?
-        output[:total_count] - output[:success_count] - output[:failed_count] <= 0
-      else
-        false
-      end
-    end
-
-    def run_progress
-      if counts_set?
-        (output[:success_count] + output[:failed_count]).to_f / output[:total_count]
-      else
-        0.1
-      end
-    end
-
-    def counts_set?
-      output[:total_count] && output[:success_count] && output[:failed_count]
-    end
-
-    def set_parent_task_id(sub_plan_ids)
-      ForemanTasks::Task::DynflowTask.
-          where(external_id: sub_plan_ids).
-          update_all(parent_task_id: task.id)
-    end
-
     def check_targets!(targets)
       if targets.empty?
         fail _("Empty bulk action")
@@ -162,9 +57,5 @@ def check_targets!(targets)
       end
     end
 
-    def check_for_errors!
-      fail _("A sub task failed") if output[:failed_count] > 0
-    end
-
   end
 end

app/models/foreman_tasks/lock.rb

@@ -57,7 +57,8 @@ def available?
 
     # returns a scope of the locks colliding with this one
     def colliding_locks
-      colliding_locks_scope = Lock.active.where(Lock.arel_table[:task_id].not_eq(task_id))
+      task_ids = task.self_and_parents.map(&:id)
+      colliding_locks_scope = Lock.active.where(Lock.arel_table[:task_id].not_in(task_ids))
       colliding_locks_scope = colliding_locks_scope.where(name:          name,
                                                           resource_id:   resource_id,
                                                           resource_type: resource_type)

app/models/foreman_tasks/task.rb

@@ -77,6 +77,14 @@ def paused?
       self.state == 'paused'
     end
 
+    def self_and_parents
+      [self].tap do |ret|
+        if parent_task
+          ret.concat(parent_task.self_and_parents)
+        end
+      end
+    end
+
     def self.search_by_generic_resource(key, operator, value)
       key =  "resource_type" if key.blank?
       key_name = self.connection.quote_column_name(key.sub(/^.*\./,''))

app/models/foreman_tasks/task/dynflow_task.rb

@@ -6,12 +6,17 @@ class Task::DynflowTask < ForemanTasks::Task
     scope :for_action, ->(action_class) { where(label: action_class.name) }
 
     def update_from_dynflow(data)
-      self.external_id = data[:id]
-      self.started_at  = data[:started_at]
-      self.ended_at    = data[:ended_at]
-      self.state       = data[:state].to_s
-      self.result      = data[:result].to_s
-      self.label       ||= main_action.class.name
+      self.external_id    = data[:id]
+      self.started_at     = data[:started_at]
+      self.ended_at       = data[:ended_at]
+      self.state          = data[:state].to_s
+      self.result         = data[:result].to_s
+      self.parent_task_id ||= begin
+                                if main_action.caller_execution_plan_id
+                                  DynflowTask.find_by_external_id!(main_action.caller_execution_plan_id).id
+                                end
+                              end
+      self.label          ||= main_action.class.name
       self.save!
     end
 

foreman-tasks.gemspec

@@ -25,7 +25,7 @@ DESC
   end
   s.test_files = Dir["test/**/*"]
 
-  s.add_dependency "dynflow", '>= 0.7.2'
+  s.add_dependency "dynflow", '>= 0.7.7'
   s.add_dependency "sequel" # for Dynflow process persistence
   s.add_dependency "sinatra" # for Dynflow web console
   s.add_dependency "daemons" # for running remote executor

test/unit/actions/action_with_sub_plans_test.rb

@@ -0,0 +1,58 @@
+require "foreman_tasks_test_helper"
+
+module ForemanTasks
+  class ActionWithSubPlansTest <  ActiveSupport::TestCase
+    self.use_transactional_fixtures = false
+
+    before do
+      User.current = User.where(:login => 'apiadmin').first
+    end
+
+    # to be able to use the locking
+    class ::User < User.parent
+      include ForemanTasks::Concerns::ActionSubject
+    end
+
+    class ParentAction < Actions::ActionWithSubPlans
+      def plan(user)
+        action_subject(user)
+        plan_self(user_id: user.id)
+      end
+
+      def create_sub_plans
+        user = User.find(input[:user_id])
+        trigger(ChildAction, user)
+      end
+    end
+
+    class ChildAction < Actions::EntryAction
+      def plan(user)
+        action_subject(user)
+        plan_self(user_id: user.id)
+      end
+      def run
+      end
+    end
+
+    describe Actions::ActionWithSubPlans do
+      let(:task) do
+        user = FactoryGirl.create(:user)
+        triggered = ForemanTasks.trigger(ParentAction, user)
+        raise triggered.error if triggered.respond_to?(:error)
+        triggered.finished.wait(2)
+        ForemanTasks::Task.find_by_external_id(triggered.id)
+      end
+
+      specify "the sub-plan stores the information about its parent" do
+        task.sub_tasks.size.must_equal 1
+        task.sub_tasks.first.label.must_equal ChildAction.name
+      end
+
+      specify "the locks of the sub-plan don't colide with the locks of its parent" do
+        child_task = task.sub_tasks.first
+        assert(child_task.locks.any? { |lock| lock.name == 'write' }, "it's locks don't conflict with parent's")
+      end
+    end
+
+  end
+end