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

CHANGELOG.md

@@ -81,6 +81,11 @@ Increment the:
   deprecated C headers (`stdint.h`, `stddef.h`, `stdlib.h`, `string.h`,
   `stdio.h`, `ctype.h`, `limits.h`, `assert.h`) with their C++ equivalents
   ([#4349](https://github.com/open-telemetry/opentelemetry-cpp/pull/4349))
+* [BUG] End the Elasticsearch exporter's wait on a read or write error
+  [#4331](https://github.com/open-telemetry/opentelemetry-cpp/pull/4331)
+* [BUG] Say why an Elasticsearch export failed when the session was destroyed
+  before a response arrived
+  [#4331](https://github.com/open-telemetry/opentelemetry-cpp/pull/4331)
 
 * [CONFIGURATION] Add SDK component builder interfaces to the registry
   [#4358](https://github.com/open-telemetry/opentelemetry-cpp/issues/4358)

exporters/elasticsearch/CMakeLists.txt

@@ -55,4 +55,8 @@ if(OTELCPP_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() # OTELCPP_BUILD_TESTING

exporters/elasticsearch/src/es_log_record_exporter.cc

@@ -110,8 +110,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()
   {
@@ -135,28 +135,43 @@ class ResponseHandler : public http_client::EventHandler
   // Callback method when an http event occurs
   void OnEvent(http_client::SessionState state, nostd::string_view /* reason */) noexcept override
   {
-    // If any failure event occurs, release the condition variable to unblock main thread
+    // If any failure event occurs, release the condition variable to unblock main thread.
+    //
+    // Recording is first writer wins, and only the event that decided the outcome reports it: a
+    // line from a later event would describe a failure the caller was never told about.
     switch (state)
     {
       case http_client::SessionState::CreateFailed:
-        OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Failed to create session");
-        recordCompletion(CompletionState::Failure);
+        if (recordCompletion(CompletionState::Failure))
+        {
+          OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Failed to create session");
+        }
         break;
       case http_client::SessionState::Created:
         OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Session created");
         break;
       case http_client::SessionState::Destroyed:
-        OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Session destroyed");
         // Nothing else will arrive after this. If no outcome was recorded, the session ended
-        // without a response, so release the waiter rather than leaving it blocked forever.
-        recordCompletion(CompletionState::Failure);
+        // without a response, so release the waiter rather than leaving it blocked forever, and
+        // say why: this is the event that decided the export failed, and the default level does
+        // not show debug.
+        if (recordCompletion(CompletionState::Failure))
+        {
+          OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Session destroyed before a response");
+        }
+        else
+        {
+          OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Session destroyed");
+        }
         break;
       case http_client::SessionState::Connecting:
         OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Connecting to peer");
         break;
       case http_client::SessionState::ConnectFailed:
-        OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Failed to connect to peer");
-        recordCompletion(CompletionState::Failure);
+        if (recordCompletion(CompletionState::Failure))
+        {
+          OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Failed to connect to peer");
+        }
         break;
       case http_client::SessionState::Connected:
         OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Connected to peer");
@@ -165,33 +180,49 @@ class ResponseHandler : public http_client::EventHandler
         OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Sending request");
         break;
       case http_client::SessionState::SendFailed:
-        OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Failed to send request");
-        recordCompletion(CompletionState::Failure);
+        if (recordCompletion(CompletionState::Failure))
+        {
+          OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Failed to send request");
+        }
         break;
       case http_client::SessionState::Response:
         OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Received response");
         break;
       case http_client::SessionState::SSLHandshakeFailed:
-        OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Failed SSL Handshake");
-        recordCompletion(CompletionState::Failure);
+        if (recordCompletion(CompletionState::Failure))
+        {
+          OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Failed SSL Handshake");
+        }
         break;
       case http_client::SessionState::TimedOut:
-        OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Request timed out");
-        recordCompletion(CompletionState::Failure);
+        if (recordCompletion(CompletionState::Failure))
+        {
+          OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Request timed out");
+        }
         break;
       case http_client::SessionState::NetworkError:
-        OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Network error");
-        recordCompletion(CompletionState::Failure);
+        if (recordCompletion(CompletionState::Failure))
+        {
+          OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Network error");
+        }
         break;
       case http_client::SessionState::ReadError:
-        OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Read error");
+        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");
-        recordCompletion(CompletionState::Failure);
+        if (recordCompletion(CompletionState::Failure))
+        {
+          OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] (manually) cancelled");
+        }
         break;
     }
   }
@@ -207,23 +238,29 @@ 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.
+   *
+   * @return whether this call is the one that decided the outcome
    */
-  void recordCompletion(CompletionState state)
+  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