Cleanup

Author: danielahrnsbrak

redis/_entry_helpers.lua

@@ -0,0 +1,9 @@
+local function test_id_from_entry(value, delimiter)
+  if delimiter then
+    local pos = string.find(value, delimiter, 1, true)
+    if pos then
+      return string.sub(value, 1, pos - 1)
+    end
+  end
+  return value
+end

redis/heartbeat.lua

@@ -1,3 +1,5 @@
+-- @include _entry_helpers
+
 local zset_key = KEYS[1]
 local processed_key = KEYS[2]
 local owners_key = KEYS[3]
@@ -7,17 +9,7 @@ local current_time = ARGV[1]
 local entry = ARGV[2]
 local entry_delimiter = ARGV[3]
 
-local function test_id_from_entry(value)
-  if entry_delimiter then
-    local pos = string.find(value, entry_delimiter, 1, true)
-    if pos then
-      return string.sub(value, 1, pos - 1)
-    end
-  end
-  return value
-end
-
-local test_id = test_id_from_entry(entry)
+local test_id = test_id_from_entry(entry, entry_delimiter)
 
 -- already processed, we do not need to bump the timestamp
 if redis.call('sismember', processed_key, test_id) == 1 then

redis/reserve_lost.lua

@@ -1,3 +1,5 @@
+-- @include _entry_helpers
+
 local zset_key = KEYS[1]
 local processed_key = KEYS[2]
 local worker_queue_key = KEYS[3]
@@ -7,19 +9,9 @@ local current_time = ARGV[1]
 local timeout = ARGV[2]
 local entry_delimiter = ARGV[3]
 
-local function test_id_from_entry(entry)
-  if entry_delimiter then
-    local pos = string.find(entry, entry_delimiter, 1, true)
-    if pos then
-      return string.sub(entry, 1, pos - 1)
-    end
-  end
-  return entry
-end
-
 local lost_tests = redis.call('zrangebyscore', zset_key, 0, current_time - timeout)
 for _, test in ipairs(lost_tests) do
-  local test_id = test_id_from_entry(test)
+  local test_id = test_id_from_entry(test, entry_delimiter)
   if redis.call('sismember', processed_key, test_id) == 0 then
     redis.call('zadd', zset_key, current_time, test)
     redis.call('lpush', worker_queue_key, test)

ruby/lib/ci/queue.rb

@@ -14,6 +14,9 @@
 require 'ci/queue/file'
 require 'ci/queue/grind'
 require 'ci/queue/bisect'
+require 'ci/queue/queue_entry'
+require 'ci/queue/class_resolver'
+require 'ci/queue/file_loader'
 
 module CI
   module Queue
@@ -83,6 +86,3 @@ def from_uri(url, config)
   end
 end
 
-require 'ci/queue/queue_entry'
-require 'ci/queue/class_resolver'
-require 'ci/queue/file_loader'

ruby/lib/ci/queue/configuration.rb

@@ -55,8 +55,7 @@ def initialize(
         max_test_duration_percentile: 0.5, track_test_duration: false, max_test_failed: nil,
         queue_init_timeout: nil, redis_ttl: 8 * 60 * 60, report_timeout: nil, inactive_workers_timeout: nil,
         export_flaky_tests_file: nil, warnings_file: nil, debug_log: nil, max_missed_heartbeat_seconds: nil,
-        lazy_load: false, lazy_load_stream_batch_size: nil, lazy_load_streaming_timeout: nil, lazy_load_test_helpers: nil,
-        stream_batch_size: nil, streaming_timeout: nil, test_helpers: nil)
+        lazy_load: false, lazy_load_stream_batch_size: nil, lazy_load_streaming_timeout: nil, lazy_load_test_helpers: nil)
         @build_id = build_id
         @circuit_breakers = [CircuitBreaker::Disabled]
         @failure_file = failure_file
