Skip to content

refactor(jobs): share queue instrumentation helpers #3132

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

Closed
solnic wants to merge 4 commits into from
Closed

refactor(jobs): share queue instrumentation helpers #3132

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ebe29d4
refactor: share queue consumer instrumentation
solnic Oct 13, 2026
7b515f6
refactor: share queue producer instrumentation
solnic Oct 13, 2026
73c10dd
fix(sidekiq): retain native extension defaults
solnic Oct 13, 2026
dffd7ec
test(sidekiq): run isolated specs in separate processes
solnic Oct 14, 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
230 changes: 53 additions & 177 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 @@ -33,15 +33,15 @@ 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)
return payload if SentryReporter.producer.instrumented_by_extension?(self)

begin
sentry_data = {}
if Sentry.configuration.rails.active_job_propagate_traces
headers = Sentry.get_trace_propagation_headers
if headers && !headers.empty?
sentry_data["trace_propagation_headers"] = headers
payload = SentryReporter.call_hook(self, :inject_trace_headers, payload: payload, headers: headers)
payload = SentryReporter.producer.inject_trace_headers(self, payload, headers)
end
end

Expand Down Expand Up @@ -76,7 +76,6 @@ def already_supported_by_sentry_integration?
end

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

# Emitted as messaging.system when the configured queue adapter's
Expand All @@ -85,20 +84,21 @@ 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
}

class << self
include ErrorReporterContext

def consumer
@consumer ||= Sentry::JobExtensions::Consumer.new(integration: Sentry::Rails, origin: SPAN_ORIGIN)
end

def producer
@producer ||= Sentry::JobExtensions::Producer.new(origin: SPAN_ORIGIN)
end

def job_extensions
@job_extensions ||= Sentry.job_extensions
end
Expand All @@ -115,46 +115,24 @@ 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
enqueued = true
enqueue.call
end

begin
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(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)

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)

run_enqueue.call
end
return yield if producer.instrumented_by_extension?(job)

producer.record_span(
job,
description: job.class.name,
messaging_data: -> {
{
system: messaging_system(job),
message_id: job.job_id,
destination: job.queue_name
}.compact
},
default_retry_count: -> { executions(job) },
on_instrumentation_error: method(:log_producer_span_error),
&enqueue
)
end

def log_producer_span_error(e)
Expand All @@ -169,9 +147,9 @@ def record(job, trace_headers: nil, user: nil, payload: nil, &block)
begin
scope.set_user(user.transform_keys(&:to_sym)) if user && !user.empty?

apply_extension_scope_data(scope, job)
consumer.apply_scope_data(scope, job)

scope.set_transaction_name(call_hook(job, :transaction_name) || job.class.name, source: :task)
scope.set_transaction_name(consumer.transaction_name(job) || job.class.name, source: :task)
scope.set_tags(queue: job.queue_name)

scope.set_contexts(active_job: {
Expand All @@ -181,42 +159,28 @@ def record(job, trace_headers: nil, user: nil, payload: nil, &block)
provider_job_id: job.provider_job_id
})

transaction_options = {
name: scope.transaction_name,
source: scope.transaction_source,
op: OP_NAME,
origin: call_hook(job, :span_origin) || SPAN_ORIGIN
}

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
Sentry.start_transaction(**transaction_options)
end

if transaction
set_messaging_data(transaction, job)
set_extension_messaging_data(transaction, job)
scope.set_span(transaction)
end
transaction = consumer.start_transaction(
scope, job,
trace_headers: trace_headers,
payload: payload,
messaging_data: -> {
{
system: messaging_system(job),
message_id: job.job_id,
destination: job.queue_name,
receive_latency: compute_latency(job)
}
},
default_retry_count: executions(job)
)

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

capture_exception(job, e) if report_exception?(job, e)
capture_exception(job, e) if consumer.report_exception?(job, e)

raise
end
Expand All @@ -225,33 +189,6 @@ def record(job, trace_headers: nil, user: nil, payload: nil, &block)
Sentry.set_current_hub_internal(original_hub)
end

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 (latency = compute_latency(job))
transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_RECEIVE_LATENCY, latency)
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
Expand All @@ -264,27 +201,8 @@ def messaging_system(job)
# 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)
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
def executions(job)
job.executions.to_i if job.respond_to?(:executions)
end

def compute_latency(job)
Expand All @@ -295,49 +213,14 @@ def compute_latency(job)
end

def capture_exception(job, e)
error_data = additive_error_data(
call_hook(job, :error_data, exception: e)
consumer.capture_exception(
job, e,
contexts: Sentry.configuration.data_collection.queues ? execution_context : {},
tags: { job_id: job.job_id, provider_job_id: job.provider_job_id },
extra: sentry_context(job)
)

options = {
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 }
}

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|
Expand All @@ -358,7 +241,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 report_retried_exception?(job, error)
return unless consumer.report_retried_exception?(job, error, default: Sentry.configuration.rails.active_job_report_on_retry_error)

capture_exception(job, error)
end
Expand All @@ -369,13 +252,6 @@ def handle_error_event(*args)
yield(event.payload[:job], event.payload[:error])
end

def finish_sentry_transaction(transaction, status)
return unless transaction

transaction.set_http_status(status)
transaction.finish
end

def sentry_context(job)
context = {
active_job: job.class.name,
Expand Down
20 changes: 20 additions & 0 deletions sentry-rails/spec/active_job/shared_examples/tracing/producer_span.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,26 @@
end
end

context "when tracing is disabled" do
let(:routed_job) do
job_fixture do
attr_accessor :routed_queue

queue_as { routed_queue || "unrouted" }

before_enqueue { |job| job.routed_queue = "routed" }

def perform; end
end
end

it "resolves a queue_as block after the job's own enqueue callbacks" do
routed_job.perform_later

expect(last_enqueued_payload["queue_name"]).to eq("routed")
end
end

context "when producer-span instrumentation raises" do
let(:configure_sentry) { proc { |config| config.traces_sample_rate = 1.0 } }

Expand Down
16 changes: 16 additions & 0 deletions sentry-rails/spec/active_job/shared_examples/tracing/trace_propagation.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,4 +98,20 @@
expect(publish_span[:origin]).to eq("auto.queue.active_job")
end
end

context "when tracing is disabled and the job carries no trace headers" do
let(:configure_sentry) { proc { |config| config.rails.active_job_propagate_traces = false } }

it "keeps the worker's trace on the captured error event" do
worker_trace_id = Sentry.get_main_hub.current_scope.propagation_context.trace_id

expect do
failing_job.perform_later
drain
end.to raise_error(RuntimeError, /boom from failing_job spec/)

error_event = sentry_events.find { |e| e.is_a?(Sentry::ErrorEvent) }
expect(error_event.contexts.dig(:trace, :trace_id)).to eq(worker_trace_id)
end
end
end
3 changes: 3 additions & 0 deletions sentry-ruby/lib/sentry/job_extensions.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,3 +4,6 @@
require "sentry/job_extensions/hooks"
require "sentry/job_extensions/extension"
require "sentry/job_extensions/registry"
require "sentry/job_extensions/instrumentation"
require "sentry/job_extensions/consumer"
require "sentry/job_extensions/producer"
Loading
Loading