feat: add basic metrics view

exporter/otlp-metrics/test/test_helper.rb

@@ -17,7 +17,7 @@
 OpenTelemetry.logger = Logger.new(File::NULL)
 
 module MockSum
-  def collect(start_time, end_time)
+  def collect(start_time, end_time, data_points)
     start_time = 1_699_593_427_329_946_585 # rubocop:disable Lint/ShadowedArgument
     end_time   = 1_699_593_427_329_946_586 # rubocop:disable Lint/ShadowedArgument
     super

metrics_sdk/lib/opentelemetry/sdk/metrics.rb

@@ -20,3 +20,4 @@ module Metrics
 require 'opentelemetry/sdk/metrics/meter'
 require 'opentelemetry/sdk/metrics/meter_provider'
 require 'opentelemetry/sdk/metrics/state'
+require 'opentelemetry/sdk/metrics/view'

metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation.rb

@@ -19,3 +19,5 @@ module Aggregation
 require 'opentelemetry/sdk/metrics/aggregation/histogram_data_point'
 require 'opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram'
 require 'opentelemetry/sdk/metrics/aggregation/sum'
+require 'opentelemetry/sdk/metrics/aggregation/last_value'
+require 'opentelemetry/sdk/metrics/aggregation/drop'

metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/drop.rb

@@ -0,0 +1,37 @@
+# frozen_string_literal: true
+
+# Copyright The OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+module OpenTelemetry
+  module SDK
+    module Metrics
+      module Aggregation
+        # Contains the implementation of the Drop aggregation
+        class Drop
+          attr_reader :aggregation_temporality
+
+          def initialize(aggregation_temporality: :delta)
+            @aggregation_temporality = aggregation_temporality
+          end
+
+          def collect(start_time, end_time, data_points)
+            data_points.values.map!(&:dup)
+          end
+
+          def update(increment, attributes, data_points)
+            data_points[attributes] = NumberDataPoint.new(
+              {},
+              0,
+              0,
+              0,
+              0
+            )
+            nil
+          end
+        end
+      end
+    end
+  end
+end

metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram.rb

@@ -25,25 +25,24 @@ def initialize(
             boundaries: DEFAULT_BOUNDARIES,
             record_min_max: true
           )
-            @data_points = {}
             @aggregation_temporality = aggregation_temporality
             @boundaries = boundaries && !boundaries.empty? ? boundaries.sort : nil
             @record_min_max = record_min_max
           end
 
-          def collect(start_time, end_time)
+          def collect(start_time, end_time, data_points)
             if @aggregation_temporality == :delta
               # Set timestamps and 'move' data point values to result.
-              hdps = @data_points.values.map! do |hdp|
+              hdps = data_points.values.map! do |hdp|
                 hdp.start_time_unix_nano = start_time
                 hdp.time_unix_nano = end_time
                 hdp
               end
-              @data_points.clear
+              data_points.clear
               hdps
             else
               # Update timestamps and take a snapshot.
-              @data_points.values.map! do |hdp|
+              data_points.values.map! do |hdp|
                 hdp.start_time_unix_nano ||= start_time # Start time of a data point is from the first observation.
                 hdp.time_unix_nano = end_time
                 hdp = hdp.dup
@@ -53,14 +52,14 @@ def collect(start_time, end_time)
             end
           end
 
-          def update(amount, attributes)
-            hdp = @data_points.fetch(attributes) do
+          def update(amount, attributes, data_points)
+            hdp = data_points.fetch(attributes) do
               if @record_min_max
                 min = Float::INFINITY
                 max = -Float::INFINITY
               end
 
