Only register queue workers using drb for dequeue

app/models/miq_queue_worker_base/runner.rb

@@ -15,7 +15,7 @@ def sync_dequeue_method
   end
 
   def dequeue_method_via_drb?
-    @dequeue_method == :drb
+    @dequeue_method == :drb && drb_dequeue_available?
   end
 
   def thresholds_exceeded?
@@ -130,6 +130,8 @@ def deliver_message(msg)
   end
 
   def do_work
+    register_worker_with_worker_monitor if dequeue_method_via_drb?
+
     # Keep collecting messages from the queue until the queue is empty,
     #   so we don't sleep in between messages
     loop do
@@ -143,6 +145,21 @@ def do_work
 
   private
 
+  def register_worker_with_worker_monitor
+    worker_monitor_drb.register_worker(@worker.pid, @worker.class.name, @worker.queue_name)
+  rescue DRb::DRbError => err
+    do_exit("Failed to register worker with worker monitor: #{err.class.name}: #{err.message}", 1)
+  end
+
+  def drb_dequeue_available?
+    @drb_dequeue_available ||=
+      begin
+        server.drb_uri.present? && worker_monitor_drb.respond_to?(:register_worker)
+      rescue DRb::DRbError
+        false
+      end
+  end
+
   # Only for file based heartbeating
   def heartbeat_message_timeout(message)
     if message.msg_timeout

app/models/miq_server/worker_management/dequeue.rb

@@ -97,6 +97,17 @@ def get_worker_count_and_priority_by_queue_name
     queue_names
   end
 
+  def register_worker(worker_pid, worker_class, queue_name)
+    worker_class = worker_class.constantize if worker_class.kind_of?(String)
+
+    @workers_lock.synchronize(:EX) do
+      worker_add(worker_pid)
+      h = @workers[worker_pid]
+      h[:class] ||= worker_class
+      h[:queue_name] ||= queue_name
+    end unless @workers_lock.nil?
+  end
+
   def populate_queue_messages
     queue_names = get_worker_count_and_priority_by_queue_name
     @queue_messages_lock.synchronize(:EX) do

app/models/miq_server/worker_management/heartbeat.rb

@@ -1,17 +1,6 @@
 module MiqServer::WorkerManagement::Heartbeat
   extend ActiveSupport::Concern
 
-  def register_worker(worker_pid, worker_class, queue_name)
-    worker_class = worker_class.constantize if worker_class.kind_of?(String)
-
-    @workers_lock.synchronize(:EX) do
-      worker_add(worker_pid)
-      h = @workers[worker_pid]
-      h[:class] ||= worker_class
-      h[:queue_name] ||= queue_name
-    end unless @workers_lock.nil?
-  end
-
   def persist_last_heartbeat(w)
     last_heartbeat = workers_last_heartbeat(w)
 

app/models/miq_worker/runner.rb

@@ -288,8 +288,6 @@ def heartbeat
       _log.info("#{log_prefix} Synchronizing configuration complete...")
     end
 
-    register_worker_with_worker_monitor unless MiqEnvironment::Command.is_podified?
-
     @last_hb = now
     do_heartbeat_work
   rescue SystemExit, SignalException
@@ -298,12 +296,6 @@ def heartbeat
     do_exit("Error heartbeating because #{err.class.name}: #{err.message}\n#{err.backtrace.join('\n')}", 1)
   end
 
-  def register_worker_with_worker_monitor
-    worker_monitor_drb.register_worker(@worker.pid, @worker.class.name, @worker.queue_name)
-  rescue DRb::DRbError => err
-    do_exit("Error processing messages from MiqServer because #{err.class.name}: #{err.message}", 1)
-  end
-
   def heartbeat_to_file(timeout = nil)
     # Disable heartbeat check.  Useful if a worker is running in isolation
     # without the oversight of MiqServer::WorkerManagement

spec/models/miq_queue_worker_base/runner_spec.rb

@@ -20,4 +20,48 @@
       expect(q1.reload.state).to eql(MiqQueue::STATUS_ERROR)
     end
   end
+
+  describe "#dequeue_method_via_drb?" do
+    let(:server) { EvmSpecHelper.local_miq_server }
+    let(:worker) { FactoryBot.create(:miq_generic_worker, :miq_server => server) }
+    let(:runner) do
+      described_class.allocate.tap do |r|
+        r.worker = worker
+        r.instance_variable_set(:@server, server)
+      end
+    end
+
+    it "returns false when the instance variable is not :drb" do
+      runner.instance_variable_set(:@dequeue_method, :sql)
+      expect(runner.dequeue_method_via_drb?).to be_falsey
+    end
+
+    context "when the instance variable is set to :drb" do
+      before { runner.instance_variable_set(:@dequeue_method, :drb) }
+
+      it "returns false if the server's drb uri is nil" do
+        server.update(:drb_uri => nil)
+        expect(runner.dequeue_method_via_drb?).to be_falsey
+      end
+
+      context "with a drb uri" do
+        let(:drb_object) { instance_double(DRbObject) }
+
+        before do
+          server.update(:drb_uri => "drbunix:///tmp/worker_monitor20200211-24337-1ma102h")
+          expect(runner).to receive(:worker_monitor_drb).and_return(drb_object)
+        end
+
+        it "returns false when the drb server is not accessible" do
+          expect(drb_object).to receive(:respond_to?).with(:register_worker).and_raise(DRb::DRbError)
+          expect(runner.dequeue_method_via_drb?).to be_falsey
+        end
+
+        it "returns true when the server is accessible" do
+          expect(drb_object).to receive(:respond_to?).with(:register_worker).and_return(true)
+          expect(runner.dequeue_method_via_drb?).to be_truthy
+        end
+      end
+    end
+  end
 end