Add execution plan chaining

This commit enables execution plans to be chained. Assuming there is an
execution plan EP1, another execution plan EP2 can be chained onto EP1.  When
chained, EP2 will stay in scheduled state until EP1 goes to stopped state. An
execution plan can be chained onto multiple prerequisite execution plans, in
which case it will be run once all the prerequisite execution plans are stopped
and have not failed. If the prerequisite execution plan ends with
stopped-error, the chained execution plan(s) will fail. If the prerequisite
execution plan is halted, the chained execution plan(s) will be run.

It builds on mechanisms which were already present. When an execution
plan is chained, it behaves in the same way as if it was scheduled for
future execution. A record is created in dynflow_delayed_table and once
the conditions for it to execute are right, it is dispatched by the
delayed executor. Because of this, there might be small delay between
when the prerequisites finishs and the chained plan is started.

Author: adamruzicka

examples/execution_plan_chaining.rb

@@ -0,0 +1,56 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+require_relative 'example_helper'
+
+class DelayedAction < Dynflow::Action
+  def plan(should_fail = false)
+    plan_self :should_fail => should_fail
+  end
+
+  def run
+    sleep 5
+    raise "Controlled failure" if input[:should_fail]
+  end
+
+  def rescue_strategy
+    Dynflow::Action::Rescue::Fail
+  end
+end
+
+if $PROGRAM_NAME == __FILE__
+  world = ExampleHelper.create_world do |config|
+    config.auto_rescue = true
+  end
+  world.action_logger.level = 1
+  world.logger.level = 0
+
+  plan1 = world.trigger(DelayedAction)
+  plan2 = world.chain(plan1.execution_plan_id, DelayedAction)
+  plan3 = world.chain(plan2.execution_plan_id, DelayedAction)
+  plan4 = world.chain(plan2.execution_plan_id, DelayedAction)
+
+  plan5 = world.trigger(DelayedAction, true)
+  plan6 = world.chain(plan5.execution_plan_id, DelayedAction)
+
+  puts <<-MSG.gsub(/^.*\|/, '')
+    |
+    |  Execution Plan Chaining example
+    |  ========================
+    |
+    |  This example shows the execution plan chaining functionality of Dynflow, which allows execution plans to wait until another execution plan finishes.
+    |
+    |  Execution plans:
+    |    #{plan1.id} runs immediately and should run successfully.
+    |    #{plan2.id} is delayed and should run once #{plan1.id} finishes.
+    |    #{plan3.id} and #{plan4.id} are delayed and should run once #{plan2.id} finishes.
+    |
+    |    #{plan5.id} runs immediately and is expected to fail.
+    |    #{plan6.id} should not run at all as its prerequisite failed.
+    |
+    |  Visit #{ExampleHelper::DYNFLOW_URL} to see their status.
+    |
+  MSG
+
+  ExampleHelper.run_web_console(world)
+end

lib/dynflow/debug/telemetry/persistence.rb

@@ -19,7 +19,7 @@ module Persistence
           :load_execution_plan,
           :save_execution_plan,
           :find_old_execution_plans,
-          :find_past_delayed_plans,
+          :find_ready_delayed_plans,
           :delete_delayed_plans,
           :save_delayed_plan,
           :set_delayed_plan_frozen,

lib/dynflow/delayed_executors/abstract_core.rb

@@ -32,7 +32,7 @@ def time
 
       def delayed_execution_plans(time)
         with_error_handling([]) do
-          world.persistence.find_past_delayed_plans(time)
+          world.persistence.find_ready_delayed_plans(time)
         end
       end
 

lib/dynflow/delayed_plan.rb

@@ -31,6 +31,12 @@ def timeout
       error("Execution plan could not be started before set time (#{@start_before})", 'timeout')
     end
 
