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, timeout: nil)
             @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,30 @@ 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
+        # A View provides SDK users with the flexibility to customize the metrics that are output by the SDK.
+        #
+        # Example:
+        #
+        #   OpenTelemetry.meter_provider.add_view('test', :aggregation => Aggregation::Drop.new,
+        #                                         :type => :counter, :unit => 'smidgen',
+        #                                         :meter_name => 'test', :meter_version => '1.0')
+        #
+        #
+        # @param [String] name Name of the view.
+        # @param [optional Hash] options For more precise matching, {View} and {MetricsStream}
+        #   options may include:
+        #     aggregation: An instance of an aggregation class, e.g. {ExplicitBucketHistogram}, {Sum}, {LastValue}
+        #     type: A Symbol representing the instrument kind, e.g. :observable_gauge, :counter
+        #     unit: A String matching an instrumentation unit, e.g. 'smidgen'
+        #     meter_name: A String matching a meter name, e.g. meter_provider.meter('sample_meter_name', version: '1.2.0'), would be 'sample_meter_name'
+        #     meter_version: A String matching a meter version, e.g. meter_provider.meter('sample_meter_name', version: '1.2.0'), would be '1.2.0'
+        #
+        # @return [nil] returns nil
+        #
+        def add_view(name, **options)
+          # TODO: add schema_url as part of options
+          @registered_views << View::RegisteredView.new(name, **options)
+          nil
         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

@@ -30,30 +30,59 @@ def initialize(
             @instrument_kind = instrument_kind
             @meter_provider = meter_provider
             @instrumentation_scope = instrumentation_scope
-            @aggregation = aggregation
+            @default_aggregation = aggregation
+            @data_points = {}
+            @registered_views = []
 
+            find_registered_view
             @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 = []
+            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) }
+            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) if view.valid_aggregation?
+                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
+            return if @meter_provider.nil?
+
+            @meter_provider.registered_views.each { |view| @registered_views << view if view.match_instrument?(self) }
           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
+      # A View provides SDK users with the flexibility to customize the metrics that are output by the SDK.
+      module View
+      end
+    end
+  end
+end
+
+require 'opentelemetry/sdk/metrics/view/registered_view'