diff --git a/CHANGELOG.md b/CHANGELOG.md index 4f7c17436..a3b3a1ee9 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/sdk/include/opentelemetry/sdk/metrics/metric_reader.h b/sdk/include/opentelemetry/sdk/metrics/metric_reader.h index 481386131..64bf62e24 100644 --- a/sdk/include/opentelemetry/sdk/metrics/metric_reader.h +++ b/sdk/include/opentelemetry/sdk/metrics/metric_reader.h @@ -6,6 +6,7 @@ #include #include #include +#include #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 shutdown_{false}; CardinalityLimits cardinality_limits_; }; diff --git a/sdk/src/metrics/metric_reader.cc b/sdk/src/metrics/metric_reader.cc index c9312fc18..3220aa61f 100644 --- a/sdk/src/metrics/metric_reader.cc +++ b/sdk/src/metrics/metric_reader.cc @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 #include "opentelemetry/sdk/metrics/metric_reader.h" +#include #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 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; diff --git a/sdk/test/metrics/metric_reader_test.cc b/sdk/test/metrics/metric_reader_test.cc index 0860fa1ef..c648ab661 100644 --- a/sdk/test/metrics/metric_reader_test.cc +++ b/sdk/test/metrics/metric_reader_test.cc @@ -2,11 +2,17 @@ // SPDX-License-Identifier: Apache-2.0 #include -#include +#include +#include +#include #include +#include #include +#include #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 shutdown_count{0}; + std::atomic 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(new internal_log::NoopLogHandler())); + + std::vector 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 entered_shutdown; + std::promise release_shutdown; + std::atomic 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(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 second_returned{false}; + std::promise 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)); +}