[WIP] RUBY-3612 OpenTelementry

gemfiles/standard.rb

@@ -4,6 +4,7 @@
 def standard_dependencies
   gem 'yard', '>= 0.9.35'
   gem 'ffi'
+  gem 'opentelemetry-sdk'
 
   group :development, :testing do
     gem 'jruby-openssl', platforms: :jruby

lib/mongo.rb

@@ -74,6 +74,7 @@
 require 'mongo/socket'
 require 'mongo/srv'
 require 'mongo/timeout'
+require 'mongo/tracing'
 require 'mongo/uri'
 require 'mongo/version'
 require 'mongo/write_concern'

lib/mongo/client.rb

@@ -112,6 +112,7 @@ class Client
       :ssl_verify_hostname,
       :ssl_verify_ocsp_endpoint,
       :timeout_ms,
+      :tracing,
       :truncate_logs,
       :user,
       :wait_queue_timeout,
@@ -437,6 +438,20 @@ def hash
     #   See Ruby's Zlib module for valid levels.
     # @option options [ Hash ] :resolv_options For internal driver use only.
     #   Options to pass through to Resolv::DNS constructor for SRV lookups.
+    # @option options [ Hash ] :tracing OpenTelemetry tracing options.
+    #   - :enabled => Boolean, whether to enable OpenTelemetry tracing. The default
+    #     value is nil that means that the configuration will be taken from the
+    #     OTEL_RUBY_INSTRUMENTATION_MONGODB_ENABLED environment variable.
+    #   - :tracer => OpenTelemetry::Trace::Tracer, the tracer to use for
+    #     tracing. Must be an implementation of OpenTelemetry::Trace::Tracer
+    #     interface.
+    #   - :query_text_max_length => Integer, the maximum length of the query text
+    #     to be included in the span attributes. If the query text exceeds this
+    #     length, it will be truncated. Value 0 means no query text
+    #     will be included in the span attributes. The default value is nil that
+    #     means that the configuration will be taken from the
+    #     OTEL_RUBY_INSTRUMENTATION_MONGODB_QUERY_TEXT_MAX_LENGTH environment
+    #     variable.
     # @option options [ Hash ] :auto_encryption_options Auto-encryption related
     #   options.
     #   - :key_vault_client => Client | nil, a client connected to the MongoDB
@@ -1195,6 +1210,15 @@ def timeout_sec
       end
     end
 
+    def tracer
+      tracing_opts = @options[:tracing] || {}
+      @tracer ||= Tracing.create_tracer(
+        enabled: tracing_opts[:enabled],
+        query_text_max_length: tracing_opts[:query_text_max_length],
+        otel_tracer: tracing_opts[:tracer],
+      )
+    end
+
     private
 
     # Attempts to parse the given list of addresses, using the provided options.

lib/mongo/collection.rb

@@ -57,6 +57,8 @@ class Collection
     # Delegate to the cluster for the next primary.
     def_delegators :cluster, :next_primary
 
+    def_delegators :client, :tracer
+
     # Options that can be updated on a new Collection instance via the #with method.
     #
     # @since 2.1.0
@@ -865,19 +867,22 @@ def insert_one(document, opts = {})
           session: session,
           operation_timeouts: operation_timeouts(opts)
           )
-        write_with_retry(write_concern, context: context) do |connection, txn_num, context|
-          Operation::Insert.new(
-            :documents => [ document ],
-            :db_name => database.name,
-            :coll_name => name,
-            :write_concern => write_concern,
-            :bypass_document_validation => !!opts[:bypass_document_validation],
-            :options => opts,
-            :id_generator => client.options[:id_generator],
-            :session => session,
-            :txn_num => txn_num,
-            :comment => opts[:comment]
-          ).execute_with_connection(connection, context: context)
+        operation = Operation::Insert.new(
+              :documents => [ document ],
+              :db_name => database.name,
+              :coll_name => name,
+              :write_concern => write_concern,
+              :bypass_document_validation => !!opts[:bypass_document_validation],
+              :options => opts,
+              :id_generator => client.options[:id_generator],
+              :session => session,
+              :comment => opts[:comment]
+            )
+        tracer.trace_operation('insert_one', operation, context) do
+          write_with_retry(write_concern, context: context) do |connection, txn_num, context|
+            operation.txn_num = txn_num
+            operation.execute_with_connection(connection, context: context)
+          end
         end
       end
     end

