fix: use condition signal to replace sleep and remove timeout.timeout… (#1826)

* fix: use condition signal to replace sleep and remove timeout.timeout (use exporter timeout function - http.timeout)

* lint

* add doc

---------

Co-authored-by: Kayla Reopelle <[email protected]>

Author: xuan-cao-swi

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

@@ -33,14 +33,22 @@ def initialize(export_interval_millis: Float(ENV.fetch('OTEL_METRIC_EXPORT_INTER
             @thread   = nil
             @continue = false
             @mutex = Mutex.new
+            @condition = ConditionVariable.new
             @export_mutex = Mutex.new
 
             start
           end
 
+          # Shuts the @thread down and set @continue to false; it will block
+          # until the shutdown thread is finished.
+          #
+          # @param [optional Numeric] timeout An optional timeout in seconds.
+          # @return [Integer] SUCCESS if no error occurred, FAILURE if a
+          #   non-specific failure occurred.
           def shutdown(timeout: nil)
             thread = lock do
               @continue = false # force termination in next iteration
+              @condition.signal
               @thread
             end
             thread&.join(@export_interval)
@@ -52,15 +60,35 @@ def shutdown(timeout: nil)
             Export::FAILURE
           end
 
+          # Export all metrics to the configured `Exporter` that have not yet
+          # been exported.
+          #
+          # This method should only be called in cases where it is absolutely
+          # necessary, such as when using some FaaS providers that may suspend
+          # the process after an invocation, but before the `PeriodicMetricReader` exports
+          # the completed metrics.
+          #
+          # @param [optional Numeric] timeout An optional timeout in seconds.
+          # @return [Integer] SUCCESS if no error occurred, FAILURE if a
+          #   non-specific failure occurred.
           def force_flush(timeout: nil)
-            export(timeout: timeout)
+            export(timeout:)
             Export::SUCCESS
           rescue StandardError
             Export::FAILURE
           end
 
+          # Check both @thread and @continue object to determine if current
+          # PeriodicMetricReader is still alive. If one of them is true/alive,
+          # then PeriodicMetricReader is determined as alive
+          def alive?
+            @continue || @thread.alive?
+          end
+
           private
 
+          # Start a thread that continously export metrics within fixed duration.
+          # The wait mechanism is using to check @mutex lock with conditional variable
           def start
             @continue = true
             if @exporter.nil?
@@ -70,19 +98,21 @@ def start
             else
               @thread = Thread.new do
                 while @continue
-                  sleep(@export_interval)
-                  begin
-                    Timeout.timeout(@export_timeout) do
-                      export(timeout: @export_timeout)
-                    end
-                  rescue Timeout::Error => e
-                    OpenTelemetry.handle_error(exception: e, message: 'PeriodicMetricReader timeout.')
+                  lock do
+                    @condition.wait(@mutex, @export_interval)
+                    export(timeout: @export_timeout)
                   end
                 end
               end
             end
           end
 
+          # Helper function for the defined exporter to export metrics.
+          # It only exports if the collected metrics are not an empty array (collect returns an Array).
+          #
+          # @param [optional Numeric] timeout An optional timeout in seconds.
+          # @return [Integer] SUCCESS if no error occurred, FAILURE if a
+          #   non-specific failure occurred
           def export(timeout: nil)
             @export_mutex.synchronize do
               collected_metrics = collect

metrics_sdk/test/integration/periodic_metric_reader_test.rb

@@ -81,5 +81,24 @@
       _(snapshot.size).must_equal(1)
       _(periodic_metric_reader.instance_variable_get(:@thread).alive?).must_equal false
     end
+
+    it 'shutdown break the export interval cycle' do
+      OpenTelemetry::SDK.configure
+
+      metric_exporter = OpenTelemetry::SDK::Metrics::Export::InMemoryMetricPullExporter.new
+      periodic_metric_reader = OpenTelemetry::SDK::Metrics::Export::PeriodicMetricReader.new(export_interval_millis: 1_000_000, export_timeout_millis: 10_000, exporter: metric_exporter)
+
+      OpenTelemetry.meter_provider.add_metric_reader(periodic_metric_reader)
+
+      _(periodic_metric_reader.alive?).must_equal true
+
+      sleep 5 # make sure the work thread start
+
+      Timeout.timeout(2) do # Fail if this block takes more than 2 seconds
+        periodic_metric_reader.shutdown
+      end
+
+      _(periodic_metric_reader.alive?).must_equal false
+    end
   end
 end

metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/drop_test.rb

@@ -8,7 +8,7 @@
 
 describe OpenTelemetry::SDK::Metrics::Aggregation::LastValue do
   let(:data_points) { {} }
-  let(:drop_aggregation) { OpenTelemetry::SDK::Metrics::Aggregation::Drop.new(aggregation_temporality: aggregation_temporality) }
+  let(:drop_aggregation) { OpenTelemetry::SDK::Metrics::Aggregation::Drop.new(aggregation_temporality:) }
   let(:aggregation_temporality) { :delta }
 
   # Time in nano

metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/explicit_bucket_histogram_test.rb

@@ -10,9 +10,9 @@
   let(:data_points) { {} }
   let(:ebh) do
     OpenTelemetry::SDK::Metrics::Aggregation::ExplicitBucketHistogram.new(
-      aggregation_temporality: aggregation_temporality,
-      boundaries: boundaries,
-      record_min_max: record_min_max
+      aggregation_temporality:,
+      boundaries:,
+      record_min_max:
     )
   end
   let(:boundaries) { [0, 5, 10, 25, 50, 75, 100, 250, 500, 1000] }

metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/last_value_test.rb

@@ -8,7 +8,7 @@
 
 describe OpenTelemetry::SDK::Metrics::Aggregation::LastValue do
   let(:data_points) { {} }
-  let(:last_value_aggregation) { OpenTelemetry::SDK::Metrics::Aggregation::LastValue.new(aggregation_temporality: aggregation_temporality) }
+  let(:last_value_aggregation) { OpenTelemetry::SDK::Metrics::Aggregation::LastValue.new(aggregation_temporality:) }
   let(:aggregation_temporality) { :delta }
 
   # Time in nano

metrics_sdk/test/opentelemetry/sdk/metrics/aggregation/sum_test.rb

@@ -8,7 +8,7 @@
 
 describe OpenTelemetry::SDK::Metrics::Aggregation::Sum do
   let(:data_points) { {} }
-  let(:sum_aggregation) { OpenTelemetry::SDK::Metrics::Aggregation::Sum.new(aggregation_temporality: aggregation_temporality, monotonic: monotonic) }
+  let(:sum_aggregation) { OpenTelemetry::SDK::Metrics::Aggregation::Sum.new(aggregation_temporality:, monotonic:) }
   let(:aggregation_temporality) { :delta }
   let(:monotonic) { false }