feat: add basic support for metrics exemplar

metrics_api/lib/opentelemetry/internal/proxy_instrument.rb

@@ -8,21 +8,23 @@ module OpenTelemetry
   module Internal
     # @api private
     class ProxyInstrument
-      def initialize(kind, name, unit, desc, callable)
+      def initialize(kind, name, unit, desc, callable, exemplar_filter, exemplar_reservoir)
         @kind = kind
         @name = name
         @unit = unit
         @desc = desc
         @callable = callable
+        @exemplar_filter    = exemplar_filter
+        @exemplar_reservoir = exemplar_reservoir
         @delegate = nil
       end
 
       def upgrade_with(meter)
         @delegate = case @kind
-                    when :counter, :histogram, :up_down_counter, :gauge
-                      meter.send("create_#{@kind}", @name, unit: @unit, description: @desc)
+                    when :counter, :histogram, :up_down_counter
+                      meter.send("create_#{@kind}", @name, unit: @unit, description: @desc, exemplar_filter: @exemplar_filter, exemplar_reservoir: @exemplar_reservoir)
                     when :observable_counter, :observable_gauge, :observable_up_down_counter
-                      meter.send("create_#{@kind}", @name, unit: @unit, description: @desc, callback: @callback)
+                      meter.send("create_#{@kind}", @name, unit: @unit, description: @desc, exemplar_filter: @exemplar_filter, exemplar_reservoir: @exemplar_reservoir, callback: @callback)
                     end
       end
 

metrics_api/lib/opentelemetry/internal/proxy_meter.rb

@@ -38,18 +38,18 @@ def delegate=(meter)
 
       private
 
-      def create_instrument(kind, name, unit, description, callback)
+      def create_instrument(kind, name, unit, description, callback, exemplar_filter, exemplar_reservoir)
         super do
-          next ProxyInstrument.new(kind, name, unit, description, callback) if @delegate.nil?
+          next ProxyInstrument.new(kind, name, unit, description, callback, exemplar_filter, exemplar_reservoir) if @delegate.nil?
 
           case kind
-          when :counter then @delegate.create_counter(name, unit: unit, description: description)
-          when :histogram then @delegate.create_histogram(name, unit: unit, description: description)
-          when :gauge then @delegate.create_gauge(name, unit: unit, description: description)
-          when :up_down_counter then @delegate.create_up_down_counter(name, unit: unit, description: description)
-          when :observable_counter then @delegate.create_observable_counter(name, unit: unit, description: description, callback: callback)
-          when :observable_gauge then @delegate.create_observable_gauge(name, unit: unit, description: description, callback: callback)
-          when :observable_up_down_counter then @delegate.create_observable_up_down_counter(name, unit: unit, description: description, callback: callback)
+          when :counter then @delegate.create_counter(name, unit: unit, description: description, exemplar_filter: exemplar_filter, exemplar_reservoir: exemplar_reservoir)
+          when :histogram then @delegate.create_histogram(name, unit: unit, description: description, exemplar_filter: exemplar_filter, exemplar_reservoir: exemplar_reservoir)
+          when :up_down_counter then @delegate.create_up_down_counter(name, unit: unit, description: description, exemplar_filter: exemplar_filter, exemplar_reservoir: exemplar_reservoir)
+          when :gauge then @delegate.create_gauge(name, unit: unit, description: description, exemplar_filter: exemplar_filter, exemplar_reservoir: exemplar_reservoir)
+          when :observable_counter then @delegate.create_observable_counter(name, unit: unit, description: description, exemplar_filter: exemplar_filter, exemplar_reservoir: exemplar_reservoir, callback: callback)
+          when :observable_gauge then @delegate.create_observable_gauge(name, unit: unit, description: description, exemplar_filter: exemplar_filter, exemplar_reservoir: exemplar_reservoir, callback: callback)
+          when :observable_up_down_counter then @delegate.create_observable_up_down_counter(name, unit: unit, description: description, exemplar_filter: exemplar_filter, exemplar_reservoir: exemplar_reservoir, callback: callback)
           end
         end
       end

metrics_api/lib/opentelemetry/metrics/meter.rb

@@ -43,8 +43,8 @@ def initialize
       # @param description [optional String] an optional free-form text provided by user.
       #
       # @return [nil] after creation of counter, it will be stored in instrument_registry
-      def create_counter(name, unit: nil, description: nil)
-        create_instrument(:counter, name, unit, description, nil) { COUNTER }
+      def create_counter(name, unit: nil, description: nil, exemplar_filter: nil, exemplar_reservoir: nil)
+        create_instrument(:counter, name, unit, description, nil, exemplar_filter, exemplar_reservoir) { COUNTER }
       end
 
       # Histogram is a synchronous Instrument which can be used to report arbitrary values that are likely
