feat: draft Mongo semantic stability

instrumentation/mongo/Appraisals

@@ -4,9 +4,18 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-appraise 'mongo-2.22' do
-  gem 'mongo', '~> 2.22.0'
+# To facilitate database semantic convention stability migration, we are using
+# appraisal to test the different semantic convention modes along with different
+# gem versions. For more information on the semantic convention modes, see:
+# https://opentelemetry.io/docs/specs/semconv/non-normative/db-migration/
 
-  # TODO: bson 5.1.0 isn't compatible with JRuby as of 2025/06/17
-  gem 'bson', '< 5.1.0' if defined?(JRUBY_VERSION)
+semconv_stability = %w[old stable dup]
+
+semconv_stability.each do |mode|
+  appraise "mongo-latest-#{mode}" do
+    gem 'mongo', '< 2.23.0' # Mongo 2.23.0+ has native OpenTelemetry instrumentation
+
+    # TODO: bson 5.1.0 isn't compatible with JRuby as of 2025/06/17
+    gem 'bson', '< 5.1.0' if defined?(JRUBY_VERSION)
+  end
 end

instrumentation/mongo/README.md

@@ -75,6 +75,22 @@ The `opentelemetry-instrumentation-mongo` gem source is [on github][repo-github]
 
 The OpenTelemetry Ruby gems are maintained by the OpenTelemetry Ruby special interest group (SIG). You can get involved by joining us on our [GitHub Discussions][discussions-url], [Slack Channel][slack-channel] or attending our weekly meeting. See the [meeting calendar][community-meetings] for dates and times. For more information on this and other language SIGs, see the OpenTelemetry [community page][ruby-sig].
 
