diff --git a/lib/instana/instrumentation/active_job.rb b/lib/instana/instrumentation/active_job.rb index 5a27c0f6..b908a698 100644 --- a/lib/instana/instrumentation/active_job.rb +++ b/lib/instana/instrumentation/active_job.rb @@ -70,25 +70,10 @@ def self.extract_context(job) end def self.parse_w3c_traceparent(traceparent, tracestate) - m = traceparent.match(/[0-9a-f]{2}-(?[0-9a-f]{32})-(?[0-9a-f]{16})-(?[0-9a-f]{2})/) - return nil unless m + ctx = ::Instana::Util.parse_w3c_trace_context(traceparent, tracestate) + return nil unless ctx - w3c_sampled = (m['flags'].to_i(16) & 0x01) == 1 - ts_match = tracestate&.match(/in=(?[0-9a-f]+);(?[0-9a-f]+)/) - - if ts_match - ::Instana::SpanContext.new( - trace_id: ::Instana::Util.header_to_id(ts_match['t']), - span_id: ::Instana::Util.header_to_id(ts_match['s']), - level: w3c_sampled ? 1 : 0 - ) - else - ::Instana::SpanContext.new( - trace_id: ::Instana::Util.header_to_id(m['trace'][16..]), - span_id: ::Instana::Util.header_to_id(m['parent']), - level: w3c_sampled ? 1 : 0 - ) - end + ::Instana::SpanContext.new(trace_id: ctx[:trace_id], span_id: ctx[:span_id], level: ctx[:level]) end end end diff --git a/lib/instana/instrumentation/aws_sdk_lambda.rb b/lib/instana/instrumentation/aws_sdk_lambda.rb index e97cde87..f696c37b 100644 --- a/lib/instana/instrumentation/aws_sdk_lambda.rb +++ b/lib/instana/instrumentation/aws_sdk_lambda.rb @@ -10,18 +10,23 @@ def call(context) return @handler.call(context) unless [:invoke_async, :invoke].include?(context.operation_name) if ::Instana.tracer.tracing? && context.operation_name == :invoke - span_context = ::Instana.tracer.context - trace_payload = { - 'X-INSTANA-T' => span_context.trace_id, - 'X-INSTANA-S' => span_context.span_id, - 'X-INSTANA-L' => span_context.level.to_s - } - merged = if context.params[:client_context] - JSON.parse(Base64.strict_decode64(context.params[:client_context])).merge(trace_payload) - else - trace_payload - end - context.params[:client_context] = Base64.strict_encode64(JSON.dump(merged)) + span_context = ::Instana.tracer.context + if span_context + trace_payload = { + 'X-INSTANA-T' => span_context.trace_id, + 'X-INSTANA-S' => span_context.span_id, + 'X-INSTANA-L' => span_context.level.to_s, + 'traceparent' => span_context.trace_parent_header, + 'tracestate' => span_context.trace_state_header + }.reject { |_, v| v.nil? || v.empty? } + + merged = if context.params[:client_context] + JSON.parse(Base64.strict_decode64(context.params[:client_context])).merge(trace_payload) + else + trace_payload + end + context.params[:client_context] = Base64.strict_encode64(JSON.dump(merged)) + end end tags = { diff --git a/lib/instana/instrumentation/aws_sdk_sns.rb b/lib/instana/instrumentation/aws_sdk_sns.rb index 0736c54f..87093dfd 100644 --- a/lib/instana/instrumentation/aws_sdk_sns.rb +++ b/lib/instana/instrumentation/aws_sdk_sns.rb @@ -14,15 +14,18 @@ def call(context) }.reject { |_, v| v.nil? } if context.operation_name == :publish - ::Instana.tracer.in_span(:sns, attributes: {sns: sns_tags}) do |_span| + ::Instana.tracer.in_span(:sns, attributes: { sns: sns_tags }) do |_span| if ::Instana.tracer.tracing? - span_context = ::Instana.tracer.context - context.params[:message_attributes] ||= {} - context.params[:message_attributes].merge!( - 'X_INSTANA_T' => { data_type: 'String', string_value: span_context.trace_id_header }, - 'X_INSTANA_S' => { data_type: 'String', string_value: span_context.span_id_header }, - 'X_INSTANA_L' => { data_type: 'String', string_value: span_context.level.to_s } - ) + inject_trace_attributes(context.params) + end + @handler.call(context) + end + elsif context.operation_name == :publish_batch + ::Instana.tracer.in_span(:sns, attributes: { sns: sns_tags }) do |_span| + if ::Instana.tracer.tracing? + context.params[:publish_batch_request_entries]&.each do |entry| + inject_trace_attributes(entry) + end end @handler.call(context) end @@ -30,6 +33,22 @@ def call(context) @handler.call(context) end end + + private + + def inject_trace_attributes(params) + span_context = ::Instana.tracer.context + return unless span_context + + params[:message_attributes] ||= {} + available_slots = 10 - params[:message_attributes].size + + extra = ::Instana::Util.aws_message_attributes_w3c(span_context) + extra.merge!(::Instana::Util.aws_message_attributes_instana(span_context)) if available_slots >= 5 + + extra.reject! { |_, v| v[:string_value].nil? || v[:string_value].empty? } + params[:message_attributes].merge!(extra) + end end def add_handlers(handlers, _config) diff --git a/lib/instana/instrumentation/aws_sdk_sqs.rb b/lib/instana/instrumentation/aws_sdk_sqs.rb index b1f9790b..02af7090 100644 --- a/lib/instana/instrumentation/aws_sdk_sqs.rb +++ b/lib/instana/instrumentation/aws_sdk_sqs.rb @@ -36,11 +36,16 @@ def call(context) def inject_instana_headers(span, params) params[:message_attributes] ||= {} - params[:message_attributes].merge!({ - "X_INSTANA_T" => {data_type: 'String', string_value: span.context.trace_id}, - "X_INSTANA_S" => {data_type: 'String', string_value: span.context.span_id}, - "X_INSTANA_L" => {data_type: 'String', string_value: span.context.level.to_s} - }) + span_context = span.context + return unless span_context + + available_slots = 10 - params[:message_attributes].size + + extra = ::Instana::Util.aws_message_attributes_w3c(span_context) + extra.merge!(::Instana::Util.aws_message_attributes_instana(span_context)) if available_slots >= 5 + + extra.reject! { |_, v| v[:string_value].nil? || v[:string_value].empty? } + params[:message_attributes].merge!(extra) end def tags_for(operation_name, params) diff --git a/lib/instana/instrumentation/bunny.rb b/lib/instana/instrumentation/bunny.rb index 83ebb42c..6503978b 100644 --- a/lib/instana/instrumentation/bunny.rb +++ b/lib/instana/instrumentation/bunny.rb @@ -134,21 +134,7 @@ def extract_instana_context(headers) end def extract_w3c_context(headers) - m = headers['traceparent'].match( - /[0-9a-f][0-9a-e]-(?[0-9a-f]{32})-(?[0-9a-f]{16})-(?[0-9a-f]{2})/ - ) - return {} unless m - - level = (m['flags'].to_i(16) & 0x01) == 1 ? 1 : 0 - ts_match = headers['tracestate']&.match(/in=(?[0-9a-f]+);(?[0-9a-f]+)/) - - if ts_match - { trace_id: ts_match['t'], span_id: ts_match['s'], level: level } - else - { trace_id: ::Instana::Util.header_to_id(m['trace'][16..]), - span_id: ::Instana::Util.header_to_id(m['parent']), - level: level } - end + ::Instana::Util.parse_w3c_trace_context(headers['traceparent'], headers['tracestate']) || {} end def log_error(error) diff --git a/lib/instana/instrumentation/grpc.rb b/lib/instana/instrumentation/grpc.rb index c7487c26..ce73834d 100644 --- a/lib/instana/instrumentation/grpc.rb +++ b/lib/instana/instrumentation/grpc.rb @@ -107,28 +107,13 @@ def extract_instana_grpc_context(metadata) end def extract_w3c_grpc_context(metadata) - m = metadata['traceparent'].match( - /[0-9a-f][0-9a-e]-(?[0-9a-f]{32})-(?[0-9a-f]{16})-(?[0-9a-f]{2})/ - ) - return nil unless m - - level = (m['flags'].to_i(16) & 0x01) == 1 ? 1 : 0 - ts_match = metadata['tracestate']&.match(/in=(?[0-9a-f]+);(?[0-9a-f]+)/) - baggage = { - external_trace_id: m['trace'], - external_trace_flags: m['flags'], - external_state: metadata['tracestate'] - } - - if ts_match - SpanContext.new(trace_id: ::Instana::Util.header_to_id(ts_match['t']), - span_id: ::Instana::Util.header_to_id(ts_match['s']), - level: level, baggage: baggage) - else - SpanContext.new(trace_id: ::Instana::Util.header_to_id(m['trace'][16..]), - span_id: ::Instana::Util.header_to_id(m['parent']), - level: level, baggage: baggage) - end + ctx = ::Instana::Util.parse_w3c_trace_context(metadata['traceparent'], metadata['tracestate']) + return nil unless ctx + + SpanContext.new(trace_id: ctx[:trace_id], + span_id: ctx[:span_id], + level: ctx[:level], + baggage: ctx.slice(:external_trace_id, :external_trace_flags, :external_state)) end end end diff --git a/lib/instana/instrumentation/shoryuken.rb b/lib/instana/instrumentation/shoryuken.rb index 99598bf4..acdb5283 100644 --- a/lib/instana/instrumentation/shoryuken.rb +++ b/lib/instana/instrumentation/shoryuken.rb @@ -15,24 +15,45 @@ def call(_worker_instance, _queue, sqs_message, _body, &block) } context = incomming_context_from(sqs_message.message_attributes) - instana_context = Instana::SpanContext.new(trace_id: context[:trace_id], span_id: context[:span_id], level: context[:level]) - Trace.with_span(OpenTelemetry::Trace.non_recording_span(instana_context)) do - ::Instana.tracer.in_span(:sqs, attributes: {sqs: sqs_tags}, &block) + if context[:trace_id] + instana_context = Instana::SpanContext.new( + trace_id: context[:trace_id], + span_id: context[:span_id], + level: context[:level] || 1 + ) + Trace.with_span(OpenTelemetry::Trace.non_recording_span(instana_context)) do + ::Instana.tracer.in_span(:sqs, attributes: { sqs: sqs_tags }, &block) + end + else + ::Instana.tracer.in_span(:sqs, attributes: { sqs: sqs_tags }, &block) end end private def incomming_context_from(attributes) - trace_id = read_message_header(attributes, 'X_INSTANA_T') - span_id = read_message_header(attributes, 'X_INSTANA_S') - level = read_message_header(attributes, 'X_INSTANA_L') - - { - trace_id: trace_id, - span_id: span_id, - level: level - }.reject { |_, v| v.nil? } + return {} unless attributes + + if read_message_header(attributes, 'X_INSTANA_T') + trace_id = read_message_header(attributes, 'X_INSTANA_T') + span_id = read_message_header(attributes, 'X_INSTANA_S') + level = read_message_header(attributes, 'X_INSTANA_L') + + { + trace_id: trace_id, + span_id: span_id, + level: level ? level.to_i : 1 + }.reject { |_, v| v.nil? } + else + w3c_context_from(attributes) + end + end + + def w3c_context_from(attributes) + ::Instana::Util.parse_w3c_trace_context( + read_message_header(attributes, 'traceparent'), + read_message_header(attributes, 'tracestate') + ) || {} end def read_message_header(attributes, key) diff --git a/lib/instana/util.rb b/lib/instana/util.rb index 7a1bcdfb..114f757e 100644 --- a/lib/instana/util.rb +++ b/lib/instana/util.rb @@ -2,7 +2,7 @@ # (c) Copyright Instana Inc. 2016 module Instana - module Util + module Util # rubocop:disable Metrics/ModuleLength class << self ID_RANGE = -2**63..2**63-1 # Retrieves and returns the source code for any ruby @@ -171,6 +171,44 @@ def header_to_id(given) given end + # Parse W3C +traceparent+ / +tracestate+ values using the OpenTelemetry + # trace-context propagator. + # + # When the tracestate carries an Instana entry (+in=;+) it + # takes precedence over the ids in the traceparent. + # + # @param traceparent [String] the traceparent header value + # @param tracestate [String, nil] the tracestate header value + # + # @return [Hash, nil] +nil+ when the traceparent is missing or invalid, + # otherwise a hash with: + # - +:trace_id+, +:span_id+, +:level+ for continuing the trace + # - +:external_trace_id+, +:external_trace_flags+, +:external_state+ + # for re-emitting the W3C context downstream + # + def parse_w3c_trace_context(traceparent, tracestate = nil) + return nil unless traceparent.is_a?(String) + + carrier = { 'traceparent' => traceparent, 'tracestate' => tracestate.is_a?(String) ? tracestate : nil } + extracted = OpenTelemetry::Trace::Propagation::TraceContext.text_map_propagator.extract( + carrier, context: OpenTelemetry::Context.empty + ) + span_context = OpenTelemetry::Trace.current_span(extracted).context + return nil unless span_context.valid? + + sampled = span_context.trace_flags.sampled? + instana_state = span_context.tracestate.value('in')&.match(/\A(?[0-9a-f]+);(?[0-9a-f]+)\z/) + + { + trace_id: header_to_id(instana_state ? instana_state[:trace] : span_context.hex_trace_id[16..]), + span_id: header_to_id(instana_state ? instana_state[:span] : span_context.hex_span_id), + level: sampled ? 1 : 0, + external_trace_id: span_context.hex_trace_id, + external_trace_flags: sampled ? '01' : '00', + external_state: tracestate + } + end + def timeout_timestamp Process.clock_gettime(Process::CLOCK_MONOTONIC) end @@ -229,6 +267,35 @@ def http_reason_phrase(status_code) klass.name.sub('Net::HTTP', '').gsub(/([A-Z])/, ' \1').strip end + + # Build the W3C trace-context message attributes for an AWS message + # (SQS or SNS). Returns a hash suitable for merging into + # +params[:message_attributes]+. + # + # @param span_context [Instana::SpanContext] + # @return [Hash] + # + def aws_message_attributes_w3c(span_context) + { + 'traceparent' => { data_type: 'String', string_value: span_context.trace_parent_header }, + 'tracestate' => { data_type: 'String', string_value: span_context.trace_state_header } + } + end + + # Build the Instana proprietary message attributes for an AWS message + # (SQS or SNS). Returns a hash suitable for merging into + # +params[:message_attributes]+. + # + # @param span_context [Instana::SpanContext] + # @return [Hash] + # + def aws_message_attributes_instana(span_context) + { + 'X_INSTANA_T' => { data_type: 'String', string_value: span_context.trace_id }, + 'X_INSTANA_S' => { data_type: 'String', string_value: span_context.span_id }, + 'X_INSTANA_L' => { data_type: 'String', string_value: span_context.level.to_s } + } + end end end end diff --git a/test/instrumentation/aws_test.rb b/test/instrumentation/aws_test.rb index 96d654dc..9052c723 100644 --- a/test/instrumentation/aws_test.rb +++ b/test/instrumentation/aws_test.rb @@ -478,6 +478,23 @@ def test_lambda_with_five_hundred_status end class AwsAgentReadyTest < Minitest::Test + SQS_SEND_MESSAGE_XML_RESPONSE = '' \ + 'd41d8cd98f00b204e9800998ecf8427e' \ + 'msg-123' \ + '' \ + 'req-123' \ + ''.freeze + + SQS_SEND_MESSAGE_BATCH_XML_RESPONSE = '' \ + '' \ + '1' \ + 'msg-1' \ + 'd41d8cd98f00b204e9800998ecf8427e' \ + '' \ + '' \ + 'req-123' \ + ''.freeze + def setup clear_all! end @@ -514,8 +531,96 @@ def test_sns_publish_injects_trace_attributes assert injected.key?('X_INSTANA_T'), 'Expected X_INSTANA_T in message_attributes' assert injected.key?('X_INSTANA_S'), 'Expected X_INSTANA_S in message_attributes' assert injected.key?('X_INSTANA_L'), 'Expected X_INSTANA_L in message_attributes' + assert injected.key?('traceparent'), 'Expected traceparent in message_attributes' assert_equal 'String', injected['X_INSTANA_T'][:data_type] assert_equal 'String', injected['X_INSTANA_L'][:data_type] + assert_equal 'String', injected['traceparent'][:data_type] + assert_match(/00-[0-9a-f]{32}-[0-9a-f]{16}-[0-9a-f]{2}/, injected['traceparent'][:string_value]) + end + + def test_sqs_send_message_injects_trace_attributes + xml_proto = defined?(Aws::SQS::Client) && Aws::SQS::Client.api.metadata['protocol'] == 'query' + body = xml_proto ? SQS_SEND_MESSAGE_XML_RESPONSE : '{}' + headers = xml_proto ? { 'Content-Type' => 'text/xml' } : { 'Content-Type' => 'application/x-amz-json-1.0' } + + stub_request(:post, %r{http://localhost:9324/.*}) + .to_return( + status: 200, + body: body, + headers: headers + ) + + injected = nil + sqs = Aws::SQS::Client.new( + region: "local", + access_key_id: "test", + secret_access_key: "test", + endpoint: "http://localhost:9324" + ) + sqs.handlers.remove(Aws::SQS::Plugins::Md5s::Handler) if defined?(Aws::SQS::Plugins::Md5s::Handler) + sqs.handlers.add( + Class.new(Seahorse::Client::Handler) do + define_method(:call) do |ctx| + injected = ctx.params[:message_attributes]&.dup + super(ctx) + end + end, + step: :initialize, priority: 0 + ) + + Instana.tracer.in_span(:sqs_attrs_test, attributes: {}) do + sqs.send_message(queue_url: 'http://localhost:9324/queue/test', message_body: 'Hello SQS') + end + + refute_nil injected, 'Expected message_attributes to be set' + assert injected.key?('X_INSTANA_T'), 'Expected X_INSTANA_T in message_attributes' + assert injected.key?('X_INSTANA_S'), 'Expected X_INSTANA_S in message_attributes' + assert injected.key?('X_INSTANA_L'), 'Expected X_INSTANA_L in message_attributes' + assert injected.key?('traceparent'), 'Expected traceparent in message_attributes' + assert_equal 'String', injected['X_INSTANA_T'][:data_type] + assert_equal 'String', injected['traceparent'][:data_type] + assert_match(/00-[0-9a-f]{32}-[0-9a-f]{16}-[0-9a-f]{2}/, injected['traceparent'][:string_value]) + end + + def test_sqs_send_message_batch_injects_trace_attributes + xml_proto = defined?(Aws::SQS::Client) && Aws::SQS::Client.api.metadata['protocol'] == 'query' + body = xml_proto ? SQS_SEND_MESSAGE_BATCH_XML_RESPONSE : '{}' + headers = xml_proto ? { 'Content-Type' => 'text/xml' } : { 'Content-Type' => 'application/x-amz-json-1.0' } + + stub_request(:post, %r{http://localhost:9324/.*}) + .to_return(status: 200, body: body, headers: headers) + + injected_entries = [] + sqs = Aws::SQS::Client.new( + region: "local", + access_key_id: "test", + secret_access_key: "test", + endpoint: "http://localhost:9324" + ) + sqs.handlers.remove(Aws::SQS::Plugins::Md5s::Handler) if defined?(Aws::SQS::Plugins::Md5s::Handler) + sqs.handlers.add( + Class.new(Seahorse::Client::Handler) do + define_method(:call) do |ctx| + injected_entries = ctx.params[:entries]&.map { |e| e[:message_attributes]&.dup } + super(ctx) + end + end, + step: :initialize, priority: 0 + ) + + Instana.tracer.in_span(:sqs_batch_attrs_test, attributes: {}) do + sqs.send_message_batch( + queue_url: 'http://localhost:9324/queue/test', + entries: [{ id: '1', message_body: 'Hello batch' }] + ) + end + + refute_empty injected_entries + entry_attrs = injected_entries.first + refute_nil entry_attrs + assert entry_attrs.key?('X_INSTANA_T') + assert entry_attrs.key?('traceparent') + assert_equal 'String', entry_attrs['traceparent'][:data_type] end def test_sns_non_publish_not_modified @@ -577,6 +682,8 @@ def test_lambda_invoke_injects_client_context assert decoded.key?('X-INSTANA-T'), 'Expected X-INSTANA-T in ClientContext' assert decoded.key?('X-INSTANA-S'), 'Expected X-INSTANA-S in ClientContext' assert decoded.key?('X-INSTANA-L'), 'Expected X-INSTANA-L in ClientContext' + assert decoded.key?('traceparent'), 'Expected traceparent in ClientContext' + assert_match(/00-[0-9a-f]{32}-[0-9a-f]{16}-[0-9a-f]{2}/, decoded['traceparent']) end def test_lambda_invoke_merges_existing_client_context @@ -610,9 +717,17 @@ def test_lambda_invoke_merges_existing_client_context assert decoded.key?('X-INSTANA-T'), 'Expected X-INSTANA-T merged in' assert decoded.key?('X-INSTANA-S'), 'Expected X-INSTANA-S merged in' assert decoded.key?('X-INSTANA-L'), 'Expected X-INSTANA-L merged in' + assert decoded.key?('traceparent'), 'Expected traceparent merged in' end def test_no_error_is_raised_and_no_spans_are_created_when_agent_is_not_ready + stub_request(:post, "http://localhost:8000/") + .to_return( + status: 400, + body: '{"__type":"com.amazonaws.dynamodb.v20120810#ResourceNotFoundException","message":"Cannot find table"}', + headers: { 'Content-Type' => 'application/x-amz-json-1.0' } + ) + error = nil ::Instana.agent.stub(:ready?, false) do @@ -638,3 +753,225 @@ def test_no_error_is_raised_and_no_spans_are_created_when_agent_is_not_ready assert_empty ::Instana.processor.queued_spans end end + +class AwsContextPropagationTest < Minitest::Test + SNS_PUBLISH_RESPONSE = '' \ + 'abc' \ + ''.freeze + + SNS_PUBLISH_BATCH_TWO_RESPONSE = '' \ + '' \ + '1m1' \ + '2m2' \ + '' \ + ''.freeze + + SNS_PUBLISH_BATCH_ONE_RESPONSE = '' \ + '' \ + '1m1' \ + '' \ + ''.freeze + + SNS_XML_HEADERS = { 'Content-Type' => 'text/xml' }.freeze + SQS_XML_RESPONSE = '' \ + 'd41d8cd98f00b204e9800998ecf8427e' \ + 'msg-123' \ + '' \ + 'req-123' \ + ''.freeze + + def setup + clear_all! + end + + # --------------------------------------------------------------------------- + # SNS helpers + # --------------------------------------------------------------------------- + + def build_sns_client + Aws::SNS::Client.new( + region: "local", + access_key_id: "test", + secret_access_key: "test", + endpoint: "http://localhost:9911" + ) + end + + def add_sns_message_attribute_interceptor(client, &block) + client.handlers.add( + Class.new(Seahorse::Client::Handler) do + define_method(:call) do |ctx| + block.call(ctx) + super(ctx) + end + end, + step: :initialize, priority: 0 + ) + end + + # --------------------------------------------------------------------------- + # SQS helpers + # --------------------------------------------------------------------------- + + def build_sqs_client + sqs = Aws::SQS::Client.new( + region: "local", + access_key_id: "test", + secret_access_key: "test", + endpoint: "http://localhost:9324" + ) + sqs.handlers.remove(Aws::SQS::Plugins::Md5s::Handler) if defined?(Aws::SQS::Plugins::Md5s::Handler) + sqs + end + + def add_sqs_message_attribute_interceptor(client, &block) + client.handlers.add( + Class.new(Seahorse::Client::Handler) do + define_method(:call) do |ctx| + block.call(ctx) + super(ctx) + end + end, + step: :initialize, priority: 0 + ) + end + + def full_attrs + (1..6).each_with_object({}) do |i, h| + h["app_attr_#{i}"] = { data_type: 'String', string_value: "val#{i}" } + end + end + + # --------------------------------------------------------------------------- + # SNS publish — all 5 headers injected when slots are available + # --------------------------------------------------------------------------- + + def test_sns_publish_injects_only_w3c_headers_when_attributes_almost_full + stub_request(:post, "http://localhost:9911/") + .to_return(status: 200, body: SNS_PUBLISH_RESPONSE, headers: SNS_XML_HEADERS) + + injected = nil + sns = build_sns_client + add_sns_message_attribute_interceptor(sns) { |ctx| injected = ctx.params[:message_attributes]&.dup } + + Instana.tracer.in_span(:sns_fallback_test, attributes: {}) do + sns.publish(topic_arn: 'topic:arn', message: 'Hello', message_attributes: full_attrs) + end + + refute_nil injected + assert injected.key?('traceparent'), 'Expected traceparent when slots < 5' + assert injected.key?('tracestate'), 'Expected tracestate when slots < 5' + refute injected.key?('X_INSTANA_T'), 'Expected X_INSTANA_T to be omitted when slots < 5' + refute injected.key?('X_INSTANA_S'), 'Expected X_INSTANA_S to be omitted when slots < 5' + refute injected.key?('X_INSTANA_L'), 'Expected X_INSTANA_L to be omitted when slots < 5' + assert injected.size <= 10, 'Total message attributes must not exceed SNS limit of 10' + end + + # --------------------------------------------------------------------------- + # SNS publish_batch — all 5 headers injected into every entry (empty attrs) + # --------------------------------------------------------------------------- + + def test_sns_publish_batch_injects_trace_attributes_into_each_entry + stub_request(:post, "http://localhost:9911/") + .to_return(status: 200, body: SNS_PUBLISH_BATCH_TWO_RESPONSE, headers: SNS_XML_HEADERS) + + injected_entries = [] + sns = build_sns_client + add_sns_message_attribute_interceptor(sns) do |ctx| + injected_entries = ctx.params[:publish_batch_request_entries]&.map do |e| + e[:message_attributes]&.dup + end + end + + Instana.tracer.in_span(:sns_batch_test, attributes: {}) do + sns.publish_batch( + topic_arn: 'topic:arn', + publish_batch_request_entries: [ + { id: '1', message: 'First' }, + { id: '2', message: 'Second' } + ] + ) + end + + assert_equal 2, injected_entries.size, 'Expected attributes captured for both entries' + injected_entries.each_with_index do |attrs, i| + refute_nil attrs, "Expected message_attributes on entry #{i + 1}" + assert attrs.key?('X_INSTANA_T'), "Expected X_INSTANA_T on entry #{i + 1}" + assert attrs.key?('X_INSTANA_S'), "Expected X_INSTANA_S on entry #{i + 1}" + assert attrs.key?('X_INSTANA_L'), "Expected X_INSTANA_L on entry #{i + 1}" + assert attrs.key?('traceparent'), "Expected traceparent on entry #{i + 1}" + assert_match( + /00-[0-9a-f]{32}-[0-9a-f]{16}-[0-9a-f]{2}/, + attrs['traceparent'][:string_value] + ) + end + end + + # --------------------------------------------------------------------------- + # SNS publish_batch — falls back to W3C-only when entry slots < 5 + # --------------------------------------------------------------------------- + + def test_sns_publish_batch_injects_only_w3c_headers_when_entry_attributes_almost_full + stub_request(:post, "http://localhost:9911/") + .to_return(status: 200, body: SNS_PUBLISH_BATCH_ONE_RESPONSE, headers: SNS_XML_HEADERS) + + injected_entries = [] + sns = build_sns_client + add_sns_message_attribute_interceptor(sns) do |ctx| + injected_entries = ctx.params[:publish_batch_request_entries]&.map do |e| + e[:message_attributes]&.dup + end + end + + Instana.tracer.in_span(:sns_batch_fallback_test, attributes: {}) do + sns.publish_batch( + topic_arn: 'topic:arn', + publish_batch_request_entries: [ + { id: '1', message: 'Hello', message_attributes: full_attrs } + ] + ) + end + + attrs = injected_entries.first + refute_nil attrs + assert attrs.key?('traceparent'), 'Expected traceparent when slots < 5' + assert attrs.key?('tracestate'), 'Expected tracestate when slots < 5' + refute attrs.key?('X_INSTANA_T'), 'Expected X_INSTANA_T to be omitted when slots < 5' + refute attrs.key?('X_INSTANA_S'), 'Expected X_INSTANA_S to be omitted when slots < 5' + refute attrs.key?('X_INSTANA_L'), 'Expected X_INSTANA_L to be omitted when slots < 5' + assert attrs.size <= 10, 'Total message attributes per batch entry must not exceed SNS limit of 10' + end + + # --------------------------------------------------------------------------- + # SQS send_message — falls back to W3C-only when slots < 5 + # --------------------------------------------------------------------------- + + def test_sqs_send_message_injects_only_w3c_headers_when_attributes_almost_full + xml_proto = defined?(Aws::SQS::Client) && Aws::SQS::Client.api.metadata['protocol'] == 'query' + body = xml_proto ? SQS_XML_RESPONSE : '{}' + headers = xml_proto ? { 'Content-Type' => 'text/xml' } : { 'Content-Type' => 'application/x-amz-json-1.0' } + + stub_request(:post, %r{http://localhost:9324/.*}) + .to_return(status: 200, body: body, headers: headers) + + injected = nil + sqs = build_sqs_client + add_sqs_message_attribute_interceptor(sqs) { |ctx| injected = ctx.params[:message_attributes]&.dup } + + Instana.tracer.in_span(:sqs_fallback_test, attributes: {}) do + sqs.send_message( + queue_url: 'http://localhost:9324/queue/test', + message_body: 'Hello SQS', + message_attributes: full_attrs + ) + end + + refute_nil injected + assert injected.key?('traceparent'), 'Expected traceparent when slots < 5' + assert injected.key?('tracestate'), 'Expected tracestate when slots < 5' + refute injected.key?('X_INSTANA_T'), 'Expected X_INSTANA_T to be omitted when slots < 5' + refute injected.key?('X_INSTANA_S'), 'Expected X_INSTANA_S to be omitted when slots < 5' + refute injected.key?('X_INSTANA_L'), 'Expected X_INSTANA_L to be omitted when slots < 5' + assert injected.size <= 10, 'Total message attributes must not exceed SQS limit of 10' + end +end diff --git a/test/instrumentation/shoryuken_test.rb b/test/instrumentation/shoryuken_test.rb index e29873d0..5d499b45 100644 --- a/test/instrumentation/shoryuken_test.rb +++ b/test/instrumentation/shoryuken_test.rb @@ -102,4 +102,50 @@ def test_message_with_non_string_value_attribute refute_nil span assert_nil span[:p] end + + def test_start_trace_with_w3c_traceparent + message = OpenStruct.new( + queue_url: 'http://example.com', + message_attributes: { + 'traceparent' => OpenStruct.new(string_value: '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01') + } + ) + + @middleware.call(nil, nil, message, nil) {} + + span = ::Instana.processor.queued_spans.first + assert_equal 'a3ce929d0e0e4736', span[:t] + assert_equal '00f067aa0ba902b7', span[:p] + assert_equal 'entry', span[:data][:sqs][:sort] + end + + def test_start_trace_with_w3c_traceparent_and_tracestate + message = OpenStruct.new( + queue_url: 'http://example.com', + message_attributes: { + 'traceparent' => OpenStruct.new(string_value: '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01'), + 'tracestate' => OpenStruct.new(string_value: 'in=1234567890abcdef;fedcba0987654321') + } + ) + + @middleware.call(nil, nil, message, nil) {} + + span = ::Instana.processor.queued_spans.first + assert_equal '1234567890abcdef', span[:t] + assert_equal 'fedcba0987654321', span[:p] + end + + def test_start_trace_with_w3c_traceparent_unsampled + message = OpenStruct.new( + queue_url: 'http://example.com', + message_attributes: { + 'traceparent' => OpenStruct.new(string_value: '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00') + } + ) + + @middleware.call(nil, nil, message, nil) {} + + # Since level is 0 (unsampled), no recording span should be queued + assert_empty ::Instana.processor.queued_spans + end end