feat: add basic implementation of asynchronous metrics

metrics_api/lib/opentelemetry/metrics/instrument/observable_counter.rb

@@ -9,7 +9,14 @@ module Metrics
     module Instrument
       # No-op implementation of ObservableCounter.
       class ObservableCounter
-        # TODO
+        # Observe the ObservableCounter with fixed timeout duration.
+        #
+        # @param [int] timeout The timeout duration for callback to run, which MUST be a non-negative numeric value.
+        # @param [Hash{String => String, Numeric, Boolean, Array<String, Numeric, Boolean>}] attributes
+        #   Values must be non-nil and (array of) string, boolean or numeric type.
+        #   Array values must not contain nil elements and all elements must be of
+        #   the same basic type (string, numeric, boolean).
+        def observe(timeout: nil, attributes: {}); end
       end
     end
   end

metrics_api/lib/opentelemetry/metrics/instrument/observable_gauge.rb

@@ -9,7 +9,14 @@ module Metrics
     module Instrument
       # No-op implementation of ObservableGauge.
       class ObservableGauge
-        # TODO
+        # Observe the ObservableGauge with fixed timeout duration.
+        #
+        # @param [int] timeout The timeout duration for callback to run, which MUST be a non-negative numeric value.
+        # @param [Hash{String => String, Numeric, Boolean, Array<String, Numeric, Boolean>}] attributes
+        #   Values must be non-nil and (array of) string, boolean or numeric type.
+        #   Array values must not contain nil elements and all elements must be of
+        #   the same basic type (string, numeric, boolean).
+        def observe(timeout: nil, attributes: {}); end
       end
     end
   end

metrics_api/lib/opentelemetry/metrics/instrument/observable_up_down_counter.rb

@@ -9,7 +9,14 @@ module Metrics
     module Instrument
       # No-op implementation of ObservableUpDownCounter.
       class ObservableUpDownCounter
-        # TODO
+        # Observe the ObservableUpDownCounter with fixed timeout duration.
+        #
+        # @param [int] timeout The timeout duration for callback to run, which MUST be a non-negative numeric value.
+        # @param [Hash{String => String, Numeric, Boolean, Array<String, Numeric, Boolean>}] attributes
+        #   Values must be non-nil and (array of) string, boolean or numeric type.
+        #   Array values must not contain nil elements and all elements must be of
+        #   the same basic type (string, numeric, boolean).
+        def observe(timeout: nil, attributes: {}); end
       end
     end
   end

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

@@ -14,6 +14,7 @@ module Instrument
 end
 
 require 'opentelemetry/sdk/metrics/instrument/synchronous_instrument'
+require 'opentelemetry/sdk/metrics/instrument/asynchronous_instrument'
 require 'opentelemetry/sdk/metrics/instrument/counter'
 require 'opentelemetry/sdk/metrics/instrument/histogram'
 require 'opentelemetry/sdk/metrics/instrument/observable_counter'

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

