[EXPORTER] Fix Elasticsearch log exporter Shutdown ignoring its timeout

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)
 

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;

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<std::recursive_mutex> 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<AsyncResponseHandler>(
@@ -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<std::recursive_mutex> 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

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<FakeHttpClient>();
+  auto exporter    = std::unique_ptr<sdklogs::LogRecordExporter>(
+      new logs_exporter::ElasticsearchLogRecordExporter(options, http_client));
+
+  auto record = exporter->MakeRecordable();
+  record->SetBody("shutdown regression test");
+  auto export_result =
+      exporter->Export(nostd::span<std::unique_ptr<sdklogs::Recordable>>(&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<FakeHttpClient>();
+  auto exporter    = std::unique_ptr<sdklogs::LogRecordExporter>(
+      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<std::unique_ptr<sdklogs::Recordable>>(&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)
 {