Skip to content

feat(rails): run ActiveJob reporting through job extensions #3130

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
solnic wants to merge 2 commits into from
+1,059 −187
Open

feat(rails): run ActiveJob reporting through job extensions #3130

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
caf345a
feat(rails): run ActiveJob reporting through job extensions
solnic Nov 4, 2026
15f96fd
fix(rails): keep the _sentry payload when injecting trace headers fails
solnic Nov 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 45 additions & 0 deletions sentry-rails/lib/sentry/rails.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
require "sentry/integrable"
require "sentry/rails/tracing"
require "sentry/rails/configuration"
require "sentry/rails/active_job_extension"
require "sentry/rails/structured_logging"
require "sentry/rails/engine"
require "sentry/rails/railtie"
Expand All @@ -14,5 +15,49 @@ module Sentry
module Rails
extend Integrable
register_integration name: "rails", version: Sentry::Rails::VERSION

DEFAULT_ACTIVE_JOB_EXTENSION = ActiveJobExtension.new
private_constant :DEFAULT_ACTIVE_JOB_EXTENSION

# Returns whether sentry-rails reports ActiveJob jobs run through the given
# queue adapter, in which case the backend gem should not report them itself.
# The result depends on the current configuration, so check it when a job is
# handled rather than caching it.
#
# @param adapter_class_name [String, Class]
# @return [Boolean]
def self.active_job_owns_adapter?(adapter_class_name)
Sentry.initialized? && !Sentry.configuration.rails.skippable_job_adapters.include?(adapter_class_name.to_s)
end

# Registers the extension for ActiveJob jobs run through the given queue
# adapter. Raises ArgumentError if it is not an {ActiveJobExtension}.
#
# @param adapter [Symbol] the adapter name, such as `:sidekiq`
# @param extension [ActiveJobExtension]
# @return [ActiveJobExtension]
def self.register_active_job_extension(adapter, extension)
unless extension.is_a?(ActiveJobExtension)
raise ArgumentError, "ActiveJob extension must be an instance of #{ActiveJobExtension}, got #{extension.inspect}"
end

active_job_extensions[adapter.to_sym] = extension
end

# Returns the extension for the job class' queue adapter, or the default
# {ActiveJobExtension} when none is registered.
#
# @param job_class [Class]
# @return [ActiveJobExtension]
def self.active_job_extension_for(job_class)
active_job_extensions.fetch(ActiveJobExtension.adapter_name(job_class), DEFAULT_ACTIVE_JOB_EXTENSION)
end

# Returns the registered extensions by adapter name.
#
# @return [Hash{Symbol => ActiveJobExtension}]
def self.active_job_extensions
@active_job_extensions ||= {}
end
end
end
231 changes: 45 additions & 186 deletions sentry-rails/lib/sentry/rails/active_job.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
require "set"
require "sentry/rails/serializer"
require "sentry/rails/error_reporter_context"
require "sentry/rails/active_job_extension"

module Sentry
module Rails
Expand All @@ -13,77 +14,52 @@ module ActiveJobExtensions

def self.prepended(base)
base.attr_accessor :_sentry
base.attr_reader :_sentry_job_data
end

def perform_now
if !Sentry.initialized? || already_supported_by_sentry_integration?
super
else
data = _sentry || {}
SentryReporter.record(
self,
trace_headers: data["trace_propagation_headers"],
user: data["user"]
) { super }
SentryReporter.record(self) { super }
end
end

def serialize
payload = super
return payload if !Sentry.initialized? || already_supported_by_sentry_integration?

begin
sentry_data = {}
if Sentry.configuration.rails.active_job_propagate_traces
headers = Sentry.get_trace_propagation_headers
sentry_data["trace_propagation_headers"] = headers if headers && !headers.empty?
end

if Sentry.configuration.data_collection.user_info
user = Sentry.get_current_scope.user
allowed = user.transform_keys(&:to_s).slice(*USER_FIELDS_ALLOWLIST)
sentry_data["user"] = allowed unless allowed.empty?
end

