diff --git a/examples/remote_executor.rb b/examples/remote_executor.rb index 8b3749c6..348e384a 100755 --- a/examples/remote_executor.rb +++ b/examples/remote_executor.rb @@ -62,8 +62,9 @@ def run_server run(world) end - def initialize_sidekiq_orchestrator + def initialize_sidekiq_orchestrator(id) ExampleHelper.create_world do |config| + config.id = id if id config.persistence_adapter = persistence_adapter config.connector = connector config.executor = ::Dynflow::Executors::Sidekiq::Core @@ -164,8 +165,9 @@ def run_client(count) # assuming the remote executor was required as part of initialization # of the ActiveJob worker queues = Sidekiq.configure_server { |c| c.options[:queues] } - world = if queues.include?("dynflow_orchestrator") - RemoteExecutorExample.initialize_sidekiq_orchestrator + world = if queues.include?("dynflow_orchestrator") || (orchestrator_queue = queues.find { |q| q.start_with?('dynflow_orchestrator:') }) + orchestrator_queue ||= 'dynflow_orchestrator' + RemoteExecutorExample.initialize_sidekiq_orchestrator(orchestrator_queue.split(':')[1]) elsif (queues - ['dynflow_orchestrator']).any? RemoteExecutorExample.initialize_sidekiq_worker end diff --git a/lib/dynflow/config.rb b/lib/dynflow/config.rb index 79bd4178..1eef91aa 100644 --- a/lib/dynflow/config.rb +++ b/lib/dynflow/config.rb @@ -79,6 +79,10 @@ def queues @queues ||= QueuesConfig.new end + config_attr :id, String do |world, config| + SecureRandom.uuid + end + config_attr :logger_adapter, LoggerAdapters::Abstract do LoggerAdapters::Simple.new end diff --git a/lib/dynflow/executors/abstract/core.rb b/lib/dynflow/executors/abstract/core.rb index ff3fa7b2..b95bf632 100644 --- a/lib/dynflow/executors/abstract/core.rb +++ b/lib/dynflow/executors/abstract/core.rb @@ -89,6 +89,9 @@ def execution_status(execution_plan_id = nil) {} end + def prune_orphaned_queues + end + def heartbeat @logger.debug('Executor heartbeat') record = @world.coordinator.find_records(:id => @world.id, diff --git a/lib/dynflow/executors/parallel.rb b/lib/dynflow/executors/parallel.rb index ba735cbb..2bc1c2fa 100644 --- a/lib/dynflow/executors/parallel.rb +++ b/lib/dynflow/executors/parallel.rb @@ -61,6 +61,10 @@ def halt(execution_plan_id) @core.tell([:halt, execution_plan_id]) end + def prune_orphaned_queues + @core.tell(:prune_orphaned_queues) + end + def initialized @core_initialized end diff --git a/lib/dynflow/executors/sidekiq/core.rb b/lib/dynflow/executors/sidekiq/core.rb index 2bfbf126..edaa7bad 100644 --- a/lib/dynflow/executors/sidekiq/core.rb +++ b/lib/dynflow/executors/sidekiq/core.rb @@ -6,6 +6,7 @@ require 'dynflow/executors/sidekiq/worker_jobs' require 'dynflow/executors/sidekiq/redis_locking' +require 'sidekiq/api' require 'sidekiq-reliable-fetch' Sidekiq.configure_server do |config| # Use semi-reliable fetch @@ -28,20 +29,22 @@ class Core < Abstract::Core def initialize(world, *_args) @world = world @logger = world.logger - wait_for_orchestrator_lock + @subqueue = ::Sidekiq.configure_server { |c| c[:queues].find { |q| q.start_with? 'dynflow_orchestrator:' } } + @reply_queue = @subqueue || 'dynflow_orchestrator' + wait_for_orchestrator_lock unless @subqueue super schedule_update_telemetry - begin_startup! + begin_startup! unless @subqueue end def heartbeat super - reacquire_orchestrator_lock + reacquire_orchestrator_lock unless @subqueue end def start_termination(*args) super - release_orchestrator_lock + release_orchestrator_lock unless @subqueue finish_termination end @@ -50,9 +53,20 @@ def execution_status(execution_plan_id = nil) {} end + def prune_orphaned_queues + active_world_ids = @world.coordinator.find_worlds(true).map(&:id) + ::Sidekiq::Queue.all.each do |queue| + next unless queue.name.start_with?('dynflow_orchestrator:') + world_id = queue.name.split(':', 2)[1] + next if active_world_ids.include?(world_id) + logger.info("Removing orphaned orchestrator queue #{queue.name}") + queue.clear + end + end + def feed_pool(work_items) work_items.each do |new_work| - WorkerJobs::PerformWork.set(queue: suggest_queue(new_work)).perform_async(new_work) + WorkerJobs::PerformWork.set(queue: suggest_queue(new_work)).perform_async(new_work, @reply_queue) end end @@ -68,6 +82,8 @@ def update_telemetry end def work_finished(work, delayed_events = nil) + return super if @subqueue + # If the work item is sent in reply to a request from the current orchestrator, proceed if work.sender_orchestrator_id == @world.id super diff --git a/lib/dynflow/executors/sidekiq/orchestrator_jobs.rb b/lib/dynflow/executors/sidekiq/orchestrator_jobs.rb index 9044a95a..0f5f80cd 100644 --- a/lib/dynflow/executors/sidekiq/orchestrator_jobs.rb +++ b/lib/dynflow/executors/sidekiq/orchestrator_jobs.rb @@ -7,8 +7,6 @@ module OrchestratorJobs # handles resposnes about finished work form the workers # or some event to handle on orchestrator side class WorkerDone < InternalJobBase - sidekiq_options queue: :dynflow_orchestrator - # @param request_envelope [Dispatcher::Request] - request to handle on orchestrator side # usually to start new execution or to pass some event def perform(work_item, delayed_events = nil) @@ -23,8 +21,6 @@ def perform(work_item, delayed_events = nil) # handles setting up an event on orchestrator class PlanEvent < InternalJobBase - sidekiq_options queue: :dynflow_orchestrator - # @param event_envelope [Dispatcher::Event] - request to handle on orchestrator side # usually to start new execution or to pass some event def perform(execution_plan_id, step_id, event, time) @@ -33,8 +29,6 @@ def perform(execution_plan_id, step_id, event, time) end class HandlePersistenceError < InternalJobBase - sidekiq_options queue: :dynflow_orchestrator - # @param request_envelope [Dispatcher::Request] - request to handle on orchestrator side # usually to start new execution or to pass some event def perform(error, work_item) diff --git a/lib/dynflow/executors/sidekiq/worker_jobs.rb b/lib/dynflow/executors/sidekiq/worker_jobs.rb index 98abdd85..46446073 100644 --- a/lib/dynflow/executors/sidekiq/worker_jobs.rb +++ b/lib/dynflow/executors/sidekiq/worker_jobs.rb @@ -5,7 +5,7 @@ module Executors module Sidekiq module WorkerJobs class PerformWork < InternalJobBase - def perform(work_item) + def perform(work_item, respond_to = 'dynflow_orchestrator') with_telemetry(work_item) do Executors.run_user_code do work_item.world = Dynflow.process_world @@ -13,10 +13,10 @@ def perform(work_item) end end rescue Errors::PersistenceError => e - OrchestratorJobs::HandlePersistenceError.perform_async(e, work_item) + OrchestratorJobs::HandlePersistenceError.set(queue: respond_to).perform_async(e, work_item) ensure step = work_item.step if work_item.is_a?(Director::StepWorkItem) - OrchestratorJobs::WorkerDone.perform_async(work_item, step && step.delayed_events) + OrchestratorJobs::WorkerDone.set(queue: respond_to).perform_async(work_item, step && step.delayed_events) end private diff --git a/lib/dynflow/world.rb b/lib/dynflow/world.rb index 2dc1856f..212fbd8a 100644 --- a/lib/dynflow/world.rb +++ b/lib/dynflow/world.rb @@ -23,7 +23,7 @@ def initialize(config) Dynflow::Telemetry.set_adapter @config.telemetry_adapter Dynflow::Telemetry.register_metrics! - @id = SecureRandom.uuid + @id = @config.id @logger_adapter = @config.logger_adapter @clock = spawn_and_wait(Clock, 'clock', logger) @config.validate diff --git a/lib/dynflow/world/invalidation.rb b/lib/dynflow/world/invalidation.rb index 8f1fd87b..f773ed57 100644 --- a/lib/dynflow/world/invalidation.rb +++ b/lib/dynflow/world/invalidation.rb @@ -142,6 +142,7 @@ def perform_validity_checks end pruned = connector.prune_undeliverable_envelopes(self) logger.error("Pruned #{pruned} undeliverable envelopes") unless pruned.zero? + executor.prune_orphaned_queues if executor world_invalidation_result.values.select { |result| result == :invalidated }.size end diff --git a/test/bats/sidekiq-orchestrator.bats b/test/bats/sidekiq-orchestrator.bats index b75ae5f3..2c7ad09d 100644 --- a/test/bats/sidekiq-orchestrator.bats +++ b/test/bats/sidekiq-orchestrator.bats @@ -176,3 +176,27 @@ teardown() { timeout 10 bundle exec ruby examples/remote_executor.rb client 1 wait_for 1 1 grep -P 'dynflow: ExecutionPlan.*running >>.*stopped' "$(bg_output_file o2)" } + +@test "multi-orchestrator — jobs spread across two orchestrators using per-orchestrator queues" { + cd "$(get_project_root)" + + # Start orchestrator 1 with subqueue dynflow_orchestrator:orch-a + uuid1=$(uuidgen) + run_background 'o1' bundle exec sidekiq -c 1 -r ./examples/remote_executor.rb -q dynflow_orchestrator:${uuid1} + wait_for 30 1 grep -q "World ${uuid1} started..." "$(bg_output_file o1)" + + # Start orchestrator 2 with subqueue dynflow_orchestrator:orch-b + uuid2=$(uuidgen) + run_background 'o2' bundle exec sidekiq -c 1 -r ./examples/remote_executor.rb -q dynflow_orchestrator:${uuid2} + wait_for 30 1 grep -q "World ${uuid2} started..." "$(bg_output_file o2)" + + # Start one worker + run_background 'w1' bundle exec sidekiq -r ./examples/remote_executor.rb -q default + + # Trigger 10 jobs — with 2 orchestrators round-robin, each should get ~5 + timeout 60 bundle exec ruby examples/remote_executor.rb client 10 + + # Assert both orchestrators handled at least one job + wait_for 5 1 grep -qP 'ExecutionPlan.*running >>.*stopped' "$(bg_output_file o1)" + wait_for 5 1 grep -qP 'ExecutionPlan.*running >>.*stopped' "$(bg_output_file o2)" +} diff --git a/test/sidekiq_multi_orchestrator_test.rb b/test/sidekiq_multi_orchestrator_test.rb new file mode 100644 index 00000000..eed02c77 --- /dev/null +++ b/test/sidekiq_multi_orchestrator_test.rb @@ -0,0 +1,176 @@ +# frozen_string_literal: true + +require_relative 'test_helper' +require 'mocha/minitest' + +require 'sidekiq' +require 'sidekiq/api' +require 'dynflow/executors/sidekiq/core' + +module Dynflow + module SidekiqMultiOrchestratorTest + describe Executors::Sidekiq::Core do + after do + ::Dynflow.instance_variable_set('@process_world', nil) + end + + # @param subqueue [String, nil] the per-orchestrator queue (e.g. "dynflow_orchestrator:abc") this + # "process" is configured to listen on, or nil to simulate the legacy shared-queue setup + def build_world(subqueue: nil, id: nil) + ::Sidekiq.stubs(:configure_server).returns(subqueue) + world = WorldFactory.create_world do |c| + c.executor = Executors::Sidekiq::Core + c.id = id if id + end + ::Dynflow.instance_variable_set('@process_world', world) + world + end + + # digs out the actual Sidekiq::Core instance behind the actor reference, mirroring + # the get_director helper in test_helper.rb + def raw_core(world) + world.executor.instance_variable_get('@core').instance_variable_get('@core').context + end + + describe 'reply queue selection' do + it 'falls back to the shared queue when no per-orchestrator queue is configured' do + Executors::Sidekiq::Core.any_instance.expects(:wait_for_orchestrator_lock) + Executors::Sidekiq::Core.any_instance.expects(:begin_startup!) + world = build_world(subqueue: nil) + _(raw_core(world).instance_variable_get('@reply_queue')).must_equal 'dynflow_orchestrator' + end + + it 'uses the per-orchestrator queue when one is configured' do + Executors::Sidekiq::Core.any_instance.expects(:wait_for_orchestrator_lock).never + Executors::Sidekiq::Core.any_instance.expects(:begin_startup!).never + world = build_world(subqueue: 'dynflow_orchestrator:some-world-id') + _(raw_core(world).instance_variable_get('@reply_queue')).must_equal 'dynflow_orchestrator:some-world-id' + end + end + + describe 'heartbeat lock reacquisition' do + it 'reacquires the orchestrator lock when no subqueue is configured' do + Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock) + Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!) + Executors::Sidekiq::Core.any_instance.expects(:reacquire_orchestrator_lock) + world = build_world(subqueue: nil) + world.executor.core.ask!(:heartbeat) + end + + it 'does not touch the lock when a subqueue is configured' do + Executors::Sidekiq::Core.any_instance.expects(:reacquire_orchestrator_lock).never + world = build_world(subqueue: 'dynflow_orchestrator:some-world-id') + world.executor.core.ask!(:heartbeat) + end + end + + describe 'termination lock release' do + it 'releases the orchestrator lock when no subqueue is configured' do + Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock) + Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!) + Executors::Sidekiq::Core.any_instance.expects(:release_orchestrator_lock) + world = build_world(subqueue: nil) + world.terminate.wait(5) + end + + it 'does not touch the lock when a subqueue is configured' do + Executors::Sidekiq::Core.any_instance.expects(:release_orchestrator_lock).never + world = build_world(subqueue: 'dynflow_orchestrator:some-world-id') + world.terminate.wait(5) + end + end + + describe '#feed_pool' do + it 'tells workers to reply on the shared queue by default' do + Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock) + Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!) + world = build_world(subqueue: nil) + work_item = stub('work_item', :queue => :default) + proxy = mock('perform_async_proxy') + proxy.expects(:perform_async).with(work_item, 'dynflow_orchestrator') + Executors::Sidekiq::WorkerJobs::PerformWork.expects(:set).with(:queue => :default).returns(proxy) + raw_core(world).feed_pool([work_item]) + end + + it 'tells workers to reply on the orchestrator-specific queue when configured' do + reply_queue = 'dynflow_orchestrator:some-world-id' + world = build_world(subqueue: reply_queue) + work_item = stub('work_item', :queue => :default) + proxy = mock('perform_async_proxy') + proxy.expects(:perform_async).with(work_item, reply_queue) + Executors::Sidekiq::WorkerJobs::PerformWork.expects(:set).with(:queue => :default).returns(proxy) + raw_core(world).feed_pool([work_item]) + end + end + + describe '#work_finished' do + it 'processes any work unconditionally when a subqueue is configured' do + world = build_world(subqueue: 'dynflow_orchestrator:some-world-id') + work = stub('work', :sender_orchestrator_id => 'someone-else') + Executors::Abstract::Core.any_instance.expects(:work_finished).with(work, nil) + raw_core(world).work_finished(work) + end + + it 'processes work sent by this orchestrator even without a subqueue' do + Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock) + Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!) + world = build_world(subqueue: nil) + work = stub('work', :sender_orchestrator_id => world.id) + Executors::Abstract::Core.any_instance.expects(:work_finished).with(work, nil) + raw_core(world).work_finished(work) + end + + it 'defers foreign work to handle_unknown_work_item when no subqueue is configured' do + Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock) + Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!) + world = build_world(subqueue: nil) + core = raw_core(world) + work = stub('work', :sender_orchestrator_id => 'someone-else') + Executors::Abstract::Core.any_instance.expects(:work_finished).never + core.expects(:handle_unknown_work_item).with(work) + core.work_finished(work) + end + end + + describe 'drain / startup-complete recovery flow' do + it 'begin_startup! enqueues a DrainMarker job for this world' do + Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock) + Executors::Sidekiq::WorkerJobs::DrainMarker.expects(:perform_async).with('fixed-test-world-id') + build_world(subqueue: nil, id: 'fixed-test-world-id') + end + + it 'DrainMarker hands off to StartupComplete for the same world id' do + Executors::Sidekiq::OrchestratorJobs::StartupComplete.expects(:perform_async).with('some-world-id') + Executors::Sidekiq::WorkerJobs::DrainMarker.new.perform('some-world-id') + end + + it 'StartupComplete tells the matching orchestrator core that startup completed' do + Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock) + Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!) + world = build_world(subqueue: nil) + world.executor.core.expects(:tell).with([:startup_complete]) + Executors::Sidekiq::OrchestratorJobs::StartupComplete.new.perform(world.id) + end + + it 'StartupComplete discards notifications meant for a different world' do + Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock) + Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!) + world = build_world(subqueue: nil) + world.executor.core.expects(:tell).never + Executors::Sidekiq::OrchestratorJobs::StartupComplete.new.perform('a-different-world-id') + end + + it '#startup_complete runs validity checks and clears recovery mode' do + Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock) + Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!) + world = build_world(subqueue: nil) + core = raw_core(world) + core.instance_variable_set('@recovery', true) + world.expects(:perform_validity_checks) + core.startup_complete + _(core.instance_variable_get('@recovery')).must_equal false + end + end + end + end +end diff --git a/test/sidekiq_prune_orphaned_queues_test.rb b/test/sidekiq_prune_orphaned_queues_test.rb new file mode 100644 index 00000000..0cb7f44a --- /dev/null +++ b/test/sidekiq_prune_orphaned_queues_test.rb @@ -0,0 +1,73 @@ +# frozen_string_literal: true + +require_relative 'test_helper' +require 'mocha/minitest' + +require 'sidekiq' +require 'sidekiq/api' +require 'dynflow/executors/sidekiq/core' + +module Dynflow + module SidekiqPruneOrphanedQueuesTest + describe Executors::Sidekiq::Core do + before do + Executors::Sidekiq::Core.any_instance.stubs(:wait_for_orchestrator_lock) + Executors::Sidekiq::Core.any_instance.stubs(:begin_startup!) + end + + after do + ::Dynflow.instance_variable_set('@process_world', nil) + ::Sidekiq.redis do |conn| + conn.smembers('queues').each do |name| + next unless name.start_with?('dynflow_orchestrator:') + conn.del("queue:#{name}") + conn.srem?('queues', name) + end + end + end + + let(:world) do + world = WorldFactory.create_world { |c| c.executor = Executors::Sidekiq::Core } + ::Dynflow.instance_variable_set('@process_world', world) + world + end + + def seed_queue(name) + ::Sidekiq.redis do |conn| + conn.sadd?('queues', name) + conn.lpush("queue:#{name}", '{}') + end + end + + def queue_names + ::Sidekiq::Queue.all.map(&:name) + end + + it 'removes orchestrator queues for worlds without an active executor, keeps the rest' do + active_queue = "dynflow_orchestrator:#{world.id}" + stale_queue = 'dynflow_orchestrator:stale-world-id' + seed_queue(active_queue) + seed_queue(stale_queue) + + world.executor.core.ask!(:prune_orphaned_queues) + + _(queue_names).must_include active_queue + _(queue_names).wont_include stale_queue + end + + it 'leaves unrelated queues untouched' do + other_queue = 'default' + seed_queue(other_queue) + + world.executor.core.ask!(:prune_orphaned_queues) + + _(queue_names).must_include other_queue + + ::Sidekiq.redis do |conn| + conn.del("queue:#{other_queue}") + conn.srem?('queues', other_queue) + end + end + end + end +end diff --git a/test/world_test.rb b/test/world_test.rb index e87135ad..efcefaf4 100644 --- a/test/world_test.rb +++ b/test/world_test.rb @@ -2,6 +2,7 @@ require_relative 'test_helper' require 'fileutils' +require 'mocha/minitest' module Dynflow module WorldTest @@ -9,6 +10,32 @@ module WorldTest let(:world) { WorldFactory.create_world } let(:world_with_custom_meta) { WorldFactory.create_world { |c| c.meta = { 'fast' => true } } } + describe '#id' do + it 'defaults to a random uuid, different for each world' do + other_world = WorldFactory.create_world + _(world.id).must_match(/\A[0-9a-f-]{36}\z/) + _(world.id).wont_equal other_world.id + end + + it 'is configurable' do + custom_id = 'my-custom-world-id' + custom_world = WorldFactory.create_world { |c| c.id = custom_id } + _(custom_world.id).must_equal custom_id + end + end + + describe '#perform_validity_checks' do + it 'prunes orphaned executor queues' do + world.executor.expects(:prune_orphaned_queues) + world.perform_validity_checks + end + + it 'does not fail when there is no executor' do + client_world = WorldFactory.create_world { |c| c.executor = false } + client_world.perform_validity_checks + end + end + describe '#meta' do it 'by default informs about the hostname and the pid running the world' do registered_world = world.coordinator.find_worlds(false, id: world.id).first