diff --git a/ruby/lib/ci/queue/configuration.rb b/ruby/lib/ci/queue/configuration.rb index 33afca6..b3770be 100644 --- a/ruby/lib/ci/queue/configuration.rb +++ b/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 diff --git a/ruby/lib/ci/queue/redis/worker.rb b/ruby/lib/ci/queue/redis/worker.rb index 7338a45..6c48555 100644 --- a/ruby/lib/ci/queue/redis/worker.rb +++ b/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'))}" diff --git a/ruby/lib/minitest/queue.rb b/ruby/lib/minitest/queue.rb index 9b0770c..daf65ac 100644 --- a/ruby/lib/minitest/queue.rb +++ b/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 diff --git a/ruby/test/ci/queue/configuration_test.rb b/ruby/test/ci/queue/configuration_test.rb index a9b66d3..ce80d8b 100644 --- a/ruby/test/ci/queue/configuration_test.rb +++ b/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', diff --git a/ruby/test/ci/queue/redis/worker_idle_exit_test.rb b/ruby/test/ci/queue/redis/worker_idle_exit_test.rb new file mode 100644 index 0000000..764924c --- /dev/null +++ b/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