Skip to content

feat(rails): delegate ActiveJob to backend 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 1 commit into from
+975 −190
Open

feat(rails): delegate ActiveJob to backend extensions #3130

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
23 changes: 23 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 @@ -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
246 changes: 65 additions & 181 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,87 +3,81 @@
require "set"
require "sentry/rails/serializer"
require "sentry/rails/error_reporter_context"
require "sentry/rails/active_job/job_extension"
require "sentry/rails/active_job/producer"

module Sentry
module Rails
module ActiveJobExtensions
SENTRY_PAYLOAD_KEY = "_sentry"

USER_FIELDS_ALLOWLIST = %w[id email username].freeze

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 || {}
current_scope = Sentry.get_current_scope
scope = current_scope if data["execution_scope_id"] == current_scope.object_id
transaction = scope&.get_span
SentryReporter.record(
self,
trace_headers: data["trace_propagation_headers"],
user: data["user"]
user: data["user"],
payload: _sentry_job_data,
transaction: transaction,
scope: scope
) { 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.producer(self).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.
MESSAGING_SYSTEM_FALLBACK = "activejob"

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

class << self
include ErrorReporterContext
def default_extension
@default_extension ||= JobExtension.new
end

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

def producer(job)
job_extensions.call(job_extensions.for(job), :producer, job: job)
end

def job_extensions
default_extension.registry
end

def producer_callback_registered?
@producer_callback_registered ||= false
Expand All @@ -95,144 +89,51 @@ def producer_callback_registered!

def record_producer_span(job, &enqueue)
return yield if !Sentry.initialized? || job.already_supported_by_sentry_integration?
producer(job).record_span(job, &enqueue)
end

enqueued = false
def record(job, trace_headers: nil, user: nil, payload: nil, transaction: nil, scope: nil, &block)
options = { trace_headers: trace_headers, user: user, payload: payload, transaction: transaction }
return record_in_scope(scope, job, **options, backend_scope: true, &block) if scope

run_enqueue = lambda do
enqueued = true
enqueue.call
with_isolated_scope do |isolated_scope|
record_in_scope(isolated_scope, job, **options, &block)
end
end

def record_in_scope(scope, job, trace_headers: nil, user: nil, payload: nil, transaction: nil, backend_scope: false)
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
scope.set_user(user.transform_keys(&:to_sym)) if user && !user.empty?

log_producer_span_error(e)
consumer.apply_scope_data(scope, job)

run_enqueue.call
end
end
scope.set_transaction_name(consumer.transaction_name(job), source: :task)

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)
original_hub = Sentry.get_current_hub_internal
Sentry.clone_hub_to_current_thread
transaction = consumer.start_transaction(
scope, job,
transaction: transaction.is_a?(Sentry::Transaction) ? transaction : nil,
trace_headers: trace_headers,
payload: payload
)

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)
yield.tap do
consumer.finish_transaction(transaction, 200)
end
rescue Exception => e # rubocop:disable Lint/RescueException
consumer.finish_transaction(transaction, 500)

if backend_scope
scope.update_from_options(**default_extension.error_data(job: job, exception: e))
elsif consumer.report_exception?(job, e)
capture_exception(job, e)

raise
end
end
ensure
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)
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)
consumer.capture_exception(job, e, **default_extension.error_data(job: job, exception: e))
end

def register_event_handlers
Expand All @@ -255,7 +156,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 consumer.report_retried_exception?(job, error, default: nil)

capture_exception(job, error)
end
Expand All @@ -266,35 +167,18 @@ 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
private :record_in_scope

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