Merge branch 'main' into update-error-handling-api

Author: kaylareopelle

.toys/.data/releases.yml

@@ -229,6 +229,10 @@ gems:
     directory: processor/baggage
     version_constant: [OpenTelemetry, Processor, Baggage, VERSION]
 
+  - name: opentelemetry-propagator-google_cloud_trace_context
+    directory: propagator/google_cloud_trace_context
+    version_constant: [OpenTelemetry, Propagator, GoogleCloudTraceContext, VERSION]
+
   - name: opentelemetry-propagator-ottrace
     directory: propagator/ottrace
     version_constant: [OpenTelemetry, Propagator, OTTrace, VERSION]

instrumentation/rdkafka/Appraisals

@@ -4,7 +4,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-%w[0.12.0 0.13.0 0.14.0 0.15.0 0.16.0 0.17.0 0.18.0 0.19.0].each do |version|
+%w[0.18.0 0.19.0 0.20.0 0.21.0].each do |version|
   appraise "rdkafka-#{version}" do
     gem 'rdkafka', "~> #{version}"
   end

instrumentation/rdkafka/lib/opentelemetry/instrumentation/rdkafka/instrumentation.rb

@@ -11,7 +11,7 @@ module Rdkafka
       class Instrumentation < OpenTelemetry::Instrumentation::Base
         compatible do
           gem_version = Gem::Version.new(::Rdkafka::VERSION)
-          Gem::Requirement.new('>= 0.10.0', '< 0.20.0').satisfied_by?(gem_version)
+          Gem::Requirement.new('>= 0.18.0').satisfied_by?(gem_version)
         end
 
         install do |_config|

instrumentation/rdkafka/lib/opentelemetry/instrumentation/rdkafka/patches/consumer.rb

@@ -10,11 +10,7 @@ module Rdkafka
       module Patches
         # The Consumer module contains the instrumentation patch for the Consumer class
         module Consumer
-          GETTER = if Gem::Version.new(::Rdkafka::VERSION) >= Gem::Version.new('0.13.0')
-                     Context::Propagation.text_map_getter
-                   else
-                     OpenTelemetry::Common::Propagation.symbol_key_getter
-                   end
+          GETTER = Context::Propagation.text_map_getter
           private_constant :GETTER
 
           def each
@@ -42,26 +38,30 @@ def each
             end
           end
 
-          def each_batch(max_items: 100, bytes_threshold: Float::INFINITY, timeout_ms: 250, yield_on_error: false, &block)
-            super do |messages, error|
-              if messages.empty?
-                yield messages, error
-              else
-                attributes = {
-                  'messaging.system' => 'kafka',
-                  'messaging.destination_kind' => 'topic',
-                  'messaging.kafka.message_count' => messages.size
-                }
+          # each_batch method is deleted in rdkafka-ruby-0.20.0
+          # But, rdkafka-ruby-0.19.x and 0.18.x are still maintained
+          if Gem::Version.new(::Rdkafka::VERSION) < Gem::Version.new('0.20.0')
+            def each_batch(max_items: 100, bytes_threshold: Float::INFINITY, timeout_ms: 250, yield_on_error: false, &block)
+              super do |messages, error|
+                if messages.empty?
+                  yield messages, error
+                else
+                  attributes = {
+                    'messaging.system' => 'kafka',
+                    'messaging.destination_kind' => 'topic',
+                    'messaging.kafka.message_count' => messages.size
+                  }
 
-                links = messages.map do |message|
-                  trace_context = OpenTelemetry.propagation.extract(message.headers, getter: GETTER)
-                  span_context = OpenTelemetry::Trace.current_span(trace_context).context
-                  OpenTelemetry::Trace::Link.new(span_context) if span_context.valid?
-                end
-                links.compact!
+                  links = messages.map do |message|
+                    trace_context = OpenTelemetry.propagation.extract(message.headers, getter: GETTER)
+                    span_context = OpenTelemetry::Trace.current_span(trace_context).context
+                    OpenTelemetry::Trace::Link.new(span_context) if span_context.valid?
+                  end
+                  links.compact!
 
-                tracer.in_span('batch process', attributes: attributes, links: links, kind: :consumer) do
-                  yield messages, error
+                  tracer.in_span('batch process', attributes: attributes, links: links, kind: :consumer) do
+                    yield messages, error
+                  end
                 end
               end
             end

instrumentation/rdkafka/test/opentelemetry/instrumentation/rdkafka/patches/consumer_test.rb

@@ -166,68 +166,71 @@
       end
     end
 
