chore(rails): AJ tracing convention compliance

sentry-rails/lib/sentry/rails/active_job.rb

@@ -70,6 +70,12 @@ 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
         }
@@ -97,8 +103,13 @@ def record_producer_span(job, &enqueue)
               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
@@ -169,20 +180,39 @@ def record(job, trace_headers: nil, user: nil, &block)
           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)
-            transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_RETRY_COUNT, [job.executions.to_i - 1, 0].max)
+            if (count = retry_count(job))
+              transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_RETRY_COUNT, count)
+            end
 
             if (latency = compute_latency(job))
               transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_RECEIVE_LATENCY, latency)
             end
           end
 
+          def 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).round
+            (Time.now.to_f - enqueued_time.to_f) * 1000.0
           end
 
           def capture_exception(job, e)

sentry-rails/spec/active_job/shared_examples/tracing/messaging_span_data.rb

@@ -5,13 +5,22 @@
 
   let(:configure_sentry) { proc { |config| config.traces_sample_rate = 1.0 } }
 
-  it "records messaging.message.id and messaging.destination.name on the consumer transaction" do
+  it "records the required messaging attribute set on the consumer transaction" do
     successful_job.set(queue: "critical").perform_later
     drain
 
     data = consumer_transaction.contexts.dig(:trace, :data)
     expect(data["messaging.message.id"]).to be_a(String).and(satisfy { |v| !v.empty? })
     expect(data["messaging.destination.name"]).to eq("critical")
+    expect(data["messaging.system"]).to be_a(String).and(satisfy { |v| !v.empty? })
+  end
+
+  it "records messaging.system derived from the configured queue adapter on the consumer transaction" do
+    successful_job.perform_later
+    drain
+
+    data = consumer_transaction.contexts.dig(:trace, :data)
+    expect(data["messaging.system"]).to eq(successful_job.queue_adapter_name)
   end
 
   it "records messaging.message.retry.count = 0 for non-retryable jobs" do
@@ -53,7 +62,24 @@ def perform
 
       consumer_txns = transactions.select { |t| t.contexts.dig(:trace, :op) == "queue.process" }
       retry_counts = consumer_txns.map { |t| t.contexts.dig(:trace, :data, "messaging.message.retry.count") }
-      expect(retry_counts).to eq([0, 0, 1])
+      expect(retry_counts).to eq([0, 1, 2])
+    end
+
+    it "records messaging.message.retry.count on the producer spans emitted for retry enqueues", skip: RAILS_VERSION < 6.0 do
+      retried_job = job_fixture do
+        retry_on StandardError, attempts: 3, wait: 0
+
+        def perform
+          raise StandardError, "trigger retry" if executions < 3
+        end
+      end
+
+      retried_job.perform_later
+      drain
+
+      publish_spans = transactions.flat_map(&:spans).select { |s| s[:op] == "queue.publish" }
+      retry_counts = publish_spans.map { |s| s[:data]["messaging.message.retry.count"] }
+      expect(retry_counts).to eq([1, 2])
     end
   end
 
@@ -65,6 +91,7 @@ def perform
 
     latency = consumer_transaction.contexts.dig(:trace, :data, "messaging.message.receive.latency")
 
-    expect(latency).to eq(5_000)
+    expect(latency).to be_a(Float)
+    expect(latency).to be_within(1.0).of(5_000.0)
   end
 end

sentry-rails/spec/active_job/shared_examples/tracing/producer_span.rb

@@ -16,11 +16,31 @@
       expect(publish_span).not_to be_nil
       expect(publish_span[:description]).to eq(successful_job.name)
       expect(publish_span[:origin]).to eq("auto.queue.active_job")
+      expect(publish_span[:data]["messaging.system"]).to be_a(String).and(satisfy { |v| !v.empty? })
+      expect(publish_span[:data]["messaging.system"]).to eq(successful_job.queue_adapter_name)
       expect(publish_span[:data]["messaging.message.id"]).to be_a(String).and(satisfy { |v| !v.empty? })
       expect(publish_span[:data]["messaging.destination.name"]).to eq("events")
