feat(rails): delegate ActiveJob to backend extensions

Co-Authored-By: Codex <noreply@example.com>

Author: solnic

sentry-rails/lib/sentry/rails.rb

@@ -13,5 +13,28 @@ module Sentry
   module Rails
     extend Integrable
     register_integration name: "rails", version: Sentry::Rails::VERSION
+
+    # Returns whether sentry-rails' ActiveJob integration reports jobs run
+    # through the given queue adapter.
+    #
+    # When it returns true, the integration captures errors from those jobs,
+    # records their queue.process transactions and queue.publish spans,
+    # propagates traces, and consults extensions registered with
+    # {Sentry.register_job_extension}. A backend gem should then not report the
+    # same ActiveJob jobs itself.
+    #
+    # The result depends on {Sentry.initialized?} and the current configuration,
+    # so it can change after {Sentry.init}. Call it when a job is handled rather
+    # than caching the result at load time.
+    #
+    # @param adapter_class_name [String, Class] the ActiveJob queue adapter's
+    #   class name, such as "ActiveJob::QueueAdapters::SidekiqAdapter", or the
+    #   adapter class itself. It is compared against
+    #   +config.rails.skippable_job_adapters+.
+    # @return [Boolean] true if the SDK is initialized and the adapter is not
+    #   listed in +config.rails.skippable_job_adapters+, false otherwise
+    def self.active_job_owns_adapter?(adapter_class_name)
+      Sentry.initialized? && !Sentry.configuration.rails.skippable_job_adapters.include?(adapter_class_name.to_s)
+    end
   end
 end

sentry-rails/lib/sentry/rails/active_job.rb

@@ -13,30 +13,42 @@ module ActiveJobExtensions
 
       def self.prepended(base)
         base.attr_accessor :_sentry
+        base.attr_reader :_sentry_job_data
       end
 
       def perform_now
         if !Sentry.initialized? || already_supported_by_sentry_integration?
           super
         else
           data = _sentry || {}
+          current_scope = Sentry.get_current_scope
+          scope = current_scope if data["execution_scope_id"] == current_scope.object_id
+          transaction = scope&.get_span
           SentryReporter.record(
             self,
             trace_headers: data["trace_propagation_headers"],
-            user: data["user"]
+            user: data["user"],
+            delivery_queue: data["delivery_queue"],
+            payload: _sentry_job_data,
+            transaction: transaction,
+            scope: scope
           ) { super }
         end
       end
 
       def serialize
         payload = super
         return payload if !Sentry.initialized? || already_supported_by_sentry_integration?
+        return payload if SentryReporter.producer.instrumented_by_extension?(self)
 
         begin
           sentry_data = {}
           if Sentry.configuration.rails.active_job_propagate_traces
             headers = Sentry.get_trace_propagation_headers
-            sentry_data["trace_propagation_headers"] = headers if headers && !headers.empty?
+            if headers && !headers.empty?
+              sentry_data["trace_propagation_headers"] = headers
+              payload = SentryReporter.producer.inject_trace_headers(self, payload, headers)
+            end
           end
 
           if Sentry.configuration.data_collection.user_info
@@ -58,18 +70,18 @@ def deserialize(job_data)
         return if !Sentry.initialized? || already_supported_by_sentry_integration?
 
         begin
+          @_sentry_job_data = job_data
           self._sentry = job_data[SENTRY_PAYLOAD_KEY]
         rescue StandardError => e
           Sentry.sdk_logger&.error("sentry-rails: failed to extract _sentry payload: #{e.class}: #{e.message}\n  #{Array(e.backtrace).first(5).join("\n  ")}")
         end
       end
 
       def already_supported_by_sentry_integration?
-        Sentry.configuration.rails.skippable_job_adapters.include?(self.class.queue_adapter.class.to_s)
+        !Sentry::Rails.active_job_owns_adapter?(self.class.queue_adapter.class)
       end
 
       class SentryReporter
-        OP_NAME = "queue.process"
         SPAN_ORIGIN = "auto.queue.active_job"
 
         # Emitted as messaging.system when the configured queue adapter's
@@ -85,6 +97,22 @@ class SentryReporter
         class << self
           include ErrorReporterContext
 
+          def consumer
+            @consumer ||= Sentry::JobExtensions::Consumer.new(integration: Sentry::Rails, origin: SPAN_ORIGIN)
+          end
+
+          def producer
+            @producer ||= Sentry::JobExtensions::Producer.new(origin: SPAN_ORIGIN)
+          end
+
+          def job_extensions
+            @job_extensions ||= Sentry.job_extensions
+          end
+
+          def call_hook(job, hook, **arguments)
+            job_extensions.call(job_extensions.for(job), hook, job: job, **arguments)
+          end
+
           def producer_callback_registered?
             @producer_callback_registered ||= false
           end