@@ -62,8 +62,8 @@ def create_counter(name, unit: nil, description: nil)
       # @param description [optional String] an optional free-form text provided by user.
       #
       # @return [nil] after creation of histogram, it will be stored in instrument_registry
-      def create_histogram(name, unit: nil, description: nil)
-        create_instrument(:histogram, name, unit, description, nil) { HISTOGRAM }
+      def create_histogram(name, unit: nil, description: nil, exemplar_filter: nil, exemplar_reservoir: nil)
+        create_instrument(:histogram, name, unit, description, nil, exemplar_filter, exemplar_reservoir) { HISTOGRAM }
       end
 
       # Gauge is an synchronous Instrument which reports non-additive value(s)
@@ -80,8 +80,8 @@ def create_histogram(name, unit: nil, description: nil)
       # @param description [optional String] an optional free-form text provided by user.
       #
       # @return [nil] after creation of gauge, it will be stored in instrument_registry
-      def create_gauge(name, unit: nil, description: nil)
-        create_instrument(:gauge, name, unit, description, nil) { GAUGE }
+      def create_gauge(name, unit: nil, description: nil, exemplar_filter: nil, exemplar_reservoir: nil)
+        create_instrument(:gauge, name, unit, description, nil, exemplar_filter, exemplar_reservoir) { GAUGE }
       end
 
       # UpDownCounter is a synchronous Instrument which supports increments and decrements.
@@ -97,8 +97,8 @@ def create_gauge(name, unit: nil, description: nil)
       # @param description [optional String] an optional free-form text provided by user.
       #
       # @return [nil] after creation of up_down_counter, it will be stored in instrument_registry
-      def create_up_down_counter(name, unit: nil, description: nil)
-        create_instrument(:up_down_counter, name, unit, description, nil) { UP_DOWN_COUNTER }
+      def create_up_down_counter(name, unit: nil, description: nil, exemplar_filter: nil, exemplar_reservoir: nil)
+        create_instrument(:up_down_counter, name, unit, description, nil, exemplar_filter, exemplar_reservoir) { UP_DOWN_COUNTER }
       end
 
       # ObservableCounter is an asynchronous Instrument which reports monotonically
@@ -119,8 +119,8 @@ def create_up_down_counter(name, unit: nil, description: nil)
       # @param description [optional String] an optional free-form text provided by user.
       #
       # @return [nil] after creation of observable_counter, it will be stored in instrument_registry
-      def create_observable_counter(name, callback:, unit: nil, description: nil)
-        create_instrument(:observable_counter, name, unit, description, callback) { OBSERVABLE_COUNTER }
+      def create_observable_counter(name, callback:, unit: nil, description: nil, exemplar_filter: nil, exemplar_reservoir: nil)
+        create_instrument(:observable_counter, name, unit, description, callback, exemplar_filter, exemplar_reservoir) { OBSERVABLE_COUNTER }
       end
 
       # ObservableGauge is an asynchronous Instrument which reports non-additive value(s)
@@ -142,8 +142,8 @@ def create_observable_counter(name, callback:, unit: nil, description: nil)
       # @param description [optional String] an optional free-form text provided by user.
       #
       # @return [nil] after creation of observable_gauge, it will be stored in instrument_registry
-      def create_observable_gauge(name, callback:, unit: nil, description: nil)
-        create_instrument(:observable_gauge, name, unit, description, callback) { OBSERVABLE_GAUGE }
+      def create_observable_gauge(name, callback:, unit: nil, description: nil, exemplar_filter: nil, exemplar_reservoir: nil)
+        create_instrument(:observable_gauge, name, unit, description, callback, exemplar_filter, exemplar_reservoir) { OBSERVABLE_GAUGE }
       end
 
       # ObservableUpDownCounter is an asynchronous Instrument which reports additive value(s)
@@ -165,13 +165,13 @@ def create_observable_gauge(name, callback:, unit: nil, description: nil)
       # @param description [optional String] an optional free-form text provided by user.
       #
       # @return [nil] after creation of observable_up_down_counter, it will be stored in instrument_registry
-      def create_observable_up_down_counter(name, callback:, unit: nil, description: nil)
-        create_instrument(:observable_up_down_counter, name, unit, description, callback) { OBSERVABLE_UP_DOWN_COUNTER }
+      def create_observable_up_down_counter(name, callback:, unit: nil, description: nil, exemplar_filter: nil, exemplar_reservoir: nil)
+        create_instrument(:observable_up_down_counter, name, unit, description, callback, exemplar_filter, exemplar_reservoir) { OBSERVABLE_UP_DOWN_COUNTER }
       end
 
       private
 