-    describe '#each_batch' do
-      it 'traces each_batch call' do
-        skip "#{Rdkafka::VERSION} is not supported" unless instrumentation.compatible?
-
-        rand_hash = SecureRandom.hex(10)
-        topic_name = "consumer-patch-batch-trace-#{rand_hash}"
-        config = { 'bootstrap.servers': "#{host}:#{port}" }
-
-        producer = Rdkafka::Config.new(config).producer
-        delivery_handles = []
-
-        delivery_handles << producer.produce(
-          topic: topic_name,
-          payload: 'wow',
-          key: 'Key 1'
-        )
-
-        delivery_handles << producer.produce(
-          topic: topic_name,
-          payload: 'super',
-          key: 'Key 2'
-        )
-
-        delivery_handles.each(&:wait)
-
-        consumer_config = config.merge(
-          'group.id': 'me',
-          'auto.offset.reset': 'smallest' # https://stackoverflow.com/a/51081649
-        )
-        consumer = Rdkafka::Config.new(config.merge(consumer_config)).consumer
-        consumer.subscribe(topic_name)
-
-        begin
-          consumer.each_batch(max_items: 2) do |messages|
-            raise 'oops' unless messages.empty?
+    # each_batch method is deleted in rdkafka 0.20.0
+    if Gem::Version.new(Rdkafka::VERSION) < Gem::Version.new('0.20.0')
+      describe '#each_batch' do
+        it 'traces each_batch call' do
+          skip "#{Rdkafka::VERSION} is not supported" unless instrumentation.compatible?
+
+          rand_hash = SecureRandom.hex(10)
+          topic_name = "consumer-patch-batch-trace-#{rand_hash}"
+          config = { 'bootstrap.servers': "#{host}:#{port}" }
+
+          producer = Rdkafka::Config.new(config).producer
+          delivery_handles = []
+
+          delivery_handles << producer.produce(
+            topic: topic_name,
+            payload: 'wow',
+            key: 'Key 1'
+          )
+
+          delivery_handles << producer.produce(
+            topic: topic_name,
+            payload: 'super',
+            key: 'Key 2'
+          )
+
+          delivery_handles.each(&:wait)
+
+          consumer_config = config.merge(
+            'group.id': 'me',
+            'auto.offset.reset': 'smallest' # https://stackoverflow.com/a/51081649
+          )
+          consumer = Rdkafka::Config.new(config.merge(consumer_config)).consumer
+          consumer.subscribe(topic_name)
+
+          begin
+            consumer.each_batch(max_items: 2) do |messages|
+              raise 'oops' unless messages.empty?
+            end
+          rescue StandardError
           end
-        rescue StandardError
-        end
-
-        span = spans.find { |s| s.name == 'batch process' }
-        _(span.kind).must_equal(:consumer)
-        _(span.attributes['messaging.kafka.message_count']).must_equal(2)
 
-        event = span.events.first
-        _(event.name).must_equal('exception')
-        _(event.attributes['exception.type']).must_equal('RuntimeError')
-        _(event.attributes['exception.message']).must_equal('oops')
-
-        first_link = span.links[0]
-        linked_span_context = first_link.span_context
-        _(linked_span_context.trace_id).must_equal(spans[0].trace_id)
-        _(linked_span_context.span_id).must_equal(spans[0].span_id)
-
-        second_link = span.links[1]
-        linked_span_context = second_link.span_context
-        _(linked_span_context.trace_id).must_equal(spans[1].trace_id)
-        _(linked_span_context.span_id).must_equal(spans[1].span_id)
-
-        _(spans.size).must_equal(3)
-      ensure
-        begin; producer&.close; rescue StandardError; end
-        begin; consumer&.close; rescue StandardError; end
+          span = spans.find { |s| s.name == 'batch process' }
+          _(span.kind).must_equal(:consumer)
+          _(span.attributes['messaging.kafka.message_count']).must_equal(2)
+
+          event = span.events.first
+          _(event.name).must_equal('exception')
+          _(event.attributes['exception.type']).must_equal('RuntimeError')
+          _(event.attributes['exception.message']).must_equal('oops')
+
+          first_link = span.links[0]
+          linked_span_context = first_link.span_context
+          _(linked_span_context.trace_id).must_equal(spans[0].trace_id)
+          _(linked_span_context.span_id).must_equal(spans[0].span_id)
+
+          second_link = span.links[1]
+          linked_span_context = second_link.span_context
+          _(linked_span_context.trace_id).must_equal(spans[1].trace_id)
+          _(linked_span_context.span_id).must_equal(spans[1].span_id)
+
+          _(spans.size).must_equal(3)
+        ensure
+          begin; producer&.close; rescue StandardError; end
+          begin; consumer&.close; rescue StandardError; end
+        end
       end
     end
   end