payload[SENTRY_PAYLOAD_KEY] = sentry_data unless sentry_data.empty?
rescue StandardError => e
Sentry.sdk_logger&.error("sentry-rails: failed to inject _sentry payload: #{e.class}: #{e.message}\n #{Array(e.backtrace).first(5).join("\n ")}")
end

payload
SentryReporter.prepare_payload(self, payload)
end

def deserialize(job_data)
super
return if !Sentry.initialized? || already_supported_by_sentry_integration?

begin
@_sentry_job_data = job_data
self._sentry = job_data[SENTRY_PAYLOAD_KEY]
rescue StandardError => e
Sentry.sdk_logger&.error("sentry-rails: failed to extract _sentry payload: #{e.class}: #{e.message}\n #{Array(e.backtrace).first(5).join("\n ")}")
end
end

def already_supported_by_sentry_integration?
Sentry.configuration.rails.skippable_job_adapters.include?(self.class.queue_adapter.class.to_s)
!Sentry::Rails.active_job_owns_adapter?(self.class.queue_adapter.class)
end

class SentryReporter
OP_NAME = "queue.process"
SPAN_ORIGIN = "auto.queue.active_job"

# Emitted as messaging.system when the configured queue adapter's
# identity cannot be resolved from the job. ActiveJob is adapter-agnostic
# and supports arbitrary third-party backends, so the concrete value is
# normally derived from the adapter itself; this is only the last resort.
OP_NAME = Sentry::JobExtensions::Extension::PROCESS_OP
SPAN_ORIGIN = ActiveJobExtension::SPAN_ORIGIN
MESSAGING_SYSTEM_FALLBACK = "activejob"

EVENT_HANDLERS = {
"enqueue_retry.active_job" => :retry_handler
}

class << self
include ErrorReporterContext
def prepare_payload(job, payload)
Sentry::Rails.active_job_extension_for(job.class).prepare_payload(job: job, payload: payload)
end

def producer_callback_registered?
@producer_callback_registered ||= false
Expand All @@ -96,143 +72,38 @@ def producer_callback_registered!
def record_producer_span(job, &enqueue)
return yield if !Sentry.initialized? || job.already_supported_by_sentry_integration?

enqueued = false

run_enqueue = lambda do
enqueued = true
enqueue.call
end

begin
Sentry.with_child_span(op: "queue.publish", description: job.class.name) do |span|
if span
span.set_origin(SPAN_ORIGIN)
span.set_data(Sentry::Span::DataConventions::MESSAGING_SYSTEM, messaging_system(job))
span.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_ID, job.job_id)
span.set_data(Sentry::Span::DataConventions::MESSAGING_DESTINATION_NAME, job.queue_name)

if (count = retry_count(job))
span.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_RETRY_COUNT, count)
end
end

run_enqueue.call
end
rescue StandardError => e
raise if enqueued

log_producer_span_error(e)
Sentry::Rails.active_job_extension_for(job.class).instrument_enqueue(job: job, &enqueue)
end

run_enqueue.call
def record(job, &block)
with_isolated_scope do |scope|
record_in_scope(scope, job, &block)
end
end

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_in_scope(scope, job)
extension = Sentry::Rails.active_job_extension_for(job.class)
transaction = nil

def record(job, trace_headers: nil, user: nil, &block)
original_hub = Sentry.get_current_hub_internal
Sentry.clone_hub_to_current_thread
begin
transaction = extension.start_job(scope: scope, job: job)

Sentry.with_scope do |scope|
begin
scope.set_user(user.transform_keys(&:to_sym)) if user && !user.empty?

scope.set_transaction_name(job.class.name, source: :task)
scope.set_tags(queue: job.queue_name)

scope.set_contexts(active_job: {
job_class: job.class.name,
job_id: job.job_id,
queue: job.queue_name,
provider_job_id: job.provider_job_id
})

transaction_options = {
name: scope.transaction_name,
source: scope.transaction_source,
op: OP_NAME,
origin: SPAN_ORIGIN
}

transaction = if trace_headers && !trace_headers.empty?
continued = Sentry.continue_trace(trace_headers, **transaction_options)
Sentry.start_transaction(transaction: continued, **transaction_options)
else
Sentry.start_transaction(**transaction_options)
end

