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

@@ -3,87 +3,81 @@
 require "set"
 require "sentry/rails/serializer"
 require "sentry/rails/error_reporter_context"
+require "sentry/rails/active_job/job_extension"
+require "sentry/rails/active_job/producer"
 
 module Sentry
   module Rails
     module ActiveJobExtensions
       SENTRY_PAYLOAD_KEY = "_sentry"
 
-      USER_FIELDS_ALLOWLIST = %w[id email username].freeze
-
       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"],
+            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?
-
-        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?
-          end
-
-          if Sentry.configuration.data_collection.user_info
-            user = Sentry.get_current_scope.user
-            allowed = user.transform_keys(&:to_s).slice(*USER_FIELDS_ALLOWLIST)
-            sentry_data["user"] = allowed unless allowed.empty?
-          end
-
-          payload[SENTRY_PAYLOAD_KEY] = sentry_data unless sentry_data.empty?
-        rescue StandardError => e
-          Sentry.sdk_logger&.error("sentry-rails: failed to inject _sentry payload: #{e.class}: #{e.message}\n  #{Array(e.backtrace).first(5).join("\n  ")}")
-        end
-
-        payload
+        SentryReporter.producer(self).prepare_payload(self, payload)
       end
 
       def deserialize(job_data)
         super
         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
-        # identity cannot be resolved from the job. ActiveJob is adapter-agnostic
-        # and supports arbitrary third-party backends, so the concrete value is
-        # normally derived from the adapter itself; this is only the last resort.
-        MESSAGING_SYSTEM_FALLBACK = "activejob"
-
         EVENT_HANDLERS = {
           "enqueue_retry.active_job" => :retry_handler
         }
 
         class << self
-          include ErrorReporterContext
+          def default_extension
+            @default_extension ||= JobExtension.new
+          end
+
+          def consumer
+            @consumer ||= Sentry::JobExtensions::Consumer.new(integration: Sentry::Rails, origin: JobExtension::SPAN_ORIGIN, registry: job_extensions)
+          end
+
+          def producer(job)
+            job_extensions.call(job_extensions.for(job), :producer, job: job)
+          end
+
+          def job_extensions
+            default_extension.registry
+          end
 
           def producer_callback_registered?
             @producer_callback_registered ||= false
@@ -95,144 +89,51 @@ def producer_callback_registered!
 
           def record_producer_span(job, &enqueue)
             return yield if !Sentry.initialized? || job.already_supported_by_sentry_integration?
+            producer(job).record_span(job, &enqueue)
+          end
 
-            enqueued = false
+          def record(job, trace_headers: nil, user: nil, payload: nil, transaction: nil, scope: nil, &block)
+            options = { trace_headers: trace_headers, user: user, payload: payload, transaction: transaction }
+            return record_in_scope(scope, job, **options, backend_scope: true, &block) if scope
 
-            run_enqueue = lambda do
-              enqueued = true
-              enqueue.call
+            with_isolated_scope do |isolated_scope|
+              record_in_scope(isolated_scope, job, **options, &block)
             end
+          end
 
+          def record_in_scope(scope, job, trace_headers: nil, user: nil, payload: nil, transaction: nil, backend_scope: false)
             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
+              scope.set_user(user.transform_keys(&:to_sym)) if user && !user.empty?
 
-              log_producer_span_error(e)
+              consumer.apply_scope_data(scope, job)
 
-              run_enqueue.call
-            end
-          end
+              scope.set_transaction_name(consumer.transaction_name(job), source: :task)
 
-          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
+              transaction = consumer.start_transaction(
+                scope, job,
+                transaction: transaction.is_a?(Sentry::Transaction) ? transaction : nil,
+                trace_headers: trace_headers,
+                payload: payload
+              )
 
-            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)
+              yield.tap do
+                consumer.finish_transaction(transaction, 200)
+              end
+            rescue Exception => e # rubocop:disable Lint/RescueException
+              consumer.finish_transaction(transaction, 500)
 
+              if backend_scope
+                scope.update_from_options(**default_extension.error_data(job: job, exception: e))
+              elsif consumer.report_exception?(job, e)
                 capture_exception(job, e)
-
-                raise
               end
-            end
-          ensure
-            Sentry.set_current_hub_internal(original_hub)
-          end
-
-          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 (latency = compute_latency(job))
-              transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_RECEIVE_LATENCY, latency)
+              raise
             end
           end
 
-          def messaging_system(job)
-            name = job.class.queue_adapter_name if job.class.respond_to?(:queue_adapter_name)
-            name = name.to_s
-            name.empty? ? MESSAGING_SYSTEM_FALLBACK : name
-          end
-
-          # Number of retries the job has already gone through, as observed at
-          # the moment the span is opened. On the consumer the span opens before
-          # ActiveJob increments +executions+, so a job's first attempt reads 0,
-          # 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)
-            job.executions.to_i if job.respond_to?(:executions)
-          end
-
-          def compute_latency(job)
-            return unless job.respond_to?(:enqueued_at) && job.enqueued_at
-
-            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
-          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
-
-            Sentry::Rails.capture_exception(e, **options)
+            consumer.capture_exception(job, e, **default_extension.error_data(job: job, exception: e))
           end
 
           def register_event_handlers
@@ -255,7 +156,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: nil)
 
               capture_exception(job, error)
             end
@@ -266,35 +167,18 @@ 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
+          private :record_in_scope
 
-          def sentry_context(job)
-            context = {
-              active_job: job.class.name,
-              scheduled_at: job.scheduled_at,
-              job_id: job.job_id,
-              provider_job_id: job.provider_job_id,
-              locale: job.locale
-            }
-
-            if Sentry.configuration.data_collection.queues
-              context[:arguments] = sentry_serialize_arguments(job.arguments)
-            end
-
-            context
-          end
+          private
 
-          def sentry_serialize_arguments(argument)
-            Sentry::Rails::Serializer.serialize(argument)
+          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
 
-          private
-
           def subscribers
             @__subscribers__ ||= Set.new
           end