diff --git a/sentry-rails/lib/sentry/rails/active_job.rb b/sentry-rails/lib/sentry/rails/active_job.rb index c8108e7e9..bd86df0ca 100644 --- a/sentry-rails/lib/sentry/rails/active_job.rb +++ b/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) diff --git a/sentry-rails/spec/active_job/shared_examples/tracing/messaging_span_data.rb b/sentry-rails/spec/active_job/shared_examples/tracing/messaging_span_data.rb index 8cbab029c..16af73cb3 100644 --- a/sentry-rails/spec/active_job/shared_examples/tracing/messaging_span_data.rb +++ b/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 diff --git a/sentry-rails/spec/active_job/shared_examples/tracing/producer_span.rb b/sentry-rails/spec/active_job/shared_examples/tracing/producer_span.rb index ff692534d..102093ab1 100644 --- a/sentry-rails/spec/active_job/shared_examples/tracing/producer_span.rb +++ b/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 diff --git a/sentry-rails/spec/active_job/shared_examples/tracing/trace_propagation.rb b/sentry-rails/spec/active_job/shared_examples/tracing/trace_propagation.rb index 53b04c560..570ae87ac 100644 --- a/sentry-rails/spec/active_job/shared_examples/tracing/trace_propagation.rb +++ b/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 diff --git a/sentry-rails/spec/active_job/solid_queue_spec.rb b/sentry-rails/spec/active_job/solid_queue_spec.rb index 4dc0de457..6d511b367 100644 --- a/sentry-rails/spec/active_job/solid_queue_spec.rb +++ b/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 diff --git a/sentry-ruby/lib/sentry/span.rb b/sentry-ruby/lib/sentry/span.rb index 0d35e4fab..b53554dc4 100644 --- a/sentry-ruby/lib/sentry/span.rb +++ b/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"