feat: Metrics SDK Synchronous aggregation (#1346)

* Feat: add sync aggregation

* More tests

* add more tests

* Chore: Apply Francis suggested changes

Co-authored-by: Francis Bogsanyi <[email protected]>

* fix: syntax error on review suggestion

* fix: assert histogram bounds are sorted

* fix: remove dead code

* chore: update comment for default boundaries

* fix: add counter instrument tests

* fix: Add histogram test

* fix: add up down counter test

* fix: remove shutdown check on test exporter

* fix: only set min max when enabled

* chore: appease the cop

* fix: remove unused mutex in sum agg

* fix: remove unused flags field

* fix: apply Francis code suggestions

Co-authored-by: Francis Bogsanyi <[email protected]>

* chore: appease the cop

* fix: dup data points on collection

* chore: appease the cop

* fix: histogram min max conditional assignment

* fix: Apply Francis feedback

Co-authored-by: Francis Bogsanyi <[email protected]>

* fix: handle nil boundaries on explicit histogram

* fix: empty boundaries to behave same as nil

* fix: Francis temporality refactoring

Co-authored-by: Francis Bogsanyi <[email protected]>

* fix: add temporality tests for sum aggregation

* fix: add histogram aggregation tests

* fix: rebase fixups

* fix: Apply Francis feedback

Co-authored-by: Francis Bogsanyi <[email protected]>

* fix: Set differing start end times

Co-authored-by: Francis Bogsanyi <[email protected]>
Co-authored-by: Andrew Hayworth <[email protected]>

Author: 3 people

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

@@ -4,19 +4,18 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-require 'opentelemetry/sdk/metrics/aggregation/histogram'
-
 module OpenTelemetry
   module SDK
     module Metrics
       # The Aggregation module contains the OpenTelemetry metrics reference
       # aggregation implementations.
       module Aggregation
-        extend self
-
-        SUM = ->(v1, v2) { v1 + v2 }
-        EXPLICIT_BUCKET_HISTOGRAM = ExplicitBucketHistogram.new
       end
     end
   end
 end
+
+require 'opentelemetry/sdk/metrics/aggregation/number_data_point'
+require 'opentelemetry/sdk/metrics/aggregation/histogram_data_point'
+require 'opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram'
+require 'opentelemetry/sdk/metrics/aggregation/sum'

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

@@ -0,0 +1,97 @@
+# 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 ExplicitBucketHistogram aggregation
+        # https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#explicit-bucket-histogram-aggregation
+        class ExplicitBucketHistogram
+          DEFAULT_BOUNDARIES = [0, 5, 10, 25, 50, 75, 100, 250, 500, 1000].freeze
+          private_constant :DEFAULT_BOUNDARIES
+
+          # The default value for boundaries represents the following buckets:
+          # (-inf, 0], (0, 5.0], (5.0, 10.0], (10.0, 25.0], (25.0, 50.0],
+          # (50.0, 75.0], (75.0, 100.0], (100.0, 250.0], (250.0, 500.0],
+          # (500.0, 1000.0], (1000.0, +inf)
+          def initialize(
+            aggregation_temporality: :delta,
+            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)
+            if @aggregation_temporality == :delta
+              # Set timestamps and 'move' data point values to result.
+              hdps = @data_points.values.each do |hdp|
+                hdp.start_time_unix_nano = start_time
+                hdp.time_unix_nano = end_time
+              end
+              @data_points.clear
+              hdps
+            else
+              # Update timestamps and take a snapshot.
+              @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
+                hdp.bucket_counts = hdp.bucket_counts.dup
+                hdp
+              end
+            end
+          end
+
+          def update(amount, attributes)
+            hdp = @data_points.fetch(attributes) do
+              if @record_min_max
+                min = Float::INFINITY
+                max = -Float::INFINITY
+              end
+
+              @data_points[attributes] = HistogramDataPoint.new(
+                attributes,
+                nil,                 # :start_time_unix_nano
+                nil,                 # :time_unix_nano
+                0,                   # :count
+                0,                   # :sum
+                empty_bucket_counts, # :bucket_counts
+                @boundaries,         # :explicit_bounds
+                nil,                 # :exemplars
+                min,                 # :min
+                max                  # :max
+              )
+            end
+
+            if @record_min_max
+              hdp.max = amount if amount > hdp.max
+              hdp.min = amount if amount < hdp.min
+            end
+
+            hdp.sum += amount
+            hdp.count += 1
+            if @boundaries
+              bucket_index = @boundaries.bsearch_index { _1 >= amount } || @boundaries.size
+              hdp.bucket_counts[bucket_index] += 1
+            end
+            nil
+          end
+
+          private
+
+          def empty_bucket_counts
+            @boundaries ? Array.new(@boundaries.size + 1, 0) : nil
+          end
+        end
+      end
+    end
+  end
+end

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

@@ -0,0 +1,24 @@
+# frozen_string_literal: true
+
+# Copyright The OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+module OpenTelemetry
+  module SDK
+    module Metrics
+      module Aggregation
+        HistogramDataPoint = Struct.new(:attributes,            # optional Hash{String => String, Numeric, Boolean, Array<String, Numeric, Boolean>}
+                                        :start_time_unix_nano,  # Integer nanoseconds since Epoch
+                                        :time_unix_nano,        # Integer nanoseconds since Epoch
+                                        :count,                 # Integer count is the number of values in the population. Must be non-negative.
+                                        :sum,                   # Integer sum of the values in the population. If count is zero then this field then this field must be zero
+                                        :bucket_counts,         # optional Array[Integer] field contains the count values of histogram for each bucket.
+                                        :explicit_bounds,       # Array[Float] specifies buckets with explicitly defined bounds for values.
+                                        :exemplars,             # optional List of exemplars collected from measurements that were used to form the data point
+                                        :min,                   # optional Float min is the minimum value over (start_time, end_time].
+                                        :max)                   # optional Float max is the maximum value over (start_time, end_time].
+      end
+    end
+  end
+end

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

@@ -0,0 +1,19 @@
+# frozen_string_literal: true
+
+# Copyright The OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+module OpenTelemetry
+  module SDK
+    module Metrics
+      module Aggregation
+        NumberDataPoint = Struct.new(:attributes,            # Hash{String => String, Numeric, Boolean, Array<String, Numeric, Boolean>}
+                                     :start_time_unix_nano,  # Integer nanoseconds since Epoch
+                                     :time_unix_nano,        # Integer nanoseconds since Epoch
+                                     :value,                 # Integer
+                                     :exemplars)             # optional List of exemplars collected from measurements that were used to form the data point
+      end
+    end
+  end
+end

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

@@ -0,0 +1,54 @@
+# 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 Sum aggregation
+        # https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#sum-aggregation
+        class Sum
+          def initialize(aggregation_temporality: :delta)
+            @aggregation_temporality = aggregation_temporality
+            @data_points = {}
+          end
+
+          def collect(start_time, end_time)
+            if @aggregation_temporality == :delta
+              # Set timestamps and 'move' data point values to result.
+              ndps = @data_points.values.each do |ndp|
+                ndp.start_time_unix_nano = start_time
+                ndp.time_unix_nano = end_time
+              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)
+            ndp = @data_points[attributes] || @data_points[attributes] = NumberDataPoint.new(
+              attributes,
+              nil,
+              nil,
+              0,
+              nil
+            )
+
+            ndp.value += increment
+            nil
+          end
+        end
+      end
+    end
+  end
+end

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

