feat(sidekiq): migrate to ActiveJob-first integration

Author: solnic

sentry-sidekiq/Rakefile

@@ -1,13 +1,26 @@
 # frozen_string_literal: true
 
 require "bundler/gem_tasks"
-require_relative "../lib/sentry/test/rake_tasks"
-
-Sentry::Test::RakeTasks.define_spec_tasks(
-  spec_pattern: "spec/sentry/**/*_spec.rb",
-  spec_rspec_opts: "--order rand --format progress",
-  isolated_specs_pattern: "spec/isolated/**/*_spec.rb",
-  isolated_rspec_opts: "--format progress"
-)
+require "rspec/core/rake_task"
+
+RSpec::Core::RakeTask.new(:spec) do |task|
+  task.pattern = "spec/sentry/**/*_spec.rb"
+  task.rspec_opts = "--order rand --format progress"
+end
+
+namespace :spec do
+  isolated_tasks = Dir["spec/isolated/**/*_spec.rb"].sort.map do |file|
+    task_name = "isolated:#{file.delete_prefix("spec/isolated/").delete_suffix("_spec.rb")}"
+
+    RSpec::Core::RakeTask.new(task_name) do |task|
+      task.pattern = file
+      task.rspec_opts = "--format progress"
+    end
+
+    task_name
+  end
+
+  task isolated: isolated_tasks
+end
 
 task default: [:spec, :"spec:isolated"]

sentry-sidekiq/lib/sentry-sidekiq.rb

@@ -5,6 +5,7 @@
 require "sentry/integrable"
 require "sentry/sidekiq/version"
 require "sentry/sidekiq/configuration"
+require "sentry/sidekiq/job_extension"
 require "sentry/sidekiq/error_handler"
 require "sentry/sidekiq/sentry_context_middleware"
 
@@ -14,18 +15,16 @@ module Sidekiq
 
     register_integration name: "sidekiq", version: Sentry::Sidekiq::VERSION
 
-    if defined?(::Rails::Railtie)
-      class Railtie < ::Rails::Railtie
-        config.after_initialize do
-          next unless Sentry.initialized? && defined?(::Sentry::Rails)
-
-          Sentry.configuration.rails.skippable_job_adapters << "ActiveJob::QueueAdapters::SidekiqAdapter"
-        end
-      end
+    def self.job_extension
+      @job_extension ||= JobExtension.new
     end
+
+    Sentry.register_job_extension(job_extension)
   end
 end
 
+require "sentry/sidekiq/railtie" if defined?(::Rails::Railtie)
+
 Sidekiq.configure_server do |config|
   config.error_handlers << Sentry::Sidekiq::ErrorHandler.new
   config.server_middleware do |chain|

sentry-sidekiq/lib/sentry/sidekiq/active_job_support/job_extension.rb

@@ -0,0 +1,106 @@
+# frozen_string_literal: true
+
+module Sentry
+  module Sidekiq
+    module ActiveJobSupport
+      module JobExtension
+        ACTIVE_JOB_ADAPTER = "ActiveJob::QueueAdapters::SidekiqAdapter"
+        SIDEKIQ_JOB_KEY = "_sentry_sidekiq_job"
+
+        def handles?(job)
+          super || active_job?(job)
+        end
+
+        def scope_data(job:)
+          return super unless active_job?(job)
+
+          sidekiq_job = handed_over_sidekiq_job(job)
+          sidekiq_job ? super(job: sidekiq_job) : {}
+        end
+
+        def messaging_data(job:)
+          return super unless active_job?(job)
+
+          sidekiq_job = handed_over_sidekiq_job(job)
+          return {} unless sidekiq_job
+
+          super(job: sidekiq_job).merge(body_size: body_size(sidekiq_job)).compact
+        end
+
+        def retry_info(job:)
+          return super unless active_job?(job)
+
+          sidekiq_job = handed_over_sidekiq_job(job)
+          super(job: sidekiq_job) if sidekiq_job
+        end
+
+        def extract_trace_headers(job:, payload:)
+          return super if sidekiq_job?(payload)
+
+          sidekiq_job = sidekiq_job_in(payload)
+          super(job: job, payload: sidekiq_job) if sidekiq_job
+        end
+
+        def transaction_name(job:)
+          active_job?(job) ? "#{self.class::SIDEKIQ_NAME}/#{job.class.name}" : super
+        end
+
+        def span_origin(job:)
+          active_job?(job) ? self.class::SPAN_ORIGIN : super
+        end
+
+        def integration(job:)
+          active_job?(job) ? Sentry::Sidekiq : super
+        end
+
+        def report_retried_exception?(job:, exception:)
+          active_job?(job) ? false : super
+        end
+
+        def performed_by_active_job?(sidekiq_job)
+          !!sidekiq_job["wrapped"] && active_job_owns_adapter?
+        end
+
+        def active_job_args(sidekiq_job, scope:)
+          payload, *arguments = sidekiq_job["args"]
+          return sidekiq_job["args"] unless payload.is_a?(Hash)
+
+          sentry_data = (payload["_sentry"] || {}).merge("delivery_queue" => sidekiq_job["queue"])
+          sentry_data["execution_scope_id"] = scope.object_id
+          [payload.merge(
+            SIDEKIQ_JOB_KEY => sidekiq_job,
+            "_sentry" => sentry_data
+          ), *arguments]
+        end
+
+        private
+
+        def active_job?(job)
+          return false unless defined?(::ActiveJob::Base) && job.is_a?(::ActiveJob::Base)
+
+          job.class.queue_adapter.class.name == ACTIVE_JOB_ADAPTER
+        end
+
+        def active_job_owns_adapter?
+          return false unless defined?(::Sentry::Rails::ActiveJobExtensions)
+          return false unless ::Sentry::Rails.respond_to?(:active_job_owns_adapter?)
+
+          ::Sentry::Rails.active_job_owns_adapter?(ACTIVE_JOB_ADAPTER)
+        end
+
+        def handed_over_sidekiq_job(job)
+          return unless active_job?(job) && job.respond_to?(:_sentry_job_data)
+
+          sidekiq_job_in(job._sentry_job_data)
+        end
+
+        def sidekiq_job_in(active_job_payload)
+          return unless active_job_payload.is_a?(Hash)
+
+          sidekiq_job = active_job_payload[SIDEKIQ_JOB_KEY]
+          sidekiq_job if sidekiq_job?(sidekiq_job)
+        end
+      end
+    end
+  end
+end

