fix(sidekiq): trace failures before ActiveJob performs

Author: solnic

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

@@ -22,12 +22,15 @@ def perform_now
           super
         else
           data = _sentry || {}
+          transaction = Sentry.get_current_scope.get_span
+          transaction = nil unless transaction.is_a?(Sentry::Transaction) && transaction.span_id == data["transaction_span_id"]
           SentryReporter.record(
             self,
             trace_headers: data["trace_propagation_headers"],
             user: data["user"],
             delivery_queue: data["delivery_queue"],
-            payload: _sentry_job_data
+            payload: _sentry_job_data,
+            transaction: transaction
           ) { super }
         end
       end
@@ -141,7 +144,7 @@ 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, delivery_queue: nil, payload: nil, &block)
+          def record(job, trace_headers: nil, user: nil, delivery_queue: nil, payload: nil, transaction: nil, &block)
             original_hub = Sentry.get_current_hub_internal
             original_extension_scope_contexts = job._sentry_extension_scope_contexts
             Sentry.clone_hub_to_current_thread
@@ -164,6 +167,7 @@ def record(job, trace_headers: nil, user: nil, delivery_queue: nil, payload: nil
 
                 transaction = consumer.start_transaction(
                   scope, job,
+                  transaction: transaction.is_a?(Sentry::Transaction) ? transaction : nil,
                   trace_headers: trace_headers,
                   payload: payload,
                   messaging_data: -> {

sentry-ruby/lib/sentry/job_extensions/consumer.rb

@@ -25,7 +25,7 @@ def transaction_name(job)
         registry.call(registry.for(job), :transaction_name, job: job)
       end
 
-      def start_transaction(scope, job, trace_headers: nil, payload: nil, always_continue_trace: false, messaging_data: {}, default_retry_count: nil)
+      def start_transaction(scope, job, transaction: nil, trace_headers: nil, payload: nil, always_continue_trace: false, messaging_data: {}, default_retry_count: nil)
         extension = registry.for(job)
         options = {
           name: scope.transaction_name,
@@ -43,11 +43,16 @@ def start_transaction(scope, job, trace_headers: nil, payload: nil, always_conti
           end
         end
 
-        transaction = if always_continue_trace || (trace_headers && !trace_headers.empty?)
-          continued = Sentry.continue_trace(trace_headers, **options)
-          Sentry.start_transaction(transaction: continued, **options)
+        if transaction
+          transaction.set_name(options[:name], source: options[:source])
+          transaction.set_origin(options[:origin])
         else
-          Sentry.start_transaction(**options)
+          transaction = if always_continue_trace || (trace_headers && !trace_headers.empty?)
+            continued = Sentry.continue_trace(trace_headers, **options)
+            Sentry.start_transaction(transaction: continued, **options)
+          else
+            Sentry.start_transaction(**options)
+          end
         end
         return unless transaction
 
@@ -62,7 +67,7 @@ def start_transaction(scope, job, trace_headers: nil, payload: nil, always_conti
       end
 
       def finish_transaction(transaction, status)
-        return unless transaction
+        return if !transaction || transaction.timestamp
 
         transaction.set_http_status(status)
         transaction.finish

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

@@ -30,11 +30,12 @@ def performed_by_active_job?(sidekiq_job)
         !!sidekiq_job["wrapped"] && active_job_owns_adapter?
       end
 
-      def active_job_args(sidekiq_job, sidekiq_config: nil)
+      def active_job_args(sidekiq_job, sidekiq_config: nil, transaction: nil)
         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["transaction_span_id"] = transaction.span_id if transaction
         [payload.merge(
           SIDEKIQ_JOB_KEY => sidekiq_job,
           SIDEKIQ_RETRY_LIMIT_KEY => @retry_policy.limit(sidekiq_job, sidekiq_config),

sentry-sidekiq/lib/sentry/sidekiq/sentry_context_middleware.rb

@@ -56,15 +56,25 @@ def perform_through_active_job(job, sidekiq_job)
           scope.clear_span
           consumer.apply_scope_data(scope, sidekiq_job, default_scope_data: job_extension.scope_data(job: sidekiq_job))
           scope.set_transaction_name(job_extension.transaction_name(job: job), source: :task)
-          Sentry.continue_trace(job_extension.extract_trace_headers(job: sidekiq_job, payload: sidekiq_job))
-          job["args"] = job_extension.active_job_args(sidekiq_job, sidekiq_config: sidekiq_config)
+          trace_headers = args.first.is_a?(Hash) && args.first.dig("_sentry", "trace_propagation_headers")
+          trace_headers = job_extension.extract_trace_headers(job: sidekiq_job, payload: sidekiq_job) if !trace_headers || trace_headers.empty?
+          transaction = consumer.start_transaction(
+            scope, sidekiq_job,
+            payload: sidekiq_job,
+            trace_headers: trace_headers,
+            messaging_data: -> { job_extension.consumer_messaging_data(sidekiq_job) },
+            default_retry_count: job_extension.retry_info(job: sidekiq_job)[:count],
+            always_continue_trace: true
+          )
+          job["args"] = job_extension.active_job_args(sidekiq_job, sidekiq_config: sidekiq_config, transaction: transaction)
 
           completed = false
           begin
             result = yield
             completed = true
             result
           ensure
+            consumer.finish_transaction(transaction, completed ? 200 : 500)
             if !completed && (exception = $!)
               context = { context: JobExtension::JOB_ERROR_CONTEXT, job: sidekiq_job }
               context[:jobstr] = ::Sidekiq.dump_json(sidekiq_job) if JobExtension::JOB_STRING_IN_ERROR_CONTEXT

sentry-sidekiq/spec/isolated/active_job_spec.rb

@@ -483,6 +483,32 @@ def perform; end
       end
     end
 
+    it "records one failed consumer transaction when the queued job class is missing" do
+      Sentry.configuration.traces_sample_rate = 1.0
+      message = sidekiq_message(
+        parity_job, retry: false, queue: "delivery",
+        trace_propagation_headers: { "sentry-trace" => "#{"a" * 32}-#{"b" * 16}-1" }
+      )
+      message["wrapped"] = "RemovedJob"
+      message["args"].first["job_class"] = "RemovedJob"
+
+      [true, false].each do |adapter_skipped|
+        events = process(message, adapter_skipped: adapter_skipped)
+        transactions = consumer_transactions(events)
+        errors = events.grep(Sentry::ErrorEvent)
+
+        expect(transactions.size).to eq(1)
+        expect(errors.size).to eq(1)
+        expect(transactions.first.transaction).to eq("Sidekiq/RemovedJob")
+        expect(transactions.first.contexts.dig(:trace, :status)).to eq("internal_error")
+        expect(transactions.first.contexts.dig(:trace, :trace_id)).to eq("a" * 32)
+        expect(transactions.first.contexts.dig(:trace, :data)).to include(
+          "messaging.message.id" => message["jid"], "messaging.destination.name" => "delivery"
+        )
+        expect(errors.first.contexts.dig(:trace, :span_id)).to eq(transactions.first.contexts.dig(:trace, :span_id))
+      end
+    end
+
     it "restores the worker hub and payload after a failure before perform_now" do
       skip_sidekiq_adapter(false)
       message = sidekiq_message(parity_job, retry: false)