Merge pull request #16685 from jerryk55/smart_proxy_heartbeat_thread

Add Heartbeat Thread to SmartProxy Worker

app/models/miq_smart_proxy_worker/runner.rb

@@ -1,3 +1,73 @@
 class MiqSmartProxyWorker::Runner < MiqQueueWorkerBase::Runner
   self.delay_startup_for_vim_broker = true # NOTE: For smartproxy role
+
+  def do_before_work_loop
+    @tid = start_heartbeat_thread
+  end
+
+  def before_exit(message, _exit_code)
+    @exit_requested = true
+    #
+    # Stop the Heartbeat Thread
+    #
+    safe_log("#{message} Stopping Heartbeat Thread.")
+
+    #
+    # Wait for the Heartbeat Thread to stop
+    #
+    if @tid
+      safe_log("#{message} Waiting for Heartbeat Thread to Stop.")
+      begin
+        @tid.join(worker_settings[:heartbeat_thread_shutdown_timeout])
+      rescue NoMethodError => join_err
+        safe_log(join_err)
+      end
+    end
+  end
+
+  def start_heartbeat_thread
+    @exit_requested    = false
+    @heartbeat_started = Concurrent::Event.new
+    _log.info("#{log_prefix} Starting Heartbeat Thread")
+
+    tid = Thread.new do
+      begin
+        heartbeat_thread
+      rescue => err
+        _log.error("#{log_prefix} Heartbeat Thread aborted because [#{err.message}]")
+        _log.log_backtrace(err)
+        Thread.exit
+      ensure
+        @heartbeat_started.set
+      end
+    end
+
+    @heartbeat_started.wait
+    _log.info("#{log_prefix} Started Heartbeat Thread")
+
+    tid
+  end
+
+  def heartbeat_thread
+    @heartbeat_started.set
+    until @exit_requested
+      heartbeat
+      sleep 30
+    end
+  end
+
+  def do_work
+    if @tid.nil? || !@tid.alive?
+      unless @tid.try(:status)
+        dead_tid, @tid = @tid, nil
+        _log.info("#{log_prefix} Waiting for the Heartbeat Thread to exit...")
+        dead_tid.join # raise the exception the dead thread failed with
+      end
+
+      _log.info("#{log_prefix} Heartbeat Thread gone. Restarting...")
+      @tid = start_heartbeat_thread
+    end
+
+    super
+  end
 end

config/settings.yml

@@ -1163,6 +1163,7 @@
         :memory_threshold: 2.gigabytes
         :queue_timeout: 20.minutes
         :restart_interval: 6.hours
+        :heartbeat_thread_shutdown_timeout: 10.seconds
     :schedule_worker:
       :container_entities_purge_interval: 1.day
       :binary_blob_purge_interval: 1.hour