lib/mongo/collection/view.rb

@@ -72,6 +72,8 @@ class View
       # Delegate to the cluster for the next primary.
       def_delegators :cluster, :next_primary
 
+      def_delegators :client, :tracer
+
       alias :selector :filter
 
       # @return [ Integer | nil | The timeout_ms value that was passed as an

lib/mongo/collection/view/iterable.rb

@@ -88,19 +88,21 @@ def select_cursor(session)
             operation_timeouts: operation_timeouts,
             view: self
           )
-
-          if respond_to?(:write?, true) && write?
-            server = server_selector.select_server(cluster, nil, session, write_aggregation: true)
-            result = send_initial_query(server, context)
-
-            if use_query_cache?
-              CachingCursor.new(view, result, server, session: session, context: context)
+          op = initial_query_op(session)
+          tracer.trace_operation('get_more', op, context) do
+            if respond_to?(:write?, true) && write?
+              server = server_selector.select_server(cluster, nil, session, write_aggregation: true)
+              result = send_initial_query(server, context)
+
+              if use_query_cache?
+                CachingCursor.new(view, result, server, session: session, context: context)
+              else
+                Cursor.new(view, result, server, session: session, context: context)
+              end
             else
-              Cursor.new(view, result, server, session: session, context: context)
-            end
-          else
-            read_with_retry_cursor(session, server_selector, view, context: context) do |server|
-              send_initial_query(server, context)
+              read_with_retry_cursor(session, server_selector, view, context: context) do |server|
+                send_initial_query(server, context)
+              end
             end
           end
         end

lib/mongo/cursor.rb

@@ -39,6 +39,7 @@ class Cursor
     def_delegators :@view, :collection
     def_delegators :collection, :client, :database
     def_delegators :@server, :cluster
+    def_delegators :client, :tracer
 
     # @return [ Collection::View ] view The collection view.
     attr_reader :view
@@ -514,10 +515,12 @@ def unregister
 
     def execute_operation(op, context: nil)
       op_context = context || possibly_refreshed_context
-      if @connection.nil?
-        op.execute(@server, context: op_context)
-      else
-        op.execute_with_connection(@connection, context: op_context)
+      tracer.trace_operation('find', op, op_context) do
+        if @connection.nil?
+          op.execute(@server, context: op_context)
+        else
+          op.execute_with_connection(@connection, context: op_context)
+        end
       end
     end
 

lib/mongo/operation/context.rb

