[SDK] Fix MetricReader post shutdown handling

CHANGELOG.md

@@ -20,6 +20,14 @@ Increment the:
   `CURLE_SEND_FAIL_REWIND` and dropping the batch
   ([#4549](https://github.com/open-telemetry/opentelemetry-cpp/issues/4549))
 
+* [SDK] Fix `MetricReader::ForceFlush()` invoking `OnForceFlush()` on a
+  shutdown reader.
+  [#4548](https://github.com/open-telemetry/opentelemetry-cpp/pull/4548)
+
+* [SDK] Fix `MetricReader::Shutdown()` invoking `OnShutDown()` multiple times.
+  Concurrent calls now block until the first call's shutdown has completed.
+  [#4536](https://github.com/open-telemetry/opentelemetry-cpp/issues/4536)
+
 * [DOC] Fix and clarify the `StartSpanOptions` documentation
   [#4526](https://github.com/open-telemetry/opentelemetry-cpp/pull/4526)
 

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

@@ -6,6 +6,7 @@
 #include <atomic>
 #include <chrono>
 #include <cstddef>
+#include <mutex>
 
 #include "opentelemetry/nostd/function_ref.h"
 #include "opentelemetry/sdk/metrics/cardinality_limits.h"
@@ -71,11 +72,19 @@ class MetricReader
 
   /**
    * Shutdown the metric reader.
+   *
+   * Idempotent and a completion barrier: only the first call runs OnShutDown(), and a
+   * concurrent call blocks until the first call has finished before returning.
+   *
+   * @return the result of OnShutDown() for the first call, true for any subsequent call.
    */
   bool Shutdown(std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept;
 
   /**
    * Force flush the metric read by the reader.
+   *
+   * @return false without invoking OnForceFlush() if the reader is already shut down, otherwise
+   * the result of OnForceFlush().
    */
   bool ForceFlush(std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept;
 
@@ -96,6 +105,7 @@ class MetricReader
 protected:
 private:
   MetricProducer *metric_producer_{nullptr};
+  std::mutex shutdown_m_;
   std::atomic<bool> shutdown_{false};
   CardinalityLimits cardinality_limits_;
 };

sdk/src/metrics/metric_reader.cc

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 #include "opentelemetry/sdk/metrics/metric_reader.h"
+#include <mutex>
 #include "opentelemetry/sdk/common/global_log_handler.h"
 #include "opentelemetry/sdk/metrics/cardinality_limits.h"
 #include "opentelemetry/sdk/metrics/export/metric_producer.h"
@@ -48,13 +49,14 @@ bool MetricReader::Collect(
 
 bool MetricReader::Shutdown(std::chrono::microseconds timeout) noexcept
 {
-  bool status = true;
-  if (IsShutdown())
+  // Serialize so concurrent calls block until the first call's shutdown has completed.
+  std::lock_guard<std::mutex> shutdown_guard{shutdown_m_};
+  if (shutdown_.exchange(true, std::memory_order_release))
   {
-    OTEL_INTERNAL_LOG_WARN("MetricReader::Shutdown - Cannot invoke shutdown twice!");
+    OTEL_INTERNAL_LOG_WARN("MetricReader::Shutdown - Already shutdown!");
+    return true;
   }
-
-  shutdown_.store(true, std::memory_order_release);
+  bool status = true;
 
   if (!OnShutDown(timeout))
   {
@@ -67,11 +69,14 @@ bool MetricReader::Shutdown(std::chrono::microseconds timeout) noexcept
 /** Flush metric read by this reader **/
 bool MetricReader::ForceFlush(std::chrono::microseconds timeout) noexcept
 {
-  bool status = true;
   if (IsShutdown())
   {
-    OTEL_INTERNAL_LOG_WARN("MetricReader::Shutdown Cannot invoke Force flush on shutdown reader!");
+    OTEL_INTERNAL_LOG_WARN(
+        "MetricReader::ForceFlush Cannot invoke Force flush on shutdown reader!");
+    return false;
   }
+
+  bool status = true;
   if (!OnForceFlush(timeout))
   {
     status = false;

sdk/test/metrics/metric_reader_test.cc

@@ -2,11 +2,17 @@
 // SPDX-License-Identifier: Apache-2.0
 
 #include <gtest/gtest.h>
-#include <memory>
+#include <atomic>
+#include <chrono>
+#include <future>
 #include <string>
+#include <thread>
 #include <utility>
+#include <vector>
 #include "common.h"
 
+#include "opentelemetry/nostd/shared_ptr.h"
+#include "opentelemetry/sdk/common/global_log_handler.h"
 #include "opentelemetry/sdk/instrumentationscope/instrumentation_scope.h"
 #include "opentelemetry/sdk/metrics/cardinality_limits.h"
 #include "opentelemetry/sdk/metrics/export/metric_producer.h"
@@ -147,3 +153,184 @@ TEST(MetricReaderTest, CardinalityLimitsExplicitSdkDefaultIsHonoured)
   EXPECT_EQ(metric_reader->GetCardinalityLimit(InstrumentType::kHistogram), 1500);
   EXPECT_EQ(metric_reader->GetCardinalityLimit(InstrumentType::kUpDownCounter), 1500);
 }
+
+namespace
+{
+
+class CountingMetricReader : public MetricReader
+{
+public:
+  // hook_result is what both OnForceFlush() and OnShutDown() report back.
+  explicit CountingMetricReader(bool hook_result = true) : hook_result_(hook_result) {}
+
+  AggregationTemporality GetAggregationTemporality(InstrumentType) const noexcept override
+  {
+    return AggregationTemporality::kCumulative;
+  }
+
+  std::atomic<int> shutdown_count{0};
+  std::atomic<int> force_flush_count{0};
+
+private:
+  bool OnForceFlush(std::chrono::microseconds) noexcept override
+  {
+    ++force_flush_count;
+    return hook_result_;
+  }
+
+  bool OnShutDown(std::chrono::microseconds) noexcept override
+  {
+    ++shutdown_count;
+    return hook_result_;
+  }
+
+  const bool hook_result_;
+};
+
+}  // namespace
+
+TEST(MetricReaderTest, ShutdownIsInvokedOnce)
+{
+  CountingMetricReader reader;
+
+  EXPECT_TRUE(reader.Shutdown());
+  EXPECT_TRUE(reader.IsShutdown());
+  EXPECT_EQ(reader.shutdown_count.load(), 1);
+
+  EXPECT_TRUE(reader.Shutdown());
+  EXPECT_TRUE(reader.Shutdown());
+  EXPECT_EQ(reader.shutdown_count.load(), 1);
+}
+
+TEST(MetricReaderTest, ConcurrentShutdownIsInvokedOnce)
+{
+  namespace internal_log = opentelemetry::sdk::common::internal_log;
+  CountingMetricReader reader;
+
+  // default logger is not thread-safe
+  auto previous_handler = internal_log::GlobalLogHandler::GetLogHandler();
+  internal_log::GlobalLogHandler::SetLogHandler(
+      nostd::shared_ptr<internal_log::LogHandler>(new internal_log::NoopLogHandler()));
+
+  std::vector<std::thread> threads;
+  threads.reserve(8);
+  for (int i = 0; i < 8; i++)
+  {
+    threads.emplace_back([&reader]() { reader.Shutdown(); });
+  }
+  for (auto &thread : threads)
+  {
+    thread.join();
+  }
+
+  internal_log::GlobalLogHandler::SetLogHandler(previous_handler);
+
+  EXPECT_EQ(reader.shutdown_count.load(), 1);
+}
+
+TEST(MetricReaderTest, ForceFlushAfterShutdownIsNoOp)
+{
+  CountingMetricReader reader;
+
+  EXPECT_TRUE(reader.ForceFlush());
+  EXPECT_EQ(reader.force_flush_count.load(), 1);
+
+  EXPECT_TRUE(reader.Shutdown());
+
+  EXPECT_FALSE(reader.ForceFlush());
+  EXPECT_EQ(reader.force_flush_count.load(), 1);
+}
+
+TEST(MetricReaderTest, FailedShutdownIsReportedAndNotRetried)
+{
+  CountingMetricReader reader{/* hook_result= */ false};
+
+  EXPECT_FALSE(reader.ForceFlush());
+  EXPECT_EQ(reader.force_flush_count.load(), 1);
+
+  // The first call reports the hook's failure.
+  EXPECT_FALSE(reader.Shutdown());
+  EXPECT_TRUE(reader.IsShutdown());
+  EXPECT_EQ(reader.shutdown_count.load(), 1);
+
+  // A later call is a no-op that succeeds, without re-entering the failed hook.
+  EXPECT_TRUE(reader.Shutdown());
+  EXPECT_EQ(reader.shutdown_count.load(), 1);
+
+  // Flush stays rejected even after a failed shutdown, without re-entering the hook.
+  EXPECT_FALSE(reader.ForceFlush());
+  EXPECT_EQ(reader.force_flush_count.load(), 1);
+}
+
+namespace
+{
+
+// Parks inside OnShutDown() until released, to observe what a concurrent caller sees.
+class BlockingMetricReader : public MetricReader
+{
+public:
+  AggregationTemporality GetAggregationTemporality(InstrumentType) const noexcept override
+  {
+    return AggregationTemporality::kCumulative;
+  }
+
+  std::promise<void> entered_shutdown;
+  std::promise<void> release_shutdown;
+  std::atomic<bool> shutdown_finished{false};
+
+private:
+  bool OnForceFlush(std::chrono::microseconds) noexcept override { return true; }
+
+  bool OnShutDown(std::chrono::microseconds) noexcept override
+  {
+    entered_shutdown.set_value();
+    release_shutdown.get_future().wait();
+    shutdown_finished.store(true, std::memory_order_release);
+    return true;
+  }
+};
+
+}  // namespace
+
+TEST(MetricReaderTest, ConcurrentShutdownWaitsForCleanupToComplete)
+{
+  namespace internal_log = opentelemetry::sdk::common::internal_log;
+  BlockingMetricReader reader;
+
+  // default logger is not thread-safe
+  auto previous_handler = internal_log::GlobalLogHandler::GetLogHandler();
+  internal_log::GlobalLogHandler::SetLogHandler(
+      nostd::shared_ptr<internal_log::LogHandler>(new internal_log::NoopLogHandler()));
+
+  auto entered = reader.entered_shutdown.get_future();
+  std::thread first([&reader]() { EXPECT_TRUE(reader.Shutdown()); });
+
+  // The first caller owns the shutdown and is now parked inside OnShutDown().
+  entered.wait();
+  EXPECT_TRUE(reader.IsShutdown());
+  EXPECT_FALSE(reader.shutdown_finished.load(std::memory_order_acquire));
+
+  std::atomic<bool> second_returned{false};
+  std::promise<void> second_started;
+  auto started = second_started.get_future();
+  std::thread second([&]() {
+    second_started.set_value();
+    // Block until first caller releases the shutdown, then return true.
+    EXPECT_TRUE(reader.Shutdown());
+    EXPECT_TRUE(reader.shutdown_finished.load(std::memory_order_acquire));
+    second_returned.store(true, std::memory_order_release);
+  });
+
+  started.wait();
+  // Arbitrary sleep to ensure second is still blocked and not just we were too fast to check.
+  std::this_thread::sleep_for(std::chrono::milliseconds(50));
+  EXPECT_FALSE(second_returned.load(std::memory_order_acquire));
+
+  reader.release_shutdown.set_value();
+  second.join();
+  first.join();
+
+  internal_log::GlobalLogHandler::SetLogHandler(previous_handler);
+
+  EXPECT_TRUE(second_returned.load(std::memory_order_acquire));
+}