Check for timed out active tasks

app/models/miq_schedule_worker/jobs.rb

@@ -172,6 +172,10 @@ def generate_chargeback_for_service(args = {})
     queue_work(:class_name => "Service", :method_name => "queue_chargeback_reports", :zone => nil, :args => args)
   end
 
+  def check_for_timed_out_active_tasks
+    queue_work(:class_name => "MiqTask", :method_name => "update_status_for_timed_out_active_tasks", :zone => nil)
+  end
+
   private
 
   def queue_work(options)

app/models/miq_schedule_worker/runner.rb

@@ -234,11 +234,22 @@ def schedules_for_scheduler_role
       end
     end
 
-    # run run chargeback generation every day at specific time
+    # run chargeback generation every day at specific time
     schedule_chargeback_report_for_service_daily
+
+    schedule_check_for_task_timeout
+
     @schedules[:scheduler]
   end
 
+  def schedule_check_for_task_timeout
+    every = worker_settings[:task_timeout_check_frequency]
+    scheduler = scheduler_for(:scheduler)
+    scheduler.schedule_every(every, :first_at => Time.current + 1.minute) do
+      enqueue :check_for_timed_out_active_tasks
+    end
+  end
+
   def schedule_chargeback_report_for_service_daily
     every = worker_settings[:chargeback_generation_interval]
     at = worker_settings[:chargeback_generation_time_utc]

app/models/miq_task.rb

@@ -33,6 +33,17 @@ class MiqTask < ApplicationRecord
     t.grouping(Arel::Nodes::Case.new(t[:state]).when(STATE_FINISHED).then(t[:status]).else(t[:state]))
   end)
 
+  scope :active,            -> { where(:state => STATE_ACTIVE) }
+  scope :no_associated_job, -> { where.not("id IN (SELECT miq_task_id from jobs)") }
+  scope :timed_out,         -> { where("updated_on < ?", Time.now.utc - ::Settings.task.active_task_timeout.to_i_with_method) }
+
+  def self.update_status_for_timed_out_active_tasks
+    MiqTask.active.timed_out.no_associated_job.find_each do |task|
+      task.update_status(STATE_FINISHED, STATUS_ERROR,
+                         "Task [#{task.id}] timed out - not active for more than #{::Settings.task.active_task_timeout}")
+    end
+  end
+
   def active?
     ![STATE_QUEUED, STATE_FINISHED].include?(state)
   end

config/settings.yml

@@ -1129,6 +1129,8 @@
   :max_parallel_scans_per_host: 1
   :max_qitems_per_scan_request: 0
   :watchdog_interval: 1.minute
+:task:
+  :active_task_timeout: 6.hours
 :ui:
   :mark_translated_strings: false
 :webservices:
@@ -1299,6 +1301,7 @@
       :session_timeout_interval: 30.seconds
       :storage_file_collection_interval: 1.days
       :storage_file_collection_time_utc: 21600
+      :task_timeout_check_frequency: 1.hour
       :vm_retired_interval: 10.minutes
       :yum_update_check: 12.hours
     :ui_worker:

spec/models/miq_schedule_worker/jobs_spec.rb

@@ -49,4 +49,15 @@
       )
     end
   end
+
+  describe "#check_for_timed_out_active_tasks" do
+    it "enqueues update_status_for_timed_out_active_tasks" do
+      allow(MiqServer).to receive(:my_zone)
+      described_class.new.check_for_timed_out_active_tasks
+      expect(MiqQueue.first).to have_attributes(
+        :class_name  => "MiqTask",
+        :method_name => "update_status_for_timed_out_active_tasks"
+      )
+    end
+  end
 end

spec/models/miq_schedule_worker/runner_spec.rb

@@ -460,15 +460,18 @@
           end
         end
 
-        context "Chargeback reports for Services" do
+        context "schedule for 'scheduler' role" do
           before do
             allow(@schedule_worker).to receive(:heartbeat)
             @schedule_worker.instance_variable_set(:@active_roles, ["scheduler"])
