From 94cc294fc23285eb0264cded2fc77561440799a6 Mon Sep 17 00:00:00 2001 From: om7057 Date: Thu, 3 Sep 2026 08:33:02 +0530 Subject: [PATCH 1/7] [EXPORTER] Fix Elasticsearch log exporter Shutdown ignoring its timeout Shutdown() never read its timeout parameter and always returned true regardless of whether anything had actually flushed. It now flushes pending exports against the caller's deadline via ForceFlush(timeout) before cancelling sessions, and returns what that flush reported. Also closes an admission race: Export() could pass its isShutdown() check and still register a session after Shutdown() had already taken its session_counter_ snapshot in ForceFlush(), so the flush could return without ever having waited for it. Both the shutdown flag and session registration now share one lock. Fixes #4359 --- CHANGELOG.md | 9 +++++ .../elasticsearch/es_log_record_exporter.h | 6 ++- .../src/es_log_record_exporter.cc | 34 +++++++++++++++-- .../test/es_log_record_exporter_test.cc | 37 +++++++++++++++++++ 4 files changed, 80 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 122cf6be0c..6f85c5136f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,15 @@ Increment the: change. [#4624](https://github.com/open-telemetry/opentelemetry-cpp/pull/4624) +* [EXPORTER] Fix the Elasticsearch log exporter's `Shutdown()` ignoring its + timeout and always reporting success. It now flushes pending exports against + the caller's deadline before cancelling sessions, and returns whether that + flush actually completed in time. Also closes a race where a session could + register for export after `Shutdown()` had already taken its snapshot of + in-flight sessions, so `ForceFlush()` could return without ever waiting for + it. + [#4359](https://github.com/open-telemetry/opentelemetry-cpp/issues/4359) + * [EXAMPLES] Fix random attribute selection in metrics foo example to include all key-value pairs [#4585](https://github.com/open-telemetry/opentelemetry-cpp/pull/4585) diff --git a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h index 6234df59a2..4d32d0a70f 100644 --- a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h +++ b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h @@ -129,8 +129,10 @@ class ElasticsearchLogRecordExporter final : public opentelemetry::sdk::logs::Lo std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept override; /** - * Shutdown this exporter. - * @param timeout The maximum time to wait for the shutdown method to return + * Shutdown this exporter. Flushes any pending export within the given timeout before + * cancelling remaining sessions, then reports whether the flush completed in time. + * @param timeout The maximum time to wait for pending exports to flush before shutting down + * @return true if all pending exports flushed before the timeout, false otherwise */ bool Shutdown( std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept override; diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index cd3b1bdbfd..18747a31f7 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -441,8 +441,20 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( request->SetBody(body_vec); #ifdef ENABLE_ASYNC_EXPORT - // Send the request - synchronization_data_->session_counter_.fetch_add(1, std::memory_order_release); + // Send the request. Registration has to happen under the same lock Shutdown() takes to + // flip is_shutdown_ and snapshot session_counter_ (see ForceFlush()) - otherwise a session + // that passes the isShutdown() check above can still register after Shutdown() has already + // taken its snapshot, and ForceFlush() would return without ever having waited for it. + { + std::lock_guard lock_guard{synchronization_data_->force_flush_m}; + if (isShutdown()) + { + OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Exporting " + << records.size() << " log(s) failed, exporter is shutdown"); + return sdk::common::ExportResult::kFailure; + } + synchronization_data_->session_counter_.fetch_add(1, std::memory_order_release); + } std::size_t span_count = records.size(); auto synchronization_data = synchronization_data_; auto handler = std::make_shared( @@ -549,15 +561,29 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou #endif } -bool ElasticsearchLogRecordExporter::Shutdown(std::chrono::microseconds /* timeout */) noexcept +bool ElasticsearchLogRecordExporter::Shutdown(std::chrono::microseconds timeout) noexcept { +#ifdef ENABLE_ASYNC_EXPORT + { + // Same lock Export() takes around its isShutdown() check and registration, so that by the + // time ForceFlush() below takes its session_counter_ snapshot, every session that is + // going to register for this shutdown already has. + std::lock_guard lock_guard{synchronization_data_->force_flush_m}; + is_shutdown_ = true; + } +#else is_shutdown_ = true; +#endif + + // Flush with the caller's deadline before cancelling anything, so the wait below has + // something to wait for. Cancelling first would leave nothing pending to flush. + const bool flushed = ForceFlush(timeout); // Shutdown the session manager http_client_->CancelAllSessions(); http_client_->FinishAllSessions(); - return true; + return flushed; } bool ElasticsearchLogRecordExporter::isShutdown() const noexcept diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 52a5d9b6ef..3eb2e32ff9 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -157,6 +157,43 @@ TEST(ElasticsearchLogsExporterTests, CustomClientConstructionSucceeds) ASSERT_NE(exporter, nullptr); } +// Regression test: Shutdown() used to ignore its timeout parameter entirely and always return +// true, whether or not anything had actually flushed. It should now report what flushing (via +// ForceFlush) actually found. With FakeHttpClient, every export completes synchronously inside +// Export() itself, so there is nothing left pending by the time Shutdown() runs. +TEST(ElasticsearchLogsExporterTests, ShutdownReportsFlushCompletion) +{ + logs_exporter::ElasticsearchExporterOptions options; + auto http_client = std::make_shared(); + auto exporter = std::unique_ptr( + new logs_exporter::ElasticsearchLogRecordExporter(options, http_client)); + + auto record = exporter->MakeRecordable(); + record->SetBody("shutdown regression test"); + auto export_result = + exporter->Export(nostd::span>(&record, 1)); + ASSERT_EQ(export_result, opentelemetry::sdk::common::ExportResult::kSuccess); + + EXPECT_TRUE(exporter->Shutdown(std::chrono::seconds(1))); +} + +// Regression test: once Shutdown() has been called, any later Export() must fail rather than +// silently trying to register a session against an exporter that is already tearing down. +TEST(ElasticsearchLogsExporterTests, ExportAfterShutdownFails) +{ + logs_exporter::ElasticsearchExporterOptions options; + auto http_client = std::make_shared(); + auto exporter = std::unique_ptr( + new logs_exporter::ElasticsearchLogRecordExporter(options, http_client)); + + ASSERT_TRUE(exporter->Shutdown(std::chrono::seconds(1))); + + auto record = exporter->MakeRecordable(); + auto result = exporter->Export(nostd::span>(&record, 1)); + + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure); +} + // Attempt to write a log to an invalid host/port, test that the Export() returns failure TEST(DISABLED_ElasticsearchLogsExporterTests, InvalidEndpoint) { From 96f4e6c35dfa9df1888c3132dddde6a271cbf401 Mon Sep 17 00:00:00 2001 From: Om Kulkarni Date: Sun, 20 Sep 2026 19:13:01 +0530 Subject: [PATCH 2/7] [EXPORTER] Clamp ForceFlush wait to caller's deadline, verify completion directly ForceFlush() always waited response_timeout_ on its condition variable regardless of how much of the caller's own timeout was left, so Shutdown(timeout) could block far longer than requested whenever an export never completes. The wait is now clamped to whatever remains of the caller's deadline, matching the pattern already used by the OTLP HTTP client's ForceFlush(). Clamping surfaced a second, pre-existing bug: the loop treated any cv_status::no_timeout return from wait_for() as proof of completion, but the standard permits a spurious wakeup to report no_timeout too, indistinguishable from a real notification. A short-lived wait makes that far more likely to be hit, so ForceFlush() could report success without anything having actually finished. Completion is now always verified directly against finished_session_counter_ rather than inferred from the wait's return value. Reported by @thc1006 in review of open-telemetry/opentelemetry-cpp#4523. --- .../src/es_log_record_exporter.cc | 23 ++++-- .../test/es_log_record_exporter_test.cc | 72 +++++++++++++++++++ 2 files changed, 88 insertions(+), 7 deletions(-) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index 18747a31f7..b57d331e49 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -537,7 +537,10 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou } std::unique_lock lk_cv(synchronization_data_->force_flush_cv_m); - // Wait for all the sessions to finish + // Wait for all the sessions to finish. The condition variable's return value is not trusted + // on its own: the standard permits wait_for() to report cv_status::no_timeout on a spurious + // wakeup, indistinguishable from a real notification, so completion is always verified + // against finished_session_counter_ directly instead of inferred from the wait's return. while (timeout_steady > std::chrono::steady_clock::duration::zero()) { if (synchronization_data_->finished_session_counter_.load(std::memory_order_acquire) >= @@ -546,16 +549,22 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou break; } + // Clamp the wait to whatever is left of the caller's deadline: waiting the full + // response_timeout_ regardless of timeout_steady would let Shutdown(timeout) block far + // longer than the timeout it was given whenever nothing ever notifies this condition + // variable (e.g. an export that never completes). + const std::chrono::steady_clock::duration wait_interval = (std::min)( + std::chrono::duration_cast( + std::chrono::seconds{options_.response_timeout_}), + timeout_steady); + std::chrono::steady_clock::time_point start_timepoint = std::chrono::steady_clock::now(); - if (std::cv_status::no_timeout != synchronization_data_->force_flush_cv.wait_for( - lk_cv, std::chrono::seconds{options_.response_timeout_})) - { - break; - } + synchronization_data_->force_flush_cv.wait_for(lk_cv, wait_interval); timeout_steady -= std::chrono::steady_clock::now() - start_timepoint; } - return timeout_steady > std::chrono::steady_clock::duration::zero(); + return synchronization_data_->finished_session_counter_.load(std::memory_order_acquire) >= + running_counter; #else return true; #endif diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 3eb2e32ff9..fd768799cd 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -120,6 +120,49 @@ class FakeHttpClient final : public http_client::HttpClient void SetMaxSessionsPerConnection(std::size_t) noexcept override {} }; +// A session that accepts a handler and keeps it forever, never calling back into it. Used to +// keep an async export outstanding by the time Shutdown()/ForceFlush() runs, so their wait +// loop actually has something to wait for instead of finding nothing pending. +// +// Only meaningful under ENABLE_ASYNC_EXPORT: that is the only build where ForceFlush() waits on +// anything at all (see ElasticsearchLogRecordExporter::ForceFlush). +#ifdef ENABLE_ASYNC_EXPORT +class HoldingSession final : public http_client::Session +{ +public: + std::shared_ptr CreateRequest() noexcept override + { + return std::make_shared(); + } + + void SendRequest(std::shared_ptr handler) noexcept override + { + held_ = std::move(handler); + } + + bool IsSessionActive() noexcept override { return true; } + bool CancelSession() noexcept override { return true; } + bool FinishSession() noexcept override { return true; } + +private: + std::shared_ptr held_; +}; + +class HoldingHttpClient final : public http_client::HttpClient +{ +public: + std::shared_ptr CreateSession( + opentelemetry::nostd::string_view) noexcept override + { + return std::make_shared(); + } + + bool CancelAllSessions() noexcept override { return true; } + bool FinishAllSessions() noexcept override { return true; } + void SetMaxSessionsPerConnection(std::size_t) noexcept override {} +}; +#endif // ENABLE_ASYNC_EXPORT + } // namespace namespace sdklogs = opentelemetry::sdk::logs; @@ -177,6 +220,35 @@ TEST(ElasticsearchLogsExporterTests, ShutdownReportsFlushCompletion) EXPECT_TRUE(exporter->Shutdown(std::chrono::seconds(1))); } +// Regression test: ForceFlush()'s wait loop always waited the full response_timeout_ on its +// condition variable, regardless of how much of the caller's own timeout was left. With an +// export still outstanding and nothing to notify the condition variable, Shutdown(1us) waited +// the full response_timeout_ (30s by default) instead of returning after about 1us. The wait +// is now clamped to whatever remains of the caller's deadline, matching the pattern already +// used by the OTLP HTTP client's ForceFlush(). +#ifdef ENABLE_ASYNC_EXPORT +TEST(ElasticsearchLogsExporterTests, ShutdownClampsWaitToCallerTimeoutWhenExportIsOutstanding) +{ + logs_exporter::ElasticsearchExporterOptions options; + auto http_client = std::make_shared(); + auto exporter = std::unique_ptr( + new logs_exporter::ElasticsearchLogRecordExporter(options, http_client)); + + auto record = exporter->MakeRecordable(); + record->SetBody("this export never completes"); + auto export_result = + exporter->Export(nostd::span>(&record, 1)); + ASSERT_EQ(export_result, opentelemetry::sdk::common::ExportResult::kSuccess); + + auto start = std::chrono::steady_clock::now(); + bool result = exporter->Shutdown(std::chrono::microseconds(1)); + auto elapsed = std::chrono::steady_clock::now() - start; + + EXPECT_FALSE(result); + EXPECT_LT(elapsed, std::chrono::seconds(1)); +} +#endif // ENABLE_ASYNC_EXPORT + // Regression test: once Shutdown() has been called, any later Export() must fail rather than // silently trying to register a session against an exporter that is already tearing down. TEST(ElasticsearchLogsExporterTests, ExportAfterShutdownFails) From 3dd61668af25204f0d7bdcdf59938faaf6909084 Mon Sep 17 00:00:00 2001 From: Om Kulkarni Date: Mon, 21 Sep 2026 23:39:14 +0530 Subject: [PATCH 3/7] Fix CI failures on this branch: format, IWYU, and a test leak - clang-format wanted wait_interval's line-wrap reflowed. - IWYU wanted an explicit include for std::min. - HoldingSession held a shared_ptr back to the AsyncResponseHandler it was given, which itself holds a shared_ptr to the Session, forming a reference cycle that ASan/Valgrind flagged as a leak. Nothing on the AsyncResponseHandler destruction path touches finished_session_counter_ regardless of whether the handler is kept alive, so the retention was unnecessary; dropping the handler without storing it keeps the export "outstanding" just as well and removes the cycle. --- .../src/es_log_record_exporter.cc | 9 +++++---- .../test/es_log_record_exporter_test.cc | 20 ++++++++----------- 2 files changed, 13 insertions(+), 16 deletions(-) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index b57d331e49..8fb4ca0ab6 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -1,6 +1,7 @@ // Copyright The OpenTelemetry Authors // SPDX-License-Identifier: Apache-2.0 +#include #include #include #include @@ -553,10 +554,10 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou // response_timeout_ regardless of timeout_steady would let Shutdown(timeout) block far // longer than the timeout it was given whenever nothing ever notifies this condition // variable (e.g. an export that never completes). - const std::chrono::steady_clock::duration wait_interval = (std::min)( - std::chrono::duration_cast( - std::chrono::seconds{options_.response_timeout_}), - timeout_steady); + const std::chrono::steady_clock::duration wait_interval = + (std::min)(std::chrono::duration_cast( + std::chrono::seconds{options_.response_timeout_}), + timeout_steady); std::chrono::steady_clock::time_point start_timepoint = std::chrono::steady_clock::now(); synchronization_data_->force_flush_cv.wait_for(lk_cv, wait_interval); diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index fd768799cd..4859329cf7 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -120,9 +120,11 @@ class FakeHttpClient final : public http_client::HttpClient void SetMaxSessionsPerConnection(std::size_t) noexcept override {} }; -// A session that accepts a handler and keeps it forever, never calling back into it. Used to -// keep an async export outstanding by the time Shutdown()/ForceFlush() runs, so their wait -// loop actually has something to wait for instead of finding nothing pending. +// A session that drops the handler it is given without ever calling back into it. Nothing on +// the AsyncResponseHandler destruction path touches finished_session_counter_, so an export +// through this session stays counted as outstanding for as long as the exporter lives, whether +// or not the handler itself is retained; not retaining it avoids a Session/AsyncResponseHandler +// reference cycle (they hold shared_ptrs to each other) that a leak sanitizer would flag. // // Only meaningful under ENABLE_ASYNC_EXPORT: that is the only build where ForceFlush() waits on // anything at all (see ElasticsearchLogRecordExporter::ForceFlush). @@ -135,17 +137,11 @@ class HoldingSession final : public http_client::Session return std::make_shared(); } - void SendRequest(std::shared_ptr handler) noexcept override - { - held_ = std::move(handler); - } + void SendRequest(std::shared_ptr) noexcept override {} bool IsSessionActive() noexcept override { return true; } bool CancelSession() noexcept override { return true; } bool FinishSession() noexcept override { return true; } - -private: - std::shared_ptr held_; }; class HoldingHttpClient final : public http_client::HttpClient @@ -240,8 +236,8 @@ TEST(ElasticsearchLogsExporterTests, ShutdownClampsWaitToCallerTimeoutWhenExport exporter->Export(nostd::span>(&record, 1)); ASSERT_EQ(export_result, opentelemetry::sdk::common::ExportResult::kSuccess); - auto start = std::chrono::steady_clock::now(); - bool result = exporter->Shutdown(std::chrono::microseconds(1)); + auto start = std::chrono::steady_clock::now(); + bool result = exporter->Shutdown(std::chrono::microseconds(1)); auto elapsed = std::chrono::steady_clock::now() - start; EXPECT_FALSE(result); From a0640b2283eee2d951df34a4e2817092a8efe10c Mon Sep 17 00:00:00 2001 From: Om Kulkarni Date: Tue, 22 Sep 2026 19:46:07 +0530 Subject: [PATCH 4/7] Fix IWYU: mark as keep since std::min is only used under ENABLE_ASYNC_EXPORT --- exporters/elasticsearch/src/es_log_record_exporter.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index 8fb4ca0ab6..25aa4e942e 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -1,7 +1,7 @@ // Copyright The OpenTelemetry Authors // SPDX-License-Identifier: Apache-2.0 -#include +#include // IWYU pragma: keep #include #include #include From a2372e995c7508d57eb35bcd43a20333868438b9 Mon Sep 17 00:00:00 2001 From: Om Kulkarni Date: Tue, 22 Sep 2026 19:52:26 +0530 Subject: [PATCH 5/7] Fix HoldingSession dropping the handler, which conflicts with #4502 Applied thc1006's patch from PR review: HoldingSession now parks the handler in a variable the test case owns instead of dropping it, so an export through it stays outstanding regardless of what #4502 changes about AsyncResponseHandler's destructor. A raw pointer to the parked variable (not a shared_ptr) keeps the session/client from forming a reference cycle with the handler. --- .../test/es_log_record_exporter_test.cc | 38 +++++++++++++++---- 1 file changed, 30 insertions(+), 8 deletions(-) diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 4859329cf7..2847752503 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -120,11 +120,12 @@ class FakeHttpClient final : public http_client::HttpClient void SetMaxSessionsPerConnection(std::size_t) noexcept override {} }; -// A session that drops the handler it is given without ever calling back into it. Nothing on -// the AsyncResponseHandler destruction path touches finished_session_counter_, so an export -// through this session stays counted as outstanding for as long as the exporter lives, whether -// or not the handler itself is retained; not retaining it avoids a Session/AsyncResponseHandler -// reference cycle (they hold shared_ptrs to each other) that a leak sanitizer would flag. +// A session that parks the handler it is given, in the test case rather than in the session +// itself, without ever calling back into it. The handler stays alive (so the export it +// represents keeps counting as outstanding) without creating a Session/AsyncResponseHandler +// reference cycle: AsyncResponseHandler holds a shared_ptr to its session, so a session that +// held the handler back would keep both alive for the exporter's own lifetime, which a leak +// sanitizer would flag. // // Only meaningful under ENABLE_ASYNC_EXPORT: that is the only build where ForceFlush() waits on // anything at all (see ElasticsearchLogRecordExporter::ForceFlush). @@ -132,30 +133,46 @@ class FakeHttpClient final : public http_client::HttpClient class HoldingSession final : public http_client::Session { public: + explicit HoldingSession(std::shared_ptr *parked) : parked_(parked) {} + std::shared_ptr CreateRequest() noexcept override { return std::make_shared(); } - void SendRequest(std::shared_ptr) noexcept override {} + // Parked where the case can see it, not in this session: the handler owns its session, so a + // session that owned the handler back would keep the pair alive. + void SendRequest(std::shared_ptr handler) noexcept override + { + *parked_ = std::move(handler); + } bool IsSessionActive() noexcept override { return true; } bool CancelSession() noexcept override { return true; } bool FinishSession() noexcept override { return true; } + +private: + std::shared_ptr *parked_; }; class HoldingHttpClient final : public http_client::HttpClient { public: + explicit HoldingHttpClient(std::shared_ptr *parked) : parked_(parked) + {} + std::shared_ptr CreateSession( opentelemetry::nostd::string_view) noexcept override { - return std::make_shared(); + return std::make_shared(parked_); } bool CancelAllSessions() noexcept override { return true; } bool FinishAllSessions() noexcept override { return true; } void SetMaxSessionsPerConnection(std::size_t) noexcept override {} + +private: + std::shared_ptr *parked_; }; #endif // ENABLE_ASYNC_EXPORT @@ -225,8 +242,10 @@ TEST(ElasticsearchLogsExporterTests, ShutdownReportsFlushCompletion) #ifdef ENABLE_ASYNC_EXPORT TEST(ElasticsearchLogsExporterTests, ShutdownClampsWaitToCallerTimeoutWhenExportIsOutstanding) { + // Declared first so it outlives the client and the session that point at it. + std::shared_ptr parked; logs_exporter::ElasticsearchExporterOptions options; - auto http_client = std::make_shared(); + auto http_client = std::make_shared(&parked); auto exporter = std::unique_ptr( new logs_exporter::ElasticsearchLogRecordExporter(options, http_client)); @@ -242,6 +261,9 @@ TEST(ElasticsearchLogsExporterTests, ShutdownClampsWaitToCallerTimeoutWhenExport EXPECT_FALSE(result); EXPECT_LT(elapsed, std::chrono::seconds(1)); + + // Let the export finish now that the assertions are done. + parked.reset(); } #endif // ENABLE_ASYNC_EXPORT From 9a14aa8faf1b69935edc0ac7efb46a4ab044cdb7 Mon Sep 17 00:00:00 2001 From: Om Kulkarni Date: Tue, 22 Sep 2026 21:37:15 +0530 Subject: [PATCH 6/7] Fix session leak when Export() is rejected mid-CreateSession by Shutdown() The second shutdown check added earlier runs after CreateSession(), so a real client that retains a shared_ptr to every session it creates (e.g. curl) never gets that session back on the rejected path, since the handler that would normally call FinishSession() is never built. Move the rejection decision under the lock but call session->FinishSession() outside it before returning failure, so the client's own session bookkeeping is not left dangling. Applied thc1006's patch and regression test (ARejectedExportHandsItsSessionBack) from PR review, using a RetainingHttpClient/RetainingSession pair that mimics curl's session retention and gates CreateSession() so Shutdown() can land mid-export. Verified locally: 7/7 pass with async export on, 5 pass + 1 skip with it off, --gtest_repeat=30 on the new test is stable, and ASan+UBSan+LSan report nothing. --- .../src/es_log_record_exporter.cc | 19 ++- .../test/es_log_record_exporter_test.cc | 150 ++++++++++++++++++ 2 files changed, 164 insertions(+), 5 deletions(-) diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index 25aa4e942e..144d9e6f2a 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -442,19 +442,28 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( request->SetBody(body_vec); #ifdef ENABLE_ASYNC_EXPORT + bool rejected = false; // Send the request. Registration has to happen under the same lock Shutdown() takes to // flip is_shutdown_ and snapshot session_counter_ (see ForceFlush()) - otherwise a session // that passes the isShutdown() check above can still register after Shutdown() has already // taken its snapshot, and ForceFlush() would return without ever having waited for it. { std::lock_guard lock_guard{synchronization_data_->force_flush_m}; - if (isShutdown()) + rejected = isShutdown(); + if (!rejected) { - OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Exporting " - << records.size() << " log(s) failed, exporter is shutdown"); - return sdk::common::ExportResult::kFailure; + synchronization_data_->session_counter_.fetch_add(1, std::memory_order_release); } - synchronization_data_->session_counter_.fetch_add(1, std::memory_order_release); + } + + // Outside the lock: the client owns the session until somebody hands it back, and this is the + // only path that can, since the handler that would do it later is never built. + if (rejected) + { + session->FinishSession(); + OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Exporting " + << records.size() << " log(s) failed, exporter is shutdown"); + return sdk::common::ExportResult::kFailure; } std::size_t span_count = records.size(); auto synchronization_data = synchronization_data_; diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index 2847752503..dc06115b0e 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -19,10 +19,14 @@ #include #include #include +#include #include #include +#include #include +#include #include +#include #include "nlohmann/json.hpp" namespace @@ -106,6 +110,114 @@ class FakeSession final : public http_client::Session bool FinishSession() noexcept override { return true; } }; +// A client that owns what it creates, the way the curl client does: CreateSession() keeps a +// reference in the client and only FinishSession() gives it back. CreateSession() also waits on +// a gate, so a case can run Shutdown() while an export sits between its two shutdown checks. +class RetainingSession final : public http_client::Session +{ +public: + std::shared_ptr CreateRequest() noexcept override + { + return std::make_shared(); + } + + void SendRequest(std::shared_ptr) noexcept override { sent_ = true; } + + bool IsSessionActive() noexcept override { return finish_calls_ == 0; } + bool CancelSession() noexcept override { return true; } + // Counted rather than flagged: handing a session back twice is as wrong as not at all, and a + // flag reads the same either way. + bool FinishSession() noexcept override + { + ++finish_calls_; + return true; + } + + bool sent_ = false; + std::size_t finish_calls_ = 0; +}; + +class RetainingHttpClient final : public http_client::HttpClient +{ +public: + std::shared_ptr CreateSession( + opentelemetry::nostd::string_view) noexcept override + { + { + std::unique_lock lock{gate_m_}; + entered_ = true; + gate_cv_.notify_all(); + gate_cv_.wait(lock, [this] { return released_; }); + } + auto session = std::make_shared(); + std::lock_guard lock{sessions_m_}; + sessions_.push_back(session); + return session; + } + + bool CancelAllSessions() noexcept override { return true; } + bool FinishAllSessions() noexcept override { return true; } + void SetMaxSessionsPerConnection(std::size_t) noexcept override {} + + void WaitUntilCreating() + { + std::unique_lock lock{gate_m_}; + gate_cv_.wait(lock, [this] { return entered_; }); + } + + void Release() + { + { + std::lock_guard lock{gate_m_}; + released_ = true; + } + gate_cv_.notify_all(); + } + + std::size_t Sent() + { + std::lock_guard lock{sessions_m_}; + std::size_t n = 0; + for (const auto &s : sessions_) + { + n += s->sent_ ? 1 : 0; + } + return n; + } + + std::size_t FinishCalls() + { + std::lock_guard lock{sessions_m_}; + std::size_t n = 0; + for (const auto &s : sessions_) + { + n += s->finish_calls_; + } + return n; + } + + // What the client is still holding that nobody handed back. + std::size_t Retained() + { + std::lock_guard lock{sessions_m_}; + std::size_t n = 0; + for (const auto &s : sessions_) + { + n += s->finish_calls_ == 0 ? 1 : 0; + } + return n; + } + +private: + std::mutex gate_m_; + std::condition_variable gate_cv_; + bool entered_ = false; + bool released_ = false; + + std::mutex sessions_m_; + std::vector> sessions_; +}; + class FakeHttpClient final : public http_client::HttpClient { public: @@ -269,6 +381,44 @@ TEST(ElasticsearchLogsExporterTests, ShutdownClampsWaitToCallerTimeoutWhenExport // Regression test: once Shutdown() has been called, any later Export() must fail rather than // silently trying to register a session against an exporter that is already tearing down. +// The case is only meaningful where the second shutdown check exists, but it is registered in +// both builds: gtest_add_tests reads the source, so a case behind #ifdef is still handed to CTest +// in the build that does not compile it and reports a pass it never ran. +TEST(ElasticsearchLogsExporterTests, ARejectedExportHandsItsSessionBack) +{ +#ifndef ENABLE_ASYNC_EXPORT + GTEST_SKIP() << "the shutdown re-check this covers is compiled only with async export"; +#else + auto client = std::make_shared(); + logs_exporter::ElasticsearchExporterOptions options; + auto exporter = std::unique_ptr( + new logs_exporter::ElasticsearchLogRecordExporter(options, client)); + + auto record = exporter->MakeRecordable(); + record->SetBody("a record the exporter will refuse"); + std::array, 1> batch = {std::move(record)}; + + auto result = opentelemetry::sdk::common::ExportResult::kSuccess; + std::thread exporting([&] { + result = exporter->Export( + nostd::span>(batch.data(), batch.size())); + }); + + // Shutdown lands while the export is inside CreateSession, so it sees no registered session + // and returns, and the export then meets the second check on its way back. + client->WaitUntilCreating(); + exporter->Shutdown(); + client->Release(); + exporting.join(); + + EXPECT_EQ(opentelemetry::sdk::common::ExportResult::kFailure, result); + EXPECT_EQ(0u, client->Sent()) << "a rejected export must not send"; + EXPECT_EQ(0u, client->Retained()) + << "the rejected session is still held by the client, so nothing will call FinishSession"; + EXPECT_EQ(1u, client->FinishCalls()) << "handed back exactly once, not twice"; +#endif +} + TEST(ElasticsearchLogsExporterTests, ExportAfterShutdownFails) { logs_exporter::ElasticsearchExporterOptions options; From 694036db5a2b6a8b40bff2235593dc009681d385 Mon Sep 17 00:00:00 2001 From: Om Kulkarni Date: Wed, 23 Sep 2026 00:24:31 +0530 Subject: [PATCH 7/7] Fix IWYU: mark as keep, only used under ENABLE_ASYNC_EXPORT --- exporters/elasticsearch/test/es_log_record_exporter_test.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index dc06115b0e..8fd02acc47 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -24,7 +24,7 @@ #include #include #include -#include +#include // IWYU pragma: keep #include #include #include "nlohmann/json.hpp"