ruby: Measure worker slack and let a sample of workers exit idle

ruby/lib/ci/queue/configuration.rb

@@ -16,6 +16,7 @@ class Configuration
       attr_accessor :write_duration_averages
       attr_accessor :heartbeat_grace_period, :heartbeat_interval
       attr_accessor :retry_selection
+      attr_accessor :idle_exit_probability, :idle_exit_grace
       attr_reader :circuit_breakers
       attr_writer :seed, :build_id
       attr_writer :queue_init_timeout, :report_timeout, :inactive_workers_timeout
@@ -31,6 +32,8 @@ def from_env(env)
             redis_ttl: env['CI_QUEUE_REDIS_TTL']&.to_i ||  8 * 60 * 60,
             known_flaky_tests: load_known_flaky_tests(env['CI_QUEUE_KNOWN_FLAKY_TESTS']),
             branch: env['BUILDKITE_BRANCH'],
+            idle_exit_probability: env['CI_QUEUE_IDLE_EXIT_PROBABILITY']&.to_f || 0.0,
+            idle_exit_grace: env['CI_QUEUE_IDLE_EXIT_GRACE']&.to_f || 30.0,
           )
         end
 
@@ -68,7 +71,9 @@ def initialize(
         timing_redis_url: nil,
         heartbeat_grace_period: 30,
         heartbeat_interval: 10,
-        retry_selection: :failed_tests
+        retry_selection: :failed_tests,
+        idle_exit_probability: 0.0,
+        idle_exit_grace: 30.0
       )
         @build_id = build_id
         @circuit_breakers = [CircuitBreaker::Disabled]