-      def create_instrument(kind, name, unit, description, callback)
+      def create_instrument(kind, name, unit, description, callback, exemplar_filter, exemplar_reservoir)
         @mutex.synchronize do
           OpenTelemetry.logger.warn("duplicate instrument registration occurred for instrument #{name}") if @instrument_registry.include? name
 

metrics_sdk/lib/opentelemetry/sdk/metrics.rb

@@ -13,6 +13,7 @@ module Metrics
   end
 end
 
+require 'opentelemetry/sdk/metrics/exemplar'
 require 'opentelemetry/sdk/metrics/aggregation'
 require 'opentelemetry/sdk/metrics/configurator_patch'
 require 'opentelemetry/sdk/metrics/export'

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

@@ -14,16 +14,22 @@ class ExplicitBucketHistogram
           DEFAULT_BOUNDARIES = [0, 5, 10, 25, 50, 75, 100, 250, 500, 1000].freeze
           private_constant :DEFAULT_BOUNDARIES
 
+          # if no reservior pass from instrument, then use this empty reservior to avoid no method found error
+          DEFAULT_RESERVOIR = Metrics::Exemplar::FixedSizeExemplarReservoir.new
+          private_constant :DEFAULT_RESERVOIR
+
           # 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: ENV.fetch('OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE', :cumulative),
             boundaries: DEFAULT_BOUNDARIES,
-            record_min_max: true
+            record_min_max: true,
+            exemplar_reservoir: DEFAULT_RESERVOIR
           )
             @aggregation_temporality = AggregationTemporality.determine_temporality(aggregation_temporality: aggregation_temporality, default: :cumulative)
+            @exemplar_reservoir = exemplar_reservoir
             @boundaries = boundaries && !boundaries.empty? ? boundaries.sort : nil
             @record_min_max = record_min_max
           end
@@ -65,7 +71,7 @@ def update(amount, attributes, data_points)
                 0,                   # :sum
                 empty_bucket_counts, # :bucket_counts
                 @boundaries,         # :explicit_bounds
-                nil,                 # :exemplars
+                @exemplar_reservoir.collect(attributes: attributes, aggregation_temporality: @aggregation_temporality), # :exemplars
                 min,                 # :min
                 max                  # :max
               )

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

@@ -3,17 +3,24 @@
 # 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: ENV.fetch('OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE', :cumulative), monotonic: false, instrument_kind: nil)
+          # if no reservior pass from instrument, then use this empty reservior to avoid no method found error
+          DEFAULT_RESERVOIR = Metrics::Exemplar::FixedSizeExemplarReservoir.new
+          private_constant :DEFAULT_RESERVOIR
+
+          def initialize(aggregation_temporality: ENV.fetch('OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE', :cumulative),
+                         monotonic: false,
+                         instrument_kind: nil,
+                         exemplar_reservoir: DEFAULT_RESERVOIR)
             @aggregation_temporality = AggregationTemporality.determine_temporality(aggregation_temporality: aggregation_temporality, instrument_kind: instrument_kind, default: :cumulative)
             @monotonic = monotonic
+            @exemplar_reservoir = exemplar_reservoir
           end
 
           def collect(start_time, end_time, data_points)
@@ -48,7 +55,8 @@ def update(increment, attributes, data_points)
               nil,
               nil,
               0,
-              nil
+              # will this cause the reservoir overloaded with old exemplars?
+              @exemplar_reservoir.collect(attributes: attributes, aggregation_temporality: @aggregation_temporality) # exemplar
             )
 
             ndp.value += increment

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

@@ -0,0 +1,26 @@
+# frozen_string_literal: true
+
+# Copyright The OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+module OpenTelemetry
+  module SDK
+    module Metrics
+      # The Exemplar module contains the OpenTelemetry metrics reference
+      # exemplar implementations.
+      module Exemplar
+      end
+    end
+  end
+end
+
+require 'opentelemetry/sdk/metrics/exemplar/exemplar'
+require 'opentelemetry/sdk/metrics/exemplar/exemplar_filter'
+require 'opentelemetry/sdk/metrics/exemplar/exemplar_reservoir'
+require 'opentelemetry/sdk/metrics/exemplar/always_off_exemplar_filter'
+require 'opentelemetry/sdk/metrics/exemplar/always_on_exemplar_filter'
+require 'opentelemetry/sdk/metrics/exemplar/trace_based_exemplar_filter'
+require 'opentelemetry/sdk/metrics/exemplar/noop_exemplar_reservoir'
+require 'opentelemetry/sdk/metrics/exemplar/fixed_size_exemplar_reservoir'
+require 'opentelemetry/sdk/metrics/exemplar/histogram_exemplar_reservoir'