+      expect(publish_span[:data]["messaging.message.retry.count"]).to eq(0)
       expect(publish_span[:timestamp]).not_to be_nil
     end
 
+    it "records the same messaging.system on the producer span and the consumer transaction" do
+      within_parent_transaction do
+        successful_job.perform_later
+      end
+      drain
+
+      parent = transactions.find { |t| t.contexts.dig(:trace, :op) == "test" }
+      publish_span = parent.spans.find { |s| s[:op] == "queue.publish" }
+      consumer = transactions.find { |t| t.contexts.dig(:trace, :op) == "queue.process" }
+
+      producer_system = publish_span[:data]["messaging.system"]
+      consumer_system = consumer.contexts.dig(:trace, :data, "messaging.system")
+
+      expect(producer_system).not_to be_empty
+      expect(consumer_system).to eq(producer_system)
+    end
+
     it "does not raise or capture an orphan span when no parent transaction is active" do
       expect { successful_job.perform_later }.not_to raise_error
 

sentry-rails/spec/active_job/shared_examples/tracing/trace_propagation.rb

@@ -20,6 +20,17 @@
     expect(consumer_transaction.contexts.dig(:trace, :parent_span_id)).to eq(publish_span_id)
   end
 
+  it "injects both sentry-trace and baggage propagation headers into the serialized payload" do
+    within_parent_transaction do
+      successful_job.perform_later
+    end
+
+    headers = last_enqueued_payload["_sentry"]["trace_propagation_headers"]
+    expect(headers.keys).to include("sentry-trace", "baggage")
+    expect(headers["sentry-trace"]).not_to be_empty
+    expect(headers["baggage"]).not_to be_empty
+  end
+
   it "captures a consumer transaction without raising when no parent transaction was active at enqueue" do
     expect { successful_job.perform_later }.not_to raise_error
     expect { drain }.not_to raise_error

sentry-rails/spec/active_job/solid_queue_spec.rb

@@ -70,5 +70,43 @@ def last_enqueued_payload
 
     it_behaves_like "a Sentry-instrumented ActiveJob backend"
     it_behaves_like "an ActiveJob backend that supports distributed tracing"
+
+    context "adapter-generated span parentage" do
+      let(:configure_sentry) { proc { |config| config.traces_sample_rate = 1.0 } }
+
+      it "nests the Solid Queue enqueue DB span directly under the queue.publish span" do
+        within_parent_transaction do
+          successful_job.perform_later
+        end
+
+        parent = transactions.find { |t| t.contexts.dig(:trace, :op) == "test" }
+        publish_span = parent.spans.find { |s| s[:op] == "queue.publish" }
+        expect(publish_span).not_to be_nil
+
+        enqueue_db_span = parent.spans.find do |s|
+          s[:op] == "db.sql.active_record" && s[:parent_span_id] == publish_span[:span_id]
+        end
+        expect(enqueue_db_span).not_to be_nil
+      end
+
+      it "nests a job-body DB span directly under the queue.process consumer transaction" do
+        db_job = job_fixture do
+          def perform
+            SolidQueue::Job.count
+          end
+        end
+
+        db_job.perform_later
+        drain
+
+        expect(consumer_transaction).not_to be_nil
+
+        consumer_span_id = consumer_transaction.contexts.dig(:trace, :span_id)
+        body_db_span = consumer_transaction.spans.find do |s|
+          s[:op] == "db.sql.active_record" && s[:parent_span_id] == consumer_span_id
+        end
+        expect(body_db_span).not_to be_nil
+      end
+    end
   end
 end

sentry-ruby/lib/sentry/span.rb

@@ -45,6 +45,7 @@ module DataConventions
       FUNCTION = "code.function"
       NAMESPACE = "code.namespace"
 
+      MESSAGING_SYSTEM = "messaging.system"
       MESSAGING_MESSAGE_ID = "messaging.message.id"
       MESSAGING_DESTINATION_NAME = "messaging.destination.name"
       MESSAGING_MESSAGE_RECEIVE_LATENCY = "messaging.message.receive.latency"