sentry-sidekiq/lib/sentry/sidekiq/active_job_support/server_middleware.rb

@@ -0,0 +1,32 @@
+# frozen_string_literal: true
+
+module Sentry
+  module Sidekiq
+    module ActiveJobSupport
+      module ServerMiddleware
+        private
+
+        def execution_trace_headers(job, sidekiq_job)
+          if job_extension.performed_by_active_job?(job) && job["args"].first.is_a?(Hash)
+            headers = job["args"].first.dig("_sentry", "trace_propagation_headers")
+            return headers if headers && !headers.empty?
+          end
+
+          super
+        end
+
+        def perform_job(job, sidekiq_job, scope:)
+          return super unless job_extension.performed_by_active_job?(job)
+
+          args = job["args"]
+          begin
+            job["args"] = job_extension.active_job_args(sidekiq_job, scope: scope)
+            super
+          ensure
+            job["args"] = args
+          end
+        end
+      end
+    end
+  end
+end

sentry-sidekiq/lib/sentry/sidekiq/context_filter.rb

@@ -35,17 +35,7 @@ def filtered
       end
 
       def transaction_name
-        class_name = (context["wrapped"] || context["class"] ||
-                      (context[:job] && (context[:job]["wrapped"] || context[:job]["class"]))
-                    )
-
-        if class_name
-          "#{SIDEKIQ_NAME}/#{class_name}"
-        elsif context[:event]
-          "#{SIDEKIQ_NAME}/#{context[:event]}"
-        else
-          SIDEKIQ_NAME
-        end
+        Sentry::Sidekiq.job_extension.error_context_transaction_name(context)
       end
 
       private

sentry-sidekiq/lib/sentry/sidekiq/error_handler.rb

@@ -1,11 +1,17 @@
 # frozen_string_literal: true
 
 require "sentry/sidekiq/context_filter"
+require "sentry/sidekiq/failure_reporting"
+require "sentry/sidekiq/sentry_context_middleware"
 
 module Sentry
   module Sidekiq
     class ErrorHandler
-      WITH_SIDEKIQ_7 = ::Gem::Version.new(::Sidekiq::VERSION) >= ::Gem::Version.new("7.0")
+      def initialize(failure_reporting: FailureReporting.new, consumer: SentryContextServerMiddleware.consumer, job_extension: Sentry::Sidekiq.job_extension)
+        @failure_reporting = failure_reporting
+        @consumer = consumer
+        @job_extension = job_extension
+      end
 
       # @param ex [Exception] the exception / error that occured
       # @param context [Hash or Array] Sidekiq error context
@@ -16,83 +22,19 @@ class ErrorHandler
       def call(ex, context, sidekiq_config = nil)
         return unless Sentry.initialized?
 
-        context_filter = Sentry::Sidekiq::ContextFilter.new(context)
-
         scope = Sentry.get_current_scope
