Ensure ems workers are killed by their server/orchestrator pod

app/models/ext_management_system.rb

@@ -622,11 +622,33 @@ def destroy_queue
     task.id
   end
 
+  def ems_workers
+    MiqWorker.find_alive.where(:queue_name => queue_name)
+  end
+
+  def wait_for_ems_workers_removal
+    return if Rails.env.test?
+
+    quiesce_loop_timeout = ::Settings.server.worker_monitor.quiesce_loop_timeout || 5.minutes
+    worker_monitor_poll  = (::Settings.server.worker_monitor.poll || 1.second).to_i_with_method
+    kill_ems_workers_started_on = Time.now.utc
+
+    loop do
+      # killed workers will have their row removed, so we wait for this
+      break unless ems_workers.exists?
+      break if (Time.now.utc - kill_ems_workers_started_on) > quiesce_loop_timeout
+
+      sleep worker_monitor_poll
+    end
+  end
+
   def destroy(task_id = nil)
     disable!(:validate => false) if enabled?
 
-    # kill workers
-    MiqWorker.find_alive.where(:queue_name => queue_name).each(&:kill)
+    # Async kill each ems worker and wait until their row is removed before we delete
+    # the ems/managers to ensure a worker doesn't recreate the ems/manager.
+    ems_workers.each(&:kill_async)
+    wait_for_ems_workers_removal
 
     _log.info("Destroying #{child_managers.count} child_managers")
     child_managers.destroy_all

app/models/miq_worker.rb

@@ -394,6 +394,21 @@ def kill
     destroy
   end
 
+  # kill needs be done by the worker's orchestrator pod / server process
+  # TODO: Note, stop is async through the queue, while kill is sync.  Should kill be async too?
+  # Also, this looks a lot like MiqServer#stop_worker_queue except stop_worker is called on the server row whereas
+  # we're calling kill on the worker row.
+  def kill_async
+    MiqQueue.put_unless_exists(
+      :class_name  => self.class.name,
+      :instance_id => id,
+      :method_name => 'kill',
+      :queue_name  => 'miq_server',
+      :server_guid => miq_server.guid,
+      :zone        => miq_server.my_zone
+    )
+  end
+
   def kill_process
     if containerized_worker?
       delete_container_objects

spec/models/ext_management_system_spec.rb

@@ -666,8 +666,15 @@
 
     it "destroys an ems with active workers" do
       ems = FactoryBot.create(:ext_management_system)
-      worker = FactoryBot.create(:miq_ems_refresh_worker, :queue_name => ems.queue_name, :status => "started")
+      worker = FactoryBot.create(:miq_ems_refresh_worker, :queue_name => ems.queue_name, :status => "started", :miq_server => EvmSpecHelper.local_miq_server)
+
       ems.destroy
+
+      # Simulate another process delivering the worker kill message
+      queue_message = MiqQueue.order(:id).first
+      status, message, result = queue_message.deliver
+      queue_message.delivered(status, message, result)
+
       expect(ExtManagementSystem.count).to eq(0)
       expect(worker.class.exists?(worker.id)).to eq(false)
     end
@@ -719,8 +726,8 @@
       ems.destroy_queue
 
       expect(MiqQueue.count).to eq(1)
-
-      deliver_queue_message
+      deliver_queue_message #  ems destroy message
+      deliver_queue_message #  worker kill message
 
       expect(MiqQueue.count).to eq(0)
       expect(ExtManagementSystem.count).to eq(0)

spec/models/miq_worker_spec.rb

@@ -392,6 +392,33 @@ def check_has_required_role(worker_role_names, expected_result)
       end
     end
 
+    describe "#kill_async" do
+      let!(:remote_server) { EvmSpecHelper.remote_guid_miq_server_zone[1] }
+      let!(:local_server)  { EvmSpecHelper.local_guid_miq_server_zone[1] }
+
+      it "queues local worker to local server" do
+        worker = FactoryBot.create(:miq_worker, :miq_server => local_server)
+        worker.kill_async
+        msg = MiqQueue.where(:method_name => 'kill', :class_name => worker.class.name).first
+        expect(msg).to have_attributes(
+          :queue_name  => 'miq_server',
+          :server_guid => local_server.guid,
+          :zone        => local_server.my_zone
+        )
+      end
+
+      it "queues remote worker to remote server" do
+        worker = FactoryBot.create(:miq_worker, :miq_server => remote_server)
+        worker.kill_async
+        msg = MiqQueue.where(:method_name => 'kill', :class_name => worker.class.name).first
+        expect(msg).to have_attributes(
+          :queue_name  => 'miq_server',
+          :server_guid => remote_server.guid,
+          :zone        => remote_server.my_zone
+        )
+      end
+    end
+
     describe "#stopping_for_too_long?" do
       subject { @worker.stopping_for_too_long? }