diff --git a/ruby/lib/ci/queue/redis/base.rb b/ruby/lib/ci/queue/redis/base.rb index 9cc8366..5a638e9 100644 --- a/ruby/lib/ci/queue/redis/base.rb +++ b/ruby/lib/ci/queue/redis/base.rb @@ -1,4 +1,5 @@ # frozen_string_literal: true + module CI module Queue module Redis @@ -8,7 +9,7 @@ class Base TEN_MINUTES = 60 * 10 CONNECTION_ERRORS = [ ::Redis::BaseConnectionError, - ::SocketError, # https://github.com/redis/redis-rb/pull/631 + ::SocketError # https://github.com/redis/redis-rb/pull/631 ].freeze def initialize(redis_url, config) @@ -54,12 +55,11 @@ def progress def wait_for_master(timeout: 120) return true if master? + (timeout * 10 + 1).to_i.times do - if queue_initialized? - return true - else - sleep 0.1 - end + return true if queue_initialized? + + sleep 0.1 end raise LostMaster, "The master worker is still `#{master_status}` after #{timeout} seconds waiting." end @@ -71,7 +71,7 @@ def workers_count def queue_initialized? @queue_initialized ||= begin status = master_status - status == 'ready' || status == 'finished' + %w[ready finished].include?(status) end end @@ -93,6 +93,10 @@ def max_test_failed? test_failed >= config.max_test_failed end + def master_worker_id + redis.get(key('master-worker-id')) + end + private attr_reader :redis, :redis_url diff --git a/ruby/lib/ci/queue/redis/worker.rb b/ruby/lib/ci/queue/redis/worker.rb index b553d32..2d5d99d 100644 --- a/ruby/lib/ci/queue/redis/worker.rb +++ b/ruby/lib/ci/queue/redis/worker.rb @@ -1,4 +1,5 @@ # frozen_string_literal: true + require 'ci/queue/static' require 'set' @@ -67,7 +68,7 @@ def master? end def idle? - !(@idle_since.nil?) + !@idle_since.nil? end def poll @@ -95,7 +96,7 @@ def poll puts " Running tests: #{running_tests.size}. #{running_tests}" puts " Owners: #{redis.hgetall(key('owners'))}" unless running_tests.empty? - puts " Checking if running tests are in processed set:" + puts ' Checking if running tests are in processed set:' running_tests.each do |test, _score| puts " #{test}: #{redis.sismember(key('processed'), test)}" end @@ -147,25 +148,25 @@ def acknowledge(test_or_id) # Accept either an object with .id or a string ID test_key = test_or_id.respond_to?(:id) ? test_or_id.id : test_or_id raise_on_mismatching_test(test_key) - + max_retries = 5 retry_count = 0 - + begin eval_script( :acknowledge, keys: [key('running'), key('processed'), key('owners')], - argv: [test_key], + argv: [test_key] ) == 1 rescue StandardError => e retry_count += 1 if retry_count < max_retries # Exponential backoff: 1s, 2s, ... - sleep(0.1 * (2 ** (retry_count - 1))) + sleep(0.1 * (2**(retry_count - 1))) retry else warn("Failed to acknowledge test #{test_key.inspect} after #{max_retries} retries: #{e.class} - #{e.message}. " \ - "Test remains in running set and may be picked up by reserve_lost after timeout.") + 'Test remains in running set and may be picked up by reserve_lost after timeout.') raise end end @@ -184,9 +185,9 @@ def requeue(test, offset: Redis.requeue_offset, skip_reservation_check: false) key('queue'), key('running'), key('worker', worker_id, 'queue'), - key('owners'), + key('owners') ], - argv: [config.max_requeues, global_max_requeues, test_key, offset], + argv: [config.max_requeues, global_max_requeues, test_key, offset] ) == 1 @reserved_test = test_key unless requeued || skip_reservation_check @@ -197,7 +198,7 @@ def release! eval_script( :release, keys: [key('running'), key('worker', worker_id, 'queue'), key('owners')], - argv: [], + argv: [] ) nil end @@ -215,20 +216,20 @@ def timeout end def raise_on_mismatching_test(test) - if @reserved_test == test - @reserved_test = nil - else + unless @reserved_test == test raise ReservationError, "Acknowledged #{test.inspect} but #{@reserved_test.inspect} was reserved" end + + @reserved_test = nil end def reserve if @reserved_test raise ReservationError, "#{@reserved_test.inspect} is already reserved. " \ - "You have to acknowledge it before you can reserve another one" + 'You have to acknowledge it before you can reserve another one' end - @reserved_test = (try_to_reserve_lost_test || try_to_reserve_test) + @reserved_test = try_to_reserve_lost_test || try_to_reserve_test end def try_to_reserve_test @@ -240,9 +241,9 @@ def try_to_reserve_test key('processed'), key('worker', worker_id, 'queue'), key('owners'), - key('test-group-timeout'), + key('test-group-timeout') ], - argv: [CI::Queue.time_now.to_f, 'true', config.timeout], + argv: [CI::Queue.time_now.to_f, 'true', config.timeout] ) end @@ -254,20 +255,18 @@ def try_to_reserve_lost_test key('completed'), key('worker', worker_id, 'queue'), key('owners'), - key('test-group-timeout'), + key('test-group-timeout') ], - argv: [CI::Queue.time_now.to_f, timeout, 'true', config.timeout], + argv: [CI::Queue.time_now.to_f, timeout, 'true', config.timeout] ) if lost_test.nil? && idle? puts "Worker #{worker_id} could not reserve a lost test while idle" - puts "Printing running tests:" + puts 'Printing running tests:' puts "#{redis.zrange(key('running'), 0, -1, withscores: true)}" end - if lost_test - build.record_warning(Warnings::RESERVED_LOST_TEST, test: lost_test, timeout: timeout) - end + build.record_warning(Warnings::RESERVED_LOST_TEST, test: lost_test, timeout: timeout) if lost_test lost_test end @@ -294,8 +293,6 @@ def register redis.sadd(key('workers'), [worker_id]) end - private - def acquire_master_role? return true if @master diff --git a/ruby/lib/ci/queue/strategy/suite_bin_packing.rb b/ruby/lib/ci/queue/strategy/suite_bin_packing.rb index a1531f3..b752518 100644 --- a/ruby/lib/ci/queue/strategy/suite_bin_packing.rb +++ b/ruby/lib/ci/queue/strategy/suite_bin_packing.rb @@ -1,4 +1,5 @@ # frozen_string_literal: true + require_relative 'base' require 'json' @@ -20,20 +21,18 @@ def load_timing_data(file_path) def initialize(config, redis: nil) super(config) - if redis - @moving_average = CI::Queue::Redis::TestDurationMovingAverages.new(redis) - end + @moving_average = CI::Queue::Redis::TestDurationMovingAverages.new(redis) if redis - if config&.timing_file - @timing_data = self.class.load_timing_data(config.timing_file) - else - @timing_data = {} - end + @timing_data = if config&.timing_file + self.class.load_timing_data(config.timing_file) + else + {} + end @max_duration = config&.suite_max_duration || 120_000 @fallback_duration = config&.timing_fallback_duration || 100.0 @buffer_percent = config&.suite_buffer_percent || 10 - + # Cache for test durations to avoid redundant lookups @duration_cache = {} end @@ -41,10 +40,10 @@ def initialize(config, redis: nil) def order_tests(tests, random: ::Random.new, redis: nil) # Clear duration cache for this ordering run @duration_cache.clear - + # Calculate dynamic max_duration based on total duration and parallelism dynamic_max_duration = calculate_dynamic_max_duration(tests) - + # Group tests by suite name suites = tests.group_by { |test| extract_suite_name(test.id) } @@ -55,7 +54,7 @@ def order_tests(tests, random: ::Random.new, redis: nil) create_chunks_for_suite( suite_name, suite_tests, - dynamic_max_duration, + dynamic_max_duration ) ) end @@ -82,19 +81,19 @@ def get_test_duration(test_id) return @duration_cache[test_id] if @duration_cache.key?(test_id) duration = if @moving_average - avg = @moving_average[test_id] - if avg - avg - elsif @timing_data.key?(test_id) - @timing_data[test_id] - else - @fallback_duration - end - elsif @timing_data.key?(test_id) - @timing_data[test_id] - else - @fallback_duration - end + avg = @moving_average[test_id] + if avg + avg + elsif @timing_data.key?(test_id) + @timing_data[test_id] + else + @fallback_duration + end + elsif @timing_data.key?(test_id) + @timing_data[test_id] + else + @fallback_duration + end # Cache the result @duration_cache[test_id] = duration @@ -104,7 +103,7 @@ def get_test_duration(test_id) def calculate_dynamic_max_duration(tests) # Get parallel job count from environment variable parallel_job_count = ENV['BUILDKITE_PARALLEL_JOB_COUNT']&.to_i - + puts "parallel_job_count: #{parallel_job_count}" # If no parallel job count, fall back to configured max_duration @@ -120,7 +119,7 @@ def calculate_dynamic_max_duration(tests) base_max_duration = total_duration.to_f / parallel_job_count puts "base_max_duration: #{base_max_duration}, @max_duration: #{@max_duration}" - + # Ensure we don't go below a minimum reasonable value # Use configured max_duration as a floor to prevent extremely small chunks [base_max_duration, @max_duration].max diff --git a/ruby/test/ci/queue/redis/dynamic_timeout_test.rb b/ruby/test/ci/queue/redis/dynamic_timeout_test.rb index fcabd61..110495a 100644 --- a/ruby/test/ci/queue/redis/dynamic_timeout_test.rb +++ b/ruby/test/ci/queue/redis/dynamic_timeout_test.rb @@ -1,4 +1,5 @@ # frozen_string_literal: true + require 'test_helper' class CI::Queue::DynamicTimeoutTest < Minitest::Test @@ -74,15 +75,16 @@ def test_chunk_timeout_scales_with_test_count def test_multiple_chunks_stored_with_different_timeouts tests = create_mock_tests([ - 'TestA#test_1', 'TestA#test_2', # 2 tests - 'TestB#test_1', 'TestB#test_2', 'TestB#test_3', # 3 tests - 'TestC#test_1', # 1 test - ]) + 'TestA#test_1', 'TestA#test_2', # 2 tests + 'TestB#test_1', 'TestB#test_2', 'TestB#test_3', # 3 tests + 'TestC#test_1' # 1 test + ]) chunks = [ CI::Queue::TestChunk.new('TestA:chunk_0', 'TestA', ['TestA#test_1', 'TestA#test_2'], 2000.0, test_count: 2), - CI::Queue::TestChunk.new('TestB:chunk_0', 'TestB', ['TestB#test_1', 'TestB#test_2', 'TestB#test_3'], 3000.0, test_count: 3), - CI::Queue::TestChunk.new('TestC:chunk_0', 'TestC', ['TestC#test_1'], 1000.0, test_count: 1), + CI::Queue::TestChunk.new('TestB:chunk_0', 'TestB', ['TestB#test_1', 'TestB#test_2', 'TestB#test_3'], 3000.0, + test_count: 3), + CI::Queue::TestChunk.new('TestC:chunk_0', 'TestC', ['TestC#test_1'], 1000.0, test_count: 1) ] @worker.stub(:reorder_tests, chunks) do @@ -126,15 +128,16 @@ def test_single_test_not_in_timeout_hash def test_mixed_chunks_and_tests_only_chunks_have_timeouts tests = create_mock_tests([ - 'TestA#test_1', 'TestA#test_2', - 'TestB#test_1', - 'TestC#test_1', 'TestC#test_2', 'TestC#test_3' - ]) + 'TestA#test_1', 'TestA#test_2', + 'TestB#test_1', + 'TestC#test_1', 'TestC#test_2', 'TestC#test_3' + ]) chunks = [ CI::Queue::TestChunk.new('TestA:chunk_0', 'TestA', ['TestA#test_1', 'TestA#test_2'], 2000.0, test_count: 2), tests[2], # Individual test TestB#test_1 - CI::Queue::TestChunk.new('TestC:chunk_0', 'TestC', ['TestC#test_1', 'TestC#test_2', 'TestC#test_3'], 3000.0, test_count: 3), + CI::Queue::TestChunk.new('TestC:chunk_0', 'TestC', ['TestC#test_1', 'TestC#test_2', 'TestC#test_3'], 3000.0, + test_count: 3) ] @worker.stub(:reorder_tests, chunks) do @@ -163,7 +166,7 @@ def test_reserve_test_passes_dynamic_deadline_flag 'build:42:processed', 'build:42:worker:1:queue', 'build:42:owners', - 'build:42:test-group-timeout', # 6th key for dynamic deadline + 'build:42:test-group-timeout' # 6th key for dynamic deadline ] @worker.stub(:eval_script, proc { |script, keys:, argv:| @@ -191,7 +194,7 @@ def test_reserve_lost_test_passes_dynamic_deadline_flag 'build:42:completed', 'build:42:worker:1:queue', 'build:42:owners', - 'build:42:test-group-timeout', # 5th key for dynamic deadline + 'build:42:test-group-timeout' # 5th key for dynamic deadline ] @worker.stub(:eval_script, proc { |script, keys:, argv:| @@ -241,7 +244,7 @@ def test_chunk_not_marked_lost_before_dynamic_timeout worker2 = CI::Queue::Redis.new(@redis_url, worker2_config) lost_test = worker2.send(:try_to_reserve_lost_test) - assert_nil lost_test, "Chunk should not be marked as lost before dynamic timeout" + assert_nil lost_test, 'Chunk should not be marked as lost before dynamic timeout' end def test_single_test_marked_lost_after_default_timeout @@ -249,7 +252,7 @@ def test_single_test_marked_lost_after_default_timeout config = CI::Queue::Configuration.new( build_id: 'single-timeout-test', worker_id: '1', - timeout: 0.5, # 0.5 seconds + timeout: 0.5 # 0.5 seconds ) worker1 = CI::Queue::Redis.new(@redis_url, config) @@ -274,7 +277,7 @@ def test_single_test_marked_lost_after_default_timeout worker2 = CI::Queue::Redis.new(@redis_url, worker2_config) lost_test = worker2.send(:try_to_reserve_lost_test) - assert_equal 'TestA#test_1', lost_test, "Single test should be marked as lost after default timeout" + assert_equal 'TestA#test_1', lost_test, 'Single test should be marked as lost after default timeout' end def test_batching_with_many_chunks