+    def failed_dependencies(uuids)
+      bullets = uuids.map { |u| "- #{u}" }.join("\n")
+      msg = "Execution plan could not be started because some of its prerequisite execution plans failed:\n#{bullets}"
+      error(msg, 'failed-dependency')
+    end
+
     def error(message, history_entry = nil)
       execution_plan.root_plan_step.state = :error
       execution_plan.root_plan_step.error = ::Dynflow::ExecutionPlan::Steps::Error.new(message)

lib/dynflow/director.rb

@@ -114,7 +114,15 @@ def execute
         plan = world.persistence.load_delayed_plan(execution_plan_id)
         return if plan.nil? || plan.execution_plan.state != :scheduled
 
-        if !plan.start_before.nil? && plan.start_before < Time.now.utc()
+        if plan.start_before.nil?
+          blocker_ids = world.persistence.find_execution_plan_dependencies(execution_plan_id)
+          statuses = world.persistence.find_execution_plan_statuses({ filters: { uuid: blocker_ids } })
+          failed = statuses.select { |_uuid, status| status[:state] == 'stopped' && status[:result] == 'error' }
+          if failed.any?
+            plan.failed_dependencies(failed.keys)
+            return
+          end
+        elsif plan.start_before < Time.now.utc()
           plan.timeout
           return
         end

lib/dynflow/persistence.rb

@@ -101,8 +101,16 @@ def find_old_execution_plans(age)
       end
     end
 
-    def find_past_delayed_plans(time)
-      adapter.find_past_delayed_plans(time).map do |plan|
+    def find_execution_plan_dependencies(execution_plan_id)
+      adapter.find_execution_plan_dependencies(execution_plan_id)
+    end
+
+    def find_blocked_execution_plans(execution_plan_id)
+      adapter.find_blocked_execution_plans(execution_plan_id)
+    end
+
+    def find_ready_delayed_plans(time)
+      adapter.find_ready_delayed_plans(time).map do |plan|
         DelayedPlan.new_from_hash(@world, plan)
       end
     end
@@ -163,5 +171,9 @@ def prune_envelopes(receiver_ids)
     def prune_undeliverable_envelopes
       adapter.prune_undeliverable_envelopes
     end
+
+    def chain_execution_plan(first, second)
+      adapter.chain_execution_plan(first, second)
+    end
   end
 end

lib/dynflow/persistence_adapters/abstract.rb

@@ -72,7 +72,15 @@ def save_execution_plan(execution_plan_id, value)
         raise NotImplementedError
       end
 
-      def find_past_delayed_plans(options = {})
+      def find_execution_plan_dependencies(execution_plan_id)
+        raise NotImplementedError
+      end
+
+      def find_blocked_execution_plans(execution_plan_id)
+        raise NotImplementedError
+      end
+
+      def find_ready_delayed_plans(options = {})
         raise NotImplementedError
       end
 

lib/dynflow/persistence_adapters/sequel.rb

@@ -39,7 +39,8 @@ class action_class execution_plan_uuid queue),
                     envelope:            %w(receiver_id),
                     coordinator_record:  %w(id owner_id class),
                     delayed:             %w(execution_plan_uuid start_at start_before args_serializer frozen),