metrics_sdk/lib/opentelemetry/sdk/metrics/exemplar/always_off_exemplar_filter.rb

@@ -0,0 +1,21 @@
+# frozen_string_literal: true
+
+# Copyright The OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+module OpenTelemetry
+  module SDK
+    module Metrics
+      module Exemplar
+        # AlwaysOffExemplarFilter makes no measurements eligible for being an Exemplar.
+        # Using this ExemplarFilter is as good as disabling Exemplar feature.
+        class AlwaysOffExemplarFilter < ExemplarFilter
+          def self.should_sample?(value, timestamp, attributes, context)
+            false
+          end
+        end
+      end
+    end
+  end
+end

metrics_sdk/lib/opentelemetry/sdk/metrics/exemplar/always_on_exemplar_filter.rb

@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+# Copyright The OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+module OpenTelemetry
+  module SDK
+    module Metrics
+      module Exemplar
+        # AlwaysOnExemplarFilter
+        class AlwaysOnExemplarFilter < ExemplarFilter
+          def self.should_sample?(value, timestamp, attributes, context)
+            true
+          end
+        end
+      end
+    end
+  end
+end

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

@@ -0,0 +1,26 @@
+# frozen_string_literal: true
+
+# Copyright The OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+module OpenTelemetry
+  module SDK
+    module Metrics
+      module Exemplar
+        # Exemplar
+        class Exemplar
+          attr_reader :value, :time_unix_nano, :attributes, :span_id, :trace_id
+
+          def initialize(value, time_unix_nano, attributes, span_id, trace_id)
+            @value = value
+            @time_unix_nano = time_unix_nano
+            @attributes = attributes
+            @span_id  = span_id
+            @trace_id = trace_id
+          end
+        end
+      end
+    end
+  end
+end

metrics_sdk/lib/opentelemetry/sdk/metrics/exemplar/exemplar_filter.rb

@@ -0,0 +1,25 @@
+# frozen_string_literal: true
+
+# Copyright The OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+module OpenTelemetry
+  module SDK
+    module Metrics
+      module Exemplar
+        # ExemplarFilter
+        class ExemplarFilter
+          # Returns a {Boolean} value.
+          #
+          # @param [Integer] value Value of the measurement
+          # @param [Hash] attributes Complete set of Attributes of the measurement
+          # @param [Context] context Context of the measurement, which covers the Baggage and the current active Span.
+          #
+          # @return [Boolean]
+          def self.should_sample?(value, attributes, context); end
+        end
+      end
+    end
+  end
+end

metrics_sdk/lib/opentelemetry/sdk/metrics/exemplar/exemplar_reservoir.rb

@@ -0,0 +1,51 @@
+# frozen_string_literal: true
+
+# Copyright The OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+module OpenTelemetry
+  module SDK
+    module Metrics
+      module Exemplar
+        # ExemplarReservoir
+        class ExemplarReservoir
+          def initialize
+            @exemplars = []
+          end
+
+          # Store the info into exemplars bucket
+          #
+          # @param [Integer] value Value of the measurement
+          # @param [Integer] timestamp Time of recording
+          # @param [Hash] attributes Complete set of Attributes of the measurement
+          # @param [Context] context SpanContext of the measurement, which covers the Baggage and the current active Span.
+          #
+          # @return [Nil]
+          def offer(value: nil, timestamp: nil, attributes: nil, context: nil)
+            span_context = current_span_context(context)
+            @exemplars << Exemplar.new(value, timestamp, attributes, span_context.hex_span_id, span_context.hex_trace_id)
+            nil
+          end
+
+          # return list of Exemplars based on given attributes
+          #
+          # @param [Hash] attributes Value of the measurement
+          # @param [Boolean] aggregation_temporality Should remove the original exemplars or not, default delta
+          #
+          # @return [Array] exemplars Array of exemplars
+          def collect(attributes: nil, aggregation_temporality: :delta)
+            exemplars = []
+            @exemplars.each { |exemplar| exemplars << exemplar if exemplar } # TODO: Addition operation on selecting exemplar
+            @exemplars.clear if aggregation_temporality == :delta
+            exemplars
+          end
+
+          def current_span_context(context)
+            ::OpenTelemetry::Trace.current_span(context).context
+          end
+        end
+      end
+    end
+  end
+end