From b119f94acec45d2e280a646d7c89eb5b1f321d1e Mon Sep 17 00:00:00 2001 From: om7057 Date: Thu, 3 Sep 2026 08:33:02 +0530 Subject: [PATCH] [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 e20b19e972..950ca3bd1e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,15 @@ Increment the: ## [Unreleased] +* [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) + * [EXPORTER] Fix the Elasticsearch log exporter aborting the process when a log record's body or attributes contain bytes that are not valid UTF-8. The exporter now substitutes the replacement character for the invalid bytes 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) {