From 3d4d8e6fd3b3952f1c470a6dda761d93e42ca8a4 Mon Sep 17 00:00:00 2001
From: Adam Ruzicka <aruzicka@redhat.com>
Date: Fri, 21 Jul 2023 13:00:50 +0000
Subject: [PATCH] Introduce v2 for sub-plans

The previous implementation of sub plans API comprised of several
modules which could be layered on top of each other. This made the code
somewhat hard to follow, not all of the modules worked in all scenarios
as some relied on an executor being a single process.

The v2 flattens all the modules into a single one and drastically
reduces the amount of possible combinations. Sub plans are spawned in
batches and then the parent polls until the sub plans finish.

It also includes support for limitting the number of sub-plans that may
be running at a given time, without relying on the throttle limitter and
thus should be safe in multi-process environments. Under the hood it
dynamically alters the batch size to honor the concurrency limit.
---
 examples/sub_plans_v2.rb                |  75 ++++++++
 lib/dynflow/action.rb                   |   1 +
 lib/dynflow/action/v2.rb                |   9 +
 lib/dynflow/action/v2/with_sub_plans.rb | 219 ++++++++++++++++++++++++
 test/v2_sub_plans_test.rb               | 189 ++++++++++++++++++++
 5 files changed, 493 insertions(+)
 create mode 100755 examples/sub_plans_v2.rb
 create mode 100644 lib/dynflow/action/v2.rb
 create mode 100644 lib/dynflow/action/v2/with_sub_plans.rb
 create mode 100644 test/v2_sub_plans_test.rb

diff --git a/examples/sub_plans_v2.rb b/examples/sub_plans_v2.rb
new file mode 100755
index 00000000..7fd90fd3
--- /dev/null
+++ b/examples/sub_plans_v2.rb
@@ -0,0 +1,75 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+example_description = <<DESC
+  Sub Plans Example
+  ===================
+
+  This example shows, how to trigger the execution plans from within a
+  run method of some action and waing for them to finish.
+
+  This is useful, when doing bulk actions, where having one big
+  execution plan would not be effective, or in case all the data are
+  not available by the time of original action planning.
+
+DESC
+
+require_relative 'example_helper'
+require_relative 'orchestrate_evented'
+
+COUNT = (ARGV[0] || 25).to_i
+
+class Foo < Dynflow::Action
+  def plan
+    plan_self
+  end
+
+  def run(event = nil)
+    case event
+    when nil
+      rng = Random.new
+      plan_event(:ping, rng.rand(25) + 1)
+      suspend
+    when :ping
+      # Finish
+    end
+  end
+end
+
+class SubPlansExample < Dynflow::Action
+  include Dynflow::Action::V2::WithSubPlans
+
+  def initiate
+    limit_concurrency_level! 3
+    super
+  end
+
+  def create_sub_plans
+    current_batch.map { |i| trigger(Foo) }
+  end
+
+  def batch_size
+    15
+  end
+
+  def batch(from, size)
+    COUNT.times.drop(from).take(size)
+  end
+
+  def total_count
+    COUNT
+  end
+end
+
+if $0 == __FILE__
+  ExampleHelper.world.action_logger.level = Logger::DEBUG
+  ExampleHelper.world
+  t1 = ExampleHelper.world.trigger(SubPlansExample)
+  puts example_description
+  puts <<-MSG.gsub(/^.*\|/, '')
+    |  Execution plans #{t1.id} with sub plans triggered
+    |  You can see the details at
+    |    #{ExampleHelper::DYNFLOW_URL}/#{t1.id}
+  MSG
+
+  ExampleHelper.run_web_console
+end
diff --git a/lib/dynflow/action.rb b/lib/dynflow/action.rb
index 8e118476..3774db6f 100644
--- a/lib/dynflow/action.rb
+++ b/lib/dynflow/action.rb
@@ -26,6 +26,7 @@ class Action < Serializable
     require 'dynflow/action/with_sub_plans'
     require 'dynflow/action/with_bulk_sub_plans'
     require 'dynflow/action/with_polling_sub_plans'
+    require 'dynflow/action/v2'
 
     def self.all_children
       children.values.inject(children.values) do |children, child|