if transaction
set_messaging_data(transaction, job)
scope.set_span(transaction)
end

yield.tap do
finish_sentry_transaction(transaction, 200)
end
rescue Exception => e # rubocop:disable Lint/RescueException
finish_sentry_transaction(transaction, 500)

capture_exception(job, e)

raise
yield.tap do
extension.finish_job(transaction: transaction, status: 200)
end
end
ensure
Sentry.set_current_hub_internal(original_hub)
end
rescue Exception => e # rubocop:disable Lint/RescueException
extension.finish_job(transaction: transaction, status: 500)

def set_messaging_data(transaction, job)
transaction.set_data(Sentry::Span::DataConventions::MESSAGING_SYSTEM, messaging_system(job))
transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_ID, job.job_id)
transaction.set_data(Sentry::Span::DataConventions::MESSAGING_DESTINATION_NAME, job.queue_name)
if (count = retry_count(job))
transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_RETRY_COUNT, count)
end
if extension.safely("decide whether to report the job exception", fallback: true) { extension.report_exception?(job: job, exception: e) }
extension.capture_exception(job: job, exception: e)
end

if (latency = compute_latency(job))
transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_RECEIVE_LATENCY, latency)
raise
end
end

def messaging_system(job)
name = job.class.queue_adapter_name if job.class.respond_to?(:queue_adapter_name)
name = name.to_s
name.empty? ? MESSAGING_SYSTEM_FALLBACK : name
end

# Number of retries the job has already gone through, as observed at
# the moment the span is opened. On the consumer the span opens before
# ActiveJob increments +executions+, so a job's first attempt reads 0,
# its first retry reads 1, and so on. On the producer the retry enqueue
# runs after the failed attempt bumped +executions+, so it observes the
# same progression.
def retry_count(job)
job.executions.to_i if job.respond_to?(:executions)
end

def compute_latency(job)
return unless job.respond_to?(:enqueued_at) && job.enqueued_at

enqueued_time = job.enqueued_at.is_a?(String) ? Time.parse(job.enqueued_at) : job.enqueued_at
(Time.now.to_f - enqueued_time.to_f) * 1000.0
end

def capture_exception(job, e)
options = {
extra: sentry_context(job),
tags: {
job_id: job.job_id,
provider_job_id: job.provider_job_id
},
# Send synchronously: a worker process may exit before the async
# background worker flushes its queue, which would drop the event.
hint: { background: false }
}
options[:contexts] = execution_context if Sentry.configuration.data_collection.queues

Sentry::Rails.capture_exception(e, **options)
Sentry::Rails.active_job_extension_for(job.class).capture_exception(job: job, exception: e)
end

def register_event_handlers
Expand All @@ -255,9 +126,14 @@ 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

capture_exception(job, error)
extension = Sentry::Rails.active_job_extension_for(job.class)
report = extension.safely("decide whether to report the retried exception", fallback: Sentry.configuration.rails.active_job_report_on_retry_error) do
extension.report_retried_exception?(job: job, exception: error)
end
return unless report

extension.capture_exception(job: job, exception: error)
end
end

Expand All @@ -266,35 +142,18 @@ def handle_error_event(*args)
yield(event.payload[:job], event.payload[:error])
end

def finish_sentry_transaction(transaction, status)
return unless transaction
private :record_in_scope

transaction.set_http_status(status)
transaction.finish
end

def sentry_context(job)
context = {
active_job: job.class.name,
scheduled_at: job.scheduled_at,
job_id: job.job_id,
provider_job_id: job.provider_job_id,
locale: job.locale
}

if Sentry.configuration.data_collection.queues
context[:arguments] = sentry_serialize_arguments(job.arguments)
end

context
end
private

def sentry_serialize_arguments(argument)
Sentry::Rails::Serializer.serialize(argument)
def with_isolated_scope(&block)
original_hub = Sentry.get_current_hub_internal
Sentry.clone_hub_to_current_thread
Sentry.with_scope(&block)
ensure
Sentry.set_current_hub_internal(original_hub)
end

private

def subscribers
@__subscribers__ ||= Set.new
end
Expand Down
Loading
Loading