@@ -0,0 +1,93 @@
+# frozen_string_literal: true
+
+# Copyright The OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+module OpenTelemetry
+  module SDK
+    module Metrics
+      module Instrument
+        # {AsynchronousInstrument} contains the common functionality shared across
+        # the asynchronous instruments SDK instruments.
+        class AsynchronousInstrument
+          def initialize(name, unit, description, callback, instrumentation_scope, meter_provider)
+            @name = name
+            @unit = unit
+            @description = description
+            @instrumentation_scope = instrumentation_scope
+            @meter_provider = meter_provider
+            @metric_streams = []
+            @callbacks = []
+            @timeout   = nil
+            @attributes = {}
+
+            init_callback(callback)
+            meter_provider.register_asynchronous_instrument(self)
+          end
+
+          # @api private
+          def register_with_new_metric_store(metric_store, aggregation: default_aggregation)
+            ms = OpenTelemetry::SDK::Metrics::State::AsynchronousMetricStream.new(
+              @name,
+              @description,
+              @unit,
+              instrument_kind,
+              @meter_provider,
+              @instrumentation_scope,
+              aggregation,
+              @callbacks,
+              @timeout,
+              @attributes
+            )
+            @metric_streams << ms
+            metric_store.add_metric_stream(ms)
+          end
+
+          # The API MUST support creation of asynchronous instruments by passing zero or more callback functions
+          # to be permanently registered to the newly created instrument.
+          def init_callback(callback)
+            if callback.instance_of?(Proc)
+              @callbacks << callback
+            elsif callback.instance_of?(Array)
+              callback.each { |cb| @callbacks << cb if cb.instance_of?(Proc) }
+            else
+              OpenTelemetry.logger.warn "Only accept single Proc or Array of Proc for initialization with callback (given callback #{callback.class}"
+            end
+          end
+
+          # Where the API supports registration of callback functions after asynchronous instrumentation creation,
+          # the user MUST be able to undo registration of the specific callback after its registration by some means.
+          def register_callback(callback)
+            if callback.instance_of?(Proc)
+              @callbacks << callback
+              callback
+            else
+              OpenTelemetry.logger.warn "Only accept single Proc for registering callback (given callback #{callback.class}"
+            end
+          end
+
+          def unregister(callback)
+            @callbacks.delete(callback)
+          end
+
+          def timeout(timeout)
+            @timeout = timeout
+          end
+
+          def add_attributes(attributes)
+            @attributes.merge!(attributes) if attributes.instance_of?(Hash)
+          end
+
+          private
+
+          # update the observed value (after calling observe)
+          # invoke callback will execute callback and export metric_data that is observed
+          def update(timeout, attributes)
+            @metric_streams.each { |ms| ms.invoke_callback(timeout, attributes) }
+          end
+        end
+      end
+    end
+  end
+end

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

@@ -8,16 +8,31 @@ module OpenTelemetry
   module SDK
     module Metrics
       module Instrument
-        # {ObservableCounter} is the SDK implementation of {OpenTelemetry::Metrics::ObservableCounter}.
-        class ObservableCounter < OpenTelemetry::Metrics::Instrument::ObservableCounter
-          attr_reader :name, :unit, :description
+        # {ObservableCounter} is the SDK implementation of {OpenTelemetry::SDK::Metrics::Instrument::AsynchronousInstrument}.
+        # Asynchronous Counter is an asynchronous Instrument which reports monotonically increasing value(s) when the instrument is being observed.
+        class ObservableCounter < OpenTelemetry::SDK::Metrics::Instrument::AsynchronousInstrument
+          # Returns the instrument kind as a Symbol
+          #
+          # @return [Symbol]
+          def instrument_kind
+            :observable_counter
+          end
+
+          # Observe the Counter with fixed timeout duration.
+          #
+          # @param [int] timeout The timeout duration for callback to run, which MUST be a non-negative numeric value.
+          # @param [Hash{String => String, Numeric, Boolean, Array<String, Numeric, Boolean>}] attributes
+          #   Values must be non-nil and (array of) string, boolean or numeric type.
+          #   Array values must not contain nil elements and all elements must be of
+          #   the same basic type (string, numeric, boolean).
+          def observe(timeout: nil, attributes: {})
+            update(timeout, attributes)
+          end
+
+          private
 
-          def initialize(name, unit, description, callback, meter)
-            @name = name
-            @unit = unit
-            @description = description
-            @callback = callback
-            @meter = meter
+          def default_aggregation
+            OpenTelemetry::SDK::Metrics::Aggregation::Sum.new
           end
         end
       end

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

@@ -8,16 +8,31 @@ module OpenTelemetry
   module SDK
     module Metrics
       module Instrument
-        # {ObservableGauge} is the SDK implementation of {OpenTelemetry::Metrics::ObservableGauge}.
-        class ObservableGauge < OpenTelemetry::Metrics::Instrument::ObservableGauge
-          attr_reader :name, :unit, :description
+        # {ObservableGauge} is the SDK implementation of {OpenTelemetry::SDK::Metrics::Instrument::AsynchronousInstrument}.
+        # Asynchronous Gauge is an asynchronous Instrument which reports non-additive value(s) (e.g. the room temperature)
+        class ObservableGauge < OpenTelemetry::SDK::Metrics::Instrument::AsynchronousInstrument
+          # Returns the instrument kind as a Symbol
+          #
+          # @return [Symbol]
+          def instrument_kind
+            :observable_gauge
+          end
+
+          # Observe the Gauge with fixed timeout duration.
+          #
+          # @param [int] timeout The timeout duration for callback to run, which MUST be a non-negative numeric value.
+          # @param [Hash{String => String, Numeric, Boolean, Array<String, Numeric, Boolean>}] attributes
+          #   Values must be non-nil and (array of) string, boolean or numeric type.
+          #   Array values must not contain nil elements and all elements must be of
+          #   the same basic type (string, numeric, boolean).
+          def observe(timeout: nil, attributes: {})
+            update(timeout, attributes)
+          end
+
+          private
 
