From 77e02b10ddc0f09d6d1d81747fe87d64ec287ff2 Mon Sep 17 00:00:00 2001 From: Peter Solnica Date: Thu, 15 Oct 2026 10:58:07 +0000 Subject: [PATCH] feat(ruby): add job extensions and shared instrumentation --- sentry-ruby/lib/sentry-ruby.rb | 24 + sentry-ruby/lib/sentry/job_extensions.rb | 10 + .../lib/sentry/job_extensions/composite.rb | 23 + .../lib/sentry/job_extensions/consumer.rb | 135 +++++ .../lib/sentry/job_extensions/extension.rb | 245 ++++++++ .../lib/sentry/job_extensions/hooks.rb | 69 +++ .../sentry/job_extensions/instrumentation.rb | 39 ++ .../lib/sentry/job_extensions/producer.rb | 72 +++ .../lib/sentry/job_extensions/registry.rb | 138 +++++ .../lib/sentry/job_extensions/validators.rb | 107 ++++ .../sentry/job_extensions/consumer_spec.rb | 387 +++++++++++++ .../sentry/job_extensions/producer_spec.rb | 173 ++++++ .../spec/sentry/job_extensions_spec.rb | 546 ++++++++++++++++++ 13 files changed, 1968 insertions(+) create mode 100644 sentry-ruby/lib/sentry/job_extensions.rb create mode 100644 sentry-ruby/lib/sentry/job_extensions/composite.rb create mode 100644 sentry-ruby/lib/sentry/job_extensions/consumer.rb create mode 100644 sentry-ruby/lib/sentry/job_extensions/extension.rb create mode 100644 sentry-ruby/lib/sentry/job_extensions/hooks.rb create mode 100644 sentry-ruby/lib/sentry/job_extensions/instrumentation.rb create mode 100644 sentry-ruby/lib/sentry/job_extensions/producer.rb create mode 100644 sentry-ruby/lib/sentry/job_extensions/registry.rb create mode 100644 sentry-ruby/lib/sentry/job_extensions/validators.rb create mode 100644 sentry-ruby/spec/sentry/job_extensions/consumer_spec.rb create mode 100644 sentry-ruby/spec/sentry/job_extensions/producer_spec.rb create mode 100644 sentry-ruby/spec/sentry/job_extensions_spec.rb 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..fb53d4009 --- /dev/null +++ b/sentry-ruby/lib/sentry/job_extensions.rb @@ -0,0 +1,10 @@ +# frozen_string_literal: true + +require "sentry/job_extensions/validators" +require "sentry/job_extensions/hooks" +require "sentry/job_extensions/extension" +require "sentry/job_extensions/composite" +require "sentry/job_extensions/registry" +require "sentry/job_extensions/instrumentation" +require "sentry/job_extensions/consumer" +require "sentry/job_extensions/producer" diff --git a/sentry-ruby/lib/sentry/job_extensions/composite.rb b/sentry-ruby/lib/sentry/job_extensions/composite.rb new file mode 100644 index 000000000..94a26b3a0 --- /dev/null +++ b/sentry-ruby/lib/sentry/job_extensions/composite.rb @@ -0,0 +1,23 @@ +# frozen_string_literal: true + +module Sentry + module JobExtensions + class Composite < Extension + def initialize(extension, default_extension:, registry:) + super(name: extension.name) + @extension = extension + @default_extension = default_extension + @registry = registry + end + + Hooks::CONTRACTS.each_key do |hook| + define_method(hook) do |**arguments| + default = @registry.call(@default_extension, hook, **arguments) + arguments = arguments.merge(payload: default) if hook == :inject_trace_headers + override = @registry.call(@extension, hook, **arguments) + @default_extension.merge_hook(hook, default: default, override: override) + end + end + end + end +end diff --git a/sentry-ruby/lib/sentry/job_extensions/consumer.rb b/sentry-ruby/lib/sentry/job_extensions/consumer.rb new file mode 100644 index 000000000..b78c3be6a --- /dev/null +++ b/sentry-ruby/lib/sentry/job_extensions/consumer.rb @@ -0,0 +1,135 @@ +# frozen_string_literal: true + +module Sentry + module JobExtensions + # @api private + class Consumer < Instrumentation + OP_NAME = "queue.process" + + attr_reader :integration, :origin + + def initialize(integration:, origin:, registry: Sentry.job_extensions) + super(registry: registry) + @integration = integration + @origin = origin + end + + def apply_scope_data(scope, job, default_scope_data: {}) + apply_scope_values(scope, default_scope_data) + data = registry.call(registry.for(job), :scope_data, job: job) + apply_scope_values(scope, data) + end + + def transaction_name(job) + registry.call(registry.for(job), :transaction_name, job: job) + end + + def start_transaction(scope, job, transaction: nil, trace_headers: nil, payload: nil, always_continue_trace: false, messaging_data: {}, default_retry_count: nil) + extension = registry.for(job) + options = { + name: scope.transaction_name, + source: scope.transaction_source, + op: OP_NAME, + origin: registry.call(extension, :span_origin, job: job) || origin + } + + if (trace_headers.nil? || trace_headers.empty?) && payload + extracted_headers = registry.call(extension, :extract_trace_headers, job: job, payload: payload) + + unless extracted_headers.nil? + trace_headers = extracted_headers + always_continue_trace = true + end + end + + if transaction + transaction.set_name(options[:name], source: options[:source]) + transaction.set_origin(options[:origin]) + else + transaction = if always_continue_trace || (trace_headers && !trace_headers.empty?) + continued = Sentry.continue_trace(trace_headers, **options) + Sentry.start_transaction(transaction: continued, **options) + else + Sentry.start_transaction(**options) + end + end + return unless transaction + + set_messaging_data(transaction, resolve(messaging_data).reject { |_key, value| value.nil? }) + set_messaging_data(transaction, registry.call(extension, :messaging_data, job: job)) + set_retry_count(transaction, extension, job, default: resolve(default_retry_count)) + + scope.set_span(transaction) + transaction + end + + def finish_transaction(transaction, status) + return if !transaction || transaction.timestamp + + transaction.set_http_status(status) + transaction.finish + end + + def will_retry?(job) + info = registry.call(registry.for(job), :retry_info, job: job) + info ? info[:will_retry] : false + end + + def report_exception?(job, exception) + decision = registry.call(registry.for(job), :report_exception?, job: job, exception: exception) + decision.nil? ? !will_retry?(job) : decision + end + + def report_retried_exception?(job, exception, default:) + decision = registry.call(registry.for(job), :report_retried_exception?, job: job, exception: exception) + decision.nil? ? default : decision + end + + def capture_exception(job, exception, contexts: {}, tags: {}, extra: {}) + extension = registry.for(job) + error_data = additive_error_data( + registry.call(extension, :error_data, job: job, exception: exception) + ) + + options = { + extra: error_data[:extra].merge(matching_keys(extra, error_data[:extra])), + tags: error_data[:tags].merge(matching_keys(tags, error_data[:tags])), + hint: { background: false } + } + contexts = error_data[:contexts].merge(matching_keys(contexts, error_data[:contexts])) + options[:contexts] = contexts unless contexts.empty? + + capturing_integration = registry.call(extension, :integration, job: job) || integration + capturing_integration.capture_exception(exception, **options) + end + + private + + def apply_scope_values(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 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 + end + end +end 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..a7cdc37a9 --- /dev/null +++ b/sentry-ruby/lib/sentry/job_extensions/extension.rb @@ -0,0 +1,245 @@ +# 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 + + def merge_hook(hook, default:, override:) + override.nil? ? default : override + 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 + + def producer(job:) + nil + end + + def publish_messaging_data(job:) + messaging_data(job: job) + 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..d098a428b --- /dev/null +++ b/sentry-ruby/lib/sentry/job_extensions/hooks.rb @@ -0,0 +1,69 @@ +# 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), + producer: Contract.new(:producer, fallback: NO_OVERRIDE), + publish_messaging_data: Contract.new(:messaging_data, fallback: EMPTY_HASH) + }.merge( + %i[report_exception? report_retried_exception?] + .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/instrumentation.rb b/sentry-ruby/lib/sentry/job_extensions/instrumentation.rb new file mode 100644 index 000000000..736fb1f67 --- /dev/null +++ b/sentry-ruby/lib/sentry/job_extensions/instrumentation.rb @@ -0,0 +1,39 @@ +# frozen_string_literal: true + +module Sentry + module JobExtensions + # @api private + class Instrumentation + MESSAGING_DATA_ATTRIBUTES = { + system: Span::DataConventions::MESSAGING_SYSTEM, + message_id: Span::DataConventions::MESSAGING_MESSAGE_ID, + destination: Span::DataConventions::MESSAGING_DESTINATION_NAME, + receive_latency: Span::DataConventions::MESSAGING_MESSAGE_RECEIVE_LATENCY + }.freeze + + attr_reader :registry + + def initialize(registry: Sentry.job_extensions) + @registry = registry + end + + private + + def set_messaging_data(span, data) + data.each do |key, value| + span.set_data(MESSAGING_DATA_ATTRIBUTES.fetch(key), value) + end + end + + def set_retry_count(span, extension, job, default: nil) + info = registry.call(extension, :retry_info, job: job) + count = info ? info[:count] : default + span.set_data(Span::DataConventions::MESSAGING_MESSAGE_RETRY_COUNT, count) if count + end + + def resolve(value) + value.respond_to?(:call) ? value.call : value + end + end + end +end diff --git a/sentry-ruby/lib/sentry/job_extensions/producer.rb b/sentry-ruby/lib/sentry/job_extensions/producer.rb new file mode 100644 index 000000000..acb4e0be9 --- /dev/null +++ b/sentry-ruby/lib/sentry/job_extensions/producer.rb @@ -0,0 +1,72 @@ +# frozen_string_literal: true + +module Sentry + module JobExtensions + # @api private + class Producer < Instrumentation + OP_NAME = "queue.publish" + + attr_reader :origin + + def initialize(origin: nil, registry: Sentry.job_extensions) + super(registry: registry) + @origin = origin + end + + def record_span(job, on_instrumentation_error: nil, **span_options, &enqueue) + return with_span(job, **span_options, &enqueue) unless on_instrumentation_error + + enqueued = false + run_enqueue = lambda do + enqueued = true + enqueue.call + end + + begin + with_span(job, **span_options, &run_enqueue) + rescue StandardError => e + raise if enqueued + + on_instrumentation_error.call(e) + run_enqueue.call + end + end + + def prepare_payload(job, payload) + payload + end + + def inject_trace_headers(job, payload, headers) + registry.call(registry.for(job), :inject_trace_headers, job: job, payload: payload, headers: headers) + end + + private + + def with_span(job, description: nil, messaging_data: {}, default_messaging_data: {}, extension_messaging_keys: nil, retry_count: true, default_retry_count: nil) + extension = registry.for(job) + span_description = registry.call(extension, :publish_description, job: job) || description + + Sentry.with_child_span(op: OP_NAME, description: span_description) do |span| + if span + span_origin = registry.call(extension, :span_origin, job: job) || origin + span.set_origin(span_origin) if span_origin + + set_messaging_data(span, resolve(default_messaging_data)) + extension_data = registry.call(extension, :publish_messaging_data, job: job) + extension_data = extension_data.slice(*extension_messaging_keys) if extension_messaging_keys + set_messaging_data(span, extension_data) + set_messaging_data(span, resolve(messaging_data)) + set_retry_count(span, extension, job, default: resolve(default_retry_count)) if retry_count + end + + yield + end + end + class Delegated < Producer + def record_span(job, **options) + yield + 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..5706df406 --- /dev/null +++ b/sentry-ruby/lib/sentry/job_extensions/registry.rb @@ -0,0 +1,138 @@ +# 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(registry: nil, default_extension: nil) + @registry = registry + @default_extension = default_extension + @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) + return @registry.register(extension) if @registry + + 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 + + def with_default(extension) + unless extension.is_a?(Extension) + raise ArgumentError, "default job extension must be an instance of #{Extension}, got #{extension.inspect}" + end + + Registry.new(registry: self, default_extension: 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) + extension = @registry ? @registry.for(job) : @extensions.find { |candidate| handles?(candidate, job) } + return extension unless @default_extension + return @default_extension unless extension + + Composite.new(extension, default_extension: @default_extension, registry: self) + 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 + return @registry.clear if @registry + + @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..d4ac51708 --- /dev/null +++ b/sentry-ruby/lib/sentry/job_extensions/validators.rb @@ -0,0 +1,107 @@ +# 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 producer(result, **) + return if result.nil? + return result if result.is_a?(Producer) + + 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/consumer_spec.rb b/sentry-ruby/spec/sentry/job_extensions/consumer_spec.rb new file mode 100644 index 000000000..491e3fee5 --- /dev/null +++ b/sentry-ruby/spec/sentry/job_extensions/consumer_spec.rb @@ -0,0 +1,387 @@ +# frozen_string_literal: true + +require "sentry/integrable" + +module Sentry + module TestJobsIntegration + extend Sentry::Integrable + + register_integration name: "test_jobs", version: "1.0" + end +end + +RSpec.describe Sentry::JobExtensions::Consumer do + let(:job) { Struct.new(:id, :queue).new("job-1", "critical") } + let(:integration) { Sentry::TestJobsIntegration } + let(:string_io) { StringIO.new } + + def consumer(**hooks) + registry = Sentry::JobExtensions::Registry.new + registry.register( + Class.new(Sentry::JobExtensions::Extension) do + define_method(:handles?) { |_job| true } + hooks.each { |hook, body| define_method(hook, &body) } + end.new(name: "test_backend") + ) + described_class.new(integration: integration, origin: "auto.queue.test", registry: registry) + end + + def scope + Sentry.get_current_scope + end + + def start_transaction(consumer, **options) + scope.set_transaction_name("TestJob", source: :task) + consumer.start_transaction(scope, job, **options) + end + + def finished_transaction(consumer, **options) + transaction = start_transaction(consumer, **options) + consumer.finish_transaction(transaction, 200) + sentry_events.last + end + + def capture(consumer, exception = RuntimeError.new("boom"), **options) + consumer.capture_exception(job, exception, **options) + sentry_events.last + end + + def enriching_consumer + consumer( + messaging_data: ->(job:) { { system: "backend", message_id: job.id, destination: job.queue, receive_latency: 42.5 } }, + retry_info: ->(job:) { { count: 3, will_retry: true } }, + scope_data: ->(job:) { { contexts: { backend: { "jid" => "backend-jid" } }, tags: { "backend.queue" => "low" }, extra: { size: 12 } } } + ) + end + + let(:foreign_trace_id) { "a" * 32 } + let(:foreign_trace_headers) { { "sentry-trace" => "#{foreign_trace_id}-#{"b" * 16}-1" } } + + before do + perform_basic_setup do |config| + config.traces_sample_rate = 1.0 + config.sdk_logger = Logger.new(string_io) + end + setup_sentry_test + end + + after do + teardown_sentry_test + end + + describe "#start_transaction" do + it "starts a queue.process transaction on the scope named after the scope's transaction" do + transaction = start_transaction(consumer) + + expect(scope.get_span).to eq(transaction) + expect(transaction.op).to eq("queue.process") + expect(transaction.origin).to eq("auto.queue.test") + expect(transaction.name).to eq("TestJob") + expect(transaction.source).to eq(:task) + end + + it "reuses the supplied transaction while applying the job's identity and messaging data" do + started = enriching_consumer + transaction = Sentry.start_transaction(name: "BeforeHandoff", op: "queue.process", origin: "auto.queue.previous", trace_id: foreign_trace_id) + span_id = transaction.span_id + start_timestamp = transaction.start_timestamp + + reused = start_transaction(started, transaction: transaction) + started.finish_transaction(reused, 200) + + expect(reused).to equal(transaction) + expect(scope.get_span).to equal(transaction) + expect(reused.start_timestamp).to eq(start_timestamp) + expect(sentry_events.size).to eq(1) + event = sentry_events.last + expect(event.transaction).to eq("TestJob") + expect(event.transaction_info).to include(source: :task) + expect(event.contexts[:trace]).to include(trace_id: foreign_trace_id, span_id: span_id, origin: "auto.queue.test") + expect(event.contexts.dig(:trace, :data)).to include( + "messaging.system" => "backend", + "messaging.message.id" => "job-1", + "messaging.message.retry.count" => 3 + ) + end + + it "records the extension's messaging data and retry count as span data" do + event = finished_transaction(enriching_consumer) + + expect(event.contexts.dig(:trace, :data)).to include( + "messaging.system" => "backend", + "messaging.message.id" => "job-1", + "messaging.destination.name" => "critical", + "messaging.message.receive.latency" => 42.5, + "messaging.message.retry.count" => 3 + ) + end + + it "lets the extension's messaging data win over the caller's and fills the rest from non-nil caller values" do + partial_consumer = consumer(messaging_data: ->(job:) { { system: "backend", message_id: job.id } }) + + event = finished_transaction( + partial_consumer, + messaging_data: { system: "caller", message_id: "caller-id", destination: "caller-queue", receive_latency: nil } + ) + + data = event.contexts.dig(:trace, :data) + expect(data).to include( + "messaging.system" => "backend", + "messaging.message.id" => "job-1", + "messaging.destination.name" => "caller-queue" + ) + expect(data).not_to have_key("messaging.message.receive.latency") + end + + it "uses the caller's default retry count when the extension reports no retry state" do + expect(finished_transaction(consumer, default_retry_count: 7).contexts.dig(:trace, :data)) + .to include("messaging.message.retry.count" => 7) + end + + it "evaluates a callable default retry count when the extension reports no retry state" do + event = finished_transaction(consumer, default_retry_count: -> { 7 }) + + expect(event.contexts.dig(:trace, :data)).to include("messaging.message.retry.count" => 7) + end + + it "prefers the extension's retry count over the caller's default" do + expect(finished_transaction(enriching_consumer, default_retry_count: 7).contexts.dig(:trace, :data)) + .to include("messaging.message.retry.count" => 3) + end + + it "records only the caller's data when no registered extension handles the job" do + unhandled = described_class.new(integration: integration, origin: "auto.queue.test", registry: Sentry::JobExtensions::Registry.new) + + event = finished_transaction(unhandled, default_retry_count: 7) + + expect(event.contexts.dig(:trace, :data)).to include("messaging.message.retry.count" => 7) + expect(event.contexts.dig(:trace, :data)).not_to include("messaging.system") + end + + it "continues the trace from the caller's trace headers" do + transaction = start_transaction(consumer, trace_headers: foreign_trace_headers) + + expect(transaction.trace_id).to eq(foreign_trace_id) + end + + it "continues the trace the extension extracts from the payload when the caller has no trace headers" do + transaction = start_transaction(consumer(extract_trace_headers: ->(job:, payload:) { payload["headers"] }), trace_headers: {}, payload: { "headers" => foreign_trace_headers }) + + expect(transaction.trace_id).to eq(foreign_trace_id) + end + + it "prefers the caller's trace headers over the ones the extension extracts" do + transaction = start_transaction(consumer(extract_trace_headers: ->(job:, payload:) { { "sentry-trace" => "#{"c" * 32}-#{"d" * 16}-1" } }), trace_headers: foreign_trace_headers, payload: {}) + + expect(transaction.trace_id).to eq(foreign_trace_id) + end + + it "keeps the scope's trace when there are no trace headers" do + scope_trace_id = scope.propagation_context.trace_id + + start_transaction(consumer) + + expect(scope.propagation_context.trace_id).to eq(scope_trace_id) + end + + it "starts a new trace on the scope without trace headers when told to always continue the trace" do + scope_trace_id = scope.propagation_context.trace_id + + start_transaction(consumer, always_continue_trace: true) + + expect(scope.propagation_context.trace_id).not_to eq(scope_trace_id) + end + + it "starts a new trace on the scope when the extension extracts no trace headers from the payload" do + scope_trace_id = scope.propagation_context.trace_id + + start_transaction(consumer(extract_trace_headers: ->(job:, payload:) { {} }), payload: {}) + + expect(scope.propagation_context.trace_id).not_to eq(scope_trace_id) + end + + it "returns nil when tracing is disabled" do + Sentry.configuration.traces_sample_rate = nil + + expect(start_transaction(enriching_consumer)).to be_nil + end + + { + "raise" => ->(**) { raise "hook exploded" }, + "return malformed data" => ->(**) { "malformed" } + }.each do |behavior, body| + it "starts the transaction without extension data when the hooks #{behavior}" do + event = finished_transaction(consumer(messaging_data: body, retry_info: body, extract_trace_headers: body), payload: {}) + + expect(event.contexts.dig(:trace, :status)).to eq("ok") + expect(event.contexts.dig(:trace, :data).keys.grep(/\Amessaging\./)).to be_empty + expect(string_io.string).to include("test_backend") + end + end + end + + describe "#finish_transaction" do + it "finishes the transaction with the status for the given HTTP status code" do + started = consumer + transaction = start_transaction(started) + + started.finish_transaction(transaction, 500) + + expect(sentry_events.last.contexts.dig(:trace, :status)).to eq("internal_error") + end + + it "keeps the first completion status and sends the transaction only once" do + started = consumer + transaction = start_transaction(started) + started.finish_transaction(transaction, 200) + timestamp = transaction.timestamp + + started.finish_transaction(transaction, 500) + + expect(transaction.timestamp).to eq(timestamp) + expect(transaction.status).to eq("ok") + expect(sentry_events.size).to eq(1) + expect(sentry_events.last.contexts.dig(:trace, :status)).to eq("ok") + end + + it "ignores a missing transaction" do + expect { consumer.finish_transaction(nil, 200) }.not_to raise_error + end + end + + describe "#apply_scope_data" do + it "adds the extension's contexts, tags and extra to the scope" do + enriching_consumer.apply_scope_data(scope, job) + + expect(scope.contexts[:backend]).to eq("jid" => "backend-jid") + expect(scope.tags).to include("backend.queue" => "low") + expect(scope.extra).to include(size: 12) + end + + it "sets the user the extension supplies on the scope" do + consumer(scope_data: ->(job:) { { user: { "id" => "42", "segment" => "beta" } } }).apply_scope_data(scope, job) + + expect(scope.user).to eq("id" => "42", "segment" => "beta") + end + + it "leaves the scope unchanged when the hook fails" do + consumer(scope_data: ->(job:) { raise "boom" }).apply_scope_data(scope, job) + + expect(scope.tags).to be_empty + expect(scope.extra).to be_empty + end + end + + describe "#will_retry?" do + it "reports whether the extension says the backend will retry the job" do + expect(enriching_consumer.will_retry?(job)).to be(true) + expect(consumer(retry_info: ->(job:) { { count: 3, will_retry: false } }).will_retry?(job)).to be(false) + end + + it "reports no retry when the backend's retry state is unknown" do + expect(consumer.will_retry?(job)).to be(false) + expect(consumer(retry_info: ->(job:) { "malformed" }).will_retry?(job)).to be(false) + end + end + + describe "#report_exception?" do + let(:exception) { RuntimeError.new("boom") } + + it "follows the extension's decision whether the backend will retry the job or not" do + reporting = consumer(report_exception?: ->(job:, exception:) { true }, retry_info: ->(job:) { { count: 0, will_retry: true } }) + silent = consumer(report_exception?: ->(job:, exception:) { false }, retry_info: ->(job:) { { count: 0, will_retry: false } }) + + expect(reporting.report_exception?(job, exception)).to be(true) + expect(silent.report_exception?(job, exception)).to be(false) + end + + it "reports only failures the backend will not retry when the extension does not decide" do + expect(consumer.report_exception?(job, exception)).to be(true) + expect(enriching_consumer.report_exception?(job, exception)).to be(false) + end + + { + "raises" => ->(**) { raise "hook exploded" }, + "returns malformed data" => ->(**) { "yes" } + }.each do |behavior, body| + it "falls back to the backend's retry state when the hook #{behavior}" do + expect(consumer(report_exception?: body, retry_info: ->(job:) { { count: 0, will_retry: true } }).report_exception?(job, exception)).to be(false) + expect(string_io.string).to include("test_backend") + end + end + end + + describe "#capture_exception" do + def error_consumer + consumer( + error_data: lambda do |job:, exception:| + { + contexts: { backend_error: { "retried" => true }, job: { "id" => "backend", "attempt" => 2 } }, + tags: { "backend.error" => "yes", queue: "backend-queue", job_id: "backend" }, + extra: { attempts: 2, size: 99 } + } + end + ) + end + + it "captures the exception through the integration" do + event = capture(consumer) + + expect(event.exception.values.last.type).to eq("RuntimeError") + expect(event.exception.values.last.mechanism.type).to eq("test_jobs") + expect(event.sdk[:name]).to eq("sentry.ruby.test_jobs") + end + + it "adds the extension's error data without replacing values already on the scope" do + scope.set_contexts(job: { "id" => "scope" }) + scope.set_tags(queue: "default") + scope.set_extras(size: 12) + + event = capture(error_consumer) + + expect(event.contexts[:backend_error]).to eq("retried" => true) + expect(event.contexts[:job]).to eq("id" => "scope", "attempt" => 2) + expect(event.tags).to include("backend.error" => "yes", queue: "default") + expect(event.extra).to include(attempts: 2, size: 12) + end + + it "keeps existing extension and user context values while adding error-specific fields" do + refining_consumer = consumer( + scope_data: ->(job:) { { contexts: { job: { "id" => "backend", "args" => ["raw"] } } } }, + error_data: ->(job:, exception:) { { contexts: { job: { "args" => ["filtered"], "id" => "error", "attempt" => 2 } } } } + ) + refining_consumer.apply_scope_data(scope, job) + scope.set_contexts(job: { "id" => "user" }) + + event = capture(refining_consumer) + + expect(event.contexts[:job]).to eq("id" => "user", "args" => ["raw"], "attempt" => 2) + end + + it "lets the caller's contexts, tags and extra win over the extension's error data" do + event = capture(error_consumer, contexts: { job: { "id" => "caller" } }, tags: { job_id: "caller" }, extra: { attempts: 5 }) + + expect(event.contexts[:job]).to eq("id" => "caller") + expect(event.tags).to include(job_id: "caller", "backend.error" => "yes") + expect(event.extra).to include(attempts: 5) + end + + it "lets caller tags and extra override extension values with equivalent string or symbol keys" do + event = capture(error_consumer, tags: { "job_id" => "caller", :"backend.error" => "caller-error" }, extra: { "attempts" => 5 }) + + expect(event.tags).to eq(job_id: "caller", queue: "backend-queue", "backend.error" => "caller-error") + expect(event.extra).to eq(attempts: 5, size: 99) + expect(event.to_json_compatible).to include( + "tags" => { "job_id" => "caller", "queue" => "backend-queue", "backend.error" => "caller-error" }, + "extra" => { "attempts" => 5, "size" => 99 } + ) + end + + it "captures the exception with the caller's data when the hook fails" do + event = capture(consumer(error_data: ->(**) { raise "hook exploded" }), tags: { job_id: "caller" }) + + expect(event.exception.values.last.type).to eq("RuntimeError") + expect(event.tags).to eq(job_id: "caller") + end + end +end diff --git a/sentry-ruby/spec/sentry/job_extensions/producer_spec.rb b/sentry-ruby/spec/sentry/job_extensions/producer_spec.rb new file mode 100644 index 000000000..a858d627e --- /dev/null +++ b/sentry-ruby/spec/sentry/job_extensions/producer_spec.rb @@ -0,0 +1,173 @@ +# frozen_string_literal: true + +RSpec.describe Sentry::JobExtensions::Producer do + let(:job) { Struct.new(:id, :queue).new("job-1", "critical") } + let(:string_io) { StringIO.new } + + def producer(origin: nil, **hooks) + registry = Sentry::JobExtensions::Registry.new + registry.register( + Class.new(Sentry::JobExtensions::Extension) do + define_method(:handles?) { |_job| true } + hooks.each { |hook, body| define_method(hook, &body) } + end.new(name: "test_backend") + ) + described_class.new(origin: origin, registry: registry) + end + + def enriching_producer(origin: nil) + producer( + origin: origin, + messaging_data: ->(job:) { { system: "backend", message_id: job.id, destination: job.queue, receive_latency: 42.5 } }, + retry_info: ->(job:) { { count: 3, will_retry: true } } + ) + end + + def within_transaction + transaction = Sentry.start_transaction(name: "parent", op: "test") + Sentry.get_current_scope.set_span(transaction) + yield + ensure + transaction.finish + end + + def publish_span + sentry_events.last.spans.find { |span| span[:op] == "queue.publish" } + end + + def record_span(producer, **options, &enqueue) + producer.record_span(job, description: "TestJob", **options, &(enqueue || -> { :enqueued })) + end + + before do + perform_basic_setup do |config| + config.traces_sample_rate = 1.0 + config.sdk_logger = Logger.new(string_io) + end + setup_sentry_test + end + + after do + teardown_sentry_test + end + + describe "#record_span" do + it "records a queue.publish span with the extension's messaging data and retry count around the enqueue" do + result = within_transaction { record_span(enriching_producer(origin: "auto.queue.test")) } + + expect(result).to eq(:enqueued) + expect(publish_span).to include(description: "TestJob", origin: "auto.queue.test") + expect(publish_span[:data]).to include( + "messaging.system" => "backend", + "messaging.message.id" => "job-1", + "messaging.destination.name" => "critical", + "messaging.message.receive.latency" => 42.5, + "messaging.message.retry.count" => 3 + ) + end + + it "lets the caller's messaging data win over the extension's" do + within_transaction do + record_span(enriching_producer, messaging_data: { system: "caller", message_id: "caller-id" }) + end + + expect(publish_span[:data]).to include( + "messaging.system" => "caller", + "messaging.message.id" => "caller-id", + "messaging.destination.name" => "critical" + ) + end + + it "applies only the requested keys of the extension's messaging data" do + within_transaction do + record_span(enriching_producer, extension_messaging_keys: %i[message_id destination]) + end + + expect(publish_span[:data]).to include("messaging.message.id" => "job-1", "messaging.destination.name" => "critical") + expect(publish_span[:data]).not_to include("messaging.system", "messaging.message.receive.latency") + end + + it "falls back to the caller's retry count when the extension has no retry info" do + within_transaction { record_span(producer, default_retry_count: 2) } + + expect(publish_span[:data]).to include("messaging.message.retry.count" => 2) + end + + it "records no retry count when the caller opts out" do + within_transaction { record_span(enriching_producer, retry_count: false) } + + expect(publish_span[:data]).not_to include("messaging.message.retry.count") + end + + it "degrades a failing extension hook to its neutral result" do + within_transaction { record_span(producer(messaging_data: ->(job:) { raise "boom" }, retry_info: ->(job:) { :malformed }), default_retry_count: 1) } + + expect(publish_span[:data]).to include("messaging.message.retry.count" => 1) + expect(publish_span[:data]).not_to include("messaging.message.id") + end + + it "enqueues without a span when no transaction is active" do + expect(record_span(enriching_producer)).to eq(:enqueued) + expect(sentry_events).to be_empty + end + + context "when recording the span raises" do + before do + allow(Sentry).to receive(:with_child_span).and_raise(StandardError, "boom from instrumentation") + end + + it "propagates the failure without enqueueing when the caller wants no isolation" do + enqueued = 0 + + expect do + within_transaction { record_span(enriching_producer) { enqueued += 1 } } + end.to raise_error(StandardError, /boom from instrumentation/) + expect(enqueued).to eq(0) + end + + it "reports the failure to the caller and enqueues once when the caller wants isolation" do + errors = [] + enqueued = 0 + + result = within_transaction do + record_span(enriching_producer, on_instrumentation_error: ->(e) { errors << e }) { enqueued += 1 } + end + + expect(result).to eq(1) + expect(enqueued).to eq(1) + expect(errors.map(&:message)).to eq(["boom from instrumentation"]) + end + end + + it "re-raises a failing enqueue without reporting it as an instrumentation failure" do + errors = [] + + expect do + within_transaction do + record_span(enriching_producer, on_instrumentation_error: ->(e) { errors << e }) { raise "boom from enqueue" } + end + end.to raise_error(RuntimeError, /boom from enqueue/) + expect(errors).to be_empty + end + end + + describe "#inject_trace_headers" do + let(:payload) { { "job_class" => "TestJob" } } + let(:headers) { { "sentry-trace" => "trace" } } + + it "returns the payload the extension built with the headers" do + extension_producer = producer(inject_trace_headers: ->(job:, payload:, headers:) { payload.merge("trace" => headers) }) + + expect(extension_producer.inject_trace_headers(job, payload, headers)).to eq( + "job_class" => "TestJob", "trace" => headers + ) + end + + it "returns the payload unchanged when the extension fails" do + failing_producer = producer(inject_trace_headers: ->(job:, payload:, headers:) { raise "boom" }) + + expect(failing_producer.inject_trace_headers(job, payload, headers)).to equal(payload) + expect(string_io.string).to include("failed in inject_trace_headers") + 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..dacced275 --- /dev/null +++ b/sentry-ruby/spec/sentry/job_extensions_spec.rb @@ -0,0 +1,546 @@ +# 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 "#with_default" do + let(:default_extension) do + build_extension("default", claims: ->(_job) { true }).tap do |extension| + extension.define_singleton_method(:transaction_name) { |job:| "Default/#{job.adapter}" } + extension.define_singleton_method(:report_retried_exception?) { |job:, exception:| true } + end + end + let(:registry) { Sentry.job_extensions.with_default(default_extension) } + + it "uses the default until a matching backend is registered" do + expect(registry.call(registry.for(sidekiq_job), :transaction_name, job: sidekiq_job)).to eq("Default/sidekiq") + backend = build_extension("sidekiq") + backend.define_singleton_method(:transaction_name) { |job:| "SidekiqJob" } + + Sentry.register_job_extension(backend) + + expect(registry.call(registry.for(sidekiq_job), :transaction_name, job: sidekiq_job)).to eq("SidekiqJob") + expect(registry.call(registry.for(good_job_job), :transaction_name, job: good_job_job)).to eq("Default/good_job") + end + + it "fills missing hooks from the default and preserves a backend's false decision" do + backend = build_extension("sidekiq") + backend.define_singleton_method(:report_retried_exception?) { |job:, exception:| false } + Sentry.register_job_extension(backend) + + extension = registry.for(sidekiq_job) + + expect(registry.call(extension, :transaction_name, job: sidekiq_job)).to eq("Default/sidekiq") + expect(registry.call(extension, :report_retried_exception?, job: sidekiq_job, exception: RuntimeError.new)).to be(false) + end + + context "with default trace propagation" do + let(:payload) { { "class" => "ExampleJob", "args" => [] } } + let(:headers) { { "sentry-trace" => "#{"a" * 32}-#{"b" * 16}-1" } } + let(:default_extension) do + Class.new(Sentry::JobExtensions::Extension) do + def inject_trace_headers(job:, payload:, headers:) + payload.merge("sentry" => headers) + end + + def extract_trace_headers(job:, payload:) + payload["sentry"] + end + end.new(name: "default") + end + + it "preserves the default trace when the backend only supplies metadata" do + backend = build_extension("sidekiq") + backend.define_singleton_method(:scope_data) { |job:| { tags: { queue: "default" } } } + Sentry.register_job_extension(backend) + extension = registry.for(sidekiq_job) + + injected = registry.call(extension, :inject_trace_headers, job: sidekiq_job, payload: payload, headers: headers) + + expect(injected).to eq(payload.merge("sentry" => headers)) + expect(registry.call(extension, :extract_trace_headers, job: sidekiq_job, payload: injected)).to eq(headers) + expect(payload).to eq("class" => "ExampleJob", "args" => []) + end + + it "lets the backend add its headers to the default payload" do + backend = build_extension("sidekiq") + backend.define_singleton_method(:inject_trace_headers) do |job:, payload:, headers:| + payload.merge("backend_trace" => headers) + end + Sentry.register_job_extension(backend) + extension = registry.for(sidekiq_job) + + injected = registry.call(extension, :inject_trace_headers, job: sidekiq_job, payload: payload, headers: headers) + + expect(injected).to eq(payload.merge("sentry" => headers, "backend_trace" => headers)) + expect(registry.call(extension, :extract_trace_headers, job: sidekiq_job, payload: injected)).to eq(headers) + end + 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 }, + producer: { job: job }, + publish_messaging_data: { 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, + producer: nil, + publish_messaging_data: {} + } + 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 the producer an extension supplies for enqueueing the job" do + producer = Sentry::JobExtensions::Producer::Delegated.new + extension = build_hooked_extension(producer: ->(job:) { producer }) + + expect(call_hook(extension, :producer)).to be(producer) + 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], + producer: ["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