-        scope.set_transaction_name(context_filter.transaction_name, source: :task) unless scope.transaction_name
-
-        retry_lim = retry_limit(context, sidekiq_config)
-
-        # If Sentry is configured to only report an error _after_ all retries have been exhausted,
-        # and if the job is retryable, and have not exceeded the retry_limit,
-        # return early.
-        if Sentry.configuration.sidekiq.report_after_job_retries && retryable?(context)
-          retry_count = context.dig(:job, "retry_count")
-          if retry_count.nil? || retry_count < retry_lim - 1
-            return
-          end
-        end
-
-        # See if we want to ignore jobs that have dead: false
-        return if Sentry.configuration.sidekiq.report_only_dead_jobs && context.dig(:job, "dead") == false
-
-        # Check if the retry count is below the attempt_threshold
-        attempt_threshold = context.dig(:job, "attempt_threshold")
-        if attempt_threshold && retryable?(context)
-          attempt_threshold = attempt_threshold.to_i
-          retry_count = context.dig(:job, "retry_count")
-          # attempt 1 - retry_count is nil
-          # attempt 2 - this is your first retry so retry_count is 0
-          # attempt 3 - you have retried once, retry_count is 1
-          attempt = retry_count.nil? ? 1 : retry_count.to_i + 2
-          # Cap at the final attempt so jobs with fewer retries than the threshold still report.
-          effective_threshold = [attempt_threshold, retry_lim + 1].min
+        scope.set_transaction_name(@job_extension.error_context_transaction_name(context), source: :task) unless scope.transaction_name
 
-          return if attempt < effective_threshold
-        end
+        job = context[:job]
+        return unless @failure_reporting.report?(job, sidekiq_config)
 
-        Sentry::Sidekiq.capture_exception(
-          ex,
-          contexts: { sidekiq: context_filter.filtered },
-          hint: { background: false }
+        @consumer.capture_exception(
+          job, ex,
+          contexts: { sidekiq: ContextFilter.new(context).filtered }
         )
       ensure
         scope&.clear
       end
-
-      private
-
-      def retryable?(context)
-        retry_option = context.dig(:job, "retry")
-        # when `retry` is not specified, it's default is `true` and it means 25 retries.
-        retry_option.nil? || retry_option == true || (retry_option.is_a?(Integer) && retry_option.positive?)
-      end
-
-      # @return [Integer] the number of retries allowed for the job
-      # Tries to fetch the retry limit from the job config first,
-      # then falls back to Sidekiq's configuration.
-      def retry_limit(context, sidekiq_config)
-        limit = context.dig(:job, "retry")
-
-        case limit
-        when Integer
-          limit
-        when TrueClass, NilClass
-          max_retries =
-            if WITH_SIDEKIQ_7
-              # Sidekiq 7.1.5+ passes the config to the error handler, so we should use that.
-              # Sidekiq 7.0 -> 7.1.5 provides ::Sidekiq.default_configuration.
-              sidekiq_config.is_a?(::Sidekiq::Config) ?
-                sidekiq_config[:max_retries] :
-                ::Sidekiq.default_configuration[:max_retries]
-            else
-              ::Sidekiq.options[:max_retries]
-            end
-          max_retries || 25
-        else
-          0
-        end
-      end
     end
   end
 end

sentry-sidekiq/lib/sentry/sidekiq/failure_reporting.rb

@@ -0,0 +1,49 @@
+# frozen_string_literal: true
+
+require "sentry/sidekiq/retry_policy"
+
+module Sentry
+  module Sidekiq
+    class FailureReporting
+      def initialize(retry_policy: RetryPolicy.build)
+        @retry_policy = retry_policy
+      end
+
+      def report?(job, sidekiq_config = nil)
+        config = Sentry.configuration.sidekiq
+        retry_limit = @retry_policy.limit(job, sidekiq_config)
+
+        return false if config.report_after_job_retries && retries_pending?(job, retry_limit)
+        return false if config.report_only_dead_jobs && job_value(job, "dead") == false
+        return false if below_attempt_threshold?(job, retry_limit)
+
+        true
+      end
+
+      private
+
+      def retries_pending?(job, retry_limit)
+        return false unless @retry_policy.retryable?(job)
+
+        retry_count = job_value(job, "retry_count")
+        retry_count.nil? || retry_count < retry_limit - 1
+      end
+
+      def below_attempt_threshold?(job, retry_limit)
+        attempt_threshold = job_value(job, "attempt_threshold")
+        return false unless attempt_threshold && @retry_policy.retryable?(job)
+
+        [attempt_threshold.to_i, retry_limit + 1].min > attempt(job)
+      end
+
+      def attempt(job)
+        retry_count = job_value(job, "retry_count")
+        retry_count.nil? ? 1 : retry_count.to_i + 2
+      end
+
+      def job_value(job, key)
+        job && job[key]
+      end
+    end
+  end
+end