-          def initialize(name, unit, description, callback, meter)
-            @name = name
-            @unit = unit
-            @description = description
-            @callback = callback
-            @meter = meter
+          def default_aggregation
+            OpenTelemetry::SDK::Metrics::Aggregation::Sum.new
           end
         end
       end

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

@@ -8,16 +8,31 @@ module OpenTelemetry
   module SDK
     module Metrics
       module Instrument
-        # {ObservableUpDownCounter} is the SDK implementation of {OpenTelemetry::Metrics::ObservableUpDownCounter}.
-        class ObservableUpDownCounter < OpenTelemetry::Metrics::Instrument::ObservableUpDownCounter
-          attr_reader :name, :unit, :description
+        # {ObservableUpDownCounter} is the SDK implementation of {OpenTelemetry::SDK::Metrics::Instrument::AsynchronousInstrument}.
+        # Asynchronous UpDownCounter is an asynchronous Instrument which reports additive value(s) (e.g. the process heap size)
+        class ObservableUpDownCounter < OpenTelemetry::SDK::Metrics::Instrument::AsynchronousInstrument
+          # Returns the instrument kind as a Symbol
+          #
+          # @return [Symbol]
+          def instrument_kind
+            :observable_up_down_counter
+          end
+
+          # Observe the UpDownCounter with fixed timeout duration.
+          #
+          # @param [int] timeout The timeout duration for callback to run, which MUST be a non-negative numeric value.
+          # @param [Hash{String => String, Numeric, Boolean, Array<String, Numeric, Boolean>}] attributes
+          #   Values must be non-nil and (array of) string, boolean or numeric type.
+          #   Array values must not contain nil elements and all elements must be of
+          #   the same basic type (string, numeric, boolean).
+          def observe(timeout: nil, attributes: {})
+            update(timeout, attributes)
+          end
+
+          private
 
-          def initialize(name, unit, description, callback, meter)
-            @name = name
-            @unit = unit
-            @description = description
-            @callback = callback
-            @meter = meter
+          def default_aggregation
+            OpenTelemetry::SDK::Metrics::Aggregation::Sum.new(aggregation_temporality: :delta)
           end
         end
       end

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

@@ -28,6 +28,30 @@ def initialize(name, version, meter_provider)
           @meter_provider = meter_provider
         end
 
+        # Multiple-instrument callbacks
+        # Callbacks registered after the time of instrument creation MAY be associated with multiple instruments.
+        # Related spec: https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/api.md#multiple-instrument-callbacks
+        # Related spec: https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/api.md#synchronous-instrument-api
+        #
+        # @param [Array] instruments A list (or tuple, etc.) of Instruments used in the callback function.
+        # @param [Proc] callback A callback function
+        #
+        # It is RECOMMENDED that the API authors use one of the following forms for the callback function:
+        # The list (or tuple, etc.) returned by the callback function contains (Instrument, Measurement) pairs.
+        # the Observable Result parameter receives an additional (Instrument, Measurement) pairs
+        # Here it chose the second form
+        def register_callback(instruments, callback)
+          instruments.each do |instrument|
+            instrument.register_callback(callback)
+          end
+        end
+
+        def unregister(instruments, callback)
+          instruments.each do |instrument|
+            instrument.unregister(callback)
+          end
+        end
+
         # @api private
         def add_metric_reader(metric_reader)
           @instrument_registry.each_value do |instrument|

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

@@ -126,6 +126,7 @@ def register_synchronous_instrument(instrument)
             end
           end
         end
+        alias register_asynchronous_instrument register_synchronous_instrument
 
         # A View provides SDK users with the flexibility to customize the metrics that are output by the SDK.
         #

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

@@ -20,3 +20,4 @@ module State
 require 'opentelemetry/sdk/metrics/state/metric_data'
 require 'opentelemetry/sdk/metrics/state/metric_store'
 require 'opentelemetry/sdk/metrics/state/metric_stream'
+require 'opentelemetry/sdk/metrics/state/asynchronous_metric_stream'