[SDK] Fix crash in PeriodicExportingMetricReader.

sdk/include/opentelemetry/sdk/metrics/metric_reader.h

@@ -5,6 +5,7 @@
 
 #include <atomic>
 #include <chrono>
+#include <memory>
 
 #include "opentelemetry/nostd/function_ref.h"
 #include "opentelemetry/sdk/metrics/export/metric_producer.h"
@@ -19,7 +20,7 @@ namespace metrics
 /**
  * MetricReader defines the interface to collect metrics from SDK
  */
-class MetricReader
+class MetricReader : public std::enable_shared_from_this<MetricReader>
 {
 public:
   MetricReader();

sdk/src/metrics/export/periodic_exporting_metric_reader.cc

@@ -89,19 +89,26 @@ void PeriodicExportingMetricReader::DoBackgroundWork()
 
 bool PeriodicExportingMetricReader::CollectAndExportOnce()
 {
-  std::atomic<bool> cancel_export_for_timeout{false};
-  auto future_receive = std::async(std::launch::async, [this, &cancel_export_for_timeout] {
-    Collect([this, &cancel_export_for_timeout](ResourceMetrics &metric_data) {
-      if (cancel_export_for_timeout)
-      {
-        OTEL_INTERNAL_LOG_ERROR(
-            "[Periodic Exporting Metric Reader] Collect took longer configured time: "
-            << export_timeout_millis_.count() << " ms, and timed out");
-        return false;
-      }
-      this->exporter_->Export(metric_data);
-      return true;
-    });
+  std::shared_ptr<std::atomic<bool>> cancel_export_for_timeout =
+      std::make_shared<std::atomic<bool>>(false);
+  auto keep_lifetime = shared_from_this();
+
+  auto future_receive = std::async(std::launch::async, [keep_lifetime, cancel_export_for_timeout] {
+    keep_lifetime->Collect(
+        [keep_lifetime, cancel_export_for_timeout](ResourceMetrics &metric_data) {
+          if (cancel_export_for_timeout->load(std::memory_order_acquire))
+          {
+            OTEL_INTERNAL_LOG_ERROR(
+                "[Periodic Exporting Metric Reader] Collect took longer configured time: "
+                << static_cast<PeriodicExportingMetricReader *>(keep_lifetime.get())
+                       ->export_timeout_millis_.count()
+                << " ms, and timed out");
+            return false;
+          }
+          static_cast<PeriodicExportingMetricReader *>(keep_lifetime.get())
+              ->exporter_->Export(metric_data);
+          return true;
+        });
   });
 
   std::future_status status;
@@ -111,7 +118,7 @@ bool PeriodicExportingMetricReader::CollectAndExportOnce()
     status = future_receive.wait_for(std::chrono::milliseconds(export_timeout_millis_));
     if (status == std::future_status::timeout)
     {
-      cancel_export_for_timeout = true;
+      cancel_export_for_timeout->store(true, std::memory_order_release);
       break;
     }
   } while (status != std::future_status::ready);

sdk/test/metrics/periodic_exporting_metric_reader_test.cc

@@ -7,15 +7,25 @@
 
 #include <gtest/gtest.h>
 
+#include <chrono>
+#include <memory>
+#include <thread>
+
 using namespace opentelemetry;
 using namespace opentelemetry::sdk::instrumentationscope;
 using namespace opentelemetry::sdk::metrics;
 
 class MockPushMetricExporter : public PushMetricExporter
 {
 public:
+  MockPushMetricExporter(std::chrono::milliseconds wait) : wait_(wait) {}
+
   opentelemetry::sdk::common::ExportResult Export(const ResourceMetrics &record) noexcept override
   {
+    if (wait_ > std::chrono::milliseconds::zero())
+    {
+      std::this_thread::sleep_for(wait_);
+    }
     records_.push_back(record);
     return opentelemetry::sdk::common::ExportResult::kSuccess;
   }
@@ -34,6 +44,7 @@ class MockPushMetricExporter : public PushMetricExporter
 
 private:
   std::vector<ResourceMetrics> records_;
+  std::chrono::milliseconds wait_;
 };
 
 class MockMetricProducer : public MetricProducer
@@ -61,17 +72,34 @@ class MockMetricProducer : public MetricProducer
 
 TEST(PeriodicExporingMetricReader, BasicTests)
 {
-  std::unique_ptr<PushMetricExporter> exporter(new MockPushMetricExporter());
+  std::unique_ptr<PushMetricExporter> exporter(
+      new MockPushMetricExporter(std::chrono::milliseconds{0}));
   PeriodicExportingMetricReaderOptions options;
   options.export_timeout_millis  = std::chrono::milliseconds(200);
   options.export_interval_millis = std::chrono::milliseconds(500);
   auto exporter_ptr              = exporter.get();
-  PeriodicExportingMetricReader reader(std::move(exporter), options);
+  std::shared_ptr<PeriodicExportingMetricReader> reader =
+      std::make_shared<PeriodicExportingMetricReader>(std::move(exporter), options);
   MockMetricProducer producer;
-  reader.SetMetricProducer(&producer);
+  reader->SetMetricProducer(&producer);
   std::this_thread::sleep_for(std::chrono::milliseconds(2000));
-  EXPECT_NO_THROW(reader.ForceFlush());
-  reader.Shutdown();
+  EXPECT_NO_THROW(reader->ForceFlush());
+  reader->Shutdown();
   EXPECT_EQ(static_cast<MockPushMetricExporter *>(exporter_ptr)->GetDataCount(),
             static_cast<MockMetricProducer *>(&producer)->GetDataCount());
 }
+
+TEST(PeriodicExporingMetricReader, Timeout)
+{
+  std::unique_ptr<PushMetricExporter> exporter(
+      new MockPushMetricExporter(std::chrono::milliseconds{2000}));
+  PeriodicExportingMetricReaderOptions options;
+  options.export_timeout_millis  = std::chrono::milliseconds(200);
+  options.export_interval_millis = std::chrono::milliseconds(500);
+  std::shared_ptr<PeriodicExportingMetricReader> reader =
+      std::make_shared<PeriodicExportingMetricReader>(std::move(exporter), options);
+  MockMetricProducer producer;
+  reader->SetMetricProducer(&producer);
+  std::this_thread::sleep_for(std::chrono::milliseconds(1000));
+  reader->Shutdown();
+}