feat(queues): improve core queue conformance

Co-Authored-By: Codex <noreply@example.com>

Author: solnic

sentry-ruby/lib/sentry-ruby.rb

@@ -12,6 +12,7 @@
 require "sentry/utils/logging_helper"
 require "sentry/utils/callback_helper"
 require "sentry/utils/sample_rand"
+require "sentry/utils/latency"
 require "sentry/configuration"
 require "sentry/structured_logger"
 require "sentry/debug_structured_logger"

sentry-ruby/lib/sentry/job_extensions/consumer.rb

@@ -55,9 +55,11 @@ def start_transaction(scope, job, transaction: nil, trace_headers: nil, payload:
         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: default_retry_count)
+        if transaction.sampled
+          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: default_retry_count)
+        end
 
         scope.set_span(transaction)
         transaction

sentry-ruby/lib/sentry/job_extensions/extension.rb

@@ -181,7 +181,9 @@ def extract_trace_headers(job:, payload:)
       # @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.
+      #   milliseconds between enqueueing and the start of the job) and
+      #   +:body_size+ (a non-negative Integer, the size of the message body
+      #   in bytes). The fallback result is an empty Hash.
       def messaging_data(job:)
         {}
       end

sentry-ruby/lib/sentry/job_extensions/instrumentation.rb

@@ -8,7 +8,8 @@ class Instrumentation
         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
+        receive_latency: Span::DataConventions::MESSAGING_MESSAGE_RECEIVE_LATENCY,
+        body_size: Span::DataConventions::MESSAGING_MESSAGE_BODY_SIZE
       }.freeze
 
       attr_reader :registry

sentry-ruby/lib/sentry/job_extensions/producer.rb

@@ -47,7 +47,7 @@ def with_span(job, description: nil, messaging_data: {}, default_messaging_data:
         span_description = registry.call(extension, :publish_description, job: job) || description
 
         Sentry.with_child_span(op: OP_NAME, description: span_description) do |span|
-          if span
+          if span && span.sampled
             span_origin = registry.call(extension, :span_origin, job: job) || origin
             span.set_origin(span_origin) if span_origin
 

sentry-ruby/lib/sentry/job_extensions/validators.rb

@@ -14,7 +14,8 @@ class 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) }
+        receive_latency: ->(value) { value.is_a?(Numeric) },
+        body_size: ->(value) { value.is_a?(Integer) && value >= 0 }
       }.freeze
 
       def initialize(event_keys: EVENT_DATA_KEYS, scope_keys: SCOPE_DATA_KEYS, messaging_fields: MESSAGING_DATA_VALIDATORS)

sentry-ruby/lib/sentry/span.rb

@@ -50,6 +50,7 @@ module DataConventions
       MESSAGING_DESTINATION_NAME = "messaging.destination.name"
       MESSAGING_MESSAGE_RECEIVE_LATENCY = "messaging.message.receive.latency"
       MESSAGING_MESSAGE_RETRY_COUNT = "messaging.message.retry.count"
+      MESSAGING_MESSAGE_BODY_SIZE = "messaging.message.body.size"
 
       # Time in ms the request spent in the server queue before processing began.
       HTTP_QUEUE_TIME_MS = "http.server.request.time_in_queue"

sentry-ruby/lib/sentry/utils/latency.rb

@@ -0,0 +1,18 @@
+# frozen_string_literal: true
+
+require "time"
+
+module Sentry
+  module Utils
+    module Latency
+      module_function
+
+      def milliseconds_since(time)
+        return unless time
+
+        time = Time.parse(time) if time.is_a?(String)
+        (Time.now.to_f - time.to_f) * 1000.0
+      end
+    end
+  end
+end

sentry-ruby/spec/sentry/job_extensions/consumer_spec.rb

@@ -48,7 +48,7 @@ def capture(consumer, exception = RuntimeError.new("boom"), **options)
 
   def enriching_consumer
     consumer(
-      messaging_data: ->(job:) { { system: "backend", message_id: job.id, destination: job.queue, receive_latency: 42.5 } },
+      messaging_data: ->(job:) { { system: "backend", message_id: job.id, destination: job.queue, receive_latency: 42.5, body_size: 128 } },
       retry_info: ->(job:) { { count: 3, will_retry: true } },
       scope_data: ->(job:) { { contexts: { backend: { "jid" => "backend-jid" } }, tags: { "backend.queue" => "low" }, extra: { size: 12 } } }
     )
@@ -112,6 +112,7 @@ def enriching_consumer
         "messaging.message.id" => "job-1",
         "messaging.destination.name" => "critical",
         "messaging.message.receive.latency" => 42.5,
+        "messaging.message.body.size" => 128,
         "messaging.message.retry.count" => 3
       )
     end

sentry-ruby/spec/sentry/job_extensions/producer_spec.rb

@@ -18,7 +18,7 @@ def producer(origin: nil, **hooks)
   def enriching_producer(origin: nil)
     producer(
       origin: origin,
-      messaging_data: ->(job:) { { system: "backend", message_id: job.id, destination: job.queue, receive_latency: 42.5 } },
+      messaging_data: ->(job:) { { system: "backend", message_id: job.id, destination: job.queue, receive_latency: 42.5, body_size: 128 } },
       retry_info: ->(job:) { { count: 3, will_retry: true } }
     )
   end
@@ -62,6 +62,7 @@ def record_span(producer, **options, &enqueue)
         "messaging.message.id" => "job-1",
         "messaging.destination.name" => "critical",
         "messaging.message.receive.latency" => 42.5,
+        "messaging.message.body.size" => 128,
         "messaging.message.retry.count" => 3
       )
     end