@@ -69,6 +69,7 @@ def initialize(
       attr_reader :session
       attr_reader :view
       attr_reader :options
+      attr_accessor :tracer
 
       # Returns a new Operation::Context with the deadline refreshed
       # and relative to the current moment.

lib/mongo/operation/insert/op_msg.rb

@@ -34,8 +34,14 @@ class OpMsg < OpMsgBase
         private
 
         def get_result(connection, context, options = {})
-          # This is a Mongo::Operation::Insert::Result
-          Result.new(*dispatch_message(connection, context), @ids, context: context)
+          message = build_message(connection, context)
+          if (tracer = context.tracer)
+            tracer.trace_command(message, context, connection) do
+              Result.new(*dispatch_message(message, connection, context), @ids, context: context)
+            end
+          else
+            Result.new(*dispatch_message(message, connection, context), @ids, context: context)
+          end
         end
 
         def selector(connection)

lib/mongo/operation/shared/executable.rb

@@ -104,12 +104,18 @@ def result_class
       end
 
       def get_result(connection, context, options = {})
-        result_class.new(*dispatch_message(connection, context, options), context: context, connection: connection)
+        message = build_message(connection, context)
+        if (tracer = context.tracer)
+          tracer.trace_command(message, context, connection) do
+            result_class.new(*dispatch_message(message, connection, context, options), context: context, connection: connection)
+          end
+        else
+          result_class.new(*dispatch_message(message, connection, context, options), context: context, connection: connection)
+        end
       end
 
       # Returns a Protocol::Message or nil as reply.
-      def dispatch_message(connection, context, options = {})
-        message = build_message(connection, context)
+      def dispatch_message(message, connection, context, options = {})
         message = message.maybe_encrypt(connection, context)
         reply = connection.dispatch([ message ], context, options)
         [reply, connection.description, connection.global_id]

lib/mongo/operation/shared/specifiable.rb

@@ -526,6 +526,10 @@ def txn_num
         @spec[:txn_num]
       end
 
+      def txn_num=(num)
+        @spec[:txn_num] = num
+      end
+
       # The command.
       #
       # @return [ Hash ] The command.

lib/mongo/server/connection.rb

@@ -388,6 +388,17 @@ def record_checkin!
         self
       end
 
+      def transport
+        return nil if @socket.nil?
+
+        case @socket
+        when Mongo::Socket::Unix
+          :unix
+        else
+          :tcp
+        end
+      end
+
       private
 
       def deliver(message, client, options = {})

lib/mongo/tracing.rb

@@ -0,0 +1,16 @@
+# frozen_string_literal: true
+
+module Mongo
+  module Tracing
+    def create_tracer(enabled: nil, query_text_max_length: nil, otel_tracer: nil)
+      OpenTelemetry::Tracer.new(
+        enabled: enabled,
+        query_text_max_length: query_text_max_length,
+        otel_tracer: otel_tracer,
+      )
+    end
+    module_function :create_tracer
+  end
+end
+
+require 'mongo/tracing/open_telemetry'

lib/mongo/tracing/open_telemetry.rb

@@ -0,0 +1,26 @@
+# frozen_string_literal: true
+
+# Copyright (C) 2025-present MongoDB Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the 'License');
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an 'AS IS' BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+module Mongo
+  module Tracing
+    module OpenTelemetry
+    end
+  end
+end
+
+require 'mongo/tracing/open_telemetry/command_tracer'
+require 'mongo/tracing/open_telemetry/operation_tracer'
+require 'mongo/tracing/open_telemetry/tracer'

lib/mongo/tracing/open_telemetry/command_tracer.rb

@@ -0,0 +1,94 @@
+# frozen_string_literal: true
+
+# Copyright (C) 2025-present MongoDB Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the 'License');
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an 'AS IS' BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+module Mongo
+  module Tracing
+    module OpenTelemetry
+      class CommandTracer
+        def initialize(otel_tracer, query_text_max_length: 0)
+          @otel_tracer = otel_tracer
+          @query_text_max_length = query_text_max_length
+        end
+
+        def trace_command(message, _operation_context, connection)
+          @otel_tracer.in_span(
+            command_span_name(message),
+            attributes: span_attributes(message, connection),
+            kind: :client
+          ) do |span, _context|
+            yield.tap do |result|
+              if result.respond_to?(:cursor_id) && result.cursor_id.positive?
+                span.set_attribute('db.mongodb.cursor_id', result.cursor_id)
+              end
+            end
+          end
+        end
+
+        private
+
+        def span_attributes(message, connection)
+          {
+            'db.system' => 'mongodb',
+            'db.namespace' => message.documents.first['$db'],
+            'db.collection.name' => collection_name(message),
+            'db.operation.name' => message.documents.first.keys.first,
+            'server.port' => connection.address.port,
+            'server.address' => connection.address.host,
+            'network.transport' => connection.transport.to_s,
+            'db.mongodb.server_connection_id' => connection.server.description.server_connection_id,
+            'db.mongodb.driver_connection_id' => connection.id,
+            'db.query.text' => query_text(message)
+          }.compact
+        end
+
+        def command_span_name(message)
+          message.documents.first.keys.first
+        end
+
+        def collection_name(message)
+          case message.documents.first.keys.first
+          when 'getMore'
+            message.documents.first['collection']
+          else
+            message.documents.first.values.first
+          end
+        end
+
+        def query_text?
+          @query_text_max_length.positive?
+        end
+
+        EXCLUDED_KEYS = %w[lsid $db $clusterTime signature].freeze
+        ELLIPSES = '...'
+
+        def query_text(message)
+          return unless query_text?
+
+          text = message
+                 .documents
+                 .first
+                 .reject { |key, _| EXCLUDED_KEYS.include?(key) }
+                 .to_json
+          if text.length > @query_text_max_length
+            "#{text[0...@query_text_max_length]}#{ELLIPSES}"
+          else
+            text
+          end
+        end
+      end
+    end
+  end
+end