Merge pull request #33751 from steves/add_retry_notifications_to_aj

Add hooks to ActiveJob around retries and discards

5 files changed, 152 insertions, 35 deletions

diff --git a/activejob/CHANGELOG.md b/activejob/CHANGELOG.md
index 2417ea3a87..c47465cb43 100644
--- a/activejob/CHANGELOG.md
+++ b/activejob/CHANGELOG.md
@@ -1,3 +1,7 @@
+*   Added `enqueue_retry.active_job`, `retry_stopped.active_job`, and `discard.active_job` hooks.
+
+    *steves*
+
 *   Allow `assert_performed_with` to be called without a block.
 
     *bogdanvlviv*
diff --git a/activejob/lib/active_job/exceptions.rb b/activejob/lib/active_job/exceptions.rb
index 1e57dbcb1c..9a14c33d80 100644
--- a/activejob/lib/active_job/exceptions.rb
+++ b/activejob/lib/active_job/exceptions.rb
@@ -44,14 +44,24 @@ module ActiveJob
       #  end
       def retry_on(*exceptions, wait: 3.seconds, attempts: 5, queue: nil, priority: nil)
         rescue_from(*exceptions) do |error|
+          payload = {
+            job: self,
+            adapter: self.class.queue_adapter,
+            error: error,
+            wait: wait
+          }
+
           if executions < attempts
-            logger.error "Retrying #{self.class} in #{wait} seconds, due to a #{error.class}. The original exception was #{error.cause.inspect}."
-            retry_job wait: determine_delay(wait), queue: queue, priority: priority
+            ActiveSupport::Notifications.instrument("enqueue_retry.active_job", payload) do
+              retry_job wait: determine_delay(wait), queue: queue, priority: priority
+            end
           else
             if block_given?
-              yield self, error
+              ActiveSupport::Notifications.instrument("retry_stopped.active_job", payload) do
+                yield self, error
+              end
             else
-              logger.error "Stopped retrying #{self.class} due to a #{error.class}, which reoccurred on #{executions} attempts. The original exception was #{error.cause.inspect}."
+              ActiveSupport::Notifications.instrument("retry_stopped.active_job", payload)
               raise error
             end
           end
@@ -78,10 +88,16 @@ module ActiveJob
       #  end
       def discard_on(*exceptions)
         rescue_from(*exceptions) do |error|
-          if block_given?
-            yield self, error
-          else
-            logger.error "Discarded #{self.class} due to a #{error.class}. The original exception was #{error.cause.inspect}."
+          payload = {
+            job: self,
+            adapter: self.class.queue_adapter,
+            error: error
+          }
+
+          ActiveSupport::Notifications.instrument("discard.active_job", payload) do
+            if block_given?
+              yield self, error
+            end
           end
         end
       end
diff --git a/activejob/lib/active_job/logging.rb b/activejob/lib/active_job/logging.rb
index 9ffd60ad53..96a3e6bf48 100644
--- a/activejob/lib/active_job/logging.rb
+++ b/activejob/lib/active_job/logging.rb
@@ -88,6 +88,34 @@ module ActiveJob
           end
         end
 
+        def enqueue_retry(event)
+          job = event.payload[:job]
+          ex = event.payload[:error]
+          wait = event.payload[:wait]
+
+          error do
+            "Retrying #{job.class} in #{wait} seconds, due to a #{ex.class}. The original exception was #{ex.cause.inspect}."
+          end
+        end
+
+        def retry_stopped(event)
+          job = event.payload[:job]
+          ex = event.payload[:error]
+
+          error do
+            "Stopped retrying #{job.class} due to a #{ex.class}, which reoccurred on #{job.executions} attempts. The original exception was #{ex.cause.inspect}."
+          end
+        end
+
+        def discard(event)
+          job = event.payload[:job]
+          ex = event.payload[:error]
+
+          error do
+            "Discarded #{job.class} due to a #{ex.class}. The original exception was #{ex.cause.inspect}."
+          end
+        end
+
         private
           def queue_name(event)
             event.payload[:adapter].class.name.demodulize.remove("Adapter") + "(#{event.payload[:job].queue_name})"
diff --git a/activejob/test/cases/logging_test.rb b/activejob/test/cases/logging_test.rb
index a1107a07fd..4041f5f8c0 100644
--- a/activejob/test/cases/logging_test.rb
+++ b/activejob/test/cases/logging_test.rb
@@ -8,9 +8,11 @@ require "jobs/logging_job"
 require "jobs/overridden_logging_job"
 require "jobs/nested_job"
 require "jobs/rescue_job"
+require "jobs/retry_job"
 require "models/person"
 
 class LoggingTest < ActiveSupport::TestCase
+  include ActiveJob::TestHelper
   include ActiveSupport::LogSubscriber::TestHelper
   include ActiveSupport::Logger::Severity
 
