diff --git a/sentry-sidekiq/README.md b/sentry-sidekiq/README.md index 19faf2c7c..cc52587ba 100644 --- a/sentry-sidekiq/README.md +++ b/sentry-sidekiq/README.md @@ -34,3 +34,17 @@ gem "sentry-sidekiq" ``` Then you're all set! `sentry-sidekiq` will automatically insert a custom middleware and error handler to capture exceptions from your workers! + +### ActiveJob on Sidekiq + +By default, `ActiveJob::QueueAdapters::SidekiqAdapter` is listed in `config.rails.skippable_job_adapters`, so ActiveJob jobs running on Sidekiq are reported by `sentry-sidekiq`, exactly as before. Rails apps can opt in to the `sentry-rails` ActiveJob integration instead by removing the adapter from that list: + +```ruby +Sentry.init do |config| + config.rails.skippable_job_adapters.delete("ActiveJob::QueueAdapters::SidekiqAdapter") +end +``` + +The choice is made once, when the Rails app boots. Once opted in, `sentry-rails` reports those jobs under Sidekiq's transaction names and span origin, and `sentry-sidekiq` only reports errors raised before an ActiveJob job starts, such as a job class that can no longer be loaded. Sidekiq's own retry count is not recorded for those jobs, and `config.sidekiq.report_after_job_retries`, `config.sidekiq.report_only_dead_jobs` and the `attempt_threshold` job option only apply to the errors `sentry-sidekiq` reports. Plain Sidekiq jobs are not affected. + +Errors from ActiveJob jobs on Sidekiq that `retry_on` retries are not reported, regardless of `config.rails.active_job_report_on_retry_error`. diff --git a/sentry-sidekiq/Rakefile b/sentry-sidekiq/Rakefile index 9207ea5fe..648ec9fe3 100644 --- a/sentry-sidekiq/Rakefile +++ b/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"] diff --git a/sentry-sidekiq/lib/sentry-sidekiq.rb b/sentry-sidekiq/lib/sentry-sidekiq.rb index 4faeffcd2..1f28337ce 100644 --- a/sentry-sidekiq/lib/sentry-sidekiq.rb +++ b/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,14 @@ 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 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| diff --git a/sentry-sidekiq/lib/sentry/sidekiq/active_job_extension.rb b/sentry-sidekiq/lib/sentry/sidekiq/active_job_extension.rb new file mode 100644 index 000000000..e4aae23d6 --- /dev/null +++ b/sentry-sidekiq/lib/sentry/sidekiq/active_job_extension.rb @@ -0,0 +1,42 @@ +# frozen_string_literal: true + +require "sentry/rails/active_job_extension" +require "sentry/rails/active_job/producer" + +module Sentry + module Sidekiq + class ActiveJobExtension < Sentry::Rails::ActiveJobExtension + class Producer < Sentry::Rails::ActiveJobExtension::Producer + def record_span(job) + yield + end + end + + def initialize + super(name: "sidekiq_active_job") + end + + def transaction_name(job:) + "#{JobExtension::SIDEKIQ_NAME}/#{super}" + end + + def span_origin(job:) + JobExtension::SPAN_ORIGIN + end + + def integration(job:) + Sentry::Sidekiq + end + + def report_retried_exception?(job:, exception:) + false + end + + def producer(job:) + @producer ||= Producer.new(extension: self) + end + end + + Sentry::Rails.register_active_job_extension(:sidekiq, ActiveJobExtension.new) + end +end diff --git a/sentry-sidekiq/lib/sentry/sidekiq/active_job_support/server_middleware.rb b/sentry-sidekiq/lib/sentry/sidekiq/active_job_support/server_middleware.rb new file mode 100644 index 000000000..ce75da4f6 --- /dev/null +++ b/sentry-sidekiq/lib/sentry/sidekiq/active_job_support/server_middleware.rb @@ -0,0 +1,17 @@ +# frozen_string_literal: true + +module Sentry + module Sidekiq + module ActiveJobSupport + ACTIVE_JOB_ADAPTER = "ActiveJob::QueueAdapters::SidekiqAdapter" + + module ServerMiddleware + def call(worker, job, queue) + return yield if job["wrapped"] + + super + end + end + end + end +end diff --git a/sentry-sidekiq/lib/sentry/sidekiq/context_filter.rb b/sentry-sidekiq/lib/sentry/sidekiq/context_filter.rb index ff9f014c3..40890db5c 100644 --- a/sentry-sidekiq/lib/sentry/sidekiq/context_filter.rb +++ b/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 diff --git a/sentry-sidekiq/lib/sentry/sidekiq/error_handler.rb b/sentry-sidekiq/lib/sentry/sidekiq/error_handler.rb index 024364e96..84b91f517 100644 --- a/sentry-sidekiq/lib/sentry/sidekiq/error_handler.rb +++ b/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,21 @@ 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 + scope.set_transaction_name(@job_extension.error_context_transaction_name(context), source: :task) unless scope.transaction_name - retry_lim = retry_limit(context, sidekiq_config) + job = context[:job] + return unless @failure_reporting.report?(job, 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 + contexts = { sidekiq: ContextFilter.new(context).filtered } + captured = @consumer.safely("capture the job exception", fallback: false) do + @consumer.capture_exception(job, ex, contexts: contexts) + true 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 - - return if attempt < effective_threshold - end - - Sentry::Sidekiq.capture_exception( - ex, - contexts: { sidekiq: context_filter.filtered }, - hint: { background: false } - ) + Sentry::Sidekiq.capture_exception(ex, contexts: contexts, hint: { background: false }) unless captured 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 diff --git a/sentry-sidekiq/lib/sentry/sidekiq/failure_reporting.rb b/sentry-sidekiq/lib/sentry/sidekiq/failure_reporting.rb new file mode 100644 index 000000000..3798ce16f --- /dev/null +++ b/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 diff --git a/sentry-sidekiq/lib/sentry/sidekiq/job_extension.rb b/sentry-sidekiq/lib/sentry/sidekiq/job_extension.rb new file mode 100644 index 000000000..d2af7aefc --- /dev/null +++ b/sentry-sidekiq/lib/sentry/sidekiq/job_extension.rb @@ -0,0 +1,127 @@ +# frozen_string_literal: true + +require "sentry/sidekiq/retry_policy" +require "sentry/sidekiq/context_filter" + +module Sentry + module Sidekiq + class JobExtension < Sentry::JobExtensions::Extension + SPAN_ORIGIN = "auto.queue.sidekiq" + SIDEKIQ_NAME = "Sidekiq" + ENQUEUED_AT_IN_MILLISECONDS = ::Gem::Version.new(::Sidekiq::VERSION) >= ::Gem::Version.new("8.0.0") + + def initialize(name: "sidekiq", retry_policy: RetryPolicy.build) + super(name: name) + @retry_policy = retry_policy + end + + def error_context_transaction_name(context) + job = context["wrapped"] || context["class"] ? context : context[:job] + return transaction_name(job: job) if job_class(job) + + context[:event] ? "#{SIDEKIQ_NAME}/#{context[:event]}" : SIDEKIQ_NAME + end + + def scope_data(job:) + return {} unless sidekiq_job?(job) + + data = { + contexts: { sidekiq: ContextFilter.new(job).filtered }, + tags: { queue: job["queue"], jid: job["jid"] }.merge(job_tags(job)) + } + user = job["sentry_user"] + data[:user] = user if user.is_a?(Hash) + data + end + + def messaging_data(job:) + return {} unless sidekiq_job?(job) + + { + system: name, + message_id: job["jid"], + destination: job["queue"], + receive_latency: receive_latency(job), + body_size: body_size(job) + }.compact + end + + def publish_messaging_data(job:) + return {} unless sidekiq_job?(job) + + { system: name, message_id: job["jid"], destination: job["queue"] } + end + + def retry_info(job:) + return super unless sidekiq_job?(job) + + { count: @retry_policy.retry_count(job) || 0, will_retry: @retry_policy.will_retry?(job) } + end + + def inject_trace_headers(job:, payload:, headers:) + return payload unless sidekiq_job?(payload) + return payload if payload["trace_propagation_headers"] + + payload.merge("trace_propagation_headers" => headers) + end + + def extract_trace_headers(job:, payload:) + return {} unless sidekiq_job?(payload) + + payload["trace_propagation_headers"] || {} + end + + def transaction_name(job:) + name = job_class(job) + name ? "#{SIDEKIQ_NAME}/#{name}" : SIDEKIQ_NAME + end + + def publish_description(job:) + job["class"].to_s + end + + def span_origin(job:) + SPAN_ORIGIN + end + + def integration(job:) + Sentry::Sidekiq + end + + def producer(job:) + @producer ||= Sentry::JobExtensions::Producer::Delegated.new(extension: self) + end + + private + + def sidekiq_job?(job) + job.is_a?(Hash) && job.key?("jid") + end + + def job_class(job) + job["wrapped"] || job["class"] if job.is_a?(Hash) + end + + def job_tags(job) + Array(job["tags"]).each_with_object({}) { |name, tags_hash| tags_hash[:"sidekiq.#{name}"] = true } + end + + def body_size(job) + ::Sidekiq.dump_json(job).bytesize + rescue StandardError + nil + end + + def enqueued_at(job) + enqueued_at = job["enqueued_at"] + return unless enqueued_at + + ENQUEUED_AT_IN_MILLISECONDS ? Time.at(Rational(enqueued_at, 1000)) : Time.at(enqueued_at) + end + + def receive_latency(job) + Sentry::Utils::Latency.milliseconds_since(enqueued_at(job))&.round + end + end + end +end diff --git a/sentry-sidekiq/lib/sentry/sidekiq/producer.rb b/sentry-sidekiq/lib/sentry/sidekiq/producer.rb new file mode 100644 index 000000000..f5922a81a --- /dev/null +++ b/sentry-sidekiq/lib/sentry/sidekiq/producer.rb @@ -0,0 +1,14 @@ +# frozen_string_literal: true + +module Sentry + module Sidekiq + class Producer < Sentry::JobExtensions::Producer + private + + def describe_span(span, job) + span.set_description(extension.publish_description(job: job)) + set_messaging_data(span, extension.publish_messaging_data(job: job)) + end + end + end +end diff --git a/sentry-sidekiq/lib/sentry/sidekiq/railtie.rb b/sentry-sidekiq/lib/sentry/sidekiq/railtie.rb new file mode 100644 index 000000000..47839f0ba --- /dev/null +++ b/sentry-sidekiq/lib/sentry/sidekiq/railtie.rb @@ -0,0 +1,23 @@ +# frozen_string_literal: true + +require "sentry/sidekiq/active_job_support/server_middleware" + +module Sentry + module Sidekiq + class Railtie < ::Rails::Railtie + initializer "sentry.sidekiq.skippable_job_adapter", before: :load_environment_config do + next unless defined?(::Sentry::Rails) + + Sentry::Configuration.after(:initialize) { rails.skippable_job_adapters << ActiveJobSupport::ACTIVE_JOB_ADAPTER } + Sentry.configuration.rails.skippable_job_adapters << ActiveJobSupport::ACTIVE_JOB_ADAPTER if Sentry.initialized? + end + + config.after_initialize do + next unless defined?(::Sentry::Rails) && Sentry::Rails.active_job_owns_adapter?(ActiveJobSupport::ACTIVE_JOB_ADAPTER) + + require "sentry/sidekiq/active_job_extension" + SentryContextServerMiddleware.prepend(ActiveJobSupport::ServerMiddleware) + end + end + end +end diff --git a/sentry-sidekiq/lib/sentry/sidekiq/retry_policy.rb b/sentry-sidekiq/lib/sentry/sidekiq/retry_policy.rb new file mode 100644 index 000000000..796132b26 --- /dev/null +++ b/sentry-sidekiq/lib/sentry/sidekiq/retry_policy.rb @@ -0,0 +1,100 @@ +# frozen_string_literal: true + +module Sentry + module Sidekiq + class RetryPolicy + DEFAULT_MAX_RETRIES = 25 + + def self.build + if ::Gem::Version.new(::Sidekiq::VERSION) >= ::Gem::Version.new("7.0") + Sidekiq7.new + elsif ::Sidekiq.respond_to?(:[]) + Sidekiq6.new + else + Legacy.new + end + end + + attr_reader :default_config + + def initialize(default_config) + @default_config = default_config + end + + def retryable?(job) + retry_option = retry_option(job) + # 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 limit(job, sidekiq_config = nil) + retry_option = retry_option(job) + + case retry_option + when Integer + retry_option + when TrueClass, NilClass + max_retries(sidekiq_config) || DEFAULT_MAX_RETRIES + else + 0 + end + end + + def retry_count(job) + job && job["retry_count"] + end + + def will_retry?(job) + count = retry_count(job) + next_count = count.nil? ? 0 : count + 1 + retryable?(job) && next_count < limit(job) + end + + private + + def retry_option(job) + job && job["retry"] + end + + def max_retries(_sidekiq_config) + default_config[:max_retries] + end + + class Sidekiq7 < RetryPolicy + def initialize(default_config = ::Sidekiq.default_configuration) + super + end + + private + + # 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. + def max_retries(sidekiq_config) + config = sidekiq_config.is_a?(::Sidekiq::Config) ? sidekiq_config : default_config + config[:max_retries] + end + end + + class Sidekiq6 < RetryPolicy + def initialize(default_config = ::Sidekiq) + super + end + end + + class Legacy < RetryPolicy + def initialize(default_config = ::Sidekiq) + super + end + + private + + def max_retries(_sidekiq_config) + default_config.options[:max_retries] + end + end + end + end +end diff --git a/sentry-sidekiq/lib/sentry/sidekiq/sentry_context_middleware.rb b/sentry-sidekiq/lib/sentry/sidekiq/sentry_context_middleware.rb index 8a06b38d5..7c82e5a84 100644 --- a/sentry-sidekiq/lib/sentry/sidekiq/sentry_context_middleware.rb +++ b/sentry-sidekiq/lib/sentry/sidekiq/sentry_context_middleware.rb @@ -1,109 +1,70 @@ # frozen_string_literal: true -require "sentry/sidekiq/context_filter" +require "sentry/sidekiq/job_extension" +require "sentry/sidekiq/producer" module Sentry module Sidekiq - module Helpers - def set_span_data(span, id:, queue:, latency: nil, retry_count: nil) - return unless span - - span.set_data(Span::DataConventions::MESSAGING_MESSAGE_ID, id) - span.set_data(Span::DataConventions::MESSAGING_DESTINATION_NAME, queue) - span.set_data(Span::DataConventions::MESSAGING_MESSAGE_RECEIVE_LATENCY, latency) if latency - span.set_data(Span::DataConventions::MESSAGING_MESSAGE_RETRY_COUNT, retry_count) if retry_count - end + class SentryContextServerMiddleware + SPAN_ORIGIN = JobExtension::SPAN_ORIGIN - if ::Gem::Version.new(::Sidekiq::VERSION) >= ::Gem::Version.new("8.0.0") - def calculate_latency(job) - now_in_ms - job["enqueued_at"] if job["enqueued_at"] - end - else - def calculate_latency(job) - ((Time.now.to_f - job["enqueued_at"]) * 1000).round if job["enqueued_at"] - end + def self.consumer + @consumer ||= Sentry::JobExtensions::Consumer.new(extension: Sentry::Sidekiq.job_extension) end - def now_in_ms - ::Process.clock_gettime(::Process::CLOCK_REALTIME, :millisecond) + def initialize(consumer = self.class.consumer) + @consumer = consumer end - end - - class SentryContextServerMiddleware - include Sentry::Sidekiq::Helpers - - OP_NAME = "queue.process" - SPAN_ORIGIN = "auto.queue.sidekiq" def call(worker, job, queue) return yield unless Sentry.initialized? - context_filter = Sentry::Sidekiq::ContextFilter.new(job) + sidekiq_job = job.merge("queue" => queue) + extension = @consumer.extension Sentry.clone_hub_to_current_thread scope = Sentry.get_current_scope - if (user = job["sentry_user"]) - scope.set_user(user) + safely("apply scope data") { @consumer.apply_scope_data(scope, sidekiq_job) } + name = safely("name the transaction", fallback: Sentry::Sidekiq.job_extension.transaction_name(job: sidekiq_job)) do + extension.transaction_name(job: sidekiq_job) end - scope.set_tags(queue: queue, jid: job["jid"]) - scope.set_tags(build_tags(job["tags"])) - scope.set_contexts(sidekiq: job.merge("queue" => queue)) - scope.set_transaction_name(context_filter.transaction_name, source: :task) - transaction = start_transaction(scope, job["trace_propagation_headers"]) - - if transaction - scope.set_span(transaction) - - latency = calculate_latency(job) - - set_span_data( - transaction, - id: job["jid"], - queue: queue, - latency: latency, - retry_count: job["retry_count"] || 0 - ) + scope.set_transaction_name(name, source: :task) + trace_headers = safely("extract trace headers", fallback: {}) do + extension.extract_trace_headers(job: sidekiq_job, payload: sidekiq_job) + end + scope.generate_propagation_context if trace_headers.empty? + transaction = safely("start the transaction") do + @consumer.start_transaction(scope, sidekiq_job, trace_headers: trace_headers) end begin yield rescue - finish_transaction(transaction, 500) + @consumer.finish_transaction(transaction, 500) raise end - finish_transaction(transaction, 200) + @consumer.finish_transaction(transaction, 200) # don't need to use ensure here # if the job failed, we need to keep the scope for error handler. and the scope will be cleared there scope.clear end - def build_tags(tags) - Array(tags).each_with_object({}) { |name, tags_hash| tags_hash[:"sidekiq.#{name}"] = true } - end - - def start_transaction(scope, env) - options = { - name: scope.transaction_name, - source: scope.transaction_source, - op: OP_NAME, - origin: SPAN_ORIGIN - } + private - transaction = Sentry.continue_trace(env, **options) - Sentry.start_transaction(transaction: transaction, **options) - end - - def finish_transaction(transaction, status) - return unless transaction - - transaction.set_http_status(status) - transaction.finish + def safely(context, fallback: nil, &block) + @consumer.safely(context, fallback: fallback, &block) end end class SentryContextClientMiddleware - include Sentry::Sidekiq::Helpers + def self.producer + @producer ||= Producer.new(extension: Sentry::Sidekiq.job_extension) + end + + def initialize(producer = self.class.producer) + @producer = producer + end def call(worker_class, job, queue, _redis_pool) return yield unless Sentry.initialized? @@ -114,14 +75,11 @@ def call(worker_class, job, queue, _redis_pool) end if Sentry.configuration.sidekiq.propagate_traces - job["trace_propagation_headers"] ||= Sentry.get_trace_propagation_headers + headers = Sentry.get_trace_propagation_headers + job.replace(@producer.safely("inject trace headers", fallback: job) { @producer.inject_trace_headers(job, job, headers) }) end - Sentry.with_child_span(op: "queue.publish", description: worker_class.to_s) do |span| - set_span_data(span, id: job["jid"], queue: queue) - - yield - end + @producer.record_span(job.merge("queue" => queue)) { yield } end end end diff --git a/sentry-sidekiq/spec/isolated/active_job_spec.rb b/sentry-sidekiq/spec/isolated/active_job_spec.rb new file mode 100644 index 000000000..3bec646e9 --- /dev/null +++ b/sentry-sidekiq/spec/isolated/active_job_spec.rb @@ -0,0 +1,121 @@ +# frozen_string_literal: true + +require_relative "support/rails_app" +require "active_job/railtie" +require "sidekiq/api" + +RSpec.describe "ActiveJob jobs on Sidekiq when the Sidekiq adapter is not skippable" do + let(:processor) { new_processor } + let(:transport) { Sentry.get_current_client.transport } + + let(:succeeding_job) do + stub_const("SucceedingJob", Class.new(ActiveJob::Base) do + self.queue_adapter = :sidekiq + queue_as :events + + def perform(*); end + end) + end + + let(:failing_job) do + stub_const("FailingJob", Class.new(ActiveJob::Base) do + self.queue_adapter = :sidekiq + queue_as :events + + def perform(*) + raise "job failed" + end + end) + end + + before do + make_basic_app do |config| + config.traces_sample_rate = 1.0 + config.rails.skippable_job_adapters.delete("ActiveJob::QueueAdapters::SidekiqAdapter") + end + end + + def sidekiq_message(job_class, *arguments, **attributes) + job = job_class.new(*arguments) + enqueued_at = WITH_SIDEKIQ_8 ? now_in_ms : Time.now.to_f + { + "class" => ActiveJob::QueueAdapters::SidekiqAdapter::JobWrapper.to_s, + "wrapped" => job_class.name, + "queue" => job.queue_name, + "args" => [job.serialize], + "jid" => SecureRandom.hex(12), + "retry" => false, + "created_at" => enqueued_at, + "enqueued_at" => enqueued_at + }.merge(attributes.transform_keys(&:to_s)) + end + + def deliver(message) + transport.events.clear + process_work(processor, Sidekiq::BasicFetch::UnitOfWork.new("queue:#{message["queue"]}", Sidekiq.dump_json(message))) + transport.events.dup + end + + def consumer_transactions(events) + events.select { |event| event.is_a?(Sentry::TransactionEvent) && event.contexts.dig(:trace, :op) == "queue.process" } + end + + it "reports the job once through sentry-rails under Sidekiq's transaction name and origin" do + transactions = consumer_transactions(deliver(sidekiq_message(succeeding_job))) + + expect(transactions.size).to eq(1) + expect(transactions.first.transaction).to eq("Sidekiq/SucceedingJob") + expect(transactions.first.contexts.dig(:trace, :origin)).to eq("auto.queue.sidekiq") + expect(transactions.first.contexts[:active_job]).to include(job_class: "SucceedingJob", queue: "events") + end + + it "captures a failing job once with its ActiveJob context" do + message = sidekiq_message(failing_job) + + events = deliver(message) + + errors = events.grep(Sentry::ErrorEvent) + expect(errors.size).to eq(1) + expect(errors.first.transaction).to eq("Sidekiq/FailingJob") + expect(errors.first.contexts[:active_job]).to include(job_class: "FailingJob", provider_job_id: message["jid"]) + expect(errors.first.contexts.dig(:trace, :span_id)).to eq(consumer_transactions(events).first.contexts.dig(:trace, :span_id)) + end + + it "captures an error raised before the job runs once through sentry-sidekiq" do + message = sidekiq_message(succeeding_job, retry: false) + message["wrapped"] = "RemovedJob" + message["args"].first["job_class"] = "RemovedJob" + + errors = deliver(message).grep(Sentry::ErrorEvent) + + expect(errors.size).to eq(1) + expect(errors.first.exception.values.first.mechanism.type).to eq("sidekiq") + expect(errors.first.contexts[:sidekiq]).to include("jid" => message["jid"], "queue" => "events") + end + + it "continues the trace of the transaction that enqueued the job" do + queue = Sidekiq::Queue.new("events") + queue.clear + parent = Sentry.start_transaction(name: "enqueuer", op: "test") + Sentry.get_current_scope.set_span(parent) + + succeeding_job.perform_later("sku") + + parent.finish + Sentry.get_current_scope.clear + transaction = consumer_transactions(deliver(queue.first.item)).first + queue.clear + + expect(transaction.contexts[:trace]).to include(trace_id: parent.trace_id) + end + + it "keeps reporting plain Sidekiq workers through sentry-sidekiq" do + transport.events.clear + + execute_worker(processor, HappyWorker) + + transaction = consumer_transactions(transport.events).first + expect(transaction.transaction).to eq("Sidekiq/HappyWorker") + expect(transaction.contexts).not_to have_key(:active_job) + end +end diff --git a/sentry-sidekiq/spec/isolated/rails_spec.rb b/sentry-sidekiq/spec/isolated/rails_spec.rb index 7fad22de8..023f8b5c0 100644 --- a/sentry-sidekiq/spec/isolated/rails_spec.rb +++ b/sentry-sidekiq/spec/isolated/rails_spec.rb @@ -1,47 +1,6 @@ # frozen_string_literal: true -begin - require "simplecov" - SimpleCov.command_name "SidekiqRails" -rescue LoadError -end - -require "sentry-rails" - -# This MUST be required after sentry-rails because it requires sentry-sidekiq -# which checks if Railtie is defined to properly set things up -require_relative "../spec_helper" - -# This is needed to prevent Sidekiq 6.5 crash -if SIDEKIQ_VERSION >= Gem::Version.new("6.5") && SIDEKIQ_VERSION < Gem::Version.new("7.0") - # NoMethodError: - # undefined method 'broadcast' for class ActiveSupport::Logger - # /workspace/sentry/vendor/gems/3.4.5/gems/sidekiq-6.5.7/lib/sidekiq/rails.rb:46:in 'block (2 levels) in ' - Rails.logger = Logger.new($stdout) -end - -class TestApp < Rails::Application -end - -def make_basic_app - app = Class.new(TestApp) do - def self.name - "RailsTestApp" - end - end - - app.config.hosts = nil - app.config.secret_key_base = "test" - app.config.eager_load = false - - app.initializer :configure_sentry do - perform_basic_setup - end - - app.initialize! - Rails.application = app - app -end +require_relative "support/rails_app" RSpec.describe Sentry::Sidekiq do before do diff --git a/sentry-sidekiq/spec/isolated/support/rails_app.rb b/sentry-sidekiq/spec/isolated/support/rails_app.rb new file mode 100644 index 000000000..ad4ddf82d --- /dev/null +++ b/sentry-sidekiq/spec/isolated/support/rails_app.rb @@ -0,0 +1,58 @@ +# frozen_string_literal: true + +begin + require "simplecov" + SimpleCov.command_name "SidekiqRails" +rescue LoadError +end + +require "sentry-rails" + +# This MUST be required after sentry-rails because it requires sentry-sidekiq +# which checks if Railtie is defined to properly set things up +require_relative "../../spec_helper" + +# This is needed to prevent Sidekiq 6.5 crash +if SIDEKIQ_VERSION >= Gem::Version.new("6.5") && SIDEKIQ_VERSION < Gem::Version.new("7.0") + # NoMethodError: + # undefined method 'broadcast' for class ActiveSupport::Logger + # /workspace/sentry/vendor/gems/3.4.5/gems/sidekiq-6.5.7/lib/sidekiq/rails.rb:46:in 'block (2 levels) in ' + Rails.logger = Logger.new($stdout) +end + +class TestApp < Rails::Application +end + +BOOTED_APPS = [] + +def make_basic_app(&sentry_config) + return reinitialize_basic_app(BOOTED_APPS.first, &sentry_config) if BOOTED_APPS.any? + + boot_basic_app(&sentry_config).tap { |app| BOOTED_APPS << app } +end + +def reinitialize_basic_app(app, &sentry_config) + perform_basic_setup(&sentry_config) + ActiveSupport.run_load_hooks(:after_initialize, app) + app +end + +def boot_basic_app(&sentry_config) + app = Class.new(TestApp) do + def self.name + "RailsTestApp" + end + end + + app.config.hosts = nil + app.config.secret_key_base = "test" + app.config.eager_load = false + + app.initializer :configure_sentry do + perform_basic_setup(&sentry_config) + end + + app.initialize! + Rails.application = app + app +end diff --git a/sentry-sidekiq/spec/sentry/sidekiq/error_handler_spec.rb b/sentry-sidekiq/spec/sentry/sidekiq/error_handler_spec.rb index 42acd8615..24efbafef 100644 --- a/sentry-sidekiq/spec/sentry/sidekiq/error_handler_spec.rb +++ b/sentry-sidekiq/spec/sentry/sidekiq/error_handler_spec.rb @@ -91,6 +91,26 @@ expect(event[:contexts][:sidekiq]).to eq(expected_context) end + context "when the extension handling the job raises" do + let(:failing_extension_class) do + Class.new(Sentry::Sidekiq::JobExtension) do + define_method(:error_data) { |**| raise "extension failure" } + define_method(:integration) { |**| raise "extension failure" } + end + end + + subject { described_class.new(consumer: Sentry::JobExtensions::Consumer.new(extension: failing_extension_class.new(name: "failing"))) } + + it "still captures the exception with the Sidekiq context" do + exception = build_exception + + subject.call(exception, { context: "Job raised exception", job: context }) + + expect(transport.events.count).to eq(1) + expect(transport.events.first.contexts[:sidekiq]).to include("jid" => context["jid"], "class" => "HardWorker") + end + end + context "when the job is wrapped" do let(:context) { super().merge("class" => "WrapperJob", "wrapped" => "HardWorker") } diff --git a/sentry-sidekiq/spec/sentry/sidekiq/sentry_context_middleware_spec.rb b/sentry-sidekiq/spec/sentry/sidekiq/sentry_context_middleware_spec.rb index c1b1a4b62..bedd7a568 100644 --- a/sentry-sidekiq/spec/sentry/sidekiq/sentry_context_middleware_spec.rb +++ b/sentry-sidekiq/spec/sentry/sidekiq/sentry_context_middleware_spec.rb @@ -7,6 +7,21 @@ RSpec.shared_context "sidekiq", shared_context: :metadata do let(:user) { { "id" => rand(10_000) } } + let(:sdk_log) { StringIO.new } + + let(:failing_extension_class) do + Class.new(Sentry::Sidekiq::JobExtension) do + %i[ + scope_data transaction_name extract_trace_headers inject_trace_headers messaging_data + publish_messaging_data publish_description retry_info span_origin + ].each do |hook| + define_method(hook) { |**| raise "extension failure" } + end + end + end + + let(:failing_extension) { failing_extension_class.new(name: "failing") } + let(:processor) do new_processor end @@ -30,6 +45,44 @@ expect(event.user).to eq(user) end + it "starts a new trace for a job without trace propagation headers" do + perform_basic_setup + worker_trace_id = Sentry.get_main_hub.current_scope.propagation_context.trace_id + + execute_worker(processor, SadWorker) + + event = transport.events.last + expect(event.contexts.dig(:trace, :trace_id)).to be_a(String) + expect(event.contexts.dig(:trace, :trace_id)).not_to eq(worker_trace_id) + end + + it "continues the trace from trace propagation headers when tracing is disabled" do + perform_basic_setup + trace_id = "a" * 32 + + execute_worker(processor, SadWorker, trace_propagation_headers: { "sentry-trace" => "#{trace_id}-#{"b" * 16}-1" }) + + expect(transport.events.last.contexts.dig(:trace, :trace_id)).to eq(trace_id) + end + + context "when the extension handling the job raises" do + before do + perform_basic_setup do |config| + config.traces_sample_rate = 1.0 + config.sdk_logger = Logger.new(sdk_log) + end + end + + it "performs the job once and logs the extension's failure" do + performed = 0 + + described_class.new(Sentry::JobExtensions::Consumer.new(extension: failing_extension)).call(HappyWorker.new, { "jid" => "123456", "class" => "HappyWorker", "args" => [] }, "default") { performed += 1 } + + expect(performed).to eq(1) + expect(sdk_log.string).to include('Job extension "failing" failed') + end + end + context "with tracing enabled" do before do perform_basic_setup { |config| config.traces_sample_rate = 1.0 } @@ -58,7 +111,8 @@ it "sets sidekiq tags to the event" do execute_worker(processor, TagsWorker) event = transport.events.last - expect(event.tags.keys).to include(:"sidekiq.marvel", :"sidekiq.dc") + expect(event.tags).to include(queue: "default", jid: "123123", "sidekiq.marvel": true, "sidekiq.dc": true) + expect(event.contexts[:sidekiq]).to include("class" => "TagsWorker", "jid" => "123123", "queue" => "default", "tags" => ["marvel", "dc"]) end it "has the correct origin" do @@ -93,6 +147,38 @@ def timecop_delay expect(trace[:data]['messaging.message.receive.latency']).to be_within(20).of(expected_latency) end + it "reports sidekiq as the messaging system on the queue.process transaction" do + execute_worker(processor, HappyWorker) + + expect(transport.events[0].contexts.dig(:trace, :data)['messaging.system']).to eq('sidekiq') + end + + it "reports the job's retry count on the queue.process transaction" do + execute_worker(processor, HappyWorker, retry_count: 3) + + expect(transport.events[0].contexts.dig(:trace, :data)['messaging.message.retry.count']).to eq(3) + end + + it "reports the byte length of the job encoded as JSON as the message body size on the queue.process transaction" do + message = Sidekiq.dump_json("jid" => "123456", "class" => "HappyWorker", "args" => ["héllo"], "queue" => "default") + + process_work(processor, Sidekiq::BasicFetch::UnitOfWork.new("queue:default", message)) + + expect(transport.events[0].contexts.dig(:trace, :data)['messaging.message.body.size']).to eq(message.bytesize) + end + + it "performs the job and omits the message body size when the job cannot be encoded as JSON" do + job = { "jid" => "123456", "class" => "HappyWorker", "args" => ["\xFF".dup.force_encoding(Encoding::UTF_8)], "queue" => "default" } + performed = false + + described_class.new.call(HappyWorker.new, job, "default") { performed = true } + + trace_data = transport.events[0].contexts.dig(:trace, :data) + expect(performed).to be(true) + expect(trace_data).to include('messaging.message.id' => '123456', 'messaging.system' => 'sidekiq') + expect(trace_data).not_to have_key('messaging.message.body.size') + end + if MIN_SIDEKIQ_6 it "does not fail for latency when performed inline" do HappyWorker.perform_inline @@ -181,6 +267,18 @@ def ensure_queue_empty(queue, timeout: 0.1) expect(queue.first["sentry_user"]).to eq(user) end + context "when the scope user has fields beyond id, email and username" do + let(:user) { { "id" => 1, "ip_address" => "1.2.3.4", "segment" => "vip" } } + + it "propagates the whole user" do + Sentry.configuration.data_collection.user_info = true + + client.push('queue' => 'default', 'class' => HappyWorker, 'args' => []) + + expect(queue.first["sentry_user"]).to eq(user) + end + end + it "does not propagate user information when user_info is disabled" do Sentry.configuration.data_collection.user_info = false @@ -239,6 +337,94 @@ def ensure_queue_empty(queue, timeout: 0.1) expect(span[:data]['messaging.destination.name']).to eq('default') end + it "only reports a producer body size that matches the finalized queued message" do + client.push('queue' => 'default', 'class' => HappyWorker, 'args' => ["héllo"]) + + transaction.finish + + pushed_job = queue.first.item + span = transport.events.last.spans.detect { |span| span[:op] == "queue.publish" } + expect(span[:data]['messaging.message.body.size']).to be_nil.or eq(Sidekiq.dump_json(pushed_job).bytesize) + end + + context "when an earlier client middleware reroutes the job" do + let(:rerouted_queue) { Sidekiq::Queue.new("rerouted") } + + let(:rerouting_middleware) do + Class.new do + def call(_worker_class, job, _queue, _redis_pool) + job["queue"] = "rerouted" + yield + end + end + end + + let(:client) do + Sidekiq::Client.new.tap do |client| + client.middleware do |chain| + chain.add rerouting_middleware + chain.add described_class + end + end + end + + after do + client.middleware { |chain| chain.remove(rerouting_middleware) } + ensure_queue_empty(rerouted_queue) + end + + it "reports the queue the push was made to as the destination" do + client.push('queue' => 'default', 'class' => HappyWorker, 'args' => []) + + transaction.finish + + span = transport.events.last.spans.detect { |span| span[:op] == "queue.publish" } + expect(span[:data]['messaging.destination.name']).to eq('default') + end + end + + it "names the queue.publish span after the worker with the sidekiq messaging system and without an origin or retry count" do + client.push('queue' => 'default', 'class' => HappyWorker, 'args' => [], 'retry_count' => 2) + + transaction.finish + + span = transport.events.last.spans.detect { |span| span[:op] == "queue.publish" } + expect(span[:description]).to eq("HappyWorker") + expect(span[:origin]).to eq("manual") + expect(span[:data]['messaging.system']).to eq('sidekiq') + expect(span[:data]).not_to have_key('messaging.message.retry.count') + end + + it "does not enqueue the job when recording the queue.publish span raises" do + allow(Sentry).to receive(:with_child_span).and_call_original + allow(Sentry).to receive(:with_child_span) + .with(hash_including(op: "queue.publish")) + .and_raise(StandardError, "boom from instrumentation") + + expect do + client.push('queue' => 'default', 'class' => HappyWorker, 'args' => []) + end.to raise_error(StandardError, /boom from instrumentation/) + + expect(queue.size).to be(0) + end + + context "when the extension handling the job raises" do + before do + Sentry.configuration.sdk_logger = Logger.new(sdk_log) + end + + it "enqueues the job once and logs the extension's failure" do + middleware = described_class.new(Sentry::Sidekiq::Producer.new(extension: failing_extension)) + job = { "jid" => "123456", "class" => "HappyWorker", "args" => [], "queue" => "default" } + enqueued = 0 + + middleware.call(HappyWorker, job, "default", nil) { enqueued += 1 } + + expect(enqueued).to eq(1) + expect(sdk_log.string).to include('Job extension "failing" failed') + end + end + it "does not propagate headers with propagate_traces = false" do perform_basic_setup do |config| config.traces_sample_rate = 1.0 diff --git a/sentry-sidekiq/spec/sentry/sidekiq_spec.rb b/sentry-sidekiq/spec/sentry/sidekiq_spec.rb index a7b727eb3..d87d7f6a5 100644 --- a/sentry-sidekiq/spec/sentry/sidekiq_spec.rb +++ b/sentry-sidekiq/spec/sentry/sidekiq_spec.rb @@ -117,6 +117,18 @@ def retry_last_failed_job end context "with attempt_threshold" do + [false, 0].each do |retry_option| + it "reports the first failure below the threshold when retry is #{retry_option}" do + worker = Class.new(SadWorker) + worker.sidekiq_options attempt_threshold: 3, retry: retry_option + + execute_worker(processor, worker) + + expect(transport.events.count).to eq(1) + expect(retry_set.count).to eq(0) + end + end + it "doesn't report the error until attempts equal the threshold" do worker = Class.new(SadWorker) worker.sidekiq_options attempt_threshold: 3 @@ -320,6 +332,18 @@ def retry_last_failed_job expect(event.contexts.dig(:trace, :trace_id)).to eq(transaction.contexts.dig(:trace, :trace_id)) end + it "does not log Sidekiq deprecation warnings while processing a job" do + log = StringIO.new + logger = Logger.new(log) + WITH_SIDEKIQ_7 ? Sidekiq.instance_variable_get(:@config).logger = logger : Sidekiq.logger = logger + + execute_worker(processor, HappyWorker) + + expect(log.string).not_to include("deprecated") + ensure + silence_sidekiq + end + context "with instrumenter :otel" do before do perform_basic_setup do |config|