Fix #1663 Threading issue between Meter::RegisterSyncMetricStorage and Meter::Collect

sdk/include/opentelemetry/sdk/metrics/meter.h

@@ -120,6 +120,7 @@ class Meter final : public opentelemetry::metrics::Meter
       InstrumentDescriptor &instrument_descriptor);
   std::unique_ptr<AsyncWritableMetricStorage> RegisterAsyncMetricStorage(
       InstrumentDescriptor &instrument_descriptor);
+  opentelemetry::common::SpinLockMutex storage_lock_;
 };
 }  // namespace metrics
 }  // namespace sdk

sdk/src/metrics/meter.cc

@@ -208,6 +208,7 @@ const sdk::instrumentationscope::InstrumentationScope *Meter::GetInstrumentation
 std::unique_ptr<SyncWritableMetricStorage> Meter::RegisterSyncMetricStorage(
     InstrumentDescriptor &instrument_descriptor)
 {
+  std::lock_guard<opentelemetry::common::SpinLockMutex> guard(storage_lock_);
   auto ctx = meter_context_.lock();
   if (!ctx)
   {
@@ -251,6 +252,7 @@ std::unique_ptr<SyncWritableMetricStorage> Meter::RegisterSyncMetricStorage(
 std::unique_ptr<AsyncWritableMetricStorage> Meter::RegisterAsyncMetricStorage(
     InstrumentDescriptor &instrument_descriptor)
 {
+  std::lock_guard<opentelemetry::common::SpinLockMutex> guard(storage_lock_);
   auto ctx = meter_context_.lock();
   if (!ctx)
   {
@@ -293,6 +295,7 @@ std::unique_ptr<AsyncWritableMetricStorage> Meter::RegisterAsyncMetricStorage(
 std::vector<MetricData> Meter::Collect(CollectorHandle *collector,
                                        opentelemetry::common::SystemTimestamp collect_ts) noexcept
 {
+  std::lock_guard<opentelemetry::common::SpinLockMutex> guard(storage_lock_);
   observable_registry_->Observe(collect_ts);
   std::vector<MetricData> metric_data_list;
   auto ctx = meter_context_.lock();

sdk/test/metrics/meter_test.cc

@@ -30,14 +30,15 @@ class MockMetricReader : public MetricReader
 
 namespace
 {
-nostd::shared_ptr<metrics::Meter> InitMeter(MetricReader **metricReaderPtr)
+nostd::shared_ptr<metrics::Meter> InitMeter(MetricReader **metricReaderPtr,
+                                            std::string meter_name = "meter_name")
 {
   static std::shared_ptr<metrics::MeterProvider> provider(new MeterProvider());
   std::unique_ptr<MetricReader> metric_reader(new MockMetricReader());
   *metricReaderPtr = metric_reader.get();
   auto p           = std::static_pointer_cast<MeterProvider>(provider);
   p->AddMetricReader(std::move(metric_reader));
-  auto meter = provider->GetMeter("meter_name");
+  auto meter = provider->GetMeter(meter_name);
   return meter;
 }
 }  // namespace
@@ -70,6 +71,55 @@ TEST(MeterTest, BasicAsyncTests)
     }
     return true;
   });
+  observable_counter->RemoveCallback(asyc_generate_measurements, nullptr);
+}
+
+constexpr static unsigned MAX_THREADS       = 25;
+constexpr static unsigned MAX_ITERATIONS_MT = 1000;
+
+TEST(MeterTest, StressMultiThread)
+{
+  MetricReader *metric_reader_ptr = nullptr;
+  auto meter                      = InitMeter(&metric_reader_ptr, "stress_test_meter");
+  std::atomic<unsigned> threadCount(0);
+  size_t numIterations = MAX_ITERATIONS_MT;
+  std::atomic<bool> do_collect{false}, do_sync_create{true}, do_async_create{false};
+  std::vector<nostd::shared_ptr<opentelemetry::metrics::ObservableInstrument>>
+      observable_instruments;
+  while (numIterations--)
+  {
+    for (size_t i = 0; i < MAX_THREADS; i++)
+    {
+      if (threadCount++ < MAX_THREADS)
+      {
+        auto t = std::thread([&]() {
+          std::this_thread::yield();
+          if (do_sync_create.exchange(false))
+          {
+            std::string instrument_name = "test_couter_" + std::to_string(numIterations);
+            meter->CreateLongCounter(instrument_name, "", "");
+            do_async_create.store(true);
+          }
+          if (do_async_create.exchange(false))
+          {
+            auto observable_instrument =
+                meter->CreateLongObservableGauge("test_gauge" + std::to_string(numIterations));
+            observable_instrument->AddCallback(asyc_generate_measurements, nullptr);
+            observable_instruments.push_back(std::move(observable_instrument));
+            do_collect.store(true);
+          }
+          if (do_collect.exchange(false))
+          {
+            metric_reader_ptr->Collect([](ResourceMetrics &metric_data) { return true; });
+            do_sync_create.store(true);
+          }
+        });
+        t.detach();
+      }
+    }
+  }
+  // random wait for all callbacks to complete
+  std::this_thread::sleep_for(std::chrono::milliseconds(1000));
 }
 
 #endif