Skip to content

[CI] add more logs and master worker info #76

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lihao-figma merged 1 commit into from
Nov 25, 2025
Merged

[CI] add more logs and master worker info #76

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 11 additions & 7 deletions ruby/lib/ci/queue/redis/base.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
# frozen_string_literal: true

module CI
module Queue
module Redis
Expand All @@ -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)
Expand Down Expand Up @@ -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
Expand All @@ -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

Expand All @@ -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
Expand Down
47 changes: 22 additions & 25 deletions ruby/lib/ci/queue/redis/worker.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
# frozen_string_literal: true

require 'ci/queue/static'
require 'set'

Expand Down Expand Up @@ -67,7 +68,7 @@ def master?
end

def idle?
!(@idle_since.nil?)
!@idle_since.nil?
end

def poll
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand All @@ -197,7 +198,7 @@ def release!
eval_script(
:release,
keys: [key('running'), key('worker', worker_id, 'queue'), key('owners')],
argv: [],
argv: []
)
nil
end
Expand All @@ -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
Expand All @@ -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

Expand All @@ -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
Expand All @@ -294,8 +293,6 @@ def register
redis.sadd(key('workers'), [worker_id])
end

private

def acquire_master_role?
return true if @master

Expand Down
53 changes: 26 additions & 27 deletions ruby/lib/ci/queue/strategy/suite_bin_packing.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
# frozen_string_literal: true

require_relative 'base'
require 'json'

Expand All @@ -20,31 +21,29 @@ 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

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) }

Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand Down
Loading