fix: recover PeriodicMetricReader in forked processes

Author: chrisholmes

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

@@ -25,6 +25,7 @@ def initialize
         def metrics_configuration_hook
           OpenTelemetry.meter_provider = Metrics::MeterProvider.new(resource: @resource)
           configure_metric_readers
+          attach_fork_hooks!
         end
 
         def configure_metric_readers
@@ -52,6 +53,10 @@ def wrapped_metric_exporters_from_env
             end
           end
         end
+
+        def attach_fork_hooks!
+          ForkHooks.attach!
+        end
       end
     end
   end

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

@@ -59,6 +59,13 @@ def force_flush(timeout: nil)
             Export::FAILURE
           end
 
+          def after_fork
+            @exporter.reset if @exporter.respond_to?(:reset)
+            collect # move past previously reported metrics from parent process
+            @thread = nil
+            start
+          end
+
           private
 
           def start

metrics_sdk/test/integration/periodic_metric_reader_test.rb

@@ -5,6 +5,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 require 'test_helper'
+require 'json'
 
 describe OpenTelemetry::SDK do
   describe '#periodic_metric_reader' do
@@ -81,5 +82,43 @@
       _(snapshot.size).must_equal(1)
       _(periodic_metric_reader.instance_variable_get(:@thread).alive?).must_equal false
     end
+
+    it 'is restarted after forking' do
+      OpenTelemetry::SDK.configure
+
+      metric_exporter = OpenTelemetry::SDK::Metrics::Export::InMemoryMetricPullExporter.new
+      periodic_metric_reader = OpenTelemetry::SDK::Metrics::Export::PeriodicMetricReader.new(export_interval_millis: 5000, export_timeout_millis: 5000, exporter: metric_exporter)
+
+      OpenTelemetry.meter_provider.add_metric_reader(periodic_metric_reader)
+
+      read, write = IO.pipe
+
+      pid = fork do
+        meter = OpenTelemetry.meter_provider.meter('test')
+        counter = meter.create_counter('counter', unit: 'smidgen', description: 'a small amount of something')
+
+        counter.add(1)
+        counter.add(2, attributes: { 'a' => 'b' })
+        counter.add(2, attributes: { 'a' => 'b' })
+        counter.add(3, attributes: { 'b' => 'c' })
+        counter.add(4, attributes: { 'd' => 'e' })
+
+        sleep(8)
+        snapshot = metric_exporter.metric_snapshots
+
+        json = snapshot.map { |reading| { name: reading.name } }.to_json
+        write.puts json
+      end
+
+      Timeout.timeout(10) do
+        Process.waitpid(pid)
+      end
+
+      periodic_metric_reader.shutdown
+      snapshot = JSON.parse(read.gets.chomp)
+      _(snapshot.size).must_equal(1)
+      _(snapshot[0]).must_equal('name' => 'counter')
+      _(periodic_metric_reader.instance_variable_get(:@thread).alive?).must_equal false
+    end
   end
 end

metrics_sdk/test/test_helper.rb

@@ -20,7 +20,7 @@ def reset_metrics_sdk
     :@meter_provider,
     OpenTelemetry::Internal::ProxyMeterProvider.new
   )
-
+  OpenTelemetry::SDK::Metrics::ForkHooks.instance_variable_set(:@fork_hooks_attached, false)
   OpenTelemetry.logger = Logger.new(File::NULL)
   OpenTelemetry.error_handler = nil
 end