diff --git a/CHANGELOG.md b/CHANGELOG.md index 528a0e1733..5f9f7cf300 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) + * [DOC] Fix and clarify the `StartSpanOptions` documentation [#4526](https://github.com/open-telemetry/opentelemetry-cpp/pull/4526) 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) {