-                    output_chunk:        %w(execution_plan_uuid action_id kind timestamp) }
+                    output_chunk:        %w(execution_plan_uuid action_id kind timestamp),
+                    execution_plan_dependency: %w(execution_plan_uuid blocked_by_uuid) }
 
       SERIALIZABLE_COLUMNS = { action:  %w(input output),
                                delayed: %w(serialized_args),
@@ -153,12 +154,31 @@ def find_old_execution_plans(age)
         records.map { |plan| execution_plan_column_map(load_data plan, table_name) }
       end
 
-      def find_past_delayed_plans(time)
+      def find_execution_plan_dependencies(execution_plan_id)
+        table(:execution_plan_dependency)
+          .where(execution_plan_uuid: execution_plan_id)
+          .select_map(:blocked_by_uuid)
+      end
+
+      def find_blocked_execution_plans(execution_plan_id)
+        table(:execution_plan_dependency)
+          .where(blocked_by_uuid: execution_plan_id)
+          .select_map(:execution_plan_uuid)
+      end
+
+      def find_ready_delayed_plans(time)
         table_name = :delayed
+        # Subquery to find delayed plans that have at least one non-stopped dependency
+        plans_with_unfinished_deps = table(:execution_plan_dependency)
+                                     .join(TABLES[:execution_plan], uuid: :blocked_by_uuid)
+                                     .where(::Sequel.~(state: 'stopped'))
+                                     .select(:execution_plan_uuid)
+
         records = with_retry do
           table(table_name)
-            .where(::Sequel.lit('start_at <= ? OR (start_before IS NOT NULL AND start_before <= ?)', time, time))
+            .where(::Sequel.lit('start_at IS NULL OR (start_at <= ? OR (start_before IS NOT NULL AND start_before <= ?))', time, time))
             .where(:frozen => false)
+            .exclude(execution_plan_uuid: plans_with_unfinished_deps)
             .order_by(:start_at)
             .all
         end
@@ -175,6 +195,10 @@ def save_delayed_plan(execution_plan_id, value)
         save :delayed, { execution_plan_uuid: execution_plan_id }, value, with_data: false
       end
 
+      def chain_execution_plan(first, second)
+        save :execution_plan_dependency, {}, { execution_plan_uuid: second, blocked_by_uuid: first }, with_data: false
+      end
+
       def load_step(execution_plan_id, step_id)
         load :step, execution_plan_uuid: execution_plan_id, id: step_id
       end
@@ -319,7 +343,8 @@ def abort_if_pending_migrations!
                  envelope:            :dynflow_envelopes,
                  coordinator_record:  :dynflow_coordinator_records,
                  delayed:             :dynflow_delayed_plans,
-                 output_chunk:        :dynflow_output_chunks }
+                 output_chunk:        :dynflow_output_chunks,
+                 execution_plan_dependency: :dynflow_execution_plan_dependencies }
 
       def table(which)
         db[TABLES.fetch(which)]

lib/dynflow/persistence_adapters/sequel_migrations/025_create_execution_plan_dependencies.rb

@@ -0,0 +1,22 @@
+# frozen_string_literal: true
+
+Sequel.migration do
+  up do
+    type = database_type
+    create_table(:dynflow_execution_plan_dependencies) do
+      column_properties = if type.to_s.include?('postgres')
+                            { type: :uuid }
+                          else
+                            { type: String, size: 36, fixed: true, null: false }
+                          end
+      foreign_key :execution_plan_uuid, :dynflow_execution_plans, on_delete: :cascade, **column_properties
+      foreign_key :blocked_by_uuid, :dynflow_execution_plans, on_delete: :cascade, **column_properties
+      index :blocked_by_uuid
+      index :execution_plan_uuid
+    end
+  end
+
+  down do
+    drop_table(:dynflow_execution_plan_dependencies)
+  end
+end

lib/dynflow/world.rb

@@ -202,6 +202,16 @@ def delay_with_options(action_class:, args:, delay_options:, id: nil, caller_act
       Scheduled[execution_plan.id]
     end
 
+    def chain(plan_uuids, action_class, *args)
+      plan_uuids = [plan_uuids] unless plan_uuids.is_a? Array
+      result = delay_with_options(action_class: action_class, args: args, delay_options: { frozen: true })
+      plan_uuids.each do |plan_uuid|
+        persistence.chain_execution_plan(plan_uuid, result.execution_plan_id)
+      end
+      persistence.set_delayed_plan_frozen(result.execution_plan_id, false)
+      result
+    end
+
     def plan_elsewhere(action_class, *args)
       execution_plan = ExecutionPlan.new(self, nil)
       execution_plan.delay(nil, action_class, {}, *args)