diff --git a/sentry-rails/lib/sentry/rails.rb b/sentry-rails/lib/sentry/rails.rb index 8af39f372..896968b2b 100644 --- a/sentry-rails/lib/sentry/rails.rb +++ b/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 diff --git a/sentry-rails/lib/sentry/rails/active_job.rb b/sentry-rails/lib/sentry/rails/active_job.rb index 69d48a1dd..49fd6433c 100644 --- a/sentry-rails/lib/sentry/rails/active_job.rb +++ b/sentry-rails/lib/sentry/rails/active_job.rb @@ -13,6 +13,7 @@ module ActiveJobExtensions def self.prepended(base) base.attr_accessor :_sentry + base.attr_reader :_sentry_job_data end def perform_now @@ -23,7 +24,8 @@ def perform_now SentryReporter.record( self, trace_headers: data["trace_propagation_headers"], - user: data["user"] + user: data["user"], + payload: _sentry_job_data ) { super } end end @@ -31,12 +33,16 @@ def perform_now def serialize payload = super return payload if !Sentry.initialized? || already_supported_by_sentry_integration? + return payload if SentryReporter.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.call_hook(self, :inject_trace_headers, payload: payload, headers: headers) + end end if Sentry.configuration.data_collection.user_info @@ -58,6 +64,7 @@ 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 ")}") @@ -65,7 +72,7 @@ def deserialize(job_data) 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 @@ -78,6 +85,13 @@ class SentryReporter # normally derived from the adapter itself; this is only the last resort. MESSAGING_SYSTEM_FALLBACK = "activejob" + MESSAGING_DATA_ATTRIBUTES = { + system: Sentry::Span::DataConventions::MESSAGING_SYSTEM, + message_id: Sentry::Span::DataConventions::MESSAGING_MESSAGE_ID, + destination: Sentry::Span::DataConventions::MESSAGING_DESTINATION_NAME, + receive_latency: Sentry::Span::DataConventions::MESSAGING_MESSAGE_RECEIVE_LATENCY + }.freeze + EVENT_HANDLERS = { "enqueue_retry.active_job" => :retry_handler } @@ -85,6 +99,14 @@ class SentryReporter class << self include ErrorReporterContext + 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 @@ -93,9 +115,15 @@ def producer_callback_registered! @producer_callback_registered = true end + def instrumented_by_extension?(job) + call_hook(job, :instruments_enqueue?) == true + end + def record_producer_span(job, &enqueue) return yield if !Sentry.initialized? || job.already_supported_by_sentry_integration? + return yield if instrumented_by_extension?(job) + enqueued = false run_enqueue = lambda do @@ -104,9 +132,11 @@ def record_producer_span(job, &enqueue) end begin - Sentry.with_child_span(op: "queue.publish", description: job.class.name) do |span| + description = call_hook(job, :publish_description) || job.class.name + Sentry.with_child_span(op: "queue.publish", description: description) do |span| if span - span.set_origin(SPAN_ORIGIN) + span.set_origin(call_hook(job, :span_origin) || SPAN_ORIGIN) + set_extension_messaging_data(span, job) 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) @@ -131,7 +161,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, &block) + def record(job, trace_headers: nil, user: nil, payload: nil, &block) original_hub = Sentry.get_current_hub_internal Sentry.clone_hub_to_current_thread @@ -139,7 +169,9 @@ def record(job, trace_headers: nil, user: nil, &block) begin scope.set_user(user.transform_keys(&:to_sym)) if user && !user.empty? - scope.set_transaction_name(job.class.name, source: :task) + apply_extension_scope_data(scope, job) + + scope.set_transaction_name(call_hook(job, :transaction_name) || job.class.name, source: :task) scope.set_tags(queue: job.queue_name) scope.set_contexts(active_job: { @@ -153,10 +185,19 @@ def record(job, trace_headers: nil, user: nil, &block) name: scope.transaction_name, source: scope.transaction_source, op: OP_NAME, - origin: SPAN_ORIGIN + origin: call_hook(job, :span_origin) || SPAN_ORIGIN } - transaction = if trace_headers && !trace_headers.empty? + extracted_trace = false + if (trace_headers.nil? || trace_headers.empty?) && payload + extracted_headers = call_hook(job, :extract_trace_headers, payload: payload) + unless extracted_headers.nil? + trace_headers = extracted_headers + extracted_trace = true + end + end + + transaction = if extracted_trace || (trace_headers && !trace_headers.empty?) continued = Sentry.continue_trace(trace_headers, **transaction_options) Sentry.start_transaction(transaction: continued, **transaction_options) else @@ -165,6 +206,7 @@ def record(job, trace_headers: nil, user: nil, &block) if transaction set_messaging_data(transaction, job) + set_extension_messaging_data(transaction, job) scope.set_span(transaction) end @@ -174,7 +216,7 @@ def record(job, trace_headers: nil, user: nil, &block) rescue Exception => e # rubocop:disable Lint/RescueException finish_sentry_transaction(transaction, 500) - capture_exception(job, e) + capture_exception(job, e) if report_exception?(job, e) raise end @@ -196,6 +238,20 @@ def set_messaging_data(transaction, job) end end + def apply_extension_scope_data(scope, job) + data = call_hook(job, :scope_data) + scope.set_contexts(data[:contexts]) if data[:contexts] + scope.set_tags(data[:tags]) if data[:tags] + scope.set_extras(data[:extra]) if data[:extra] + scope.set_user(data[:user]) if data[:user] + end + + def set_extension_messaging_data(span, job) + call_hook(job, :messaging_data).each do |key, value| + span.set_data(MESSAGING_DATA_ATTRIBUTES.fetch(key), value) + end + end + def messaging_system(job) name = job.class.queue_adapter_name if job.class.respond_to?(:queue_adapter_name) name = name.to_s @@ -209,7 +265,26 @@ def messaging_system(job) # 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) + if (info = call_hook(job, :retry_info)) + info[:count] + elsif job.respond_to?(:executions) + job.executions.to_i + end + end + + def backend_will_retry?(job) + info = call_hook(job, :retry_info) + info ? info[:will_retry] : false + end + + def report_exception?(job, exception) + decision = call_hook(job, :report_exception?, exception: exception) + decision.nil? ? !backend_will_retry?(job) : decision + end + + def report_retried_exception?(job, exception) + decision = call_hook(job, :report_retried_exception?, exception: exception) + decision.nil? ? Sentry.configuration.rails.active_job_report_on_retry_error : decision end def compute_latency(job) @@ -220,21 +295,49 @@ def compute_latency(job) end def capture_exception(job, e) + error_data = additive_error_data( + call_hook(job, :error_data, exception: e) + ) + options = { - extra: sentry_context(job), - tags: { + extra: error_data[:extra].merge(sentry_context(job)), + tags: error_data[:tags].merge( 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) + contexts = error_data[:contexts] + contexts = contexts.merge(matching_keys(execution_context, contexts)) if Sentry.configuration.data_collection.queues + options[:contexts] = contexts if Sentry.configuration.data_collection.queues || !contexts.empty? + + integration = call_hook(job, :integration) || Sentry::Rails + integration.capture_exception(e, **options) + end + + def additive_error_data(data) + scope = Sentry.get_current_scope + + { + contexts: matching_keys(data.fetch(:contexts, {}), scope.contexts).to_h do |name, context| + scope_context = scope.contexts.fetch(name, {}) + context = matching_keys(context, scope_context) + [name, context.merge(scope_context)] + end, + tags: matching_keys(data.fetch(:tags, {}), scope.tags).reject { |key, _value| scope.tags.key?(key) }, + extra: matching_keys(data.fetch(:extra, {}), scope.extra).reject { |key, _value| scope.extra.key?(key) } + } + end + + def matching_keys(data, reference) + keys = reference.keys.to_h { |key| [key.to_s, key] } + data.transform_keys { |key| keys.fetch(key.to_s, key) } end + def register_event_handlers EVENT_HANDLERS.each do |name, handler| subscribers << ActiveSupport::Notifications.subscribe(name) do |*args| @@ -255,7 +358,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 report_retried_exception?(job, error) capture_exception(job, error) end diff --git a/sentry-rails/spec/sentry/rails/active_job_job_extensions_spec.rb b/sentry-rails/spec/sentry/rails/active_job_job_extensions_spec.rb new file mode 100644 index 000000000..04479a142 --- /dev/null +++ b/sentry-rails/spec/sentry/rails/active_job_job_extensions_spec.rb @@ -0,0 +1,551 @@ +# frozen_string_literal: true + +require "spec_helper" + +RSpec.describe "Sentry + ActiveJob with a registered job extension", type: :job do + include_context "active_job backend harness", adapter: :test + include_context "test adapter" + + let(:configure_sentry) { proc { |config| config.traces_sample_rate = 1.0 } } + + def register_extension(&body) + klass = Class.new(Sentry::JobExtensions::Extension) do + def handles?(_job) + true + end + + class_eval(&body) if body + end + + Sentry.register_job_extension(klass.new(name: "test_backend")) + end + + def perform_failing_job + expect do + failing_job.perform_later + drain + end.to raise_error(RuntimeError, /boom from failing_job spec/) + + error_event + end + + def error_event + sentry_events.find { |event| event.is_a?(Sentry::ErrorEvent) } + end + + def publish_span + parent = transactions.find { |t| t.contexts.dig(:trace, :op) == "test" } + parent.spans.find { |span| span[:op] == "queue.publish" } + end + + def register_enriching_extension + register_extension do + def scope_data(job:) + { + contexts: { backend: { "jid" => "backend-jid" }, active_job: { job_id: "backend", attempt: 3 } }, + tags: { "backend.queue" => "low", queue: "backend-queue" }, + extra: { backend_payload_size: 12, job_id: "backend" } + } + end + + def error_data(job:, exception:) + { + contexts: { backend_error: { "retried" => true }, active_job: { job_id: "backend" } }, + tags: { "backend.error" => "yes", job_id: "backend", provider_job_id: "backend" }, + extra: { backend_attempts: 2, active_job: "backend" } + } + end + + def messaging_data(job:) + { system: "backend", message_id: "backend", destination: "backend", receive_latency: 42.0 } + end + end + end + + describe "scope data" do + before { register_enriching_extension } + + it "adds the extension's contexts, tags and extra to the consumer transaction while keeping ActiveJob's values" do + successful_job.perform_later + drain + + transaction = consumer_transaction + + expect(transaction.contexts[:backend]).to eq("jid" => "backend-jid") + expect(transaction.contexts[:active_job]).to include(job_class: successful_job.name, attempt: 3) + expect(transaction.contexts[:active_job][:job_id]).not_to eq("backend") + expect(transaction.tags).to include("backend.queue" => "low", queue: "default") + expect(transaction.extra).to include(backend_payload_size: 12) + end + + it "adds the extension's scope data to the error event while keeping ActiveJob's values" do + event = perform_failing_job + + expect(event.contexts[:backend]).to eq("jid" => "backend-jid") + expect(event.tags).to include("backend.queue" => "low", queue: "default") + expect(event.extra).to include(backend_payload_size: 12) + expect(event.extra[:job_id]).not_to eq("backend") + end + end + + describe "error data" do + before { register_enriching_extension } + + it "adds the extension's error data to the captured event while keeping ActiveJob's values" do + event = perform_failing_job + + expect(event.contexts[:backend_error]).to eq("retried" => true) + expect(event.tags).to include("backend.error" => "yes") + expect(event.extra).to include(backend_attempts: 2, active_job: failing_job.name) + expect(event.tags[:job_id]).to eq(event.extra[:job_id]) + expect(event.tags[:provider_job_id]).to eq(event.extra[:provider_job_id]) + expect(event.contexts[:active_job][:job_id]).to eq(event.extra[:job_id]) + end + end + + describe "error data from a changing job" do + it "keeps extension context values from before execution when the job fails" do + register_extension do + def scope_data(job:) + { contexts: { backend: { attempt: job.executions } } } + end + + def error_data(job:, exception:) + { contexts: { backend: { attempt: job.executions } } } + end + end + + expect(perform_failing_job.contexts[:backend]).to eq(attempt: 0) + end + end + + describe "string-keyed error context" do + it "preserves serialized ActiveJob metadata while adding backend fields" do + register_extension do + def error_data(job:, exception:) + { contexts: { "active_job" => { "job_id" => "backend-id", "attempt" => 3 } } } + end + end + + event = JSON.parse(JSON.generate(perform_failing_job.to_h)) + + expect(event.dig("contexts", "active_job")).to include( + "job_id" => event.dig("extra", "job_id"), + "job_class" => failing_job.name, + "queue" => "default", + "provider_job_id" => event.dig("extra", "provider_job_id"), + "attempt" => 3 + ) + end + end + + describe "error data overlapping the job's scope" do + it "does not let the extension's error data replace tags and extra already on the job's scope" do + register_extension do + def scope_data(job:) + { tags: { "backend.queue" => "low" }, extra: { backend_payload_size: 12 } } + end + + def error_data(job:, exception:) + { tags: { "backend.queue" => "high", queue: "backend-queue" }, extra: { backend_payload_size: 99 } } + end + end + + event = perform_failing_job + + expect(event.tags).to include("backend.queue" => "low", queue: "default") + expect(event.extra).to include(backend_payload_size: 12) + end + end + + describe "error data with Rails execution context", skip: RAILS_VERSION < 7.0 do + context "when queue data collection is enabled" do + let(:configure_sentry) do + proc do |config| + config.traces_sample_rate = 1.0 + config.data_collection.queues = true + end + end + + let(:failing_job) do + job_fixture do + def perform + Rails.error.set_context(debug_key: "from_job") + raise "boom from failing_job spec" + end + end + end + + it "keeps ActiveJob's execution context alongside the extension's contexts" do + register_extension do + def error_data(job:, exception:) + { contexts: { "rails.error" => { debug_key: "backend" }, backend_error: { "retried" => true } } } + end + end + + event = perform_failing_job + + expect(event.contexts["rails.error"]).to include(debug_key: "from_job") + expect(event.contexts[:backend_error]).to eq("retried" => true) + end + end + end + + describe "messaging data" do + before { register_enriching_extension } + + it "adds the extension's messaging data to the producer span only where ActiveJob sets none" do + within_parent_transaction do + successful_job.set(queue: "events").perform_later + end + + expect(publish_span[:data]).to include( + "messaging.system" => successful_job.queue_adapter_name, + "messaging.destination.name" => "events", + "messaging.message.receive.latency" => 42.0 + ) + expect(publish_span[:data]["messaging.message.id"]).not_to eq("backend") + end + + it "reports the extension's messaging data on the consumer transaction in place of ActiveJob's" do + successful_job.set(queue: "events").perform_later + drain + + expect(consumer_transaction.contexts.dig(:trace, :data)).to include( + "messaging.system" => "backend", + "messaging.message.id" => "backend", + "messaging.destination.name" => "backend", + "messaging.message.receive.latency" => 42.0 + ) + end + end + + describe "user" do + let(:configure_sentry) do + proc do |config| + config.traces_sample_rate = 1.0 + config.send_default_pii = true + end + end + + it "replaces the user the job carries with the user the extension supplies" do + register_extension do + def scope_data(job:) + { user: { "id" => "backend-user", "segment" => "vip" } } + end + end + Sentry.set_user(id: "web-user") + + successful_job.perform_later + Sentry.set_user({}) + + expect(last_enqueued_payload.dig("_sentry", "user")).to eq("id" => "web-user") + + drain + + expect(consumer_transaction.user).to eq("id" => "backend-user", "segment" => "vip") + end + end + + describe "retry info" do + def register_retrying_extension(count:, will_retry:) + register_extension do + define_method(:retry_info) { |job:| { count: count, will_retry: will_retry } } + end + end + + it "reports the extension's retry count on the producer span and the consumer transaction" do + register_retrying_extension(count: 4, will_retry: false) + + within_parent_transaction { successful_job.perform_later } + drain + + expect(publish_span[:data]).to include("messaging.message.retry.count" => 4) + expect(consumer_transaction.contexts.dig(:trace, :data)).to include("messaging.message.retry.count" => 4) + end + + it "does not capture the job's exception but still fails the transaction when the backend will retry" do + register_retrying_extension(count: 1, will_retry: true) + + expect do + failing_job.perform_later + drain + end.to raise_error(RuntimeError, /boom from failing_job spec/) + + expect(error_event).to be_nil + expect(consumer_transaction.contexts.dig(:trace, :status)).to eq("internal_error") + end + + it "captures the job's exception when the backend will not retry" do + register_retrying_extension(count: 5, will_retry: false) + + expect(perform_failing_job).not_to be_nil + end + end + + describe "exception reporting" do + def register_deciding_extension(report:, will_retry:) + register_extension do + define_method(:retry_info) { |job:| { count: 1, will_retry: will_retry } } + define_method(:report_exception?) { |job:, exception:| report } + end + end + + it "captures the job's exception when the extension reports it, even though the backend will retry" do + register_deciding_extension(report: true, will_retry: true) + + expect(perform_failing_job).not_to be_nil + end + + it "does not capture the job's exception when the extension declines it, but still fails the transaction" do + register_deciding_extension(report: false, will_retry: false) + + expect(perform_failing_job).to be_nil + expect(consumer_transaction.contexts.dig(:trace, :status)).to eq("internal_error") + end + end + + describe "retried exception reporting", skip: RAILS_VERSION < 6.0 do + let(:retrying_job) do + job_fixture do + retry_on StandardError, attempts: 2, wait: 0 + + def perform + raise "boom from retrying_job spec" if executions == 1 + end + end + end + + def register_retry_deciding_extension(report) + register_extension do + define_method(:report_retried_exception?) { |job:, exception:| report } + end + end + + def perform_retrying_job + retrying_job.perform_later + drain + error_event + end + + it "does not capture an exception ActiveJob retries when the extension declines it, even with active_job_report_on_retry_error" do + Sentry.configuration.rails.active_job_report_on_retry_error = true + register_retry_deciding_extension(false) + + expect(perform_retrying_job).to be_nil + end + + it "captures an exception ActiveJob retries when the extension reports it, even without active_job_report_on_retry_error" do + register_retry_deciding_extension(true) + + expect(perform_retrying_job&.exception&.values&.first&.value).to include("boom from retrying_job spec") + end + end + + describe "trace headers" do + let(:foreign_trace_id) { "a" * 32 } + let(:foreign_trace_headers) { { "sentry-trace" => "#{foreign_trace_id}-#{"b" * 16}-1" } } + + def register_extension_extracting(headers) + register_extension do + define_method(:extract_trace_headers) { |job:, payload:| headers } + end + end + + it "lets the extension carry the trace headers in the serialized payload alongside _sentry" do + register_extension do + def inject_trace_headers(job:, payload:, headers:) + payload.merge("backend_trace" => headers["sentry-trace"]) + end + end + + within_parent_transaction { successful_job.perform_later } + + payload = last_enqueued_payload + expect(payload["backend_trace"]).not_to be_nil + expect(payload["backend_trace"]).to eq(payload["_sentry"]["trace_propagation_headers"]["sentry-trace"]) + end + + context "when the job carries no _sentry trace headers" do + let(:configure_sentry) do + proc do |config| + config.traces_sample_rate = 1.0 + config.rails.active_job_propagate_traces = false + end + end + + it "continues the trace the extension extracts" do + register_extension_extracting(foreign_trace_headers) + + successful_job.perform_later + drain + + expect(consumer_transaction.contexts.dig(:trace, :trace_id)).to eq(foreign_trace_id) + end + end + + it "continues the _sentry trace even when the extension extracts different headers" do + register_extension_extracting(foreign_trace_headers) + parent_trace_id = nil + + within_parent_transaction do |parent| + parent_trace_id = parent.trace_id + successful_job.perform_later + end + drain + + expect(consumer_transaction.contexts.dig(:trace, :trace_id)).to eq(parent_trace_id) + end + end + + describe "naming" do + let(:backend_integration) do + Module.new do + extend Sentry::Integrable + + def self.integration_name + "backend" + end + end + end + + def mechanism_type(event) + event.exception.values.first.mechanism.type + end + + before do + integration = backend_integration + register_extension do + define_method(:transaction_name) { |job:| "Backend/#{job.class.name}" } + define_method(:span_origin) { |job:| "auto.queue.backend" } + define_method(:publish_description) { |job:| "Backend/#{job.class.name}" } + define_method(:integration) { |job:| integration } + end + end + + it "names the consumer transaction and sets its origin as the extension supplies" do + successful_job.perform_later + drain + + transaction = consumer_transaction + + expect(transaction.transaction).to eq("Backend/#{successful_job.name}") + expect(transaction.transaction_info).to eq(source: :task) + expect(transaction.contexts.dig(:trace, :origin)).to eq("auto.queue.backend") + end + + it "describes the producer span and sets its origin as the extension supplies" do + within_parent_transaction { successful_job.perform_later } + + expect(publish_span[:description]).to eq("Backend/#{successful_job.name}") + expect(publish_span[:origin]).to eq("auto.queue.backend") + end + + it "captures the job's exception under the extension's integration and transaction name" do + event = perform_failing_job + + expect(mechanism_type(event)).to eq("backend") + expect(event.transaction).to eq("Backend/#{failing_job.name}") + end + end + + describe "enqueue instrumentation" do + let(:configure_sentry) do + proc do |config| + config.traces_sample_rate = 1.0 + config.data_collection.user_info = true + end + end + + it "leaves the producer span, trace headers and user to an extension that instruments enqueueing the job" do + register_extension do + def instruments_enqueue?(job:) + true + end + end + Sentry.set_user(id: "42") + + within_parent_transaction { successful_job.perform_later } + + expect(publish_span).to be_nil + expect(last_enqueued_payload).not_to have_key("_sentry") + end + end + + describe "when the extension adds nothing" do + let(:extension_data_keys) { [:backend, "backend.queue", :backend_payload_size, :backend_error, "backend.error", :backend_attempts] } + + def expect_no_extension_data(event) + expect(event.contexts.keys & extension_data_keys).to be_empty + expect(event.tags.keys & extension_data_keys).to be_empty + expect(event.extra.keys & extension_data_keys).to be_empty + end + + it "sends nothing from an extension that does not handle the job" do + register_enriching_extension.define_singleton_method(:handles?) { |_job| false } + + within_parent_transaction { failing_job.perform_later } + expect { drain }.to raise_error(RuntimeError, /boom from failing_job spec/) + + expect_no_extension_data(error_event) + expect_no_extension_data(consumer_transaction) + expect(publish_span[:data]).not_to have_key("messaging.message.receive.latency") + end + + it "never consults the registry for jobs on a skipped adapter" do + consulted = [] + Sentry.register_job_extension( + Class.new(Sentry::JobExtensions::Extension) do + define_method(:handles?) do |job| + consulted << job + true + end + end.new(name: "observer") + ) + Sentry.configuration.rails.skippable_job_adapters = [failing_job.queue_adapter.class.to_s] + + expect do + failing_job.perform_later + drain + end.to raise_error(RuntimeError, /boom from failing_job spec/) + + expect(consulted).to be_empty + end + + { + "raises" => proc { raise "extension failure" }, + "returns malformed data" => proc { :malformed } + }.each do |behavior, hook_body| + context "when every hook #{behavior}" do + before do + register_extension do + %i[scope_data error_data messaging_data transaction_name span_origin publish_description integration instruments_enqueue?].each do |hook| + define_method(hook) { |**_args| instance_exec(&hook_body) } + end + end + end + + it "still performs the job and finishes its transaction" do + within_parent_transaction { successful_job.perform_later } + drain + + expect(consumer_transaction.contexts.dig(:trace, :status)).to eq("ok") + expect(consumer_transaction.transaction).to eq(successful_job.name) + expect(consumer_transaction.contexts.dig(:trace, :origin)).to eq("auto.queue.active_job") + expect(publish_span[:data]).not_to have_key("messaging.message.receive.latency") + expect(publish_span[:description]).to eq(successful_job.name) + expect(publish_span[:origin]).to eq("auto.queue.active_job") + end + + it "still captures the job's exception with ActiveJob's data" do + event = perform_failing_job + + expect(event.extra).to include(active_job: failing_job.name) + expect(event.tags[:job_id]).to eq(event.extra[:job_id]) + expect(event.transaction).to eq(failing_job.name) + expect(event.exception.values.first.mechanism.type).to eq("rails") + end + end + end + end +end diff --git a/sentry-rails/spec/sentry/rails/active_job_ownership_spec.rb b/sentry-rails/spec/sentry/rails/active_job_ownership_spec.rb new file mode 100644 index 000000000..3cc14c575 --- /dev/null +++ b/sentry-rails/spec/sentry/rails/active_job_ownership_spec.rb @@ -0,0 +1,25 @@ +# frozen_string_literal: true + +require "spec_helper" + +RSpec.describe "Sentry::Rails.active_job_owns_adapter?" do + let(:adapter_class_name) { "ActiveJob::QueueAdapters::SidekiqAdapter" } + + it "is true when Sentry is initialized and the adapter is not skipped" do + make_basic_app + + expect(Sentry::Rails.active_job_owns_adapter?(adapter_class_name)).to be(true) + end + + it "is false when the adapter is in skippable_job_adapters" do + make_basic_app do |config| + config.rails.skippable_job_adapters = [adapter_class_name] + end + + expect(Sentry::Rails.active_job_owns_adapter?(adapter_class_name)).to be(false) + end + + it "is false when Sentry is not initialized" do + expect(Sentry::Rails.active_job_owns_adapter?(adapter_class_name)).to be(false) + end +end diff --git a/sentry-rails/spec/spec_helper.rb b/sentry-rails/spec/spec_helper.rb index 1cdd3e617..82c56f508 100644 --- a/sentry-rails/spec/spec_helper.rb +++ b/sentry-rails/spec/spec_helper.rb @@ -88,6 +88,8 @@ reset_sentry_globals! + Sentry.job_extensions.clear + Rails.application&.cleanup! end diff --git a/sentry-ruby/lib/sentry-ruby.rb b/sentry-ruby/lib/sentry-ruby.rb index c961d732a..e7111373b 100644 --- a/sentry-ruby/lib/sentry-ruby.rb +++ b/sentry-ruby/lib/sentry-ruby.rb @@ -29,6 +29,7 @@ require "sentry/cron/monitor_check_ins" require "sentry/vernier/profiler" require "sentry/metrics" +require "sentry/job_extensions" [ "sentry/rake", @@ -63,6 +64,8 @@ module Sentry exception_locals_tp ].freeze + JOB_EXTENSIONS = JobExtensions::Registry.new + class << self # @!visibility private def exception_locals_tp @@ -148,6 +151,27 @@ def register_integration(name, version) integrations[name.to_s] = meta end + ##### Job Extensions ##### + + # Returns the process-wide registry of background-job extensions. + # + # @return [JobExtensions::Registry] + def job_extensions + JOB_EXTENSIONS + end + + # Registers a background-job extension built by a backend gem. It can be + # called at load time, before or after {Sentry.init}. + # + # @param extension [JobExtensions::Extension] an extension instance, + # constructed with its own configuration and collaborators + # @return [JobExtensions::Extension] the registered extension + # @raise [ArgumentError] if the argument is not a {JobExtensions::Extension} + # instance, or an extension with the same name is already registered + def register_job_extension(extension) + job_extensions.register(extension) + end + ##### Method Delegation ##### extend Forwardable diff --git a/sentry-ruby/lib/sentry/job_extensions.rb b/sentry-ruby/lib/sentry/job_extensions.rb new file mode 100644 index 000000000..de3b1bfcc --- /dev/null +++ b/sentry-ruby/lib/sentry/job_extensions.rb @@ -0,0 +1,6 @@ +# frozen_string_literal: true + +require "sentry/job_extensions/validators" +require "sentry/job_extensions/hooks" +require "sentry/job_extensions/extension" +require "sentry/job_extensions/registry" diff --git a/sentry-ruby/lib/sentry/job_extensions/extension.rb b/sentry-ruby/lib/sentry/job_extensions/extension.rb new file mode 100644 index 000000000..c1cfc852e --- /dev/null +++ b/sentry-ruby/lib/sentry/job_extensions/extension.rb @@ -0,0 +1,246 @@ +# frozen_string_literal: true + +module Sentry + module JobExtensions + # Base class for objects that let a background-job backend enrich Sentry's + # job integrations. + # + # A backend gem subclasses this class, receives everything the extension + # depends on (configuration, collaborators) through its constructor, and + # registers one instance with {Sentry.register_job_extension}. Sentry never + # instantiates extensions itself and treats jobs as opaque objects: only the + # extension interprets the job it is given. + # + # @example Defining and registering an extension + # class MyBackend::SentryJobExtension < Sentry::JobExtensions::Extension + # def initialize(config:) + # super(name: "my_backend") + # @config = config + # end + # + # def handles?(job) + # job.class.queue_adapter_name.to_s == "my_backend" + # end + # end + # + # Sentry.register_job_extension(MyBackend::SentryJobExtension.new(config: MyBackend.config)) + class Extension + # The extension's unique name. No two registered extensions can share it. + # + # @return [String] + attr_reader :name, :hooks + + # @param name [String, Symbol] a non-empty name identifying the backend, + # stored as a frozen String + # @raise [ArgumentError] if the name is not a non-empty String or Symbol + def initialize(name:, validators: Validators.new) + unless (name.is_a?(String) || name.is_a?(Symbol)) && !name.empty? + raise ArgumentError, "job extension name must be a non-empty String or Symbol, got #{name.inspect}" + end + + @name = name.to_s.dup.freeze + @hooks = Hooks.new(validators: validators) + end + + # Whether this extension handles the given job. Override it to claim + # the jobs run by the extension's backend. + # + # If this method raises a StandardError, the failure is logged through the + # SDK logger and the extension is treated as not handling the job. + # + # @param job [Object] the job, as passed by the job integration + # @return [Boolean] false unless overridden + def handles?(job) + false + end + + # @!group Hooks + # + # Hooks are called by job integrations through {Registry#call}, only on + # the extension that handles the job. Each hook maps its input to data + # and must not mutate the objects it is given; the job integration + # applies the returned data. Override only the hooks the backend can + # supply: the defaults return the fallback result, which makes the + # integration keep its own behavior. If a hook raises a StandardError or + # returns data that does not match its documented shape, the failure is + # logged through the SDK logger and the fallback result is used instead. + # Hash keys beyond those documented for a hook are dropped. + + # Data to apply to the scope the job is performed in. Called before the + # job is performed. A supplied user replaces the user the job + # integration sets on the scope. + # + # @param job [Object] the job being performed + # @return [Hash] a Hash with any of the keys +:contexts+ (a Hash of + # context name to Hash), +:tags+ (a Hash), +:extra+ (a Hash) and + # +:user+ (a Hash). The fallback result is an empty Hash. + def scope_data(job:) + {} + end + + # Data to add to the event captured for an exception raised by the job. + # Called when the job fails and its exception is about to be captured. + # + # Supplied tags and extra never replace a key already on the job's + # scope. Each supplied context is merged into the scope's context of the + # same name, keeping values already on the scope. Data the job integration captures + # explicitly with the exception takes precedence over this data. + # + # @param job [Object] the job that failed + # @param exception [Exception] the exception raised by the job + # @return [Hash] a Hash with any of the keys +:contexts+ (a Hash of + # context name to Hash), +:tags+ (a Hash) and +:extra+ (a Hash). + # The fallback result is an empty Hash. + def error_data(job:, exception:) + {} + end + + # The backend's own retry state for the job. Called when the job is + # enqueued, before it is performed, and when it fails. Only +:count+ is + # used when the job is enqueued or about to be performed; +:will_retry+ + # is used only when the job fails, to decide how the failure is + # reported. The hook may be called several times for the same attempt, + # so it must not change any state. + # + # @param job [Object] the job being enqueued, performed, or that failed + # @return [Hash, nil] a Hash with both the keys +:count+ (a + # non-negative Integer, the number of retries already attempted) and + # +:will_retry+ (true or false, whether the backend will retry the + # failed attempt). The fallback result is nil, meaning the backend's + # retry state is unknown. + def retry_info(job:) + nil + end + + # Whether the job integration captures the exception raised by the job. + # Called when the job fails. Declining to capture does not change the + # job's outcome: the job still fails and its transaction is still + # finished as failed. + # + # @param job [Object] the job that failed + # @param exception [Exception] the exception raised by the job + # @return [Boolean, nil] true to capture the exception, false not to + # capture it. The fallback result is nil, which captures the exception + # only when the backend will not retry the job, as reported by + # +:will_retry+ from #retry_info. + def report_exception?(job:, exception:) + nil + end + + # Whether the job integration captures an exception that the job + # integration's own retry mechanism rescued and retries, so the backend + # never sees it. Called when the job integration schedules the retry. + # + # @param job [Object] the job being retried + # @param exception [Exception] the exception the job raised + # @return [Boolean, nil] true to capture the exception, false not to + # capture it. The fallback result is nil, which keeps the job + # integration's own setting. + def report_retried_exception?(job:, exception:) + nil + end + + # Adds trace propagation headers to the backend's job payload. Called + # when a job is enqueued. + # + # @param payload [Object] the backend's job payload + # @param headers [Hash{String => String}] the trace propagation headers + # to carry, such as +sentry-trace+ and +baggage+ + # @return [Object] a new payload of the same class as the one given, + # carrying the headers. The fallback result is the given payload, + # unchanged. + def inject_trace_headers(job:, payload:, headers:) + payload + end + + # Reads the trace propagation headers carried by the backend's job + # payload. Called before the job is performed, to continue the trace + # started when it was enqueued. Only called when the job integration + # has no trace headers of its own for the job. + # + # @param payload [Object] the backend's job payload + # @return [Hash{String => String}, nil] the headers the payload carries. + # The job continues the trace they describe, and an empty Hash makes + # the job start a new trace. The fallback result is nil, which keeps + # the job integration's own trace handling. + def extract_trace_headers(job:, payload:) + nil + end + + # Data for the messaging attributes of the producer span created when the + # job is enqueued and the consumer span created when it is performed. On + # the consumer span, supplied values replace the job integration's own; + # on the producer span, they fill only attributes the job integration + # does not set. + # + # @param job [Object] the job being enqueued or performed + # @return [Hash] a Hash with any of the keys +:system+ (a String naming + # the backend), +:message_id+ (a String), +:destination+ (a String, + # such as the queue name), +:receive_latency+ (a Numeric, the + # milliseconds between enqueueing and the start of the job). The fallback result is an empty Hash. + def messaging_data(job:) + {} + end + + # The name of the transaction the job is performed in. Called before the + # job is performed. Errors captured while the job runs carry the same + # transaction name. + # + # @param job [Object] the job being performed + # @return [String, nil] a non-empty String. The fallback result is nil, + # which keeps the name the job integration gives the transaction. + def transaction_name(job:) + nil + end + + # The origin of the producer span created when the job is enqueued and + # of the consumer transaction created when it is performed. + # + # @param job [Object] the job being enqueued or performed + # @return [String, nil] a non-empty String, such as + # +"auto.queue.my_backend"+. The fallback result is nil, which keeps the + # job integration's own origin. + def span_origin(job:) + nil + end + + # The description of the producer span created when the job is enqueued. + # + # @param job [Object] the job being enqueued + # @return [String, nil] a non-empty String. The fallback result is nil, + # which keeps the description the job integration gives the span. + def publish_description(job:) + nil + end + + # The integration the job's exceptions are captured under. Called when + # the job fails and its exception is about to be captured. The + # integration's name is recorded on the event and as its exception + # mechanism. + # + # @param job [Object] the job that failed + # @return [Sentry::Integrable, nil] an object extended with + # {Sentry::Integrable}, such as a registered integration module. The + # fallback result is nil, which keeps the job integration's own + # integration. + def integration(job:) + nil + end + + # Whether the backend instruments enqueueing the job itself. Called when + # the job is enqueued. When true, the job integration records no + # producer span for the job and adds no trace propagation headers or + # user to the job's payload, leaving both to the backend. + # + # @param job [Object] the job being enqueued + # @return [Boolean, nil] true when the backend instruments enqueueing + # the job. The fallback result is nil, which keeps the job + # integration's own producer instrumentation. + def instruments_enqueue?(job:) + nil + end + + # @!endgroup + end + end +end diff --git a/sentry-ruby/lib/sentry/job_extensions/hooks.rb b/sentry-ruby/lib/sentry/job_extensions/hooks.rb new file mode 100644 index 000000000..59f284f26 --- /dev/null +++ b/sentry-ruby/lib/sentry/job_extensions/hooks.rb @@ -0,0 +1,67 @@ +# frozen_string_literal: true + +require "sentry/integrable" + +module Sentry + module JobExtensions + # @!visibility private + class Hooks + MALFORMED = Validators::MALFORMED + + class Contract + def initialize(validator, fallback:) + @validator = validator + @fallback = fallback + freeze + end + + def fallback_result(arguments) + @fallback.call(**arguments) + end + + def normalize(validators, result, arguments) + validators.public_send(@validator, result, **arguments) + end + end + + EMPTY_HASH = ->(**) { {} }.freeze + NO_OVERRIDE = ->(**) { nil }.freeze + ORIGINAL_PAYLOAD = ->(payload:, **) { payload }.freeze + + CONTRACTS = { + scope_data: Contract.new(:scope_data, fallback: EMPTY_HASH), + error_data: Contract.new(:event_data, fallback: EMPTY_HASH), + messaging_data: Contract.new(:messaging_data, fallback: EMPTY_HASH), + retry_info: Contract.new(:retry_info, fallback: NO_OVERRIDE), + inject_trace_headers: Contract.new(:payload, fallback: ORIGINAL_PAYLOAD), + extract_trace_headers: Contract.new(:trace_headers, fallback: NO_OVERRIDE), + integration: Contract.new(:integration, fallback: NO_OVERRIDE) + }.merge( + %i[report_exception? report_retried_exception? instruments_enqueue?] + .to_h { |hook| [hook, Contract.new(:decision, fallback: NO_OVERRIDE)] }, + %i[transaction_name span_origin publish_description] + .to_h { |hook| [hook, Contract.new(:label, fallback: NO_OVERRIDE)] } + ).freeze + + def initialize(validators: Validators.new) + @validators = validators + end + + def normalize(hook, result, arguments) + fetch(hook).normalize(@validators, result, arguments) + end + + def fallback_result(hook, arguments) + fetch(hook).fallback_result(arguments) + end + + private + + def fetch(hook) + CONTRACTS.fetch(hook) do + raise ArgumentError, "unknown job extension hook #{hook.inspect}, expected one of #{CONTRACTS.keys.inspect}" + end + end + end + end +end diff --git a/sentry-ruby/lib/sentry/job_extensions/registry.rb b/sentry-ruby/lib/sentry/job_extensions/registry.rb new file mode 100644 index 000000000..4103b4ecb --- /dev/null +++ b/sentry-ruby/lib/sentry/job_extensions/registry.rb @@ -0,0 +1,120 @@ +# frozen_string_literal: true + +module Sentry + module JobExtensions + # Holds the registered {Extension} instances, in registration order, and + # selects the one that handles a given job. + # + # The registry is process-wide and available through {Sentry.job_extensions}. + # It is independent of {Sentry.init}: extensions can be registered before or + # after the SDK is initialized, and stay registered when it is re-initialized. + class Registry + include LoggingHelper + + def initialize + @mutex = Mutex.new + @extensions = [].freeze + @hooks = Hooks.new + end + + # Registers an extension. + # + # @param extension [Extension] an instance of {Extension} or a subclass of it + # @return [Extension] the registered extension + # @raise [ArgumentError] if the argument is not an {Extension} instance, or + # an extension with the same name is already registered. Existing + # registrations are left unchanged. + def register(extension) + unless extension.is_a?(Extension) + raise ArgumentError, "job extension must be an instance of #{Extension}, got #{extension.inspect}" + end + + @mutex.synchronize do + if @extensions.any? { |registered| registered.name == extension.name } + raise ArgumentError, "a job extension named #{extension.name.inspect} is already registered" + end + + @extensions = (@extensions + [extension]).freeze + end + + extension + end + + # Returns the extension that handles the given job. + # + # Extensions are asked through {Extension#handles?} in registration order, + # and the first one that answers truthy is returned. An extension whose + # {Extension#handles?} raises a StandardError is logged through the SDK + # logger and skipped. Exceptions outside StandardError propagate. + # + # @param job [Object] the job, passed as-is to each extension + # @return [Extension, nil] the earliest registered extension that handles + # the job, or nil if none does + def for(job) + @extensions.find { |extension| handles?(extension, job) } + end + + # Calls a hook on an extension and returns its result, or the hook's + # fallback result when the extension cannot supply valid data. + # + # The fallback result is returned when the extension is nil, when the hook + # raises a StandardError, or when it returns data that does not match the + # shape documented on the corresponding {Extension} method. Keys a Hash + # result carries beyond those documented are dropped. Failures and + # malformed results are logged through the SDK logger. Exceptions outside + # StandardError propagate. + # + # Arguments are passed to the extension as given. Not mutating them is + # the extension's obligation, documented on {Extension}. + # + # @example Applying an extension's scope data + # extension = Sentry.job_extensions.for(job) + # data = Sentry.job_extensions.call(extension, :scope_data, job: job) + # scope.set_contexts(data[:contexts]) if data[:contexts] + # + # @param extension [Extension, nil] the extension returned by {#for} + # @param hook [Symbol] the name of a hook method documented on {Extension} + # @param arguments [Array] the arguments documented on the hook's + # {Extension} method + # @return [Object] the extension's result or the hook's fallback result + # @raise [ArgumentError] if the hook is not a hook documented on {Extension} + def call(extension, hook, **arguments) + hooks = extension ? extension.hooks : @hooks + fallback_result = hooks.fallback_result(hook, arguments) + return fallback_result unless extension + + invoke(extension, hook, hooks, arguments, fallback_result) + end + + # @!visibility private + def clear + @mutex.synchronize { @extensions = [].freeze } + end + + private + + def handles?(extension, job) + extension.handles?(job) + rescue StandardError => e + log_error("Job extension #{extension.name.inspect} failed to check whether it handles a job", e) + false + end + + def invoke(extension, hook, hooks, arguments, fallback_result) + result = extension.public_send(hook, **arguments) + normalized = hooks.normalize(hook, result, arguments) + return normalized unless normalized.equal?(Hooks::MALFORMED) + + log_warn("Job extension #{extension.name.inspect} returned malformed data from #{hook}: a #{result.class}") + fallback_result + rescue StandardError => e + log_error("Job extension #{extension.name.inspect} failed in #{hook}", e) + fallback_result + end + + def sdk_logger + Sentry.sdk_logger if Sentry.initialized? + end + end + end +end diff --git a/sentry-ruby/lib/sentry/job_extensions/validators.rb b/sentry-ruby/lib/sentry/job_extensions/validators.rb new file mode 100644 index 000000000..3521c48cb --- /dev/null +++ b/sentry-ruby/lib/sentry/job_extensions/validators.rb @@ -0,0 +1,100 @@ +# frozen_string_literal: true + +require "sentry/integrable" + +module Sentry + module JobExtensions + class Validators + MALFORMED = Object.new.freeze + + EVENT_DATA_KEYS = %i[contexts tags extra].freeze + SCOPE_DATA_KEYS = (EVENT_DATA_KEYS + %i[user]).freeze + + MESSAGING_DATA_VALIDATORS = { + system: ->(value) { value.is_a?(String) }, + message_id: ->(value) { value.is_a?(String) }, + destination: ->(value) { value.is_a?(String) }, + receive_latency: ->(value) { value.is_a?(Numeric) } + }.freeze + + def initialize(event_keys: EVENT_DATA_KEYS, scope_keys: SCOPE_DATA_KEYS, messaging_fields: MESSAGING_DATA_VALIDATORS) + @event_keys = event_keys.equal?(EVENT_DATA_KEYS) ? event_keys : event_keys.dup.freeze + @scope_keys = scope_keys.equal?(SCOPE_DATA_KEYS) ? scope_keys : scope_keys.dup.freeze + @messaging_fields = messaging_fields.equal?(MESSAGING_DATA_VALIDATORS) ? messaging_fields : messaging_fields.dup.freeze + end + + def known_entries(result, keys) + result.select { |key, _value| keys.include?(key) } + end + + def scope_data(result, **) + normalize_event_data(result, @scope_keys) + end + + def event_data(result, **) + normalize_event_data(result, @event_keys) + end + + def payload(result, payload:, **) + result.is_a?(payload.class) ? result : MALFORMED + end + + def normalize_event_data(result, keys) + return MALFORMED unless result.is_a?(Hash) + + data = known_entries(result, keys) + return MALFORMED unless data.values.all? { |value| value.is_a?(Hash) } + return MALFORMED unless data.fetch(:contexts, {}).values.all? { |context| context.is_a?(Hash) } + + data + end + + def retry_info(result, **) + return if result.nil? + return MALFORMED unless result.is_a?(Hash) + + count = result[:count] + will_retry = result[:will_retry] + return MALFORMED unless count.is_a?(Integer) && count >= 0 && [true, false].include?(will_retry) + + { count: count, will_retry: will_retry } + end + + def trace_headers(result, **) + return if result.nil? + return MALFORMED unless result.is_a?(Hash) && result.all? { |key, value| key.is_a?(String) && value.is_a?(String) } + + result + end + + def messaging_data(result, **) + return MALFORMED unless result.is_a?(Hash) + + data = known_entries(result, @messaging_fields.keys) + return MALFORMED unless data.all? { |key, value| @messaging_fields.fetch(key).call(value) } + + data + end + + def label(result, **) + return if result.nil? + return MALFORMED unless result.is_a?(String) && !result.empty? + + result + end + + def decision(result, **) + return result if result.nil? || result == true || result == false + + MALFORMED + end + + def integration(result, **) + return if result.nil? + return MALFORMED unless result.is_a?(Integrable) + + result + end + end + end +end diff --git a/sentry-ruby/spec/sentry/job_extensions_spec.rb b/sentry-ruby/spec/sentry/job_extensions_spec.rb new file mode 100644 index 000000000..eb4a3d74b --- /dev/null +++ b/sentry-ruby/spec/sentry/job_extensions_spec.rb @@ -0,0 +1,468 @@ +# frozen_string_literal: true + +RSpec.describe "Sentry job extensions" do + let(:extension_class) do + Class.new(Sentry::JobExtensions::Extension) do + def initialize(name:, claim_check:) + super(name: name) + @claim_check = claim_check + end + + def handles?(job) + @claim_check.call(job) + end + end + end + + let(:sidekiq_job) { double("job", adapter: "sidekiq") } + let(:good_job_job) { double("job", adapter: "good_job") } + + def build_extension(name, claims: ->(job) { job.adapter == name }) + extension_class.new(name: name, claim_check: claims) + end + + after { Sentry.job_extensions.clear } + + describe ".register_job_extension" do + it "makes an extension available before Sentry is initialized" do + extension = build_extension("sidekiq") + + Sentry.register_job_extension(extension) + + expect(Sentry.initialized?).to eq(false) + expect(Sentry.job_extensions.for(sidekiq_job)).to be(extension) + end + + it "makes an extension available after Sentry is initialized" do + perform_basic_setup + extension = build_extension("sidekiq") + + Sentry.register_job_extension(extension) + + expect(Sentry.job_extensions.for(sidekiq_job)).to be(extension) + end + + it "keeps registrations across Sentry re-initialization" do + extension = build_extension("sidekiq") + Sentry.register_job_extension(extension) + + perform_basic_setup + + expect(Sentry.job_extensions.for(sidekiq_job)).to be(extension) + end + + it "rejects an extension whose name is already registered and keeps the original" do + original = build_extension("sidekiq") + Sentry.register_job_extension(original) + + expect { Sentry.register_job_extension(build_extension("sidekiq")) } + .to raise_error(ArgumentError, /sidekiq/) + expect(Sentry.job_extensions.for(sidekiq_job)).to be(original) + end + + it "rejects an object that is not an extension instance and keeps existing registrations" do + extension = build_extension("sidekiq") + Sentry.register_job_extension(extension) + duck = double("duck", name: "good_job", handles?: true) + + expect { Sentry.register_job_extension(duck) }.to raise_error(ArgumentError) + expect { Sentry.register_job_extension(extension_class) }.to raise_error(ArgumentError) + expect(Sentry.job_extensions.for(sidekiq_job)).to be(extension) + expect(Sentry.job_extensions.for(good_job_job)).to be_nil + end + end + + describe "Sentry::JobExtensions::Extension" do + it "rejects a blank name" do + expect { build_extension("") }.to raise_error(ArgumentError) + expect { build_extension(nil) }.to raise_error(ArgumentError) + end + + it "does not claim any job by default" do + Sentry.register_job_extension(Sentry::JobExtensions::Extension.new(name: "noop")) + + expect(Sentry.job_extensions.for(sidekiq_job)).to be_nil + end + end + + describe "#for" do + it "returns the extension that claims the job" do + sidekiq = build_extension("sidekiq") + good_job = build_extension("good_job") + Sentry.register_job_extension(sidekiq) + Sentry.register_job_extension(good_job) + + expect(Sentry.job_extensions.for(good_job_job)).to be(good_job) + end + + it "returns nil when no extension claims the job" do + Sentry.register_job_extension(build_extension("sidekiq")) + + expect(Sentry.job_extensions.for(good_job_job)).to be_nil + end + + it "returns the earliest registered extension when several claim the job" do + first = build_extension("first", claims: ->(_job) { true }) + second = build_extension("second", claims: ->(_job) { true }) + Sentry.register_job_extension(first) + Sentry.register_job_extension(second) + + expect(Sentry.job_extensions.for(sidekiq_job)).to be(first) + end + + context "when a claim check raises a StandardError" do + let(:string_io) { StringIO.new } + let(:broken) { build_extension("broken", claims: ->(_job) { raise "claim failed" }) } + + before do + perform_basic_setup { |config| config.sdk_logger = Logger.new(string_io) } + end + + it "logs the failure and considers the remaining extensions" do + sidekiq = build_extension("sidekiq") + Sentry.register_job_extension(broken) + Sentry.register_job_extension(sidekiq) + + expect(Sentry.job_extensions.for(sidekiq_job)).to be(sidekiq) + expect(string_io.string).to include("broken", "claim failed") + end + + it "returns nil when no other extension claims the job" do + Sentry.register_job_extension(broken) + + expect(Sentry.job_extensions.for(sidekiq_job)).to be_nil + end + end + + it "treats a raising claim check as not claiming when Sentry is not initialized" do + Sentry.register_job_extension(build_extension("broken", claims: ->(_job) { raise "claim failed" })) + + expect(Sentry.job_extensions.for(sidekiq_job)).to be_nil + end + + it "propagates exceptions outside StandardError from a claim check" do + Sentry.register_job_extension(build_extension("interrupted", claims: ->(_job) { raise Interrupt })) + + expect { Sentry.job_extensions.for(sidekiq_job) }.to raise_error(Interrupt) + end + end + + describe "#call" do + let(:hooked_extension_class) do + Class.new(Sentry::JobExtensions::Extension) do + def initialize(name:, hooks:) + super(name: name) + hooks.each { |hook, implementation| define_singleton_method(hook, &implementation) } + end + end + end + + let(:job) { double("job", jid: "abc123", queue: "default", retry_count: 2) } + let(:exception) { RuntimeError.new("job failed") } + let(:payload) { { "class" => "HardJob", "args" => [1] } } + let(:headers) { { "sentry-trace" => "trace-value", "baggage" => "baggage-value" } } + let(:string_io) { StringIO.new } + let(:backend_integration) do + Module.new do + extend Sentry::Integrable + + def self.integration_name + "backend" + end + end + end + + let(:hook_arguments) do + { + scope_data: { job: job }, + error_data: { job: job, exception: exception }, + retry_info: { job: job }, + report_retried_exception?: { job: job, exception: exception }, + inject_trace_headers: { job: job, payload: payload, headers: headers }, + extract_trace_headers: { job: job, payload: payload }, + messaging_data: { job: job }, + transaction_name: { job: job }, + span_origin: { job: job }, + publish_description: { job: job }, + integration: { job: job }, + instruments_enqueue?: { job: job } + } + end + + let(:fallback_results) do + { + scope_data: {}, + error_data: {}, + retry_info: nil, + report_retried_exception?: nil, + inject_trace_headers: payload, + extract_trace_headers: nil, + messaging_data: {}, + transaction_name: nil, + span_origin: nil, + publish_description: nil, + integration: nil, + instruments_enqueue?: nil + } + end + + def build_hooked_extension(**hooks) + hooked_extension_class.new(name: "sidekiq", hooks: hooks) + end + + def call_hook(extension, hook) + Sentry.job_extensions.call(extension, hook, **hook_arguments.fetch(hook)) + end + + def call_every_hook(extension) + hook_arguments.keys.to_h { |hook| [hook, call_hook(extension, hook)] } + end + + before do + perform_basic_setup { |config| config.sdk_logger = Logger.new(string_io) } + end + + it "uses custom validators without changing validation for other extensions" do + validators = Class.new(Sentry::JobExtensions::Validators) do + def label(result, **) + super(result.is_a?(Symbol) ? result.to_s : result) + end + end.new + custom = Class.new(Sentry::JobExtensions::Extension) do + def transaction_name(job:) + :CustomJob + end + end + extension = custom.new(name: "custom", validators: validators) + default_extension = custom.new(name: "default") + + expect(call_hook(extension, :transaction_name)).to eq("CustomJob") + expect(call_hook(default_extension, :transaction_name)).to be_nil + expect(call_hook(nil, :transaction_name)).to be_nil + end + + it "uses configured metadata keys without following later changes to the supplied rules" do + scope_keys = [:tags] + event_keys = [:extra] + validators = Sentry::JobExtensions::Validators.new(scope_keys: scope_keys, event_keys: event_keys) + custom = Class.new(Sentry::JobExtensions::Extension) do + def scope_data(job:) + { tags: { queue: "default" }, user: { id: "42" } } + end + + def error_data(job:, exception:) + { tags: { failed: true }, extra: { attempt: 2 } } + end + end + extension = custom.new(name: "custom", validators: validators) + scope_keys << :user + event_keys << :tags + + expect(call_hook(extension, :scope_data)).to eq(tags: { queue: "default" }) + expect(call_hook(extension, :error_data)).to eq(extra: { attempt: 2 }) + expect(call_hook(custom.new(name: "default"), :scope_data)) + .to eq(tags: { queue: "default" }, user: { id: "42" }) + end + + it "uses configured messaging validators without changing other extensions or following rule mutations" do + fields = Sentry::JobExtensions::Validators::MESSAGING_DATA_VALIDATORS.merge( + message_id: ->(value) { value.is_a?(String) && !value.empty? } + ) + validators = Sentry::JobExtensions::Validators.new(messaging_fields: fields) + custom = Class.new(Sentry::JobExtensions::Extension) do + def messaging_data(job:) + { system: "sidekiq", message_id: job } + end + end + extension = custom.new(name: "custom", validators: validators) + default_extension = custom.new(name: "default") + fields[:message_id] = ->(**) { true } + + expect(Sentry.job_extensions.call(extension, :messaging_data, job: "abc123")) + .to eq(system: "sidekiq", message_id: "abc123") + expect(Sentry.job_extensions.call(extension, :messaging_data, job: "")).to eq({}) + expect(Sentry.job_extensions.call(default_extension, :messaging_data, job: "")) + .to eq(system: "sidekiq", message_id: "") + end + + it "returns the scope contexts and tags an extension derives from the job" do + extension = build_hooked_extension( + scope_data: lambda do |job:| + { contexts: { sidekiq: { jid: job.jid } }, tags: { queue: job.queue } } + end + ) + + expect(call_hook(extension, :scope_data)) + .to eq(contexts: { sidekiq: { jid: "abc123" } }, tags: { queue: "default" }) + end + + it "returns the event tags, extra and contexts an extension derives from the job and exception" do + extension = build_hooked_extension( + error_data: lambda do |job:, exception:| + { + tags: { queue: job.queue }, + extra: { error_message: exception.message }, + contexts: { sidekiq: { jid: job.jid } } + } + end + ) + + expect(call_hook(extension, :error_data)).to eq( + tags: { queue: "default" }, + extra: { error_message: "job failed" }, + contexts: { sidekiq: { jid: "abc123" } } + ) + end + + it "returns the retry count and policy an extension derives from the job" do + extension = build_hooked_extension( + retry_info: ->(job:) { { count: job.retry_count, will_retry: job.retry_count < 5 } } + ) + + expect(call_hook(extension, :retry_info)).to eq(count: 2, will_retry: true) + end + + it "returns a payload carrying the trace headers" do + extension = build_hooked_extension( + inject_trace_headers: ->(job:, payload:, headers:) { payload.merge("sentry" => headers) } + ) + + expect(call_hook(extension, :inject_trace_headers)) + .to eq("class" => "HardJob", "args" => [1], "sentry" => headers) + end + + it "returns the trace headers an extension reads from the payload" do + carrying_payload = { "class" => "HardJob", "sentry" => headers } + extension = build_hooked_extension(extract_trace_headers: ->(job:, payload:) { payload["sentry"] }) + + expect(Sentry.job_extensions.call(extension, :extract_trace_headers, job: job, payload: carrying_payload)).to eq(headers) + end + + it "returns the messaging span data an extension derives from the job" do + extension = build_hooked_extension( + messaging_data: lambda do |job:| + { system: "sidekiq", message_id: job.jid, destination: job.queue, receive_latency: 1500 } + end + ) + + expect(call_hook(extension, :messaging_data)) + .to eq(system: "sidekiq", message_id: "abc123", destination: "default", receive_latency: 1500) + end + + it "returns the names an extension gives the job and the integration it captures errors under" do + integration = backend_integration + extension = build_hooked_extension( + transaction_name: ->(job:) { "Backend/#{job.queue}" }, + span_origin: ->(job:) { "auto.queue.backend" }, + publish_description: ->(job:) { "publish #{job.jid}" }, + integration: ->(job:) { integration } + ) + + expect(call_every_hook(extension)).to include( + transaction_name: "Backend/default", + span_origin: "auto.queue.backend", + publish_description: "publish abc123", + integration: backend_integration + ) + end + + it "returns whether an extension instruments enqueueing the job itself" do + extension = build_hooked_extension(instruments_enqueue?: ->(job:) { job.queue == "default" }) + + expect(call_hook(extension, :instruments_enqueue?)).to be(true) + end + + it "returns the fallback result of every hook an extension does not override" do + extension = build_hooked_extension(scope_data: ->(job:) { { tags: { queue: "default" } } }) + + expect(call_every_hook(extension)) + .to eq(fallback_results.merge(scope_data: { tags: { queue: "default" } })) + end + + it "returns the fallback result of every hook for a bare extension" do + extension = Sentry::JobExtensions::Extension.new(name: "noop") + + expect(call_every_hook(extension)).to eq(fallback_results) + end + + it "returns the fallback result of every hook when no extension claims the job" do + expect(call_every_hook(nil)).to eq(fallback_results) + end + + it "rejects a hook that is not part of the contract" do + expect { Sentry.job_extensions.call(build_hooked_extension, :handles?, job: job) } + .to raise_error(ArgumentError, /handles\?/) + end + + it "logs and returns the fallback result when a hook raises a StandardError" do + failing_hooks = hook_arguments.keys.to_h { |hook| [hook, ->(**) { raise "#{hook} exploded" }] } + extension = build_hooked_extension(**failing_hooks) + + expect(call_every_hook(extension)).to eq(fallback_results) + hook_arguments.each_key do |hook| + expect(string_io.string).to include("sidekiq", hook.to_s, "#{hook} exploded") + end + end + + it "returns the fallback result when a hook raises and Sentry is not initialized" do + reset_sentry_globals! + extension = build_hooked_extension(scope_data: ->(job:) { raise "scope exploded" }) + + expect(call_hook(extension, :scope_data)).to eq({}) + end + + it "propagates exceptions outside StandardError from a hook" do + extension = build_hooked_extension(retry_info: ->(job:) { raise Interrupt }) + + expect { call_hook(extension, :retry_info) }.to raise_error(Interrupt) + end + + { + scope_data: [{ tags: { queue: "default" }, body: "x" }, { tags: { queue: "default" } }], + error_data: [{ extra: { attempt: 1 }, unknown: nil }, { extra: { attempt: 1 } }], + retry_info: [{ count: 2, will_retry: true, backoff: 30 }, { count: 2, will_retry: true }], + messaging_data: [{ system: "sidekiq", payload: "{}" }, { system: "sidekiq" }] + }.each do |hook, (result, known_data)| + it "keeps the documented data and drops unknown keys when #{hook} returns #{result.inspect}" do + extension = build_hooked_extension(hook => ->(**) { result }) + + expect(call_hook(extension, hook)).to eq(known_data) + expect(string_io.string).not_to include("malformed") + end + end + + it "does not write a malformed result into the log" do + extension = build_hooked_extension(inject_trace_headers: ->(job:, payload:, headers:) { payload["args"].map(&:to_s) }) + + expect(call_hook(extension, :inject_trace_headers)).to eq(payload) + expect(string_io.string).to include("sidekiq", "inject_trace_headers", "Array") + expect(string_io.string).not_to include('"1"') + end + + { + scope_data: [nil, [], { contexts: { sidekiq: "jid" } }, { tags: [] }, { user: "42" }], + error_data: [nil, "tags", { extra: nil }, { contexts: { sidekiq: 1 } }], + retry_info: [{}, 2, { count: "2", will_retry: true }, { count: -1, will_retry: true }, + { count: 2, will_retry: "yes" }, { count: 2 }], + inject_trace_headers: [nil, "payload"], + extract_trace_headers: ["sentry-trace", [], { "sentry-trace" => 1 }, { sentry_trace: "trace" }], + messaging_data: [nil, [], { system: :sidekiq }, { message_id: 1 }, { destination: nil }, + { receive_latency: "fast" }], + transaction_name: ["", :HardJob, 1], + span_origin: ["", :"auto.queue.backend"], + publish_description: ["", :HardJob], + integration: ["backend", Module.new, Kernel], + instruments_enqueue?: ["yes", 1], + report_retried_exception?: ["yes", 1] + }.each do |hook, malformed_results| + malformed_results.each do |malformed| + it "logs and returns the fallback result when #{hook} returns #{malformed.inspect}" do + extension = build_hooked_extension(hook => ->(**) { malformed }) + + expect(call_hook(extension, hook)).to eq(fallback_results.fetch(hook)) + expect(string_io.string).to include("sidekiq", hook.to_s) + end + end + end + end +end diff --git a/sentry-sidekiq/lib/sentry-sidekiq.rb b/sentry-sidekiq/lib/sentry-sidekiq.rb index 4faeffcd2..f51ab38c8 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,6 +15,12 @@ module Sidekiq register_integration name: "sidekiq", version: Sentry::Sidekiq::VERSION + def self.job_extension + @job_extension ||= JobExtension.new + end + + Sentry.register_job_extension(job_extension) + if defined?(::Rails::Railtie) class Railtie < ::Rails::Railtie config.after_initialize do 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..0e945cbce 100644 --- a/sentry-sidekiq/lib/sentry/sidekiq/error_handler.rb +++ b/sentry-sidekiq/lib/sentry/sidekiq/error_handler.rb @@ -1,11 +1,14 @@ # frozen_string_literal: true require "sentry/sidekiq/context_filter" +require "sentry/sidekiq/failure_reporting" 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) + @failure_reporting = failure_reporting + end # @param ex [Exception] the exception / error that occured # @param context [Hash or Array] Sidekiq error context @@ -16,83 +19,20 @@ 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(ContextFilter.new(context).transaction_name, 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 } + hint: { background: false }, + 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 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..196e71096 --- /dev/null +++ b/sentry-sidekiq/lib/sentry/sidekiq/job_extension.rb @@ -0,0 +1,94 @@ +# frozen_string_literal: true + +require "sentry/sidekiq/context_filter" +require "sentry/sidekiq/retry_policy" + +module Sentry + module Sidekiq + class JobExtension < Sentry::JobExtensions::Extension + SIDEKIQ_NAME = "Sidekiq" + WITH_SIDEKIQ_8 = ::Gem::Version.new(::Sidekiq::VERSION) >= ::Gem::Version.new("8.0.0") + + def initialize(retry_policy: RetryPolicy.build) + super(name: "sidekiq") + @retry_policy = retry_policy + end + + def handles?(job) + job.is_a?(Hash) && job.key?("jid") + end + + def scope_data(job:) + sidekiq_job = job + return {} unless sidekiq_job + + data = { + contexts: { sidekiq: ContextFilter.new(sidekiq_job).filtered }, + tags: { queue: sidekiq_job["queue"], jid: sidekiq_job["jid"] }.merge(job_tags(sidekiq_job)) + } + user = sidekiq_job["sentry_user"] + data[:user] = user if user.is_a?(Hash) + data + end + + def messaging_data(job:) + { + message_id: job["jid"], + destination: job["queue"], + receive_latency: receive_latency(job) + }.compact + end + + def retry_info(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 if payload["trace_propagation_headers"] + + payload.merge("trace_propagation_headers" => headers) + end + + def extract_trace_headers(job:, payload:) + payload["trace_propagation_headers"] + end + + def transaction_name(job:) + name = job_class(job) + "#{SIDEKIQ_NAME}/#{name}" if name + 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 + + private + + 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 + + if WITH_SIDEKIQ_8 + def receive_latency(job) + now_in_ms - job["enqueued_at"] if job["enqueued_at"] + end + else + def receive_latency(job) + ((Time.now.to_f - job["enqueued_at"]) * 1000).round if job["enqueued_at"] + end + end + + def now_in_ms + ::Process.clock_gettime(::Process::CLOCK_REALTIME, :millisecond) + 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..19b40a5e0 100644 --- a/sentry-sidekiq/lib/sentry/sidekiq/sentry_context_middleware.rb +++ b/sentry-sidekiq/lib/sentry/sidekiq/sentry_context_middleware.rb @@ -13,20 +13,6 @@ def set_span_data(span, id:, queue:, latency: nil, retry_count: nil) 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 - - 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 - end - - def now_in_ms - ::Process.clock_gettime(::Process::CLOCK_REALTIME, :millisecond) - end end class SentryContextServerMiddleware @@ -54,7 +40,7 @@ def call(worker, job, queue) if transaction scope.set_span(transaction) - latency = calculate_latency(job) + latency = Sentry::Sidekiq.job_extension.messaging_data(job: job)[:receive_latency] set_span_data( transaction, diff --git a/sentry-sidekiq/spec/sentry/sidekiq/failure_reporting_spec.rb b/sentry-sidekiq/spec/sentry/sidekiq/failure_reporting_spec.rb new file mode 100644 index 000000000..94cb541a1 --- /dev/null +++ b/sentry-sidekiq/spec/sentry/sidekiq/failure_reporting_spec.rb @@ -0,0 +1,63 @@ +# frozen_string_literal: true + +require "spec_helper" + +RSpec.describe Sentry::Sidekiq::FailureReporting do + subject(:reporting) { described_class.new } + + before do + perform_basic_setup + end + + def job(**attributes) + { "class" => "SadWorker", "jid" => "abc123", "queue" => "default" }.merge(attributes.transform_keys(&:to_s)) + end + + context "with config.report_after_job_retries = true" do + before do + Sentry.configuration.sidekiq.report_after_job_retries = true + end + + [false, 0].each do |retry_option| + it "reports the first failure when retry is #{retry_option}" do + expect(reporting.report?(job(retry: retry_option))).to be(true) + end + end + + if WITH_SIDEKIQ_7 + context "when the passed Sidekiq config sets max_retries" do + let(:config) { sidekiq_config(max_retries: 3) } + + it "skips failures before the configured retries are exhausted" do + expect(reporting.report?(job(retry: true, retry_count: 1), config)).to be(false) + end + + it "reports the failure once the configured retries are exhausted" do + expect(reporting.report?(job(retry: true, retry_count: 2), config)).to be(true) + end + end + end + end + + context "with attempt_threshold" do + [false, 0].each do |retry_option| + it "reports the first failure when retry is #{retry_option}" do + expect(reporting.report?(job(retry: retry_option, attempt_threshold: 3))).to be(true) + end + end + + if WITH_SIDEKIQ_7 + context "when the passed Sidekiq config sets max_retries below the threshold" do + let(:config) { sidekiq_config(max_retries: 2) } + + it "skips failures before the final attempt" do + expect(reporting.report?(job(retry: true, retry_count: 0, attempt_threshold: 5), config)).to be(false) + end + + it "reports the failure on the final attempt" do + expect(reporting.report?(job(retry: true, retry_count: 1, attempt_threshold: 5), config)).to be(true) + end + end + end + end +end diff --git a/sentry-sidekiq/spec/sentry/sidekiq/job_extension_spec.rb b/sentry-sidekiq/spec/sentry/sidekiq/job_extension_spec.rb new file mode 100644 index 000000000..0bae6fb99 --- /dev/null +++ b/sentry-sidekiq/spec/sentry/sidekiq/job_extension_spec.rb @@ -0,0 +1,94 @@ +# frozen_string_literal: true + +require "spec_helper" +require "timecop" +require "sidekiq/api" + +RSpec.describe Sentry::Sidekiq::JobExtension do + let(:queue) { Sidekiq::Queue.new("default") } + let(:trace_headers) { { "sentry-trace" => "#{"a" * 32}-#{"b" * 16}-1", "baggage" => "sentry-trace_id=#{"a" * 32}" } } + + def delivered_job(**options) + Sidekiq::Client.push({ "queue" => "default", "class" => HappyWorker, "args" => [] }.merge(options.transform_keys(&:to_s))) + queue.first.item.merge("queue" => "default") + end + + def call_hook(hook, job, *arguments) + keywords = { job: job } + keywords[:payload] = job if %i[inject_trace_headers extract_trace_headers].include?(hook) + keywords[:headers] = arguments.first if hook == :inject_trace_headers + Sentry.job_extensions.call(Sentry.job_extensions.for(job), hook, **keywords) + end + + before do + perform_basic_setup + queue.clear + end + + after do + queue.clear + end + + it "describes the delivered job's message id, queue and receive latency in milliseconds" do + job = delivered_job + + data = Timecop.freeze(Time.now + 10) do + Timecop.mock_process_clock = true + call_hook(:messaging_data, job) + ensure + Timecop.mock_process_clock = false + end + + expect(data).to include(message_id: job["jid"], destination: "default") + expect(data[:receive_latency]).to be_within(1_000).of(10_000) + end + + it "omits the receive latency of a job without an enqueue time" do + job = delivered_job.reject { |key, _value| key == "enqueued_at" } + + expect(call_hook(:messaging_data, job)).to eq(message_id: job["jid"], destination: "default") + end + + it "reports a first attempt as zero retries that Sidekiq will retry" do + expect(call_hook(:retry_info, delivered_job(retry: 1))).to eq(count: 0, will_retry: true) + end + + it "reports the retries a retried job has gone through" do + expect(call_hook(:retry_info, delivered_job(retry: 5, retry_count: 2))).to eq(count: 2, will_retry: true) + end + + it "reports that Sidekiq will not retry a job with retries disabled or exhausted" do + expect(call_hook(:retry_info, delivered_job(retry: false))).to eq(count: 0, will_retry: false) + expect(call_hook(:retry_info, delivered_job(retry: 3, retry_count: 3))).to eq(count: 3, will_retry: false) + end + + it "reports that Sidekiq will not retry the final delivered attempt" do + expect(call_hook(:retry_info, delivered_job(retry: 1, retry_count: 0))).to eq(count: 0, will_retry: false) + end + + it "describes the job as the sidekiq context with queue, jid and sidekiq tags" do + job = delivered_job(tags: ["marvel", "dc"]) + + expect(call_hook(:scope_data, job)).to eq( + contexts: { sidekiq: job }, + tags: { queue: "default", jid: job["jid"], "sidekiq.marvel": true, "sidekiq.dc": true } + ) + end + + it "carries trace propagation headers in a copy of the job" do + job = { "class" => "HappyWorker", "jid" => "123" } + + expect(call_hook(:inject_trace_headers, job, trace_headers)).to eq(job.merge("trace_propagation_headers" => trace_headers)) + expect(job).not_to have_key("trace_propagation_headers") + end + + it "keeps the trace propagation headers a job already carries" do + job = { "class" => "HappyWorker", "jid" => "123", "trace_propagation_headers" => trace_headers } + + expect(call_hook(:inject_trace_headers, job, { "sentry-trace" => "other" })).to eq(job) + end + + it "reads the trace propagation headers the job carries" do + expect(call_hook(:extract_trace_headers, delivered_job(trace_propagation_headers: trace_headers))).to eq(trace_headers) + end +end