[SDK] Fix crash in PeriodicExportingMetricReader.

api/include/opentelemetry/common/macros.h

@@ -387,6 +387,37 @@ point.
 
 #endif
 
+// OPENTELEMETRY_HAVE_EXCEPTIONS
+//
+// Checks whether the compiler both supports and enables exceptions. Many
+// compilers support a "no exceptions" mode that disables exceptions.
+//
+// Generally, when OPENTELEMETRY_HAVE_EXCEPTIONS is not defined:
+//
+// * Code using `throw` and `try` may not compile.
+// * The `noexcept` specifier will still compile and behave as normal.
+// * The `noexcept` operator may still return `false`.
+//
+// For further details, consult the compiler's documentation.
+#ifndef OPENTELEMETRY_HAVE_EXCEPTIONS
+# if defined(__clang__) && ((__clang_major__ * 100) + __clang_minor__) < 306
+// Clang < 3.6
+// http://releases.llvm.org/3.6.0/tools/clang/docs/ReleaseNotes.html#the-exceptions-macro
+# if defined(__EXCEPTIONS) && OPENTELEMETRY_HAVE_FEATURE(cxx_exceptions)
+# define OPENTELEMETRY_HAVE_EXCEPTIONS 1
+# endif // defined(__EXCEPTIONS) && OPENTELEMETRY_HAVE_FEATURE(cxx_exceptions)
+# elif OPENTELEMETRY_HAVE_FEATURE(cxx_exceptions)
+# define OPENTELEMETRY_HAVE_EXCEPTIONS 1
+// Handle remaining special cases and default to exceptions being supported.
+# elif !(defined(__GNUC__) && !defined(__EXCEPTIONS) && !defined(__cpp_exceptions)) && \
+ !(defined(_MSC_VER) && !defined(_CPPUNWIND))
+# define OPENTELEMETRY_HAVE_EXCEPTIONS 1
+# endif
+#endif
+#ifndef OPENTELEMETRY_HAVE_EXCEPTIONS
+# define OPENTELEMETRY_HAVE_EXCEPTIONS 0
+#endif
+
 /*
  OPENTELEMETRY_ATTRIBUTE_LIFETIME_BOUND indicates that a resource owned by a function
  parameter or implicit object parameter is retained by the return value of the

examples/plugin/plugin/tracer.cc

@@ -9,6 +9,7 @@
 #include "opentelemetry/common/attribute_value.h"
 #include "opentelemetry/common/timestamp.h"
 #include "opentelemetry/context/context_value.h"
+#include "opentelemetry/nostd/utility.h"
 #include "opentelemetry/trace/span_context.h"
 #include "opentelemetry/trace/span_metadata.h"
 #include "tracer.h"

exporters/otlp/src/otlp_file_client.cc

@@ -14,8 +14,6 @@
 // clang-format on
 
 #include "google/protobuf/message.h"
-#include "google/protobuf/reflection.h"
-#include "google/protobuf/stubs/common.h"
 #include "nlohmann/json.hpp"
 
 // clang-format off
@@ -28,15 +26,26 @@
 #include "opentelemetry/sdk/common/global_log_handler.h"
 #include "opentelemetry/version.h"
 
+#ifdef _MSC_VER
+# include <string.h>
+# define strcasecmp _stricmp
+#else
+# include <strings.h>
+#endif
+
+#include <limits.h>
 #include <atomic>
 #include <condition_variable>
+#include <cstdint>
 #include <cstdio>
-#include <cstdlib>
 #include <cstring>
+#include <ctime>
 #include <fstream>
+#include <functional>
 #include <mutex>
 #include <string>
 #include <thread>
+#include <utility>
 #include <vector>
 
 #if !defined(__CYGWIN__) && defined(_WIN32)
@@ -64,11 +73,8 @@
 
 #else
 
-# include <dirent.h>
-# include <errno.h>
 # include <fcntl.h>
 # include <sys/stat.h>
-# include <sys/types.h>
 # include <unistd.h>
 
 # define FS_ACCESS(x) access(x, F_OK)
@@ -89,10 +95,6 @@
 # undef GetMessage
 #endif
 
-#ifdef _MSC_VER
-# define strcasecmp _stricmp
-#endif
-
 #if (defined(_MSC_VER) && _MSC_VER >= 1600) || \
  (defined(__STDC_VERSION__) && __STDC_VERSION__ >= 201112L) || defined(__STDC_LIB_EXT1__)
 # ifdef _MSC_VER

sdk/src/metrics/export/periodic_exporting_metric_reader.cc

@@ -13,6 +13,7 @@
 #include <thread>
 #include <utility>
 
+#include "opentelemetry/common/macros.h"
 #include "opentelemetry/common/timestamp.h"
 #include "opentelemetry/sdk/common/global_log_handler.h"
 #include "opentelemetry/sdk/metrics/export/metric_producer.h"
@@ -29,6 +30,10 @@
 # include <future>
 #endif
 
+#if OPENTELEMETRY_HAVE_EXCEPTIONS
+# include <exception>
+#endif
+
 OPENTELEMETRY_BEGIN_NAMESPACE
 namespace sdk
 {
@@ -90,31 +95,61 @@ 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::future_status status;
  std::uint64_t notify_force_flush = force_flush_pending_sequence_.load(std::memory_order_acquire);
- do
+ std::unique_ptr<std::thread> task_thread;
+
+#if OPENTELEMETRY_HAVE_EXCEPTIONS
+ try
  {
- status = future_receive.wait_for(std::chrono::milliseconds(export_timeout_millis_));
- if (status == std::future_status::timeout)
+#endif
+ std::promise<void> sender;
+ auto receiver = sender.get_future();
+
+ task_thread.reset(new std::thread([this, &cancel_export_for_timeout] {
+ this->Collect([this, &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: "
+ << this->export_timeout_millis_.count() << " ms, and timed out");
+ return false;
+ }
+ this->exporter_->Export(metric_data);
+ return true;
+ });
+ }));
+
+ std::future_status status;
+ do
  {
- cancel_export_for_timeout = true;
- break;
- }
- } while (status != std::future_status::ready);
+ status = receiver.wait_for(std::chrono::milliseconds(export_timeout_millis_));
+ if (status == std::future_status::timeout)
+ {
+ cancel_export_for_timeout.store(true, std::memory_order_release);
+ break;
+ }
+ } while (status != std::future_status::ready);
+#if OPENTELEMETRY_HAVE_EXCEPTIONS
+ }
+ catch (std::exception &e)
+ {
+ OTEL_INTERNAL_LOG_ERROR("[Periodic Exporting Metric Reader] Collect failed with exception "
+ << e.what());
+ return false;
+ }
+ catch (...)
+ {
+ OTEL_INTERNAL_LOG_ERROR(
+ "[Periodic Exporting Metric Reader] Collect failed with unknown exception");
+ return false;
+ }
+#endif
+
+ if (task_thread && task_thread->joinable())
+ {
+ task_thread->join();
+ }
 
  std::uint64_t notified_sequence = force_flush_notified_sequence_.load(std::memory_order_acquire);
  while (notify_force_flush > notified_sequence)

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();
+}