Merge pull request #188 from Shopify/cbruckmayer/worker-crash

Exit early if all workers die

Author: ChrisBr

ruby/lib/ci/queue/configuration.rb

@@ -8,7 +8,7 @@ class Configuration
       attr_accessor :max_test_failed, :redis_ttl
       attr_reader :circuit_breakers
       attr_writer :seed, :build_id
-      attr_writer :queue_init_timeout
+      attr_writer :queue_init_timeout, :report_timeout, :inactive_workers_timeout
 
       class << self
         def from_env(env)
@@ -35,7 +35,7 @@ def initialize(
         namespace: nil, seed: nil, flaky_tests: [], statsd_endpoint: nil, max_consecutive_failures: nil,
         grind_count: nil, max_duration: nil, failure_file: nil, max_test_duration: nil,
         max_test_duration_percentile: 0.5, track_test_duration: false, max_test_failed: nil,
-        queue_init_timeout: nil, redis_ttl: 8 * 60 * 60
+        queue_init_timeout: nil, redis_ttl: 8 * 60 * 60, report_timeout: nil, inactive_workers_timeout: nil
       )
         @build_id = build_id
         @circuit_breakers = [CircuitBreaker::Disabled]
@@ -57,12 +57,22 @@ def initialize(
         self.max_consecutive_failures = max_consecutive_failures
         self.max_duration = max_duration
         @redis_ttl = redis_ttl
+        @report_timeout = report_timeout
+        @inactive_workers_timeout = inactive_workers_timeout
       end
 
       def queue_init_timeout
         @queue_init_timeout || timeout
       end
 
+      def report_timeout
+        @report_timeout || timeout
+      end
+
+      def inactive_workers_timeout
+        @inactive_workers_timeout || timeout
+      end
+
       def max_consecutive_failures=(max)
         if max
           @circuit_breakers << CircuitBreaker::MaxConsecutiveFailures.new(max_consecutive_failures: max)

ruby/lib/ci/queue/redis/supervisor.rb

@@ -21,17 +21,33 @@ def wait_for_workers
 
           yield if block_given?
 
-          time_left = config.timeout
-          until exhausted? || time_left <= 0 || max_test_failed?
-            sleep 1
+          time_left = config.report_timeout
+          time_left_with_no_workers = config.inactive_workers_timeout
+          until exhausted? || time_left <= 0 || max_test_failed? || time_left_with_no_workers <= 0
             time_left -= 1
+            sleep 1
+
+            if active_workers?
+              time_left_with_no_workers = config.inactive_workers_timeout
+            else
+              time_left_with_no_workers -= 1
+            end
 
             yield if block_given?
           end
+
+          puts "Aborting, it seems all workers died." if time_left_with_no_workers <= 0
           exhausted?
         rescue CI::Queue::Redis::LostMaster
           false
         end
+
+        private
+
+        def active_workers?
+          # if there are running jobs we assume there are still agents active
+          redis.zrangebyscore(key('running'), Time.now.to_f - config.timeout, "+inf", limit: [0,1]).count > 0
+        end
       end
     end
   end

ruby/lib/minitest/queue/runner.rb

@@ -370,6 +370,24 @@ def parser
             queue_config.timeout = timeout
           end
 
+          help = <<~EOS
+            Specify a timeout after which the report command will fail if not all tests have been processed.
+            Defaults to the value set for --timeout.
+          EOS
+          opts.separator ""
+          opts.on('--report-timeout TIMEOUT', Float, help) do |timeout|
+            queue_config.report_timeout = timeout
+          end
+
+          help = <<~EOS
+            Specify a timeout after the report will fail if all workers are inactive (e.g. died).
+            Defaults to the value set for --timeout.
+          EOS
+          opts.separator ""
+          opts.on('--inactive-workers-timeout TIMEOUT', Float, help) do |timeout|
+            queue_config.inactive_workers_timeout = timeout
+          end
+
           help = <<~EOS
             Specify a timeout to elect the leader and populate the queue.
             Defaults to the value set for --timeout.

ruby/test/ci/queue/configuration_test.rb

@@ -112,5 +112,21 @@ def test_queue_init_timeout_set
 
       assert_equal 45, config.queue_init_timeout
     end
+
+    def test_report_timeout_unset_timeout_set
+      config = Configuration.from_env({})
+      config.timeout = 120
+
+      assert_equal config.timeout, config.report_timeout
+    end
+
+    def test_report_timeout_set
+      config = Configuration.from_env({})
+      config.report_timeout = 45
+      config.timeout = 120
+
+      assert_equal 45, config.report_timeout
+    end
+
   end
 end

ruby/test/ci/queue/redis_supervisor_test.rb

@@ -39,6 +39,18 @@ def test_wait_for_workers
     assert_equal true, workers_done
   end
 
+  def test_wait_for_workers_timeout
+    @supervisor = supervisor(timeout: 10, queue_init_timeout: 0.1)
+    io = nil
+    thread = Thread.start do
+      io = capture_io { @supervisor.wait_for_workers }
+    end
+    thread.wakeup
+    worker(1)
+    thread.join
+    assert_includes io, "Aborting, it seems all workers died.\n"
+  end
+
   def test_num_workers
     assert_equal 0, @supervisor.workers_count
     worker(1)
@@ -58,10 +70,14 @@ def worker(id)
     ).populate(SharedQueueAssertions::TEST_LIST)
   end
 
-  def supervisor
+  def supervisor(timeout: 30, queue_init_timeout: nil)
     CI::Queue::Redis::Supervisor.new(
       @redis_url,
-      CI::Queue::Configuration.new(build_id: '42'),
+      CI::Queue::Configuration.new(
+        build_id: '42',
+        timeout: timeout,
+        queue_init_timeout: queue_init_timeout
+      ),
     )
   end
 end