diff --git a/lib/dynflow/action/v2.rb b/lib/dynflow/action/v2.rb
new file mode 100644
index 00000000..b2dc3ce2
--- /dev/null
+++ b/lib/dynflow/action/v2.rb
@@ -0,0 +1,9 @@
+# frozen_string_literal: true
+
+module Dynflow
+  class Action
+    module V2
+      require 'dynflow/action/v2/with_sub_plans'
+    end
+  end
+end
diff --git a/lib/dynflow/action/v2/with_sub_plans.rb b/lib/dynflow/action/v2/with_sub_plans.rb
new file mode 100644
index 00000000..1c818402
--- /dev/null
+++ b/lib/dynflow/action/v2/with_sub_plans.rb
@@ -0,0 +1,219 @@
+# frozen_string_literal: true
+
+module Dynflow::Action::V2
+  module WithSubPlans
+    include Dynflow::Action::Cancellable
+
+    DEFAULT_BATCH_SIZE = 100
+    DEFAULT_POLLING_INTERVAL = 15
+    Ping = Algebrick.atom
+
+    class SubtaskFailedException < RuntimeError
+      def backtrace
+        []
+      end
+    end
+
+    # Methods to be overridden
+    def create_sub_plans
+      raise NotImplementedError
+    end
+
+    # Should return the expected total count of tasks
+    def total_count
+      raise NotImplementedError
+    end
+
+    def batch_size
+      DEFAULT_BATCH_SIZE
+    end
+
+    # Should return a slice of size items starting from item with index from
+    def batch(from, size)
+      raise NotImplementedError
+    end
+
+    # Polling
+    def polling_interval
+      DEFAULT_POLLING_INTERVAL
+    end
+
+    # Callbacks
+    def on_finish
+    end
+
+    def on_planning_finished
+    end
+
+    def run(event = nil)
+      case event
+      when nil
+        if output[:total_count]
+          resume
+        else
+          initiate
+        end
+      when Ping
+        tick
+      when ::Dynflow::Action::Cancellable::Cancel
+        cancel!
+      when ::Dynflow::Action::Cancellable::Abort
+        abort!
+      end
+      try_to_finish || suspend_and_ping
+    end
+
+    def initiate
+      output[:planned_count] = 0
+      output[:cancelled_count] = 0
+      output[:total_count] = total_count
+      spawn_plans
+    end
+
+    def resume
+      if sub_plans.all? { |sub_plan| sub_plan.error_in_plan? }
+        output[:resumed_count] ||= 0
+        output[:resumed_count] += output[:failed_count]
+        # We're starting over and need to reset the counts
+        %w(total failed pending success).each { |key| output.delete("#{key}_count".to_sym) }
+        initiate
+      else
+        tick
+      end
+    end
+
+    def tick
+      recalculate_counts
+      spawn_plans if can_spawn_next_batch?
+    end
+
+    def suspend_and_ping
+      delay = (concurrency_limit.nil? || concurrency_limit_capacity > 0) && can_spawn_next_batch? ? nil : polling_interval
+      plan_event(Ping, delay)
+      suspend
+    end
+
+    def spawn_plans
+      sub_plans = create_sub_plans
+      sub_plans = Array[sub_plans] unless sub_plans.is_a? Array
+      increase_counts(sub_plans.count, 0)
+      on_planning_finished unless can_spawn_next_batch?
+    end
+
+    def increase_counts(planned, failed)
+      output[:planned_count] += planned + failed
+      output[:failed_count]  = output.fetch(:failed_count, 0) + failed
+      output[:pending_count] = output.fetch(:pending_count, 0) + planned
+      output[:success_count] ||= 0
+    end
+
+    def try_to_finish
+      return false unless done?
+
+      check_for_errors!
+      on_finish
+      true
+    end
+
+    def done?
+      return false if can_spawn_next_batch? || !counts_set?
+
+      total_count - output[:success_count] - output[:failed_count] - output[:cancelled_count] <= 0
+    end
+
+    def run_progress
+      return 0.1 unless counts_set? && total_count > 0
+
+      sum = output.values_at(:success_count, :cancelled_count, :failed_count).reduce(:+)
+      sum.to_f / total_count
+    end
+
+    def recalculate_counts
+      total   = total_count
+      failed  = sub_plans_count('state' => %w(paused stopped), 'result' => 'error')
+      success = sub_plans_count('state' => 'stopped', 'result' => 'success')
+      output.update(:pending_count => total - failed - success,
+                    :failed_count  => failed - output.fetch(:resumed_count, 0),
+                    :success_count => success)
+    end
+
+    def counts_set?
+      output[:total_count] && output[:success_count] && output[:failed_count] && output[:pending_count]
+    end
+
+    def check_for_errors!
+      raise SubtaskFailedException.new("A sub task failed") if output[:failed_count] > 0
+    end
+
+    # Helper for creating sub plans
+    def trigger(action_class, *args)
+      world.trigger { world.plan_with_options(action_class: action_class, args: args, caller_action: self) }
+    end
+
+    # Concurrency limitting
+    def limit_concurrency_level!(level)
+      input[:dynflow] ||= {}
+      input[:dynflow][:concurrency_limit] = level
+    end
+
+    def concurrency_limit
+      input[:dynflow] ||= {}
+      input[:dynflow][:concurrency_limit]
+    end
+
+    def concurrency_limit_capacity
+      if limit = concurrency_limit
+        return limit unless counts_set?
+        capacity = limit - (output[:planned_count] - (output[:success_count] + output[:failed_count]))
+        [0, capacity].max
+      end
+    end
+
+    # Cancellation handling
+    def cancel!(force = false)
+      # Count the not-yet-planned tasks as cancelled
+      output[:cancelled_count] = total_count - output[:planned_count]
+      # Pass the cancel event to running sub plans if they can be cancelled
+      sub_plans(:state => 'running').each { |sub_plan| sub_plan.cancel(force) if sub_plan.cancellable? }
+      suspend
+    end
+
+    def abort!
+      cancel! true
+    end
+
+    # Batching
+    # Returns the items in the current batch
+    def current_batch
+      start_position = output[:planned_count]
+      size = batch_size
+      size = concurrency_limit_capacity if concurrency_limit
+      size = start_position + size > total_count ? total_count - start_position : size
+      batch(start_position, size)
+    end
+
+    def can_spawn_next_batch?
+      remaining_count > 0
+    end
+
+    def remaining_count
+      total_count - output[:cancelled_count] - output[:planned_count]
+    end
+
+    private
+
+    # Sub-plan lookup
+    def sub_plan_filter
+      { 'caller_execution_plan_id' => execution_plan_id,
+        'caller_action_id' => self.id }
+    end
+
+    def sub_plans(filter = {})
+      world.persistence.find_execution_plans(filters: sub_plan_filter.merge(filter))
+    end
+
+    def sub_plans_count(filter = {})
+      world.persistence.find_execution_plan_counts(filters: sub_plan_filter.merge(filter))
+    end
+  end
+end
diff --git a/test/v2_sub_plans_test.rb b/test/v2_sub_plans_test.rb
new file mode 100644
index 00000000..b6c644e7
--- /dev/null
+++ b/test/v2_sub_plans_test.rb
@@ -0,0 +1,189 @@
+# frozen_string_literal: true
+require_relative 'test_helper'
+require 'mocha/minitest'
+
+module Dynflow
+  module V2SubPlansTest
+    describe 'V2 sub-plans' do
+      include PlanAssertions
+      include Dynflow::Testing::Assertions
+      include Dynflow::Testing::Factories
+      include TestHelpers
+
+      let(:world) { WorldFactory.create_world }
+
+      class ChildAction < ::Dynflow::Action
+        def run
+        end
+      end
+
+      class ParentAction < ::Dynflow::Action
+        include Dynflow::Action::V2::WithSubPlans
+
+        def plan(count, concurrency_level = nil)
+          limit_concurrency_level!(concurrency_level) if concurrency_level
+          plan_self :count => count
+        end
+
+        def create_sub_plans
+          output[:batch_count] ||= 0
+          output[:batch_count] += 1
+          current_batch.map { |i| trigger(ChildAction) }
+        end
+
+        def batch(from, size)
+          (1..total_count).to_a.slice(from, size)
+        end
+
+        def batch_size
+          5
+        end
+
+        def total_count
+          input[:count]
+        end
+      end
+
+      describe 'normal operation' do
+        it 'spawns all sub-plans in one go with high-enough batch size and polls until they are done' do
+          action = create_and_plan_action ParentAction, 3
+          action.world.expects(:trigger).times(3)
+          action = run_action action
+          _(action.output['total_count']).must_equal 3
+          _(action.output['planned_count']).must_equal 3
+          _(action.output['pending_count']).must_equal 3
+
+          ping = action.world.clock.pending_pings.first
+          _(ping.what.value.event).must_equal Dynflow::Action::V2::WithSubPlans::Ping
+          _(ping.when).must_be_within_delta(Time.now + action.polling_interval, 1)
+          persistence = mock()
+          persistence.stubs(:find_execution_plan_counts).returns(0)
+          action.world.stubs(:persistence).returns(persistence)
+
+          action.world.clock.progress
+          action.world.executor.progress
+          ping = action.world.clock.pending_pings.first
+          _(ping.what.value.event).must_equal Dynflow::Action::V2::WithSubPlans::Ping
+          _(ping.when).must_be_within_delta(Time.now + action.polling_interval * 2, 1)
+
+          persistence = mock()
+          persistence.stubs(:find_execution_plan_counts).returns(0).then.returns(3)
+          action.world.stubs(:persistence).returns(persistence)
+          action.world.clock.progress
+          action.world.executor.progress
+
+          _(action.state).must_equal :success
+          _(action.done?).must_equal true
+        end
+
+        it 'spawns sub-plans in multiple batches and polls until they are done' do
+          action = create_and_plan_action ParentAction, 7
+          action.world.expects(:trigger).times(5)
+          action = run_action action
+          _(action.output['total_count']).must_equal 7
+          _(action.output['planned_count']).must_equal 5
+          _(action.output['pending_count']).must_equal 5
+
+          _(action.world.clock.pending_pings).must_be :empty?
+          _, _, event, * = action.world.executor.events_to_process.first
+          _(event).must_equal Dynflow::Action::V2::WithSubPlans::Ping
+          persistence = mock()
+          # Simulate 3 finished
+          persistence.stubs(:find_execution_plan_counts).returns(0).then.returns(3)
+          action.world.stubs(:persistence).returns(persistence)
+
+          action.world.expects(:trigger).times(2)
+          action.world.executor.progress
+
+          ping = action.world.clock.pending_pings.first
+          _(ping.what.value.event).must_equal Dynflow::Action::V2::WithSubPlans::Ping
+          _(ping.when).must_be_within_delta(Time.now + action.polling_interval, 1)
+
+          persistence.stubs(:find_execution_plan_counts).returns(0).then.returns(7)
+          action.world.stubs(:persistence).returns(persistence)
+          action.world.clock.progress
+          action.world.executor.progress
+
+          _(action.state).must_equal :success
+          _(action.done?).must_equal true
+        end
+      end
+
+      describe 'with concurrency control' do
+        include Dynflow::Testing
+
+        it 'allows storage and retrieval' do
+          action = create_and_plan_action ParentAction, 0
+          action = run_action action
+          _(action.concurrency_limit).must_be_nil
+          _(action.concurrency_limit_capacity).must_be_nil
+
+          action = create_and_plan_action ParentAction, 0, 1
+          action = run_action action
+
+          _(action.input['dynflow']['concurrency_limit']).must_equal 1
+          _(action.concurrency_limit).must_equal 1
+          _(action.concurrency_limit_capacity).must_equal 1
+        end
+
+        it 'reduces the batch size to fit within the concurrency limit' do
+          action = create_and_plan_action ParentAction, 5, 2
+
+          # Plan first 2 sub-plans
+          action.world.expects(:trigger).times(2)
+
+          action = run_action action
+          _(action.output['total_count']).must_equal 5
+          _(action.output['planned_count']).must_equal 2
+          _(action.output['pending_count']).must_equal 2
+          _(action.concurrency_limit_capacity).must_equal 0
+          _(action.output['batch_count']).must_equal 1
+
+          ping = action.world.clock.pending_pings.first
+          _(ping.what.value.event).must_equal Dynflow::Action::V2::WithSubPlans::Ping
+          _(ping.when).must_be_within_delta(Time.now + action.polling_interval, 1)
+          persistence = mock()
+          # Simulate 1 sub-plan finished
+          persistence.stubs(:find_execution_plan_counts).returns(0).then.returns(1)
+          action.world.stubs(:persistence).returns(persistence)
+
+          # Only 1 sub-plans fits into the capacity
+          action.world.expects(:trigger).times(1)
+          action.world.clock.progress
+          action.world.executor.progress
+
+          _(action.output['planned_count']).must_equal 3
+
+          persistence = mock()
+          persistence.stubs(:find_execution_plan_counts).returns(0).then.returns(2)
+          action.world.stubs(:persistence).returns(persistence)
+          action.world.expects(:trigger).times(1)
+          action.world.clock.progress
+          action.world.executor.progress
+
+          _(action.output['planned_count']).must_equal 4
+
+          persistence = mock()
+          persistence.stubs(:find_execution_plan_counts).returns(0).then.returns(4)
+          action.world.stubs(:persistence).returns(persistence)
+          action.world.expects(:trigger).times(1)
+          action.world.clock.progress
+          action.world.executor.progress
+
+          _(action.output['planned_count']).must_equal 5
+          _(action.concurrency_limit_capacity).must_equal 1
+
+          persistence = mock()
+          persistence.stubs(:find_execution_plan_counts).returns(0).then.returns(5)
+          action.world.stubs(:persistence).returns(persistence)
+          action.world.expects(:trigger).never
+          action.world.clock.progress
+          action.world.executor.progress
+          _(action.state).must_equal :success
+          _(action.done?).must_equal true
+          _(action.concurrency_limit_capacity).must_equal 2
+        end
+      end
+    end
+  end
+end