[SDK] Fix crash in PeriodicExportingMetricReader.

sdk/include/opentelemetry/sdk/metrics/export/periodic_exporting_metric_reader.h

@@ -23,7 +23,9 @@ namespace sdk
 namespace metrics
 {
 
-class PeriodicExportingMetricReader : public MetricReader
+class PeriodicExportingMetricReader
+ : public MetricReader,
+ public std::enable_shared_from_this<PeriodicExportingMetricReader>
 {
 
 public:

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"

sdk/src/metrics/export/periodic_exporting_metric_reader.cc

@@ -89,19 +89,23 @@ 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: "
+ << keep_lifetime->export_timeout_millis_.count() << " ms, and timed out");
+ return false;
+ }
+ keep_lifetime->exporter_->Export(metric_data);
+ return true;
+ });
  });
 
  std::future_status status;
@@ -111,7 +115,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();
+}