-            allow(@schedule_worker).to receive(:worker_settings).and_return(:chargeback_generation_interval => 1.day)
             @schedule_worker.instance_variable_set(:@schedules, :scheduler => [])
           end
 
           describe "#schedule_chargeback_report_for_service_daily" do
+            before do
+              allow(@schedule_worker).to receive(:worker_settings).and_return(:chargeback_generation_interval => 1.day)
+            end
+
             it "queues daily generation of Chargeback report for each service" do
               job = @schedule_worker.schedule_chargeback_report_for_service_daily[0]
               expect(job).to be_kind_of(Rufus::Scheduler::EveryJob)
@@ -484,6 +487,24 @@
               job.unschedule
             end
           end
+
+          describe "#schedule_check_for_task_timeout" do
+            let(:interval) { 1.hour }
+            before do
+              allow(@schedule_worker).to receive(:worker_settings).and_return(:task_timeout_check_frequency => interval)
+            end
+
+            it "queues check for timed out tasks" do
+              job = @schedule_worker.schedule_check_for_task_timeout[0]
+              job.call
+              @schedule_worker.do_work
+              queue = MiqQueue.first
+              expect(queue.method_name).to eq "update_status_for_timed_out_active_tasks"
+              expect(queue.class_name).to eq "MiqTask"
+              MiqQueue.delete_all
+              job.unschedule
+            end
+          end
         end
       end
 

spec/models/miq_task_spec.rb

@@ -381,4 +381,68 @@
       expect(miq_task.miq_server.id).to eq server.id
     end
   end
+
+  describe ".update_status_for_timed_out_active_tasks" do
+    let(:timeout) { "1.hour" }
+    before do
+      stub_settings(:task => {:active_task_timeout => timeout})
+    end
+
+    context "task does not linked to job" do
+      let(:miq_task) { FactoryGirl.create(:miq_task_plain) }
+
+      context "task is active" do
+        before do
+          miq_task.update_attributes(:state => MiqTask::STATE_ACTIVE)
+        end
+
+        it "updates status to 'Error' for timed out task" do
+          miq_task.update_attributes(:updated_on => miq_task.updated_on - timeout.to_i_with_method)
+          expect(miq_task.status).not_to eq MiqTask::STATUS_ERROR
+          MiqTask.update_status_for_timed_out_active_tasks
+          miq_task.reload
+          expect(miq_task.status).to eq MiqTask::STATUS_ERROR
+        end
+
+        it "does not update status if task not timed out" do
+          MiqTask.update_status_for_timed_out_active_tasks
+          miq_task.reload
+          expect(miq_task.status).not_to eq MiqTask::STATUS_ERROR
+        end
+      end
+
+      context "task is not active" do
+        it "does not update status to 'Error' if task state is 'Finished'" do
+          miq_task.update_attributes(:state      => MiqTask::STATE_FINISHED,
+                                     :updated_on => miq_task.updated_on - timeout.to_i_with_method)
+          MiqTask.update_status_for_timed_out_active_tasks
+          miq_task.reload
+          expect(miq_task.status).not_to eq MiqTask::STATUS_ERROR
+        end
+
+        it "does not update status to 'Error' if task state is 'Queued'" do
+          miq_task.update_attributes(:state      => MiqTask::STATE_QUEUED,
+                                     :updated_on => miq_task.updated_on - timeout.to_i_with_method)
+          MiqTask.update_status_for_timed_out_active_tasks
+          miq_task.reload
+          expect(miq_task.status).not_to eq MiqTask::STATUS_ERROR
+        end
+      end
+    end
+
+    context "task linked to job" do
+      let(:miq_task) do
+        job = Job.create_job("VmScan")
+        job.miq_task
+      end
+
+      it "does not update status to 'Error'" do
+        miq_task.update_attributes(:state      => MiqTask::STATE_ACTIVE,
+                                   :updated_on => miq_task.updated_on - timeout.to_i_with_method)
+        MiqTask.update_status_for_timed_out_active_tasks
+        miq_task.reload
+        expect(miq_task.status).not_to eq MiqTask::STATUS_ERROR
+      end
+    end
+  end
 end