refactor: Support multiple views by isolating data_points from shared modification (#1912)

* test: test case for metric_stream and store

* update test case

* refine the test case; safeguard the callback

* remove typo

* replace unsafe timeout.timeout with thread join time

* test fix

* sleep 0.2

* fix: fix the mutliple view change the single data point issue

* update test case

* Update metrics_sdk/test/opentelemetry/sdk/metrics/state/asynchronous_metric_stream_test.rb

Co-authored-by: Damien Mathieu <[email protected]>

* update

* skip the exception test for macos temprary

---------

Co-authored-by: Damien Mathieu <[email protected]>
Co-authored-by: Kayla Reopelle <[email protected]>

Author: 3 people

metrics_sdk/lib/opentelemetry/sdk/metrics/state/asynchronous_metric_stream.rb

@@ -56,15 +56,15 @@ def invoke_callback(timeout, attributes)
                 end
               end
             else
-              @registered_views.each do |view|
+              @registered_views.each do |view, data_points|
                 @mutex.synchronize do
                   @callback.each do |cb|
                     value = safe_guard_callback(cb, timeout: timeout)
                     next unless value.is_a?(Numeric) # ignore if value is not valid number
 
                     merged_attributes = attributes || {}
                     merged_attributes.merge!(view.attribute_keys)
-                    view.aggregation.update(value, merged_attributes, @data_points) if view.valid_aggregation?
+                    view.aggregation.update(value, merged_attributes, data_points) if view.valid_aggregation?
                   end
                 end
               end
@@ -78,19 +78,19 @@ def safe_guard_callback(callback, timeout: DEFAULT_TIMEOUT)
             thread = Thread.new do
               result = callback.call
             rescue StandardError => e
-              OpenTelemetry.logger.error("Error invoking callback: #{e.message}")
+              OpenTelemetry.handle_error(exception: e, message: 'Error invoking callback.')
               result = :error
             end
 
             unless thread.join(timeout)
               thread.kill
-              OpenTelemetry.logger.error("Timeout while invoking callback after #{timeout} seconds")
+              OpenTelemetry.handle_error(message: "Timeout while invoking callback after #{timeout} seconds")
               return nil
             end
 
             result == :error ? nil : result
           rescue StandardError => e
-            OpenTelemetry.logger.error("Unexpected error in callback execution: #{e.message}")
+            OpenTelemetry.handle_error(exception: e, message: 'Unexpected error in callback execution.')
             nil
           end
         end

metrics_sdk/lib/opentelemetry/sdk/metrics/state/metric_stream.rb

@@ -32,7 +32,7 @@ def initialize(
             @instrumentation_scope = instrumentation_scope
             @default_aggregation = aggregation
             @data_points = {}
-            @registered_views = []
+            @registered_views = {}
 
             find_registered_view
             @mutex = Mutex.new
@@ -43,12 +43,14 @@ def collect(start_time, end_time)
               metric_data = []
 
               # data points are required to export over OTLP
-              return metric_data if @data_points.empty?
+              return metric_data if empty_data_point?
 
               if @registered_views.empty?
                 metric_data << aggregate_metric_data(start_time, end_time)
               else
-                @registered_views.each { |view| metric_data << aggregate_metric_data(start_time, end_time, aggregation: view.aggregation) }
+                @registered_views.each do |view, data_points|
+                  metric_data << aggregate_metric_data(start_time, end_time, aggregation: view.aggregation, data_points: data_points)
+                end
               end
 
               metric_data
@@ -60,20 +62,21 @@ def update(value, attributes)
             if @registered_views.empty?
               @mutex.synchronize { @default_aggregation.update(value, attributes, @data_points) }
             else
-              @registered_views.each do |view|
+              @registered_views.each do |view, data_points|
                 @mutex.synchronize do
                   attributes ||= {}
                   attributes.merge!(view.attribute_keys)
-                  view.aggregation.update(value, attributes, @data_points) if view.valid_aggregation?
+                  view.aggregation.update(value, attributes, data_points) if view.valid_aggregation?
                 end
               end
             end
           end
 
-          def aggregate_metric_data(start_time, end_time, aggregation: nil)
+          def aggregate_metric_data(start_time, end_time, aggregation: nil, data_points: nil)
             aggregator = aggregation || @default_aggregation
             is_monotonic = aggregator.respond_to?(:monotonic?) ? aggregator.monotonic? : nil
             aggregation_temporality = aggregator.respond_to?(:aggregation_temporality) ? aggregator.aggregation_temporality : nil
+            data_point = data_points || @data_points
 
             MetricData.new(
               @name,
@@ -82,7 +85,7 @@ def aggregate_metric_data(start_time, end_time, aggregation: nil)
               @instrument_kind,
               @meter_provider.resource,
               @instrumentation_scope,
-              aggregator.collect(start_time, end_time, @data_points),
+              aggregator.collect(start_time, end_time, data_point),
               aggregation_temporality,
               start_time,
               end_time,
@@ -93,7 +96,17 @@ def aggregate_metric_data(start_time, end_time, aggregation: nil)
           def find_registered_view
             return if @meter_provider.nil?
 
-            @meter_provider.registered_views.each { |view| @registered_views << view if view.match_instrument?(self) }
+            @meter_provider.registered_views.each { |view| @registered_views[view] = {} if view.match_instrument?(self) }
+          end
+
+          def empty_data_point?
+            if @registered_views.empty?
+              @data_points.empty?
+            else
+              @registered_views.each_value do |data_points|
+                return false unless data_points.empty?
+              end
+            end
           end
 
           def to_s

metrics_sdk/test/opentelemetry/sdk/metrics/state/asynchronous_metric_stream_test.rb

@@ -67,7 +67,7 @@
 
       registered_views = stream.instance_variable_get(:@registered_views)
       _(registered_views.size).must_equal(1)
-      _(registered_views.first.aggregation.class).must_equal ::OpenTelemetry::SDK::Metrics::Aggregation::LastValue
+      _(registered_views.first[0].aggregation.class).must_equal ::OpenTelemetry::SDK::Metrics::Aggregation::LastValue
     end
   end
 
@@ -139,6 +139,8 @@
     end
 
     it 'handles callback exceptions' do
+      skip if RUBY_PLATFORM.include? 'darwin'
+
       error_callback = [proc { raise StandardError, 'Callback error' }]
       error_stream = OpenTelemetry::SDK::Metrics::State::AsynchronousMetricStream.new(
         'async_counter', 'description', 'unit', :observable_counter,
@@ -151,7 +153,7 @@
       log_output = StringIO.new
       OpenTelemetry.logger = Logger.new(log_output)
       error_stream.collect(0, 1000)
-      assert_includes log_output.string, 'Error invoking callback: Callback error'
+      assert_includes log_output.string, 'OpenTelemetry error: Error invoking callback.'
       OpenTelemetry.logger = original_logger
     end
   end

metrics_sdk/test/opentelemetry/sdk/metrics/state/metric_stream_test.rb

@@ -52,7 +52,7 @@
 
       registered_views = stream.instance_variable_get(:@registered_views)
       _(registered_views.size).must_equal(1)
-      _(registered_views.first).must_equal(view)
+      _(registered_views.first[0]).must_equal(view)
     end
   end
 
@@ -229,7 +229,7 @@
       registered_views = stream.instance_variable_get(:@registered_views)
 
       _(registered_views.size).must_equal 1
-      _(registered_views[0].aggregation.class).must_equal ::OpenTelemetry::SDK::Metrics::Aggregation::LastValue
+      _(registered_views.first[0].aggregation.class).must_equal ::OpenTelemetry::SDK::Metrics::Aggregation::LastValue
     end
   end
 

metrics_sdk/test/opentelemetry/sdk/metrics/view/registered_view_test.rb

@@ -181,7 +181,9 @@
       OpenTelemetry.meter_provider.add_view('async_counter', aggregation: ::OpenTelemetry::SDK::Metrics::Aggregation::LastValue.new)
 
       callback = proc { 25 }
-      meter.create_observable_counter('async_counter', unit: 'smidgen', description: 'an async counter', callback: callback)
+      asynch_counter = meter.create_observable_counter('async_counter', unit: 'smidgen', description: 'an async counter', callback: callback)
+      asynch_counter.observe
+      asynch_counter.observe
 
       metric_exporter.pull
       last_snapshot = metric_exporter.metric_snapshots
@@ -197,6 +199,9 @@
         _(snapshot.instrumentation_scope.name).must_equal('test')
         _(snapshot.data_points).wont_be_empty
       end
+
+      _(last_snapshot[0].data_points.first.value).must_equal 75 # view aggregation sum
+      _(last_snapshot[1].data_points.first.value).must_equal 25 # view aggregation last value
     end
 
     it 'emits asynchronous counter metrics with view attribute filtering' do