@@ -59,13 +61,17 @@ class LoggingTest < ActiveSupport::TestCase
   end
 
   def test_uses_job_name_as_tag
-    LoggingJob.perform_later "Dummy"
-    assert_match(/\[LoggingJob\]/, @logger.messages)
+    perform_enqueued_jobs do
+      LoggingJob.perform_later "Dummy"
+      assert_match(/\[LoggingJob\]/, @logger.messages)
+    end
   end
 
   def test_uses_job_id_as_tag
-    LoggingJob.perform_later "Dummy"
-    assert_match(/\[LOGGING-JOB-ID\]/, @logger.messages)
+    perform_enqueued_jobs do
+      LoggingJob.perform_later "Dummy"
+      assert_match(/\[LOGGING-JOB-ID\]/, @logger.messages)
+    end
   end
 
   def test_logs_correct_queue_name
@@ -78,19 +84,23 @@ class LoggingTest < ActiveSupport::TestCase
   end
 
   def test_globalid_parameter_logging
-    person = Person.new(123)
-    LoggingJob.perform_later person
-    assert_match(%r{Enqueued.*gid://aj/Person/123}, @logger.messages)
-    assert_match(%r{Dummy, here is it: #<Person:.*>}, @logger.messages)
-    assert_match(%r{Performing.*gid://aj/Person/123}, @logger.messages)
+    perform_enqueued_jobs do
+      person = Person.new(123)
+      LoggingJob.perform_later person
+      assert_match(%r{Enqueued.*gid://aj/Person/123}, @logger.messages)
+      assert_match(%r{Dummy, here is it: #<Person:.*>}, @logger.messages)
+      assert_match(%r{Performing.*gid://aj/Person/123}, @logger.messages)
+    end
   end
 
   def test_globalid_nested_parameter_logging
-    person = Person.new(123)
-    LoggingJob.perform_later(person: person)
-    assert_match(%r{Enqueued.*gid://aj/Person/123}, @logger.messages)
-    assert_match(%r{Dummy, here is it: .*#<Person:.*>}, @logger.messages)
-    assert_match(%r{Performing.*gid://aj/Person/123}, @logger.messages)
+    perform_enqueued_jobs do
+      person = Person.new(123)
+      LoggingJob.perform_later(person: person)
+      assert_match(%r{Enqueued.*gid://aj/Person/123}, @logger.messages)
+      assert_match(%r{Dummy, here is it: .*#<Person:.*>}, @logger.messages)
+      assert_match(%r{Performing.*gid://aj/Person/123}, @logger.messages)
+    end
   end
 
   def test_enqueue_job_logging
@@ -102,22 +112,26 @@ class LoggingTest < ActiveSupport::TestCase
   end
 
   def test_perform_job_logging
-    LoggingJob.perform_later "Dummy"
-    assert_match(/Performing LoggingJob \(Job ID: .*?\) from .*? with arguments:.*Dummy/, @logger.messages)
-    assert_match(/Dummy, here is it: Dummy/, @logger.messages)
-    assert_match(/Performed LoggingJob \(Job ID: .*?\) from .*? in .*ms/, @logger.messages)
+    perform_enqueued_jobs do
+      LoggingJob.perform_later "Dummy"
+      assert_match(/Performing LoggingJob \(Job ID: .*?\) from .*? with arguments:.*Dummy/, @logger.messages)
+      assert_match(/Dummy, here is it: Dummy/, @logger.messages)
+      assert_match(/Performed LoggingJob \(Job ID: .*?\) from .*? in .*ms/, @logger.messages)
+    end
   end
 
   def test_perform_nested_jobs_logging
-    NestedJob.perform_later
-    assert_match(/\[LoggingJob\] \[.*?\]/, @logger.messages)
-    assert_match(/\[ActiveJob\] Enqueued NestedJob \(Job ID: .*\) to/, @logger.messages)
-    assert_match(/\[ActiveJob\] \[NestedJob\] \[NESTED-JOB-ID\] Performing NestedJob \(Job ID: .*?\) from/, @logger.messages)
-    assert_match(/\[ActiveJob\] \[NestedJob\] \[NESTED-JOB-ID\] Enqueued LoggingJob \(Job ID: .*?\) to .* with arguments: "NestedJob"/, @logger.messages)
-    assert_match(/\[ActiveJob\].*\[LoggingJob\] \[LOGGING-JOB-ID\] Performing LoggingJob \(Job ID: .*?\) from .* with arguments: "NestedJob"/, @logger.messages)
-    assert_match(/\[ActiveJob\].*\[LoggingJob\] \[LOGGING-JOB-ID\] Dummy, here is it: NestedJob/, @logger.messages)
-    assert_match(/\[ActiveJob\].*\[LoggingJob\] \[LOGGING-JOB-ID\] Performed LoggingJob \(Job ID: .*?\) from .* in/, @logger.messages)
-    assert_match(/\[ActiveJob\] \[NestedJob\] \[NESTED-JOB-ID\] Performed NestedJob \(Job ID: .*?\) from .* in/, @logger.messages)
+    perform_enqueued_jobs do
+      NestedJob.perform_later
+      assert_match(/\[LoggingJob\] \[.*?\]/, @logger.messages)
+      assert_match(/\[ActiveJob\] Enqueued NestedJob \(Job ID: .*\) to/, @logger.messages)
+      assert_match(/\[ActiveJob\] \[NestedJob\] \[NESTED-JOB-ID\] Performing NestedJob \(Job ID: .*?\) from/, @logger.messages)
+      assert_match(/\[ActiveJob\] \[NestedJob\] \[NESTED-JOB-ID\] Enqueued LoggingJob \(Job ID: .*?\) to .* with arguments: "NestedJob"/, @logger.messages)
+      assert_match(/\[ActiveJob\].*\[LoggingJob\] \[LOGGING-JOB-ID\] Performing LoggingJob \(Job ID: .*?\) from .* with arguments: "NestedJob"/, @logger.messages)
+      assert_match(/\[ActiveJob\].*\[LoggingJob\] \[LOGGING-JOB-ID\] Dummy, here is it: NestedJob/, @logger.messages)
+      assert_match(/\[ActiveJob\].*\[LoggingJob\] \[LOGGING-JOB-ID\] Performed LoggingJob \(Job ID: .*?\) from .* in/, @logger.messages)
+      assert_match(/\[ActiveJob\] \[NestedJob\] \[NESTED-JOB-ID\] Performed NestedJob \(Job ID: .*?\) from .* in/, @logger.messages)
+    end
   end
 
   def test_enqueue_at_job_logging
@@ -151,4 +165,35 @@ class LoggingTest < ActiveSupport::TestCase
     assert_match(/Performing RescueJob \(Job ID: .*?\) from .*? with arguments:.*other/, @logger.messages)
     assert_match(/Error performing RescueJob \(Job ID: .*?\) from .*? in .*ms: RescueJob::OtherError \(Bad hair\):\n.*\brescue_job\.rb:\d+:in `perform'/, @logger.messages)
   end
+
+  def test_enqueue_retry_logging
+    perform_enqueued_jobs do
+      RetryJob.perform_later "DefaultsError", 2
+      assert_match(/Retrying RetryJob in \d+ seconds, due to a DefaultsError\. The original exception was nil\./, @logger.messages)
+    end
+  end
+
+  def test_retry_stopped_logging
+    perform_enqueued_jobs do
+      RetryJob.perform_later "CustomCatchError", 6
+      assert_match(/Stopped retrying RetryJob due to a CustomCatchError, which reoccurred on \d+ attempts\. The original exception was #<CustomCatchError: CustomCatchError>\./, @logger.messages)
+    end
+  end
+
+  def test_retry_stopped_logging_without_block
+    perform_enqueued_jobs do
+      begin
+        RetryJob.perform_later "DefaultsError", 6
+      rescue DefaultsError
+        assert_match(/Stopped retrying RetryJob due to a DefaultsError, which reoccurred on \d+ attempts\. The original exception was #<DefaultsError: DefaultsError>\./, @logger.messages)
+      end
+    end
+  end
+
+  def test_discard_logging
+    perform_enqueued_jobs do
+      RetryJob.perform_later "DiscardableError", 2
+      assert_match(/Discarded RetryJob due to a DiscardableError\. The original exception was nil\./, @logger.messages)
+    end
+  end
 end
diff --git a/guides/source/active_support_instrumentation.md b/guides/source/active_support_instrumentation.md
index 3568c47dd8..8581817d71 100644
--- a/guides/source/active_support_instrumentation.md
+++ b/guides/source/active_support_instrumentation.md
@@ -458,6 +458,14 @@ Active Job
 | `:adapter`   | QueueAdapter object processing the job |
 | `:job`       | Job object                             |
 
+### enqueue_retry.active_job
+
+| Key          | Value                                  |
+| ------------ | -------------------------------------- |
+| `:job`       | Job object                             |
+| `:adapter`   | QueueAdapter object processing the job |
+| `:error`     | The error that caused the retry        |
+
 ### perform_start.active_job
 
 | Key          | Value                                  |
@@ -472,6 +480,22 @@ Active Job
 | `:adapter`   | QueueAdapter object processing the job |
 | `:job`       | Job object                             |
 
+### retry_stopped.active_job
+
+| Key          | Value                                  |
+| ------------ | -------------------------------------- |
+| `:adapter`   | QueueAdapter object processing the job |
+| `:job`       | Job object                             |
+| `:error`     | The error that caused the retry        |
+
+### discard.active_job
+
+| Key          | Value                                  |
+| ------------ | -------------------------------------- |
+| `:adapter`   | QueueAdapter object processing the job |
+| `:job`       | Job object                             |
+| `:error`     | The error that caused the discard      |
+
 Action Cable
 ------------