@@ -25,8 +25,6 @@ def pull
 
           def export(metrics)
             @mutex.synchronize do
-              return FAILURE if @stopped
-
               @metric_snapshots << metrics
             end
             SUCCESS

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

@@ -10,8 +10,6 @@ module Metrics
       module Instrument
         # {Counter} is the SDK implementation of {OpenTelemetry::Metrics::Counter}.
         class Counter < OpenTelemetry::SDK::Metrics::Instrument::SynchronousInstrument
-          DEFAULT_AGGREGATION = OpenTelemetry::SDK::Metrics::Aggregation::SUM
-
           # Returns the instrument kind as a Symbol
           #
           # @return [Symbol]
@@ -33,16 +31,19 @@ def add(increment, attributes: {})
             if increment.negative?
               OpenTelemetry.logger.warn("#{@name} received a negative value")
             else
-              update(
-                OpenTelemetry::Metrics::Measurement.new(increment, attributes),
-                DEFAULT_AGGREGATION
-              )
+              update(increment, attributes)
             end
             nil
           rescue StandardError => e
             OpenTelemetry.handle_error(exception: e)
             nil
           end
+
+          private
+
+          def default_aggregation
+            OpenTelemetry::SDK::Metrics::Aggregation::Sum.new
+          end
         end
       end
     end

metrics_sdk/lib/opentelemetry/sdk/metrics/instrument/counter.rb

@@ -10,8 +10,6 @@ module Metrics
       module Instrument
         # {Histogram} is the SDK implementation of {OpenTelemetry::Metrics::Histogram}.
         class Histogram < OpenTelemetry::SDK::Metrics::Instrument::SynchronousInstrument
-          DEFAULT_AGGREGATION = OpenTelemetry::SDK::Metrics::Aggregation::EXPLICIT_BUCKET_HISTOGRAM
-
           # Returns the instrument kind as a Symbol
           #
           # @return [Symbol]
@@ -27,12 +25,18 @@ def instrument_kind
           #   Array values must not contain nil elements and all elements must be of
           #   the same basic type (string, numeric, boolean).
           def record(amount, attributes: nil)
-            update(OpenTelemetry::Metrics::Measurement.new(amount, attributes))
+            update(amount, attributes)
             nil
           rescue StandardError => e
             OpenTelemetry.handle_error(exception: e)
             nil
           end
+
+          private
+
+          def default_aggregation
+            OpenTelemetry::SDK::Metrics::Aggregation::ExplicitBucketHistogram.new
+          end
         end
       end
     end

metrics_sdk/lib/opentelemetry/sdk/metrics/instrument/histogram.rb

@@ -19,31 +19,28 @@ def initialize(name, unit, description, instrumentation_scope, meter_provider)
             @meter_provider = meter_provider
             @metric_streams = []
 
-            meter_provider.metric_readers.each do |metric_reader|
-              register_with_new_metric_store(metric_reader.metric_store)
-            end
+            meter_provider.register_synchronous_instrument(self)
           end
 
           # @api private
-          def register_with_new_metric_store(metric_store)
+          def register_with_new_metric_store(metric_store, aggregation: default_aggregation)
             ms = OpenTelemetry::SDK::Metrics::State::MetricStream.new(
               @name,
               @description,
               @unit,
               instrument_kind,
               @meter_provider,
-              @instrumentation_scope
+              @instrumentation_scope,
+              aggregation
             )
             @metric_streams << ms
             metric_store.add_metric_stream(ms)
           end
 
           private
 
-          def update(measurement, aggregation)
-            @metric_streams.each do |ms|
-              ms.update(measurement, aggregation)
-            end
+          def update(value, attributes)
+            @metric_streams.each { |ms| ms.update(value, attributes) }
           end
         end
       end