propagator/google_cloud_trace_context/CHANGELOG.md

@@ -1 +1,5 @@
 # Release History: opentelemetry-propagator-google_cloud_trace_context
+
+### v0.1.0 / 2025-05-01
+
+Initial release.

propagator/google_cloud_trace_context/lib/opentelemetry/propagator/google_cloud_trace_context/version.rb

@@ -16,7 +16,7 @@ module Propagator
     # Namespace for OpenTelemetry GoogleCloudTraceContext propagation
     module GoogleCloudTraceContext
       # Current gem version
-      VERSION = '0.0.0'
+      VERSION = '0.1.0'
     end
   end
 end

propagator/xray/lib/opentelemetry/propagator/xray.rb

@@ -12,6 +12,7 @@
 # See the documentation for the `opentelemetry-api` gem for details.
 
 require_relative 'xray/text_map_propagator'
+require_relative 'xray/lambda_text_map_propagator'
 
 module OpenTelemetry
   # Namespace for OpenTelemetry propagator extension libraries
@@ -22,7 +23,8 @@ module XRay
 
       DEBUG_CONTEXT_KEY = Context.create_key('xray-debug-key')
       TEXT_MAP_PROPAGATOR = TextMapPropagator.new
-      private_constant :DEBUG_CONTEXT_KEY, :TEXT_MAP_PROPAGATOR
+      LAMBDA_TEXT_MAP_PROPAGATOR = LambdaTextMapPropagator.new
+      private_constant :DEBUG_CONTEXT_KEY, :TEXT_MAP_PROPAGATOR, :LAMBDA_TEXT_MAP_PROPAGATOR
 
       # @api private
       # Returns a new context with the xray debug flag enabled
@@ -41,6 +43,12 @@ def debug?(context)
       def text_map_propagator
         TEXT_MAP_PROPAGATOR
       end
+
+      # Returns a text map propagator that propagates context in the XRay
+      # format with special handling for Lambda environment.
+      def lambda_text_map_propagator
+        LAMBDA_TEXT_MAP_PROPAGATOR
+      end
     end
   end
 end

propagator/xray/lib/opentelemetry/propagator/xray/lambda_text_map_propagator.rb

@@ -0,0 +1,61 @@
+# frozen_string_literal: true
+
+# Copyright OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+# OpenTelemetry is an open source observability framework, providing a
+# general-purpose API, SDK, and related tools required for the instrumentation
+# of cloud-native software, frameworks, and libraries.
+#
+# The OpenTelemetry module provides global accessors for telemetry objects.
+# See the documentation for the `opentelemetry-api` gem for details.
+module OpenTelemetry
+  # Namespace for OpenTelemetry propagator extension libraries
+  module Propagator
+    # Namespace for OpenTelemetry XRay propagation
+    module XRay
+      # Implementation of AWS X-Ray Trace Header propagation with special handling for
+      # Lambda's _X_AMZN_TRACE_ID environment variable
+      class LambdaTextMapPropagator < TextMapPropagator
+        AWS_TRACE_HEADER_ENV_KEY = '_X_AMZN_TRACE_ID'
+
+        # Extract trace context from the supplied carrier or from Lambda environment variable
+        # If extraction fails, the original context will be returned
+        #
+        # @param [Carrier] carrier The carrier to get the header from
+        # @param [optional Context] context Context to be updated with the trace context
+        #   extracted from the carrier. Defaults to +Context.current+.
+        # @param [optional Getter] getter If the optional getter is provided, it
+        #   will be used to read the header from the carrier, otherwise the default
+        #   text map getter will be used.
+        #
+        # @return [Context] context updated with extracted baggage, or the original context
+        #   if extraction fails
+        def extract(carrier, context: Context.current, getter: Context::Propagation.text_map_getter)
+          # Check if the original input context already has a valid span
+          span_context = Trace.current_span(context).context
+          # If original context is valid, just return it - do not extract from carrier
+          return context if span_context.valid?
+
+          # First try to extract from the carrier using the standard X-Ray propagator
+          xray_context = super
+
+          # Check if we successfully extracted a context from the carrier
+          span_context = Trace.current_span(xray_context).context
+          return xray_context if span_context.valid?
+
+          # If not, check for the Lambda environment variable
+          trace_header = ENV.fetch(AWS_TRACE_HEADER_ENV_KEY, nil)
+          return xray_context unless trace_header
+
+          # Create a carrier with the trace header and extract from it
+          env_carrier = { XRAY_CONTEXT_KEY => trace_header }
+          super(env_carrier, context: xray_context, getter: getter)
+        rescue OpenTelemetry::Error
+          context
+        end
+      end
+    end
+  end
+end