-              @data_points[attributes] = HistogramDataPoint.new(
+              data_points[attributes] = HistogramDataPoint.new(
                 attributes,
                 nil,                 # :start_time_unix_nano
                 nil,                 # :time_unix_nano

metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/last_value.rb

@@ -0,0 +1,53 @@
+# frozen_string_literal: true
+
+# Copyright The OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+module OpenTelemetry
+  module SDK
+    module Metrics
+      module Aggregation
+        # Contains the implementation of the LastValue aggregation
+        class LastValue
+          attr_reader :aggregation_temporality
+
+          def initialize(aggregation_temporality: :delta)
+            @aggregation_temporality = aggregation_temporality
+          end
+
+          def collect(start_time, end_time, data_points)
+            if @aggregation_temporality == :delta
+              # Set timestamps and 'move' data point values to result.
+              ndps = data_points.values.map! do |ndp|
+                ndp.start_time_unix_nano = start_time
+                ndp.time_unix_nano = end_time
+                ndp
+              end
+              data_points.clear
+              ndps
+            else
+              # Update timestamps and take a snapshot.
+              data_points.values.map! do |ndp|
+                ndp.start_time_unix_nano ||= start_time # Start time of a data point is from the first observation.
+                ndp.time_unix_nano = end_time
+                ndp.dup
+              end
+            end
+          end
+
+          def update(increment, attributes, data_points)
+            data_points[attributes] = NumberDataPoint.new(
+              attributes,
+              nil,
+              nil,
+              increment,
+              nil
+            )
+            nil
+          end
+        end
+      end
+    end
+  end
+end

metrics_sdk/lib/opentelemetry/sdk/metrics/aggregation/sum.rb

@@ -16,31 +16,30 @@ class Sum
           def initialize(aggregation_temporality: ENV.fetch('OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE', :delta))
             # TODO: the default should be :cumulative, see issue #1555
             @aggregation_temporality = aggregation_temporality
-            @data_points = {}
           end
 
-          def collect(start_time, end_time)
+          def collect(start_time, end_time, data_points)
             if @aggregation_temporality == :delta
               # Set timestamps and 'move' data point values to result.
-              ndps = @data_points.values.map! do |ndp|
+              ndps = data_points.values.map! do |ndp|
                 ndp.start_time_unix_nano = start_time
                 ndp.time_unix_nano = end_time
                 ndp
               end
-              @data_points.clear
+              data_points.clear
               ndps
             else
               # Update timestamps and take a snapshot.
-              @data_points.values.map! do |ndp|
+              data_points.values.map! do |ndp|
                 ndp.start_time_unix_nano ||= start_time # Start time of a data point is from the first observation.
                 ndp.time_unix_nano = end_time
                 ndp.dup
               end
             end
           end
 
-          def update(increment, attributes)
-            ndp = @data_points[attributes] || @data_points[attributes] = NumberDataPoint.new(
+          def update(increment, attributes, data_points)
+            ndp = data_points[attributes] || data_points[attributes] = NumberDataPoint.new(
               attributes,
               nil,
               nil,

metrics_sdk/lib/opentelemetry/sdk/metrics/export/in_memory_metric_pull_exporter.rb

@@ -25,7 +25,7 @@ def pull
 
           def export(metrics)
             @mutex.synchronize do
-              @metric_snapshots << metrics
+              metrics.instance_of?(Array) ? @metric_snapshots.concat(metrics) : @metric_snapshots << metrics
             end
             SUCCESS
           end

metrics_sdk/lib/opentelemetry/sdk/metrics/meter_provider.rb

@@ -14,14 +14,15 @@ class MeterProvider < OpenTelemetry::Metrics::MeterProvider
         Key = Struct.new(:name, :version)
         private_constant(:Key)
 
-        attr_reader :resource, :metric_readers
+        attr_reader :resource, :metric_readers, :registered_views
 
         def initialize(resource: OpenTelemetry::SDK::Resources::Resource.create)
           @mutex = Mutex.new
           @meter_registry = {}
           @stopped = false
           @metric_readers = []
           @resource = resource
+          @registered_views = []
         end
 
         # Returns a {Meter} instance.
@@ -125,13 +126,12 @@ def register_synchronous_instrument(instrument)
           end
         end
 
-        # The type of the Instrument(s) (optional).
         # The name of the Instrument(s). OpenTelemetry SDK authors MAY choose to support wildcard characters, with the question mark (?) matching exactly one character and the asterisk character (*) matching zero or more characters.
-        # The name of the Meter (optional).
-        # The version of the Meter (optional).
-        # The schema_url of the Meter (optional).
-        def add_view
-          # TODO: For each meter add this view to all applicable instruments
+        # The options of the Instrument(s). Useful key include: aggregation, type, unit, meter_name, meter_version, attribute_keys
+        #
+        # TODO: add schema_url as part of options
+        def add_view(name, **options)
+          @registered_views << View::RegisteredView.new(name, **options)
         end
       end
     end

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

@@ -25,7 +25,7 @@ def collect
               @epoch_end_time = now_in_nano
               snapshot = @metric_streams.map { |ms| ms.collect(@epoch_start_time, @epoch_end_time) }
               @epoch_start_time = @epoch_end_time
-              snapshot
+              snapshot.flatten!
             end
           end
 

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

@@ -28,32 +28,62 @@ def initialize(
             @description = description
             @unit = unit
             @instrument_kind = instrument_kind
-            @meter_provider = meter_provider
+            @meter_provider  = meter_provider
             @instrumentation_scope = instrumentation_scope
-            @aggregation = aggregation
+            @default_aggregation = aggregation
+            @data_points = {}
 
             @mutex = Mutex.new
           end
 
           def collect(start_time, end_time)
-            @mutex.synchronize do
-              MetricData.new(
-                @name,
-                @description,
-                @unit,
-                @instrument_kind,
-                @meter_provider.resource,
-                @instrumentation_scope,
-                @aggregation.collect(start_time, end_time),
-                @aggregation.aggregation_temporality,
-                start_time,
-                end_time
-              )
+            metric_data = []
+            registered_views = find_registered_view
+
+            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) }
             end
+
+            metric_data
           end
 
           def update(value, attributes)
-            @mutex.synchronize { @aggregation.update(value, attributes) }
+            registered_views = find_registered_view
+            if registered_views.empty?
+              @mutex.synchronize { @default_aggregation.update(value, attributes, @data_points) }
+            else
+              registered_views.each do |view|
+                @mutex.synchronize do
+                  attributes ||= {}
+                  attributes.merge!(view.attribute_keys)
+                  view.aggregation.update(value, attributes, @data_points)
+                end
+              end
+            end
+          end
+
+          def aggregate_metric_data(start_time, end_time, aggregation: nil)
+            aggregator = aggregation || @default_aggregation
+            MetricData.new(
+              @name,
+              @description,
+              @unit,
+              @instrument_kind,
+              @meter_provider.resource,
+              @instrumentation_scope,
+              aggregator.collect(start_time, end_time, @data_points),
+              aggregator.aggregation_temporality,
+              start_time,
+              end_time
+            )
+          end
+
+          def find_registered_view
+            registered_views = []
+            @meter_provider.registered_views.each { |view| registered_views << view if view.match_instrument(self) }
+            registered_views
           end
 
           def to_s

metrics_sdk/lib/opentelemetry/sdk/metrics/view.rb

@@ -0,0 +1,17 @@
+# frozen_string_literal: true
+
+# Copyright The OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+module OpenTelemetry
+  module SDK
+    module Metrics
+      # View
+      module View
+      end
+    end
+  end
+end
+
+require 'opentelemetry/sdk/metrics/view/registered_view'

metrics_sdk/lib/opentelemetry/sdk/metrics/view/registered_view.rb

@@ -0,0 +1,36 @@
+# frozen_string_literal: true
+
+# Copyright The OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+module OpenTelemetry
+  module SDK
+    module Metrics
+      module View
+        # RegisteredView
+        class RegisteredView
+          attr_reader :name, :aggregation, :attribute_keys
+
+          def initialize(name, **options)
+            @name = name
+            @options = options
+            @aggregation = options[:aggregation]
+            @attribute_keys = options[:attribute_keys] || {}
+          end
+
+          def match_instrument(metric_stream)
+            return false if @aggregation.nil?
+            return false if @name && @name != metric_stream.name
+            return false if @options[:type] && @options[:type] != metric_stream.instrument_kind
+            return false if @options[:unit] && @options[:unit] != metric_stream.unit
+            return false if @options[:meter_name] && @options[:meter_name] != metric_stream.instrumentation_scope.name
+            return false if @options[:meter_version] && @options[:meter_version] != metric_stream.instrumentation_scope.version
+
+            true
+          end
+        end
+      end
+    end
+  end
+end

metrics_sdk/test/integration/in_memory_metric_pull_exporter_test.rb

@@ -26,7 +26,7 @@
       counter.add(4, attributes: { 'd' => 'e' })
 
       metric_exporter.pull
-      last_snapshot = metric_exporter.metric_snapshots.last
+      last_snapshot = metric_exporter.metric_snapshots
 
       _(last_snapshot).wont_be_empty
       _(last_snapshot[0].name).must_equal('counter')