@@ -84,9 +83,9 @@ def initialize(
         @debug_log = debug_log
         @max_missed_heartbeat_seconds = max_missed_heartbeat_seconds
         @lazy_load = lazy_load
-        @lazy_load_stream_batch_size = lazy_load_stream_batch_size || stream_batch_size || 10_000
-        @lazy_load_streaming_timeout = lazy_load_streaming_timeout || streaming_timeout || 300
-        @lazy_load_test_helpers = lazy_load_test_helpers || test_helpers
+        @lazy_load_stream_batch_size = lazy_load_stream_batch_size || 10_000
+        @lazy_load_streaming_timeout = lazy_load_streaming_timeout || 300
+        @lazy_load_test_helpers = lazy_load_test_helpers
       end
 
       def lazy_load_test_helper_paths

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

@@ -246,9 +246,21 @@ def load_script(script)
         end
 
         def read_script(name)
-          ::File.read(::File.join(CI::Queue::DEV_SCRIPTS_ROOT, "#{name}.lua"))
+          resolve_lua_includes(
+            ::File.read(::File.join(CI::Queue::DEV_SCRIPTS_ROOT, "#{name}.lua")),
+            CI::Queue::DEV_SCRIPTS_ROOT,
+          )
         rescue SystemCallError
-          ::File.read(::File.join(CI::Queue::RELEASE_SCRIPTS_ROOT, "#{name}.lua"))
+          resolve_lua_includes(
+            ::File.read(::File.join(CI::Queue::RELEASE_SCRIPTS_ROOT, "#{name}.lua")),
+            CI::Queue::RELEASE_SCRIPTS_ROOT,
+          )
+        end
+
+        def resolve_lua_includes(script, root)
+          script.gsub(/^-- @include (\S+)$/) do
+            ::File.read(::File.join(root, "#{$1}.lua"))
+          end
         end
 
         class HeartbeatProcess

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

@@ -57,9 +57,21 @@ def load_script(script)
         end
 
         def read_script(name)
-          ::File.read(::File.join(DEV_SCRIPTS_ROOT, "#{name}.lua"))
+          resolve_lua_includes(
+            ::File.read(::File.join(DEV_SCRIPTS_ROOT, "#{name}.lua")),
+            DEV_SCRIPTS_ROOT,
+          )
         rescue SystemCallError
-          ::File.read(::File.join(RELEASE_SCRIPTS_ROOT, "#{name}.lua"))
+          resolve_lua_includes(
+            ::File.read(::File.join(RELEASE_SCRIPTS_ROOT, "#{name}.lua")),
+            RELEASE_SCRIPTS_ROOT,
+          )
+        end
+
+        def resolve_lua_includes(script, root)
+          script.gsub(/^-- @include (\S+)$/) do
+            ::File.read(::File.join(root, "#{$1}.lua"))
+          end
         end
 
         HEADER = 'L'

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

@@ -168,6 +168,12 @@ def file_loader
           @file_loader ||= CI::Queue::FileLoader.new
         end
 
+        def worker_queue_length
+          redis.llen(key('worker', worker_id, 'queue'))
+        rescue *CONNECTION_ERRORS
+          nil
+        end
+
         def report_worker_error(error)
           build.report_worker_error(error)
         end

ruby/lib/minitest/queue.rb

@@ -2,6 +2,7 @@
 require 'shellwords'
 require 'minitest'
 require 'minitest/reporters'
+require 'concurrent/map'
 
 require 'minitest/queue/failure_formatter'
 require 'minitest/queue/error_report'
@@ -254,7 +255,7 @@ def store_worker_profile(queue)
         first_test = queue.respond_to?(:first_reserve_at) ? queue.first_reserve_at : nil
         profile['time_to_first_test'] = (first_test - run_start).round(2) if first_test
 
-        tests_run = queue.rescue_connection_errors { queue.send(:redis).llen(queue.send(:key, 'worker', config.worker_id, 'queue')) }
+        tests_run = queue.rescue_connection_errors { queue.worker_queue_length } if queue.respond_to?(:worker_queue_length)
         profile['tests_run'] = tests_run.to_i if tests_run
 
         load_tests_duration = Minitest::Queue::Runner.load_tests_duration
@@ -393,7 +394,7 @@ def <=>(other)
         id <=> other.id
       end
 
-      RUNNABLE_METHODS_TRIGGERED = {} # :nodoc:
+      RUNNABLE_METHODS_TRIGGERED = Concurrent::Map.new # :nodoc:
 
       def runnable
         @runnable ||= begin