wip

Author: comandeo-mongo

lib/mongo/collection/view/aggregation.rb

@@ -25,11 +25,14 @@ class View
       #
       # @since 2.0.0
       class Aggregation
+        extend Forwardable
         include Behavior
 
         # @return [ Array<Hash> ] pipeline The aggregation pipeline.
         attr_reader :pipeline
 
+        def_delegators :view, :tracer
+
         # Initialize the aggregation for the provided collection view, pipeline
         # and options.
         #
@@ -80,7 +83,7 @@ def new(options)
           Aggregation.new(view, pipeline, options)
         end
 
-        def initial_query_op(session, read_preference)
+        def initial_query_op(session, read_preference = nil)
           Operation::Aggregate.new(aggregate_spec(session, read_preference))
         end
 

lib/mongo/collection/view/aggregation/behavior.rb

@@ -88,7 +88,7 @@ def server_selector
             @view.send(:server_selector)
           end
 
-          def aggregate_spec(session, read_preference)
+          def aggregate_spec(session, read_preference = nil)
             Builder::Aggregation.new(
               pipeline,
               view,

lib/mongo/collection/view/iterable.rb

@@ -89,7 +89,15 @@ def select_cursor(session)
             view: self
           )
           op = initial_query_op(session)
-          tracer.trace_operation('get_more', op, context) do
+          op_name = case op
+                    when Mongo::Operation::Find
+                      'find'
+                    when Mongo::Operation::Aggregate
+                      'aggregate'
+                    else
+                      op.class.name.split('::').last.downcase
+                    end
+          tracer.trace_operation(op_name, 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)

lib/mongo/cursor.rb

@@ -514,6 +514,12 @@ def unregister
     end
 
     def execute_operation(op, context: nil)
+      op_name = case op
+                when Mongo::Operation::GetMore
+                  'get_more'
+                when Mongo::Operation::Close
+                  'close'
+                end
       op_context = context || possibly_refreshed_context
       tracer.trace_operation('find', op, op_context) do
         if @connection.nil?

lib/mongo/operation/shared/specifiable.rb

@@ -233,7 +233,7 @@ def documents
       #
       # @since 2.0.0
       def coll_name
-        spec.fetch(COLL_NAME)
+        spec[COLL_NAME]
       end
 
       # The id of the cursor created on the server.

lib/mongo/tracing/open_telemetry/operation_tracer.rb

@@ -28,30 +28,29 @@ def initialize(otel_tracer, parent_tracer)
         end
 
         def trace_operation(name, operation, operation_context)
-          parent_context = parent_context_for(operation_context, operation.cursor_id)
-          operation_context.tracer = @parent_tracer
-          span = @otel_tracer.start_span(
+          @otel_tracer.in_span(
             operation_span_name(name, operation),
             attributes: span_attributes(name, operation),
-            with_parent: parent_context,
             kind: :client
-          )
-          ::OpenTelemetry::Trace.with_span(span) do |_s, c|
+          ) do |span, _context|
+            operation_context.tracer = @parent_tracer
             yield.tap do |result|
-              process_cursor_context(result, operation.cursor_id, c)
+              if result.is_a?(Cursor) && result.id.positive?
+                span.set_attribute('db.mongodb.cursor_id', result.id)
+              end
             end
           end
-        rescue Exception => e
-          span&.record_exception(e)
-          span&.status = ::OpenTelemetry::Trace::Status.error("Unhandled exception of type: #{e.class}")
-          raise e
         ensure
-          span&.finish
           operation_context.tracer = nil
         end
 
         private
 
+        # Returns a hash of attributes for the OpenTelemetry span for the operation.
+        #
+        # @param name [String] The name of the operation.
+        # @param operation [Operation] The operation being traced.
+        # @return [Hash] A hash of attributes for the span.
         def span_attributes(name, operation)
           {
             'db.system' => 'mongodb',
@@ -63,49 +62,11 @@ def span_attributes(name, operation)
           }.compact
         end
 
-        def parent_context_for(operation_context, cursor_id)
-          if (key = transaction_map_key(operation_context.session))
-            transaction_context_map[key]
-          elsif cursor_id
-            cursor_context_map[cursor_id]
-          end
-        end
-
-        # This map is used to store OpenTelemetry context for cursor_id.
-        # This allows to group all operations related to a cursor under the same context.
-        #
-        # # @return [Hash] a map of cursor_id to OpenTelemetry context.
-        def cursor_context_map
-          @cursor_context_map ||= {}
-        end
-
-        def process_cursor_context(result, cursor_id, context)
-          return unless result.is_a?(Cursor)
-
-          if result.id.zero?
-            # If the cursor is closed, remove it from the context map.
-            cursor_context_map.delete(cursor_id)
-          elsif result.id && cursor_id.nil?
-            # New cursor created, store its context.
-            cursor_context_map[result.id] = context
-          end
-        end
-
-        # This map is used to store OpenTelemetry context for transaction.
-        #   This allows to group all operations related to a transaction under the same context.
+        # Returns the name of the span for the operation.
         #
-        #   @return [Hash] a map of transaction_id to OpenTelemetry context.
-        def transaction_context_map
-          @transaction_context_map ||= {}
-        end
-
-        # @param session [Mongo::Session] the session for which to get the transaction map key.
-        def transaction_map_key(session)
-          return if session.nil? || session.implicit? || !session.in_transaction?
-
-          "#{session.id}-#{session.txn_num}"
-        end
-
+        # @param name [String] The name of the operation.
+        # @param operation [Operation] The operation being traced.
+        # # @return [String] The name of the span.
         def operation_span_name(name, operation)
           if operation.coll_name
             "#{name} #{operation.db_name}.#{operation.coll_name}"

spec/runners/unified/test.rb

@@ -200,7 +200,7 @@ def generate_entities(es)
           if observe_spans
             opts[:tracing] = {
               enabled: true,
-              tracer: tracer,
+              # tracer: tracer,
             }
           end