@@ -95,104 +123,83 @@ def producer_callback_registered!
 
           def record_producer_span(job, &enqueue)
             return yield if !Sentry.initialized? || job.already_supported_by_sentry_integration?
-
-            enqueued = false
-
-            run_enqueue = lambda do
-              enqueued = true
-              enqueue.call
-            end
-
-            begin
-              Sentry.with_child_span(op: "queue.publish", description: job.class.name) do |span|
-                if span
-                  span.set_origin(SPAN_ORIGIN)
-                  span.set_data(Sentry::Span::DataConventions::MESSAGING_SYSTEM, messaging_system(job))
-                  span.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_ID, job.job_id)
-                  span.set_data(Sentry::Span::DataConventions::MESSAGING_DESTINATION_NAME, job.queue_name)
-
-                  if (count = retry_count(job))
-                    span.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_RETRY_COUNT, count)
-                  end
-                end
-
-                run_enqueue.call
-              end
-            rescue StandardError => e
-              raise if enqueued
-
-              log_producer_span_error(e)
-
-              run_enqueue.call
-            end
+            return yield if producer.instrumented_by_extension?(job)
+
+            producer.record_span(
+              job,
+              description: job.class.name,
+              messaging_data: -> {
+                {
+                  system: messaging_system(job),
+                  message_id: job.job_id,
+                  destination: job.queue_name
+                }.compact
+              },
+              default_retry_count: -> { executions(job) },
+              on_instrumentation_error: method(:log_producer_span_error),
+              &enqueue
+            )
           end
 
           def log_producer_span_error(e)
             Sentry.sdk_logger&.error("sentry-rails: failed to record producer span: #{e.class}: #{e.message}\n  #{Array(e.backtrace).first(5).join("\n  ")}")
           end
 
-          def record(job, trace_headers: nil, user: nil, &block)
-            original_hub = Sentry.get_current_hub_internal
-            Sentry.clone_hub_to_current_thread
+          def record(job, trace_headers: nil, user: nil, delivery_queue: nil, payload: nil, transaction: nil, scope: nil, &block)
+            options = { trace_headers: trace_headers, user: user, delivery_queue: delivery_queue, payload: payload, transaction: transaction }
+            return record_in_scope(scope, job, **options, backend_scope: true, &block) if scope
 
-            Sentry.with_scope do |scope|
-              begin
-                scope.set_user(user.transform_keys(&:to_sym)) if user && !user.empty?
-
-                scope.set_transaction_name(job.class.name, source: :task)
-                scope.set_tags(queue: job.queue_name)
-
-                scope.set_contexts(active_job: {
-                  job_class: job.class.name,
-                  job_id: job.job_id,
-                  queue: job.queue_name,
-                  provider_job_id: job.provider_job_id
-                })
-
-                transaction_options = {
-                  name: scope.transaction_name,
-                  source: scope.transaction_source,
-                  op: OP_NAME,
-                  origin: SPAN_ORIGIN
-                }
-
-                transaction = if trace_headers && !trace_headers.empty?
-                  continued = Sentry.continue_trace(trace_headers, **transaction_options)
-                  Sentry.start_transaction(transaction: continued, **transaction_options)
-                else
-                  Sentry.start_transaction(**transaction_options)
-                end
-
-                if transaction
-                  set_messaging_data(transaction, job)
-                  scope.set_span(transaction)
-                end
-
-                yield.tap do
-                  finish_sentry_transaction(transaction, 200)
-                end
-              rescue Exception => e # rubocop:disable Lint/RescueException
-                finish_sentry_transaction(transaction, 500)
+            with_isolated_scope do |isolated_scope|
+              record_in_scope(isolated_scope, job, **options, &block)
+            end
+          end
 
-                capture_exception(job, e)
+          def record_in_scope(scope, job, trace_headers: nil, user: nil, delivery_queue: nil, payload: nil, transaction: nil, backend_scope: false)
+            begin
+              scope.set_user(user.transform_keys(&:to_sym)) if user && !user.empty?
+
+              consumer.apply_scope_data(scope, job)
+
+              scope.set_transaction_name(consumer.transaction_name(job) || job.class.name, source: :task)
+              scope.set_tags(queue: delivery_queue || job.queue_name)
 
