Merge pull request #427 from Shopify/use-rails-logger

Use the Rails application logger

Author: etiennebarrie

CHANGELOG.md

@@ -1,6 +1,10 @@
 ### Main (unreleased)
 
-- [367](https://github.com/Shopify/job-iteration/pull/367) - Iteration can use multiple Active Job backends simultaneously by inferring the interruption adapter from the job's `queue_adapter_name`. Iteration will now raise `JobIteration::Integrations::LoadError` if no interruption adapter is found for the job's queue adapter, instead of never interrupting the job. `JobIteration.interruption_adapter` and `.load_integrations` have been removed. `JobIteration::Integrations.register` has been added.
+## v1.4.1 (Sep 5, 2023)
+
+### Bug fixes
+
+- [427](https://github.com/Shopify/job-iteration/pull/427) - Use the Rails application logger. Changes from [338](https://github.com/Shopify/job-iteration/pull/338) resulted in logging to the original value of ActiveJob.logger, not the one configured by the Rails application.
 
 ## v1.4.0 (Aug 23, 2023)
 

Gemfile.lock

@@ -7,7 +7,7 @@ GIT
 PATH
   remote: .
   specs:
-    job-iteration (1.4.0)
+    job-iteration (1.4.1)
       activejob (>= 5.2)
 
 GEM

README.md

@@ -152,7 +152,6 @@ end
 ```
 
 Iteration hooks into Sidekiq and Resque out of the box to support graceful interruption. No extra configuration is required.
-Adapters for other Active Job backends can be registered with `JobIteration::Integrations.register("my_queue_adapter_name", object)`, where `object` must implement the `call` method returning `true` if the job must be interrupted and `false` otherwise.
 
 ## Guides
 
@@ -184,7 +183,7 @@ There a few configuration assumptions that are required for Iteration to work wi
 
 **Why can't I just iterate in `#perform` method and do whatever I want?** You can, but then your job has to comply with a long list of requirements, such as the ones above. This creates leaky abstractions more easily, when instead we can expose a more powerful abstraction for developers--without exposing the underlying infrastructure.
 
-**What happens when my job is interrupted?** A checkpoint will be persisted after the current `each_iteration`, and the job will be re-enqueued. Once it's popped off the queue, the worker will work off from the next iteration.
+**What happens when my job is interrupted?** A checkpoint will be persisted to Redis after the current `each_iteration`, and the job will be re-enqueued. Once it's popped off the queue, the worker will work off from the next iteration.
 
 **What happens with retries?** An interruption of a job does not count as a retry. The iteration of job that caused the job to fail will be retried and progress will continue from there on.
 

lib/job-iteration.rb

@@ -2,17 +2,24 @@
 
 require "active_job"
 require_relative "./job-iteration/version"
-require_relative "./job-iteration/integrations"
 require_relative "./job-iteration/enumerator_builder"
 require_relative "./job-iteration/iteration"
 require_relative "./job-iteration/log_subscriber"
 
 module JobIteration
+  IntegrationLoadError = Class.new(StandardError)
+
+  INTEGRATIONS = [:resque, :sidekiq]
+
   extend self
 
-  attr_accessor :logger
+  attr_writer :logger
 
-  self.logger = ActiveJob::Base.logger
+  class << self
+    def logger
+      @logger || ActiveJob::Base.logger
+    end
+  end
 
   # Use this to _always_ interrupt the job after it's been running for more than N seconds.
   # @example
@@ -42,6 +49,11 @@ module JobIteration
   # where the throttle backoff value will take precedence over this setting.
   attr_accessor :default_retry_backoff
 
+  # Used internally for hooking into job processing frameworks like Sidekiq and Resque.
+  attr_accessor :interruption_adapter
+
+  self.interruption_adapter = -> { false }
+
   # Set if you want to use your own enumerator builder instead of default EnumeratorBuilder.
   # @example
   #
@@ -53,4 +65,29 @@ module JobIteration
   attr_accessor :enumerator_builder
 
   self.enumerator_builder = JobIteration::EnumeratorBuilder
+
+  def load_integrations
+    loaded = nil
+    INTEGRATIONS.each do |integration|
+      load_integration(integration)
+      if loaded
+        raise IntegrationLoadError,
+          "#{loaded} integration has already been loaded, but #{integration} is also available. " \
+            "Iteration will only work with one integration."
+      end
+      loaded = integration
+    rescue LoadError
+    end
+  end
+
+  def load_integration(integration)
+    unless INTEGRATIONS.include?(integration)
+      raise IntegrationLoadError,
+        "#{integration} integration is not supported. Available integrations: #{INTEGRATIONS.join(", ")}"
+    end
+
+    require_relative "./job-iteration/integrations/#{integration}"
+  end
 end
+
+JobIteration.load_integrations unless ENV["ITERATION_DISABLE_AUTOCONFIGURE"]

lib/job-iteration/active_record_enumerator.rb

@@ -31,7 +31,7 @@ def records
     def batches
       cursor = finder_cursor
       Enumerator.new(method(:size)) do |yielder|
-        while (records = instrument_next_batch(cursor))
+        while (records = cursor.next_batch(@batch_size))
           yielder.yield(records, cursor_value(records.last)) if records.any?
         end
       end
@@ -43,12 +43,6 @@ def size
 
     private
 
-    def instrument_next_batch(cursor)
-      ActiveSupport::Notifications.instrument("active_record_cursor.iteration") do
-        cursor.next_batch(@batch_size)
-      end
-    end
-
     def cursor_value(record)
       positions = @columns.map do |column|
         attribute_name = column.to_s.split(".").last

lib/job-iteration/integrations.rb

@@ -4,27 +4,21 @@
 
 module JobIteration
   module Integrations
-    module Resque
-      module IterationExtension
-        def initialize(*)
-          $resque_worker = self
-          super
-        end
-      end
-
-      # @private
-      module ::Resque
-        class Worker
-          # The patch is required in order to call shutdown? on a Resque::Worker instance
-          prepend(IterationExtension)
-        end
+    module ResqueIterationExtension # @private
+      def initialize(*) # @private
+        $resque_worker = self
+        super
       end
+    end
 
-      class << self
-        def call
-          $resque_worker.try!(:shutdown?)
-        end
+    # @private
+    module ::Resque
+      class Worker
+        # The patch is required in order to call shutdown? on a Resque::Worker instance
+        prepend(ResqueIterationExtension)
       end
     end
+
+    JobIteration.interruption_adapter = -> { $resque_worker.try!(:shutdown?) }
   end
 end

lib/job-iteration/integrations/resque.rb

@@ -1,16 +1,14 @@
 # frozen_string_literal: true
 
+require "sidekiq"
+
 module JobIteration
-  module Integrations
-    module Sidekiq
-      class << self
-        def call
-          if defined?(::Sidekiq::CLI) && (instance = ::Sidekiq::CLI.instance)
-            instance.launcher.stopping?
-          else
-            false
-          end
-        end
+  module Integrations # @private
+    JobIteration.interruption_adapter = -> do
+      if defined?(Sidekiq::CLI) && Sidekiq::CLI.instance
+        Sidekiq::CLI.instance.launcher.stopping?
+      else
+        false
       end
     end
   end

lib/job-iteration/integrations/sidekiq.rb

@@ -136,10 +136,6 @@ def retry_job(*, **)
 
     private
 
-    def interruption_adapter
-      @interruption_adapter ||= JobIteration::Integrations.load(self.class.queue_adapter_name)
-    end
-
     def enumerator_builder
       JobIteration.enumerator_builder.new(self)
     end
@@ -299,7 +295,7 @@ def job_should_exit?
         return true
       end
 
-      interruption_adapter.call || (defined?(super) && super)
+      JobIteration.interruption_adapter.call || (defined?(super) && super)
     end
 
     def handle_completed(completed)

lib/job-iteration/iteration.rb

@@ -23,7 +23,7 @@ def call
     #     MyJob.perform_now
     #   end
     def iterate_exact_times(n_times)
-      JobIteration::Integrations.stubs(:load).returns(StoppingSupervisor.new(n_times.size))
+      JobIteration.stubs(:interruption_adapter).returns(StoppingSupervisor.new(n_times.size))
     end
 
     # Stubs interruption adapter to interrupt the job after every sing iteration.
@@ -47,7 +47,7 @@ def mark_job_worker_as_interrupted
     def stub_shutdown_adapter_to_return(value)
       adapter = mock
       adapter.stubs(:call).returns(value)
-      JobIteration::Integrations.stubs(:load).returns(adapter)
+      JobIteration.stubs(:interruption_adapter).returns(adapter)
     end
   end
 end