diff --git a/CHANGELOG.md b/CHANGELOG.md index 97cde89252..2ef13659ef 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,10 @@ Increment the: ## [Unreleased] +* [SDK] Preserve the default unlimited timeout when batch span and log record + processors shut down their exporters. + [#4686](https://github.com/open-telemetry/opentelemetry-cpp/pull/4686) + * [CODE HEALTH] Prevent `std::out_of_range` escaping `noexcept` substr callers. [#4562](https://github.com/open-telemetry/opentelemetry-cpp/pull/4562) diff --git a/sdk/src/logs/batch_log_record_processor.cc b/sdk/src/logs/batch_log_record_processor.cc index ec5f449978..f5e01671ba 100644 --- a/sdk/src/logs/batch_log_record_processor.cc +++ b/sdk/src/logs/batch_log_record_processor.cc @@ -359,6 +359,11 @@ void BatchLogRecordProcessor::GetWaitAdjustedTime( std::chrono::microseconds &timeout, std::chrono::time_point &start_time) { + if (timeout == (std::chrono::microseconds::max)()) + { + return; + } + auto end_time = std::chrono::system_clock::now(); auto offset = std::chrono::duration_cast(end_time - start_time); start_time = end_time; diff --git a/sdk/src/logs/multi_log_record_processor.cc b/sdk/src/logs/multi_log_record_processor.cc index e74d8b3bb5..398df9828f 100644 --- a/sdk/src/logs/multi_log_record_processor.cc +++ b/sdk/src/logs/multi_log_record_processor.cc @@ -169,9 +169,10 @@ bool MultiLogRecordProcessor::InternalForceFlush(std::chrono::microseconds timeo bool MultiLogRecordProcessor::InternalShutdown(std::chrono::microseconds timeout) noexcept { - bool result = true; - auto start_time = std::chrono::system_clock::now(); - auto overflow_checker = std::chrono::system_clock::time_point::max(); + bool result = true; + bool has_unlimited_timeout = timeout == (std::chrono::microseconds::max)(); + auto start_time = std::chrono::system_clock::now(); + auto overflow_checker = std::chrono::system_clock::time_point::max(); std::chrono::system_clock::time_point expire_time; if (std::chrono::duration_cast(overflow_checker - start_time) <= timeout) @@ -189,6 +190,10 @@ bool MultiLogRecordProcessor::InternalShutdown(std::chrono::microseconds timeout { result = false; } + if (has_unlimited_timeout) + { + continue; + } start_time = std::chrono::system_clock::now(); if (expire_time > start_time) { diff --git a/sdk/src/trace/batch_span_processor.cc b/sdk/src/trace/batch_span_processor.cc index 403fcb55af..8f9eec8507 100644 --- a/sdk/src/trace/batch_span_processor.cc +++ b/sdk/src/trace/batch_span_processor.cc @@ -348,6 +348,11 @@ void BatchSpanProcessor::GetWaitAdjustedTime( std::chrono::microseconds &timeout, std::chrono::time_point &start_time) { + if (timeout == (std::chrono::microseconds::max)()) + { + return; + } + auto end_time = std::chrono::system_clock::now(); auto offset = std::chrono::duration_cast(end_time - start_time); start_time = end_time; diff --git a/sdk/test/logs/batch_log_record_processor_test.cc b/sdk/test/logs/batch_log_record_processor_test.cc index f9297153c5..22862eb2a2 100644 --- a/sdk/test/logs/batch_log_record_processor_test.cc +++ b/sdk/test/logs/batch_log_record_processor_test.cc @@ -142,18 +142,22 @@ class MockLogExporter final : public LogRecordExporter } // toggles the boolean flag marking this exporter as shut down - bool Shutdown(std::chrono::microseconds /* timeout */) noexcept override + bool Shutdown(std::chrono::microseconds timeout) noexcept override { - *is_shutdown_ = true; + shutdown_timeout_ = timeout; + *is_shutdown_ = true; return true; } + std::chrono::microseconds GetShutdownTimeout() const noexcept { return shutdown_timeout_; } + private: std::shared_ptr>> logs_received_; std::shared_ptr> force_flush_counter_; std::shared_ptr> is_shutdown_; std::shared_ptr> is_export_completed_; std::chrono::milliseconds export_delay_; + std::chrono::microseconds shutdown_timeout_{std::chrono::microseconds::zero()}; }; /** @@ -191,8 +195,12 @@ TEST_F(BatchLogRecordProcessorTest, TestShutdown) new std::vector>); std::shared_ptr> force_flush_counter(new std::atomic(0)); std::shared_ptr> is_shutdown(new std::atomic(false)); + std::shared_ptr> is_export_completed(new std::atomic(false)); - auto batch_processor = GetMockProcessor(logs_received, force_flush_counter, is_shutdown); + auto exporter = + new MockLogExporter(logs_received, force_flush_counter, is_shutdown, is_export_completed); + auto batch_processor = std::shared_ptr( + new BatchLogRecordProcessor(std::unique_ptr(exporter))); // Create a few test log records and send them to the processor const int num_logs = 3; @@ -221,6 +229,7 @@ TEST_F(BatchLogRecordProcessorTest, TestShutdown) // Also check that the processor is shut down at the end EXPECT_TRUE(is_shutdown->load()); + EXPECT_EQ((std::chrono::microseconds::max)(), exporter->GetShutdownTimeout()); } TEST_F(BatchLogRecordProcessorTest, TestForceFlush) diff --git a/sdk/test/logs/multi_log_record_processor_test.cc b/sdk/test/logs/multi_log_record_processor_test.cc index 9b930bb213..7b3a97eeb7 100644 --- a/sdk/test/logs/multi_log_record_processor_test.cc +++ b/sdk/test/logs/multi_log_record_processor_test.cc @@ -7,6 +7,7 @@ #include #include #include +#include #include #include @@ -172,7 +173,14 @@ TEST(MultiLogRecordProcessorTest, EmptyProcessorIsDisabled) class FixedResultProcessor final : public LogRecordProcessor { public: - explicit FixedResultProcessor(bool result) : result_(result) {} + explicit FixedResultProcessor( + bool result, + std::shared_ptr shutdown_timeout = nullptr, + std::chrono::microseconds shutdown_delay = std::chrono::microseconds::zero()) + : result_(result), + shutdown_timeout_(std::move(shutdown_timeout)), + shutdown_delay_(shutdown_delay) + {} std::unique_ptr MakeRecordable() noexcept override { @@ -187,10 +195,20 @@ class FixedResultProcessor final : public LogRecordProcessor bool ForceFlush(std::chrono::microseconds /* timeout */) noexcept override { return result_; } - bool Shutdown(std::chrono::microseconds /* timeout */) noexcept override { return result_; } + bool Shutdown(std::chrono::microseconds timeout) noexcept override + { + if (shutdown_timeout_ != nullptr) + { + *shutdown_timeout_ = timeout; + } + std::this_thread::sleep_for(shutdown_delay_); + return result_; + } private: bool result_; + std::shared_ptr shutdown_timeout_; + std::chrono::microseconds shutdown_delay_; }; std::unique_ptr MakeMultiProcessor(bool first, bool second) @@ -217,4 +235,36 @@ TEST(MultiLogRecordProcessorTest, ShutdownFailsWhenAnyChildFails) EXPECT_FALSE(MakeMultiProcessor(false, true)->Shutdown()); } +TEST(MultiLogRecordProcessorTest, DefaultShutdownPassesUnlimitedTimeoutToAllChildren) +{ + auto first_timeout = std::make_shared(); + auto second_timeout = std::make_shared(); + + std::vector> processors; + processors.emplace_back(new FixedResultProcessor(true, first_timeout)); + processors.emplace_back(new FixedResultProcessor(true, second_timeout)); + MultiLogRecordProcessor processor(std::move(processors)); + + EXPECT_TRUE(processor.Shutdown()); + EXPECT_EQ((std::chrono::microseconds::max)(), *first_timeout); + EXPECT_EQ((std::chrono::microseconds::max)(), *second_timeout); +} + +TEST(MultiLogRecordProcessorTest, FiniteShutdownSharesTimeoutBudgetAcrossChildren) +{ + auto first_timeout = std::make_shared(); + auto second_timeout = std::make_shared(); + auto timeout = std::chrono::milliseconds(100); + + std::vector> processors; + processors.emplace_back( + new FixedResultProcessor(true, first_timeout, std::chrono::milliseconds(2))); + processors.emplace_back(new FixedResultProcessor(true, second_timeout)); + MultiLogRecordProcessor processor(std::move(processors)); + + EXPECT_TRUE(processor.Shutdown(timeout)); + EXPECT_EQ(timeout, *first_timeout); + EXPECT_LT(*second_timeout, timeout); +} + } // namespace diff --git a/sdk/test/trace/batch_span_processor_test.cc b/sdk/test/trace/batch_span_processor_test.cc index 1de105d902..b675e0406c 100644 --- a/sdk/test/trace/batch_span_processor_test.cc +++ b/sdk/test/trace/batch_span_processor_test.cc @@ -88,14 +88,17 @@ class MockSpanExporter final : public sdk::trace::SpanExporter return true; } - bool Shutdown(std::chrono::microseconds /* timeout */) noexcept override + bool Shutdown(std::chrono::microseconds timeout) noexcept override { - *is_shutdown_ = true; + shutdown_timeout_ = timeout; + *is_shutdown_ = true; return true; } bool IsExportCompleted() { return is_export_completed_->load(); } + std::chrono::microseconds GetShutdownTimeout() const noexcept { return shutdown_timeout_; } + private: std::shared_ptr>> spans_received_; std::shared_ptr> shut_down_counter_; @@ -103,6 +106,7 @@ class MockSpanExporter final : public sdk::trace::SpanExporter std::shared_ptr> is_export_completed_; // Meant exclusively to test force flush timeout std::chrono::milliseconds export_delay_; + std::chrono::microseconds shutdown_timeout_{std::chrono::microseconds::zero()}; }; /** @@ -138,11 +142,10 @@ TEST_F(BatchSpanProcessorTestPeer, TestShutdown) std::shared_ptr>> spans_received( new std::vector>); - auto batch_processor = - std::shared_ptr(new sdk::trace::BatchSpanProcessor( - std::unique_ptr( - new MockSpanExporter(spans_received, shut_down_counter, is_shutdown)), - sdk::trace::BatchSpanProcessorOptions())); + auto exporter = new MockSpanExporter(spans_received, shut_down_counter, is_shutdown); + auto batch_processor = std::shared_ptr( + new sdk::trace::BatchSpanProcessor(std::unique_ptr(exporter), + sdk::trace::BatchSpanProcessorOptions())); const int num_spans = 3; auto test_spans = GetTestSpans(batch_processor, num_spans); @@ -163,6 +166,7 @@ TEST_F(BatchSpanProcessorTestPeer, TestShutdown) } EXPECT_TRUE(is_shutdown->load()); + EXPECT_EQ((std::chrono::microseconds::max)(), exporter->GetShutdownTimeout()); } TEST_F(BatchSpanProcessorTestPeer, TestForceFlush)