-                raise
+              scope.set_contexts(active_job: {
+                job_class: job.class.name,
+                job_id: job.job_id,
+                queue: job.queue_name,
+                provider_job_id: job.provider_job_id
+              })
+
+              transaction = consumer.start_transaction(
+                scope, job,
+                transaction: transaction.is_a?(Sentry::Transaction) ? transaction : nil,
+                trace_headers: trace_headers,
+                payload: payload,
+                messaging_data: -> {
+                  {
+                    system: messaging_system(job),
+                    message_id: job.job_id,
+                    destination: job.queue_name,
+                    receive_latency: receive_latency(job),
+                    body_size: payload && body_size { payload }
+                  }
+                },
+                default_retry_count: executions(job)
+              )
+
+              yield.tap do
+                consumer.finish_transaction(transaction, 200)
               end
-            end
-          ensure
-            Sentry.set_current_hub_internal(original_hub)
-          end
+            rescue Exception => e # rubocop:disable Lint/RescueException
+              consumer.finish_transaction(transaction, 500)
 
-          def set_messaging_data(transaction, job)
-            transaction.set_data(Sentry::Span::DataConventions::MESSAGING_SYSTEM, messaging_system(job))
-            transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_ID, job.job_id)
-            transaction.set_data(Sentry::Span::DataConventions::MESSAGING_DESTINATION_NAME, job.queue_name)
-            if (count = retry_count(job))
-              transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_RETRY_COUNT, count)
-            end
+              if backend_scope
+                scope.update_from_options(**exception_data(job))
+              elsif consumer.report_exception?(job, e)
+                capture_exception(job, e)
+              end
 
-            if (latency = compute_latency(job))
-              transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_RECEIVE_LATENCY, latency)
+              raise
             end
           end
 
@@ -208,31 +215,30 @@ def messaging_system(job)
           # its first retry reads 1, and so on. On the producer the retry enqueue
           # runs after the failed attempt bumped +executions+, so it observes the
           # same progression.
-          def retry_count(job)
+          def executions(job)
             job.executions.to_i if job.respond_to?(:executions)
           end
 
-          def compute_latency(job)
-            return unless job.respond_to?(:enqueued_at) && job.enqueued_at
+          def receive_latency(job)
+            Sentry::Utils::Latency.milliseconds_since(job.enqueued_at) if job.respond_to?(:enqueued_at)
+          end
 
-            enqueued_time = job.enqueued_at.is_a?(String) ? Time.parse(job.enqueued_at) : job.enqueued_at
-            (Time.now.to_f - enqueued_time.to_f) * 1000.0
+          def body_size
+            ::JSON.generate(yield).bytesize
+          rescue StandardError
+            nil
           end
 
           def capture_exception(job, e)
-            options = {
-              extra: sentry_context(job),
-              tags: {
-                job_id: job.job_id,
-                provider_job_id: job.provider_job_id
-              },
-              # Send synchronously: a worker process may exit before the async
-              # background worker flushes its queue, which would drop the event.
-              hint: { background: false }
-            }
-            options[:contexts] = execution_context if Sentry.configuration.data_collection.queues
+            consumer.capture_exception(job, e, **exception_data(job))
+          end
 
-            Sentry::Rails.capture_exception(e, **options)
+          def exception_data(job)
+            {
+              contexts: Sentry.configuration.data_collection.queues ? execution_context : {},
+              tags: { job_id: job.job_id, provider_job_id: job.provider_job_id },
+              extra: sentry_context(job)
+            }
           end
 
           def register_event_handlers
@@ -255,7 +261,7 @@ def retry_handler(*args)
             handle_error_event(*args) do |job, error|
               return if error.nil?
               return if !Sentry.initialized? || job.already_supported_by_sentry_integration?
-              return unless Sentry.configuration.rails.active_job_report_on_retry_error
+              return unless consumer.report_retried_exception?(job, error, default: Sentry.configuration.rails.active_job_report_on_retry_error)
 
               capture_exception(job, error)
             end
@@ -266,13 +272,6 @@ def handle_error_event(*args)
             yield(event.payload[:job], event.payload[:error])
           end
 
-          def finish_sentry_transaction(transaction, status)
-            return unless transaction
-
-            transaction.set_http_status(status)
-            transaction.finish
-          end
-
           def sentry_context(job)
             context = {
               active_job: job.class.name,
@@ -293,8 +292,18 @@ def sentry_serialize_arguments(argument)
             Sentry::Rails::Serializer.serialize(argument)
           end
 
+          private :record_in_scope, :exception_data
+
           private
 
+          def with_isolated_scope(&block)
+            original_hub = Sentry.get_current_hub_internal
+            Sentry.clone_hub_to_current_thread
+            Sentry.with_scope(&block)
+          ensure
+            Sentry.set_current_hub_internal(original_hub)
+          end
+
           def subscribers
             @__subscribers__ ||= Set.new
           end