+## Database semantic convention stability
+
+In the OpenTelemetry ecosystem, database semantic conventions have now reached a stable state. However, the initial Mongo instrumentation was introduced before this stability was achieved, which resulted in database attributes being based on an older version of the semantic conventions.
+
+To facilitate the migration to stable semantic conventions, you can use the `OTEL_SEMCONV_STABILITY_OPT_IN` environment variable. This variable allows you to opt-in to the new stable conventions, ensuring compatibility and future-proofing your instrumentation.
+
+When setting the value for `OTEL_SEMCONV_STABILITY_OPT_IN`, you can specify which conventions you wish to adopt:
+
+- `database` - Emits the stable database and networking conventions and ceases emitting the old conventions previously emitted by the instrumentation.
+- `database/dup` - Emits both the old and stable database and networking conventions, enabling a phased rollout of the stable semantic conventions.
+- Default behavior (in the absence of either value) is to continue emitting the old database and networking conventions the instrumentation previously emitted.
+
+During the transition from old to stable conventions, Mongo instrumentation code comes in three patch versions: `dup`, `old`, and `stable`. These versions are identical except for the attributes they send. Any changes to Mongo instrumentation should consider all three patches.
+
+For additional information on migration, please refer to our [documentation](https://opentelemetry.io/docs/specs/semconv/non-normative/db-migration/).
+
 ## License
 
 Apache 2.0 license. See [LICENSE][license-github] for more information.

instrumentation/mongo/lib/opentelemetry/instrumentation/mongo/instrumentation.rb

@@ -35,19 +35,53 @@ class Instrumentation < OpenTelemetry::Instrumentation::Base
         option :peer_service, default: nil, validate: :string
         option :db_statement, default: :obfuscate, validate: %I[omit obfuscate include]
 
+        attr_reader :semconv
+
         private
 
         def gem_version
           Gem::Version.new(::Mongo::VERSION)
         end
 
+        def determine_semconv
+          opt_in = ENV.fetch('OTEL_SEMCONV_STABILITY_OPT_IN', nil)
+          return :old if opt_in.nil?
+
+          opt_in_values = opt_in.split(',').map(&:strip)
+
+          if opt_in_values.include?('database/dup')
+            :dup
+          elsif opt_in_values.include?('database')
+            :stable
+          else
+            :old
+          end
+        end
+
         def require_dependencies
-          require_relative 'subscriber'
+          @semconv = determine_semconv
+
+          case @semconv
+          when :old
+            require_relative 'subscribers/old/subscriber'
+          when :stable
+            require_relative 'subscribers/stable/subscriber'
+          when :dup
+            require_relative 'subscribers/dup/subscriber'
+          end
         end
 
         def register_subscriber
+          subscriber_class = case @semconv
+                             when :stable
+                               Subscribers::Stable::Subscriber
+                             when :dup
+                               Subscribers::Dup::Subscriber
+                             else
+                               Subscribers::Old::Subscriber
+                             end
           # Subscribe to all COMMAND queries with our subscriber class
-          ::Mongo::Monitoring::Global.subscribe(::Mongo::Monitoring::COMMAND, Subscriber.new)
+          ::Mongo::Monitoring::Global.subscribe(::Mongo::Monitoring::COMMAND, subscriber_class.new)
         end
       end
     end

instrumentation/mongo/lib/opentelemetry/instrumentation/mongo/subscribers/dup/subscriber.rb

@@ -0,0 +1,137 @@
+# frozen_string_literal: true
+
+# Copyright The OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+require_relative '../../command_serializer'
+
+module OpenTelemetry
+  module Instrumentation
+    module Mongo
+      module Subscribers
+        module Dup
+          # Event handler class for Mongo Ruby driver (dup mode - emits both old and new semantic conventions)
+          class Subscriber
+            THREAD_KEY = :__opentelemetry_mongo_spans__
+
+            def started(event)
+              # start a trace and store it in the current thread; using the `operation_id`
+              # is safe since it's a unique id used to link events together. Also only one
+              # thread is involved in this execution so thread-local storage should be safe. Reference:
+              # https://github.com/mongodb/mongo-ruby-driver/blob/master/lib/mongo/monitoring.rb#L70
+              # https://github.com/mongodb/mongo-ruby-driver/blob/master/lib/mongo/monitoring/publishable.rb#L38-L56
+
+              collection = get_collection(event.command)
+
+              # Old conventions
+              attributes = {
+                'db.system' => 'mongodb',
+                'db.name' => event.database_name,
+                'db.operation' => event.command_name,
+                'net.peer.name' => event.address.host,
+                'net.peer.port' => event.address.port
+              }
+
+              # New stable conventions
+              attributes['db.system.name'] = 'mongodb'
+              attributes['db.namespace'] = event.database_name
+              attributes['db.operation.name'] = event.command_name
+              attributes['server.address'] = event.address.host
+              attributes['server.port'] = event.address.port
+
+              config = Mongo::Instrumentation.instance.config
+              attributes['peer.service'] = config[:peer_service] if config[:peer_service]
+              omit = config[:db_statement] == :omit
+              obfuscate = config[:db_statement] == :obfuscate
+              unless omit
+                serialized = CommandSerializer.new(event.command, obfuscate).serialize
+                # Both old and new attributes
+                attributes['db.statement'] = serialized
+                attributes['db.query.text'] = serialized
+              end
+              if collection
+                # Both old and new attributes
+                attributes['db.mongodb.collection'] = collection
+                attributes['db.collection.name'] = collection
+              end
+              attributes.compact!
+
+              span = tracer.start_span(span_name(collection, event.command_name), attributes: attributes, kind: :client)
+              set_span(event, span)
+            end
+
+            def failed(event)
+              finish_event('failed', event) do |span|
+                if event.is_a?(::Mongo::Monitoring::Event::CommandFailed)
+                  error_type = extract_error_type(event)
+                  # New stable convention attributes
+                  span.set_attribute('error.type', error_type)
+                  status_code = event.failure['code']
+                  span.set_attribute('db.response.status_code', status_code.to_s) if status_code
+                  span.add_event('exception',
+                                 attributes: {
+                                   'exception.type' => 'CommandFailed',
+                                   'exception.message' => event.message
+                                 })
+                  span.status = OpenTelemetry::Trace::Status.error(event.message)
+                end
+              end
+            end
+
+            def succeeded(event)
+              finish_event('succeeded', event)
+            end
+
+            private
+
+            def finish_event(name, event)
+              span = get_span(event)
+              return unless span
+
+              yield span if block_given?
+            rescue StandardError => e
+              OpenTelemetry.logger.debug("error when handling MongoDB '#{name}' event: #{e}")
+            ensure
+              # finish span to prevent leak and remove it from thread storage
+              span&.finish
+              clear_span(event)
+            end
+
+            def span_name(collection, command_name)
+              return command_name unless collection
+
+              "#{command_name} #{collection}"
+            end
+
+            def get_collection(command)
+              collection = command.values.first
+              collection if collection.is_a?(String)
+            end
+
+            def get_span(event)
+              Thread.current[THREAD_KEY]&.[](event.request_id)
+            end
+
+            def set_span(event, span)
+              Thread.current[THREAD_KEY] ||= {}
+              Thread.current[THREAD_KEY][event.request_id] = span
+            end
+
+            def clear_span(event)
+              Thread.current[THREAD_KEY]&.delete(event.request_id)
+            end
+
+            def extract_error_type(event)
+              event.failure['codeName'] || 'CommandFailed'
+            end
+
+            def tracer
+              Mongo::Instrumentation.instance.tracer
+            end
+          end
+        end
+      end
+    end
+  end
+end