[BUG] End the Elasticsearch exporter's wait on a read or write error

CHANGELOG.md

@@ -18,6 +18,9 @@ Increment the:
 * [OTLP/HTTP] Honor `Retry-After` response header when retrying exports,
   supporting both delay-seconds and HTTP-date formats per RFC 7231 §7.1.3.
   [#4172](https://github.com/open-telemetry/opentelemetry-cpp/issues/4172)
+
+* [BUG] End the Elasticsearch exporter's wait on a read or write error
+  [#4331](https://github.com/open-telemetry/opentelemetry-cpp/pull/4331)
 * [BUILD] Run the ext_http component install test on Windows
   [#4326](https://github.com/open-telemetry/opentelemetry-cpp/pull/4326)
 

exporters/elasticsearch/CMakeLists.txt

@@ -55,4 +55,8 @@ if(BUILD_TESTING)
     TARGET es_log_record_exporter_test
     TEST_PREFIX exporter.
     TEST_LIST es_log_record_exporter_test)
+
+  # These cases exist to catch a wait that never returns. Without a per test
+  # bound a regression stalls the job instead of failing it.
+  set_tests_properties(${es_log_record_exporter_test} PROPERTIES TIMEOUT 30)
 endif() # BUILD_TESTING

exporters/elasticsearch/src/es_log_record_exporter.cc

@@ -108,8 +108,8 @@ class ResponseHandler : public http_client::EventHandler
 
   /**
    * A method the user calls to block their thread until the request has either produced a
-   * response or failed. The longest duration is the timeout of the request, set by
-   * SetTimeoutMs(), which arrives here as a TimedOut session event.
+   * response or failed. It has no deadline of its own and relies on the HTTP client reporting
+   * one of the terminal session states.
    */
   bool waitForResponse()
   {
@@ -182,10 +182,19 @@ class ResponseHandler : public http_client::EventHandler
         recordCompletion(CompletionState::Failure);
         break;
       case http_client::SessionState::ReadError:
-        OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Read error");
+        // Reported only when this is the outcome Export() returns. Recording is first writer
+        // wins, so either of these can arrive after a response has already succeeded, and an
+        // error line there would describe a failure the caller was never told about.
+        if (recordCompletion(CompletionState::Failure))
+        {
+          OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Read error");
+        }
         break;
       case http_client::SessionState::WriteError:
-        OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Write error");
+        if (recordCompletion(CompletionState::Failure))
+        {
+          OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Write error");
+        }
         break;
       case http_client::SessionState::Cancelled:
         OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] (manually) cancelled");
@@ -206,22 +215,27 @@ class ResponseHandler : public http_client::EventHandler
    * Record the outcome of the request, first writer wins, then release any waiter. Keeping the
    * first outcome means a session destroyed after a successful response does not overwrite it.
    */
-  void recordCompletion(CompletionState state)
+  /// Returns whether this call is the one that decided the outcome.
+  bool recordCompletion(CompletionState state)
   {
+    bool recorded = false;
     {
       std::unique_lock<std::mutex> lk(mutex_);
-      recordCompletionLocked(state);
+      recorded = recordCompletionLocked(state);
     }
     cv_.notify_all();
+    return recorded;
   }
 
   /// As recordCompletion(), for callers that already hold mutex_ and notify themselves.
-  void recordCompletionLocked(CompletionState state)
+  bool recordCompletionLocked(CompletionState state)
   {
-    if (completion_ == CompletionState::Pending)
+    if (completion_ != CompletionState::Pending)
     {
-      completion_ = state;
+      return false;
     }
+    completion_ = state;
+    return true;
   }
 
   // Define a condition variable and mutex

exporters/elasticsearch/test/es_log_record_exporter_test.cc

@@ -4,7 +4,9 @@
 #include "opentelemetry/exporters/elasticsearch/es_log_record_exporter.h"
 #include "opentelemetry/common/timestamp.h"
 #include "opentelemetry/exporters/elasticsearch/es_log_recordable.h"
+#include "opentelemetry/ext/http/client/http_client.h"
 #include "opentelemetry/logs/severity.h"
+#include "opentelemetry/nostd/function_ref.h"
 #include "opentelemetry/nostd/span.h"
 #include "opentelemetry/nostd/string_view.h"
 #include "opentelemetry/nostd/utility.h"
@@ -19,6 +21,8 @@
 #include <chrono>
 #include <cstddef>
 #include <cstdint>
+#include <functional>
+#include <initializer_list>
 #include <string>
 #include <utility>
 #include "nlohmann/json.hpp"
@@ -142,3 +146,245 @@ TEST(ElasticsearchLogRecordableTests, BasicTests)
 
   EXPECT_EQ(actual, expected);
 }
+
+// ---------------------------------------------------------------------------
+// Synchronous completion path.
+//
+// A fake HTTP client drives a scripted sequence of callbacks from inside
+// SendRequest(), which runs before the exporter reaches waitForResponse(). Every
+// case here therefore also covers a completion recorded before the wait starts,
+// the notification a bare cv_.wait() would have missed.
+// ---------------------------------------------------------------------------
+namespace
+{
+namespace http_client = opentelemetry::ext::http::client;
+
+// Accepted by both the substring check and a top level "errors": false parse, so
+// these cases keep meaning the same thing whichever success check is in place.
+constexpr const char *kAcceptedBody =
+    R"({"took":30,"errors":false,"items":[{"index":{"_shards":{"failed" : 0}}}]})";
+
+class FakeResponse : public http_client::Response
+{
+public:
+  FakeResponse(http_client::StatusCode status, const std::string &body)
+      : status_(status), body_(body.begin(), body.end())
+  {}
+  const http_client::Body &GetBody() const noexcept override { return body_; }
+  bool ForEachHeader(
+      nostd::function_ref<bool(nostd::string_view, nostd::string_view)>) const noexcept override
+  {
+    return true;
+  }
+  bool ForEachHeader(
+      const nostd::string_view &,
+      nostd::function_ref<bool(nostd::string_view, nostd::string_view)>) const noexcept override
+  {
+    return true;
+  }
+  http_client::StatusCode GetStatusCode() const noexcept override { return status_; }
+
+private:
+  http_client::StatusCode status_;
+  http_client::Body body_;
+};
+
+class FakeRequest : public http_client::Request
+{
+public:
+  void SetMethod(http_client::Method) noexcept override {}
+  void SetUri(nostd::string_view) noexcept override {}
+  void SetSslOptions(const http_client::HttpSslOptions &) noexcept override {}
+  void SetBody(http_client::Body &) noexcept override {}
+  void AddHeader(nostd::string_view, nostd::string_view) noexcept override {}
+  void ReplaceHeader(nostd::string_view, nostd::string_view) noexcept override {}
+  void SetTimeoutMs(std::chrono::milliseconds) noexcept override {}
+  void SetCompression(const http_client::Compression &) noexcept override {}
+  void EnableLogging(bool) noexcept override {}
+  void SetRetryPolicy(const http_client::RetryPolicy &) noexcept override {}
+};
+
+using EventScript = std::function<void(http_client::EventHandler &)>;
+
+class FakeSession : public http_client::Session
+{
+public:
+  explicit FakeSession(EventScript script) : script_(std::move(script)) {}
+  std::shared_ptr<http_client::Request> CreateRequest() noexcept override
+  {
+    return std::make_shared<FakeRequest>();
+  }
+  void SendRequest(std::shared_ptr<http_client::EventHandler> handler) noexcept override
+  {
+    script_(*handler);
+  }
+  bool IsSessionActive() noexcept override { return false; }
+  bool CancelSession() noexcept override { return true; }
+  bool FinishSession() noexcept override { return true; }
+
+private:
+  EventScript script_;
+};
+
+class FakeHttpClient : public http_client::HttpClient
+{
+public:
+  explicit FakeHttpClient(EventScript script) : script_(std::move(script)) {}
+  std::shared_ptr<http_client::Session> CreateSession(nostd::string_view) noexcept override
+  {
+    return std::make_shared<FakeSession>(script_);
+  }
+  bool CancelAllSessions() noexcept override { return true; }
+  bool FinishAllSessions() noexcept override { return true; }
+  void SetMaxSessionsPerConnection(std::size_t) noexcept override {}
+
+private:
+  EventScript script_;
+};
+
+opentelemetry::sdk::common::ExportResult ExportWith(EventScript script)
+{
+  auto client = std::make_shared<FakeHttpClient>(std::move(script));
+  logs_exporter::ElasticsearchExporterOptions options;
+  logs_exporter::ElasticsearchLogRecordExporter exporter(options, client);
+  auto record = exporter.MakeRecordable();
+  return exporter.Export(nostd::span<std::unique_ptr<sdklogs::Recordable>>(&record, 1));
+}
+}  // namespace
+
+// The synchronous wait exists only when the exporter is built without async export, so these cases
+// skip rather than compile out: gtest_add_tests reads the source, and a case that disappeared from
+// the binary would still be registered with CTest. The skip goes in SetUp rather than at the top of
+// each body, because GTEST_SKIP returns and leaves the rest of the body unreachable, which MSVC
+// reports as C4702 and the maintainer mode jobs turn into an error.
+namespace
+{
+class ElasticsearchLogsExporterSyncTests : public ::testing::Test
+{
+protected:
+  void SetUp() override
+  {
+#ifdef ENABLE_ASYNC_EXPORT
+    GTEST_SKIP() << "Export() returns without waiting when async export is enabled";
+#endif
+  }
+};
+}  // namespace
+
+TEST_F(ElasticsearchLogsExporterSyncTests, ResponseRecordedBeforeTheWaitIsStillSeen)
+{
+  const auto result = ExportWith([](http_client::EventHandler &handler) {
+    FakeResponse response(200, kAcceptedBody);
+    handler.OnResponse(response);
+  });
+  EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kSuccess);
+}
+
+TEST_F(ElasticsearchLogsExporterSyncTests, ReadErrorEndsTheWait)
+{
+  const auto result = ExportWith([](http_client::EventHandler &handler) {
+    handler.OnEvent(http_client::SessionState::ReadError, "");
+  });
+  EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure);
+}
+
+// The whole contract in one place. Every state that ends a session has to leave a result behind,
+// otherwise a client that emits it last strands the wait. The seven states other than the two
+// this change adds have behaved this way since #4298 but had no test.
+//
+// A regression here surfaces as a CTest timeout rather than a failed assertion, because a state
+// that stops being terminal leaves Export() waiting with nothing left to wake it.
+TEST_F(ElasticsearchLogsExporterSyncTests, EveryTerminalStateEndsTheWaitInFailure)
+{
+  const http_client::SessionState terminal[] = {
+      http_client::SessionState::CreateFailed, http_client::SessionState::ConnectFailed,
+      http_client::SessionState::SendFailed,   http_client::SessionState::SSLHandshakeFailed,
+      http_client::SessionState::TimedOut,     http_client::SessionState::NetworkError,
+      http_client::SessionState::Cancelled,    http_client::SessionState::ReadError,
+      http_client::SessionState::WriteError,   http_client::SessionState::Destroyed};
+
+  for (const auto state : terminal)
+  {
+    SCOPED_TRACE(static_cast<int>(state));
+    const auto result =
+        ExportWith([state](http_client::EventHandler &handler) { handler.OnEvent(state, ""); });
+    EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure);
+  }
+}
+
+// The other side of the same contract: a state that only reports progress must not complete the
+// export on its own, or a response that arrives afterwards is never consulted.
+TEST_F(ElasticsearchLogsExporterSyncTests, ProgressStatesDoNotDecideTheResult)
+{
+  const auto result = ExportWith([](http_client::EventHandler &handler) {
+    handler.OnEvent(http_client::SessionState::Created, "");
+    handler.OnEvent(http_client::SessionState::Connecting, "");
+    handler.OnEvent(http_client::SessionState::Connected, "");
+    handler.OnEvent(http_client::SessionState::Sending, "");
+    handler.OnEvent(http_client::SessionState::Response, "");
+    FakeResponse response(200, kAcceptedBody);
+    handler.OnResponse(response);
+  });
+  EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kSuccess);
+}
+
+TEST_F(ElasticsearchLogsExporterSyncTests, WriteErrorEndsTheWait)
+{
+  const auto result = ExportWith([](http_client::EventHandler &handler) {
+    handler.OnEvent(http_client::SessionState::WriteError, "");
+  });
+  EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure);
+}
+
+TEST_F(ElasticsearchLogsExporterSyncTests, SessionDestroyedWhilePendingEndsTheWait)
+{
+  const auto result = ExportWith([](http_client::EventHandler &handler) {
+    handler.OnEvent(http_client::SessionState::Destroyed, "");
+  });
+  EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure);
+}
+
+// The first outcome recorded is the one reported, so tearing the session down after a response has
+// arrived does not turn a successful export into a failure.
+TEST_F(ElasticsearchLogsExporterSyncTests, SessionDestroyedAfterAResponseKeepsTheSuccess)
+{
+  const auto result = ExportWith([](http_client::EventHandler &handler) {
+    FakeResponse response(200, kAcceptedBody);
+    handler.OnResponse(response);
+    handler.OnEvent(http_client::SessionState::Destroyed, "");
+  });
+  EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kSuccess);
+}
+
+// Same rule for the two states this change makes terminal, and the reason their error line is
+// conditional: reaching one of them says nothing on its own about the result Export() reports.
+TEST_F(ElasticsearchLogsExporterSyncTests, IoErrorAfterAResponseKeepsTheSuccess)
+{
+  for (const auto state :
+       {http_client::SessionState::ReadError, http_client::SessionState::WriteError})
+  {
+    const auto result = ExportWith([state](http_client::EventHandler &handler) {
+      FakeResponse response(200, kAcceptedBody);
+      handler.OnResponse(response);
+      handler.OnEvent(state, "");
+    });
+    EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kSuccess);
+  }
+}
+
+// The mirror image, and the case where the error line is the only diagnostic the caller gets:
+// the I/O error is recorded first, so a response arriving afterwards does not rescue the export.
+TEST_F(ElasticsearchLogsExporterSyncTests, IoErrorBeforeAResponseKeepsTheFailure)
+{
+  for (const auto state :
+       {http_client::SessionState::ReadError, http_client::SessionState::WriteError})
+  {
+    SCOPED_TRACE(static_cast<int>(state));
+    const auto result = ExportWith([state](http_client::EventHandler &handler) {
+      handler.OnEvent(state, "");
+      FakeResponse response(200, kAcceptedBody);
+      handler.OnResponse(response);
+    });
+    EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kFailure);
+  }
+}