@@ -108,6 +113,8 @@ def initialize(
         @heartbeat_grace_period = heartbeat_grace_period
         @heartbeat_interval = heartbeat_interval
         @retry_selection = retry_selection
+        @idle_exit_probability = idle_exit_probability
+        @idle_exit_grace = idle_exit_grace
       end
 
       def queue_init_timeout

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

@@ -24,10 +24,30 @@ def initialize(redis, config)
           @reserved_tests = Concurrent::Set.new
           @shutdown_required = false
           @idle_since = nil
+          @last_test_finished_at = nil
+          # A worker with nothing left to reserve normally stays online until the whole
+          # build drains, so it can pick up a test that times out or gets requeued. Only
+          # a sample needs to do that. Random.new rather than Kernel#rand: the global RNG
+          # is seeded from --seed, which is identical across workers, so every worker
+          # would draw the same number.
+          @waits_for_requeues = Random.new.rand > config.idle_exit_probability
           super(redis, config)
         end
 
         attr_accessor :idle_since
+        attr_reader :last_test_finished_at
+
+        def waits_for_requeues?
+          @waits_for_requeues
+        end
+
+        # Time between the last test finishing and the worker leaving the queue: capacity
+        # the build paid for and did not use. Nil when the worker never reserved a test.
+        def slack_duration
+          return nil if @last_test_finished_at.nil?
+
+          CI::Queue.time_now - @last_test_finished_at
+        end
 
         def distributed?
           true
@@ -75,6 +95,15 @@ def idle?
           !@idle_since.nil?
         end
 
+        # Workers that did not draw the requeue duty leave once they have been idle for
+        # the grace period, instead of waiting for the whole build to drain.
+        def idle_exit?
+          return false if @waits_for_requeues
+          return false if @idle_since.nil?
+
+          CI::Queue.time_now - @idle_since >= config.idle_exit_grace
+        end
+
         def poll
           wait_for_master
           if master?
@@ -83,26 +112,28 @@ def poll
             master_id = master_worker_id
             warn "Worker #{worker_id} saw master worker: #{master_id}" if master_id
           end
-          idle_since = nil
+          @idle_since = nil
           idle_state_printed = false
           attempt = 0
-          until shutdown_required? || config.circuit_breakers.any?(&:open?) || exhausted? || max_test_failed?
+          until shutdown_required? || config.circuit_breakers.any?(&:open?) || exhausted? ||
+                max_test_failed? || idle_exit?
             if id = reserve
               attempt = 0
-              idle_since = nil
+              @idle_since = nil
               executable = resolve_executable(id)
 
               if executable
                 with_heartbeat(id) do
                   yield executable
                 end
+                @last_test_finished_at = CI::Queue.time_now
               else
                 warn("Warning: Could not resolve executable for ID #{id.inspect}. Acknowledging to remove from queue.")
                 acknowledge(id)
               end
             else
-              idle_since ||= CI::Queue.time_now
-              if CI::Queue.time_now - idle_since > 120 && !idle_state_printed
+              @idle_since ||= CI::Queue.time_now
+              if CI::Queue.time_now - @idle_since > 120 && !idle_state_printed
                 puts "Worker #{worker_id} has been idle for 120 seconds. Printing global state..."
                 running_tests = redis.zrange(key('running'), 0, -1, withscores: true)
                 puts "  Processed tests: #{redis.scard(key('processed'))}"

ruby/lib/minitest/queue.rb

@@ -9,6 +9,7 @@
 require 'minitest/queue/build_status_recorder'
 require 'minitest/queue/build_status_reporter'
 require 'minitest/queue/order_reporter'
+require 'minitest/queue/statsd'
 require 'minitest/queue/junit_reporter'
 require 'minitest/queue/test_data_reporter'
 require 'minitest/queue/grind_recorder'
@@ -232,10 +233,34 @@ def run_from_queue(reporter, *)
           run_single_test(executable, reporter)
         end
       end
+      report_worker_slack
     end
 
     private
 
+    # How long this worker sat idle before leaving. Emitted here rather than from the
+    # queue because ci/queue stays framework agnostic, and Statsd lives under minitest/.
+    def report_worker_slack
+      return unless queue.respond_to?(:slack_duration)
+
+      slack = queue.slack_duration
+      return if slack.nil?
+
+      endpoint = queue.config.statsd_endpoint
+      return if endpoint.nil?
+
+      statsd = Minitest::Queue::Statsd.new(
+        addr: endpoint,
+        namespace: 'minitests.queue',
+        default_tags: ["slug:#{ENV['BUILDKITE_PROJECT_SLUG']}"],
+      )
+      statsd.measure(
+        'worker.slack',
+        slack * 1000,
+        tags: ["waits_for_requeues:#{queue.waits_for_requeues?}"],
+      )
+    end
+
     def run_chunk(chunk, reporter)
       @in_chunk_context = true
 

ruby/test/ci/queue/configuration_test.rb

@@ -17,6 +17,21 @@ def test_cirleci_defaults
       assert_equal 'faa647bbb8168a77cf338e7488c3f8445c3e6554', config.seed
     end
 
+    def test_idle_exit_defaults_keep_every_worker_online
+      config = Configuration.from_env({})
+      assert_equal 0.0, config.idle_exit_probability
+      assert_equal 30.0, config.idle_exit_grace
+    end
+
+    def test_idle_exit_read_from_env
+      config = Configuration.from_env(
+        'CI_QUEUE_IDLE_EXIT_PROBABILITY' => '0.75',
+        'CI_QUEUE_IDLE_EXIT_GRACE' => '5',
+      )
+      assert_equal 0.75, config.idle_exit_probability
+      assert_equal 5.0, config.idle_exit_grace
+    end
+
     def test_heroku_ci_defaults
       config = Configuration.from_env(
         'HEROKU_TEST_RUN_ID' => 'YouAreAnAmazingPersonAndIBelieveYouCanDoIt',

ruby/test/ci/queue/redis/worker_idle_exit_test.rb

@@ -0,0 +1,49 @@
+# frozen_string_literal: true
+require 'test_helper'
+
+module CI::Queue::Redis
+  class WorkerIdleExitTest < Minitest::Test
+    REDIS_URL = 'redis://localhost:6379/0'
+
+    def test_worker_that_drew_requeue_duty_never_exits_on_idle
+      worker = build_worker(idle_exit_probability: 0.0, idle_exit_grace: 0)
+      assert worker.waits_for_requeues?
+
+      worker.idle_since = CI::Queue.time_now - 3600
+      refute worker.idle_exit?
+    end
+
+    def test_worker_without_requeue_duty_waits_out_the_grace_period
+      worker = build_worker(idle_exit_probability: 1.0, idle_exit_grace: 30)
+      refute worker.waits_for_requeues?
+
+      worker.idle_since = CI::Queue.time_now - 10
+      refute worker.idle_exit?
+
+      worker.idle_since = CI::Queue.time_now - 31
+      assert worker.idle_exit?
+    end
+
+    def test_a_busy_worker_never_exits_on_idle
+      worker = build_worker(idle_exit_probability: 1.0, idle_exit_grace: 0)
+      assert_nil worker.idle_since
+      refute worker.idle_exit?
+    end
+
+    def test_slack_duration_is_unknown_until_a_test_finishes
+      assert_nil build_worker.slack_duration
+    end
+
+    def test_slack_duration_measures_from_the_last_finished_test
+      worker = build_worker
+      worker.instance_variable_set(:@last_test_finished_at, CI::Queue.time_now - 12)
+      assert_in_delta 12, worker.slack_duration, 1
+    end
+
+    private
+
+    def build_worker(**options)
+      Worker.new(REDIS_URL, CI::Queue::Configuration.new(**options))
+    end
+  end
+end