feat: add view for asych instrument (#1887)

xuan-cao-swi · kaylareopelle · web-flow · commit 8ba003bea8cc · 2025-08-13T13:25:02.000-07:00
* feat: view for asych instrument

* refactor

* Apply suggestions from code review

Co-authored-by: Kayla Reopelle &lt;87386821+kaylareopelle@users.noreply.github.com&gt;

---------

Co-authored-by: Kayla Reopelle &lt;87386821+kaylareopelle@users.noreply.github.com&gt;
diff --git a/metrics_sdk/lib/opentelemetry/sdk/metrics/state/asynchronous_metric_stream.rb b/metrics_sdk/lib/opentelemetry/sdk/metrics/state/asynchronous_metric_stream.rb
@@ -10,11 +10,9 @@ module Metrics
       module State
         # @api private
         #
-        # The MetricStream class provides SDK internal functionality that is not a part of the
-        # public API.
-        class AsynchronousMetricStream
-          attr_reader :name, :description, :unit, :instrument_kind, :instrumentation_scope, :data_points
-
+        # The AsynchronousMetricStream class provides SDK internal functionality that is not a part of the
+        # public API. It extends MetricStream to support asynchronous instruments.
+        class AsynchronousMetricStream < MetricStream
           def initialize(
             name,
             description,
@@ -27,20 +25,14 @@ def initialize(
             timeout,
             attributes
           )
-            @name = name
-            @description = description
-            @unit = unit
-            @instrument_kind = instrument_kind
-            @meter_provider = meter_provider
-            @instrumentation_scope = instrumentation_scope
-            @aggregation = aggregation
+            # Call parent constructor with common parameters
+            super(name, description, unit, instrument_kind, meter_provider, instrumentation_scope, aggregation)
+
+            # Initialize asynchronous-specific attributes
             @callback = callback
             @start_time = now_in_nano
             @timeout = timeout
             @attributes = attributes
-            @data_points = {}
-
-            @mutex = Mutex.new
           end
 
           # When collect, if there are asynchronous SDK Instruments involved, their callback functions will be triggered.
@@ -49,47 +41,36 @@ def initialize(
           def collect(start_time, end_time)
             invoke_callback(@timeout, @attributes)
 
-            @mutex.synchronize do
-              MetricData.new(
-                @name,
-                @description,
-                @unit,
-                @instrument_kind,
-                @meter_provider.resource,
-                @instrumentation_scope,
-                @aggregation.collect(start_time, end_time, @data_points),
-                @aggregation.aggregation_temporality,
-                start_time,
-                end_time
-              )
-            end
+            # Call parent collect method for the core collection logic
+            super(start_time, end_time)
           end
 
           def invoke_callback(timeout, attributes)
-            @mutex.synchronize do
-              Timeout.timeout(timeout || 30) do
-                @callback.each do |cb|
-                  value = cb.call
-                  @aggregation.update(value, attributes, @data_points)
+            if @registered_views.empty?
+              @mutex.synchronize do
+                Timeout.timeout(timeout || 30) do
+                  @callback.each do |cb|
+                    value = cb.call
+                    @default_aggregation.update(value, attributes, @data_points)
+                  end
+                end
+              end
+            else
+              @registered_views.each do |view|
+                @mutex.synchronize do
+                  Timeout.timeout(timeout || 30) do
+                    @callback.each do |cb|
+                      value = cb.call
+                      merged_attributes = attributes || {}
+                      merged_attributes.merge!(view.attribute_keys)
+                      view.aggregation.update(value, merged_attributes, @data_points) if view.valid_aggregation?
+                    end
+                  end
                 end
               end
             end
           end
 
-          def to_s
-            instrument_info = +''
-            instrument_info << "name=#{@name}"
-            instrument_info << " description=#{@description}" if @description
-            instrument_info << " unit=#{@unit}" if @unit
-            @data_points.map do |attributes, value|
-              metric_stream_string = +''
-              metric_stream_string << instrument_info
-              metric_stream_string << " attributes=#{attributes}" if attributes
-              metric_stream_string << " #{value}"
-              metric_stream_string
-            end.join("\n")
-          end
-
           def now_in_nano
             (Time.now.to_r * 1_000_000_000).to_i
           end
diff --git a/metrics_sdk/test/opentelemetry/sdk/metrics/instrument/observable_counter_test.rb b/metrics_sdk/test/opentelemetry/sdk/metrics/instrument/observable_counter_test.rb
@@ -23,7 +23,6 @@
     metric_exporter.pull
     last_snapshot = metric_exporter.metric_snapshots
 
-    # puts "last_snapshot.inspect: #{last_snapshot.inspect}"
     _(last_snapshot[0].name).must_equal('counter')
     _(last_snapshot[0].unit).must_equal('smidgen')
     _(last_snapshot[0].description).must_equal('a small amount of something')
@@ -42,7 +41,6 @@
     metric_exporter.pull
     last_snapshot = metric_exporter.metric_snapshots
 
-    # puts "last_snapshot.inspect: #{last_snapshot.inspect}"
     _(last_snapshot[0].name).must_equal('counter')
     _(last_snapshot[0].unit).must_equal('smidgen')
     _(last_snapshot[0].description).must_equal('a small amount of something')
@@ -103,11 +101,7 @@
     metric_exporter.pull
     last_snapshot = metric_exporter.metric_snapshots
 
-    _(last_snapshot[0].name).must_equal('counter')
-    _(last_snapshot[0].unit).must_equal('smidgen')
-    _(last_snapshot[0].description).must_equal('a small amount of something')
-    _(last_snapshot[0].instrumentation_scope.name).must_equal('test')
-    _(last_snapshot[0].data_points.size).must_equal 0
+    _(last_snapshot.size).must_equal 0
   end
 
   it 'creation of instruments with more than one callabck' do
@@ -134,7 +128,7 @@
 
     metric_exporter.pull
     last_snapshot = metric_exporter.metric_snapshots
-    _(last_snapshot[0].data_points.size).must_equal 0
+    _(last_snapshot.size).must_equal 0
   end
 
   it 'creation of instruments with invalid argument result no callback' do
@@ -144,6 +138,6 @@
 
     metric_exporter.pull
     last_snapshot = metric_exporter.metric_snapshots
-    _(last_snapshot[0].data_points.size).must_equal 0
+    _(last_snapshot.size).must_equal 0
   end
 end
diff --git a/metrics_sdk/test/opentelemetry/sdk/metrics/view/registered_view_test.rb b/metrics_sdk/test/opentelemetry/sdk/metrics/view/registered_view_test.rb
@@ -90,6 +90,149 @@
     end
   end
 
+  describe '#registered_view with asynchronous counters' do
+    before { reset_metrics_sdk }
+
+    it 'emits asynchronous counter metrics with no data_points if view is drop' do
+      OpenTelemetry::SDK.configure
+
+      metric_exporter = OpenTelemetry::SDK::Metrics::Export::InMemoryMetricPullExporter.new
+      OpenTelemetry.meter_provider.add_metric_reader(metric_exporter)
+
+      meter = OpenTelemetry.meter_provider.meter('test')
+      OpenTelemetry.meter_provider.add_view('async_counter', aggregation: ::OpenTelemetry::SDK::Metrics::Aggregation::Drop.new)
+
+      callback = proc { 42 }
+      meter.create_observable_counter('async_counter', unit: 'smidgen', description: 'an async counter', callback: callback)
+
+      metric_exporter.pull
+      last_snapshot = metric_exporter.metric_snapshots
+
+      _(last_snapshot).wont_be_empty
+      _(last_snapshot[0].name).must_equal('async_counter')
+      _(last_snapshot[0].unit).must_equal('smidgen')
+      _(last_snapshot[0].description).must_equal('an async counter')
+      _(last_snapshot[0].instrumentation_scope.name).must_equal('test')
+
+      _(last_snapshot[0].data_points[0].value).must_equal 0
+      _(last_snapshot[0].data_points[0].start_time_unix_nano).must_equal 0
+      _(last_snapshot[0].data_points[0].time_unix_nano).must_equal 0
+    end
+
+    it 'emits asynchronous counter metrics with only last value in data_points if view is last_value' do
+      OpenTelemetry::SDK.configure
+
+      metric_exporter = OpenTelemetry::SDK::Metrics::Export::InMemoryMetricPullExporter.new
+      OpenTelemetry.meter_provider.add_metric_reader(metric_exporter)
+
+      meter = OpenTelemetry.meter_provider.meter('test')
+      OpenTelemetry.meter_provider.add_view('async_counter', aggregation: ::OpenTelemetry::SDK::Metrics::Aggregation::LastValue.new)
+
+      # Create a callback that returns different values each time it's called
+      call_count = 0
+      callback = proc do
+        call_count += 1
+        final_count = call_count * 10
+        final_count
+      end
+
+      meter.create_observable_counter('async_counter', unit: 'smidgen', description: 'an async counter', callback: callback)
+
+      # Trigger multiple collections to simulate multiple callback invocations
+      3.times { metric_exporter.pull }
+      last_snapshot = metric_exporter.metric_snapshots
+
+      # Reason that use 3rd from last_snapshot, because in_memory_metrics_pull exporter
+      # will store each collected metrics into its own data store unit (special case for the type of exporter)
+      _(last_snapshot[2].data_points).wont_be_empty
+      _(last_snapshot[2].data_points[0].value).must_equal 30
+    end
+
+    it 'emits asynchronous counter metrics with sum of values if view is drop but not matching to instrument' do
+      OpenTelemetry::SDK.configure
+
+      metric_exporter = OpenTelemetry::SDK::Metrics::Export::InMemoryMetricPullExporter.new
+      OpenTelemetry.meter_provider.add_metric_reader(metric_exporter)
+
+      meter = OpenTelemetry.meter_provider.meter('test')
+      # View name doesn't match the instrument name
+      OpenTelemetry.meter_provider.add_view('retnuoc_cnysa', aggregation: ::OpenTelemetry::SDK::Metrics::Aggregation::Drop.new)
+
+      callback = proc { 15 }
+      meter.create_observable_counter('async_counter', unit: 'smidgen', description: 'an async counter', callback: callback)
+
+      metric_exporter.pull
+      last_snapshot = metric_exporter.metric_snapshots
+
+      _(last_snapshot[0].data_points).wont_be_empty
+      # Since view doesn't match, it should use default aggregation (sum for counters)
+      _(last_snapshot[0].data_points[0].value).must_equal 15
+    end
+
+    it 'emits asynchronous counter metrics with multiple registered views' do
+      OpenTelemetry::SDK.configure
+
+      metric_exporter = OpenTelemetry::SDK::Metrics::Export::InMemoryMetricPullExporter.new
+      OpenTelemetry.meter_provider.add_metric_reader(metric_exporter)
+
+      meter = OpenTelemetry.meter_provider.meter('test')
+      # Add multiple views for the same instrument
+      OpenTelemetry.meter_provider.add_view('async_counter', aggregation: ::OpenTelemetry::SDK::Metrics::Aggregation::Sum.new)
+      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)
+
+      metric_exporter.pull
+      last_snapshot = metric_exporter.metric_snapshots
+
+      # Should have multiple metric data entries (one for each view)
+      _(last_snapshot.size).must_be :>=, 2
+
+      # All should have the same instrument metadata
+      last_snapshot.each do |snapshot|
+        _(snapshot.name).must_equal('async_counter')
+        _(snapshot.unit).must_equal('smidgen')
+        _(snapshot.description).must_equal('an async counter')
+        _(snapshot.instrumentation_scope.name).must_equal('test')
+        _(snapshot.data_points).wont_be_empty
+      end
+    end
+
+    it 'emits asynchronous counter metrics with view attribute filtering' do
+      OpenTelemetry::SDK.configure
+
+      metric_exporter = OpenTelemetry::SDK::Metrics::Export::InMemoryMetricPullExporter.new
+      OpenTelemetry.meter_provider.add_metric_reader(metric_exporter)
+
+      meter = OpenTelemetry.meter_provider.meter('test')
+
+      # Create a view that adds specific attributes
+      view_with_attributes = OpenTelemetry::SDK::Metrics::View::RegisteredView.new(
+        'async_counter',
+        aggregation: ::OpenTelemetry::SDK::Metrics::Aggregation::Sum.new,
+        attribute_keys: { 'environment' => 'test', 'service' => 'metrics' }
+      )
+      OpenTelemetry.meter_provider.instance_variable_get(:@registered_views) << view_with_attributes
+
+      callback = proc { 35 }
+      observable_counter = meter.create_observable_counter('async_counter', unit: 'smidgen', description: 'an async counter', callback: callback)
+      observable_counter.add_attributes({ 'original' => 'value' })
+
+      metric_exporter.pull
+      last_snapshot = metric_exporter.metric_snapshots
+
+      _(last_snapshot[0].data_points).wont_be_empty
+      _(last_snapshot[0].data_points[0].value).must_equal 35
+
+      # Check that view attributes are merged with original attributes
+      attributes = last_snapshot[0].data_points[0].attributes
+      _(attributes['environment']).must_equal 'test'
+      _(attributes['service']).must_equal 'metrics'
+      _(attributes['original']).must_equal 'value'
+    end
+  end
+
   describe '#registered_view select instrument' do
     let(:registered_view) { OpenTelemetry::SDK::Metrics::View::RegisteredView.new(nil, aggregation: ::OpenTelemetry::SDK::Metrics::Aggregation::LastValue.new) }
     let(:instrumentation_scope) do
